2026-06-03 18:41:52 +08:00

1457 lines
43 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package api
import (
"bytes"
"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 resolveModuleStatusToken(secrets InstallSecrets) string {
return firstNonEmpty(os.Getenv(envModuleStatusCallbackToken), secrets.ModuleStatusToken)
}
func materialAppEnv(req InstallRequest, secrets InstallSecrets, licenseConfig InstallLicenseConfig, localIP string, webPort int) []string {
moduleStatusToken := resolveModuleStatusToken(secrets)
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),
"PAID_CORE_MODULE_URL=http://flnxi-module-paid-core:8787",
"MODULE_RUNTIME_STATUS_TOKEN=" + moduleStatusToken,
}
}
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"
envModuleStatusCallbackURL = "FLNXI_MODULE_STATUS_CALLBACK_URL"
envModuleStatusCallbackToken = "FLNXI_MODULE_STATUS_CALLBACK_TOKEN"
)
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
moduleCallbackURL string
moduleCallbackToken string
moduleCallbackMutex sync.RWMutex
)
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()
moduleStatusCallbackURL := firstNonEmpty(
os.Getenv(envModuleStatusCallbackURL),
"http://"+localIP+":"+fmt.Sprintf("%d", webPort)+"/api/license/premium-modules/runtime-status",
)
setModuleStatusCallbackConfig(moduleStatusCallbackURL, resolveModuleStatusToken(secrets))
// 启动素材系统
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()
go reportModuleStatus(status)
}
func setModuleStatusCallbackConfig(callbackURL string, callbackToken string) {
moduleCallbackMutex.Lock()
moduleCallbackURL = strings.TrimSpace(callbackURL)
moduleCallbackToken = strings.TrimSpace(callbackToken)
moduleCallbackMutex.Unlock()
}
func getModuleStatusCallbackConfig() (string, string) {
callbackURL := strings.TrimSpace(os.Getenv(envModuleStatusCallbackURL))
callbackToken := strings.TrimSpace(os.Getenv(envModuleStatusCallbackToken))
if callbackURL != "" && callbackToken != "" {
return callbackURL, callbackToken
}
moduleCallbackMutex.RLock()
defer moduleCallbackMutex.RUnlock()
return moduleCallbackURL, moduleCallbackToken
}
func reportModuleStatus(status ModuleRuntimeStatus) {
callbackURL, callbackToken := getModuleStatusCallbackConfig()
if callbackURL == "" || callbackToken == "" {
return
}
body, err := json.Marshal(status)
if err != nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, callbackURL, bytes.NewReader(body))
if err != nil {
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+callbackToken)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return
}
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
}
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-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 ""
}