feat(delivery): switch AWX delivery to callbacks

This commit is contained in:
mac
2026-07-27 15:19:18 +08:00
parent 97f0de16bc
commit e0ee36cadb
9 changed files with 594 additions and 104 deletions
+3 -3
View File
@@ -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=
+6 -6
View File
@@ -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),
}
}
+35 -19
View File
@@ -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
+1 -1
View File
@@ -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
}
+9 -6
View File
@@ -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"`
}
+3 -2
View File
@@ -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)
+21 -7
View File
@@ -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)
+492 -53
View File
@@ -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)
}
}
}
}
+24 -7
View File
@@ -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)