feat(server): 支持 MySQL 一主多从拓扑调度与复制健康检查

- 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 提为直接依赖
2026-07-31 18:12:05 +08:00
parent 242326ccc5
commit 0ca3af39dc
6 changed files with 392 additions and 46 deletions
@@ -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(&quota).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,
@@ -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" },