From 2db8264c8ef66ae8f816bdc812edf08044953679 Mon Sep 17 00:00:00 2001 From: mac Date: Tue, 21 Jul 2026 18:26:50 +0800 Subject: [PATCH] feat(service): add deployment task integration --- frontend/components.d.ts | 1 + frontend/src/api/deployment.ts | 68 ++ frontend/src/router/index.ts | 6 + frontend/src/views/service/Catalog.vue | 1556 +++++++++++++++++++----- server/.env.example | 3 + server/internal/config/config.go | 178 +-- server/internal/database/database.go | 2 + server/internal/handler/context.go | 10 +- server/internal/handler/deployment.go | 249 ++++ server/internal/model/models.go | 28 + server/internal/router/router.go | 8 + server/internal/service/deployment.go | 499 ++++++++ 12 files changed, 2226 insertions(+), 382 deletions(-) create mode 100644 frontend/src/api/deployment.ts create mode 100644 server/internal/handler/deployment.go create mode 100644 server/internal/service/deployment.go diff --git a/frontend/components.d.ts b/frontend/components.d.ts index 67fe957..98a79b5 100644 --- a/frontend/components.d.ts +++ b/frontend/components.d.ts @@ -17,6 +17,7 @@ declare module 'vue' { ElFormItem: typeof import('element-plus/es')['ElFormItem'] ElIcon: typeof import('element-plus/es')['ElIcon'] ElInput: typeof import('element-plus/es')['ElInput'] + ElInputNumber: typeof import('element-plus/es')['ElInputNumber'] ElOption: typeof import('element-plus/es')['ElOption'] ElSelect: typeof import('element-plus/es')['ElSelect'] ElTable: typeof import('element-plus/es')['ElTable'] diff --git a/frontend/src/api/deployment.ts b/frontend/src/api/deployment.ts new file mode 100644 index 0000000..99d8118 --- /dev/null +++ b/frontend/src/api/deployment.ts @@ -0,0 +1,68 @@ +import { getToken } from '@/utils/auth' + +export interface DeploymentCreatePayload { + component: string + business_line_id: number + params: Record +} + +export interface DeploymentCreateResult { + deployment_id: string + status: string +} + +export const deploymentApi = { + async create(payload: DeploymentCreatePayload): Promise { + const data = await authRequest('/auth/api/v1/deployments', { + method: 'POST', + body: JSON.stringify(payload), + }) + return { + deployment_id: String(data.deployment_id || ''), + status: String(data.status || ''), + } + }, + + async cancel(deploymentId: string): Promise { + await authRequest(`/auth/api/v1/deployments/${encodeURIComponent(deploymentId)}/cancel`, { + method: 'POST', + }) + }, + + eventsURL(deploymentId: string): string { + const token = getToken() + const params = token ? `?access_token=${encodeURIComponent(token)}` : '' + return `/auth/api/v1/deployments/${encodeURIComponent(deploymentId)}/events${params}` + }, +} + +async function authRequest(path: string, init: RequestInit = {}) { + const token = getToken() + const response = await fetch(path, { + ...init, + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + ...(token ? { Authorization: `Bearer ${token}` } : {}), + ...init.headers, + }, + }) + const text = await response.text() + const data = parseResponseBody(text) + if (!response.ok) { + const message = data?.error || data?.message || text || `HTTP ${response.status}` + throw new Error(message) + } + return data || {} +} + +function parseResponseBody(text: string) { + if (!text.trim()) { + return {} + } + try { + return JSON.parse(text) + } catch { + return { error: text } + } +} diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index bd89c00..bc4d79a 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -105,6 +105,12 @@ const router = createRouter({ component: () => import('@/views/service/Catalog.vue'), meta: { title: '基础服务' }, }, + { + path: 'service/catalog/:component', + name: 'ServiceDelivery', + component: () => import('@/views/service/Catalog.vue'), + meta: { title: '基础服务交付' }, + }, { path: 'service/management', name: 'ServiceManagement', diff --git a/frontend/src/views/service/Catalog.vue b/frontend/src/views/service/Catalog.vue index 2257e6c..d34be53 100644 --- a/frontend/src/views/service/Catalog.vue +++ b/frontend/src/views/service/Catalog.vue @@ -1,17 +1,17 @@ diff --git a/server/.env.example b/server/.env.example index ff19953..7c2e099 100644 --- a/server/.env.example +++ b/server/.env.example @@ -11,6 +11,9 @@ WAYNE_API_BASE_URL=http://wayne-backend:8080 WAYNE_ADMIN_USERNAME=admin WAYNE_ADMIN_PASSWORD=change-this-wayne-admin-password WAYNE_TOKEN_TTL_MINUTES=1440 +ANSIBLE_SERVICE_BASE_URL=http://ansible-runner:8084 +ANSIBLE_INTERNAL_TOKEN=change-this-ansible-internal-token +DEPLOYMENT_CALLBACK_BASE_URL=http://authserver-backend:8080 MYSQL_DSN=auth:auth@tcp(127.0.0.1:3000)/authserver?charset=utf8mb4&parseTime=True&loc=Local AUTO_MIGRATE=true diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 168c450..111d3fb 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -14,49 +14,52 @@ type OAuthClient struct { } type Config struct { - AppEnv string - HTTPAddr string - PublicBaseURL string - MySQLDSN string - AutoMigrate bool - SSOEnabled bool - JWTSecret string - JWTIssuer string - JWTTTLMinutes int - SAMLEntityID string - SAMLACSURL string - SAMLSPCert string - SAMLSPKey string - SAMLIDPMetaURL string - SAMLLogoutURL string - WayenLoginURL string - WayenTargetURL string - WayenUsernameKey string - WayenPasswordKey string - WayenLoginFormat string - WayenLoginValue string - WayenOAuthRef string - WayenOAuthLoginURL string - WayneAPIBaseURL string - WayneAdminUsername string - WayneAdminPassword string - WayneTokenTTLMinutes int - WayneInternalAPIBaseURL string - WayneServiceName string - WayneServiceAPISecretKey string - OAuthClientID string - OAuthClientSecret string - OAuthRedirectURI string - OAuthCodeTTLSeconds int - OIDCIssuer string - OIDCAuthorizeURL string - OIDCTokenURL string - OIDCUserInfoURL string - OIDCJWKSURL string - CloudDMClientID string - CloudDMClientSecret string - CloudDMRedirectURI string - CloudDMTargetURL string + AppEnv string + HTTPAddr string + PublicBaseURL string + MySQLDSN string + AutoMigrate bool + SSOEnabled bool + JWTSecret string + JWTIssuer string + JWTTTLMinutes int + SAMLEntityID string + SAMLACSURL string + SAMLSPCert string + SAMLSPKey string + SAMLIDPMetaURL string + SAMLLogoutURL string + WayenLoginURL string + WayenTargetURL string + WayenUsernameKey string + WayenPasswordKey string + WayenLoginFormat string + WayenLoginValue string + WayenOAuthRef string + WayenOAuthLoginURL string + WayneAPIBaseURL string + WayneAdminUsername string + WayneAdminPassword string + WayneTokenTTLMinutes int + WayneInternalAPIBaseURL string + WayneServiceName string + WayneServiceAPISecretKey string + OAuthClientID string + OAuthClientSecret string + OAuthRedirectURI string + OAuthCodeTTLSeconds int + OIDCIssuer string + OIDCAuthorizeURL string + OIDCTokenURL string + OIDCUserInfoURL string + OIDCJWKSURL string + CloudDMClientID string + CloudDMClientSecret string + CloudDMRedirectURI string + CloudDMTargetURL string + AnsibleServiceBaseURL string + AnsibleInternalToken string + DeploymentCallbackBaseURL string } func Load() Config { @@ -69,49 +72,52 @@ func Load() Config { oidcIssuer = strings.TrimRight(oidcIssuer, "/") return Config{ - AppEnv: env("APP_ENV", "dev"), - HTTPAddr: httpAddr, - PublicBaseURL: publicBaseURL, - MySQLDSN: env("MYSQL_DSN", "auth:auth@tcp(127.0.0.1:3306)/authserver?charset=utf8mb4&parseTime=True&loc=Local"), - AutoMigrate: envBool("AUTO_MIGRATE", true), - SSOEnabled: envBool("SSO_ENABLED", true), - JWTSecret: env("JWT_SECRET", "change-this-secret"), - JWTIssuer: env("JWT_ISSUER", "authserver"), - JWTTTLMinutes: envInt("JWT_TTL_MINUTES", 120), - SAMLEntityID: samlEntityID, - SAMLACSURL: samlACSURL, - SAMLSPCert: env("SAML_SP_CERT_FILE", "certs/sp.crt"), - SAMLSPKey: env("SAML_SP_KEY_FILE", "certs/sp.key"), - SAMLIDPMetaURL: env("SAML_IDP_METADATA_URL", "http://sso-internal.dev.qiniu.io/saml2/meta"), - SAMLLogoutURL: trimURL(env("SAML_LOGOUT_URL", "")), - WayenLoginURL: env("WAYEN_LOGIN_URL", ""), - WayenTargetURL: env("WAYEN_TARGET_URL", ""), - WayenUsernameKey: env("WAYEN_USERNAME_KEY", "email"), - WayenPasswordKey: env("WAYEN_PASSWORD_KEY", "password"), - WayenLoginFormat: env("WAYEN_LOGIN_FORMAT", "form"), - WayenLoginValue: env("WAYEN_LOGIN_VALUE", "email"), - WayenOAuthRef: env("WAYEN_OAUTH_REF", "/portal/namespace/1/app"), - WayenOAuthLoginURL: trimURL(env("WAYEN_OAUTH_LOGIN_URL", "")), - WayneAPIBaseURL: trimURL(env("WAYNE_API_BASE_URL", env("WAYNE_INTERNAL_API_BASE_URL", ""))), - WayneAdminUsername: env("WAYNE_ADMIN_USERNAME", ""), - WayneAdminPassword: env("WAYNE_ADMIN_PASSWORD", ""), - WayneTokenTTLMinutes: envInt("WAYNE_TOKEN_TTL_MINUTES", 1440), - WayneInternalAPIBaseURL: trimURL(env("WAYNE_INTERNAL_API_BASE_URL", "")), - WayneServiceName: env("WAYNE_SERVICE_NAME", "xinfra"), - WayneServiceAPISecretKey: env("WAYNE_SERVICE_API_SECRET_KEY", ""), - OAuthClientID: env("OAUTH_WAYNE_CLIENT_ID", "wayne"), - OAuthClientSecret: env("OAUTH_WAYNE_CLIENT_SECRET", "wayne-secret"), - OAuthRedirectURI: env("OAUTH_WAYNE_REDIRECT_URI", ""), - OAuthCodeTTLSeconds: envInt("OAUTH_CODE_TTL_SECONDS", 120), - OIDCIssuer: oidcIssuer, - OIDCAuthorizeURL: trimURL(env("OIDC_AUTHORIZATION_ENDPOINT", oidcIssuer+"/oauth/authorize")), - OIDCTokenURL: trimURL(env("OIDC_TOKEN_ENDPOINT", oidcIssuer+"/oauth/token")), - OIDCUserInfoURL: trimURL(env("OIDC_USERINFO_ENDPOINT", oidcIssuer+"/oauth/userinfo")), - OIDCJWKSURL: trimURL(env("OIDC_JWKS_URI", oidcIssuer+"/oauth/jwks")), - CloudDMClientID: env("OIDC_CLOUDDM_CLIENT_ID", "clouddm"), - CloudDMClientSecret: env("OIDC_CLOUDDM_CLIENT_SECRET", ""), - CloudDMRedirectURI: env("OIDC_CLOUDDM_REDIRECT_URI", ""), - CloudDMTargetURL: env("CLOUDDM_TARGET_URL", ""), + AppEnv: env("APP_ENV", "dev"), + HTTPAddr: httpAddr, + PublicBaseURL: publicBaseURL, + MySQLDSN: env("MYSQL_DSN", "auth:auth@tcp(127.0.0.1:3306)/authserver?charset=utf8mb4&parseTime=True&loc=Local"), + AutoMigrate: envBool("AUTO_MIGRATE", true), + SSOEnabled: envBool("SSO_ENABLED", true), + JWTSecret: env("JWT_SECRET", "change-this-secret"), + JWTIssuer: env("JWT_ISSUER", "authserver"), + JWTTTLMinutes: envInt("JWT_TTL_MINUTES", 120), + SAMLEntityID: samlEntityID, + SAMLACSURL: samlACSURL, + SAMLSPCert: env("SAML_SP_CERT_FILE", "certs/sp.crt"), + SAMLSPKey: env("SAML_SP_KEY_FILE", "certs/sp.key"), + SAMLIDPMetaURL: env("SAML_IDP_METADATA_URL", "http://sso-internal.dev.qiniu.io/saml2/meta"), + SAMLLogoutURL: trimURL(env("SAML_LOGOUT_URL", "")), + WayenLoginURL: env("WAYEN_LOGIN_URL", ""), + WayenTargetURL: env("WAYEN_TARGET_URL", ""), + WayenUsernameKey: env("WAYEN_USERNAME_KEY", "email"), + WayenPasswordKey: env("WAYEN_PASSWORD_KEY", "password"), + WayenLoginFormat: env("WAYEN_LOGIN_FORMAT", "form"), + WayenLoginValue: env("WAYEN_LOGIN_VALUE", "email"), + WayenOAuthRef: env("WAYEN_OAUTH_REF", "/portal/namespace/1/app"), + WayenOAuthLoginURL: trimURL(env("WAYEN_OAUTH_LOGIN_URL", "")), + WayneAPIBaseURL: trimURL(env("WAYNE_API_BASE_URL", env("WAYNE_INTERNAL_API_BASE_URL", ""))), + WayneAdminUsername: env("WAYNE_ADMIN_USERNAME", ""), + WayneAdminPassword: env("WAYNE_ADMIN_PASSWORD", ""), + WayneTokenTTLMinutes: envInt("WAYNE_TOKEN_TTL_MINUTES", 1440), + WayneInternalAPIBaseURL: trimURL(env("WAYNE_INTERNAL_API_BASE_URL", "")), + WayneServiceName: env("WAYNE_SERVICE_NAME", "xinfra"), + WayneServiceAPISecretKey: env("WAYNE_SERVICE_API_SECRET_KEY", ""), + OAuthClientID: env("OAUTH_WAYNE_CLIENT_ID", "wayne"), + OAuthClientSecret: env("OAUTH_WAYNE_CLIENT_SECRET", "wayne-secret"), + OAuthRedirectURI: env("OAUTH_WAYNE_REDIRECT_URI", ""), + OAuthCodeTTLSeconds: envInt("OAUTH_CODE_TTL_SECONDS", 120), + OIDCIssuer: oidcIssuer, + OIDCAuthorizeURL: trimURL(env("OIDC_AUTHORIZATION_ENDPOINT", oidcIssuer+"/oauth/authorize")), + OIDCTokenURL: trimURL(env("OIDC_TOKEN_ENDPOINT", oidcIssuer+"/oauth/token")), + OIDCUserInfoURL: trimURL(env("OIDC_USERINFO_ENDPOINT", oidcIssuer+"/oauth/userinfo")), + OIDCJWKSURL: trimURL(env("OIDC_JWKS_URI", oidcIssuer+"/oauth/jwks")), + CloudDMClientID: env("OIDC_CLOUDDM_CLIENT_ID", "clouddm"), + CloudDMClientSecret: env("OIDC_CLOUDDM_CLIENT_SECRET", ""), + CloudDMRedirectURI: env("OIDC_CLOUDDM_REDIRECT_URI", ""), + CloudDMTargetURL: env("CLOUDDM_TARGET_URL", ""), + AnsibleServiceBaseURL: trimURL(env("ANSIBLE_SERVICE_BASE_URL", "")), + AnsibleInternalToken: env("ANSIBLE_INTERNAL_TOKEN", ""), + DeploymentCallbackBaseURL: trimURL(env("DEPLOYMENT_CALLBACK_BASE_URL", publicBaseURL)), } } diff --git a/server/internal/database/database.go b/server/internal/database/database.go index 1a25dfc..29829cf 100644 --- a/server/internal/database/database.go +++ b/server/internal/database/database.go @@ -20,6 +20,8 @@ func AutoMigrate(db *gorm.DB) error { &model.BusinessLineWayneNamespace{}, &model.AccessToken{}, &model.WayneToken{}, + &model.Deployment{}, + &model.DeploymentEvent{}, &model.AuditLog{}, ) } diff --git a/server/internal/handler/context.go b/server/internal/handler/context.go index 7af0641..d138dda 100644 --- a/server/internal/handler/context.go +++ b/server/internal/handler/context.go @@ -15,11 +15,17 @@ const ClaimsKey = "claims" func AuthMiddleware(cfg config.Config) gin.HandlerFunc { return func(c *gin.Context) { value := c.GetHeader("Authorization") - if !strings.HasPrefix(value, "Bearer ") { + tokenValue := "" + if strings.HasPrefix(value, "Bearer ") { + tokenValue = strings.TrimPrefix(value, "Bearer ") + } else { + tokenValue = strings.TrimSpace(c.Query("access_token")) + } + if tokenValue == "" { c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing bearer token"}) return } - claims, err := auth.Parse(cfg.JWTSecret, strings.TrimPrefix(value, "Bearer ")) + claims, err := auth.Parse(cfg.JWTSecret, tokenValue) if err != nil { c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid token"}) return diff --git a/server/internal/handler/deployment.go b/server/internal/handler/deployment.go new file mode 100644 index 0000000..b2a535b --- /dev/null +++ b/server/internal/handler/deployment.go @@ -0,0 +1,249 @@ +package handler + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + + "github.com/1024XEngineer/xinfra/server/internal/config" + "github.com/1024XEngineer/xinfra/server/internal/model" + "github.com/1024XEngineer/xinfra/server/internal/service" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +type DeploymentHandler struct { + cfg config.Config + db *gorm.DB + deployments *service.DeploymentService +} + +func NewDeploymentHandler(cfg config.Config, db *gorm.DB, deployments *service.DeploymentService) *DeploymentHandler { + return &DeploymentHandler{cfg: cfg, db: db, deployments: deployments} +} + +func (h *DeploymentHandler) Create(c *gin.Context) { + claims, ok := CurrentClaims(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "missing auth claims"}) + return + } + var req service.DeploymentCreateRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if !h.ensureBusinessLineMember(c, req.BusinessLineID, claims.UserID, claims.IsAdmin) { + return + } + deployment, err := h.deployments.Create(c.Request.Context(), req, claims.UserID, claims.Username) + if err != nil { + status := http.StatusBadGateway + if errors.Is(err, service.ErrDeploymentNotConfigured) { + status = http.StatusServiceUnavailable + } + c.JSON(status, gin.H{ + "error": err.Error(), + "deployment_id": deployment.DeploymentID, + "status": deployment.Status, + }) + return + } + c.JSON(http.StatusAccepted, gin.H{"deployment_id": deployment.DeploymentID, "status": deployment.Status}) +} + +func (h *DeploymentHandler) Get(c *gin.Context) { + deployment, ok := h.getAuthorizedDeployment(c) + if !ok { + return + } + events, err := h.deployments.Events(c.Request.Context(), deployment.DeploymentID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"deployment": deployment, "events": events}) +} + +func (h *DeploymentHandler) Events(c *gin.Context) { + deployment, ok := h.getAuthorizedDeployment(c) + if !ok { + return + } + events, err := h.deployments.Events(c.Request.Context(), deployment.DeploymentID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + w := c.Writer + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + for _, event := range events { + if err := writeSSE(w, event.Type, service.DeploymentEventData(event)); err != nil { + return + } + } + if isTerminalDeploymentStatus(deployment.Status) { + _ = writeSSE(w, service.DeploymentEventDone, gin.H{"status": deployment.Status}) + return + } + + ch, unsubscribe := h.deployments.Subscribe(deployment.DeploymentID) + defer unsubscribe() + flusher, _ := w.(http.Flusher) + if flusher != nil { + flusher.Flush() + } + for { + select { + case <-c.Request.Context().Done(): + return + case event := <-ch: + if err := writeSSE(w, event.Event.Type, event.Data); err != nil { + return + } + if event.Event.Type == service.DeploymentEventDone { + return + } + } + } +} + +func (h *DeploymentHandler) Cancel(c *gin.Context) { + deployment, ok := h.getAuthorizedDeployment(c) + if !ok { + return + } + if err := h.deployments.Cancel(c.Request.Context(), deployment.DeploymentID); err != nil { + writeDeploymentError(c, err) + return + } + c.JSON(http.StatusAccepted, gin.H{"deployment_id": deployment.DeploymentID, "status": service.DeploymentStatusCanceling}) +} + +func (h *DeploymentHandler) InternalEvent(c *gin.Context) { + if !h.authorizeInternal(c) { + return + } + deploymentID := strings.TrimSpace(c.Param("id")) + var req service.DeploymentEventRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + event, err := h.deployments.AppendEvent(c.Request.Context(), deploymentID, req) + if err != nil { + writeDeploymentError(c, err) + return + } + c.JSON(http.StatusAccepted, gin.H{"event_id": event.ID, "seq": event.Seq}) +} + +func (h *DeploymentHandler) InternalFinish(c *gin.Context) { + if !h.authorizeInternal(c) { + return + } + deploymentID := strings.TrimSpace(c.Param("id")) + var req service.DeploymentFinishRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + deployment, err := h.deployments.Finish(c.Request.Context(), deploymentID, req) + if err != nil { + writeDeploymentError(c, err) + return + } + c.JSON(http.StatusAccepted, gin.H{"deployment_id": deployment.DeploymentID, "status": deployment.Status}) +} + +func (h *DeploymentHandler) getAuthorizedDeployment(c *gin.Context) (model.Deployment, bool) { + claims, ok := CurrentClaims(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "missing auth claims"}) + return model.Deployment{}, false + } + deploymentID := strings.TrimSpace(c.Param("id")) + deployment, err := h.deployments.Get(c.Request.Context(), deploymentID) + if err != nil { + writeDeploymentError(c, err) + return model.Deployment{}, false + } + if !h.ensureBusinessLineMember(c, deployment.BusinessLineID, claims.UserID, claims.IsAdmin) { + return model.Deployment{}, false + } + return deployment, true +} + +func (h *DeploymentHandler) ensureBusinessLineMember(c *gin.Context, businessLineID uint64, userID uint64, isAdmin bool) bool { + if businessLineID == 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "business_line_id is required"}) + return false + } + if isAdmin { + return true + } + var binding model.BusinessLineUser + err := h.db.WithContext(c.Request.Context()).Where("business_line_id = ? AND user_id = ?", businessLineID, userID).First(&binding).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusForbidden, gin.H{"error": "current user is not assigned to this business line"}) + return false + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return false + } + return true +} + +func (h *DeploymentHandler) authorizeInternal(c *gin.Context) bool { + if h.cfg.AnsibleInternalToken == "" { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "ansible internal token is not configured"}) + return false + } + value := c.GetHeader("Authorization") + if strings.TrimPrefix(value, "Bearer ") != h.cfg.AnsibleInternalToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid internal token"}) + return false + } + return true +} + +func writeSSE(w gin.ResponseWriter, event string, data any) error { + body, err := json.Marshal(data) + if err != nil { + return err + } + if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, body); err != nil { + return err + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + return nil +} + +func writeDeploymentError(c *gin.Context, err error) { + switch { + case errors.Is(err, service.ErrDeploymentNotFound): + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + case errors.Is(err, service.ErrDeploymentForbidden): + c.JSON(http.StatusForbidden, gin.H{"error": err.Error()}) + case errors.Is(err, service.ErrDeploymentInvalidState): + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + case errors.Is(err, service.ErrDeploymentNotConfigured): + c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()}) + default: + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + } +} + +func isTerminalDeploymentStatus(status string) bool { + return status == service.DeploymentStatusSuccess || status == service.DeploymentStatusFailed || status == service.DeploymentStatusCanceled +} diff --git a/server/internal/model/models.go b/server/internal/model/models.go index 1fbcbc6..2113125 100644 --- a/server/internal/model/models.go +++ b/server/internal/model/models.go @@ -80,6 +80,34 @@ type WayneToken struct { UpdatedAt time.Time `json:"updated_at"` } +type Deployment struct { + ID uint64 `gorm:"primaryKey" json:"id"` + DeploymentID string `gorm:"size:64;not null;uniqueIndex" json:"deployment_id"` + Component string `gorm:"size:64;not null;index" json:"component"` + BusinessLineID uint64 `gorm:"not null;index" json:"business_line_id"` + BusinessLine string `gorm:"size:128;not null;default:''" json:"business_line"` + Status string `gorm:"size:32;not null;index" json:"status"` + RequestPayload string `gorm:"type:longtext" json:"request_payload"` + ResultPayload string `gorm:"type:longtext" json:"result_payload"` + CreatedBy uint64 `gorm:"not null;index" json:"created_by"` + CreatedByName string `gorm:"size:128;not null;default:''" json:"created_by_name"` + StartedAt *time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at"` + CreatedAt time.Time `gorm:"index" json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type DeploymentEvent struct { + ID uint64 `gorm:"primaryKey" json:"id"` + DeploymentID string `gorm:"size:64;not null;index:idx_deployment_events_deployment_seq,priority:1" json:"deployment_id"` + Seq uint64 `gorm:"not null;index:idx_deployment_events_deployment_seq,priority:2" json:"seq"` + Type string `gorm:"size:32;not null;index" json:"type"` + Level string `gorm:"size:32;not null;default:''" json:"level"` + Message string `gorm:"type:longtext" json:"message"` + Payload string `gorm:"type:longtext" json:"payload"` + CreatedAt time.Time `gorm:"index" json:"created_at"` +} + type AuditLog struct { ID uint64 `gorm:"primaryKey" json:"id"` RequestID string `gorm:"size:128;not null;default:''" json:"request_id"` diff --git a/server/internal/router/router.go b/server/internal/router/router.go index 8141566..f4e1bae 100644 --- a/server/internal/router/router.go +++ b/server/internal/router/router.go @@ -70,6 +70,7 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { authService := service.NewAuthService(deps.Config, deps.DB, auditService) wayenService := service.NewWayenService(deps.Config, deps.DB) wayneRoleBindingService := service.NewWayneRoleBindingService(deps.Config, deps.DB) + deploymentService := service.NewDeploymentService(deps.Config, deps.DB) healthHandler := handler.NewHealthHandler(deps.DB) authHandler := handler.NewAuthHandler(deps.Config, authService) @@ -81,6 +82,7 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { clouddmHandler := handler.NewCloudDMHandler(deps.Config, auditService) samlHandler := handler.NewSAMLHandler(deps.Config, authService) oauthHandler := handler.NewOAuthHandler(deps.Config, deps.DB, auditService) + deploymentHandler := handler.NewDeploymentHandler(deps.Config, deps.DB, deploymentService) r.GET("/healthz", healthHandler.Healthz) r.GET("/readyz", healthHandler.Readyz) @@ -89,6 +91,8 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { r.POST("/auth/oauth/token", oauthHandler.Token) r.GET("/auth/oauth/jwks", oauthHandler.JWKS) r.GET("/auth/oauth/userinfo", oauthHandler.UserInfo) + r.POST("/auth/internal/deployments/:id/events", deploymentHandler.InternalEvent) + r.POST("/auth/internal/deployments/:id/finish", deploymentHandler.InternalFinish) v1 := r.Group("/auth/api/v1") { @@ -130,6 +134,10 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { protected.PUT("/subsystem-auth/wayne/business-lines/:id/namespaces/:namespaceid/users/:username/roles", subsystemAuthHandler.BindWayneNamespaceRoles) protected.DELETE("/subsystem-auth/wayne/business-lines/:id/namespaces/:namespaceid/users/:username/roles", subsystemAuthHandler.UnbindWayneNamespaceRoles) protected.POST("/subsystem-auth/wayne/business-lines/:id/users/:userid/init", subsystemAuthHandler.InitWayneBusinessLineUser) + protected.POST("/deployments", deploymentHandler.Create) + protected.GET("/deployments/:id", deploymentHandler.Get) + protected.GET("/deployments/:id/events", deploymentHandler.Events) + protected.POST("/deployments/:id/cancel", deploymentHandler.Cancel) protected.GET("/clouddm/login", clouddmHandler.Login) } } diff --git a/server/internal/service/deployment.go b/server/internal/service/deployment.go new file mode 100644 index 0000000..8abee34 --- /dev/null +++ b/server/internal/service/deployment.go @@ -0,0 +1,499 @@ +package service + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "sort" + "strings" + "sync" + "time" + + "github.com/1024XEngineer/xinfra/server/internal/config" + "github.com/1024XEngineer/xinfra/server/internal/model" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +const ( + DeploymentStatusPending = "pending" + DeploymentStatusRunning = "running" + DeploymentStatusSuccess = "success" + DeploymentStatusFailed = "failed" + DeploymentStatusCanceling = "canceling" + DeploymentStatusCanceled = "canceled" + + DeploymentEventLog = "log" + DeploymentEventStatus = "status" + DeploymentEventResult = "result" + DeploymentEventDone = "done" + DeploymentEventError = "error" +) + +var ( + ErrDeploymentNotConfigured = errors.New("ansible deployment service is not configured") + ErrDeploymentNotFound = errors.New("deployment not found") + ErrDeploymentForbidden = errors.New("deployment permission denied") + ErrDeploymentInvalidState = errors.New("deployment state does not allow this operation") +) + +type DeploymentCreateRequest struct { + Component string `json:"component"` + BusinessLineID uint64 `json:"business_line_id"` + Params map[string]any `json:"params"` +} + +type DeploymentEventRequest struct { + Type string `json:"type"` + Level string `json:"level"` + Status string `json:"status"` + Seq uint64 `json:"seq"` + Message string `json:"message"` + Payload map[string]any `json:"payload"` +} + +type DeploymentFinishRequest struct { + Status string `json:"status"` + ExitCode int `json:"exit_code"` + Error string `json:"error"` + Summary map[string]any `json:"summary"` +} + +type DeploymentEventEnvelope struct { + Event model.DeploymentEvent + Data map[string]any +} + +type DeploymentService struct { + cfg config.Config + db *gorm.DB + client *http.Client + + mu sync.Mutex + subscribers map[string]map[chan DeploymentEventEnvelope]struct{} +} + +func NewDeploymentService(cfg config.Config, db *gorm.DB) *DeploymentService { + return &DeploymentService{ + cfg: cfg, + db: db, + client: &http.Client{Timeout: 8 * time.Second}, + subscribers: make(map[string]map[chan DeploymentEventEnvelope]struct{}), + } +} + +func (s *DeploymentService) Create(ctx context.Context, req DeploymentCreateRequest, actorID uint64, actorName string) (model.Deployment, error) { + component := strings.TrimSpace(strings.ToLower(req.Component)) + if !isSupportedDeploymentComponent(component) { + return model.Deployment{}, fmt.Errorf("unsupported deployment component: %s", req.Component) + } + if req.BusinessLineID == 0 { + return model.Deployment{}, errors.New("business_line_id is required") + } + payload, err := json.Marshal(req) + if err != nil { + return model.Deployment{}, err + } + + var businessLine model.BusinessLine + if err := s.db.WithContext(ctx).First(&businessLine, req.BusinessLineID).Error; err != nil { + return model.Deployment{}, err + } + + deployment := model.Deployment{ + DeploymentID: newDeploymentID(), + Component: component, + BusinessLineID: req.BusinessLineID, + BusinessLine: businessLine.Name, + Status: DeploymentStatusPending, + RequestPayload: string(payload), + CreatedBy: actorID, + CreatedByName: actorName, + } + + if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(&deployment).Error; err != nil { + return err + } + _, err := s.appendEventTx(ctx, tx, deployment.DeploymentID, DeploymentEventStatus, "info", "deployment created", map[string]any{"status": DeploymentStatusPending}) + return err + }); err != nil { + return model.Deployment{}, err + } + + if err := s.startPythonDeployment(ctx, deployment, req.Params); err != nil { + _ = s.FailStart(ctx, deployment.DeploymentID, err) + return deployment, err + } + return deployment, nil +} + +func (s *DeploymentService) Get(ctx context.Context, deploymentID string) (model.Deployment, error) { + var deployment model.Deployment + if err := s.db.WithContext(ctx).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return deployment, ErrDeploymentNotFound + } + return deployment, err + } + return deployment, nil +} + +func (s *DeploymentService) Events(ctx context.Context, deploymentID string) ([]model.DeploymentEvent, error) { + var events []model.DeploymentEvent + err := s.db.WithContext(ctx).Where("deployment_id = ?", deploymentID).Order("seq ASC").Find(&events).Error + return events, err +} + +func (s *DeploymentService) AppendEvent(ctx context.Context, deploymentID string, req DeploymentEventRequest) (model.DeploymentEvent, error) { + eventType := normalizeDeploymentEventType(req.Type) + level := strings.TrimSpace(req.Level) + if level == "" { + level = "info" + } + payload := req.Payload + if payload == nil { + payload = map[string]any{} + } + if req.Status != "" { + payload["status"] = normalizeDeploymentStatus(req.Status) + } + event, err := s.appendEvent(ctx, deploymentID, eventType, level, req.Message, payload) + if err != nil { + return event, err + } + if status, _ := payload["status"].(string); status != "" { + _ = s.updateStatus(ctx, deploymentID, status, "") + } + return event, nil +} + +func (s *DeploymentService) Finish(ctx context.Context, deploymentID string, req DeploymentFinishRequest) (model.Deployment, error) { + status := normalizeDeploymentStatus(req.Status) + if status == "" { + status = DeploymentStatusFailed + } + payload := map[string]any{ + "status": status, + "exit_code": req.ExitCode, + "summary": req.Summary, + } + if req.Error != "" { + payload["error"] = req.Error + } + result, _ := json.Marshal(payload) + now := time.Now() + var deployment model.Deployment + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return ErrDeploymentNotFound + } + return err + } + updates := map[string]any{"status": status, "result_payload": string(result), "finished_at": &now} + if err := tx.Model(&model.Deployment{}).Where("deployment_id = ?", deploymentID).Updates(updates).Error; err != nil { + return err + } + eventType := DeploymentEventResult + if status == DeploymentStatusFailed || status == DeploymentStatusCanceled { + eventType = DeploymentEventError + } + if _, err := s.appendEventTx(ctx, tx, deploymentID, eventType, eventLevelForStatus(status), finishMessage(status, req.Error), payload); err != nil { + return err + } + _, err := s.appendEventTx(ctx, tx, deploymentID, DeploymentEventDone, eventLevelForStatus(status), status, map[string]any{"status": status}) + return err + }) + if err != nil { + return deployment, err + } + deployment.Status = status + deployment.ResultPayload = string(result) + deployment.FinishedAt = &now + return deployment, nil +} + +func (s *DeploymentService) Cancel(ctx context.Context, deploymentID string) error { + deployment, err := s.Get(ctx, deploymentID) + if err != nil { + return err + } + if !deploymentCancelable(deployment.Status) { + return ErrDeploymentInvalidState + } + if err := s.updateStatus(ctx, deploymentID, DeploymentStatusCanceling, "cancel requested"); err != nil { + return err + } + if s.cfg.AnsibleServiceBaseURL == "" { + return ErrDeploymentNotConfigured + } + body, _ := json.Marshal(map[string]any{"deployment_id": deploymentID}) + path := s.cfg.AnsibleServiceBaseURL + "/internal/ansible/deployments/" + deploymentID + "/cancel" + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, path, bytes.NewReader(body)) + if err != nil { + return err + } + httpReq.Header.Set("Content-Type", "application/json") + if s.cfg.AnsibleInternalToken != "" { + httpReq.Header.Set("Authorization", "Bearer "+s.cfg.AnsibleInternalToken) + } + resp, err := s.client.Do(httpReq) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + data, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) + return fmt.Errorf("ansible cancel failed: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data))) + } + return nil +} + +func (s *DeploymentService) Subscribe(deploymentID string) (chan DeploymentEventEnvelope, func()) { + ch := make(chan DeploymentEventEnvelope, 64) + s.mu.Lock() + if s.subscribers[deploymentID] == nil { + s.subscribers[deploymentID] = make(map[chan DeploymentEventEnvelope]struct{}) + } + s.subscribers[deploymentID][ch] = struct{}{} + s.mu.Unlock() + return ch, func() { + s.mu.Lock() + if subscribers := s.subscribers[deploymentID]; subscribers != nil { + delete(subscribers, ch) + if len(subscribers) == 0 { + delete(s.subscribers, deploymentID) + } + } + s.mu.Unlock() + close(ch) + } +} + +func (s *DeploymentService) FailStart(ctx context.Context, deploymentID string, cause error) error { + _, err := s.Finish(ctx, deploymentID, DeploymentFinishRequest{ + Status: DeploymentStatusFailed, + Error: cause.Error(), + }) + return err +} + +func (s *DeploymentService) startPythonDeployment(ctx context.Context, deployment model.Deployment, params map[string]any) error { + if s.cfg.AnsibleServiceBaseURL == "" { + return ErrDeploymentNotConfigured + } + callbackBaseURL := strings.TrimRight(s.cfg.DeploymentCallbackBaseURL, "/") + body, err := json.Marshal(map[string]any{ + "deployment_id": deployment.DeploymentID, + "component": deployment.Component, + "callback_url": callbackBaseURL + "/auth/internal/deployments/" + deployment.DeploymentID + "/events", + "finish_url": callbackBaseURL + "/auth/internal/deployments/" + deployment.DeploymentID + "/finish", + "params": params, + }) + if err != nil { + return err + } + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.AnsibleServiceBaseURL+"/internal/ansible/deploy", bytes.NewReader(body)) + if err != nil { + return err + } + httpReq.Header.Set("Content-Type", "application/json") + if s.cfg.AnsibleInternalToken != "" { + httpReq.Header.Set("Authorization", "Bearer "+s.cfg.AnsibleInternalToken) + } + resp, err := s.client.Do(httpReq) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + data, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) + return fmt.Errorf("ansible deploy failed: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data))) + } + return nil +} + +func (s *DeploymentService) appendEvent(ctx context.Context, deploymentID, eventType, level, message string, payload map[string]any) (model.DeploymentEvent, error) { + var event model.DeploymentEvent + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var err error + event, err = s.appendEventTx(ctx, tx, deploymentID, eventType, level, message, payload) + return err + }) + return event, err +} + +func (s *DeploymentService) appendEventTx(ctx context.Context, tx *gorm.DB, deploymentID, eventType, level, message string, payload map[string]any) (model.DeploymentEvent, error) { + var deployment model.Deployment + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("deployment_id = ?", deploymentID).First(&deployment).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return model.DeploymentEvent{}, ErrDeploymentNotFound + } + return model.DeploymentEvent{}, err + } + var maxSeq uint64 + if err := tx.Model(&model.DeploymentEvent{}).Where("deployment_id = ?", deploymentID).Select("COALESCE(MAX(seq), 0)").Scan(&maxSeq).Error; err != nil { + return model.DeploymentEvent{}, err + } + payloadBody, _ := json.Marshal(payload) + event := model.DeploymentEvent{ + DeploymentID: deploymentID, + Seq: maxSeq + 1, + Type: eventType, + Level: level, + Message: message, + Payload: string(payloadBody), + } + if err := tx.Create(&event).Error; err != nil { + return model.DeploymentEvent{}, err + } + go s.broadcast(event, payload) + return event, nil +} + +func (s *DeploymentService) updateStatus(ctx context.Context, deploymentID string, status string, message string) error { + status = normalizeDeploymentStatus(status) + if status == "" { + return nil + } + now := time.Now() + updates := map[string]any{"status": status} + if status == DeploymentStatusRunning { + updates["started_at"] = &now + } + if isDeploymentTerminal(status) { + updates["finished_at"] = &now + } + if err := s.db.WithContext(ctx).Model(&model.Deployment{}).Where("deployment_id = ?", deploymentID).Updates(updates).Error; err != nil { + return err + } + if message != "" { + _, err := s.appendEvent(ctx, deploymentID, DeploymentEventStatus, eventLevelForStatus(status), message, map[string]any{"status": status}) + return err + } + return nil +} + +func (s *DeploymentService) broadcast(event model.DeploymentEvent, data map[string]any) { + envelope := DeploymentEventEnvelope{Event: event, Data: eventData(event, data)} + s.mu.Lock() + subscribers := make([]chan DeploymentEventEnvelope, 0, len(s.subscribers[event.DeploymentID])) + for ch := range s.subscribers[event.DeploymentID] { + subscribers = append(subscribers, ch) + } + s.mu.Unlock() + for _, ch := range subscribers { + select { + case ch <- envelope: + default: + } + } +} + +func eventData(event model.DeploymentEvent, data map[string]any) map[string]any { + out := map[string]any{ + "deployment_id": event.DeploymentID, + "seq": event.Seq, + "type": event.Type, + "level": event.Level, + "message": event.Message, + "created_at": event.CreatedAt, + } + for key, value := range data { + out[key] = value + } + return out +} + +func DeploymentEventData(event model.DeploymentEvent) map[string]any { + payload := map[string]any{} + if strings.TrimSpace(event.Payload) != "" { + _ = json.Unmarshal([]byte(event.Payload), &payload) + } + return eventData(event, payload) +} + +func isSupportedDeploymentComponent(component string) bool { + switch component { + case "mysql", "openresty": + return true + default: + return false + } +} + +func normalizeDeploymentEventType(value string) string { + switch strings.TrimSpace(strings.ToLower(value)) { + case DeploymentEventStatus: + return DeploymentEventStatus + case DeploymentEventResult: + return DeploymentEventResult + case DeploymentEventDone: + return DeploymentEventDone + case DeploymentEventError: + return DeploymentEventError + default: + return DeploymentEventLog + } +} + +func normalizeDeploymentStatus(value string) string { + switch strings.TrimSpace(strings.ToLower(value)) { + case DeploymentStatusPending, DeploymentStatusRunning, DeploymentStatusSuccess, DeploymentStatusFailed, DeploymentStatusCanceling, DeploymentStatusCanceled: + return strings.TrimSpace(strings.ToLower(value)) + default: + return "" + } +} + +func isDeploymentTerminal(status string) bool { + return status == DeploymentStatusSuccess || status == DeploymentStatusFailed || status == DeploymentStatusCanceled +} + +func deploymentCancelable(status string) bool { + return status == DeploymentStatusPending || status == DeploymentStatusRunning || status == DeploymentStatusCanceling +} + +func eventLevelForStatus(status string) string { + if status == DeploymentStatusFailed || status == DeploymentStatusCanceled { + return "error" + } + return "info" +} + +func finishMessage(status, fallback string) string { + if fallback != "" { + return fallback + } + switch status { + case DeploymentStatusSuccess: + return "deployment completed" + case DeploymentStatusCanceled: + return "deployment canceled" + default: + return "deployment failed" + } +} + +func newDeploymentID() string { + now := time.Now() + buf := make([]byte, 3) + if _, err := rand.Read(buf); err != nil { + return fmt.Sprintf("CMP-%s-%d", now.Format("20060102"), now.UnixNano()%1000000) + } + return fmt.Sprintf("CMP-%s-%s", now.Format("20060102"), strings.ToUpper(hex.EncodeToString(buf))) +} + +func SortedDeploymentEvents(events []model.DeploymentEvent) { + sort.Slice(events, func(i, j int) bool { + return events[i].Seq < events[j].Seq + }) +}