LokiStack/backend/services/collector_deployer.go

package services

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"strings"
	"time"

	"lokistack/models"
)

// CollectorDeployer 采集器部署器，按 type 分发到三种安装/卸载流程
type CollectorDeployer struct {
	collector *models.Collector
	loki      *models.LokiDeployment
	logFunc   func(level, message, detail string)
}

func NewCollectorDeployer(c *models.Collector, l *models.LokiDeployment, logFunc func(level, message, detail string)) *CollectorDeployer {
	return &CollectorDeployer{collector: c, loki: l, logFunc: logFunc}
}

func (d *CollectorDeployer) log(level, message, detail string) {
	if d.logFunc != nil {
		d.logFunc(level, message, detail)
	}
}

// Install 按采集器类型分发安装
func (d *CollectorDeployer) Install(ctx context.Context) error {
	endpoint := d.loki.Endpoint
	if endpoint == "" {
		return fmt.Errorf("关联的 Loki 尚未部署，缺少推送地址（endpoint），请先成功部署该 Loki")
	}
	d.log("info", fmt.Sprintf("开始部署采集器 %s（类型 %s）", d.collector.Name, d.collector.Type), "")

	var err error
	switch d.collector.Type {
	case "host-promtail":
		err = d.installHost(ctx, endpoint)
	case "k8s-daemonset":
		err = d.installDaemonset(ctx, endpoint)
	case "k8s-fluentbit-sidecar":
		err = d.installSidecar(ctx, endpoint)
	default:
		return fmt.Errorf("不支持的采集器类型: %s", d.collector.Type)
	}
	if err != nil {
		d.log("error", fmt.Sprintf("采集器部署失败: %v", err), "")
		return err
	}
	d.log("success", "采集器部署完成", "")
	return nil
}

// Uninstall 按采集器类型分发卸载
func (d *CollectorDeployer) Uninstall(ctx context.Context) error {
	d.log("info", fmt.Sprintf("开始卸载采集器 %s（类型 %s）", d.collector.Name, d.collector.Type), "")
	switch d.collector.Type {
	case "host-promtail":
		return d.uninstallHost(ctx)
	case "k8s-daemonset":
		return d.uninstallDaemonset(ctx)
	case "k8s-fluentbit-sidecar":
		return d.uninstallSidecar(ctx)
	}
	return nil
}

// ===== host-promtail：托管方式自动探测（docker 优先，systemd 兜底） =====

// installHost 主机采集器安装入口。
//
// 「进程活着」这件事不该由平台自己扛：平台的职责是「期望状态 + 配置下发 + 健康校验」，
// 存活/开机自启交给成熟的监督者。所以按目标机能力自动选托管方式：
//
//	① 有可用 Docker（docker info 成功）→ 容器（--restart=always，跨发行版抹平差异）
//	② 无 Docker 但有 systemd         → 二进制 + systemd 单元
//	③ 两者都没有                     → 明确报错，绝不裸跑进程
//
// 第 ③ 条是硬约束：`nohup promtail &` 起的进程主机重启即消失，平台不主动巡检就会
// 静默断采；要补齐就得自己实现存活检测/PID 管理/开机自启，等于重造一个更差的 supervisor。
func (d *CollectorDeployer) installHost(ctx context.Context, endpoint string) error {
	host := d.collector.Host
	if host == nil {
		return fmt.Errorf("主机采集器缺少目标主机")
	}
	client, err := NewSSHClientWithAuth(HostSSHAuth(host))
	if err != nil {
		return err
	}
	defer client.Close()

	caps := detectHostCapabilities(ctx, client)
	d.log("info", fmt.Sprintf("目标机探测：架构 %s / init %s / docker %s / SELinux %s",
		caps.Arch, caps.InitName, caps.DockerSummary(), caps.SELinux), "")

	if err := ensureHostPromtailDirs(ctx, client); err != nil {
		return err
	}

	// 先定托管方式，再渲染配置：容器模式下回环地址必须换掉（bridge 里 127.0.0.1 指容器自己）
	mode, err := chooseHostRunMode(caps)
	if err != nil {
		return err
	}

	// 配置生成与上传（写 .new，两种托管方式共用；校验通过才替换正式配置）
	conf, err := d.prepareHostConfig(ctx, client, host, endpoint, mode == HostRunModeDocker)
	if err != nil {
		return err
	}
	if mode == HostRunModeDocker {
		err := d.installHostDocker(ctx, client, conf, caps)
		if err == nil {
			return nil
		}
		// 目标机拉不到镜像（内网/无外网）时，systemd 二进制托管是可行的退路；
		// 其它错误（配置非法、容器崩溃）必须原样抛出，不能被回退掩盖。
		if errors.Is(err, errImagePullFailed) && caps.HasSystemd {
			d.log("warn", fmt.Sprintf("容器方式不可用，回退为 systemd 二进制托管：%v", err), "")
			return d.installHostSystemd(ctx, client, conf, caps)
		}
		return err
	}
	d.log("warn", "目标机没有可用的 Docker，回退为 systemd 二进制托管（跨发行版差异更大，建议先装 Docker）", "")
	return d.installHostSystemd(ctx, client, conf, caps)
}

// prepareHostConfig 探测目标机身份并渲染/上传 promtail.yaml.new。
//
// 三件事必须在安装期探测，不能写死：
//   - 真实主机名（promtail 不展开 ${HOSTNAME}，写死会在 Loki 里留下一堆字面量标签）
//   - 系统日志时间戳格式（TencentOS 是 ISO8601，Debian 是传统 syslog）
//   - 系统日志路径（Ubuntu 只有 syslog/auth.log，没有 messages）
func (d *CollectorDeployer) prepareHostConfig(ctx context.Context, client *SSHClient, host *models.Host, endpoint string, dockerMode bool) (string, error) {
	logPath := d.collector.LogPath
	if logPath == "" {
		logPath = "/var/log/*.log"
	}
	// 推送地址纠正：Loki 的 listenIP 填成 0.0.0.0 时 endpoint 是必然连不上的
	// http://0.0.0.0:3100，采集器会 running 但一条日志都推不上去。
	hostIP := ""
	if host != nil {
		hostIP = host.HostIP
	}
	pushEndpoint, err := hostPushEndpoint(endpoint, hostIP, dockerMode)
	if err != nil {
		return "", err
	}
	if pushEndpoint != endpoint {
		d.log("warn", fmt.Sprintf("Loki 推送地址不可用（%s），已改为 %s —— 建议把 Loki 部署的监听地址改成实际 IP/域名", endpoint, pushEndpoint), "")
	}
	hostname := detectHostname(ctx, client, host)
	syslogFormat := detectSyslogFormat(ctx, client)
	sysLogPaths := detectSystemLogPaths(ctx, client)
	d.log("info", fmt.Sprintf("目标机 %s，syslog 格式 %s，系统日志 %s",
		hostname, syslogFormat, strings.Join(sysLogPaths, ",")), "")

	conf, err := PromtailConfigForHost(HostPromtailOptions{
		Endpoint:       pushEndpoint,
		Cluster:        "host",
		LabelsJSON:     d.collector.Labels,
		Hostname:       hostname,
		SyslogFormat:   syslogFormat,
		SystemLogPaths: sysLogPaths,
		LogPath:        logPath,
		Jobs:           d.collector.ScrapeJobs,
	})
	if err != nil {
		return "", fmt.Errorf("生成 Promtail 配置失败（现有配置未改动）: %w", err)
	}
	d.collector.ResolvedHostname = hostname
	d.collector.SyslogFormat = syslogFormat
	d.collector.SystemLogPaths = sysLogPaths

	// 先写 .new：配置坏了会把正在跑的采集干掉，不能直接覆盖正式配置
	d.log("info", "上传 Promtail 配置", "")
	if err := uploadDockerFile(ctx, client, promtailConfigPath+".new", conf, "644"); err != nil {
		return "", fmt.Errorf("上传配置失败: %w", err)
	}
	return conf, nil
}

// installHostDocker 以容器方式托管采集器（首选路径）。
func (d *CollectorDeployer) installHostDocker(ctx context.Context, client *SSHClient, conf string, caps *HostCapabilities) error {
	name := promtailContainerName(d.collector.ID)
	image := ResolveImageFor("promtail", "grafana/promtail", DefaultPromtailVersion)

	d.log("info", "拉取采集器镜像 "+image, "")
	if _, stderr, err := client.Execute(ctx, "docker pull "+image); err != nil {
		return fmt.Errorf("%w（目标机需能访问镜像仓库，可先在平台侧配置镜像加速）: %v: %s",
			errImagePullFailed, err, truncateForMessage(strings.TrimSpace(stderr), 300))
	}

	// 用容器自己校验配置：3.x 的 -check-syntax 能挡住路径/正则/缩进错误
	d.log("info", "校验 Promtail 配置", "")
	checkCmd := fmt.Sprintf("docker run --rm -v %s:%s:ro %s -check-syntax -config.file=%s.new 2>&1",
		promtailConfigDir, promtailConfigDir, image, promtailConfigPath)
	out, _, err := client.Execute(ctx, checkCmd)
	if err != nil {
		msg := strings.TrimSpace(out)
		if strings.Contains(msg, "flag provided but not defined") {
			d.log("info", "当前 Promtail 版本不支持 -check-syntax，跳过语法校验", "")
		} else {
			return fmt.Errorf("Promtail 配置校验未通过（现有采集未改动）：%s", truncateForMessage(msg, 600))
		}
	}
	if err := promoteHostConfig(ctx, client); err != nil {
		return err
	}

	// 从 systemd 迁到容器：不先停旧服务就会两边同时采同一批文件（日志重复一倍）
	if out, _, err := client.Execute(ctx, "systemctl is-active "+promtailServiceName+" 2>/dev/null"); err == nil && strings.TrimSpace(out) == "active" {
		client.Execute(ctx, "systemctl stop "+promtailServiceName+" 2>/dev/null; systemctl disable "+promtailServiceName+" 2>/dev/null")
		d.log("info", "已停止旧的 systemd promtail 服务（改为容器托管，避免重复采集）", "")
	}

	client.Execute(ctx, "docker rm -f "+name+" >/dev/null 2>&1 || true")

	mounts := []string{
		"-v " + promtailConfigDir + ":" + promtailConfigDir + ":ro",
		"-v " + promtailPositionsDir + ":" + promtailPositionsDir,
		"-v /var/log:/var/log:ro",
	}
	if caps.HasJournald {
		// 采 systemd journal 时需要（Ubuntu 最小安装无 rsyslog 的场景）
		mounts = append(mounts,
			"-v /run/log/journal:/run/log/journal:ro",
			"-v /etc/machine-id:/etc/machine-id:ro")
	}
	// SELinux Enforcing 时 bind mount 会被拒绝（容器里读不到 /var/log）；
	// 采集器容器不承载多租户隔离，直接关标签检查，比给宿主机目录重新打标签安全。
	secOpt := ""
	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",
		name, secOpt, strings.Join(mounts, " "), image, promtailConfigPath)
	d.log("info", "启动采集器容器 "+name, "")
	if _, stderr, err := client.Execute(ctx, runCmd); err != nil {
		return fmt.Errorf("启动采集器容器失败: %v: %s", err, truncateForMessage(strings.TrimSpace(stderr), 400))
	}

	// docker run 返回 0 也可能是「起来就崩」——必须查容器状态与重启次数
	if err := waitContainerRunning(ctx, client, name, 30*time.Second); err != nil {
		logs, _, _ := client.Execute(ctx, fmt.Sprintf("docker logs --tail 30 %s 2>&1", name))
		return fmt.Errorf("采集器容器未正常运行：%v\n最近日志：\n%s", err, truncateForMessage(strings.TrimSpace(logs), 900))
	}

	d.collector.RunMode = HostRunModeDocker
	d.collector.Config = conf
	d.log("success", fmt.Sprintf("Promtail 已以容器方式运行（%s，--restart=always）", name), "")
	return nil
}

// installHostSystemd 无 Docker 环境下的兜底路径：二进制 + systemd 单元。
func (d *CollectorDeployer) installHostSystemd(ctx context.Context, client *SSHClient, conf string, caps *HostCapabilities) error {
	arch, err := normalizeArch(caps.Arch)
	if err != nil {
		return err
	}

	// 确定 Promtail 版本（优先使用关联的 Loki 版本）
	version := "3.0.0"
	if d.loki != nil && d.loki.Version != "" {
		version = d.loki.Version
	}
	d.log("info", fmt.Sprintf("下载 Promtail %s（%s）", version, arch), "")
	url := fmt.Sprintf("https://github.com/grafana/loki/releases/download/v%s/promtail-linux-%s.zip", version, arch)
	dl := fmt.Sprintf("cd /tmp && curl -fsSL -o p.zip '%s' && unzip -o p.zip -d /opt/promtail/ && chmod +x /opt/promtail/promtail-linux-%s", url, arch)
	if _, _, err := client.Execute(ctx, dl); err != nil {
		return fmt.Errorf("下载 Promtail 失败: %w", err)
	}
	client.Execute(ctx, "ln -sf /opt/promtail/promtail-linux-* /usr/local/bin/promtail 2>/dev/null || true")

	// 语法校验通过才替换正式配置
	d.log("info", "校验 Promtail 配置", "")
	if out, _, err := client.Execute(ctx, "/usr/local/bin/promtail -check-syntax -config.file="+promtailConfigPath+".new 2>&1"); err != nil {
		msg := strings.TrimSpace(out)
		// 老版本 promtail 可能没有 -check-syntax：只有明确报配置错误才阻断
		if strings.Contains(msg, "flag provided but not defined") {
			d.log("info", "当前 promtail 不支持 -check-syntax，跳过语法校验", "")
		} else {
			return fmt.Errorf("Promtail 配置校验未通过（现有配置未改动）：%s", truncateForMessage(msg, 600))
		}
	}
	if err := promoteHostConfig(ctx, client); err != nil {
		return err
	}

	// systemd 服务（生产级）
	service := `[Unit]
Description=Promtail Log Collector
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
User=root
Group=root
ExecStart=/usr/local/bin/promtail -config.file=` + promtailConfigPath + `
ExecReload=/bin/kill -HUP $MAINPID
Restart=always
RestartSec=10
LimitNOFILE=65536
StandardOutput=append:/var/log/promtail.log
StandardError=append:/var/log/promtail.log

[Install]
WantedBy=multi-user.target
`
	d.log("info", "安装 systemd 服务", "")
	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)
	}

	// 必须校验服务真的起来了：
	// 原来只判断 systemctl 命令有没有执行错误，结果 Promtail 因配置/参数错误
	// 崩溃重启 40+ 次，平台仍然显示「采集器已部署」，用户以为采集是好的。
	if err := waitServiceActive(ctx, client, promtailServiceName, 20*time.Second); err != nil {
		out, _, _ := client.Execute(ctx, "journalctl -u "+promtailServiceName+" -n 15 --no-pager 2>/dev/null | tail -15")
		tail := strings.TrimSpace(out)
		if tail == "" {
			out, _, _ = client.Execute(ctx, "tail -10 /var/log/promtail.log 2>/dev/null")
			tail = strings.TrimSpace(out)
		}
		return fmt.Errorf("Promtail 启动后未处于运行状态：%v\n最近日志：\n%s", err, truncateForMessage(tail, 900))
	}

	// 反向清理：同名容器若存在（从 docker 模式切回来），停掉避免重复采集
	client.Execute(ctx, "docker rm -f "+promtailContainerName(d.collector.ID)+" >/dev/null 2>&1 || true")

	// 保存配置快照
	d.collector.RunMode = HostRunModeSystemd
	d.collector.Config = conf
	d.log("success", "Promtail 已以 systemd 服务方式运行", "")
	return nil
}

// waitServiceActive 轮询 systemctl is-active，直到 active 或超时。
// Promtail 起不来时 systemctl restart 依然返回 0（只是排队拉起），
// 所以必须自己查状态，否则会把"崩溃重启中"误报成部署成功。
func waitServiceActive(ctx context.Context, client *SSHClient, service string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	var last string
	for time.Now().Before(deadline) {
		out, _, err := client.Execute(ctx, fmt.Sprintf("systemctl is-active %s 2>&1", service))
		last = strings.TrimSpace(out)
		if err == nil && last == "active" {
			return nil
		}
		time.Sleep(2 * time.Second)
	}
	if last == "" {
		last = "unknown"
	}
	return fmt.Errorf("%s 状态=%s", service, last)
}

func truncateForMessage(s string, max int) string {
	if len(s) <= max {
		return s
	}
	return s[:max] + "…"
}

// detectHostname 取目标机真实主机名（Bug3 修复点）。
// promtail 不做环境变量展开，把 ${HOSTNAME} 写进配置只会在 Loki 里留下一堆
// 字面量标签，无法按主机过滤。优先 `hostname -s`，失败退回平台里登记的主机名/IP。
func detectHostname(ctx context.Context, client *SSHClient, host *models.Host) string {
	if out, _, err := client.Execute(ctx, "hostname -s 2>/dev/null || hostname 2>/dev/null"); err == nil {
		if v := strings.TrimSpace(strings.SplitN(out, "\n", 2)[0]); v != "" && !strings.Contains(v, "${") {
			return v
		}
	}
	if host != nil {
		if v := strings.TrimSpace(host.Name); v != "" {
			return v
		}
		if v := strings.TrimSpace(host.HostIP); v != "" {
			return v
		}
	}
	return "unknown-host"
}

// detectSyslogFormat 探测目标机 /var/log 的时间戳格式（Bug4 修复点）。
//
//	TencentOS / RHEL 系（rsyslog 新版）: 2026-07-17T15:38:24.923784+08:00 → iso8601
//	Debian / 老 rsyslog:              Oct 28 18:11:06                  → syslog
//
// 探测不到（文件不存在/为空）时按传统 syslog 处理（保守默认）。
func detectSyslogFormat(ctx context.Context, client *SSHClient) string {
	const probe = `f=""; for p in /var/log/messages /var/log/syslog /var/log/secure; do [ -s "$p" ] && f="$p" && break; done; [ -n "$f" ] && head -n 1 "$f"`
	out, _, err := client.Execute(ctx, probe)
	if err != nil || strings.TrimSpace(out) == "" {
		return SyslogFormatClassic
	}
	line := strings.TrimSpace(strings.SplitN(out, "\n", 2)[0])
	if len(line) >= 19 && line[4] == '-' && line[7] == '-' && line[10] == 'T' {
		return SyslogFormatISO8601
	}
	return SyslogFormatClassic
}

func (d *CollectorDeployer) uninstallHost(ctx context.Context) error {
	host := d.collector.Host
	if host == nil {
		return nil
	}
	client, err := NewSSHClientWithAuth(HostSSHAuth(host))
	if err != nil {
		return err
	}
	defer client.Close()

	name := promtailContainerName(d.collector.ID)
	// 两种托管方式都清一遍：RunMode 对老记录是空的，且用户可能手工装过另一种。
	// 顺序固定：先停容器/服务，再删文件 —— 否则进程还在跑就没了配置。
	if out, _, err := client.Execute(ctx, "docker rm -f "+name+" 2>&1"); err == nil {
		d.log("info", "已删除采集器容器 "+strings.TrimSpace(out), "")
	}
	client.Execute(ctx, "systemctl stop "+promtailServiceName+" 2>/dev/null; systemctl disable "+promtailServiceName+" 2>/dev/null")
	client.Execute(ctx, "rm -f /etc/systemd/system/"+promtailServiceName+".service")
	client.Execute(ctx, "systemctl daemon-reload 2>/dev/null || true")
	client.Execute(ctx, "rm -rf "+promtailConfigDir+" /opt/promtail "+promtailPositionsDir+" /usr/local/bin/promtail")
	d.log("info", "已清理采集器配置、positions 与二进制", "")
	d.collector.RunMode = ""
	return nil
}

// ===== k8s-daemonset：以原生清单安装 Promtail DaemonSet =====

func (d *CollectorDeployer) installDaemonset(ctx context.Context, endpoint string) error {
	cluster := d.collector.Cluster
	if cluster == nil {
		return fmt.Errorf("集群采集器缺少目标集群")
	}

	// 默认部署到关联 Loki 的命名空间(统一 log 命名空间策略)
	ns := d.collector.Namespace
	if ns == "" && d.loki != nil {
		ns = d.loki.Namespace
	}
	if ns == "" {
		ns = "log"
	}

	clusterName := "unknown"
	if cluster.Name != "" {
		clusterName = cluster.Name
	}

	manifest := PromtailDaemonsetYAML(ns, endpoint, clusterName, d.collector.Labels,
		d.collector.TargetNamespaces, d.collector.AppLabelSource, d.collector.AppInclude)

	// 关键顺序：先本地校验清单，再删旧资源。
	// 否则模板有误时会出现「旧 DaemonSet 已删、新的建不出来」→ 日志采集直接中断。
	if err := validateManifest(manifest); err != nil {
		d.log("error", fmt.Sprintf("生成的 Promtail 清单无效，已中止且未改动现有资源: %v", err), "")
		return fmt.Errorf("Promtail 清单校验失败（现有采集器未受影响）: %w", err)
	}

	// 删除历史残留并通过 Kubernetes API 提交资源。
	for _, kind := range []string{"daemonset", "configmap", "serviceaccount"} {
		_ = deleteK8sResource(ctx, cluster, kind, ns, "promtail")
	}
	if err := d.applyYAML(ctx, nil, manifest, "promtail"); err != nil {
		return err
	}
	d.log("info", fmt.Sprintf("等待 DaemonSet promtail 就绪(命名空间 %s)...", ns), "")
	if err := waitDaemonSet(ctx, cluster, ns, "promtail"); err != nil {
		return fmt.Errorf("Promtail DaemonSet 未就绪: %w", err)
	}
	svcURL := endpoint

	d.collector.Config = svcURL
	d.log("success", fmt.Sprintf("Promtail DaemonSet 已部署到 %s(推送至 %s)", ns, svcURL), "")
	return nil
}

// applyYAML 通过 API Server 应用采集器清单。
func (d *CollectorDeployer) applyYAML(ctx context.Context, _ []string, manifest, label string) error {
	if d.collector.Cluster == nil {
		return fmt.Errorf("采集器缺少集群")
	}
	if err := applyManifest(ctx, d.collector.Cluster, manifest); err != nil {
		d.log("error", fmt.Sprintf("应用 %s 失败: %v", label, err), "")
		return err
	}
	return nil
}

func (d *CollectorDeployer) uninstallDaemonset(ctx context.Context) error {
	cluster := d.collector.Cluster
	if cluster == nil {
		return nil
	}

	ns := d.collector.Namespace
	if ns == "" && d.loki != nil {
		ns = d.loki.Namespace
	}
	if ns == "" {
		ns = "log"
	}
	// clusterrole / clusterrolebinding 用 loki-promtail；
	// 数组末两项是历史遗留命名，一并清理避免升级后留下孤儿资源。
	for _, item := range [][2]string{
		{"daemonset", "promtail"}, {"configmap", "promtail"}, {"serviceaccount", "promtail"},
		{"clusterrole", "loki-promtail"}, {"clusterrolebinding", "loki-promtail"},
		{"clusterrole", "opsxloki-promtail"}, {"clusterrolebinding", "opsxloki-promtail"},
	} {
		if err := deleteK8sResource(ctx, cluster, item[0], ns, item[1]); err != nil {
			return err
		}
	}
	return nil
}

// ===== k8s-fluentbit-sidecar：kubectl patch 注入 Fluent Bit sidecar =====

func (d *CollectorDeployer) installSidecar(ctx context.Context, endpoint string) error {
	cluster := d.collector.Cluster
	if cluster == nil {
		return fmt.Errorf("Sidecar 采集器缺少目标集群")
	}
	target := d.collector.TargetDeployment
	if target == "" {
		return fmt.Errorf("Sidecar 采集器缺少目标 Deployment")
	}
	client, err := clusterClient(cluster, 30*time.Second)
	if err != nil {
		return err
	}

	ns := d.collector.Namespace
	if ns == "" {
		ns = "default"
	}
	logPath := d.collector.LogPath
	if logPath == "" {
		logPath = "/var/log/app/*.log"
	}
	app := target

	// 1. 检查目标 deployment 是否存在
	d.log("info", fmt.Sprintf("检查目标 Deployment %s/%s", ns, target), "")
	deployment, err := client.AppsV1().Deployments(ns).Get(ctx, target, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("读取目标 Deployment %s/%s: %w", ns, target, err)
	}
	deploymentJSON, err := json.Marshal(deployment)
	if err != nil {
		return err
	}
	mount, err := resolveSidecarLogMount(deploymentJSON, target, logPath, d.collector.ID)
	if err != nil {
		return err
	}
	hasSidecar := false
	for _, container := range deployment.Spec.Template.Spec.Containers {
		if container.Name == "fluent-bit" {
			hasSidecar = true
			break
		}
	}

	// 3. 部署 ConfigMap
	lokiHost, lokiPort := parseLokiURL(endpoint)
	conf := FluentbitConfig(lokiHost, lokiPort, logPath, app, ns)
	configMapName := fmt.Sprintf("fluent-bit-config-%d", d.collector.ID)
	cm := FluentbitConfigMapYAML(conf, ns, configMapName)
	d.log("info", "部署 Fluent Bit ConfigMap", "")
	if err := applyManifest(ctx, cluster, cm); err != nil {
		return fmt.Errorf("ConfigMap 部署失败: %w", err)
	}
	d.log("info", fmt.Sprintf("为 %s 配置 Fluent Bit Sidecar", target), "")
	if err := patchDeployment(ctx, cluster, ns, target, SidecarPatchYAML(configMapName, mount)); err != nil {
		return fmt.Errorf("Sidecar 注入失败: %w", err)
	}
	if hasSidecar {
		if err := restartDeployment(ctx, cluster, ns, target); err != nil {
			return fmt.Errorf("Sidecar 配置已更新，但重启 Deployment 失败: %w", err)
		}
	}
	d.log("info", "等待 Pod 重建", "")
	if err := waitDeployment(ctx, cluster, ns, target); err != nil {
		return fmt.Errorf("Sidecar Deployment 未就绪: %w", err)
	}

	// DNS 自愈:CoreDNS 异常集群上 svc 域名不可解析 → 用 Loki ClusterIP 重建 ConfigMap 并滚动重启
	if strings.Contains(endpoint, ".svc") {
		host := strings.Split(strings.Split(endpoint, "//")[1], ":")[0]
		if podDNSBroken(ctx, cluster, ns, target, "fluent-bit", host) {
			d.log("warn", "fluent-bit 无法解析 Loki 服务名，回退 ClusterIP…", "")
			if ip, ierr := serviceClusterIP(ctx, cluster, lokiNamespaceOf(d.loki, endpoint), "loki"); ierr == nil && ip != "" {
				conf2 := FluentbitConfig(ip, lokiPortOf(endpoint), logPath, target, ns)
				cm2 := FluentbitConfigMapYAML(conf2, ns, configMapName)
				if err := applyManifest(ctx, cluster, cm2); err == nil {
					_ = restartDeployment(ctx, cluster, ns, target)
					d.log("info", "fluent-bit 推送地址已回退为 ClusterIP: "+ip, "")
				}
			}
		}
	}

	return nil
}

func (d *CollectorDeployer) uninstallSidecar(ctx context.Context) error {
	cluster := d.collector.Cluster
	if cluster == nil || d.collector.TargetDeployment == "" {
		return nil
	}
	client, err := clusterClient(cluster, 30*time.Second)
	if err != nil {
		return err
	}

	ns := d.collector.Namespace
	if ns == "" {
		ns = "default"
	}
	deployment, err := client.AppsV1().Deployments(ns).Get(ctx, d.collector.TargetDeployment, metav1.GetOptions{})
	if err != nil {
		return err
	}
	deploymentJSON, err := json.Marshal(deployment)
	if err != nil {
		return err
	}
	removePatch, err := sidecarRemovePatchYAML(deploymentJSON, d.collector.ID)
	if err != nil {
		return err
	}
	if err := patchDeployment(ctx, cluster, ns, d.collector.TargetDeployment, removePatch); err != nil {
		return fmt.Errorf("移除 Sidecar 失败: %w", err)
	}
	if err := deleteK8sResource(ctx, cluster, "configmap", ns, fmt.Sprintf("fluent-bit-config-%d", d.collector.ID)); err != nil {
		return err
	}

	return nil
}

// lokiPortOf 从 endpoint 提取端口(默认 3100)
func lokiPortOf(endpoint string) string {
	rest := strings.SplitN(endpoint, "//", 2)
	if len(rest) != 2 {
		return "3100"
	}
	hostPart := rest[1]
	if i := strings.Index(hostPart, ":"); i >= 0 {
		port := hostPart[i+1:]
		if j := strings.IndexAny(port, "/"); j >= 0 {
			port = port[:j]
		}
		if port != "" {
			return port
		}
	}
	return "3100"
}