diff --git a/server/.env.example b/server/.env.example index 7ce4835..02fe72b 100644 --- a/server/.env.example +++ b/server/.env.example @@ -17,16 +17,16 @@ AUTO_MIGRATE=true # MySQL service delivery (AWX is required when the scheduler is enabled) DELIVERY_SCHEDULER_ENABLED=false -DELIVERY_POLL_SECONDS=5 +DELIVERY_DISPATCH_SECONDS=5 +DELIVERY_CALLBACK_BASE_URL=http://authserver-backend.authserver.svc.cluster.local:8083 DELIVERY_RESERVATION_TTL_MINUTES=120 DELIVERY_GLOBAL_LIMIT=2 DELIVERY_TARGET_LIMIT=2 -DELIVERY_BUSINESS_LIMIT=1 AWX_BASE_URL= AWX_TOKEN= AWX_USERNAME= AWX_PASSWORD= -DELIVERY_SERVICE_TOKEN= +AWX_WEBHOOK_TOKEN= CLOUDDM_REGISTER_URL= CLOUDDM_API_TOKEN= diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 98a10e7..f560712 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -63,13 +63,13 @@ type Config struct { AWXToken string AWXUsername string AWXPassword string - DeliveryServiceToken string + AWXWebhookToken string DeliverySchedulerEnabled bool - DeliveryPollSeconds int + DeliveryDispatchSeconds int + DeliveryCallbackBaseURL string ReservationTTLMinutes int DeliveryGlobalLimit int DeliveryTargetLimit int - DeliveryBusinessLimit int } func Load() Config { @@ -131,13 +131,13 @@ func Load() Config { AWXToken: env("AWX_TOKEN", ""), AWXUsername: env("AWX_USERNAME", ""), AWXPassword: env("AWX_PASSWORD", ""), - DeliveryServiceToken: env("DELIVERY_SERVICE_TOKEN", ""), + AWXWebhookToken: env("AWX_WEBHOOK_TOKEN", ""), DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false), - DeliveryPollSeconds: envInt("DELIVERY_POLL_SECONDS", 5), + DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5), + DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)), ReservationTTLMinutes: envInt("DELIVERY_RESERVATION_TTL_MINUTES", 120), DeliveryGlobalLimit: envInt("DELIVERY_GLOBAL_LIMIT", 2), DeliveryTargetLimit: envInt("DELIVERY_TARGET_LIMIT", 2), - DeliveryBusinessLimit: envInt("DELIVERY_BUSINESS_LIMIT", 1), } } diff --git a/server/internal/handler/delivery.go b/server/internal/handler/delivery.go index f36b948..c08f1d9 100644 --- a/server/internal/handler/delivery.go +++ b/server/internal/handler/delivery.go @@ -19,42 +19,54 @@ func NewDeliveryHandler(s *service.DeliveryService) *DeliveryHandler { return &DeliveryHandler{service: s} } -type ExecutionHandler struct { +type DeliveryCallbackHandler struct { service *service.DeliveryService token string } -func NewExecutionHandler(s *service.DeliveryService, token string) *ExecutionHandler { - return &ExecutionHandler{service: s, token: strings.TrimSpace(token)} +func NewDeliveryCallbackHandler(s *service.DeliveryService, token string) *DeliveryCallbackHandler { + return &DeliveryCallbackHandler{service: s, token: strings.TrimSpace(token)} } -type executionPayload struct { - TaskID string `json:"task_id" binding:"required"` - PayloadHash string `json:"payload_hash" binding:"required"` - IdempotencyKey string `json:"idempotency_key" binding:"required"` -} - -func (h *ExecutionHandler) Create(c *gin.Context) { - provided := strings.TrimPrefix(c.GetHeader("Authorization"), "Bearer ") +func (h *DeliveryCallbackHandler) authorize(c *gin.Context) bool { + provided := strings.TrimSpace(strings.TrimPrefix(c.GetHeader("Authorization"), "Bearer ")) if h.token == "" || subtle.ConstantTimeCompare([]byte(provided), []byte(h.token)) != 1 { c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid service credential"}) + return false + } + return true +} + +func (h *DeliveryCallbackHandler) StageEvent(c *gin.Context) { + if !h.authorize(c) { return } - var req executionPayload + var req service.DeliveryStageEventInput if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - job, replay, err := h.service.CreateExecution(c.Request.Context(), req.TaskID, req.PayloadHash, req.IdempotencyKey) - if err != nil { + if err := h.service.HandleStageEvent(c.Request.Context(), c.Param("id"), req); err != nil { c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) return } - status := http.StatusCreated - if replay { - status = http.StatusOK + c.JSON(http.StatusAccepted, gin.H{"ok": true}) +} + +func (h *DeliveryCallbackHandler) AWXJobEvent(c *gin.Context) { + if !h.authorize(c) { + return } - c.JSON(status, gin.H{"execution": job, "idempotent_replay": replay}) + var req service.AWXJobNotificationInput + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid AWX webhook JSON: " + err.Error()}) + return + } + if err := h.service.HandleAWXJobNotification(c.Request.Context(), req); err != nil { + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusAccepted, gin.H{"ok": true}) } // CreateMySQL 提交 MySQL 交付请求 @@ -114,7 +126,11 @@ func (h *DeliveryHandler) List(c *gin.Context) { if raw := c.Query("business_line_id"); raw != "" { businessLineID, _ = strconv.ParseUint(raw, 10, 64) } - items, err := h.service.ListTasks(c.Request.Context(), claims.UserID, claims.IsAdmin, businessLineID) + items, err := h.service.ListTasks(c.Request.Context(), claims.UserID, claims.IsAdmin, service.DeliveryTaskListFilter{ + BusinessLineID: businessLineID, + Component: c.Query("component"), + ActiveOnly: strings.EqualFold(c.Query("active"), "true"), + }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return diff --git a/server/internal/handler/task_log.go b/server/internal/handler/task_log.go index ee0ad27..e935053 100644 --- a/server/internal/handler/task_log.go +++ b/server/internal/handler/task_log.go @@ -96,7 +96,7 @@ func (h *TaskLogHandler) Get(c *gin.Context) { } func (h *TaskLogHandler) listAWXTasks(c *gin.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]taskLogSummary, error) { - tasks, err := h.delivery.ListTasks(c.Request.Context(), userID, isAdmin, businessLineID) + tasks, err := h.delivery.ListTasks(c.Request.Context(), userID, isAdmin, service.DeliveryTaskListFilter{BusinessLineID: businessLineID}) if err != nil { return nil, err } diff --git a/server/internal/model/delivery.go b/server/internal/model/delivery.go index e30b228..64aef57 100644 --- a/server/internal/model/delivery.go +++ b/server/internal/model/delivery.go @@ -32,6 +32,7 @@ type DeliveryTask struct { ID string `gorm:"size:36;primaryKey" json:"id"` BusinessLineID uint64 `gorm:"not null;index" json:"business_line_id"` RequestedBy uint64 `gorm:"not null;index" json:"requested_by"` + Component string `gorm:"size:32;not null;default:mysql;index" json:"component"` TargetType string `gorm:"size:32;not null" json:"target_type"` TargetID uint64 `gorm:"not null;index" json:"target_id"` Namespace string `gorm:"size:63;not null;index" json:"namespace"` @@ -111,10 +112,12 @@ type ExecutionJob struct { } type TaskEvent struct { - ID uint64 `gorm:"primaryKey" json:"id"` - TaskID string `gorm:"size:36;not null;index" json:"task_id"` - FromState string `gorm:"size:32;not null" json:"from_state"` - ToState string `gorm:"size:32;not null" json:"to_state"` - Message string `gorm:"type:text" json:"message"` - CreatedAt time.Time `gorm:"index" json:"created_at"` + ID uint64 `gorm:"primaryKey" json:"id"` + TaskID string `gorm:"size:36;not null;index" json:"task_id"` + FromState string `gorm:"size:32;not null" json:"from_state"` + ToState string `gorm:"size:32;not null" json:"to_state"` + Stage string `gorm:"size:32;index" json:"stage,omitempty"` + EventStatus string `gorm:"size:32;index" json:"event_status,omitempty"` + Message string `gorm:"type:text" json:"message"` + CreatedAt time.Time `gorm:"index" json:"created_at"` } diff --git a/server/internal/router/router.go b/server/internal/router/router.go index a1bb599..2c0a548 100644 --- a/server/internal/router/router.go +++ b/server/internal/router/router.go @@ -87,13 +87,14 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { samlHandler := handler.NewSAMLHandler(deps.Config, authService) oauthHandler := handler.NewOAuthHandler(deps.Config, deps.DB, auditService) deliveryHandler := handler.NewDeliveryHandler(deliveryService) - executionHandler := handler.NewExecutionHandler(deliveryService, deps.Config.DeliveryServiceToken) + deliveryCallbackHandler := handler.NewDeliveryCallbackHandler(deliveryService, deps.Config.AWXWebhookToken) containerServiceHandler := handler.NewContainerServiceHandler(deps.DB, wayneRoleBindingService) taskLogHandler := handler.NewTaskLogHandler(deps.DB, deliveryService, wayneRoleBindingService) r.GET("/healthz", healthHandler.Healthz) r.GET("/readyz", healthHandler.Readyz) - r.POST("/api/v1/executions", executionHandler.Create) + r.POST("/auth/internal/delivery/tasks/:id/events", deliveryCallbackHandler.StageEvent) + r.POST("/auth/internal/awx/jobs/events", deliveryCallbackHandler.AWXJobEvent) r.GET("/auth/.well-known/openid-configuration", oauthHandler.Discovery) r.GET("/auth/oauth/authorize", oauthHandler.Authorize) r.POST("/auth/oauth/token", oauthHandler.Token) diff --git a/server/internal/service/awx.go b/server/internal/service/awx.go index e719798..bae334d 100644 --- a/server/internal/service/awx.go +++ b/server/internal/service/awx.go @@ -32,16 +32,19 @@ type AWXLaunchRequest struct { } type AWXJob struct { - ID uint64 `json:"id"` - Status string `json:"status"` - Failed bool `json:"failed"` + ID uint64 `json:"id"` + Status string `json:"status"` + Failed bool `json:"failed"` + IgnoredFields map[string]any `json:"ignored_fields"` } type AWXJobTemplate struct { - ID uint64 `json:"id"` - Name string `json:"name"` - Description string `json:"description"` - Inventory uint64 `json:"inventory"` + ID uint64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Inventory uint64 `json:"inventory"` + AskVariablesOnLaunch bool `json:"ask_variables_on_launch"` + AskLimitOnLaunch bool `json:"ask_limit_on_launch"` } type AWXInventoryHost struct { @@ -92,12 +95,23 @@ func (c *AWXClient) Launch(ctx context.Context, templateID uint64, input AWXLaun if err := c.request(ctx, http.MethodPost, fmt.Sprintf("/api/v2/job_templates/%d/launch/", templateID), body, &job); err != nil { return nil, err } + if len(job.IgnoredFields) > 0 { + return nil, fmt.Errorf("AWX ignored launch fields %v; enable Prompt on launch for Variables and Limit on the job template", ignoredFieldNames(job.IgnoredFields)) + } if job.ID == 0 { return nil, fmt.Errorf("AWX launch response did not include a job id") } return &job, nil } +func ignoredFieldNames(fields map[string]any) []string { + names := make([]string, 0, len(fields)) + for name := range fields { + names = append(names, name) + } + return names +} + func (c *AWXClient) GetJob(ctx context.Context, jobID string) (*AWXJob, error) { if _, err := strconv.ParseUint(jobID, 10, 64); err != nil { return nil, fmt.Errorf("invalid AWX job id %q", jobID) diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 1186212..c60086f 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "log" "net" "net/http" "regexp" @@ -25,14 +26,33 @@ import ( var dnsLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`) type MySQLDeliveryInput struct { - BusinessLineID uint64 `json:"business_line_id" binding:"required"` - TargetID uint64 `json:"target_id" binding:"required"` - Namespace string `json:"namespace" binding:"required"` - InstanceName string `json:"instance_name" binding:"required"` - MySQLVersion string `json:"mysql_version"` - CPUMilli int64 `json:"cpu_milli" binding:"required"` - MemoryMi int64 `json:"memory_mi" binding:"required"` - StorageGi int64 `json:"storage_gi" binding:"required"` + BusinessLineID uint64 `json:"business_line_id" binding:"required"` + TargetID uint64 `json:"target_id" binding:"required"` + Namespace string `json:"namespace" binding:"required"` + InstanceName string `json:"instance_name" binding:"required"` + MySQLVersion string `json:"mysql_version"` + Topology string `json:"topology"` + MySQLPort int `json:"mysql_port"` + DataDisk string `json:"data_disk"` + CPUCores int64 `json:"cpu_cores" binding:"required"` + MemoryGB int64 `json:"memory_gb" binding:"required"` + StorageGB int64 `json:"storage_gb" binding:"required"` + ParamTemplate string `json:"param_template"` + TimeZone string `json:"timezone"` + LowerCaseTableNames int `json:"lower_case_table_names"` + CharacterSet string `json:"character_set"` + Collation string `json:"collation"` + MaxConnections string `json:"max_connections"` + InnoDBRedoLogCapacity string `json:"innodb_redo_log_capacity"` + InnoDBFlushLogAtTrxCommit int `json:"innodb_flush_log_at_trx_commit"` + SyncBinlog int `json:"sync_binlog"` + InnoDBIOCapacity int `json:"innodb_io_capacity"` + LongQueryTime float64 `json:"long_query_time"` + BinlogExpireLogsSeconds int64 `json:"binlog_expire_logs_seconds"` + MaxBinlogSize string `json:"max_binlog_size"` + CPUMilli int64 `json:"-"` + MemoryMi int64 `json:"-"` + StorageGi int64 `json:"-"` } type deliveryPayload struct { @@ -50,6 +70,28 @@ type DeliveryTarget struct { Metadata string `json:"metadata"` } +type DeliveryStageEventInput struct { + Stage string `json:"stage" binding:"required"` + Status string `json:"status" binding:"required"` + Message string `json:"message"` + AWXJobID string `json:"awx_job_id"` +} + +type AWXJobNotificationInput struct { + ID uint64 `json:"id"` + Status string `json:"status"` + Name string `json:"name"` + URL string `json:"url"` + Traceback string `json:"traceback"` + ExtraVars any `json:"extra_vars"` +} + +type DeliveryTaskListFilter struct { + BusinessLineID uint64 + Component string + ActiveOnly bool +} + // targetMetadata describes the native VM候选节点池以及部署形态，由 AWX inventory hosts 动态组装。 type targetMetadata struct { Topology string `json:"topology"` @@ -71,7 +113,7 @@ func parseTargetMetadata(raw string) targetMetadata { meta.Topology = "standalone" } if meta.MySQLPort == 0 { - meta.MySQLPort = 3307 + meta.MySQLPort = 13306 } return meta } @@ -90,6 +132,19 @@ func firstFreeHost(hosts []targetHost, occupied []string) *targetHost { return nil } +func firstFreePort(occupied []int) int { + taken := make(map[int]bool, len(occupied)) + for _, port := range occupied { + taken[port] = true + } + for port := 13306; port <= 13999; port++ { + if !taken[port] { + return port + } + } + return 0 +} + type DeliveryService struct { db *gorm.DB cfg config.Config @@ -139,11 +194,14 @@ func (s *DeliveryService) getTarget(ctx context.Context, templateID uint64) (Del } func (s *DeliveryService) awxDeliveryTarget(ctx context.Context, template AWXJobTemplate) (DeliveryTarget, error) { + if !template.AskVariablesOnLaunch || !template.AskLimitOnLaunch { + return DeliveryTarget{}, fmt.Errorf("AWX job template %d must enable Prompt on launch for Variables and Limit", template.ID) + } hosts, err := s.awx.ListInventoryHosts(ctx, template.Inventory) if err != nil { return DeliveryTarget{}, err } - meta := targetMetadata{Topology: "standalone", MySQLPort: 3307} + meta := targetMetadata{Topology: "standalone", MySQLPort: 13306} for _, host := range hosts { if !host.Enabled { continue @@ -170,6 +228,7 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin if idempotencyKey == "" || len(idempotencyKey) > 128 { return nil, false, fmt.Errorf("Idempotency-Key header is required and must not exceed 128 characters") } + normalizeMySQLDeliveryInput(&input) if err := validateDeliveryInput(input); err != nil { return nil, false, err } @@ -201,9 +260,6 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin return nil, false, fmt.Errorf("user is not authorized for this business line") } } - if input.MySQLVersion == "" { - input.MySQLVersion = "8.0" - } payload := deliveryPayload{MySQLDeliveryInput: input, TargetType: target.TargetType} raw, err := json.Marshal(payload) if err != nil { @@ -214,6 +270,7 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin ID: randomUUID(), BusinessLineID: input.BusinessLineID, RequestedBy: userID, + Component: "mysql", TargetType: target.TargetType, TargetID: target.ID, Namespace: input.Namespace, @@ -234,6 +291,80 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin } var supportedMySQLVersions = map[string]bool{"8.0": true} +var supportedMySQLTopologies = map[string]bool{"standalone": true} +var supportedDataDisks = map[string]bool{"/data": true, "/disk1": true, "/mnt/vol-1": true} +var supportedParamTemplates = map[string]bool{"default": true, "high_performance": true, "high_safety": true} +var supportedCharacterSets = map[string]bool{"utf8mb4": true, "utf8": true, "gbk": true, "latin1": true} +var supportedCollations = map[string]bool{ + "utf8mb4_general_ci": true, "utf8mb4_unicode_ci": true, "utf8mb4_0900_ai_ci": true, + "utf8_general_ci": true, "gbk_chinese_ci": true, "latin1_swedish_ci": true, +} +var supportedRedoLogCapacity = map[string]bool{"auto": true, "128M": true, "256M": true, "512M": true, "1G": true} +var supportedMaxBinlogSize = map[string]bool{"128M": true, "256M": true, "512M": true, "1G": true} + +func normalizeMySQLDeliveryInput(input *MySQLDeliveryInput) { + if input.MySQLVersion == "" { + input.MySQLVersion = "8.0" + } + if input.Topology == "" { + input.Topology = "standalone" + } + if input.DataDisk == "" { + input.DataDisk = "/data" + } + if input.ParamTemplate == "" { + input.ParamTemplate = "default" + } + if input.TimeZone == "" { + input.TimeZone = "+08:00" + } + if input.CharacterSet == "" { + input.CharacterSet = "utf8mb4" + } + if input.Collation == "" { + input.Collation = defaultCollation(input.CharacterSet) + } + if input.MaxConnections == "" { + input.MaxConnections = "auto" + } + if input.InnoDBRedoLogCapacity == "" { + input.InnoDBRedoLogCapacity = "auto" + } + if input.InnoDBFlushLogAtTrxCommit == 0 { + input.InnoDBFlushLogAtTrxCommit = 1 + } + if input.SyncBinlog == 0 { + input.SyncBinlog = 1 + } + if input.InnoDBIOCapacity == 0 { + input.InnoDBIOCapacity = 2000 + } + if input.LongQueryTime == 0 { + input.LongQueryTime = 1 + } + if input.BinlogExpireLogsSeconds == 0 { + input.BinlogExpireLogsSeconds = 604800 + } + if input.MaxBinlogSize == "" { + input.MaxBinlogSize = "256M" + } + input.CPUMilli = input.CPUCores * 1000 + input.MemoryMi = input.MemoryGB * 1024 + input.StorageGi = input.StorageGB +} + +func defaultCollation(characterSet string) string { + switch characterSet { + case "utf8": + return "utf8_general_ci" + case "gbk": + return "gbk_chinese_ci" + case "latin1": + return "latin1_swedish_ci" + default: + return "utf8mb4_general_ci" + } +} func validateDeliveryInput(input MySQLDeliveryInput) error { if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) { @@ -242,19 +373,122 @@ func validateDeliveryInput(input MySQLDeliveryInput) error { if len(input.InstanceName) > 63 || !dnsLabelPattern.MatchString(input.InstanceName) { return fmt.Errorf("instance_name must be a valid Kubernetes DNS label") } - if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 1024 || input.MemoryMi > 4096 || input.StorageGi < 10 || input.StorageGi > 100 { - return fmt.Errorf("requested resources are outside the supported range (memory: 1024-4096 MiB, storage: 10-100 GiB)") + if !oneOfInt64(input.CPUCores, []int64{1, 2, 4, 8, 16}) { + return fmt.Errorf("cpu_cores must be one of 1, 2, 4, 8, 16") + } + if !oneOfInt64(input.MemoryGB, []int64{2, 4, 8, 16, 32, 64}) { + return fmt.Errorf("memory_gb must be one of 2, 4, 8, 16, 32, 64") + } + if input.StorageGB < 20 || input.StorageGB > 2000 { + return fmt.Errorf("storage_gb must be between 20 and 2000") } if input.MySQLVersion != "" && !supportedMySQLVersions[input.MySQLVersion] { return fmt.Errorf("unsupported mysql_version %q, supported: 8.0", input.MySQLVersion) } + if input.Topology != "" && !supportedMySQLTopologies[input.Topology] { + return fmt.Errorf("unsupported topology %q, supported: standalone", input.Topology) + } + if input.MySQLPort != 0 && (input.MySQLPort < 13306 || input.MySQLPort > 13999) { + return fmt.Errorf("mysql_port must be empty for auto assignment or between 13306 and 13999") + } + if !supportedDataDisks[input.DataDisk] { + return fmt.Errorf("unsupported data_disk %q", input.DataDisk) + } + if !supportedParamTemplates[input.ParamTemplate] { + return fmt.Errorf("unsupported param_template %q", input.ParamTemplate) + } + if !validTimeZone(input.TimeZone) { + return fmt.Errorf("unsupported timezone %q", input.TimeZone) + } + if input.LowerCaseTableNames != 0 && input.LowerCaseTableNames != 1 { + return fmt.Errorf("lower_case_table_names must be 0 or 1") + } + if !supportedCharacterSets[input.CharacterSet] { + return fmt.Errorf("unsupported character_set %q", input.CharacterSet) + } + if !supportedCollations[input.Collation] || !strings.HasPrefix(input.Collation, input.CharacterSet+"_") { + return fmt.Errorf("collation %q is not valid for character_set %q", input.Collation, input.CharacterSet) + } + if !validMaxConnections(input.MaxConnections) { + return fmt.Errorf("max_connections must be auto or one of 200, 500, 1000, 2000, 4000, 8000, 16000") + } + if !supportedRedoLogCapacity[input.InnoDBRedoLogCapacity] { + return fmt.Errorf("unsupported innodb_redo_log_capacity %q", input.InnoDBRedoLogCapacity) + } + if !oneOfInt(input.InnoDBFlushLogAtTrxCommit, []int{0, 1, 2}) { + return fmt.Errorf("innodb_flush_log_at_trx_commit must be one of 0, 1, 2") + } + if input.SyncBinlog != 0 && input.SyncBinlog != 1 { + return fmt.Errorf("sync_binlog must be 0 or 1") + } + if !oneOfInt(input.InnoDBIOCapacity, []int{200, 2000, 5000}) { + return fmt.Errorf("innodb_io_capacity must be one of 200, 2000, 5000") + } + if !oneOfFloat(input.LongQueryTime, []float64{0.5, 1, 2, 5, 10}) { + return fmt.Errorf("long_query_time must be one of 0.5, 1, 2, 5, 10") + } + if !oneOfInt64(input.BinlogExpireLogsSeconds, []int64{86400, 259200, 604800, 1209600}) { + return fmt.Errorf("binlog_expire_logs_seconds must be one of 86400, 259200, 604800, 1209600") + } + if !supportedMaxBinlogSize[input.MaxBinlogSize] { + return fmt.Errorf("unsupported max_binlog_size %q", input.MaxBinlogSize) + } return nil } -func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]model.DeliveryTask, error) { +func oneOfInt(value int, allowed []int) bool { + for _, item := range allowed { + if value == item { + return true + } + } + return false +} + +func oneOfInt64(value int64, allowed []int64) bool { + for _, item := range allowed { + if value == item { + return true + } + } + return false +} + +func oneOfFloat(value float64, allowed []float64) bool { + for _, item := range allowed { + if value == item { + return true + } + } + return false +} + +func validTimeZone(value string) bool { + return value == "SYSTEM" || value == "+08:00" || value == "+00:00" || value == "Asia/Shanghai" +} + +func validMaxConnections(value string) bool { + if value == "auto" { + return true + } + switch value { + case "200", "500", "1000", "2000", "4000", "8000", "16000": + return true + default: + return false + } +} + +func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin bool, filter DeliveryTaskListFilter) ([]model.DeliveryTask, error) { query := s.db.WithContext(ctx).Order("created_at DESC") - if businessLineID != 0 { - query = query.Where("business_line_id = ?", businessLineID) + if filter.BusinessLineID != 0 { + query = query.Where("business_line_id = ?", filter.BusinessLineID) + } + if filter.Component != "" { + query = query.Where("component = ?", strings.ToLower(strings.TrimSpace(filter.Component))) + } + if filter.ActiveOnly { + query = query.Where("status NOT IN ?", terminalTaskStatuses()) } if !isAdmin { query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID)) @@ -310,7 +544,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT var target DeliveryTarget dispatchable := false err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ?", model.TaskPending).Order("created_at ASC").First(&task).Error; err != nil { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ?", model.TaskPending).Order("created_at DESC").First(&task).Error; err != nil { return err } var targetErr error @@ -321,6 +555,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { return s.failInTransaction(tx, &task, model.TaskValidationFailed, "stored deployment payload is invalid") } + normalizeMySQLDeliveryInput(&payload.MySQLDeliveryInput) activeStates := []string{model.TaskValidating, model.TaskDispatching, model.TaskRunning, model.TaskCanceling} checks := []struct { query string @@ -330,8 +565,6 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT }{ {"status IN ?", []any{activeStates}, s.cfg.DeliveryGlobalLimit, "global concurrency limit reached"}, {"status IN ? AND target_id = ?", []any{activeStates, task.TargetID}, s.cfg.DeliveryTargetLimit, "target concurrency limit reached"}, - {"status IN ? AND business_line_id = ?", []any{activeStates, task.BusinessLineID}, s.cfg.DeliveryBusinessLimit, "business line concurrency limit reached"}, - {"status IN ? AND target_id = ? AND namespace = ?", []any{activeStates, task.TargetID, task.Namespace}, 1, "namespace already has an active MySQL delivery"}, } for _, check := range checks { var count int64 @@ -364,16 +597,27 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT if host == nil { return fmt.Errorf("defer: no free host available on target") } + mysqlPort := payload.MySQLPort + if mysqlPort == 0 { + var occupiedPorts []int + if err := tx.Model(&model.DeliveryTask{}).Where("target_id = ? AND target_host = ? AND mysql_port <> ? AND status NOT IN ?", task.TargetID, host.Name, 0, occupiedExclude).Pluck("mysql_port", &occupiedPorts).Error; err != nil { + return err + } + mysqlPort = firstFreePort(occupiedPorts) + if mysqlPort == 0 { + return fmt.Errorf("defer: no free MySQL port available on target host") + } + } reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "reserved", ExpiresAt: time.Now().Add(time.Duration(s.cfg.ReservationTTLMinutes) * time.Minute)} if err := tx.Create(&reservation).Error; err != nil { return err } - if err := tx.Model(&model.DeliveryTask{}).Where("id = ?", task.ID).Updates(map[string]any{"target_host": host.Name, "target_host_ip": host.IP, "mysql_port": meta.MySQLPort}).Error; err != nil { + if err := tx.Model(&model.DeliveryTask{}).Where("id = ?", task.ID).Updates(map[string]any{"target_host": host.Name, "target_host_ip": host.IP, "mysql_port": mysqlPort}).Error; err != nil { return err } task.TargetHost = host.Name task.TargetHostIP = host.IP - task.MySQLPort = meta.MySQLPort + task.MySQLPort = mysqlPort if err := s.transitionTx(tx, &task, model.TaskDispatching, "resources reserved", ""); err != nil { return err } @@ -509,7 +753,12 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { return nil, false, err } + normalizeMySQLDeliveryInput(&payload.MySQLDeliveryInput) meta := parseTargetMetadata(target.Metadata) + topology := payload.Topology + if topology == "" { + topology = meta.Topology + } // Persist execution record BEFORE launching AWX to ensure crash recovery. now := time.Now() execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: "pending", Status: "launching", StartedAt: &now} @@ -518,10 +767,18 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa } job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: map[string]any{ "task_id": task.ID, "payload_hash": task.PayloadHash, - "target_hosts": task.TargetHost, "topology": meta.Topology, + "target_hosts": task.TargetHost, "topology": topology, "instance_name": payload.InstanceName, "mysql_port": task.MySQLPort, - "memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi, - "mysql_version": payload.MySQLVersion, + "data_disk": payload.DataDisk, "cpu_cores": payload.CPUCores, + "memory_gb": payload.MemoryGB, "storage_gb": payload.StorageGB, + "mysql_version": payload.MySQLVersion, "param_template": payload.ParamTemplate, + "timezone": payload.TimeZone, "lower_case_table_names": payload.LowerCaseTableNames, + "character_set": payload.CharacterSet, "collation": payload.Collation, + "max_connections": payload.MaxConnections, "innodb_redo_log_capacity": payload.InnoDBRedoLogCapacity, + "innodb_flush_log_at_trx_commit": payload.InnoDBFlushLogAtTrxCommit, "sync_binlog": payload.SyncBinlog, + "innodb_io_capacity": payload.InnoDBIOCapacity, "long_query_time": payload.LongQueryTime, + "binlog_expire_logs_seconds": payload.BinlogExpireLogsSeconds, "max_binlog_size": payload.MaxBinlogSize, + "delivery_callback_url": s.deliveryCallbackURL(task.ID), "delivery_callback_token": s.cfg.AWXWebhookToken, }}) if err != nil { _ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}) @@ -538,35 +795,209 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa return &execution, false, nil } -func (s *DeliveryService) PollOnce(ctx context.Context) error { - var jobs []model.ExecutionJob - if err := s.db.WithContext(ctx).Where("status = ?", "running").Find(&jobs).Error; err != nil { +func (s *DeliveryService) deliveryCallbackURL(taskID string) string { + base := strings.TrimRight(strings.TrimSpace(s.cfg.DeliveryCallbackBaseURL), "/") + if base == "" { + return "" + } + return base + "/auth/internal/delivery/tasks/" + taskID + "/events" +} + +func (s *DeliveryService) HandleStageEvent(ctx context.Context, taskID string, input DeliveryStageEventInput) error { + stage := strings.ToLower(strings.TrimSpace(input.Stage)) + status := strings.ToLower(strings.TrimSpace(input.Status)) + if !validDeliveryStage(stage) { + return fmt.Errorf("invalid delivery stage %q", input.Stage) + } + if !validDeliveryStageStatus(status) { + return fmt.Errorf("invalid delivery stage status %q", input.Status) + } + message := strings.TrimSpace(input.Message) + if message == "" { + message = stage + " " + status + } + eventState := "stage_" + stage + "_" + status + + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var task model.DeliveryTask + if err := tx.First(&task, "id = ?", taskID).Error; err != nil { + return err + } + if input.AWXJobID != "" { + var execution model.ExecutionJob + if err := tx.Where("task_id = ?", task.ID).First(&execution).Error; err != nil { + return err + } + if execution.ExecutorJobID != strings.TrimSpace(input.AWXJobID) { + return fmt.Errorf("AWX job %s does not match task %s", input.AWXJobID, task.ID) + } + } + if isTerminalTaskStatus(task.Status) { + return nil + } + if err := tx.Create(&model.TaskEvent{ + TaskID: task.ID, + FromState: task.Status, + ToState: eventState, + Stage: stage, + EventStatus: status, + Message: message, + }).Error; err != nil { + return err + } + if status == "failed" { + now := time.Now() + if err := tx.Model(&model.ExecutionJob{}).Where("task_id = ?", task.ID).Updates(map[string]any{"status": "failed", "finished_at": now}).Error; err != nil { + return err + } + if err := s.transitionTx(tx, &task, model.TaskExecutionFailed, message, message); err != nil { + return err + } + return s.releaseReservation(tx, task.ID) + } + return nil + }) +} + +func (s *DeliveryService) HandleAWXJobNotification(ctx context.Context, input AWXJobNotificationInput) error { + status := strings.ToLower(strings.TrimSpace(input.Status)) + if status == "" && input.ID == 0 && awxNotificationTaskID(input.ExtraVars) == "" { + return nil + } + if status == "" { + return fmt.Errorf("missing AWX job status") + } + execution, err := s.findExecutionForAWXNotification(ctx, input) + if err != nil { return err } - for _, execution := range jobs { - job, err := s.awx.GetJob(ctx, execution.ExecutorJobID) - if err != nil { - s.finishExecution(ctx, &execution, "failed") - _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "poll AWX job "+execution.ExecutorJobID+": "+err.Error()) - continue + message := awxNotificationMessage(input) + switch status { + case "pending", "waiting", "running", "new": + return s.recordAWXEvent(ctx, execution.TaskID, status, message) + case "successful": + s.finishExecution(ctx, execution, "successful") + if err := s.completeTask(ctx, execution.TaskID); err != nil { + _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskValidationFailed, err.Error()) + return err } - switch strings.ToLower(job.Status) { - case "pending", "waiting", "running", "new": - continue - case "successful": - s.finishExecution(ctx, &execution, "successful") - if err := s.completeTask(ctx, execution.TaskID); err != nil { - _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskValidationFailed, err.Error()) - } - case "canceled": - s.finishExecution(ctx, &execution, "canceled") - _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, "AWX job was canceled") - default: - s.finishExecution(ctx, &execution, "failed") - _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "AWX job finished with status "+job.Status) + return nil + case "canceled", "cancelled": + s.finishExecution(ctx, execution, "canceled") + return s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, message) + case "failed", "error": + s.finishExecution(ctx, execution, "failed") + return s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, message) + default: + s.finishExecution(ctx, execution, "failed") + return s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "AWX job finished with status "+status) + } +} + +func (s *DeliveryService) findExecutionForAWXNotification(ctx context.Context, input AWXJobNotificationInput) (*model.ExecutionJob, error) { + var execution model.ExecutionJob + if input.ID != 0 { + if err := s.db.WithContext(ctx).Where("executor_job_id = ?", fmt.Sprint(input.ID)).First(&execution).Error; err == nil { + return &execution, nil + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, err } } - return nil + taskID := awxNotificationTaskID(input.ExtraVars) + if taskID == "" { + return nil, fmt.Errorf("AWX notification did not include a known job id or task_id") + } + if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).First(&execution).Error; err != nil { + return nil, err + } + return &execution, nil +} + +func (s *DeliveryService) recordAWXEvent(ctx context.Context, taskID, status, message string) error { + var task model.DeliveryTask + if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil { + return err + } + if isTerminalTaskStatus(task.Status) { + return nil + } + return s.db.WithContext(ctx).Create(&model.TaskEvent{ + TaskID: task.ID, + FromState: task.Status, + ToState: "awx_" + status, + Stage: "awx", + EventStatus: status, + Message: message, + }).Error +} + +func validDeliveryStage(stage string) bool { + switch stage { + case "precheck", "install", "configure", "healthcheck", "register": + return true + default: + return false + } +} + +func validDeliveryStageStatus(status string) bool { + switch status { + case "running", "success", "failed": + return true + default: + return false + } +} + +func isTerminalTaskStatus(status string) bool { + for _, item := range terminalTaskStatuses() { + if status == item { + return true + } + } + return false +} + +func terminalTaskStatuses() []string { + return []string{model.TaskFinished, model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskRegisterFailed, model.TaskCanceled} +} + +func awxNotificationMessage(input AWXJobNotificationInput) string { + status := strings.TrimSpace(input.Status) + name := strings.TrimSpace(input.Name) + if strings.TrimSpace(input.Traceback) != "" { + return strings.TrimSpace(input.Traceback) + } + if name == "" { + return "AWX job " + status + } + return "AWX job " + name + " " + status +} + +func awxNotificationTaskID(extraVars any) string { + var values map[string]any + switch v := extraVars.(type) { + case map[string]any: + values = v + case string: + if strings.TrimSpace(v) == "" { + return "" + } + _ = json.Unmarshal([]byte(v), &values) + default: + raw, err := json.Marshal(v) + if err != nil { + return "" + } + _ = json.Unmarshal(raw, &values) + } + if values == nil { + return "" + } + if taskID, ok := values["task_id"].(string); ok { + return strings.TrimSpace(taskID) + } + return "" } func (s *DeliveryService) finishExecution(ctx context.Context, execution *model.ExecutionJob, status string) { @@ -579,10 +1010,14 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil { return err } + if isTerminalTaskStatus(task.Status) { + return nil + } var payload deliveryPayload if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { return err } + normalizeMySQLDeliveryInput(&payload.MySQLDeliveryInput) addr := fmt.Sprintf("%s:%d", task.TargetHostIP, task.MySQLPort) if err := mysqlReady(ctx, addr); err != nil { return fmt.Errorf("MySQL health check failed: %w", err) @@ -664,19 +1099,23 @@ func (s *DeliveryService) failTask(ctx context.Context, task *model.DeliveryTask } func (s *DeliveryService) Run(ctx context.Context) { - interval := time.Duration(s.cfg.DeliveryPollSeconds) * time.Second + interval := time.Duration(s.cfg.DeliveryDispatchSeconds) * time.Second if interval < time.Second { interval = time.Second } ticker := time.NewTicker(interval) defer ticker.Stop() + if err := s.DispatchOnce(ctx); err != nil { + log.Printf("[delivery] dispatch pending task failed: %v", err) + } for { select { case <-ctx.Done(): return case <-ticker.C: - _ = s.DispatchOnce(ctx) - _ = s.PollOnce(ctx) + if err := s.DispatchOnce(ctx); err != nil { + log.Printf("[delivery] dispatch pending task failed: %v", err) + } } } } diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index bc18b82..8c22cf4 100644 --- a/server/internal/service/delivery_test.go +++ b/server/internal/service/delivery_test.go @@ -3,7 +3,8 @@ package service import "testing" func TestValidateDeliveryInput(t *testing.T) { - valid := MySQLDeliveryInput{BusinessLineID: 1, TargetID: 1, Namespace: "team-a", InstanceName: "mysql-01", CPUMilli: 500, MemoryMi: 1024, StorageGi: 10} + valid := MySQLDeliveryInput{BusinessLineID: 1, TargetID: 1, Namespace: "team-a", InstanceName: "mysql-01", CPUCores: 2, MemoryGB: 4, StorageGB: 50} + normalizeMySQLDeliveryInput(&valid) if err := validateDeliveryInput(valid); err != nil { t.Fatalf("valid input rejected: %v", err) } @@ -15,27 +16,43 @@ func TestValidateDeliveryInput(t *testing.T) { for name, input := range map[string]MySQLDeliveryInput{ "uppercase namespace": valid, "bad instance": valid, - "too little memory": valid, + "bad cpu cores": valid, "too much memory": valid, "too little storage": valid, "too much storage": valid, "unsupported version": valid, + "unsupported topology": valid, + "bad mysql port": valid, + "bad data disk": valid, + "unsupported charset": valid, + "bad collation": valid, } { switch name { case "uppercase namespace": input.Namespace = "Team-A" case "bad instance": input.InstanceName = "mysql_01" - case "too little memory": - input.MemoryMi = 512 + case "bad cpu cores": + input.CPUCores = 3 case "too much memory": - input.MemoryMi = 8192 + input.MemoryGB = 128 case "too little storage": - input.StorageGi = 5 + input.StorageGB = 10 case "too much storage": - input.StorageGi = 200 + input.StorageGB = 3000 case "unsupported version": input.MySQLVersion = "5.7" + case "unsupported topology": + input.Topology = "mgr" + case "bad mysql port": + input.MySQLPort = 3306 + case "bad data disk": + input.DataDisk = "/" + case "unsupported charset": + input.CharacterSet = "sjis" + case "bad collation": + input.CharacterSet = "utf8" + input.Collation = "utf8mb4_general_ci" } if err := validateDeliveryInput(input); err == nil { t.Errorf("%s was accepted", name)