filebeat 数据采集流程

本文主要是介绍filebeat 数据采集流程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

filebeat启动流程讲解了filebeat的启动流程，filebeat在构建完crawler对象，开始采集流程。

Crawler的start方法内，会启动Inputs

func (c *Crawler) Start(pipeline beat.Pipeline,r *registrar.Registrar,configInputs *common.Config,configModules *common.Config,pipelineLoaderFactory fileset.PipelineLoaderFactory,overwritePipelines bool,
) error {...for _, inputConfig := range c.inputConfigs {err := c.startInput(pipeline, inputConfig, r.GetStates())if err != nil {return err}}...
}

c.startInput(pipeline, inputConfig, r.GetStates())方法初始化Input。

首先构建Input对象
运行Input

func (c *Crawler) startInput(pipeline beat.Pipeline,config *common.Config,states []file.State,
) error {if !config.Enabled() {return nil}connector := channel.ConnectTo(pipeline, c.out)p, err := input.New(config, connector, c.beatDone, states, nil)...// 开始收集p.Start()return nil
}

p.Start()方法内启动Input，他在一个单独的协程里运行。

这里的p是对Input的封装，他的Run方法是对某个接口的实现，因为我们用来收集日志，所以我们只需要关心filebeat/input/log/input.go文件内的Run方法。Run方法内部调用了Input的scan方法，开始采集数据。

// Run runs the input
func (p *Input) Run() {...p.scan()...
}

scan方法内首先获取所有的文件。其次获取文件状态，根据状态来判定收集最新数据，还是从历史文件收集。文件收集会构建Harvester对象。

// Scan starts a scanGlob for each provided path/glob
func (p *Input) scan() {var sortInfos []FileSortInfovar files []string// 获取应该获取到的所有文件paths := p.getFiles()var err error...for i := 0; i < len(paths); i++ {var path stringvar info os.FileInfoif sortInfos == nil {path = files[i]info = paths[path]} else {path = sortInfos[i].pathinfo = sortInfos[i].info}select {case <-p.done:logp.Info("Scan aborted because input stopped.")returndefault:}newState, err := getFileState(path, info, p)if err != nil {logp.Err("Skipping file %s due to error %s", path, err)}// Load last statelastState := p.states.FindPrevious(newState)...// Decides if previous state existsif lastState.IsEmpty() {logp.Debug("input", "Start harvester for new file: %s", newState.Source)// 准备构建 harvester 了err := p.startHarvester(newState, 0)if err == errHarvesterLimit {logp.Debug("input", harvesterErrMsg, newState.Source, err)continue}if err != nil {logp.Err(harvesterErrMsg, newState.Source, err)}} else {// 从历史文件开始处理p.harvestExistingFile(newState, lastState)}}
}

p.startHarvester(newState, 0)内构建harvester。(harvester是另一个filebeat官网描述的核心组件之一)

func (p *Input) startHarvester(state file.State, offset int64) error {if p.numHarvesters.Inc() > p.config.HarvesterLimit && p.config.HarvesterLimit > 0 {p.numHarvesters.Dec()harvesterSkipped.Add(1)return errHarvesterLimit}// Set state to "not" finished to indicate that a harvester is runningstate.Finished = falsestate.Offset = offset// Create harvester with state// 这部分构建了 harvesterh, err := p.createHarvester(state, func() { p.numHarvesters.Dec() })if err != nil {p.numHarvesters.Dec()return err}// 配置 harvestererr = h.Setup()if err != nil {p.numHarvesters.Dec()return fmt.Errorf("error setting up harvester: %s", err)}// Update state before staring harvester// This makes sure the states is set to Finished: false// This is synchronous state update as part of the scanh.SendStateUpdate()// 启动 harvesterif err = p.harvesters.Start(h); err != nil {p.numHarvesters.Dec()}return err
}

p.createHarvester构建harvester
p.Setup配置harvester。Setup方法内会初始化文件相关的内容，以及构建文件reader。
p.harvesters.Start(h)运行harvester

主要还是要看harvesters.Start方法，会在单独的协程内运行harvester。

func (r *Registry) Start(h Harvester) error {// Make sure stop is not called during starting a harvesterr.Lock()defer r.Unlock()...go func() {defer func() {r.remove(h)r.wg.Done()}()// 异步运行err := h.Run()if err != nil {logp.Err("Error running input: %v", err)}}()return nil
}

harvester.Run方法真是长。。

func (h *Harvester) Run() error {// 这坨简直了for {...// 读取文件内容message, err := h.reader.Next()// 糟糕的异常处理。。。if err != nil {switch err {...}return nil}state := h.getState()startingOffset := state.Offsetstate.Offset += int64(message.Bytes)...// 读取到的文件内容text := string(message.Content)...// 数据内容都包装在 data 内，harvester 发送 data，其实就是 forwarder 转发的if !h.sendEvent(data, forwarder) {return nil}// Update state of harvester as successfully senth.state = state}
}