LogAgent —— etcd+kafka+zookeeper+go实现实时读取日志发送到kafka，并实现热加载配置读取的日志路径

工具包目录结构：

.
├── conf
│   ├── logAgent.ini
│   └── logAgentConfig.go
├── etcd
│   └── etcd.go
├── kafka
│   └── kafka.go
├── main.go
└── taillog
　　　　├── taillog.go
　　　　└── taillog_mgr.go

logAgent.ini

[kafka]
address=127.0.0.1:9092
chan_max_size=100000

[etcd]
address=127.0.0.1:2379
timeout=5
collect_log_key=xxx

[taillog]
filename="./my.log"

logAgentConf.go

/**
 * @Author: Mr.Cheng
 * @Description:
 * @File: logAgentConfig
 * @Version: 1.0.0
 * @Date: 2021/12/9 下午8:44
 */

package logAgentConfig

type AppConf struct {
    KafkaConf `ini:"kafka"`
    EtcdConf  `ini:"etcd"`
    // TaillogConf `ini:"taillog"`
}

type KafkaConf struct {
    Address string `ini:"address"`
    Size    int    `ini:"chan_max_size"`
}

type EtcdConf struct {
    Address string `ini:"address"`
    Timeout int    `ini:"timeout"`
    Key     string `ini:"collect_log_key"`
}

// ----- unused ↓️----

type TaillogConf struct {
    FileName string `ini:"filename"`
}

etcd.go

/**
 * @Author: Mr.Cheng
 * @Description:
 * @File: etcd
 * @Version: 1.0.0
 * @Date: 2021/12/9 下午9:12
 */
package etcd

import (
    "context"
    "encoding/json"
    "fmt"
    "go.etcd.io/etcd/clientv3"
    "time"
)

var (
    client *clientv3.Client
)

type LogEntry struct {
    Path  string `json:"path"`  // 日志存放的路径
    Topic string `json:"topic"` // 日志要发往kafka的topic
}

func Init(address string, interval int) (err error) {
    client, err = clientv3.New(clientv3.Config{
        Endpoints:   []string{address},
        DialTimeout: time.Duration(interval) * time.Second,
    })
    if err != nil {
        fmt.Printf("connect to etcd failed, err:%v\n", err)
        return err
    }
    return
}

// 从Etcd中根据Key获取配置项
func GetConf(key string) (LogEntryConf []*LogEntry, err error) {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    resp, err := client.Get(ctx, key)
    cancel()
    if err != nil {
        fmt.Printf("get from etcd failed, err:%v\n", err)
        return nil, err
    }
    for _, ev := range resp.Kvs {
        //fmt.Printf("%s:%s\n", ev.Key, ev.Value)
        err = json.Unmarshal(ev.Value, &LogEntryConf)
        if err != nil {
            fmt.Printf("unmarshal etcd value failed, err:%v\n", err)
            return nil, err
        }
    }
    return LogEntryConf, nil
}

// etcd watch
func WatchConf(key string, newConfChan chan<- []*LogEntry) {
    ch := client.Watch(context.Background(), key)
    for wresp := range ch {
        for _, evt := range wresp.Events {
            fmt.Printf("Type:%v key:%v value:%v\n", evt.Type, string(evt.Kv.Key), string(evt.Kv.Value))
            var newConf []*LogEntry
            // 如果是删除操作，json.Unmarshal会报错，需手动添加一个空的newConf
            if evt.Type != clientv3.EventTypeDelete {
                err := json.Unmarshal(evt.Kv.Value, &newConf)
                if err != nil {
                    fmt.Printf("unmarshal new conf failed, err:%v\n", err)
                    continue
                }
            }
            newConfChan <- newConf
        }
    }
}

kafka.go

/**
 * @Author: Mr.Cheng
 * @Description:往kafka写入日志
 * @File: kafka
 * @Version: 1.0.0
 * @Date: 2021/12/9 下午2:19
 */

package kafka

import (
    "fmt"
    "github.com/Shopify/sarama"
    "time"
)

type logData struct {
    Topic string
    Data  string
}

var (
    client      sarama.SyncProducer // 全局连接kafka的生产者
    logDataChan chan *logData
)

// 初始化连接
func Init(address []string, size int) (err error) {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送模式（需leader和follow都确认）
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 选择分区的方式（轮询）
    config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel中返回

    // 连接kafka
    client, err = sarama.NewSyncProducer(address, config)
    if err != nil {
        fmt.Printf("client kafka failed, err:%v\n", err)
        return err
    }

    // 初始化logDataChan
    logDataChan = make(chan *logData, size)

    // 从logDataChan中取数据发往kafaka
    go sendToKafka()
    return nil
}

func SendToChan(Topic, Data string) {
    data := &logData{
        Topic: Topic,
        Data:  Data,
    }
    select {
    case logDataChan <- data:
    default:
        time.Sleep(time.Millisecond * 100)
    }
}

func sendToKafka() {
    // 循环从通道logDataChan取值并发送给kafka
    for {
        select {
        case data := <-logDataChan:
            msg := &sarama.ProducerMessage{}
            msg.Topic = data.Topic
            msg.Value = sarama.StringEncoder(data.Data)
            pid, offset, err := client.SendMessage(msg)
            if err != nil {
                fmt.Printf("send msg failed, err:%v\n", err)
            }
            fmt.Printf("send msg success, pid:%v offect:%v\n", pid, offset)
        default:
            time.Sleep(time.Millisecond * 50)
        }
    }
}

taillog.go

/**
 * @Author: Mr.Cheng
 * @Description:收集日志模块
 * @File: taillog
 * @Version: 1.0.0
 * @Date: 2021/12/8 下午9:54
 */

package taillog

import (
    "context"
    "day21/02.log_agent/kafka"
    "fmt"
    "github.com/hpcloud/tail"
    "time"
)

type TailTask struct {
    Path     string
    Topic    string
    Instance *tail.Tail
    // 为了停止任务，存下context
    ctx    context.Context
    cancel context.CancelFunc
}

func NewTailTask(Path, Topic string) (tailtask *TailTask, err error) {
    config := tail.Config{
        Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, // 从文件那个地方开始读
        ReOpen:    true,                                 // 重新打开
        MustExist: false,                                // 文件不存在不报错
        Poll:      true,
        Follow:    true, // 是否跟随
    }
    ctx, cancel := context.WithCancel(context.Background())
    tailObj, err := tail.TailFile(Path, config)
    if err != nil {
        fmt.Printf("tail file failed, err:%v\n", err)
        return nil, err
    }
    tailtask = &TailTask{Path: Path, Topic: Topic, Instance: tailObj, ctx: ctx, cancel: cancel}
    // 开启读取日志并发送给kafka
    go tailtask.ReadFromTail()
    return tailtask, nil
}

func (tailtask *TailTask) ReadFromTail() {
    for {
        select {
        case <-tailtask.ctx.Done():
            return
        case line, ok := <-tailtask.Instance.Lines:
            if !ok {
                fmt.Printf("tail fail close reopen, filename:%s\n", tailtask.Path)
                time.Sleep(time.Second)
                continue
            }
            kafka.SendToChan(tailtask.Topic, line.Text)
        default:
            time.Sleep(time.Second)
        }
    }
}

taillog_mgr.go

/**
 * @Author: Mr.Cheng
 * @Description:
 * @File: taillogMgr
 * @Version: 1.0.0
 * @Date: 2021/12/14 下午3:47
 */

package taillog

import (
    "day21/02.log_agent/etcd"
    "fmt"
    "time"
)

type TailMgr struct {
    logEntry    []*etcd.LogEntry
    tskMap      map[string]*TailTask
    newConfChan chan []*etcd.LogEntry
}

var tskMgr *TailMgr

// 循环每一个日志收集项，创建tailObj，并发往kafka
func Init(logEntryConf []*etcd.LogEntry) {
    tskMgr = &TailMgr{
        logEntry:    logEntryConf,
        tskMap:      make(map[string]*TailTask, 16),
        newConfChan: make(chan []*etcd.LogEntry),
    }

    for _, LogEntry := range logEntryConf {
        // fmt.Printf("Path:%v Topic:%v\n", LogEntry.Path, LogEntry.Topic)
        tailtask, err := NewTailTask(LogEntry.Path, LogEntry.Topic)
        if err != nil {
            continue
        }
        // 在tskMap中存储一下，以便发生配置变更时做增删改操作
        key := fmt.Sprintf("%s_%s", tailtask.Path, tailtask.Topic)
        tskMgr.tskMap[key] = tailtask
    }

    go tskMgr.run()
}

// 监听newConfChan是否有数据，有数据则表示etcd配置有变化，需做相应的处理
func (t *TailMgr) run() {
    for {
        select {
        case newConf := <- t.newConfChan:
            fmt.Printf("配置发生变更，Conf:%v\n", newConf)
            // 找出新增项
            for _, logEntry := range newConf {
                key := fmt.Sprintf("%s_%s", logEntry.Path, logEntry.Topic)
                _, ok := t.tskMap[key]
                if ok {
                    // 表示该配置项原先存在
                    continue
                } else {
                    // 属于新增配置
                    fmt.Printf("新增项，path:%s topic:%s\n", logEntry.Path, logEntry.Topic)
                    tailtask, err := NewTailTask(logEntry.Path, logEntry.Topic)
                    if err != nil {
                        continue
                    }
                    // TailMgr的logEntry和tskMap增加对应项
                    t.logEntry = append(t.logEntry, logEntry)
                    t.tskMap[key] = tailtask
                    go tailtask.ReadFromTail()
                }
            }
            // 找出删除项
            for index, c1 := range t.logEntry {
                isDelete := true
                for _, c2 := range newConf {
                    if c1.Path == c2.Path && c1.Topic == c2.Topic {
                        isDelete = false
                        break
                    }
                }
                if isDelete{
                    // 表示属于删除项，从tskMap拿出tailtask对象，执行对象的cancel函数，并将该对象从tskMap中删除
                    fmt.Printf("删除项，path:%s topic:%s\n", c1.Path, c1.Topic)
                    key := fmt.Sprintf("%s_%s", c1.Path, c1.Topic)
                    t.tskMap[key].cancel()
                    // TailMgr的logEntry和tskMap删除对应项
                    delete(t.tskMap, key)
                    t.logEntry = append(t.logEntry[:index], t.logEntry[index + 1:]...)
                }
            }
        default:
            time.Sleep(time.Second)
        }
    }
}

// 向外暴露newConfChan
func NewConfChan() chan<- []*etcd.LogEntry{
    return tskMgr.newConfChan
}

main.go

/**
 * @Author: Mr.Cheng
 * @Description:
 * @File: main
 * @Version: 1.0.0
 * @Date: 2021/12/9 下午8:43
 */

package main

import (
    logAgentConfig "day21/02.log_agent/conf"
    "day21/02.log_agent/etcd"
    "day21/02.log_agent/kafka"
    "day21/02.log_agent/taillog"
    "fmt"
    "gopkg.in/ini.v1"
    "sync"
)

var (
    cfg = new(logAgentConfig.AppConf)
    wg  sync.WaitGroup
)

func main() {
    // 加载配置文件
    err := ini.MapTo(cfg, "./conf/logAgent.ini")
    if err != nil {
        fmt.Printf("load ini failed, err:%v\n", err)
        return
    }

    // 初始化kafka连接
    err = kafka.Init([]string{cfg.KafkaConf.Address}, cfg.KafkaConf.Size)
    if err != nil {
        return
    }
    fmt.Println("init kafka success")

    // 初始化etcd
    err = etcd.Init(cfg.EtcdConf.Address, cfg.EtcdConf.Timeout)
    if err != nil {
        return
    }
    fmt.Println("init etcd success")

    // 从etcd中获取日志收集项的配置信息
    logEntryConf, err := etcd.GetConf(cfg.EtcdConf.Key)
    if err != nil {
        return
    }
    fmt.Printf("get conf from etcd success, conf:%v\n", logEntryConf)

    // 收集日志发往kafka
    // 循环每一个日志收集项，创建tailObj，并发往kafka
    taillog.Init(logEntryConf)

    // 监视etcd中配置的变动，如有变动，给新的配置信息给taillog
    wg.Add(1)
    go etcd.WatchConf(cfg.EtcdConf.Key, taillog.NewConfChan())
    wg.Wait()
}