From 0ca3af39dca5274f81e39413b1eaa39c226dc63e Mon Sep 17 00:00:00 2001 From: Hungerdream <1710233908@qq.com> Date: Fri, 31 Jul 2026 18:12:05 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(server):=20=E6=94=AF=E6=8C=81=20MySQL?= =?UTF-8?q?=20=E4=B8=80=E4=B8=BB=E5=A4=9A=E4=BB=8E=E6=8B=93=E6=89=91?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E4=B8=8E=E5=A4=8D=E5=88=B6=E5=81=A5=E5=BA=B7?= =?UTF-8?q?=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DeliveryTask/DeploymentResult 新增 target_hosts/target_host_ips 完整节点列表(主节点在前) - 交付输入支持 replica_count(1-7)与有序 target_hosts,按节点数放大配额/预留/资源用量 - 端口占用与已占主机解析改为逗号列表,兼容存量单主机字段 - 指定主机校验池成员/重复/单机实例数,自动补齐剩余节点或延后调度 - AWX Limit 与回滚覆盖全部节点,extra_vars 下发主从主机与 IP 列表 - completeTask 全节点 TCP 探活,副本经 mysqlReplicationReady 校验复制线程状态 - awxTaskSummary 按服务类型生成摘要;go-sql-driver/mysql 提为直接依赖 --- server/go.mod | 2 +- server/internal/handler/task_log.go | 11 +- server/internal/handler/task_log_test.go | 27 ++ server/internal/model/delivery.go | 27 +- server/internal/service/delivery.go | 356 +++++++++++++++++++++-- server/internal/service/delivery_test.go | 15 + 6 files changed, 392 insertions(+), 46 deletions(-) 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" }, From 69a12d778e94e36f3043546d31844d97eaba77c2 Mon Sep 17 00:00:00 2001 From: Hungerdream <1710233908@qq.com> Date: Fri, 31 Jul 2026 18:12:15 +0800 Subject: [PATCH 2/5] =?UTF-8?q?feat(ansible):=20MySQL=20=E4=BA=A4=E4=BB=98?= =?UTF-8?q?=E5=89=A7=E6=9C=AC=E6=8E=A5=E5=85=A5=20GTID=20=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E4=B8=BB=E4=BB=8E=E5=A4=8D=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - mysql_expected_hosts 按 replica_count 推导预期节点数 - 主节点判定改为 mysql_primary_host 显式指定,IP 经 hostvars 回退解析 - 副本节点等待主节点端口就绪后执行 CHANGE REPLICATION SOURCE (defaults-extra-file 临时凭据文件 + SOURCE_AUTO_POSITION + GET_SOURCE_PUBLIC_KEY) - 复制线程验证含 IO/SQL Running 与错误字段空值检查 (锚点带 ^[[:space:]]* 适配 SHOW REPLICA STATUS\G 的字段名前导空格) --- ansible/mysql-deploy-callback.yml | 77 ++++++++++++++++++++++++++---- ansible/mysql-deploy.yml | 79 +++++++++++++++++++++++++++---- 2 files changed, 139 insertions(+), 17 deletions(-) diff --git a/ansible/mysql-deploy-callback.yml b/ansible/mysql-deploy-callback.yml index 945d3f9..01d2526 100644 --- a/ansible/mysql-deploy-callback.yml +++ b/ansible/mysql-deploy-callback.yml @@ -61,7 +61,13 @@ mgr_group_name: "{{ group_replication_group_name | default('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa') }}" # --- expected node count per topology --- - mysql_expected_hosts: "{{ {'standalone': 1, 'primary_replica': 2, 'mgr_3': 3}[topology | default('standalone')] }}" + mysql_expected_hosts: "{{ 1 if (topology | default('standalone')) == 'standalone' else ((replica_count | default(1) | int) + 1 if (topology | default('standalone')) == 'primary_replica' else 3) }}" + # 平台为 primary_replica 显式传入主从节点；standalone 仍按旧逻辑工作。 + mysql_primary_host_value: "{{ mysql_primary_host | default(ansible_play_hosts_all[0]) }}" + mysql_primary_ip_value: "{{ mysql_primary_ip | default(hostvars[mysql_primary_host_value].ansible_host | default(mysql_primary_host_value)) }}" + # 交付只暴露 root 管理员；复制链路复用已配置好的 root@%。 + mysql_replication_user: "root" + mysql_replication_password: "{{ mysql_root_password_value }}" # --- secrets (prefer launch extra_vars; fall back to AWX credential-injected env) --- mysql_root_password_value: "{{ mysql_root_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD'), true) }}" @@ -355,7 +361,7 @@ mysql_node_index: "{{ ansible_play_hosts_all.index(inventory_hostname) }}" mysql_node_role: >- {{ 'standalone' if topology == 'standalone' else - ('primary' if ansible_play_hosts_all.index(inventory_hostname) == 0 else + ('primary' if inventory_hostname == mysql_primary_host_value else ('replica' if topology == 'primary_replica' else 'mgr')) }} # host-wide unique: platform may pass explicit mysql_server_id; otherwise derive # port + node index (ports are unique per host in the pool model). @@ -479,6 +485,64 @@ failed_when: false no_log: true + - name: Wait for the replica to reach the primary SQL port + ansible.builtin.wait_for: + host: "{{ mysql_primary_ip_value }}" + port: "{{ mysql_port_value }}" + timeout: 60 + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + + - name: Configure and start GTID replication on the replica + ansible.builtin.shell: + cmd: | + set -euo pipefail + client_file="$(mktemp)" + sql_file="$(mktemp)" + trap 'rm -f "$client_file" "$sql_file"' EXIT + chmod 600 "$client_file" "$sql_file" + cat >"$client_file" <<'EOF' + [client] + user=root + password={{ mysql_root_password_value }} + socket={{ mysql_run_dir }}/mysql.sock + EOF + cat >"$sql_file" <<'EOF' + STOP REPLICA; + RESET REPLICA ALL; + CHANGE REPLICATION SOURCE TO SOURCE_HOST='{{ mysql_primary_ip_value }}', SOURCE_PORT={{ mysql_port_value }}, SOURCE_USER='{{ mysql_replication_user }}', SOURCE_PASSWORD='{{ mysql_replication_password | replace("'", "''") }}', SOURCE_AUTO_POSITION=1, GET_SOURCE_PUBLIC_KEY=1; + START REPLICA; + EOF + /usr/bin/mysql --defaults-extra-file="$client_file" <"$sql_file" + executable: /bin/bash + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + no_log: true + + - name: Verify GTID replication threads + ansible.builtin.shell: + cmd: | + set -euo pipefail + client_file="$(mktemp)" + trap 'rm -f "$client_file"' EXIT + chmod 600 "$client_file" + cat >"$client_file" <<'EOF' + [client] + user=root + password={{ mysql_root_password_value }} + socket={{ mysql_run_dir }}/mysql.sock + EOF + status="$(/usr/bin/mysql --defaults-extra-file="$client_file" -e 'SHOW REPLICA STATUS\G')" + grep -q 'Replica_IO_Running: Yes' <<<"$status" + grep -q 'Replica_SQL_Running: Yes' <<<"$status" + grep -qE '^[[:space:]]*Last_IO_Error:[[:space:]]*$' <<<"$status" + grep -qE '^[[:space:]]*Last_SQL_Error:[[:space:]]*$' <<<"$status" + executable: /bin/bash + register: mysql_replica_status + retries: 12 + delay: 5 + until: mysql_replica_status.rc == 0 + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + no_log: true + - name: Notify healthcheck started ansible.builtin.uri: url: "{{ delivery_callback_url_value }}" @@ -521,13 +585,10 @@ failed_when: false no_log: true - - name: Note pending HA runtime orchestration + - name: Note pending Group Replication orchestration ansible.builtin.debug: - msg: >- - topology={{ topology }} deployed with HA-ready config (GTID/binlog/relay/GR settings in place), - but replication wiring (CHANGE REPLICATION SOURCE) and Group Replication bootstrap - (START GROUP_REPLICATION) are not yet automated — nodes start config-ready only. - when: topology != 'standalone' + msg: "MGR remains unsupported; primary_replica GTID replication is configured and healthy." + when: topology == 'mgr_3' run_once: true - name: Notify platform registration handoff diff --git a/ansible/mysql-deploy.yml b/ansible/mysql-deploy.yml index ad5f69d..143c5f1 100644 --- a/ansible/mysql-deploy.yml +++ b/ansible/mysql-deploy.yml @@ -61,7 +61,13 @@ mgr_group_name: "{{ group_replication_group_name | default('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa') }}" # --- expected node count per topology --- - mysql_expected_hosts: "{{ {'standalone': 1, 'primary_replica': 2, 'mgr_3': 3}[topology | default('standalone')] }}" + mysql_expected_hosts: "{{ 1 if (topology | default('standalone')) == 'standalone' else ((replica_count | default(1) | int) + 1 if (topology | default('standalone')) == 'primary_replica' else 3) }}" + # 平台为 primary_replica 显式传入主从节点；standalone 仍按旧逻辑工作。 + mysql_primary_host_value: "{{ mysql_primary_host | default(ansible_play_hosts_all[0]) }}" + mysql_primary_ip_value: "{{ mysql_primary_ip | default(hostvars[mysql_primary_host_value].ansible_host | default(mysql_primary_host_value)) }}" + # 交付只暴露 root 管理员；复制链路复用已配置好的 root@%。 + mysql_replication_user: "root" + mysql_replication_password: "{{ mysql_root_password_value }}" # --- secrets (prefer launch extra_vars; fall back to AWX credential-injected env) --- mysql_root_password_value: "{{ mysql_root_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD'), true) }}" @@ -258,7 +264,7 @@ mysql_node_index: "{{ ansible_play_hosts_all.index(inventory_hostname) }}" mysql_node_role: >- {{ 'standalone' if topology == 'standalone' else - ('primary' if ansible_play_hosts_all.index(inventory_hostname) == 0 else + ('primary' if inventory_hostname == mysql_primary_host_value else ('replica' if topology == 'primary_replica' else 'mgr')) }} # host-wide unique: platform may pass explicit mysql_server_id; otherwise derive # port + node index (ports are unique per host in the pool model). @@ -364,19 +370,74 @@ changed_when: false no_log: true - - name: Verify MySQL TCP health + - name: Wait for the replica to reach the primary SQL port + ansible.builtin.wait_for: + host: "{{ mysql_primary_ip_value }}" + port: "{{ mysql_port_value }}" + timeout: 60 + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + + - name: Configure and start GTID replication on the replica + ansible.builtin.shell: + cmd: | + set -euo pipefail + client_file="$(mktemp)" + sql_file="$(mktemp)" + trap 'rm -f "$client_file" "$sql_file"' EXIT + chmod 600 "$client_file" "$sql_file" + cat >"$client_file" <<'EOF' + [client] + user=root + password={{ mysql_root_password_value }} + socket={{ mysql_run_dir }}/mysql.sock + EOF + cat >"$sql_file" <<'EOF' + STOP REPLICA; + RESET REPLICA ALL; + CHANGE REPLICATION SOURCE TO SOURCE_HOST='{{ mysql_primary_ip_value }}', SOURCE_PORT={{ mysql_port_value }}, SOURCE_USER='{{ mysql_replication_user }}', SOURCE_PASSWORD='{{ mysql_replication_password | replace("'", "''") }}', SOURCE_AUTO_POSITION=1, GET_SOURCE_PUBLIC_KEY=1; + START REPLICA; + EOF + /usr/bin/mysql --defaults-extra-file="$client_file" <"$sql_file" + executable: /bin/bash + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + no_log: true + + - name: Verify GTID replication threads + ansible.builtin.shell: + cmd: | + set -euo pipefail + client_file="$(mktemp)" + trap 'rm -f "$client_file"' EXIT + chmod 600 "$client_file" + cat >"$client_file" <<'EOF' + [client] + user=root + password={{ mysql_root_password_value }} + socket={{ mysql_run_dir }}/mysql.sock + EOF + status="$(/usr/bin/mysql --defaults-extra-file="$client_file" -e 'SHOW REPLICA STATUS\G')" + grep -q 'Replica_IO_Running: Yes' <<<"$status" + grep -q 'Replica_SQL_Running: Yes' <<<"$status" + grep -qE '^[[:space:]]*Last_IO_Error:[[:space:]]*$' <<<"$status" + grep -qE '^[[:space:]]*Last_SQL_Error:[[:space:]]*$' <<<"$status" + executable: /bin/bash + register: mysql_replica_status + retries: 12 + delay: 5 + until: mysql_replica_status.rc == 0 + when: topology == 'primary_replica' and inventory_hostname != mysql_primary_host_value + no_log: true + + - name: Verify MySQL TCP health on every node ansible.builtin.wait_for: host: "{{ ansible_host | default(inventory_hostname) }}" port: "{{ mysql_port_value }}" timeout: 30 - - name: Note pending HA runtime orchestration + - name: Note pending Group Replication orchestration ansible.builtin.debug: - msg: >- - topology={{ topology }} deployed with HA-ready config (GTID/binlog/relay/GR settings in place), - but replication wiring (CHANGE REPLICATION SOURCE) and Group Replication bootstrap - (START GROUP_REPLICATION) are not yet automated — nodes start config-ready only. - when: topology != 'standalone' + msg: "MGR remains unsupported; primary_replica GTID replication is configured and healthy." + when: topology == 'mgr_3' run_once: true handlers: From 716e816a6697fca4e1b7ec7fc2051a92887410d9 Mon Sep 17 00:00:00 2001 From: Hungerdream <1710233908@qq.com> Date: Fri, 31 Jul 2026 18:12:25 +0800 Subject: [PATCH 3/5] =?UTF-8?q?feat(frontend):=20=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E6=94=AF=E6=8C=81=E4=B8=80=E4=B8=BB=E5=A4=9A?= =?UTF-8?q?=E4=BB=8E=E4=BA=A4=E4=BB=98=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MySQL 新增'一主多从'模式:有序多选主机、replica_count(1-7)输入、动态拓扑预览 - precheck 校验库存主机数与选择数量,支持留空自动调度 - 交付负载携带 topology/replica_count/target_hosts,已交付主机解析 target_host_ips 列表 - 主机选择占位符退化为常量,移除冗余三元表达式 --- frontend/src/api/delivery.ts | 4 + frontend/src/views/service/Catalog.vue | 199 ++++++++++++++++++++++--- 2 files changed, 180 insertions(+), 23 deletions(-) diff --git a/frontend/src/api/delivery.ts b/frontend/src/api/delivery.ts index 5698b03..9aa1180 100644 --- a/frontend/src/api/delivery.ts +++ b/frontend/src/api/delivery.ts @@ -25,9 +25,11 @@ export interface CreateMySQLDeliveryPayload { instance_name: string mysql_version: string topology: string + replica_count?: number mysql_port?: number data_disk: string target_host?: string + target_hosts?: string[] cpu_milli: number memory_mi: number storage_gi: number @@ -77,6 +79,8 @@ export interface DeliveryTask { instance_name: string target_host?: string target_host_ip?: string + target_hosts?: string + target_host_ips?: string mysql_port?: number postgresql_port?: number status: string diff --git a/frontend/src/views/service/Catalog.vue b/frontend/src/views/service/Catalog.vue index 1b84470..4ac0e52 100644 --- a/frontend/src/views/service/Catalog.vue +++ b/frontend/src/views/service/Catalog.vue @@ -101,9 +101,9 @@ -