package services import ( "context" "fmt" "os" "strconv" "strings" "opsxloki/models" "opsxloki/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 } // ==================== Helm Values 生成 ==================== 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) } } // ==================== 清理 / 卸载 ==================== 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 [][2]string{{"statefulset", "loki"}, {"service", "loki"}, {"configmap", "loki-config"}, {"persistentvolumeclaim", "loki-data"}} { _ = deleteK8sResource(ctx, deployment.Cluster, item[0], ns, item[1]) } } // 历史 Helm release 不由此版本自动删除,避免误删旧数据。 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 安全地创建临时文件,返回文件路径和清理函数 func writeTempFile(pattern, content string, perm os.FileMode) (string, func(), error) { f, err := os.CreateTemp("", pattern) if err != nil { return "", nil, fmt.Errorf("create temp file failed: %w", err) } if _, err := f.WriteString(content); err != nil { f.Close() os.Remove(f.Name()) return "", nil, fmt.Errorf("write temp file failed: %w", err) } f.Close() if perm > 0 { os.Chmod(f.Name(), perm) } cleanup := func() { os.Remove(f.Name()) } return f.Name(), cleanup, nil } // 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") }