postFilters

Documentation

Index

• func LoadReTagMap(env string, mapi interface{}) map[string]string
• type BaseFilter
  • func (f *BaseFilter) DiscardMsg(msg *libs.FluentMsg)
  • func (f *BaseFilter) SetMsgPool(msgPool *sync.Pool)
  • func (f *BaseFilter) SetUpstream(upChan chan *libs.FluentMsg)
  • func (f *BaseFilter) SetWaitCommitChan(waitCommitChan chan<- *libs.FluentMsg)
• type CustomBigDataFilter
  • func NewCustomBigDataFilter(cfg *CustomBigDataFilterCfg) *CustomBigDataFilter
  • func (f *CustomBigDataFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg
• type CustomBigDataFilterCfg
• type DefaultFilter
  • func NewDefaultFilter(cfg *DefaultFilterCfg) *DefaultFilter
  • func (f *DefaultFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg
• type DefaultFilterCfg
• type ESDispatcherFilter
  • func NewESDispatcherFilter(cfg *ESDispatcherFilterCfg) *ESDispatcherFilter
  • func (f *ESDispatcherFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg
• type ESDispatcherFilterCfg
• type FieldsFilter
  • func NewFieldsFilter(cfg *FieldsFilterCfg) *FieldsFilter
  • func (f *FieldsFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg
• type FieldsFilterCfg
• type ForwardTagRewriterFilter
  • func NewForwardTagRewriterFilter(cfg *ForwardTagRewriterFilterCfg) *ForwardTagRewriterFilter
  • func (f *ForwardTagRewriterFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg
• type ForwardTagRewriterFilterCfg
• type PostFilterItf
• type PostPipeline
  • func NewPostPipeline(cfg *PostPipelineCfg, filters ...PostFilterItf) *PostPipeline
  • func (f *PostPipeline) Wrap(ctx context.Context, inChan chan *libs.FluentMsg) (outChan chan *libs.FluentMsg)
• type PostPipelineCfg

Functions

func LoadReTagMap

func LoadReTagMap(env string, mapi interface{}) map[string]string

LoadReTagMap parse retag config app.spring.{env}: es-general -> {app.spring.sit: es-general}

Types

type BaseFilter

type BaseFilter struct {
	// contains filtered or unexported fields
}

func (*BaseFilter) DiscardMsg

func (f *BaseFilter) DiscardMsg(msg *libs.FluentMsg)

func (*BaseFilter) SetMsgPool

func (f *BaseFilter) SetMsgPool(msgPool *sync.Pool)

func (*BaseFilter) SetUpstream

func (f *BaseFilter) SetUpstream(upChan chan *libs.FluentMsg)

func (*BaseFilter) SetWaitCommitChan

func (f *BaseFilter) SetWaitCommitChan(waitCommitChan chan<- *libs.FluentMsg)

type CustomBigDataFilter

type CustomBigDataFilter struct {
	BaseFilter
	*CustomBigDataFilterCfg
	// contains filtered or unexported fields
}

CustomBigDataFilter specific hardcoding

func NewCustomBigDataFilter

func NewCustomBigDataFilter(cfg *CustomBigDataFilterCfg) *CustomBigDataFilter

func (*CustomBigDataFilter) Filter

func (f *CustomBigDataFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg

type CustomBigDataFilterCfg

type CustomBigDataFilterCfg struct {
	Tags []string
}

type DefaultFilter

type DefaultFilter struct {
	BaseFilter
	*DefaultFilterCfg
}

func NewDefaultFilter

func NewDefaultFilter(cfg *DefaultFilterCfg) *DefaultFilter

func (*DefaultFilter) Filter

func (f *DefaultFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg

type DefaultFilterCfg

type DefaultFilterCfg struct {
	MsgKey string
	MaxLen int
}

type ESDispatcherFilter

type ESDispatcherFilter struct {
	BaseFilter
	*ESDispatcherFilterCfg
	// contains filtered or unexported fields
}

func NewESDispatcherFilter

func NewESDispatcherFilter(cfg *ESDispatcherFilterCfg) *ESDispatcherFilter

func (*ESDispatcherFilter) Filter

func (f *ESDispatcherFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg

type ESDispatcherFilterCfg

type ESDispatcherFilterCfg struct {
	TagKey   string
	Tags     []string
	ReTagMap map[string]string
}

type FieldsFilter

type FieldsFilter struct {
	BaseFilter
	*FieldsFilterCfg
	// contains filtered or unexported fields
}

func NewFieldsFilter

func NewFieldsFilter(cfg *FieldsFilterCfg) *FieldsFilter

func (*FieldsFilter) Filter

func (f *FieldsFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg

type FieldsFilterCfg

type FieldsFilterCfg struct {
	Tags,
	IncludeFields, ExcludeFields []string // filter fields
	NewFieldTemplates map[string]string
}

type ForwardTagRewriterFilter

type ForwardTagRewriterFilter struct {
	BaseFilter
	*ForwardTagRewriterFilterCfg
	// contains filtered or unexported fields
}

ForwardTagRewriterFilter rewrite tag for msgs received by forward-recv. for example, change `forward-wechat.perf` -> `forward-wechat.prod`.

func NewForwardTagRewriterFilter

func NewForwardTagRewriterFilter(cfg *ForwardTagRewriterFilterCfg) *ForwardTagRewriterFilter

func (*ForwardTagRewriterFilter) Filter

func (f *ForwardTagRewriterFilter) Filter(msg *libs.FluentMsg) *libs.FluentMsg

type ForwardTagRewriterFilterCfg

type ForwardTagRewriterFilterCfg struct {
	TagKey, Tag string
}

type PostFilterItf

type PostFilterItf interface {
	SetUpstream(chan *libs.FluentMsg)
	SetMsgPool(*sync.Pool)
	SetWaitCommitChan(chan<- *libs.FluentMsg)

	Filter(*libs.FluentMsg) *libs.FluentMsg
	DiscardMsg(*libs.FluentMsg)
}

type PostPipeline

type PostPipeline struct {
	*PostPipelineCfg
	// contains filtered or unexported fields
}

func NewPostPipeline

func NewPostPipeline(cfg *PostPipelineCfg, filters ...PostFilterItf) *PostPipeline

func (*PostPipeline) Wrap

func (f *PostPipeline) Wrap(ctx context.Context, inChan chan *libs.FluentMsg) (outChan chan *libs.FluentMsg)

type PostPipelineCfg

type PostPipelineCfg struct {
	MsgPool                             *sync.Pool
	WaitCommitChan                      chan<- *libs.FluentMsg
	ReEnterChanSize, OutChanSize, NFork int
}