134 lines
3.2 KiB
Go
134 lines
3.2 KiB
Go
package collect
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"io"
|
||
"os"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/checkpoint"
|
||
"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
|
||
)
|
||
|
||
// LogTail 持续读取单个日志文件的新增内容，并通过 checkpoint.Store 记录
|
||
// 已消费 offset，实现进程重启后的断点续传。采用轮询式 tail，避免对
|
||
// inotify 的额外依赖，跨平台且足够轻量。
|
||
type LogTail struct {
|
||
path string
|
||
store checkpoint.Store
|
||
poll time.Duration
|
||
onEntry func(model.LogRecord)
|
||
}
|
||
|
||
// NewLogTail 创建日志 tail 采集器；onEntry 每读到一行完整日志时回调。
|
||
func NewLogTail(path string, store checkpoint.Store, poll time.Duration, onEntry func(model.LogRecord)) *LogTail {
|
||
if poll <= 0 {
|
||
poll = time.Second
|
||
}
|
||
return &LogTail{path: path, store: store, poll: poll, onEntry: onEntry}
|
||
}
|
||
|
||
// Run 阻塞执行采集，直到 ctx 被取消。
|
||
func (t *LogTail) Run(ctx context.Context) {
|
||
offset, err := t.store.GetLogOffset(t.path)
|
||
if err != nil {
|
||
offset = 0
|
||
}
|
||
var partial []byte
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
default:
|
||
}
|
||
|
||
offset, partial = t.readOnce(offset, partial)
|
||
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(t.poll):
|
||
}
|
||
}
|
||
}
|
||
|
||
// readOnce 从 offset 处读取文件新增内容，返回新的 offset 与未成行的残留字节。
|
||
func (t *LogTail) readOnce(offset int64, partial []byte) (int64, []byte) {
|
||
f, err := os.Open(t.path)
|
||
if err != nil {
|
||
return offset, partial
|
||
}
|
||
defer f.Close()
|
||
|
||
info, err := f.Stat()
|
||
if err != nil {
|
||
return offset, partial
|
||
}
|
||
// 文件被截断或轮转：从头重新消费。
|
||
if info.Size() < offset {
|
||
offset = 0
|
||
partial = partial[:0]
|
||
}
|
||
|
||
if _, err := f.Seek(offset, io.SeekStart); err != nil {
|
||
return offset, partial
|
||
}
|
||
data, err := io.ReadAll(f)
|
||
if err != nil {
|
||
return offset, partial
|
||
}
|
||
if len(data) == 0 {
|
||
return offset, partial
|
||
}
|
||
|
||
partial = append(partial, data...)
|
||
for {
|
||
idx := bytes.IndexByte(partial, '\n')
|
||
if idx < 0 {
|
||
break
|
||
}
|
||
line := partial[:idx]
|
||
partial = partial[idx+1:]
|
||
offset += int64(idx) + 1
|
||
t.emit(line, offset)
|
||
}
|
||
return offset, partial
|
||
}
|
||
|
||
func (t *LogTail) emit(line []byte, offset int64) {
|
||
msg := strings.TrimRight(string(line), "\r")
|
||
rec := model.LogRecord{
|
||
Timestamp: time.Now().Unix(),
|
||
Level: detectLevel(msg),
|
||
Source: t.path,
|
||
Message: msg,
|
||
File: t.path,
|
||
Offset: offset,
|
||
}
|
||
|
||
// 先投递给上层（进入内存缓冲），再推进 offset；若进程在两步之间退出，
|
||
// 重启后会从旧 offset 重读，得到「至少一次」语义，避免丢日志。
|
||
if t.onEntry != nil {
|
||
t.onEntry(rec)
|
||
}
|
||
|
||
// 检查点写入失败不阻断采集；下轮会按旧 offset 重读（至少一次语义）。
|
||
if err := t.store.SetLogOffset(t.path, offset); err != nil {
|
||
// 可选：由上层记录日志，这里保持轻量。
|
||
}
|
||
}
|
||
|
||
// detectLevel 从日志内容中识别常见级别，未命中默认 INFO。
|
||
func detectLevel(msg string) string {
|
||
upper := strings.ToUpper(msg)
|
||
for _, lvl := range []string{"FATAL", "ERROR", "WARN", "INFO", "DEBUG"} {
|
||
if strings.Contains(upper, lvl) {
|
||
return lvl
|
||
}
|
||
}
|
||
return "INFO"
|
||
}
|