diff --git a/server/go.mod b/server/go.mod index b9d969e..2ba9677 100644 --- a/server/go.mod +++ b/server/go.mod @@ -5,6 +5,7 @@ go 1.26 require ( github.com/gin-gonic/gin v1.11.0 github.com/golang-jwt/jwt/v5 v5.3.0 + github.com/go-sql-driver/mysql v1.8.1 github.com/swaggo/files v1.0.1 github.com/swaggo/gin-swagger v1.6.1 github.com/swaggo/swag v1.16.3 @@ -30,7 +31,6 @@ require ( github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.27.0 // indirect - github.com/go-sql-driver/mysql v1.8.1 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/goccy/go-yaml v1.18.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect diff --git a/server/internal/handler/task_log.go b/server/internal/handler/task_log.go index 5857053..ec122d3 100644 --- a/server/internal/handler/task_log.go +++ b/server/internal/handler/task_log.go @@ -216,11 +216,18 @@ func (h *TaskLogHandler) getWayneTask(c *gin.Context, id string, userID uint64, } func awxTaskSummary(task model.DeliveryTask) taskLogSummary { + serviceName := "mysql" + serviceLabel := "MySQL" + if strings.EqualFold(strings.TrimSpace(task.ServiceType), "postgresql") || + (strings.TrimSpace(task.ServiceType) == "" && strings.EqualFold(strings.TrimSpace(task.Component), "postgresql")) { + serviceName = "postgresql" + serviceLabel = "PostgreSQL" + } return taskLogSummary{ ID: "awx:" + task.ID, Source: "awx", - Service: "mysql", - Name: "MySQL 标准化交付 · " + task.InstanceName, + Service: serviceName, + Name: serviceLabel + " 标准化交付 · " + task.InstanceName, Runner: "AWX Job Template #" + strconv.FormatUint(task.TargetID, 10), Status: task.Status, StatusText: textForTaskStatus(task.Status), diff --git a/server/internal/handler/task_log_test.go b/server/internal/handler/task_log_test.go index 2417413..9fdf349 100644 --- a/server/internal/handler/task_log_test.go +++ b/server/internal/handler/task_log_test.go @@ -439,6 +439,33 @@ func TestAwxTaskSummary(t *testing.T) { } } +func TestAwxTaskSummaryPostgreSQL(t *testing.T) { + tests := []struct { + name string + serviceType string + component string + }{ + {name: "service type", serviceType: "postgresql", component: "postgresql"}, + {name: "legacy component fallback", component: "postgresql"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + summary := awxTaskSummary(model.DeliveryTask{ + ID: "task-postgresql", + ServiceType: tt.serviceType, + Component: tt.component, + InstanceName: "postgresql-xinfra-n1aw28", + }) + if summary.Service != "postgresql" { + t.Fatalf("awxTaskSummary Service = %q, want %q", summary.Service, "postgresql") + } + if summary.Name != "PostgreSQL 标准化交付 · postgresql-xinfra-n1aw28" { + t.Fatalf("awxTaskSummary Name = %q, want PostgreSQL delivery name", summary.Name) + } + }) + } +} + func TestWayneTaskSummary(t *testing.T) { now := time.Now() history := service.WayneDeploymentHistory{ diff --git a/server/internal/model/delivery.go b/server/internal/model/delivery.go index 0565dcb..6dac9b3 100644 --- a/server/internal/model/delivery.go +++ b/server/internal/model/delivery.go @@ -34,17 +34,20 @@ type ResourceQuota struct { } 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"` - ServiceType string `gorm:"size:32;not null;default:'mysql';index" json:"service_type"` - TargetID uint64 `gorm:"not null;index" json:"target_id"` - Namespace string `gorm:"size:63;not null;index" json:"namespace"` - InstanceName string `gorm:"size:63;not null" json:"instance_name"` - TargetHost string `gorm:"size:1024" json:"target_host,omitempty"` - TargetHostIP string `gorm:"size:64" json:"target_host_ip,omitempty"` + 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"` + ServiceType string `gorm:"size:32;not null;default:'mysql';index" json:"service_type"` + TargetID uint64 `gorm:"not null;index" json:"target_id"` + Namespace string `gorm:"size:63;not null;index" json:"namespace"` + InstanceName string `gorm:"size:63;not null" json:"instance_name"` + TargetHost string `gorm:"size:1024" json:"target_host,omitempty"` + TargetHostIP string `gorm:"size:64" json:"target_host_ip,omitempty"` + // For primary_replica, these contain the complete node lists in primary,replica order. + TargetHosts string `gorm:"type:text" json:"target_hosts,omitempty"` + TargetHostIPs string `gorm:"type:text" json:"target_host_ips,omitempty"` MySQLPort int `gorm:"column:mysql_port;not null;default:3307" json:"mysql_port"` PostgreSQLPort int `gorm:"column:postgresql_port;not null;default:15432" json:"postgresql_port,omitempty"` Status string `gorm:"size:32;not null;index" json:"status"` @@ -101,6 +104,8 @@ type DeploymentResult struct { TargetID uint64 `gorm:"not null;index" json:"target_id"` NodeName string `gorm:"size:128;not null;default:''" json:"node_name"` Host string `gorm:"size:255;not null;default:''" json:"host"` + TargetHosts string `gorm:"type:text" json:"target_hosts,omitempty"` + TargetHostIPs string `gorm:"type:text" json:"target_host_ips,omitempty"` Port int `gorm:"not null;default:0" json:"port"` Version string `gorm:"size:64;not null;default:''" json:"version"` Status string `gorm:"size:32;not null;index" json:"status"` diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 840d177..25ea5d4 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -7,6 +7,7 @@ import ( "crypto/cipher" "crypto/rand" "crypto/sha256" + "database/sql" "encoding/base64" "encoding/hex" "encoding/json" @@ -24,6 +25,7 @@ import ( "github.com/1024XEngineer/xinfra/server/internal/config" "github.com/1024XEngineer/xinfra/server/internal/model" + mysqlDriver "github.com/go-sql-driver/mysql" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -41,16 +43,18 @@ type MySQLDeliveryInput struct { InstanceDesc string `json:"instance_desc,omitempty"` MySQLVersion string `json:"mysql_version"` Topology string `json:"topology"` + ReplicaCount int `json:"replica_count"` MySQLPort int `json:"mysql_port"` DataDisk string `json:"data_disk"` // 调度控制（选填）：点名候选池内主机跳过自动选机，端口/配额/实机守卫照常执行 - TargetHost string `json:"target_host"` - CPUMilli int64 `json:"cpu_milli" binding:"required"` - MemoryMi int64 `json:"memory_mi" binding:"required"` - StorageGi int64 `json:"storage_gi" binding:"required"` - CPUCores int64 `json:"cpu_cores,omitempty"` - MemoryGB int64 `json:"memory_gb,omitempty"` - StorageGB int64 `json:"storage_gb,omitempty"` + TargetHost string `json:"target_host"` + TargetHosts []string `json:"target_hosts"` + CPUMilli int64 `json:"cpu_milli" binding:"required"` + MemoryMi int64 `json:"memory_mi" binding:"required"` + StorageGi int64 `json:"storage_gi" binding:"required"` + CPUCores int64 `json:"cpu_cores,omitempty"` + MemoryGB int64 `json:"memory_gb,omitempty"` + StorageGB int64 `json:"storage_gb,omitempty"` // 数据库配置（选填，缺省由 playbook 基线兜底） Timezone string `json:"timezone"` LowerCaseTableNames *int `json:"lower_case_table_names"` @@ -228,6 +232,8 @@ func stringValue(value any) string { switch v := value.(type) { case string: return v + case []byte: + return string(v) case fmt.Stringer: return v.String() default: @@ -329,6 +335,35 @@ func hostTaskCount(occupied []string, name string) int { return count } +func topologyNodeCount(input MySQLDeliveryInput) int64 { + if input.Topology == "primary_replica" { + replicas := input.ReplicaCount + if replicas < 1 { + replicas = 1 + } + return int64(replicas + 1) + } + return 1 +} + +func splitHosts(raw string) []string { + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + seen := map[string]struct{}{} + for _, part := range parts { + part = strings.TrimSpace(part) + if part == "" { + continue + } + if _, ok := seen[part]; ok { + continue + } + seen[part] = struct{}{} + out = append(out, part) + } + return out +} + // 端口池 13306-13999：混合分配模型，用户留空时自动分配，可覆盖为池内指定端口。 const ( mysqlPortPoolStart = 13306 @@ -781,9 +816,8 @@ func ensureInstanceNameAvailable(tx *gorm.DB, businessLineID uint64, component, // 8.0 走 Ubuntu 自带源，8.4 走 MySQL 官方 APT 源；5.6/5.7 已 EOL 且无 noble 包，不支持。 var supportedMySQLVersions = map[string]bool{"8.0": true, "8.4": true} -// 拓扑白名单：playbook 已支持 primary_replica/mgr_3 的配置渲染， -// 但调度器仍是单主机模型且复制编排未自动化，本期仅放开 standalone。 -var supportedTopologies = map[string]bool{"standalone": true} +// 拓扑白名单：支持 standalone 和一主多从 primary_replica；MGR 暂未开放。 +var supportedTopologies = map[string]bool{"standalone": true, "primary_replica": true} const rollbackLaunchTimeout = 2 * time.Minute @@ -823,7 +857,10 @@ func validateDeliveryInput(input MySQLDeliveryInput, _ []string) error { return fmt.Errorf("unsupported mysql_version %q, supported: 8.0, 8.4 (5.6/5.7 are EOL and have no Ubuntu 24.04 packages)", input.MySQLVersion) } if input.Topology != "" && !supportedTopologies[input.Topology] { - return fmt.Errorf("unsupported topology %q, supported: standalone (primary_replica/mgr_3 pending scheduler support)", input.Topology) + return fmt.Errorf("unsupported topology %q, supported: standalone, primary_replica", input.Topology) + } + if input.Topology == "primary_replica" && (input.ReplicaCount < 1 || input.ReplicaCount > 7) { + return fmt.Errorf("replica_count must be between 1 and 7 for primary_replica") } if input.MySQLPort != 0 && (input.MySQLPort < mysqlPortPoolStart || input.MySQLPort > mysqlPortPoolEnd) { return fmt.Errorf("mysql_port must be left empty for auto allocation or within the pool %d-%d", mysqlPortPoolStart, mysqlPortPoolEnd) @@ -831,6 +868,20 @@ func validateDeliveryInput(input MySQLDeliveryInput, _ []string) error { if input.TargetHost != "" && (len(input.TargetHost) > 253 || !hostNamePattern.MatchString(input.TargetHost)) { return fmt.Errorf("target_host must be a valid inventory host name") } + if len(input.TargetHosts) > 8 { + return fmt.Errorf("target_hosts may contain at most 8 nodes") + } + seenHosts := map[string]struct{}{} + for _, host := range input.TargetHosts { + host = strings.TrimSpace(host) + if len(host) > 253 || !hostNamePattern.MatchString(host) { + return fmt.Errorf("target_hosts contains invalid inventory host %q", host) + } + if _, exists := seenHosts[host]; exists { + return fmt.Errorf("target_hosts contains duplicate host %q", host) + } + seenHosts[host] = struct{}{} + } if input.Timezone != "" && !timezonePattern.MatchString(input.Timezone) { return fmt.Errorf("timezone must be an offset like +08:00, SYSTEM, or a named zone like Asia/Shanghai") } @@ -881,6 +932,9 @@ func normalizeMySQLDeliveryInput(input *MySQLDeliveryInput) { input.InstanceName = normalizeDNSLabel(input.InstanceName) input.InstanceDesc = strings.TrimSpace(input.InstanceDesc) input.TargetHost = strings.TrimSpace(input.TargetHost) + for i := range input.TargetHosts { + input.TargetHosts[i] = strings.TrimSpace(input.TargetHosts[i]) + } input.DataDisk = strings.TrimSpace(input.DataDisk) if input.MySQLVersion == "" { input.MySQLVersion = "8.0" @@ -888,6 +942,9 @@ func normalizeMySQLDeliveryInput(input *MySQLDeliveryInput) { if input.Topology == "" { input.Topology = "standalone" } + if input.Topology == "primary_replica" && input.ReplicaCount == 0 { + input.ReplicaCount = 1 + } if input.CPUMilli == 0 && input.CPUCores != 0 { input.CPUMilli = input.CPUCores * 1000 } @@ -1641,7 +1698,8 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT } } } - quotaOK, err := checkResourceQuota(tx, task.BusinessLineID, task.TargetID, payload) + nodeCount := topologyNodeCount(payload.MySQLDeliveryInput) + quotaOK, err := checkResourceQuota(tx, task.BusinessLineID, task.TargetID, payload, nodeCount) if err != nil { return err } @@ -1654,9 +1712,26 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT } var occupied []string occupiedExclude := allocationReleasedTaskStatuses() - if err := tx.Model(&model.DeliveryTask{}).Where("target_id = ? AND target_host <> ? AND status NOT IN ?", task.TargetID, "", occupiedExclude).Pluck("target_host", &occupied).Error; err != nil { + var taskAllocations []struct{ TargetHost, TargetHosts string } + if err := tx.Model(&model.DeliveryTask{}).Select("target_host, target_hosts").Where("target_id = ? AND status NOT IN ?", task.TargetID, occupiedExclude).Find(&taskAllocations).Error; err != nil { return err } + for _, allocation := range taskAllocations { + occupied = append(occupied, splitHosts(allocation.TargetHosts)...) + if allocation.TargetHosts == "" { + occupied = append(occupied, splitHosts(allocation.TargetHost)...) + } + } + var resultAllocations []struct{ NodeName, TargetHosts string } + if err := tx.Model(&model.DeploymentResult{}).Select("node_name, target_hosts").Where("target_id = ? AND status IN ?", task.TargetID, occupiedDeploymentStatuses()).Find(&resultAllocations).Error; err != nil { + return err + } + for _, allocation := range resultAllocations { + occupied = append(occupied, splitHosts(allocation.TargetHosts)...) + if allocation.TargetHosts == "" { + occupied = append(occupied, splitHosts(allocation.NodeName)...) + } + } limit := s.cfg.DeliveryHostInstanceLimit var host *targetHost if payload.TargetHost != "" { @@ -1679,30 +1754,120 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT if host == nil { return fmt.Errorf("defer: no free host available on target") } + nodes := []*targetHost{host} + if nodeCount > 1 { + if len(payload.TargetHosts) > 0 { + if int64(len(payload.TargetHosts)) != nodeCount { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("primary_replica requires %d target_hosts, got %d", nodeCount, len(payload.TargetHosts))) + } + nodes = nodes[:0] + seen := map[string]struct{}{} + for index, requestedHost := range payload.TargetHosts { + var selected *targetHost + for i := range meta.Hosts { + if meta.Hosts[i].Name == requestedHost { + selected = &meta.Hosts[i] + break + } + } + if selected == nil { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_hosts[%d] %q is not in the candidate host pool", index, requestedHost)) + } + if _, exists := seen[selected.Name]; exists { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_hosts contains duplicate host %q", selected.Name)) + } + if hostTaskCount(occupied, selected.Name) >= max(limit, 1) { + return fmt.Errorf("defer: pinned host %s reached the per-host instance limit %d", selected.Name, max(limit, 1)) + } + seen[selected.Name] = struct{}{} + nodes = append(nodes, selected) + } + if payload.TargetHost != "" && nodes[0].Name != payload.TargetHost { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, "target_host must match the first primary_replica target_hosts entry") + } + } else { + for i := range meta.Hosts { + candidate := &meta.Hosts[i] + if candidate.Name == host.Name || hostTaskCount(occupied, candidate.Name) >= max(limit, 1) { + continue + } + nodes = append(nodes, candidate) + if int64(len(nodes)) == nodeCount { + break + } + } + if int64(len(nodes)) != nodeCount { + return fmt.Errorf("defer: only %d free hosts available for primary_replica requiring %d", len(nodes), nodeCount) + } + } + } // 端口池混合分配：同主机已占端口 = 非终态任务分配端口 ∪ 存量 active 实例端口。 var usedPorts []int - if err := tx.Model(&model.DeliveryTask{}).Where("target_host = ? AND status NOT IN ?", host.Name, occupiedExclude).Pluck("mysql_port", &usedPorts).Error; err != nil { + var portAllocations []struct { + TargetHost, TargetHosts string + MySQLPort int + } + if err := tx.Model(&model.DeliveryTask{}).Select("target_host, target_hosts, mysql_port").Where("target_id = ? AND status NOT IN ?", task.TargetID, occupiedExclude).Find(&portAllocations).Error; err != nil { return err } var instancePorts []int - if err := tx.Model(&model.DeploymentResult{}). - Where("component = ? AND service_type = ? AND node_name = ? AND status IN ?", "mysql", "database", host.Name, occupiedDeploymentStatuses()). - Pluck("port", &instancePorts).Error; err != nil { + var resultPorts []struct { + NodeName, TargetHosts string + Port int + } + if err := tx.Model(&model.DeploymentResult{}).Select("node_name, target_hosts, port").Where("target_id = ? AND component = ? AND service_type = ? AND status IN ?", task.TargetID, "mysql", "database", occupiedDeploymentStatuses()).Find(&resultPorts).Error; err != nil { return err } + for _, allocation := range portAllocations { + for _, allocatedHost := range splitHosts(allocation.TargetHosts) { + for _, node := range nodes { + if allocatedHost == node.Name { + usedPorts = append(usedPorts, allocation.MySQLPort) + } + } + } + if allocation.TargetHosts == "" && allocation.TargetHost != "" { + for _, node := range nodes { + if allocation.TargetHost == node.Name { + usedPorts = append(usedPorts, allocation.MySQLPort) + } + } + } + } + for _, allocation := range resultPorts { + hosts := splitHosts(allocation.TargetHosts) + if len(hosts) == 0 { + hosts = splitHosts(allocation.NodeName) + } + for _, allocatedHost := range hosts { + for _, node := range nodes { + if allocatedHost == node.Name { + instancePorts = append(instancePorts, allocation.Port) + } + } + } + } port, portErr := allocatePort(payload.MySQLPort, append(usedPorts, instancePorts...)) if portErr != nil { return s.failInTransaction(tx, &task, model.TaskValidationFailed, portErr.Error()) } - 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)} + primaryHostNames := make([]string, 0, len(nodes)) + primaryHostIPs := make([]string, 0, len(nodes)) + for _, node := range nodes { + primaryHostNames = append(primaryHostNames, node.Name) + primaryHostIPs = append(primaryHostIPs, node.IP) + } + reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli * nodeCount, MemoryMi: payload.MemoryMi * nodeCount, StorageGi: payload.StorageGi * nodeCount, InstanceCount: nodeCount, 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": port}).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, "target_hosts": strings.Join(primaryHostNames, ","), "target_host_ips": strings.Join(primaryHostIPs, ","), "mysql_port": port}).Error; err != nil { return err } task.TargetHost = host.Name task.TargetHostIP = host.IP + task.TargetHosts = strings.Join(primaryHostNames, ",") + task.TargetHostIPs = strings.Join(primaryHostIPs, ",") task.MySQLPort = port if err := s.transitionTx(tx, &task, model.TaskDispatching, "resources reserved", ""); err != nil { return err @@ -1716,7 +1881,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT return &task, err } -func checkResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload deliveryPayload) (bool, error) { +func checkResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload deliveryPayload, nodeCount int64) (bool, error) { var quota model.ResourceQuota if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", businessLineID, targetID).First("a).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -1732,10 +1897,10 @@ func checkResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload de if err := tx.Model(&model.ResourceReservation{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COALESCE(SUM(instance_count),0) instances").Where("business_line_id = ? AND target_id = ? AND (status = ? OR (status = ? AND expires_at > ?))", businessLineID, targetID, "rollback", "reserved", time.Now()).Scan(&reserved).Error; err != nil { return false, err } - return used.CPU+reserved.CPU+payload.CPUMilli <= quota.CPUMilli && - used.Memory+reserved.Memory+payload.MemoryMi <= quota.MemoryMi && - used.Storage+reserved.Storage+payload.StorageGi <= quota.StorageGi && - used.Instances+reserved.Instances+1 <= quota.InstanceLimit, nil + return used.CPU+reserved.CPU+payload.CPUMilli*nodeCount <= quota.CPUMilli && + used.Memory+reserved.Memory+payload.MemoryMi*nodeCount <= quota.MemoryMi && + used.Storage+reserved.Storage+payload.StorageGi*nodeCount <= quota.StorageGi && + used.Instances+reserved.Instances+nodeCount <= quota.InstanceLimit, nil } func (s *DeliveryService) failInTransaction(tx *gorm.DB, task *model.DeliveryTask, status, message string) error { @@ -1846,9 +2011,12 @@ func mysqlDeploymentMetadata(task model.DeliveryTask, payload deliveryPayload) m "instance_name": instance, "target_host": task.TargetHost, "target_host_ip": task.TargetHostIP, + "target_hosts": task.TargetHosts, + "target_host_ips": task.TargetHostIPs, "mysql_port": task.MySQLPort, "mysql_version": mysqlVersion, "topology": topology, + "replica_count": payload.ReplicaCount, "cpu_milli": payload.CPUMilli, "memory_mi": payload.MemoryMi, "storage_gi": payload.StorageGi, @@ -1911,6 +2079,73 @@ func mysqlReady(ctx context.Context, address string) error { return conn.Close() } +// mysqlReplicationReady verifies the runtime replication state instead of +// treating an open TCP port as evidence that a replica is configured. The +// delivery credential is already encrypted at rest and is only decrypted for +// this short-lived health check. +func mysqlReplicationReady(ctx context.Context, address, username, password string) error { + config := mysqlDriver.Config{ + User: username, + Passwd: password, + Net: "tcp", + Addr: address, + Timeout: 5 * time.Second, + ReadTimeout: 5 * time.Second, + WriteTimeout: 5 * time.Second, + AllowNativePasswords: true, + } + db, err := sql.Open("mysql", config.FormatDSN()) + if err != nil { + return err + } + defer db.Close() + + if err := db.PingContext(ctx); err != nil { + return fmt.Errorf("connect: %w", err) + } + rows, err := db.QueryContext(ctx, "SHOW REPLICA STATUS") + if err != nil { + return fmt.Errorf("query replica status: %w", err) + } + defer rows.Close() + columns, err := rows.Columns() + if err != nil { + return err + } + if !rows.Next() { + if err := rows.Err(); err != nil { + return err + } + return errors.New("replica status is empty; CHANGE REPLICATION SOURCE TO was not applied") + } + values := make([]any, len(columns)) + refs := make([]any, len(columns)) + for i := range values { + refs[i] = &values[i] + } + if err := rows.Scan(refs...); err != nil { + return err + } + status := make(map[string]string, len(columns)) + for i, column := range columns { + status[column] = strings.TrimSpace(stringValue(values[i])) + } + ioRunning := status["Replica_IO_Running"] + if ioRunning == "" { + ioRunning = status["Slave_IO_Running"] + } + sqlRunning := status["Replica_SQL_Running"] + if sqlRunning == "" { + sqlRunning = status["Slave_SQL_Running"] + } + if ioRunning != "Yes" || sqlRunning != "Yes" { + lastIOError := status["Last_IO_Error"] + lastSQLError := status["Last_SQL_Error"] + return fmt.Errorf("replication threads are not healthy (io=%q sql=%q io_error=%q sql_error=%q)", ioRunning, sqlRunning, lastIOError, lastSQLError) + } + return nil +} + func (s *DeliveryService) DispatchOnce(ctx context.Context) error { task, err := s.claimAndReserve(ctx) if err != nil { @@ -1980,7 +2215,11 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa } extraVars["delivery_callback_url"] = s.deliveryCallbackURL(task.ID) extraVars["delivery_callback_token"] = s.cfg.AWXWebhookToken - job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: extraVars}) + limit := task.TargetHosts + if strings.TrimSpace(limit) == "" { + limit = task.TargetHost + } + job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: limit, ExtraVars: extraVars}) if err != nil { _ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}) return nil, false, err @@ -2012,13 +2251,30 @@ func deliveryExtraVars(task *model.DeliveryTask, payload deliveryPayload, meta t // 存量任务的 ImmutablePayload 无 topology 字段，回退到 target metadata。 topology = meta.Topology } + allHosts := task.TargetHosts + if strings.TrimSpace(allHosts) == "" { + allHosts = task.TargetHost + } vars := map[string]any{ "task_id": task.ID, "payload_hash": task.PayloadHash, - "target_hosts": task.TargetHost, "topology": topology, + "target_hosts": allHosts, "topology": topology, + "replica_count": payload.ReplicaCount, "instance_name": payload.InstanceName, "mysql_port": task.MySQLPort, "memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi, "mysql_version": payload.MySQLVersion, } + if topology == "primary_replica" { + nodes := splitHosts(allHosts) + if len(nodes) >= 2 { + vars["mysql_primary_host"] = nodes[0] + vars["mysql_replica_hosts"] = strings.Join(nodes[1:], ",") + ips := splitHosts(task.TargetHostIPs) + if len(ips) >= 2 { + vars["mysql_primary_ip"] = ips[0] + vars["mysql_replica_ips"] = strings.Join(ips[1:], ",") + } + } + } if payload.DataDisk != "" { vars["data_disk"] = payload.DataDisk } @@ -2348,10 +2604,36 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil { return err } - 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) + addresses := splitHosts(task.TargetHostIPs) + if len(addresses) == 0 { + addresses = []string{task.TargetHostIP} } + for _, hostIP := range addresses { + addr := fmt.Sprintf("%s:%d", hostIP, task.MySQLPort) + if err := mysqlReady(ctx, addr); err != nil { + return fmt.Errorf("MySQL health check failed on %s: %w", addr, err) + } + } + if payload.Topology == "primary_replica" { + if len(addresses) < 2 { + return fmt.Errorf("primary_replica requires at least one replica address, got %d", len(addresses)) + } + credentialVars, err := s.deploymentCredentialVars(ctx, task.ID) + if err != nil { + return fmt.Errorf("replication health check credential unavailable: %w", err) + } + rootPassword := credentialVars["mysql_root_password"] + if rootPassword == "" { + return errors.New("replication health check credential unavailable: root password is empty") + } + for _, hostIP := range addresses[1:] { + addr := fmt.Sprintf("%s:%d", hostIP, task.MySQLPort) + if err := mysqlReplicationReady(ctx, addr, "root", rootPassword); err != nil { + return fmt.Errorf("MySQL replication health check failed on %s: %w", addr, err) + } + } + } + nodeCount := topologyNodeCount(payload.MySQLDeliveryInput) now := time.Now() if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if err := s.transitionTx(tx, &task, model.TaskRegistering, "AWX succeeded and MySQL health check passed", ""); err != nil { @@ -2368,6 +2650,8 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error TargetID: task.TargetID, NodeName: task.TargetHost, Host: task.TargetHostIP, + TargetHosts: task.TargetHosts, + TargetHostIPs: task.TargetHostIPs, Port: task.MySQLPort, Version: payload.MySQLVersion, Status: "active", @@ -2382,7 +2666,7 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error if err := tx.Model(&model.DeploymentCredential{}).Where("task_id = ? AND status = ?", task.ID, "pending").Updates(map[string]any{"deployment_result_id": result.ID, "status": "available"}).Error; err != nil { return err } - if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: result.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "active"}).Error; err != nil { + if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: result.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli * nodeCount, MemoryMi: payload.MemoryMi * nodeCount, StorageGi: payload.StorageGi * nodeCount, InstanceCount: nodeCount, Status: "active"}).Error; err != nil { return err } if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", task.ID, "reserved").Update("status", "consumed").Error; err != nil { @@ -2752,9 +3036,13 @@ func (s *DeliveryService) launchRollbackJob(ctx context.Context, task *model.Del if err := s.db.WithContext(ctx).Save(&rollback).Error; err != nil { return s.markRollbackFailed(ctx, task.ID, "cannot persist rollback job: "+err.Error()) } + rollbackLimit := task.TargetHosts + if strings.TrimSpace(rollbackLimit) == "" { + rollbackLimit = task.TargetHost + } job, err := s.awx.Launch(ctx, s.cfg.RollbackTemplateID, AWXLaunchRequest{ InventoryID: target.AWXInventoryID, - Limit: task.TargetHost, + Limit: rollbackLimit, ExtraVars: rollbackExtraVars(task, payload), }) if err != nil { @@ -2834,8 +3122,12 @@ func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID } func rollbackExtraVars(task *model.DeliveryTask, payload deliveryPayload) map[string]any { + hosts := task.TargetHosts + if strings.TrimSpace(hosts) == "" { + hosts = task.TargetHost + } return map[string]any{ - "target_hosts": task.TargetHost, + "target_hosts": hosts, "instance_name": payload.InstanceName, "data_disk": payload.DataDisk, "task_id": task.ID, diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index 006e2ce..6a5c2ef 100644 --- a/server/internal/service/delivery_test.go +++ b/server/internal/service/delivery_test.go @@ -53,6 +53,21 @@ func TestValidateDeliveryInput(t *testing.T) { if err := validateDeliveryInput(lts, dataDisks); err != nil { t.Fatalf("8.4 LTS rejected: %v", err) } + replica := valid + replica.Topology = "primary_replica" + replica.ReplicaCount = 3 + replica.TargetHosts = []string{"db-01", "db-02", "db-03", "db-04"} + if err := validateDeliveryInput(replica, dataDisks); err != nil { + t.Fatalf("primary_replica input rejected: %v", err) + } + if got := topologyNodeCount(replica); got != 4 { + t.Fatalf("topologyNodeCount = %d, want 4", got) + } + tooManyReplicas := replica + tooManyReplicas.ReplicaCount = 8 + if err := validateDeliveryInput(tooManyReplicas, dataDisks); err == nil { + t.Fatal("replica_count 8 was accepted") + } for name, mutate := range map[string]func(*MySQLDeliveryInput){ "uppercase namespace": func(in *MySQLDeliveryInput) { in.Namespace = "Team-A" }, "bad instance": func(in *MySQLDeliveryInput) { in.InstanceName = "mysql_01" },