-
+
+
+ {{ topologyNodes.length }} 节点{{ topologyNodes.length > 1 ? `(1 主 ${topologyNodes.length - 1} 从)` : '' }}
+ {{ topologyNodes.length > 1 ? 'GTID 异步复制 · 从节点默认只读' : '单实例部署' }}
+
+
主库复制链路
+
+
{{ activeService.icon }}
{{ node.name }}
@@ -346,6 +352,7 @@
{{ node.role }}
{{ storageLabel }} 配额
+
@@ -423,6 +430,13 @@
{{ connectionReady ? '可连接' : '等待地址' }}
{{ resultAddress }}
+
+
REPLICA ENDPOINTS
+
+ 从节点{{ endpoint }}
+
+
+
{{ isPostgreSQL ? '使用 AWX 托管的 postgres 管理员凭据连接。' : '使用交付时设置的 root 管理员账号连接。' }}
{{ databaseCommand }}
@@ -534,7 +548,7 @@ import { computed, onBeforeUnmount, onMounted, reactive, ref, watch } from 'vue'
import { ElMessage, ElMessageBox } from 'element-plus'
import { ArrowDown, Back, CircleCheck, Connection, CopyDocument, DocumentCopy, Download, Monitor, Promotion, Refresh, Setting, Tickets } from '@element-plus/icons-vue'
import { useRoute, useRouter } from 'vue-router'
-import { deliveryApi, type DeliveryMountPath, type DeliveryTarget, type DeploymentCredential, type TaskEvent } from '@/api/delivery'
+import { deliveryApi, type DeliveryMountPath, type DeliveryTarget, type DeliveryTask, type DeploymentCredential, type TaskEvent } from '@/api/delivery'
import { useAuthStore } from '@/stores/auth'
import { useBusinessLineStore } from '@/stores/businessLine'
import { useBusinessLineMockProfile } from '@/utils/businessLineMock'
@@ -617,6 +631,7 @@ interface WorkbenchSnapshot {
deploymentId: string
deliveryError: string
deliveredHost: string
+ deliveredHosts?: string[]
deliveredPort?: number
lastDeliveryStatus: string
deliveryLog: string
@@ -660,7 +675,7 @@ const basicServices = ref([
icon: 'My',
bgColor: 'var(--logo-w-bg)',
iconColor: 'var(--tag-blue-text)',
- description: '单实例标准化交付,完成后自动注册进 CloudDM / open-cdm 数据源,统一审核 SQL 上线。',
+ description: '支持单实例和两节点异步主从交付,完成后自动注册进 CloudDM / open-cdm 数据源。',
playbook: 'roles/mysql-deploy',
version: 'v8.0',
status: '可交付',
@@ -669,6 +684,7 @@ const basicServices = ref([
versions: ['MySQL 8.0', 'MySQL 8.4'],
modes: [
{ value: 'single', label: '单实例' },
+ { value: 'replica', label: '一主多从' },
],
specs: ['1C / 2G', '2C / 4G', '4C / 8G', '8C / 16G', '16C / 32G'],
disks: ['20 GB', '50 GB', '100 GB', '500 GB', '1000 GB'],
@@ -816,6 +832,7 @@ const canceling = ref(false)
const deploymentId = ref('')
const deliveryError = ref('')
const deliveredHost = ref('')
+const deliveredHosts = ref([])
const deliveredPort = ref()
const deliveryTargets = ref([])
const selectedTargetId = ref()
@@ -931,10 +948,15 @@ const topologyNodes = computed(() => {
]
}
return activeServiceKey.value === 'mysql'
- ? [
- { name: 'db-tx-021', ip: '10.24.18.21', role: '主节点' },
- { name: 'db-tx-022', ip: '10.24.18.22', role: '从节点' },
- ]
+ ? Array.from({ length: deliveryForm.replicaCount + 1 }, (_, index) => {
+ const selectedName = deliveryForm.targetHosts[index]
+ const host = targetHosts.value.find((item) => item.name === selectedName) || targetHosts.value[index]
+ return {
+ name: host?.name || `等待调度 ${index + 1}`,
+ ip: host?.ip || '由 AWX Inventory 分配',
+ role: index === 0 ? '主节点' : `从节点 ${index}`,
+ }
+ })
: [
{ name: 'ngx-tx-011', ip: '10.24.31.11', role: '主节点' },
{ name: 'ngx-tx-012', ip: '10.24.31.12', role: '备用节点' },
@@ -949,6 +971,7 @@ const runnerPreview = computed(() => {
`component: ${activeService.value?.key || '-'}`,
`instance: ${deliveryForm.instanceName}`,
`topology: ${currentModeLabel.value}`,
+ ...(deliveryForm.mode === 'replica' ? [`replica_count: ${deliveryForm.replicaCount}`] : []),
`limit: ${selectedTargetHostText.value}`,
`${isPostgreSQL.value ? 'data_root' : 'data_disk'}: ${deliveryForm.dataDisk}`,
`resources: ${deliveryForm.spec} / ${storageLabel.value}`,
@@ -968,13 +991,14 @@ const credentialEligible = computed(() => {
const revealedRootCredential = computed(() => revealedCredentials.value.find((item) => item.username === 'root'))
const resultReady = computed(() => !taskStateLoading.value && !running.value && isTerminalDeliveryStatus(lastDeliveryStatus.value))
const selectedTargetHostText = computed(() => {
- if (isPostgreSQL.value && deliveryForm.mode === 'primary_replica') {
+ if ((isPostgreSQL.value && deliveryForm.mode === 'primary_replica') || (activeServiceKey.value === 'mysql' && deliveryForm.mode === 'replica')) {
return deliveryForm.targetHosts.length ? deliveryForm.targetHosts.join(', ') : '由 AWX Inventory 分配'
}
return deliveryForm.targetHost || '由 AWX Inventory 分配'
})
const previewAddress = computed(() => `${selectedHost.value?.ip || 'AWX 自动分配'}:${deliveryForm.port || '自动端口'}`)
const resultAddress = computed(() => connectionReady.value ? `${deliveredHost.value}:${deliveredPort.value}` : '交付接口未返回地址')
+const replicaEndpoints = computed(() => deliveredHosts.value.slice(1).map((host) => `${host}:${deliveredPort.value || ''}`.replace(/:$/, '')))
const connectionUri = computed(() => {
if (!connectionReady.value) return ''
return isPostgreSQL.value ? `postgresql://postgres@${deliveredHost.value}:${deliveredPort.value}/postgres` : `mysql://root@${deliveredHost.value}:${deliveredPort.value}`
@@ -1134,6 +1158,7 @@ function persistWorkbench() {
deploymentId: deploymentId.value,
deliveryError: deliveryError.value,
deliveredHost: deliveredHost.value,
+ deliveredHosts: deliveredHosts.value,
deliveredPort: deliveredPort.value,
lastDeliveryStatus: lastDeliveryStatus.value,
deliveryLog: deliveryLog.value,
@@ -1197,6 +1222,7 @@ async function restoreWorkbench() {
deploymentId.value = snapshot.deploymentId
deliveryError.value = snapshot.deliveryError
deliveredHost.value = snapshot.deliveredHost
+ deliveredHosts.value = snapshot.deliveredHosts?.length ? snapshot.deliveredHosts : (snapshot.deliveredHost ? [snapshot.deliveredHost] : [])
deliveredPort.value = snapshot.deliveredPort
lastDeliveryStatus.value = snapshot.lastDeliveryStatus
deliveryLog.value = snapshot.deliveryLog
@@ -1209,7 +1235,7 @@ async function restoreWorkbench() {
taskStateLoading.value = true
try {
const data = await deliveryApi.getTask(snapshot.deploymentId)
- deliveredHost.value = data.task.target_host_ip || deliveredHost.value
+ updateDeliveredHosts(data.task)
deliveredPort.value = (isPostgreSQL.value ? data.task.postgresql_port : data.task.mysql_port) || deliveredPort.value
lastDeliveryStatus.value = data.task.status
credentialAvailable.value = isPostgreSQL.value ? false : data.task.credential_available
@@ -1304,6 +1330,7 @@ function resetExecutionState() {
canceling.value = false
deploymentId.value = ''
deliveredHost.value = ''
+ deliveredHosts.value = []
deliveredPort.value = undefined
deliveryLog.value = '[ready] 等待创建交付任务...'
seenEventIds.value = new Set()
@@ -1352,6 +1379,24 @@ async function precheck() {
ElMessage.warning('当前没有可用的 AWX 部署模板,请联系平台管理员配置')
return
}
+ if (activeServiceKey.value === 'mysql' && deliveryForm.mode === 'replica' && targetHosts.value.length < 2) {
+ precheckPassed.value = false
+ ElMessage.error('一主一从至少需要两个可用的 AWX Inventory 主机')
+ return
+ }
+ if (activeServiceKey.value === 'mysql' && deliveryForm.mode === 'replica') {
+ const requiredHosts = deliveryForm.replicaCount + 1
+ if (targetHosts.value.length < requiredHosts) {
+ precheckPassed.value = false
+ ElMessage.warning(`一主 ${deliveryForm.replicaCount} 从需要 ${requiredHosts} 台不同主机,当前 Inventory 只有 ${targetHosts.value.length} 台`)
+ return
+ }
+ if (deliveryForm.targetHosts.length > 0 && deliveryForm.targetHosts.length !== requiredHosts) {
+ precheckPassed.value = false
+ ElMessage.warning(`请按主节点、从节点顺序选择 ${requiredHosts} 台主机,或清空后由平台自动分配`)
+ return
+ }
+ }
if (!isPostgreSQL.value && deliveryForm.port !== undefined && (deliveryForm.port < 13306 || deliveryForm.port > 13999)) {
ElMessage.warning('手动端口必须在 13306–13999 范围内')
return
@@ -1488,10 +1533,12 @@ function mysqlDeliveryPayload(businessLineId: number) {
namespace: normalizeDNSLabel(currentName.value),
instance_name: normalizeDNSLabel(deliveryForm.instanceName),
mysql_version: mysqlVersionValue(deliveryForm.version),
- topology: 'standalone',
+ topology: deliveryForm.mode === 'replica' ? 'primary_replica' : 'standalone',
+ replica_count: deliveryForm.mode === 'replica' ? deliveryForm.replicaCount : 0,
mysql_port: deliveryForm.port || 0,
data_disk: deliveryForm.dataDisk,
target_host: deliveryForm.targetHost || undefined,
+ target_hosts: deliveryForm.mode === 'replica' && deliveryForm.targetHosts.length ? deliveryForm.targetHosts : undefined,
cpu_milli: resources.cpuMilli,
memory_mi: resources.memoryMi,
storage_gi: deliveryForm.storageGi,
@@ -1549,6 +1596,12 @@ function hostLabel(host: TargetHostOption) {
return host.ip ? `${host.name} · ${host.ip}` : host.name
}
+function updateDeliveredHosts(task: DeliveryTask) {
+ const hosts = (task.target_host_ips || '').split(',').map((host) => host.trim()).filter(Boolean)
+ deliveredHosts.value = hosts.length ? hosts : (task.target_host_ip ? [task.target_host_ip] : deliveredHosts.value)
+ deliveredHost.value = task.target_host_ip || deliveredHost.value || deliveredHosts.value[0] || ''
+}
+
function startTaskPolling(id: string, resetEvents = false) {
stopPolling()
if (resetEvents) seenEventIds.value = new Set()
@@ -1564,7 +1617,7 @@ function stopPolling() {
async function pollTask(id: string) {
try {
const data = await deliveryApi.getTask(id)
- deliveredHost.value = data.task.target_host_ip || deliveredHost.value
+ updateDeliveredHosts(data.task)
deliveredPort.value = (isPostgreSQL.value ? data.task.postgresql_port : data.task.mysql_port) || deliveredPort.value
credentialAvailable.value = isPostgreSQL.value ? false : data.task.credential_available
upsertDeliveryHistory(id, data.task.status, data.task.error_message || '', data.task.created_at)
@@ -2346,6 +2399,13 @@ h4 {
.preview-panel {
position: sticky;
top: 16px;
+ min-width: 0;
+}
+
+.preview-panel > .section-head {
+ margin-bottom: 10px;
+ padding-bottom: 12px;
+ border-bottom: 1px solid var(--line-soft);
}
.section-head,
@@ -2720,14 +2780,22 @@ h4 {
.preview-facts {
display: grid;
grid-template-columns: repeat(3, minmax(0, 1fr));
- gap: 10px;
+ gap: 0;
margin-bottom: 12px;
- padding-bottom: 12px;
+ padding: 10px 0;
border-bottom: 1px solid var(--line-soft);
}
.preview-facts div {
min-width: 0;
+ min-height: 42px;
+ padding: 0 12px;
+ border-left: 1px solid var(--line-soft);
+}
+
+.preview-facts div:first-child {
+ padding-left: 0;
+ border-left: 0;
}
.preview-facts span,
@@ -2742,12 +2810,11 @@ h4 {
}
.preview-facts strong {
- overflow: hidden;
color: var(--text-mid);
font-family: var(--mono);
font-size: 11px;
- text-overflow: ellipsis;
- white-space: nowrap;
+ line-height: 1.4;
+ overflow-wrap: anywhere;
}
.config-list {
@@ -2798,7 +2865,63 @@ h4 {
.topology {
display: grid;
gap: 10px;
+ max-height: 300px;
+ margin: 0;
+ overflow-y: auto;
+ padding-right: 4px;
+ align-content: start;
+ scrollbar-gutter: stable;
+}
+
+.topology::-webkit-scrollbar {
+ width: 6px;
+}
+
+.topology::-webkit-scrollbar-thumb {
+ border-radius: 3px;
+ background: var(--line);
+}
+
+.topology-shell {
+ position: relative;
margin-bottom: 14px;
+ padding: 12px;
+ border: 1px solid var(--line);
+ border-radius: 8px;
+ background: var(--bg-panel-2);
+}
+
+.topology-title {
+ display: flex;
+ align-items: baseline;
+ justify-content: space-between;
+ gap: 12px;
+ margin-bottom: 8px;
+ color: var(--text-strong);
+}
+
+.topology-title span {
+ color: var(--text-dim);
+ font-size: 12px;
+}
+
+.topology-connector {
+ position: relative;
+ height: 1px;
+ margin: 0 12px 10px;
+ border-top: 1px dashed var(--accent);
+ opacity: 0.65;
+}
+
+.topology-connector span {
+ position: absolute;
+ top: -9px;
+ left: 50%;
+ padding: 0 8px;
+ transform: translateX(-50%);
+ background: var(--bg-panel-2);
+ color: var(--accent);
+ font-size: 11px;
}
.topology.nodes-1 {
@@ -2813,19 +2936,44 @@ h4 {
grid-template-columns: repeat(3, minmax(0, 1fr));
}
+.topology.nodes-4,
+.topology.nodes-5,
+.topology.nodes-6,
+.topology.nodes-7,
+.topology.nodes-8 {
+ grid-template-columns: repeat(3, minmax(0, 1fr));
+}
+
.node-card {
min-width: 0;
- padding: 12px;
+ min-height: 112px;
+ padding: 10px 11px;
border: 1px solid var(--line);
border-radius: 8px;
background: var(--bg-panel-2);
}
+.node-card.primary {
+ border-color: color-mix(in srgb, var(--accent) 55%, var(--line));
+ background: color-mix(in srgb, var(--accent) 8%, var(--bg-panel-2));
+}
+
+.node-card.replica {
+ border-color: color-mix(in srgb, var(--text-dim) 45%, var(--line));
+}
+
.node-head {
display: flex;
align-items: center;
gap: 8px;
- margin-bottom: 8px;
+ margin-bottom: 9px;
+}
+
+.node-head strong {
+ min-width: 0;
+ overflow: hidden;
+ text-overflow: ellipsis;
+ white-space: nowrap;
}
.node-icon {
@@ -2843,6 +2991,7 @@ h4 {
color: var(--text-dim);
font-family: var(--mono);
font-size: 11px;
+ line-height: 1.35;
}
.yaml-preview,
@@ -2859,7 +3008,8 @@ h4 {
}
.yaml-preview {
- min-height: 180px;
+ min-height: 0;
+ max-height: 260px;
margin: 0 0 12px;
padding: 12px 14px;
}
@@ -2871,6 +3021,9 @@ h4 {
}
.form-actions {
+ margin-top: 12px;
+ padding-top: 12px;
+ border-top: 1px solid var(--line-soft);
justify-content: flex-end;
gap: 8px;
}
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..0200da6 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,69 @@ func hostTaskCount(occupied []string, name string) int {
return count
}
+// resolvePinnedPrimaryReplicaNodes 按用户显式指定的有序 target_hosts 解析 primary_replica 节点列表,
+// 首节点为主节点。返回值:节点列表、参数校验失败原因(终态)、延后调度错误(defer: 前缀,可重试)。
+func resolvePinnedPrimaryReplicaNodes(hosts []targetHost, requested []string, occupied []string, limit int, nodeCount int64, pinnedPrimary string) ([]*targetHost, string, error) {
+ if int64(len(requested)) != nodeCount {
+ return nil, fmt.Sprintf("primary_replica requires %d target_hosts, got %d", nodeCount, len(requested)), nil
+ }
+ nodes := make([]*targetHost, 0, len(requested))
+ seen := map[string]struct{}{}
+ for index, requestedHost := range requested {
+ var selected *targetHost
+ for i := range hosts {
+ if hosts[i].Name == requestedHost {
+ selected = &hosts[i]
+ break
+ }
+ }
+ if selected == nil {
+ return nil, fmt.Sprintf("target_hosts[%d] %q is not in the candidate host pool", index, requestedHost), nil
+ }
+ if _, exists := seen[selected.Name]; exists {
+ return nil, fmt.Sprintf("target_hosts contains duplicate host %q", selected.Name), nil
+ }
+ if hostTaskCount(occupied, selected.Name) >= max(limit, 1) {
+ return nil, "", 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 pinnedPrimary != "" && nodes[0].Name != pinnedPrimary {
+ return nil, "target_host must match the first primary_replica target_hosts entry", nil
+ }
+ return nodes, "", nil
+}
+
+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 +850,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 +891,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 +902,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 +966,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 +976,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 +1732,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 +1746,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 +1788,103 @@ 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 {
+ pinned, validationMsg, deferErr := resolvePinnedPrimaryReplicaNodes(meta.Hosts, payload.TargetHosts, occupied, limit, nodeCount, payload.TargetHost)
+ if deferErr != nil {
+ return deferErr
+ }
+ if validationMsg != "" {
+ return s.failInTransaction(tx, &task, model.TaskValidationFailed, validationMsg)
+ }
+ // 显式列表以首节点为主节点:覆盖此前默认挑选的空闲主机,
+ // 避免 task.TargetHost/IP 与台账、凭据指向未部署节点。
+ nodes = pinned
+ host = nodes[0]
+ } 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 +1898,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 +1914,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 +2028,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 +2096,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 +2232,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 +2268,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 +2621,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 +2667,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 +2683,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 +3053,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 +3139,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..a197443 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" },
@@ -123,6 +138,35 @@ func TestFirstFreeHost(t *testing.T) {
}
}
+func TestResolvePinnedPrimaryReplicaNodes(t *testing.T) {
+ pool := []targetHost{{Name: "node-a", IP: "10.0.0.1"}, {Name: "node-b", IP: "10.0.0.2"}, {Name: "node-c", IP: "10.0.0.3"}}
+
+ // 主节点不是池首空闲节点(firstFreeHost 会选 node-a):首节点必须以用户列表为准。
+ nodes, validationMsg, err := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-c", "node-a"}, nil, 4, 2, "")
+ if err != nil || validationMsg != "" {
+ t.Fatalf("unexpected rejection: msg=%q err=%v", validationMsg, err)
+ }
+ if nodes[0].Name != "node-c" || nodes[0].IP != "10.0.0.3" || nodes[1].Name != "node-a" {
+ t.Fatalf("primary must follow the user-specified order, got %+v", nodes)
+ }
+
+ if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a"}, nil, 4, 2, ""); msg == "" {
+ t.Fatal("host count mismatch was accepted")
+ }
+ if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-x"}, nil, 4, 2, ""); msg == "" {
+ t.Fatal("host outside the pool was accepted")
+ }
+ if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-a"}, nil, 4, 2, ""); msg == "" {
+ t.Fatal("duplicate host was accepted")
+ }
+ if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-b", "node-c"}, nil, 4, 2, "node-a"); msg == "" {
+ t.Fatal("target_host mismatching the first entry was accepted")
+ }
+ if _, _, err := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-b"}, []string{"node-b"}, 1, 2, ""); err == nil {
+ t.Fatal("saturated pinned host must defer scheduling")
+ }
+}
+
func TestAllocatePort(t *testing.T) {
if port, err := allocatePort(0, nil); err != nil || port != mysqlPortPoolStart {
t.Fatalf("expected first pool port %d, got %d err=%v", mysqlPortPoolStart, port, err)