log-agent/taillog/taillog.go

package taillog

import (
	"context"
	"fmt"
	"github.com/hpcloud/tail"
	"logagent/kafka"
)

//专门从日志文件收集日志的模块

//一个日志收集的任务

type TailTask struct {
	path     string
	topic    string
	instance *tail.Tail
	//为了实现退出t.run()
	ctx    context.Context
	cancel context.CancelFunc
}

func NewTailTask(path string, topic string) (tailChan *TailTask) {
	ctx, cancel := context.WithCancel(context.Background())
	tailChan = &TailTask{
		path:   path,
		topic:  topic,
		ctx:    ctx,
		cancel: cancel,
	}
	tailChan.init() //根据路径去打开日志
	return
}

func (t *TailTask) init() {
	config := tail.Config{
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //从文件的哪个地方开始读 filebeat记录了文件断点的位置
		ReOpen:    true,                                 //重新打开，切分用
		MustExist: false,                                //文件不存在不报错
		Poll:      true,
		Follow:    true, //是否跟随
	}
	var err error
	t.instance, err = tail.TailFile(t.path, config)
	if err != nil {
		fmt.Printf("tail file failed,err:%v\n", err)
		return
	}
	//当goroutine执行的函数退出的时候，goroutine就结束了
	go t.run() //直接采集日志发送到kafka
}

func (t *TailTask) ReadChan() <-chan *tail.Line {
	return t.instance.Lines
}

func (t *TailTask) run() {
	for {
		select {
		case <-t.ctx.Done():
			fmt.Printf("tail task:%v exit\n", t.path+t.topic)
			return
		case line := <-t.ReadChan():
			fmt.Printf("日志已发送channel，text：%v\n", line.Text)
			//kafka.SendToKafka(t.topic, line.Text) //函数调用函数
			//先把日志发送到一个通道中
			kafka.SendToChan(t.topic, line.Text)
			//kafka哪个包中有一个单独的goroutine去取日志数据发送到kafka
		}
	}
}