From f6af12a731db210938274218b70ca58182b2c9e7 Mon Sep 17 00:00:00 2001
From: opsx <opsx@local>
Date: Wed, 28 Oct 2026 20:27:48 +0000
Subject: [PATCH] =?UTF-8?q?fix(collectors):=20=E5=AE=B9=E5=99=A8=E6=A8=A1?=
 =?UTF-8?q?=E5=BC=8F=E9=A6=96=E6=AC=A1=E5=AE=89=E8=A3=85=E6=97=A5=E5=BF=97?=
 =?UTF-8?q?=E5=85=A8=E8=A2=AB=20Loki=20=E6=8B=92=E7=BB=9D=EF=BC=88entry=20?=
 =?UTF-8?q?too=20far=20behind=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

=== 现象 ===
容器化采集器（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
---
 backend/services/collector_deployer.go      | 20 ++++++
 backend/services/promtail_positions.go      | 77 +++++++++++++++++++++
 backend/services/promtail_positions_test.go | 46 ++++++++++++
 3 files changed, 143 insertions(+)
 create mode 100644 backend/services/promtail_positions.go
 create mode 100644 backend/services/promtail_positions_test.go

diff --git a/backend/services/collector_deployer.go b/backend/services/collector_deployer.go
index 32b4bb5..7486285 100644
--- a/backend/services/collector_deployer.go
+++ b/backend/services/collector_deployer.go
@@ -238,6 +238,17 @@ func (d *CollectorDeployer) installHostDocker(ctx context.Context, client *SSHCl
 	if strings.EqualFold(caps.SELinux, "Enforcing") {
 		secOpt = " --security-opt label=disable"
 	}
+	// 启动前预置 positions 到文件末尾，否则 promtail 会从文件头回填历史，
+	// 超过 Loki reject_old_samples_max_age 的条目让整批 400，新日志也发不出去。
+	if cmd := seedPositionsCmd(collectorPositionPatterns(d.collector, d.collector.SystemLogPaths)); cmd != "" {
+		d.log("info", "预置采集进度（跳过历史日志，避免 entry too far behind）", "")
+		if out, stderr, err := client.Execute(ctx, cmd); err != nil {
+			d.log("warn", fmt.Sprintf("预置 positions 失败（不影响启动，但首次采集可能因历史日志被拒）：%v %s", err, strings.TrimSpace(stderr)), "")
+		} else if t := strings.TrimSpace(out); t != "" {
+			d.log("info", t, "")
+		}
+	}
+
 	runCmd := fmt.Sprintf(
 		"docker run -d --name %s --restart=always --memory=256m --cpus=1 "+
 			"--log-opt max-size=10m --log-opt max-file=3%s %s %s -config.file=%s",
@@ -319,6 +330,15 @@ WantedBy=multi-user.target
 	if err := uploadDockerFile(ctx, client, "/etc/systemd/system/"+promtailServiceName+".service", service, "644"); err != nil {
 		return fmt.Errorf("创建 systemd 服务失败: %w", err)
 	}
+	// 与容器模式同因：positions 为空会从文件头回填历史 → 整批 400
+	if cmd := seedPositionsCmd(collectorPositionPatterns(d.collector, d.collector.SystemLogPaths)); cmd != "" {
+		if out, stderr, err := client.Execute(ctx, cmd); err != nil {
+			d.log("warn", fmt.Sprintf("预置 positions 失败（不影响启动）：%v %s", err, strings.TrimSpace(stderr)), "")
+		} else if t := strings.TrimSpace(out); t != "" {
+			d.log("info", t, "")
+		}
+	}
+
 	if _, _, err := client.Execute(ctx, "systemctl daemon-reload && systemctl enable "+promtailServiceName+" && systemctl restart "+promtailServiceName); err != nil {
 		return fmt.Errorf("启动 Promtail 失败: %w", err)
 	}
diff --git a/backend/services/promtail_positions.go b/backend/services/promtail_positions.go
new file mode 100644
index 0000000..2ed7c1f
--- /dev/null
+++ b/backend/services/promtail_positions.go
@@ -0,0 +1,77 @@
+package services
+
+import (
+	"fmt"
+	"strings"
+
+	"lokistack/models"
+)
+
+// seedPositionsCmd 生成「把 positions 预置到文件末尾」的远端命令。
+//
+// 为什么必须做：
+//
+//	Promtail 在 positions 里没有某个文件记录时，会从**文件开头**开始读。
+//	主机上的 /var/log/messages 往往有几周历史，而 Loki 只接受最近
+//	`reject_old_samples_max_age`（默认 1h）的条目 —— 一旦 promtail 把新旧日志
+//	混在一个 batch 里推送，整批会被 400 拒绝（entry too far behind），
+//	**连新日志也发不出去**，容器日志里只会看到无尽的
+//	`msg="final error sending batch" status=400`。
+//
+//	实测：容器模式首次安装后 10 分钟只进 47 条，其余全是这个错误。
+//
+// 只补 positions 里**还没有**的路径，因此重装/升级不会破坏已有采集进度，
+// 也不会重复采集。
+func seedPositionsCmd(patterns []string) string {
+	// 去重，保持顺序
+	seen := map[string]bool{}
+	var uniq []string
+	for _, p := range patterns {
+		p = strings.TrimSpace(p)
+		if p == "" || seen[p] {
+			continue
+		}
+		seen[p] = true
+		uniq = append(uniq, p)
+	}
+	if len(uniq) == 0 {
+		return ""
+	}
+
+	quoted := make([]string, 0, len(uniq))
+	for _, p := range uniq {
+		quoted = append(quoted, "'"+strings.ReplaceAll(p, "'", `'\''`)+"'")
+	}
+
+	return fmt.Sprintf(`mkdir -p /var/lib/promtail
+POS=/var/lib/promtail/positions.yaml
+if [ ! -s "$POS" ]; then
+  printf 'positions:\n' > "$POS"
+fi
+for f in %s; do
+  [ -f "$f" ] || continue
+  if grep -qF "  $f:" "$POS" 2>/dev/null; then continue; fi
+  printf '  %%s: "%%s"\n' "$f" "$(stat -c %%s "$f")" >> "$POS"
+done
+echo "positions 已预置: $(grep -c ':' "$POS" 2>/dev/null || echo 0) 条记录"`,
+		strings.Join(quoted, " "))
+}
+
+// collectorPositionPatterns 收集该采集器涉及的所有路径（含通配符），用于预置 positions。
+func collectorPositionPatterns(c *models.Collector, systemPaths []string) []string {
+	var out []string
+	out = append(out, systemPaths...)
+
+	if p := strings.TrimSpace(c.LogPath); p != "" {
+		out = append(out, splitPaths(p)...)
+	}
+	for _, job := range c.ScrapeJobs {
+		paths := job.Paths
+		// 内置系统日志项不写死路径时，用探测结果
+		if job.System && len(paths) == 0 {
+			paths = systemPaths
+		}
+		out = append(out, paths...)
+	}
+	return out
+}
diff --git a/backend/services/promtail_positions_test.go b/backend/services/promtail_positions_test.go
new file mode 100644
index 0000000..08b9db3
--- /dev/null
+++ b/backend/services/promtail_positions_test.go
@@ -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)
+		}
+	}
+}