task

Documentation

Index

• Constants
• func ConnectToTask(task *Task) channel.Connector
• func FormatterRegister(name string, factory FormatterFactory) error
• func SetResourceLimit(maxCpuLimit, checkTimes int)
• type Formatter
• type FormatterFactory
  • func FindFormatterFactory(name string) (FormatterFactory, error)
• type Processors
  • func NewProcessors(config *config.TaskConfig) (*Processors, error)
  • func (client *Processors) Run(event *beat.Event) *beat.Event
• type PublisherFunc
• type Sender
  • func NewSender(config *cfg.TaskConfig, taskDone chan struct{}, publisher PublisherFunc) (*Sender, error)
  • func (client *Sender) OnEvent(data *util.Data) bool
  • func (client *Sender) Start() error
  • func (client *Sender) String() string
  • func (client *Sender) Wait()
• type Task
  • func NewTask(config *cfg.TaskConfig, beatDone chan struct{}) *Task
  • func (task *Task) Close() error
  • func (task *Task) Done() <-chan struct{}
  • func (task *Task) OnEvent(data *util.Data) bool
  • func (task *Task) Reload() error
  • func (task *Task) Start(lastStates []file.State) error
  • func (task *Task) Stop() error
  • func (task *Task) String() string

Constants

const (
	EqualOperation = iota
	NotEqualOperation
)

Functions

func ConnectToTask

func ConnectToTask(task *Task) channel.Connector

ConnectToTask 返回任务实例，用于接收采集事件 OnEvent

func FormatterRegister

func FormatterRegister(name string, factory FormatterFactory) error

FormatterRegister: 注册sender输出方法

func SetResourceLimit

func SetResourceLimit(maxCpuLimit, checkTimes int)

Types

type Formatter

type Formatter interface {
	Format([]*util.Data) beat.MapStr
}

Formatter: 采集器事件包格式化接口, 根据任务配置返回相应的格式

type FormatterFactory

type FormatterFactory = func(config *config.TaskConfig) (Formatter, error)

FormatterFactory is used by output plugins to build an output instance

func FindFormatterFactory

func FindFormatterFactory(name string) (FormatterFactory, error)

FindFormatterFactory: 获取格式化器实例

type Processors

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

Processors: 兼容数据平台过滤规则

func NewProcessors

func NewProcessors(config *config.TaskConfig) (*Processors, error)

NewProcessors: 兼容原采集器处理并复用filebeat.processors

func (*Processors) Run

func (client *Processors) Run(event *beat.Event) *beat.Event

Run: 处理采集事件

type PublisherFunc

type PublisherFunc func(beat.Event) bool

Publisher: 接收采集事件并发送到outlet

type Sender

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

Sender: 对采集事件进行打包, 并调用beat发送事件

func NewSender

func NewSender(config *cfg.TaskConfig, taskDone chan struct{}, publisher PublisherFunc) (*Sender, error)

NewSender 生成采集器Sender实例

func (*Sender) OnEvent

func (client *Sender) OnEvent(data *util.Data) bool

OnEvent获取采集事件

func (*Sender) Start

func (client *Sender) Start() error

Start 启动Sender实例

func (*Sender) String

func (client *Sender) String() string

Sender 实例名称

func (*Sender) Wait

func (client *Sender) Wait()

Wait

type Task

type Task struct {
	ID string
	// contains filtered or unexported fields
}

Task： 采集任务具体实现，负责filebeat采集事件处理、过滤、打包，并发送到采集框架

func NewTask

func NewTask(config *cfg.TaskConfig, beatDone chan struct{}) *Task

NewTask 生成采集任务实例

func (*Task) Close

func (task *Task) Close() error

Close 由Filebeat在停止采集插件后调用

func (*Task) Done

func (task *Task) Done() <-chan struct{}

Done 返回任务状态channel

func (*Task) OnEvent

func (task *Task) OnEvent(data *util.Data) bool

处理input runner发送的事件

func (*Task) Reload

func (task *Task) Reload() error

Reload 通知各采集模块针对重载操作进行适配

func (*Task) Start

func (task *Task) Start(lastStates []file.State) error

Start 负责启动采集任务实例

func (*Task) Stop

func (task *Task) Stop() error

Stop 负责停止采集任务实例，在Filebeat采集插件停止后退出

func (*Task) String

func (task *Task) String() string

String 任务实例名称