fix(collectors): 容器模式首次安装日志全被 Loki 拒绝(entry too far behind)

=== 现象 ===
容器化采集器(docker 优先托管)上线后实测:10 分钟只进 47 条,
容器日志刷满:
    msg="final error sending batch" status=400
    error="entry too far behind, entry timestamp is: 2026-10-21T20:01:16Z,
           oldest acceptable timestamp is: 2026-10-28T19:..."

=== 根因 ===
Promtail 在 positions 里没有某文件记录时,会从**文件开头**开始读。
主机 /var/log/messages 有 7 天历史(约 1.35 亿字节),而 Loki 只接受最近
reject_old_samples_max_age(默认 1h)的条目。
关键危害:promtail 把新旧日志混在**同一个 batch** 里推送 → 整批 400 被拒,
**连新日志也发不出去**,不只是丢历史。

systemd 模式此前看着正常,只是因为跑了很久、positions 已建立;
全新主机(或容器换了数据目录、positions 为空)必然复现。

=== 修复 ===
安装时(容器与 systemd 两条路径都加)预置 positions 到文件末尾:
  - 只补 positions 里【还没有】的路径 → 重装/升级不破坏已有进度、不重复采集
  - 偏移用 stat -c %s 取当前文件大小
  - 路径来源:系统日志探测结果 ∪ logPath ∪ 每个 ScrapeJob 的 paths

新增 services/promtail_positions.go:
  seedPositionsCmd(patterns)          → 生成预置脚本
  collectorPositionPatterns(c, sys)   → 汇总该采集器涉及的所有路径
预置失败只 warn 不阻断启动(不影响老环境升级)。

=== 实测(172.19.16.8 / TencentOS 4)===
  修复前:final error 持续刷、400 持续、10 分钟 47 条
  修复后:final error 0 次、400 0 次、restarts=0
          positions 已预置(cron=71042092 / cloud-init=286805 …)
          日志正常入库 295 条/15 分钟(真实新增速率,不再回填历史)
          时间戳与源文件一致:20:27:01.314716+00:00 CROND[1445228]

测试:TestSeedPositionsCmd / TestCollectorPositionPatterns 通过

镜像:lghost/lokistack:latest + v0.1 = sha256:d5dd05dca465
This commit is contained in:
opsx committed 2026-10-28 20:27:48 +00:00
1 parent 7e3356e32b
commit f6af12a731
3 files changed
+143

No files matched your search

@@ -0,0 +1,46 @@
package services
import (
"strings"
"testing"
"lokistack/models"
)
// 预置 positions 的脚本必须:只补缺失路径、用文件大小做偏移、
// 已有记录不覆盖(重装不重复采集)。
func TestSeedPositionsCmd(t *testing.T) {
cmd := seedPositionsCmd([]string{"/var/log/messages", "/var/log/*.log", "/var/log/messages"})
for _, want := range []string{
"positions:", "/var/lib/promtail/positions.yaml",
"grep -qF", "stat -c %s", "/var/log/messages", "/var/log/*.log",
} {
if !strings.Contains(cmd, want) {
t.Fatalf("脚本缺少 %q:\n%s", want, cmd)
}
}
// 去重:同一路径只出现一次
if strings.Count(cmd, "'/var/log/messages'") != 1 {
t.Fatalf("路径未去重:\n%s", cmd)
}
if seedPositionsCmd(nil) != "" {
t.Fatal("无路径时应返回空串")
}
}
// 采集器涉及的路径要覆盖:系统日志 + logPath + 每个 job 的路径
func TestCollectorPositionPatterns(t *testing.T) {
c := &models.Collector{
LogPath: "/data/app/*.log",
ScrapeJobs: []models.ScrapeJob{
{Name: "a", Paths: []string{"/data/a/*.log"}},
{Name: "system", System: true}, // 无显式路径 → 用探测结果
},
}
got := strings.Join(collectorPositionPatterns(c, []string{"/var/log/messages"}), ",")
for _, want := range []string{"/var/log/messages", "/data/app/*.log", "/data/a/*.log"} {
if !strings.Contains(got, want) {
t.Fatalf("缺少路径 %s,实际:%s", want, got)
}
}
}