Golang日志收集工具实现与实战

Go需自行构建日志收集管道：用fsnotify监听文件轮转并动态添加新文件，bufio.Scanner安全读取行（设10MB缓冲），原始字节封装为JSON或gob结构化日志，控制发送背压防OOM。

Go 本身没有内置的集中式日志收集能力，log 包只负责本地输出；要实现真正的日志收集工具（比如从多个服务抓日志、转发到 Kafka / ES / 文件归档），必须自己组装管道：监听文件变化 + 解析行 + 过滤/增强 + 序列化 + 发送。

用 fsnotify 监听日志文件追加写入

Linux 下日志轮转（如 logrotate）会重命名或删除旧文件，tail -f 能跟住是因为它检测 inotify 的 IN_MOVED_FROM 和 IN_CREATE 事件。Go 没有原生支持，得靠 fsnotify 手动处理：

• 对每个目标日志路径调用 watcher.Add()，但注意：不能只 Add 一次就完事——轮转后新文件名（如 app.log.1 或 app.log.2024-06-01）不会自动被监听，需结合 filepath.Glob 定期扫描 + 对新增文件调用 Add()
• fsnotify.Event.Op 中真正代表“新内容追加”的是 fsnotify.Write，但某些文件系统（如 NFS）可能不触发该事件，此时得 fallback 到定期 stat 检查 Size 变化
• 避免重复读：记录每个文件的 os.FileInfo.Sys().(*syscall.Stat_t).Ino 和 dev，防止硬链接或同名覆盖导致 offset 错乱

用 bufio.Scanner 安全读取增量日志行

别直接用 ReadLine 或 ReadString('\n')——日志行可能超长、可能含二进制数据、可能因 crash 导致半截写入。bufio.Scanner 更稳妥，但默认 MaxScanTokenSize 是 64KB：

• 启动前务必调用 scanner.Buffer(make([]byte, 4096), 10*1024*1024)，把 max 设为 10MB（根据业务日志单行最大长度预估）
• 遇到 scanner.Err() == bufio.ErrTooLong 时，说明某行爆了 buffer，此时应丢弃该行并记录告警（log.Printf("line too long in %s, skipped", path)），而不是 panic 或阻塞
• 每次 scanner.Scan() 后立即用 scanner.Bytes() 拿原始字节，别转 string——避免 UTF-8 解码失败，后续做 JSON 封装或 Base64 编码更安全

用 gob 或 json.RawMessage 封装结构化日志再发送

原始日志行基本是纯文本，但收集端（如 Loki、Logstash）需要字段：时间戳、服务名、level、trace_id。硬解析正则太脆弱，推荐两种轻量方案：

• 如果日志已按 JSON 输出（如 zap 的 json.NewEncoder），直接用 json.RawMessage 零拷贝包装：map[string]interface{}{"ts": time.Now().UTC().Format(time.RFC3339), "host": hostname, "log": json.RawMessage(lineBytes)}
• 如果日志是文本格式（如 "INFO [2024-06-01T12:34:56Z] user login"），不要现场解析，先存原始 []byte，加固定字段：struct{ Time time.Time; Host string; Raw []byte }{time.Now(), hostname, lineBytes}，再用 gob.Encoder 编码——比 JSON 快 30%，且天然支持 []byte
• 无论哪种，发送前检查总长度，超 1MB（HTTP body 上限常见值）就拆包，否则 Kafka 默认消息上限是 1MB，ES bulk API 也常设限

用 net/http 或 sarama 发送到远端时控制背压

日志产生速度可能远高于网络吞吐，没流控会导致内存暴涨 OOM。不能简单起 goroutine 发送：

• 用带缓冲的 channel 做第一道队列，容量设为 1000～5000（根据内存预算）；生产者往里塞 logEntry，消费者从 channel 拉取后发 HTTP POST 或 Kafka
• HTTP 发送时，复用 &http.Client{Timeout: 5 * time.Second}，并检查 resp.StatusCode ——429 Too Many Requests 或 503 Service Unavailable 出现时，把当前 batch 放回 channel 前端（用 select { case ch <- entry: ... default: } 避免阻塞）
• Kafka 场景下，sarama.SyncProducer 会阻塞，改用 sarama.AsyncProducer，监听 Errors() 和 Successes() channel，并在 Errors() 回调里重试（最多 2 次，间隔 100ms），失败三次则写本地 fallback 文件

最易被忽略的是文件 inode 复用和轮转间隙丢失——当 logrotate rename + create 两个操作之间存在微小时间窗，fsnotify 可能漏掉第一条新日志；必须在每次 detect 到新文件时，从末尾向前扫描 1KB，找最后一个完整换行符位置作为起始 offset，而不是直接 Seek(0, io.SeekEnd)。