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