package api import ( "context" "encoding/base64" "encoding/json" "fmt" "io" "net" "net/http" "os" "runtime" "strings" "sync" "time" "flnxi-agent/internal/docker" "github.com/docker/docker/api/types/container" "github.com/docker/go-connections/nat" "github.com/gin-gonic/gin" ) // hostIP 存储从请求中获取的宿主机 IP var hostIP string var hostIPMutex sync.RWMutex // getHostIP 获取宿主机 IP(优先使用从请求中获取的 IP) func getHostIP() string { hostIPMutex.RLock() ip := hostIP hostIPMutex.RUnlock() if ip != "" { return ip } // 回退:尝试获取本机 IP(在容器内可能不准确) return getLocalIP() } // setHostIP 设置宿主机 IP func setHostIP(ip string) { hostIPMutex.Lock() hostIP = ip hostIPMutex.Unlock() } // extractHostIP 从请求中提取宿主机 IP // 优先从请求的 Host 头获取(用户访问 Agent 的地址就是宿主机地址) func extractHostIP(c *gin.Context) string { // 1. 从 Host 头获取(最可靠,因为用户是通过宿主机 IP 访问 Agent 的) host := c.Request.Host if host != "" { // 去掉端口号 if h, _, err := net.SplitHostPort(host); err == nil { host = h } // 验证是有效的 IP 地址 if ip := net.ParseIP(host); ip != nil && !ip.IsLoopback() { return host } } // 2. 从 X-Forwarded-For 获取 if xff := c.GetHeader("X-Forwarded-For"); xff != "" { parts := strings.Split(xff, ",") if len(parts) > 0 { ip := strings.TrimSpace(parts[0]) if parsed := net.ParseIP(ip); parsed != nil && !parsed.IsLoopback() { return ip } } } // 3. 从 X-Real-IP 获取 if xri := c.GetHeader("X-Real-IP"); xri != "" { if parsed := net.ParseIP(xri); parsed != nil && !parsed.IsLoopback() { return xri } } return "" } // getLocalIP 获取本机 IP 地址(回退方案) func getLocalIP() string { addrs, err := net.InterfaceAddrs() if err != nil { return "localhost" } for _, addr := range addrs { if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { if ipnet.IP.To4() != nil { return ipnet.IP.String() } } } return "localhost" } func firstNonEmpty(values ...string) string { for _, value := range values { trimmed := strings.TrimSpace(value) if trimmed != "" { return trimmed } } return "" } func resolveInstallImages(req InstallRequest) (string, string, string, error) { materialImage := firstNonEmpty(req.MaterialImage, os.Getenv(envMaterialImage), os.Getenv(envMaterialImageCompat)) if materialImage == "" { return "", "", "", fmt.Errorf("%s or %s is required", envMaterialImage, envMaterialImageCompat) } if err := validateFixedImageTag(materialImage); err != nil { return "", "", "", fmt.Errorf("material image: %w", err) } postgresImage := firstNonEmpty(req.PostgresImage, os.Getenv(envPostgresImage), os.Getenv(envPostgresImageCompat), "postgres:15-alpine") minioImage := firstNonEmpty(req.MinioImage, os.Getenv(envMinioImage), os.Getenv(envMinioImageCompat), "minio/minio:RELEASE.2025-04-22T22-12-26Z") return materialImage, postgresImage, minioImage, nil } type InstallLicenseConfig struct { ServerURL string ProjectCode string RequestSigningSecret string PublicKey string LocalEncryptionKey string } func resolveInstallLicenseConfig(req InstallRequest, secrets InstallSecrets) (InstallLicenseConfig, error) { config := InstallLicenseConfig{ ServerURL: firstNonEmpty(req.LicenseServerURL, os.Getenv(envLicenseServerURL)), ProjectCode: firstNonEmpty(req.LicenseProjectCode, os.Getenv(envLicenseProjectCode)), RequestSigningSecret: firstNonEmpty(req.LicenseRequestSigningSecret, os.Getenv(envLicenseRequestSigningSecret)), PublicKey: firstNonEmpty(req.LicensePublicKey, os.Getenv(envLicensePublicKey)), LocalEncryptionKey: firstNonEmpty(req.LocalEncryptionKey, os.Getenv(envLocalEncryptionKey), secrets.LocalEncryptionKey), } missing := make([]string, 0) if config.ServerURL == "" { missing = append(missing, envLicenseServerURL) } if config.ProjectCode == "" { missing = append(missing, envLicenseProjectCode) } if len(config.RequestSigningSecret) < 32 { missing = append(missing, envLicenseRequestSigningSecret+"(至少32字符)") } if config.PublicKey == "" { missing = append(missing, envLicensePublicKey) } if len(config.LocalEncryptionKey) < 32 { missing = append(missing, envLocalEncryptionKey+"(至少32字符)") } if len(missing) > 0 { return InstallLicenseConfig{}, fmt.Errorf("missing material license config: %s", strings.Join(missing, ", ")) } return config, nil } func materialAppEnv(req InstallRequest, secrets InstallSecrets, licenseConfig InstallLicenseConfig, localIP string, webPort int) []string { return []string{ "LICENSE_KEY=" + req.LicenseKey, "LICENSE_SERVER_URL=" + licenseConfig.ServerURL, "LICENSE_PROJECT_CODE=" + licenseConfig.ProjectCode, "LICENSE_REQUEST_SIGNING_SECRET=" + licenseConfig.RequestSigningSecret, "LICENSE_PUBLIC_KEY=" + licenseConfig.PublicKey, "LOCAL_ENCRYPTION_KEY=" + licenseConfig.LocalEncryptionKey, "DATABASE_URL=postgresql://" + secrets.DbUser + ":" + secrets.DbPassword + "@flnxi-db:5432/" + secrets.DbName, // S3/MinIO 配置(内部访问使用 Docker 网络) "S3_ENDPOINT=http://flnxi-minio:9000", "S3_ACCESS_KEY_ID=" + secrets.MinioRootUser, "S3_SECRET_ACCESS_KEY=" + secrets.MinioRootPassword, "S3_BUCKET=" + secrets.S3Bucket, "S3_REGION=us-east-1", "S3_ENABLE_PATH_STYLE=1", // S3 公开访问配置(浏览器访问使用宿主机 IP) "S3_PUBLIC_DOMAIN=http://" + localIP + ":9000/" + secrets.S3Bucket, "SERVER_IP=" + localIP, "MINIO_API_PORT=9000", // NextAuth 配置 "NEXTAUTH_SECRET=" + secrets.NextAuthSecret, "NEXTAUTH_URL=http://" + localIP + ":" + fmt.Sprintf("%d", webPort), "APP_PUBLIC_URL=http://" + localIP + ":" + fmt.Sprintf("%d", webPort), } } func resolveUpdateImage(req UpdateRequest) (string, error) { if image := firstNonEmpty(req.Image); image != "" { if err := validateFixedImageTag(image); err != nil { return "", err } return image, nil } repository := firstNonEmpty(os.Getenv(envMaterialImageRepository)) if repository == "" { if current := firstNonEmpty(os.Getenv(envMaterialImage), os.Getenv(envMaterialImageCompat)); current != "" { if idx := strings.LastIndex(current, ":"); idx > strings.LastIndex(current, "/") { repository = current[:idx] } } } if repository == "" { return "", fmt.Errorf("%s is required when update image is not provided", envMaterialImageRepository) } version := strings.TrimSpace(req.Version) if version == "" { return "", fmt.Errorf("version is required") } if strings.EqualFold(version, "latest") { return "", fmt.Errorf("latest is not allowed for customer updates; use a fixed version tag") } return repository + ":" + version, nil } func validateFixedImageTag(image string) error { image = strings.TrimSpace(image) if image == "" { return fmt.Errorf("image is required") } imageName := image if at := strings.Index(imageName, "@"); at >= 0 { imageName = imageName[:at] } lastSlash := strings.LastIndex(imageName, "/") lastColon := strings.LastIndex(imageName, ":") if lastColon <= lastSlash || lastColon == len(imageName)-1 { return fmt.Errorf("image must include a fixed version tag") } tag := strings.TrimSpace(imageName[lastColon+1:]) if strings.EqualFold(tag, "latest") { return fmt.Errorf("latest is not allowed for customer updates; use a fixed version tag") } return nil } const ( envMaterialImage = "FLNXI_MATERIAL_IMAGE" envMaterialImageCompat = "APP_IMAGE" envMaterialImageRepository = "FLNXI_MATERIAL_IMAGE_REPOSITORY" envPostgresImage = "FLNXI_POSTGRES_IMAGE" envPostgresImageCompat = "POSTGRES_IMAGE" envMinioImage = "FLNXI_MINIO_IMAGE" envMinioImageCompat = "MINIO_IMAGE" envVersionListURL = "FLNXI_VERSION_LIST_URL" envLicenseServerURL = "LICENSE_SERVER_URL" envLicenseProjectCode = "LICENSE_PROJECT_CODE" envLicenseRequestSigningSecret = "LICENSE_REQUEST_SIGNING_SECRET" envLicensePublicKey = "LICENSE_PUBLIC_KEY" envLocalEncryptionKey = "LOCAL_ENCRYPTION_KEY" ) var ( installStatus = &InstallStatus{Status: "idle"} updateStatus = &UpdateStatus{Status: "idle"} statusMutex sync.RWMutex moduleStatuses = map[string]ModuleRuntimeStatus{} moduleStatusesMutex sync.RWMutex moduleSyncLocks = map[string]bool{} moduleSyncLocksMutex sync.Mutex ) type InstallStatus struct { Status string `json:"status"` Step string `json:"step"` Progress int `json:"progress"` Message string `json:"message"` Error string `json:"error,omitempty"` } type UpdateStatus struct { Status string `json:"status"` Step string `json:"step"` Progress int `json:"progress"` Message string `json:"message"` Error string `json:"error,omitempty"` } type ModulePullCredential struct { RegistryHost string `json:"registryHost"` Repository string `json:"repository"` Tag string `json:"tag"` Digest string `json:"digest,omitempty"` Username string `json:"username"` PullToken string `json:"pullToken"` ExpiresAt string `json:"expiresAt"` } type ModuleSyncRequest struct { ModuleID string `json:"moduleId"` Image string `json:"image"` Digest string `json:"digest,omitempty"` ContainerName string `json:"containerName,omitempty"` Network string `json:"network,omitempty"` Env map[string]string `json:"env,omitempty"` HealthTimeoutSeconds int `json:"healthTimeoutSeconds,omitempty"` PullCredential ModulePullCredential `json:"pullCredential"` } type ModuleRuntimeStatus struct { ModuleID string `json:"moduleId"` Image string `json:"image"` Digest string `json:"digest,omitempty"` Status string `json:"status"` LastSyncAt string `json:"lastSyncAt"` LastError string `json:"lastError,omitempty"` } type InstallRequest struct { LicenseKey string `json:"licenseKey"` DataPath string `json:"dataPath"` WebPort int `json:"webPort"` AuthToken string `json:"authToken,omitempty"` MaterialImage string `json:"materialImage,omitempty"` PostgresImage string `json:"postgresImage,omitempty"` MinioImage string `json:"minioImage,omitempty"` LicenseServerURL string `json:"licenseServerUrl,omitempty"` LicenseProjectCode string `json:"licenseProjectCode,omitempty"` LicenseRequestSigningSecret string `json:"licenseRequestSigningSecret,omitempty"` LicensePublicKey string `json:"licensePublicKey,omitempty"` LocalEncryptionKey string `json:"localEncryptionKey,omitempty"` } type UpdateRequest struct { Version string `json:"version"` AuthToken string `json:"authToken"` Image string `json:"image,omitempty"` } func GetSystemInfo(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { info, err := dc.GetDockerInfo(context.Background()) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{ "os": runtime.GOOS, "arch": runtime.GOARCH, "dockerVersion": info.ServerVersion, "containers": info.Containers, "containersRunning": info.ContainersRunning, "images": info.Images, "memTotal": info.MemTotal, "cpus": info.NCPU, }) } } func Install(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { var req InstallRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } // 从请求中获取宿主机 IP(用户访问 Agent UI 的地址) clientIP := extractHostIP(c) if clientIP != "" { setHostIP(clientIP) } statusMutex.Lock() if installStatus.Status == "running" { statusMutex.Unlock() c.JSON(http.StatusConflict, gin.H{"error": "安装正在进行中"}) return } installStatus = &InstallStatus{Status: "running", Step: "初始化", Progress: 0} statusMutex.Unlock() go runInstall(dc, req) c.JSON(http.StatusOK, gin.H{"message": "安装已开始"}) } } func GetInstallStatus(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { statusMutex.RLock() defer statusMutex.RUnlock() c.JSON(http.StatusOK, installStatus) } } type installDockerClient interface { PullImageWithProgress(ctx context.Context, image string, authStr string, onProgress func(current, total int64, status string)) error ExecOnHost(ctx context.Context, cmd string) error CreateNetworkIfNotExists(ctx context.Context, name string) error CreateAndStartContainer(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, name string) (string, error) } func runInstall(dc installDockerClient, req InstallRequest) { ctx := context.Background() setStep := func(step string, progress int, msg string) { statusMutex.Lock() installStatus.Step, installStatus.Progress, installStatus.Message = step, progress, msg statusMutex.Unlock() } setErr := func(e string) { statusMutex.Lock() installStatus.Status, installStatus.Error = "failed", e statusMutex.Unlock() } authStr := "" if req.AuthToken != "" { b, _ := json.Marshal(map[string]string{"username": "token", "password": req.AuthToken}) authStr = base64.URLEncoding.EncodeToString(b) } secrets, err := generateInstallSecrets() if err != nil { setErr("生成安装凭据失败: " + err.Error()) return } materialImage, postgresImage, minioImage, err := resolveInstallImages(req) if err != nil { setErr("安装镜像配置缺失: " + err.Error()) return } licenseConfig, err := resolveInstallLicenseConfig(req, secrets) if err != nil { setErr("许可证配置缺失: " + err.Error()) return } // 拉取素材系统镜像 (0-30%) setStep("拉取镜像", 1, "正在拉取素材系统镜像...") err = dc.PullImageWithProgress(ctx, materialImage, authStr, func(current, total int64, status string) { pct := 1 if total > 0 { pct = int(float64(current)/float64(total)*29) + 1 } setStep("拉取镜像", pct, "素材系统: "+status) }) if err != nil { setErr("拉取镜像失败: " + err.Error()) return } // 拉取数据库镜像 (30-50%) setStep("拉取镜像", 30, "正在拉取数据库镜像...") err = dc.PullImageWithProgress(ctx, postgresImage, "", func(current, total int64, status string) { pct := 30 if total > 0 { pct = int(float64(current)/float64(total)*20) + 30 } setStep("拉取镜像", pct, "数据库: "+status) }) if err != nil { setErr("拉取数据库镜像失败: " + err.Error()) return } // 拉取存储镜像 (50-60%) setStep("拉取镜像", 50, "正在拉取存储镜像...") err = dc.PullImageWithProgress(ctx, minioImage, "", func(current, total int64, status string) { pct := 50 if total > 0 { pct = int(float64(current)/float64(total)*10) + 50 } setStep("拉取镜像", pct, "存储服务: "+status) }) if err != nil { setErr("拉取存储镜像失败: " + err.Error()) return } setStep("准备环境", 60, "创建数据目录...") webPort := req.WebPort if webPort == 0 { webPort = 3000 } dataPath := req.DataPath if dataPath == "" { dataPath = "/opt/flnxi" } // 创建数据目录并设置权限(通过 alpine 容器执行) if err := dc.ExecOnHost(ctx, "mkdir -p "+dataPath+"/postgres "+dataPath+"/minio "+dataPath+"/material && chmod -R 777 "+dataPath); err != nil { setErr("创建数据目录失败: " + err.Error()) return } // 创建 Docker 网络 setStep("准备环境", 62, "创建 Docker 网络...") if err := dc.CreateNetworkIfNotExists(ctx, "flnxi-network"); err != nil { setErr("创建网络失败: " + err.Error()) return } // 启动数据库容器 setStep("启动服务", 65, "启动数据库...") dbCfg := &container.Config{ Image: postgresImage, Env: []string{ "POSTGRES_USER=" + secrets.DbUser, "POSTGRES_PASSWORD=" + secrets.DbPassword, "POSTGRES_DB=" + secrets.DbName, }, } dbHcfg := &container.HostConfig{ Binds: []string{dataPath + "/postgres:/var/lib/postgresql/data"}, NetworkMode: "flnxi-network", RestartPolicy: container.RestartPolicy{Name: "unless-stopped"}, } if _, err := dc.CreateAndStartContainer(ctx, dbCfg, dbHcfg, "flnxi-db"); err != nil { setErr("启动数据库失败: " + err.Error()) return } // 启动 MinIO 存储(暴露 9000 端口供浏览器访问) setStep("启动服务", 75, "启动存储服务...") minioCfg := &container.Config{ Image: minioImage, Env: []string{ "MINIO_ROOT_USER=" + secrets.MinioRootUser, "MINIO_ROOT_PASSWORD=" + secrets.MinioRootPassword, }, Cmd: []string{"server", "/data", "--console-address", ":9001"}, ExposedPorts: nat.PortSet{"9000/tcp": struct{}{}, "9001/tcp": struct{}{}}, } minioHcfg := &container.HostConfig{ Binds: []string{dataPath + "/minio:/data"}, NetworkMode: "flnxi-network", RestartPolicy: container.RestartPolicy{Name: "unless-stopped"}, PortBindings: nat.PortMap{ "9000/tcp": []nat.PortBinding{{HostIP: "0.0.0.0", HostPort: "9000"}}, "9001/tcp": []nat.PortBinding{{HostIP: "0.0.0.0", HostPort: "9001"}}, }, } if _, err := dc.CreateAndStartContainer(ctx, minioCfg, minioHcfg, "flnxi-minio"); err != nil { setErr("启动存储服务失败: " + err.Error()) return } // 等待数据库就绪 setStep("启动服务", 80, "等待数据库就绪...") time.Sleep(5 * time.Second) // 获取宿主机 IP 用于配置公开访问地址(从用户请求中获取,确保是宿主机 IP 而非容器 IP) localIP := getHostIP() // 启动素材系统 setStep("启动服务", 90, "启动素材系统...") cfg := &container.Config{ Image: materialImage, User: "0", // 以 root 用户运行,确保有权限写入挂载目录 Env: materialAppEnv(req, secrets, licenseConfig, localIP, webPort), ExposedPorts: nat.PortSet{"3000/tcp": struct{}{}}, } hcfg := &container.HostConfig{ PortBindings: nat.PortMap{"3000/tcp": []nat.PortBinding{{HostIP: "0.0.0.0", HostPort: fmt.Sprintf("%d", webPort)}}}, Binds: []string{dataPath + "/material:/app/data"}, NetworkMode: "flnxi-network", RestartPolicy: container.RestartPolicy{Name: "unless-stopped"}, } if _, err := dc.CreateAndStartContainer(ctx, cfg, hcfg, "flnxi-material"); err != nil { setErr("启动素材系统失败: " + err.Error()) return } statusMutex.Lock() installStatus.Status, installStatus.Step, installStatus.Progress, installStatus.Message = "success", "完成", 100, "安装完成" statusMutex.Unlock() } func Update(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { var req UpdateRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } statusMutex.Lock() if updateStatus.Status == "running" { statusMutex.Unlock() c.JSON(http.StatusConflict, gin.H{"error": "更新正在进行中"}) return } updateStatus = &UpdateStatus{Status: "running", Step: "初始化", Progress: 0} statusMutex.Unlock() go runUpdate(dc, req) c.JSON(http.StatusOK, gin.H{"message": "更新已开始"}) } } func GetUpdateStatus(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { statusMutex.RLock() defer statusMutex.RUnlock() c.JSON(http.StatusOK, updateStatus) } } func GetModuleStatuses() gin.HandlerFunc { return func(c *gin.Context) { moduleStatusesMutex.RLock() defer moduleStatusesMutex.RUnlock() result := make([]ModuleRuntimeStatus, 0, len(moduleStatuses)) for _, status := range moduleStatuses { result = append(result, status) } c.JSON(http.StatusOK, result) } } func SyncModule(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { var req ModuleSyncRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } if err := validateModuleSyncRequest(req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } syncKey := moduleContainerName(req) if !tryBeginModuleSync(syncKey) { c.JSON(http.StatusConflict, gin.H{"error": "该模块正在同步中,请稍后再试", "moduleId": req.ModuleID}) return } go func() { defer finishModuleSync(syncKey) runModuleSync(dc, req) }() c.JSON(http.StatusOK, gin.H{"message": "模块同步已开始", "moduleId": req.ModuleID}) } } type moduleDockerClient interface { GetContainer(ctx context.Context, name string) (*docker.ContainerInfo, error) PullImage(ctx context.Context, image string, authStr string) error GetImageRepoDigests(ctx context.Context, image string) ([]string, error) StopContainer(ctx context.Context, containerID string) error RemoveContainer(ctx context.Context, containerID string) error StartContainer(ctx context.Context, containerID string) error CreateAndStartContainer(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, name string) (string, error) WaitForHealthy(ctx context.Context, containerID string, timeout time.Duration) error } func setModuleStatus(status ModuleRuntimeStatus) { moduleStatusesMutex.Lock() moduleStatuses[status.ModuleID] = status moduleStatusesMutex.Unlock() } func tryBeginModuleSync(syncKey string) bool { moduleSyncLocksMutex.Lock() defer moduleSyncLocksMutex.Unlock() if moduleSyncLocks[syncKey] { return false } moduleSyncLocks[syncKey] = true return true } func finishModuleSync(syncKey string) { moduleSyncLocksMutex.Lock() delete(moduleSyncLocks, syncKey) moduleSyncLocksMutex.Unlock() } func moduleStatus(req ModuleSyncRequest, status string, err error) ModuleRuntimeStatus { lastError := "" if err != nil { lastError = err.Error() } return ModuleRuntimeStatus{ ModuleID: req.ModuleID, Image: req.Image, Digest: firstNonEmpty(req.Digest, req.PullCredential.Digest), Status: status, LastSyncAt: time.Now().UTC().Format(time.RFC3339), LastError: lastError, } } func moduleContainerName(req ModuleSyncRequest) string { if name := strings.TrimSpace(req.ContainerName); name != "" { return name } builder := strings.Builder{} for _, r := range strings.ToLower(req.ModuleID) { if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '-' || r == '_' { builder.WriteRune(r) } } suffix := builder.String() if suffix == "" { suffix = "unknown" } return "flnxi-module-" + suffix } func moduleEnv(req ModuleSyncRequest) []string { env := []string{"FLNXI_MODULE_ID=" + req.ModuleID} for key, value := range req.Env { key = strings.TrimSpace(key) if key == "" || strings.Contains(key, "=") { continue } env = append(env, key+"="+value) } return env } func moduleRegistryAuth(credential ModulePullCredential) string { payload := map[string]string{ "username": credential.Username, "password": credential.PullToken, "serveraddress": credential.RegistryHost, } b, _ := json.Marshal(payload) return base64.URLEncoding.EncodeToString(b) } func parseRegistryImage(image string) (registryHost string, repository string, tag string, err error) { image = strings.TrimSpace(image) imageName := image if at := strings.Index(imageName, "@"); at >= 0 { imageName = imageName[:at] } parts := strings.Split(imageName, "/") if len(parts) < 2 || strings.TrimSpace(parts[0]) == "" { err = fmt.Errorf("模块镜像地址无效") return } registryHost = parts[0] imagePath := strings.Join(parts[1:], "/") lastSlash := strings.LastIndex(imagePath, "/") lastColon := strings.LastIndex(imagePath, ":") if lastColon <= lastSlash || lastColon == len(imagePath)-1 { err = fmt.Errorf("模块镜像缺少固定版本 tag") return } repository = imagePath[:lastColon] tag = imagePath[lastColon+1:] return } func validateModuleSyncRequest(req ModuleSyncRequest) error { if strings.TrimSpace(req.ModuleID) == "" { return fmt.Errorf("moduleId is required") } if err := validateFixedImageTag(req.Image); err != nil { return fmt.Errorf("module image: %w", err) } registryHost, repository, tag, err := parseRegistryImage(req.Image) if err != nil { return err } credential := req.PullCredential if credential.Username != "flnxi-module" { return fmt.Errorf("模块拉取用户名无效") } if strings.TrimSpace(credential.PullToken) == "" { return fmt.Errorf("模块拉取 token 不能为空") } if credential.RegistryHost != registryHost || credential.Repository != repository || credential.Tag != tag { return fmt.Errorf("模块拉取凭证和镜像地址不匹配") } expectedDigest := firstNonEmpty(req.Digest, credential.Digest) if expectedDigest == "" { return fmt.Errorf("模块 digest 不能为空") } if !isValidSha256Digest(expectedDigest) { return fmt.Errorf("模块 digest 必须是 sha256 格式") } if req.Digest != "" && !isValidSha256Digest(req.Digest) { return fmt.Errorf("模块 digest 必须是 sha256 格式") } if credential.Digest != "" && !isValidSha256Digest(credential.Digest) { return fmt.Errorf("模块拉取凭证 digest 必须是 sha256 格式") } if req.Digest != "" && credential.Digest != "" && req.Digest != credential.Digest { return fmt.Errorf("模块 digest 和拉取凭证不匹配") } if credential.ExpiresAt == "" { return fmt.Errorf("模块拉取凭证缺少过期时间") } expiresAt, err := time.Parse(time.RFC3339, credential.ExpiresAt) if err != nil { return fmt.Errorf("模块拉取凭证过期时间无效") } if !expiresAt.After(time.Now()) { return fmt.Errorf("模块拉取凭证已过期") } return nil } func isValidSha256Digest(digest string) bool { if len(digest) != len("sha256:")+64 || !strings.HasPrefix(digest, "sha256:") { return false } for _, ch := range digest[len("sha256:"):] { if (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f') || (ch >= 'A' && ch <= 'F') { continue } return false } return true } func repoDigestsContain(repoDigests []string, digest string) bool { if digest == "" { return false } for _, value := range repoDigests { if value == digest || strings.HasSuffix(value, "@"+digest) { return true } } return false } func moduleContainerConfig(req ModuleSyncRequest) (*container.Config, *container.HostConfig) { network := firstNonEmpty(req.Network, os.Getenv("FLNXI_MODULE_NETWORK"), "flnxi-app-network") return &container.Config{ Image: req.Image, Env: moduleEnv(req), }, &container.HostConfig{ NetworkMode: container.NetworkMode(network), RestartPolicy: container.RestartPolicy{Name: "unless-stopped"}, ReadonlyRootfs: true, CapDrop: []string{"ALL"}, SecurityOpt: []string{"no-new-privileges:true"}, Tmpfs: map[string]string{"/tmp": "rw,noexec,nosuid,size=64m"}, Resources: container.Resources{ Memory: 256 * 1024 * 1024, NanoCPUs: 500000000, }, } } func restoreModuleContainer(ctx context.Context, dc moduleDockerClient, old *docker.ContainerInfo, name string) error { if old == nil { return fmt.Errorf("旧模块容器不存在") } current, err := dc.GetContainer(ctx, name) if err != nil { return err } if current != nil && current.ID == old.ID { if current.State != "running" { return dc.StartContainer(ctx, current.ID) } return nil } if current != nil { if current.State == "running" { _ = dc.StopContainer(ctx, current.ID) } if err := dc.RemoveContainer(ctx, current.ID); err != nil { return err } } image := oldMaterialImage(old) if image == "" { return fmt.Errorf("旧模块镜像信息不存在") } cfg, hcfg, err := materialContainerConfig(old, image) if err != nil { return err } _, err = dc.CreateAndStartContainer(ctx, cfg, hcfg, name) return err } func runModuleSync(dc moduleDockerClient, req ModuleSyncRequest) { ctx := context.Background() name := moduleContainerName(req) setModuleStatus(moduleStatus(req, "pulling", nil)) old, err := dc.GetContainer(ctx, name) if err != nil { setModuleStatus(moduleStatus(req, "unhealthy", err)) return } authStr := moduleRegistryAuth(req.PullCredential) if err := dc.PullImage(ctx, req.Image, authStr); err != nil { setModuleStatus(moduleStatus(req, "missing", err)) return } expectedDigest := firstNonEmpty(req.Digest, req.PullCredential.Digest) repoDigests, err := dc.GetImageRepoDigests(ctx, req.Image) if err != nil { setModuleStatus(moduleStatus(req, "unhealthy", err)) return } if !repoDigestsContain(repoDigests, expectedDigest) { err := fmt.Errorf("模块镜像 digest 不匹配") setModuleStatus(moduleStatus(req, "unhealthy", err)) return } if old != nil { if old.State == "running" { _ = dc.StopContainer(ctx, old.ID) } if err := dc.RemoveContainer(ctx, old.ID); err != nil { setModuleStatus(moduleStatus(req, "unhealthy", err)) return } } cfg, hcfg := moduleContainerConfig(req) newID, err := dc.CreateAndStartContainer(ctx, cfg, hcfg, name) if err != nil { if old != nil { _ = restoreModuleContainer(ctx, dc, old, name) } setModuleStatus(moduleStatus(req, "unhealthy", err)) return } timeout := time.Duration(req.HealthTimeoutSeconds) * time.Second if timeout <= 0 { timeout = 30 * time.Second } if err := dc.WaitForHealthy(ctx, newID, timeout); err != nil { if old != nil { _ = restoreModuleContainer(ctx, dc, old, name) } setModuleStatus(moduleStatus(req, "unhealthy", err)) return } setModuleStatus(moduleStatus(req, "running", nil)) } type updateDockerClient interface { GetContainer(ctx context.Context, name string) (*docker.ContainerInfo, error) PullImage(ctx context.Context, image string, authStr string) error StopContainer(ctx context.Context, containerID string) error RemoveContainer(ctx context.Context, containerID string) error StartContainer(ctx context.Context, containerID string) error CreateAndStartContainer(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, name string) (string, error) WaitForHealthy(ctx context.Context, containerID string, timeout time.Duration) error RemoveImage(ctx context.Context, imageID string) error PruneImages(ctx context.Context) error } func runUpdate(dc updateDockerClient, req UpdateRequest) { ctx := context.Background() setStep := func(step string, progress int, msg string) { statusMutex.Lock() updateStatus.Step, updateStatus.Progress, updateStatus.Message = step, progress, msg statusMutex.Unlock() } setErr := func(e string) { statusMutex.Lock() updateStatus.Status, updateStatus.Error = "failed", e statusMutex.Unlock() } failWithRollback := func(prefix string, err error, old *docker.ContainerInfo) { setStep("回滚旧版本", 85, "更新失败,正在恢复旧版本...") rollbackErr := restoreMaterialContainer(ctx, dc, old) if rollbackErr != nil { setErr(fmt.Sprintf("%s: %v;旧版本恢复失败: %v", prefix, err, rollbackErr)) return } setErr(fmt.Sprintf("%s: %v;已恢复旧版本容器", prefix, err)) } authStr := "" if req.AuthToken != "" { b, _ := json.Marshal(map[string]string{"username": "token", "password": req.AuthToken}) authStr = base64.URLEncoding.EncodeToString(b) } setStep("准备更新", 5, "获取当前容器信息...") cont, err := dc.GetContainer(ctx, "flnxi-material") if err != nil || cont == nil { setErr("找不到素材系统容器") return } // 记录旧镜像 ID,用于后续清理 oldImageID := cont.ImageID oldConfig := cont image, err := resolveUpdateImage(req) if err != nil { setErr("更新镜像配置缺失: " + err.Error()) return } setStep("拉取镜像", 10, "正在拉取新版本镜像...") if err := dc.PullImage(ctx, image, authStr); err != nil { setErr("拉取镜像失败: " + err.Error()) return } setStep("停止服务", 40, "停止当前容器...") if err := dc.StopContainer(ctx, cont.ID); err != nil { setErr("停止容器失败: " + err.Error()) return } setStep("清理旧容器", 50, "删除旧容器...") if err := dc.RemoveContainer(ctx, cont.ID); err != nil { failWithRollback("删除容器失败", err, oldConfig) return } setStep("启动新版本", 60, "创建新容器...") cfg, hcfg, err := materialContainerConfig(oldConfig, image) if err != nil { failWithRollback("生成新容器配置失败", err, oldConfig) return } newID, err := dc.CreateAndStartContainer(ctx, cfg, hcfg, "flnxi-material") if err != nil { failWithRollback("启动新容器失败", err, oldConfig) return } setStep("健康检查", 80, "等待服务就绪...") if err := dc.WaitForHealthy(ctx, newID, 60*time.Second); err != nil { failWithRollback("健康检查失败", err, oldConfig) return } // 清理旧镜像 setStep("清理旧版本", 90, "清理旧镜像...") if oldImageID != "" { // 尝试删除旧镜像,忽略错误(可能被其他容器使用) _ = dc.RemoveImage(ctx, oldImageID) } // 清理悬空镜像 setStep("清理旧版本", 95, "清理悬空镜像...") _ = dc.PruneImages(ctx) statusMutex.Lock() updateStatus.Status, updateStatus.Step, updateStatus.Progress, updateStatus.Message = "success", "完成", 100, "更新完成" statusMutex.Unlock() } func materialContainerConfig(old *docker.ContainerInfo, image string) (*container.Config, *container.HostConfig, error) { if old == nil || old.Config == nil { return nil, nil, fmt.Errorf("旧容器配置不存在") } cfg := *old.Config cfg.Image = image hcfg := &container.HostConfig{} if old.HostConfig != nil { copied := *old.HostConfig hcfg = &copied } if hcfg.RestartPolicy.Name == "" { hcfg.RestartPolicy = container.RestartPolicy{Name: "unless-stopped"} } return &cfg, hcfg, nil } func oldMaterialImage(old *docker.ContainerInfo) string { if old == nil || old.Config == nil { return "" } image := strings.TrimSpace(old.Config.Image) if image != "" { return image } return strings.TrimSpace(old.ImageID) } func restoreMaterialContainer(ctx context.Context, dc updateDockerClient, old *docker.ContainerInfo) error { if old == nil { return fmt.Errorf("旧容器信息不存在") } current, err := dc.GetContainer(ctx, "flnxi-material") if err != nil { return fmt.Errorf("查询当前容器失败: %w", err) } if current != nil && current.ID == old.ID { if current.State != "running" { if err := dc.StartContainer(ctx, current.ID); err != nil { return fmt.Errorf("启动旧容器失败: %w", err) } } return nil } if current != nil { if current.State == "running" { _ = dc.StopContainer(ctx, current.ID) } if err := dc.RemoveContainer(ctx, current.ID); err != nil { return fmt.Errorf("删除失败的新容器失败: %w", err) } } image := oldMaterialImage(old) if image == "" { return fmt.Errorf("旧镜像信息不存在") } cfg, hcfg, err := materialContainerConfig(old, image) if err != nil { return err } if _, err := dc.CreateAndStartContainer(ctx, cfg, hcfg, "flnxi-material"); err != nil { return fmt.Errorf("重建旧容器失败: %w", err) } return nil } func ListContainers(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { containers, err := dc.ListContainers(context.Background(), "flnxi") if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } result := make([]gin.H, 0) for _, ct := range containers { result = append(result, gin.H{ "id": ct.ID[:12], "name": ct.Names[0], "image": ct.Image, "state": ct.State, "status": ct.Status, "created": ct.Created, }) } c.JSON(http.StatusOK, result) } } func RestartContainer(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { name := c.Param("name") cont, err := dc.GetContainer(context.Background(), name) if err != nil || cont == nil { c.JSON(http.StatusNotFound, gin.H{"error": "容器不存在"}) return } if err := dc.RestartContainer(context.Background(), cont.ID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "重启成功"}) } } func StopContainer(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { name := c.Param("name") cont, err := dc.GetContainer(context.Background(), name) if err != nil || cont == nil { c.JSON(http.StatusNotFound, gin.H{"error": "容器不存在"}) return } if err := dc.StopContainer(context.Background(), cont.ID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "停止成功"}) } } func StartContainer(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { name := c.Param("name") cont, err := dc.GetContainer(context.Background(), name) if err != nil || cont == nil { c.JSON(http.StatusNotFound, gin.H{"error": "容器不存在"}) return } if err := dc.StartContainer(context.Background(), cont.ID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "启动成功"}) } } func HealthCheck(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { _, err := dc.GetDockerInfo(context.Background()) if err != nil { c.JSON(http.StatusServiceUnavailable, gin.H{"status": "unhealthy", "error": "Docker 连接失败"}) return } cont, _ := dc.GetContainer(context.Background(), "flnxi-material") materialStatus := "not_found" if cont != nil { materialStatus = cont.State } c.JSON(http.StatusOK, gin.H{"status": "healthy", "agent": os.Getenv("AGENT_VERSION"), "material": materialStatus}) } } // GetVersions returns versions from a configured version-list endpoint. func GetVersions(dc *docker.Client) gin.HandlerFunc { return func(c *gin.Context) { versionListURL := strings.TrimSpace(os.Getenv(envVersionListURL)) if versionListURL == "" { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "not_configured", }) return } client := &http.Client{Timeout: 10 * time.Second} req, err := http.NewRequest("GET", versionListURL, nil) if err != nil { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "fallback", }) return } resp, err := client.Do(req) if err != nil { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "fallback", }) return } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "fallback", }) return } body, err := io.ReadAll(resp.Body) if err != nil { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "fallback", }) return } var tagsResp struct { Name string `json:"name"` Tags []string `json:"tags"` } if err := json.Unmarshal(body, &tagsResp); err != nil { c.JSON(http.StatusOK, gin.H{ "versions": []string{}, "latest": "", "source": "fallback", }) return } // 过滤并排序版本号 versions := filterAndSortVersions(tagsResp.Tags) latest := findLatestVersion(versions) c.JSON(http.StatusOK, gin.H{ "versions": versions, "latest": latest, "source": "configured", }) } } // filterAndSortVersions 过滤并排序版本号 func filterAndSortVersions(tags []string) []string { var versions []string for _, tag := range tags { if tag == "latest" { continue } // 只保留语义化版本号 (如 1.0.0, 1.0.1) if isSemanticVersion(tag) { versions = append(versions, tag) } } // 按版本号降序排序 sortVersionsDesc(versions) return versions } // isSemanticVersion 检查是否是语义化版本号 func isSemanticVersion(tag string) bool { parts := splitVersion(tag) return len(parts) >= 2 } // splitVersion 分割版本号 func splitVersion(v string) []int { var parts []int for _, p := range splitString(v, '.') { n := 0 for _, c := range p { if c >= '0' && c <= '9' { n = n*10 + int(c-'0') } else { break } } parts = append(parts, n) } return parts } // splitString 简单的字符串分割 func splitString(s string, sep byte) []string { var result []string start := 0 for i := 0; i < len(s); i++ { if s[i] == sep { result = append(result, s[start:i]) start = i + 1 } } result = append(result, s[start:]) return result } // sortVersionsDesc 按版本号降序排序 func sortVersionsDesc(versions []string) { for i := 0; i < len(versions)-1; i++ { for j := i + 1; j < len(versions); j++ { if compareVersions(versions[i], versions[j]) < 0 { versions[i], versions[j] = versions[j], versions[i] } } } } // compareVersions 比较两个版本号 func compareVersions(v1, v2 string) int { p1 := splitVersion(v1) p2 := splitVersion(v2) maxLen := len(p1) if len(p2) > maxLen { maxLen = len(p2) } for i := 0; i < maxLen; i++ { n1, n2 := 0, 0 if i < len(p1) { n1 = p1[i] } if i < len(p2) { n2 = p2[i] } if n1 != n2 { return n1 - n2 } } return 0 } // findLatestVersion 找到最新的语义化版本 func findLatestVersion(versions []string) string { for _, v := range versions { if v != "latest" && isSemanticVersion(v) { return v } } if len(versions) > 0 { return versions[0] } return "" }