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) + } + } +}