LokiStack/backend/services/loki_deployer.go

package services

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"lokistack/models"
	"lokistack/pkg/crypto"
)

// LokiDeployer 负责 Loki 的 Kubernetes、主机 Docker 及历史部署模式。
type LokiDeployer struct {
	deployment *models.LokiDeployment
	logFunc    func(level, message, detail string)
}

func NewLokiDeployer(deployment *models.LokiDeployment, logFunc func(level, message, detail string)) *LokiDeployer {
	return &LokiDeployer{deployment: deployment, logFunc: logFunc}
}

// Deploy 根据部署模式分发到对应实现
func (d *LokiDeployer) Deploy(ctx context.Context) error {
	if sc := d.deployment.StorageConfig; sc != nil {
		for _, credential := range []string{sc.AccessKey, sc.SecretKey} {
			if strings.HasPrefix(credential, "ENC:") {
				return fmt.Errorf("存储凭证解密失败，请检查 ENCRYPTION_KEY 并重新保存存储配置")
			}
		}
	}
	d.log("info", fmt.Sprintf("Starting Loki deployment: %s (mode: %s, version: %s)",
		d.deployment.Name, d.deployment.Mode, d.effectiveVersion()), "")

	// 版本兼容性检查
	if err := d.checkVersionCompatibility(); err != nil {
		d.log("warn", err.Error(), "")
	}

	var err error
	switch d.deployment.Mode {
	case "k8s-yaml":
		err = d.deployYaml(ctx)
	case "host-binary":
		err = d.deployBinary(ctx)
	case "host-docker":
		err = d.deployDocker(ctx)
	default:
		return fmt.Errorf("unsupported deployment mode: %s", d.deployment.Mode)
	}
	if err != nil {
		return err
	}

	return nil
}

// ==================== K8s YAML 部署 (StatefulSet) ====================

func (d *LokiDeployer) deployYaml(ctx context.Context) error {
	if d.deployment.Cluster == nil {
		return fmt.Errorf("please select target cluster")
	}
	ns := d.effectiveNamespace()
	if err := ensureNamespace(ctx, d.deployment.Cluster, ns); err != nil {
		return fmt.Errorf("创建命名空间 %s: %w", ns, err)
	}
	d.log("info", "Applying Loki manifest (StatefulSet + PVC + ConfigMap + Service)...", "")
	if err := applyManifest(ctx, d.deployment.Cluster, d.generateManifest(ns)); err != nil {
		return err
	}
	d.log("info", "Waiting for StatefulSet to be ready...", "")
	if err := waitStatefulSet(ctx, d.deployment.Cluster, ns, "loki"); err != nil {
		return fmt.Errorf("Loki StatefulSet 未就绪: %w", err)
	}
	// Ingress access log belongs to the target cluster, not the Loki namespace.
	// Its absence or lack of permission must not invalidate a healthy Loki deployment.
	if _, err := ApplyIngressAccess(ctx, d.deployment.Cluster); err != nil {
		d.log("warn", "未能自动配置 ingress-nginx 访问日志", err.Error())
	} else {
		d.log("info", "已配置 ingress-nginx JSON 访问日志，包含请求 Host", "")
	}
	d.log("success", "Loki deployed via Kubernetes API", "")
	return nil
}

// ==================== 主机二进制部署 ====================

func (d *LokiDeployer) deployBinary(ctx context.Context) error {
	if d.deployment.Host == nil {
		return fmt.Errorf("please select target host")
	}
	host := d.deployment.Host
	client, err := NewSSHClientWithAuth(HostSSHAuth(host))
	if err != nil {
		return err
	}
	defer client.Close()

	// 检测架构
	arch, _, _ := client.Execute(ctx, "uname -m")
	lokiArch := strings.TrimSpace(arch)
	switch lokiArch {
	case "x86_64":
		lokiArch = "amd64"
	case "aarch64":
		lokiArch = "arm64"
	}

	version := d.effectiveVersion()
	d.log("info", fmt.Sprintf("SSH connected, downloading Loki %s (%s)...", version, lokiArch), "")

	// 创建目录
	client.Execute(ctx, "mkdir -p /opt/loki /var/lib/loki /var/lib/loki/rules /var/lib/loki/rules-temp /etc/loki")

	// 下载 Loki
	downloadURL := fmt.Sprintf("https://github.com/grafana/loki/releases/download/v%s/loki-linux-%s.zip", version, lokiArch)
	cmd := fmt.Sprintf("cd /tmp && curl -fsSL -o loki.zip '%s' && unzip -o loki.zip -d /opt/loki/ && chmod +x /opt/loki/loki-linux-%s",
		downloadURL, lokiArch)
	if _, _, err := client.Execute(ctx, cmd); err != nil {
		return fmt.Errorf("download Loki failed: %w", err)
	}
	client.Execute(ctx, "ln -sf /opt/loki/loki-linux-* /usr/local/bin/loki 2>/dev/null || true")

	// 上传配置
	config := d.generateLokiConfig()
	client.Execute(ctx, fmt.Sprintf("cat > /etc/loki/loki.yaml << 'EOF'\n%s\nEOF", config))

	// systemd 服务（生产级配置）
	service := `[Unit]
Description=Grafana Loki
Wants=network-online.target
After=network-online.target

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

[Install]
WantedBy=multi-user.target
`
	client.Execute(ctx, fmt.Sprintf("cat > /etc/systemd/system/loki.service << 'EOF'\n%s\nEOF", service))

	// 启动服务
	if _, _, err := client.Execute(ctx, "systemctl daemon-reload && systemctl enable loki && systemctl restart loki"); err != nil {
		return fmt.Errorf("start Loki failed: %w", err)
	}

	// 健康检查
	d.log("info", "Waiting for Loki to be ready...", "")
	for i := 0; i < 10; i++ {
		_, _, err := client.Execute(ctx, "curl -fsSL http://localhost:3100/ready")
		if err == nil {
			break
		}
		client.Execute(ctx, "sleep 2")
	}

	d.log("success", fmt.Sprintf("Loki %s deployed on %s", version, host.HostIP), "")
	return nil
}

// ==================== 部署参数与版本校验 ====================

func (d *LokiDeployer) effectiveVersion() string {
	if d.deployment.Version != "" {
		return d.deployment.Version
	}
	return "3.0.0"
}

func (d *LokiDeployer) effectiveNamespace() string {
	if d.deployment.Namespace != "" {
		return d.deployment.Namespace
	}
	return "log"
}

func (d *LokiDeployer) effectiveStorageType() string {
	if d.deployment.StorageConfig != nil && d.deployment.StorageConfig.StorageType != "" {
		return d.deployment.StorageConfig.StorageType
	}
	return "filesystem"
}

func (d *LokiDeployer) checkVersionCompatibility() error {
	version := d.effectiveVersion()
	parts := strings.Split(version, ".")
	if len(parts) < 1 {
		return fmt.Errorf("cannot parse version: %s", version)
	}
	major, err := strconv.Atoi(parts[0])
	if err != nil {
		return fmt.Errorf("cannot parse major version: %s", version)
	}
	if major < 2 {
		return fmt.Errorf("Loki version %s is too old, recommend 2.8+ or 3.x", version)
	}
	return nil
}

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

// ==================== 清理 / 卸载 ====================

// lokiCleanupKinds 停止/清理时删除的 Kubernetes 资源清单。
// 刻意不包含 persistentvolumeclaim(loki-data):Stop/Restart 也会走到这里,
// 删 PVC 等于把 filesystem 存储的日志数据永久清空——数据只随显式卸载流程处理。
var lokiCleanupKinds = [][2]string{
	{"statefulset", "loki"},
	{"service", "loki"},
	{"configmap", "loki-config"},
}

func CleanupLoki(ctx context.Context, deployment *models.LokiDeployment) {
	if deployment.Mode == "k8s-yaml" && deployment.Cluster != nil {
		ns := deployment.Namespace
		if ns == "" {
			ns = "log"
		}
		for _, item := range lokiCleanupKinds {
			_ = deleteK8sResource(ctx, deployment.Cluster, item[0], ns, item[1])
		}
	}
	// 仅清理本产品创建的资源；其他途径部署的遗留资源不自动删除，避免误删数据。

	if deployment.Mode == "host-binary" && deployment.Host != nil {
		host := deployment.Host
		client, _ := NewSSHClientWithAuth(HostSSHAuth(host))
		if client != nil {
			defer client.Close()
			client.Execute(ctx, "systemctl stop loki 2>/dev/null || true")
			client.Execute(ctx, "systemctl disable loki 2>/dev/null || true")
			client.Execute(ctx, "rm -f /etc/systemd/system/loki.service /usr/local/bin/loki")
			client.Execute(ctx, "systemctl daemon-reload")
		}
	}
	if deployment.Mode == "host-docker" && deployment.Host != nil {
		_ = stopDockerLoki(ctx, deployment)
	}
}

// DecryptCred 解密凭证
func DecryptCred(encrypted string) (string, error) {
	if strings.HasPrefix(encrypted, "ENC:") {
		return crypto.Decrypt(encrypted[4:])
	}
	return encrypted, nil
}

// ==================== 通用工具函数 ====================

// writeTempFile 安全地创建临时文件，返回文件路径和清理函数

// indentYAML 对 YAML 内容进行缩进（用于嵌入 ConfigMap）
func indentYAML(s string, spaces int) string {
	pad := strings.Repeat(" ", spaces)
	lines := strings.Split(s, "\n")
	for i, l := range lines {
		if l != "" {
			lines[i] = pad + l
		}
	}
	return strings.Join(lines, "\n")
}