LokiStack/backend/services/collector_host_runtime.go

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