读猿码系列——3. 从filebeat和go-stash深入日志收集及处理（filebeat篇）

├── autodiscover        # 包含filebeat的autodiscover适配器（adapter），当autodiscover发现新容器时创建对应类型的输入
├── beater              # 包含与libbeat库交互相关的文件
├── channel             # 包含filebeat输出到pipeline相关的文件
├── config              # 包含filebeat配置结构和解析函数
├── fileset             # 包含module和fileset相关的结构
├── harvester           # 包含Harvester接口定义、Reader接口及实现等
├── input               # 包含所有输入类型的实现（比如: log, stdin, syslog）
├── inputsource         # 在syslog输入类型中用于读取tcp或udp syslog
├── module              # 包含各module和fileset配置
├── modules.d           # 包含各module对应的日志路径配置文件，用于修改默认路径
├── processor           # 用于从容器日志的事件字段source中提取容器id
├── registrar           # 包含Registrar结构和方法
└── ...

type crawler struct {
    log             *logp.Logger
    inputs          map[uint64]cfgfile.Runner // 包含所有输入的runner
    inputConfigs    []*conf.C+
    wg              sync.WaitGroup
    inputsFactory   cfgfile.RunnerFactory
    modulesFactory  cfgfile.RunnerFactory
    modulesReloader *cfgfile.Reloader
    inputReloader   *cfgfile.Reloader
    once            bool
    beatDone        chan struct{}
}

for _, inputConfig := range c.inputConfigs {
    err := c.startInput(pipeline, inputConfig)
    if err != nil {
        return fmt.Errorf("starting input failed: %w", err)
    }
}

for _, path := range p.config.Paths {
    matches, err := filepath.Glob(path)
    if err != nil {
        logger.Errorf("glob(%s) failed: %v", path, err)
        continue
    }...
}

// Input contains the input and its config
type Input struct {
    cfg                 *conf.C
    logger              *logp.Logger
    config              config
    states              *file.States
    harvesters          *harvester.Registry
    outlet              channel.Outleter
    stateOutlet         channel.Outleter
    done                chan struct{}
    numHarvesters       atomic.Uint32
    meta                map[string]string
    stopOnce            sync.Once
    fileStateIdentifier file.StateIdentifier
}

func (h *Harvester) Run() error {
    ...
    for {
        ...
        message, err := h.reader.Next()
        if err != nil {
            switch err {
            case ErrFileTruncate:
                logger.Info("File was truncated. Begin reading file from offset 0.")
                h.state.Offset = 0
                filesTruncated.Add(1)
            case ErrRemoved:
                logger.Info("File was removed. Closing because close_removed is enabled.")
            case ErrRenamed:
                logger.Info("File was renamed. Closing because close_renamed is enabled.")
            case ErrClosed:
                logger.Info("Reader was closed. Closing.")
            case io.EOF:
                logger.Info("End of file reached. Closing because close_eof is enabled.")
            case ErrInactive:
                logger.Infof("File is inactive. Closing because close_inactive of %v reached.", h.config.CloseInactive)
            default:
                logger.Errorf("Read line error: %v", err)
            }
            return nil
        }
        ...
        // Stop harvester in case of an error
        if !h.onMessage(forwarder, state, message, startingOffset) {
            return nil
        }
        ...
    }
}

err := forwarder.Send(beat.Event{
    Timestamp: timestamp,
    Fields:    fields,
    Meta:      meta,
    Private:   state,
})    

// Harvester contains all harvester related data
type Harvester struct {
    logger *logp.Logger

    id     uuid.UUID
    config config
    source harvester.Source // the source being watched

    // shutdown handling
    done     chan struct{}
    doneWg   *sync.WaitGroup
    stopOnce sync.Once
    stopWg   *sync.WaitGroup
    stopLock sync.Mutex

    // internal harvester state
    state  file.State
    states *file.States
    log    *Log

    // file reader pipeline
    reader          reader.Reader
    encodingFactory encoding.EncodingFactory
    encoding        encoding.Encoding

    // event/state publishing
    outletFactory OutletFactory
    publishState  func(file.State) bool

    metrics *harvesterProgressMetrics

    onTerminate func()
}

type bufferingEventLoop struct {
    broker *broker

    buf       *batchBuffer
    flushList flushList
    eventCount int

    unackedEventCount int

    minEvents    int
    maxEvents    int
    flushTimeout time.Duration

    // ack handling
    pendingACKs chanList

    // buffer flush timer state
    timer *time.Timer
    idleC <-chan time.Time
}    

func (l *bufferingEventLoop) run() {
    broker := l.broker

    for {
        var pushChan chan pushRequest
        if l.eventCount < l.maxEvents {
            pushChan = l.broker.pushChan
        }

        var getChan chan getRequest
        if !l.flushList.empty() {
            getChan = l.broker.getChan
        }

        var schedACKs chan chanList
        if !l.pendingACKs.empty() {
            schedACKs = l.broker.scheduledACKs
        }

        select {
        case <-broker.done:
            return

        case req := <-pushChan: 
            l.handleInsert(&req)

        case req := <-l.broker.cancelChan: 
            l.handleCancel(&req)

        case req := <-getChan:
            l.handleGetRequest(&req)

        case schedACKs <- l.pendingACKs:
            l.pendingACKs = chanList{}

        case count := <-l.broker.ackChan:
            l.handleACK(count)

        case req := <-l.broker.metricChan: 
            l.handleMetricsRequest(&req)

        case <-l.idleC:
            l.idleC = nil
            l.timer.Stop()
            if l.buf.length() > 0 {
                l.flushBuffer()
            }
        }
    }
}

req.responseChan <- getResponse{acker.ackChan, entries}

func (c *eventConsumer) run() {
    queueReader := makeQueueReader()
    go func() {
        queueReader.run(log)
    }()
outerLoop:
    for {
        if queueBatch == nil && !pendingRead {
            pendingRead = true
            queueReader.req <- queueReaderRequest{
                queue:      c.queue,
                retryer:    c,
                batchSize:  target.batchSize,
                timeToLive: target.timeToLive,
            }
        }
        ...
        var outputChan chan publisher.Batch
        if active != nil {
            outputChan = target.ch
        }
        select {
        case outputChan <- active:
            if len(retryBatches) > 0 {
                c.observer.eventsRetry(len(active.Events()))
                retryBatches = retryBatches[1:]
            } else {
                queueBatch = nil
            }

        case target = <-c.targetChan:

        case queueBatch = <-queueReader.resp:
            pendingRead = false

        case req := <-c.retryChan:
            ...
            retryBatches = append(retryBatches, req.batch)

        case <-c.done:
            break outerLoop
        }
    }
}

func (b *broker) Get(count int) (queue.Batch, error) {
    responseChan := make(chan getResponse, 1)
    select {
    case <-b.done:
        return nil, io.EOF
    case b.getChan <- getRequest{
        entryCount: count, responseChan: responseChan}:
    }
    resp := <-responseChan
    return &batch{
        queue:   b,
        entries: resp.entries,
        ackChan: resp.ackChan,
    }, nil
}

type getRequest struct {
    entryCount   int              
    responseChan chan getResponse 
}

type getResponse struct {
    ackChan chan batchAckMsg
    entries []queueEntry
}

type queueEntry struct {
    event  interface{}
    client clientState
}

type Client interface {
    Close() error
    Publish(context.Context, publisher.Batch) error
    String() string
}

func (c *client) Publish(_ context.Context, batch publisher.Batch) error {
    events := batch.Events()
    c.observer.NewBatch(len(events))

    ref := &msgRef{
        client: c,
        count:  int32(len(events)),
        total:  len(events),
        failed: nil,
        batch:  batch,
    }

    ch := c.producer.Input()
    for i := range events {
        d := &events[i]
        msg, err := c.getEventMessage(d)
        if err != nil {
            c.log.Errorf("Dropping event: %+v", err)
            ref.done()
            c.observer.Dropped(1)
            continue
        }

        msg.ref = ref
        msg.initProducerMessage()
        ch <- &msg.msg
    }

    return nil
}