feat(collectors): 主机采集器容器优先托管 + 采集器日志查看 + 推送地址纠正

现象 1. 主机采集器只有 systemd 一条路，跨发行版差异大（这轮已踩过 param 被删、日志路径、时间戳格式三处）；进程存活/开机自启也没有统一保证。 2. 「采集器日志」在主机模式下无法从平台查看，排「起没起来/在报什么错」必须 SSH。 3. 实测发现：Loki 部署把 listenIP 填成 0.0.0.0 时，部署记录 endpoint 是 http://0.0.0.0:3100，采集器照抄去推 -> connect: connection refused，而容器/systemd 都是 running、平台 status=active —— 日志静默丢失。根因 1. 托管方式写死在代码里（promtail 二进制 + systemd 单元），没有能力探测。 2. 没有对外暴露采集器自身日志的接口。 3. 渲染 promtail 配置时直接用 deployment.Endpoint，没有校验它是不是「可连接的推送地址」；另外容器用 bridge 网络，127.0.0.1 在容器里指向容器自身，同机 Loki 也会推失败。修法 - services/collector_host_runtime.go（新增） * detectHostCapabilities：一次 SSH 往返探测架构/init/systemd/docker可用/开机自启/SELinux/journald * chooseHostRunMode：docker 优先 -> systemd 兜底 -> 两者都没有则明确报错（不裸跑进程） * installHostDocker：grafana/promtail:3.4.0 容器，--restart=always --memory=256m --cpus=1 --log-opt max-size=10m,max-file=3，挂 /etc/promtail(:ro) /var/lib/promtail /var/log(:ro) (+journald 相关)，SELinux Enforcing 时 --security-opt label=disable；先 docker run --rm -check-syntax 校验配置，再替换正式配置 * waitContainerRunning：看 Status + RestartCount，起来就崩立即报错并带回 docker logs * hostPushEndpoint：0.0.0.0/:: 换真实主机 IP；容器模式下 127.0.0.1/localhost 也换；换不了明确报错 * CollectorLogs：主机 docker logs / journalctl，K8s DaemonSet Pod 日志、Sidecar fluent-bit 容器日志 - collector_deployer.go：installHost 探测->选模式->渲染配置->分发；systemd 路径从 docker 迁回时清理同名容器；卸载时容器与 systemd 两种方式都清（RunMode 对老记录是空的） - 拉不到镜像（内网无外网）时降级 systemd 二进制托管，其它错误原样抛出 - models/collector.go：新增 runMode 并纳入 Sanitized；services.Update 白名单同步 - 接口 GET /api/v1/collectors/:id/logs?tail=N；前端卡片加「运行方式」+「日志」弹窗 - build.sh：detect_go 的 INFO 走 stderr（原来被打进 GO_CMD，构建必失败）验证（172.19.16.8 / pre-cex-ops-jenkins，collector id=2） - 安装后 runMode=docker，容器 Up，RestartPolicy=always，systemd promtail 自动停用 - 推送地址自动纠正为 http://172.19.16.8:3100，Loki 近 10 分钟查到 4919 条 job=system/job=app、host=pre-cex-ops-jenkins - docker 垫片模拟「docker 不可用」-> 自动回退 runMode=systemd、unit active+enabled、日志接口 source=systemd；移除垫片重装 -> 切回 docker - 卸载：容器 0 个、unit 文件不存在、/etc/promtail 不存在、runMode 清空；重装恢复 running/0 重启 - K8s 采集器日志接口 source=kubernetes（顺带暴露 429 限流，见后续日志量治理） - go vet 干净；go test ./... 全绿（新增 chooseHostRunMode / hostPushEndpoint 等 6 组用例）； tsc -b 通过 - 镜像 lghost/lokistack:latest = :v0.1 = sha256:906a6d5369d7，Pod imageID 一致
2026-10-28 20:01:38 +00:00
1 parent c983d0e3c1
commit 7e3356e32b
@@ -0,0 +1,409 @@
+package services
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"strings"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"lokistack/models"
+)
+
+// 主机采集器的两种托管方式。
+//
+// 为什么要落库：安装/升级/卸载必须走**同一条**路径。不记录的话，卸载时不知道该
+// 删容器还是删 systemd 单元 → 留下孤儿进程继续采日志（重复数据 + 无法收敛）。
+const (
+	HostRunModeDocker  = "docker"
+	HostRunModeSystemd = "systemd"
+)
+
+// DefaultPromtailVersion 采集器容器镜像版本（与 DaemonSet 模板保持一致）。
+const DefaultPromtailVersion = "3.4.0"
+
+const (
+	promtailServiceName  = "promtail"
+	promtailConfigDir    = "/etc/promtail"
+	promtailConfigPath   = promtailConfigDir + "/promtail.yaml"
+	promtailPositionsDir = "/var/lib/promtail"
+)
+
+func promtailContainerName(id uint) string { return fmt.Sprintf("lokistack-promtail-%d", id) }
+
+// hostPushEndpoint 把「Loki 监听地址」纠正成「采集器真正能连的推送地址」。
+//
+// 两种必须纠正的情况（本会话实测踩到的）：
+//
+//	① 用户在主机部署 Loki 时把 listenIP 填成 0.0.0.0 / ::
+//	   → 部署记录里的 endpoint 变成 http://0.0.0.0:3100，采集器照抄过去推
+//	     → connect: connection refused，日志静默丢失（容器/serve 都是 running，最难查）
+//	② 容器模式下的回环地址
+//	   → 容器用 bridge 网络，127.0.0.1 指的是容器自己，不是宿主机
+//	     （systemd 模式不受影响，所以只在 docker 模式下改）
+//
+// 纠正不了（没有主机 IP 可替换）时明确报错，不返回一个必然失败的地址。
+func hostPushEndpoint(endpoint, hostIP string, dockerMode bool) (string, error) {
+	raw := strings.TrimSpace(endpoint)
+	if raw == "" {
+		return "", fmt.Errorf("缺少 Loki 推送地址（endpoint）")
+	}
+	u, err := url.Parse(raw)
+	if err != nil || u.Host == "" {
+		return raw, nil // 非标准形态（如 k8s svc 短名）不做处理
+	}
+	host := u.Hostname()
+	wildcard := host == "0.0.0.0" || host == "::" || host == "[::]"
+	loopback := host == "localhost" || host == "127.0.0.1" || host == "::1"
+	if !wildcard && !(dockerMode && loopback) {
+		return raw, nil
+	}
+	ip := strings.TrimSpace(hostIP)
+	if ip == "" || net.ParseIP(ip) == nil {
+		return "", fmt.Errorf("Loki 推送地址是 %s（%s），但取不到目标主机 IP 用于替换："+
+			"请把 Loki 部署的监听地址改成实际可访问的 IP 或域名", raw, map[bool]string{true: "通配地址无法作为推送目标", false: "容器内 127.0.0.1 指向容器自身"}[wildcard])
+	}
+	port := u.Port()
+	if port == "" {
+		port = "3100"
+	}
+	u.Host = net.JoinHostPort(ip, port)
+	return u.String(), nil
+}
+
+// errImagePullFailed 拉不到采集器镜像（目标机访问不了镜像仓库）。
+// 单独成哨兵错误：这种情况可以退回 systemd 二进制托管，而配置错误、容器起不来
+// 这类问题不该回退（回退只会把真实错误藏起来）。
+var errImagePullFailed = errors.New("拉取采集器镜像失败")
+
+// HostCapabilities 是目标主机的托管能力探测结果（一次 SSH 往返拿全）。
+type HostCapabilities struct {
+	Arch          string // x86_64 / aarch64 …
+	InitName      string // systemd / init / …
+	HasSystemd    bool   // /run/systemd/system 存在（systemd 正在跑，不是只装了命令）
+	HasDocker     bool   // 有 docker 命令
+	DockerUsable  bool   // docker info 成功（daemon 在跑 + 当前用户有权限）
+	DockerVersion string
+	DockerEnabled string // enabled / disabled / unknown
+	SELinux       string // Enforcing / Permissive / Disabled / unknown
+	HasJournald   bool   // 存在持久化或运行时 journal 目录
+}
+
+// DockerSummary 供操作日志展示（别在日志里输出空字符串）。
+func (c *HostCapabilities) DockerSummary() string {
+	if !c.HasDocker {
+		return "未安装"
+	}
+	if !c.DockerUsable {
+		if c.DockerVersion != "" {
+			return "已安装但不可用（daemon 未运行或无权限）"
+		}
+		return "已安装但不可用"
+	}
+	desc := c.DockerVersion
+	if desc == "" {
+		desc = "可用"
+	}
+	return desc + "（开机自启 " + c.DockerEnabled + "）"
+}
+
+// detectHostCapabilities 探测目标机能否用 docker / systemd 托管长驻进程。
+//
+// 一次合并命令的原因：每次 SSH 往返都有 100~300ms 开销，装机流程本来就要跑十几步，
+// 逐条探测会让「点部署后等很久」的体感更差。
+func detectHostCapabilities(ctx context.Context, client *SSHClient) *HostCapabilities {
+	caps := &HostCapabilities{InitName: "unknown", SELinux: "unknown", DockerEnabled: "unknown"}
+	script := `echo "ARCH=$(uname -m 2>/dev/null)"
+echo "INIT=$(ps -p 1 -o comm= 2>/dev/null | tr -d ' ')"
+[ -d /run/systemd/system ] && echo "SYSTEMD=1" || echo "SYSTEMD=0"
+if command -v docker >/dev/null 2>&1; then
+  echo "DOCKER=1"
+  echo "DOCKER_VERSION=$(docker version --format '{{.Server.Version}}' 2>/dev/null)"
+  docker info >/dev/null 2>&1 && echo "DOCKER_USABLE=1" || echo "DOCKER_USABLE=0"
+  echo "DOCKER_ENABLED=$(systemctl is-enabled docker 2>/dev/null || echo unknown)"
+else
+  echo "DOCKER=0"
+  echo "DOCKER_USABLE=0"
+fi
+echo "SELINUX=$(getenforce 2>/dev/null || echo unknown)"
+if [ -d /run/log/journal ] || [ -d /var/log/journal ]; then echo "JOURNALD=1"; else echo "JOURNALD=0"; fi`
+	out, _, err := client.Execute(ctx, script)
+	if err != nil && strings.TrimSpace(out) == "" {
+		return caps
+	}
+	for _, line := range strings.Split(out, "\n") {
+		line = strings.TrimSpace(line)
+		key, value, ok := strings.Cut(line, "=")
+		if !ok {
+			continue
+		}
+		switch key {
+		case "ARCH":
+			caps.Arch = value
+		case "INIT":
+			if value != "" {
+				caps.InitName = value
+			}
+		case "SYSTEMD":
+			caps.HasSystemd = value == "1"
+		case "DOCKER":
+			caps.HasDocker = value == "1"
+		case "DOCKER_VERSION":
+			caps.DockerVersion = value
+		case "DOCKER_USABLE":
+			caps.DockerUsable = value == "1"
+		case "DOCKER_ENABLED":
+			if value != "" {
+				caps.DockerEnabled = value
+			}
+		case "SELINUX":
+			if value != "" {
+				caps.SELinux = value
+			}
+		case "JOURNALD":
+			caps.HasJournald = value == "1"
+		}
+	}
+	return caps
+}
+
+// normalizeArch 把 uname -m 转成发行包/镜像使用的架构名。
+func normalizeArch(arch string) (string, error) {
+	switch strings.TrimSpace(arch) {
+	case "x86_64", "amd64":
+		return "amd64", nil
+	case "aarch64", "arm64":
+		return "arm64", nil
+	}
+	return "", fmt.Errorf("不支持的主机架构: %s", arch)
+}
+
+// chooseHostRunMode 决定托管方式：docker 优先 → systemd 兜底 → 都没有则明确报错。
+//
+// 抽成纯函数是为了能测：这条判断决定「卸载时要删容器还是删 unit」，
+// 判错的后果是孤儿进程继续采日志（重复数据）或卸载残留，靠人工回归太贵。
+func chooseHostRunMode(caps *HostCapabilities) (string, error) {
+	if caps == nil {
+		return "", fmt.Errorf("缺少目标机能力探测结果")
+	}
+	if caps.DockerUsable {
+		return HostRunModeDocker, nil
+	}
+	if caps.HasSystemd {
+		return HostRunModeSystemd, nil
+	}
+	return "", fmt.Errorf("目标机既没有可用的 Docker（docker info 失败）也没有 systemd，无法可靠托管采集进程：" +
+		"请先在目标机安装并启动 Docker 后重试。平台不会以裸进程方式运行 promtail（主机重启后采集会静默中断）")
+}
+
+func ensureHostPromtailDirs(ctx context.Context, client *SSHClient) error {
+	if _, _, err := client.Execute(ctx, "mkdir -p "+promtailConfigDir+" "+promtailPositionsDir); err != nil {
+		return fmt.Errorf("创建 Promtail 目录失败: %w", err)
+	}
+	return nil
+}
+
+// promoteHostConfig 把校验通过的 .new 覆盖为正式配置。
+func promoteHostConfig(ctx context.Context, client *SSHClient) error {
+	if _, _, err := client.Execute(ctx, "mv -f "+promtailConfigPath+".new "+promtailConfigPath); err != nil {
+		return fmt.Errorf("替换 Promtail 配置失败: %w", err)
+	}
+	return nil
+}
+
+// waitContainerRunning 轮询容器真实状态。
+//
+// `docker run -d` 返回 0 **不代表容器在跑**（镜像参数错会立刻退出）。
+// 这里同时看 Status 与 RestartCount：RestartCount 持续增长 = 崩溃循环，
+// 必须立即报错并带回 docker logs —— 等价于 systemd 侧 waitServiceActive 的防护。
+func waitContainerRunning(ctx context.Context, client *SSHClient, name string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	last := ""
+	for time.Now().Before(deadline) {
+		out, _, _ := client.Execute(ctx, fmt.Sprintf("docker inspect -f '{{.State.Status}}|{{.RestartCount}}' %s 2>/dev/null", name))
+		last = strings.TrimSpace(out)
+		status, restarts, ok := strings.Cut(last, "|")
+		if ok {
+			status = strings.TrimSpace(status)
+			restarts = strings.TrimSpace(restarts)
+			if status == "running" && restarts == "0" {
+				return nil
+			}
+			if restarts != "" && restarts != "0" {
+				return fmt.Errorf("容器 %s 处于重启循环（Status=%s, RestartCount=%s）", name, status, restarts)
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(2 * time.Second):
+		}
+	}
+	if last == "" {
+		last = "容器不存在"
+	}
+	return fmt.Errorf("容器 %s %s 内未进入 running 且无重启（当前 %s）", name, timeout, last)
+}
+
+// ===== 采集器日志（Web 上直接看，不用登机器） =====
+
+// CollectorLogs 返回采集器最近的运行日志。
+//
+// 主机模式按托管方式自动选入口：容器 → `docker logs`；systemd → journalctl / 文件。
+// K8s 模式读 DaemonSet/采集 Pod 的容器日志。
+func (s *CollectorService) CollectorLogs(id uint, tail int) (map[string]interface{}, error) {
+	c, err := s.Get(id)
+	if err != nil {
+		return nil, err
+	}
+	if tail <= 0 {
+		tail = 200
+	}
+	if tail > 2000 {
+		tail = 2000
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	switch c.Type {
+	case "host-promtail":
+		return hostCollectorLogs(ctx, c, tail)
+	case "k8s-daemonset", "k8s-fluentbit-sidecar":
+		return k8sCollectorLogs(ctx, c, tail)
+	}
+	return nil, fmt.Errorf("不支持的采集器类型: %s", c.Type)
+}
+
+func hostCollectorLogs(ctx context.Context, c *models.Collector, tail int) (map[string]interface{}, error) {
+	if c.Host == nil {
+		return nil, fmt.Errorf("该采集器未关联主机，无法读取日志")
+	}
+	client, err := NewSSHClientWithAuth(HostSSHAuth(c.Host))
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+
+	name := promtailContainerName(c.ID)
+	// ① 容器托管（首选）：docker logs 自带轮转，不需要额外清理
+	if _, _, err := client.Execute(ctx, "docker container inspect "+name+" >/dev/null 2>&1"); err == nil {
+		out, stderr, _ := client.Execute(ctx, fmt.Sprintf("docker logs --tail %d --timestamps %s 2>&1", tail, name))
+		return map[string]interface{}{
+			"source":      "docker",
+			"sourceLabel": "Docker 容器 " + name,
+			"content":     strings.TrimSpace(out + stderr),
+		}, nil
+	}
+
+	// ② systemd 托管（无 docker 的兜底路径，或历史安装）
+	if _, _, err := client.Execute(ctx, "[ -f /etc/systemd/system/"+promtailServiceName+".service ]"); err == nil {
+		out, _, _ := client.Execute(ctx, fmt.Sprintf("journalctl -u %s -n %d --no-pager 2>/dev/null", promtailServiceName, tail))
+		content := strings.TrimSpace(out)
+		if content == "" {
+			out, _, _ = client.Execute(ctx, fmt.Sprintf("tail -n %d /var/log/promtail.log 2>/dev/null", tail))
+			content = strings.TrimSpace(out)
+		}
+		return map[string]interface{}{
+			"source":      "systemd",
+			"sourceLabel": "systemd 服务 " + promtailServiceName,
+			"content":     content,
+		}, nil
+	}
+
+	return nil, fmt.Errorf("目标机 %s 上既没有采集器容器 %s，也没有 %s systemd 服务（可能尚未部署或已被手工清理）",
+		c.Host.HostIP, name, promtailServiceName)
+}
+
+func k8sCollectorLogs(ctx context.Context, c *models.Collector, tail int) (map[string]interface{}, error) {
+	if c.Cluster == nil {
+		return nil, fmt.Errorf("该采集器未关联集群，无法读取日志")
+	}
+	ns := c.Namespace
+	if ns == "" && c.LokiDeployment != nil {
+		ns = c.LokiDeployment.Namespace
+	}
+	if ns == "" {
+		ns = "log"
+	}
+	client, err := clusterClient(c.Cluster, 30*time.Second)
+	if err != nil {
+		return nil, err
+	}
+
+	// Sidecar 模式的日志在业务 Pod 的 fluent-bit 容器里，不是独立的 DaemonSet
+	if c.Type == "k8s-fluentbit-sidecar" {
+		return sidecarCollectorLogs(ctx, client, ns, c.TargetDeployment, tail)
+	}
+
+	pods, err := client.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: "app=promtail"})
+	if err != nil {
+		return nil, fmt.Errorf("读取采集器 Pod 失败（检查集群凭证是否具备 pods list/log 权限）: %w", err)
+	}
+	if len(pods.Items) == 0 {
+		return nil, fmt.Errorf("命名空间 %s 中没有找到 promtail Pod（采集器可能未部署成功）", ns)
+	}
+	return podContainerLogs(ctx, client, ns, runningPodName(pods.Items), "promtail", tail)
+}
+
+// runningPodName 优先挑 Running 的 Pod：DaemonSet 里常有刚重建的 Pending 实例，
+// 读它的日志只会得到一行「container is waiting to start」。
+func runningPodName(pods []corev1.Pod) string {
+	for _, p := range pods {
+		if p.Status.Phase == corev1.PodRunning {
+			return p.Name
+		}
+	}
+	if len(pods) > 0 {
+		return pods[0].Name
+	}
+	return ""
+}
+
+func podContainerLogs(ctx context.Context, client kubernetes.Interface, ns, pod, container string, tail int) (map[string]interface{}, error) {
+	if pod == "" {
+		return nil, fmt.Errorf("命名空间 %s 中没有可读取日志的采集 Pod", ns)
+	}
+	lines := int64(tail)
+	raw, err := client.CoreV1().Pods(ns).GetLogs(pod, &corev1.PodLogOptions{TailLines: &lines, Container: container}).DoRaw(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("读取 Pod %s/%s 的 %s 容器日志失败: %w", ns, pod, container, err)
+	}
+	return map[string]interface{}{
+		"source":      "kubernetes",
+		"sourceLabel": fmt.Sprintf("Pod %s/%s（容器 %s）", ns, pod, container),
+		"content":     strings.TrimSpace(string(raw)),
+	}, nil
+}
+
+// sidecarCollectorLogs 找业务 Deployment 的 Pod，读它里面 fluent-bit 容器的日志。
+func sidecarCollectorLogs(ctx context.Context, client kubernetes.Interface, ns, deployment string, tail int) (map[string]interface{}, error) {
+	if deployment == "" {
+		return nil, fmt.Errorf("该 Sidecar 采集器缺少目标 Deployment，无法定位日志")
+	}
+	pods, err := client.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("读取命名空间 %s 的 Pod 失败（检查集群凭证权限）: %w", ns, err)
+	}
+	matched := make([]corev1.Pod, 0, len(pods.Items))
+	for _, p := range pods.Items {
+		if !strings.HasPrefix(p.Name, deployment) {
+			continue
+		}
+		for _, ct := range p.Spec.Containers {
+			if ct.Name == "fluent-bit" {
+				matched = append(matched, p)
+				break
+			}
+		}
+	}
+	if len(matched) == 0 {
+		return nil, fmt.Errorf("命名空间 %s 中 %s 的 Pod 里没有 fluent-bit 容器（Sidecar 可能未注入成功）", ns, deployment)
+	}
+	return podContainerLogs(ctx, client, ns, runningPodName(matched), "fluent-bit", tail)
+}