Files
xinfra/server/internal/service/delivery.go
T
Hungerdream 95c3e98611 fix(server): 显式 target_hosts 时主机指向用户指定的主节点
- 显式列表校验通过后令 host = nodes[0],task.TargetHost/IP、台账、
  CloudDM 注册与凭据地址不再指向此前默认挑选的空闲主机
- 显式列表解析抽为纯函数 resolvePinnedPrimaryReplicaNodes,
  行为与原内联逻辑一致(校验失败终态/单机上限 defer 重试)
- 新增测试覆盖主节点非池首空闲节点及数量/池外/重复/占满场景
2026-07-31 18:49:34 +08:00

3313 lines
115 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package service
import (
"bytes"
"context"
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"crypto/sha256"
"database/sql"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"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"
)
var dnsLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`)
// AWX inventory 主机名:允许字母数字、中划线与点(FQDN)。
var hostNamePattern = regexp.MustCompile(`^[a-zA-Z0-9](?:[-a-zA-Z0-9.]*[a-zA-Z0-9])?$`)
type MySQLDeliveryInput struct {
BusinessLineID uint64 `json:"business_line_id" binding:"required"`
TargetID uint64 `json:"target_id" binding:"required"`
Namespace string `json:"namespace" binding:"required"`
InstanceName string `json:"instance_name" binding:"required"`
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"`
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"`
CharacterSet string `json:"character_set"`
Collation string `json:"collation"`
// 高级参数(选填,零值视为未设置)
MaxConnections string `json:"max_connections"`
InnodbRedoLogCapacity string `json:"innodb_redo_log_capacity"`
InnodbFlushLogAtTrxCommit *int `json:"innodb_flush_log_at_trx_commit"`
SyncBinlog *int `json:"sync_binlog"`
InnodbIOCapacity int `json:"innodb_io_capacity"`
LongQueryTime float64 `json:"long_query_time"`
BinlogExpireLogsSeconds int64 `json:"binlog_expire_logs_seconds"`
MaxBinlogSize string `json:"max_binlog_size"`
MySQLRootPassword string `json:"mysql_root_password"`
}
type deliveryPayload struct {
MySQLDeliveryInput
TargetType string `json:"target_type"`
}
type DeliveryTarget struct {
ID uint64 `json:"id"`
Name string `json:"name"`
TargetType string `json:"target_type"`
AWXInventoryID uint64 `json:"awx_inventory_id"`
AWXTemplateID uint64 `json:"awx_template_id"`
Enabled bool `json:"enabled"`
Metadata string `json:"metadata"`
}
type DeliveryMountPath struct {
Path string `json:"path"`
AvailableGi int64 `json:"available_gi"`
FSType string `json:"fstype,omitempty"`
}
type DeliveryStageEventInput struct {
Stage string `json:"stage" binding:"required"`
Status string `json:"status" binding:"required"`
Message string `json:"message"`
AWXJobID string `json:"awx_job_id"`
}
type AWXJobNotificationInput struct {
ID uint64 `json:"id"`
Status string `json:"status"`
Name string `json:"name"`
URL string `json:"url"`
Traceback string `json:"traceback"`
ExtraVars any `json:"extra_vars"`
}
type DeliveryTaskListFilter struct {
BusinessLineID uint64
Component string
ActiveOnly bool
}
type DeliveryTaskSnapshot struct {
Task *model.DeliveryTask `json:"task"`
Events []model.TaskEvent `json:"events"`
}
type MySQLServiceLedgerItem struct {
Name string `json:"name"`
Datacenter string `json:"datacenter"`
BusinessTag string `json:"business_tag"`
Instances int `json:"instances"`
Healthy int `json:"healthy"`
Address string `json:"address"`
Status string `json:"status"`
StatusClass string `json:"status_class"`
Version string `json:"version"`
Namespace string `json:"namespace"`
}
type MySQLInspectResult struct {
ID uint64 `json:"id"`
TaskID string `json:"task_id"`
InstanceName string `json:"instance_name"`
Host string `json:"host"`
Status string `json:"status"`
ServiceState string `json:"service_state"`
PortListening bool `json:"port_listening"`
ConfigExists bool `json:"config_exists"`
InstallExists bool `json:"install_exists"`
BaseExists bool `json:"base_exists"`
DataExists bool `json:"data_exists"`
RunExists bool `json:"run_exists"`
ExpectedDataDir string `json:"expected_data_dir"`
ActualDataDir string `json:"actual_data_dir"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type DeploymentCredentialView struct {
Service string `json:"service"`
InstanceName string `json:"instance_name"`
Host string `json:"host"`
Port int `json:"port"`
Username string `json:"username"`
AccountHost string `json:"account_host"`
Password string `json:"password"`
}
// targetMetadata describes the native VM候选节点池以及部署形态,由 AWX inventory hosts 动态组装。
type targetMetadata struct {
Topology string `json:"topology"`
MySQLPort int `json:"mysql_port,omitempty"`
Hosts []targetHost `json:"hosts"`
}
type targetHost struct {
Name string `json:"name"`
IP string `json:"ip"`
}
func parseTargetMetadata(raw string) targetMetadata {
meta := targetMetadata{}
if strings.TrimSpace(raw) != "" {
_ = json.Unmarshal([]byte(raw), &meta)
}
if meta.Topology == "" {
meta.Topology = "standalone"
}
if meta.MySQLPort == 0 {
meta.MySQLPort = 3307
}
return meta
}
func mountPathsFromFacts(facts map[string]any) []DeliveryMountPath {
rawMounts, ok := facts["ansible_mounts"].([]any)
if !ok {
return nil
}
items := make([]DeliveryMountPath, 0, len(rawMounts))
seen := map[string]struct{}{}
for _, raw := range rawMounts {
mount, ok := raw.(map[string]any)
if !ok {
continue
}
path := strings.TrimSpace(stringValue(mount["mount"]))
if path == "" || !strings.HasPrefix(path, "/") {
continue
}
if _, exists := seen[path]; exists {
continue
}
seen[path] = struct{}{}
items = append(items, DeliveryMountPath{
Path: path,
AvailableGi: bytesToGi(int64Value(mount["size_available"])),
FSType: strings.TrimSpace(stringValue(mount["fstype"])),
})
}
sort.Slice(items, func(i, j int) bool {
if items[i].Path == "/" {
return false
}
if items[j].Path == "/" {
return true
}
return items[i].Path < items[j].Path
})
return items
}
func stringValue(value any) string {
if value == nil {
return ""
}
switch v := value.(type) {
case string:
return v
case []byte:
return string(v)
case fmt.Stringer:
return v.String()
default:
return fmt.Sprintf("%v", value)
}
}
func int64Value(value any) int64 {
switch v := value.(type) {
case int:
return int64(v)
case int64:
return v
case float64:
return int64(v)
case json.Number:
n, _ := v.Int64()
return n
default:
return 0
}
}
func bytesToGi(bytes int64) int64 {
if bytes <= 0 {
return 0
}
return bytes / 1073741824
}
func credentialKey(secret string) []byte {
sum := sha256.Sum256([]byte(strings.TrimSpace(secret)))
return sum[:]
}
func encryptCredential(secret, plain string) (string, string, error) {
block, err := aes.NewCipher(credentialKey(secret))
if err != nil {
return "", "", err
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return "", "", err
}
nonce := make([]byte, gcm.NonceSize())
if _, err := rand.Read(nonce); err != nil {
return "", "", err
}
ciphertext := gcm.Seal(nil, nonce, []byte(plain), nil)
return base64.StdEncoding.EncodeToString(ciphertext), base64.StdEncoding.EncodeToString(nonce), nil
}
func decryptCredential(secret, ciphertextValue, nonceValue string) (string, error) {
ciphertext, err := base64.StdEncoding.DecodeString(ciphertextValue)
if err != nil {
return "", err
}
nonce, err := base64.StdEncoding.DecodeString(nonceValue)
if err != nil {
return "", err
}
block, err := aes.NewCipher(credentialKey(secret))
if err != nil {
return "", err
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return "", err
}
plain, err := gcm.Open(nil, nonce, ciphertext, nil)
if err != nil {
return "", err
}
return string(plain), nil
}
// firstFreeHost 返回候选池中非失败任务数未达单机实例上限的第一个节点;
// limit < 1 时按 1 兜底(退化为旧的一机一实例语义)。顺序遍历天然形成"先摊开、摊满一轮再叠加"。
func firstFreeHost(hosts []targetHost, occupied []string, limit int) *targetHost {
if limit < 1 {
limit = 1
}
for i := range hosts {
if hostTaskCount(occupied, hosts[i].Name) < limit {
return &hosts[i]
}
}
return nil
}
// hostTaskCount 统计某主机在占用清单(非失败任务的 target_host 列表,含重复)中的出现次数。
func hostTaskCount(occupied []string, name string) int {
count := 0
for _, h := range occupied {
if h == name {
count++
}
}
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
mysqlPortPoolEnd = 13999
)
// allocatePort 在目标主机已占用端口集上做混合分配:指定端口验冲突,未指定则取池内首个空闲端口。
func allocatePort(requested int, used []int) (int, error) {
taken := make(map[int]bool, len(used))
for _, p := range used {
taken[p] = true
}
if requested != 0 {
if taken[requested] {
return 0, fmt.Errorf("mysql_port %d is already allocated on the target host", requested)
}
return requested, nil
}
for p := mysqlPortPoolStart; p <= mysqlPortPoolEnd; p++ {
if !taken[p] {
return p, nil
}
}
return 0, fmt.Errorf("mysql port pool %d-%d is exhausted on the target host", mysqlPortPoolStart, mysqlPortPoolEnd)
}
// stdoutCacheItem 缓存 AWX Job stdout 的结果
type stdoutCacheItem struct {
stdout string
createdAt time.Time
}
type DeliveryService struct {
db *gorm.DB
cfg config.Config
awx *AWXClient
audit *AuditService
executionMu sync.Mutex
streamMu sync.Mutex
streams map[string]map[chan DeliveryTaskSnapshot]struct{}
stdoutCache map[string]*stdoutCacheItem
cacheMu sync.RWMutex
}
func (s *DeliveryService) DB() *gorm.DB { return s.db }
func NewDeliveryService(cfg config.Config, db *gorm.DB, audit *AuditService) *DeliveryService {
return &DeliveryService{
db: db,
cfg: cfg,
awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword),
audit: audit,
streams: make(map[string]map[chan DeliveryTaskSnapshot]struct{}),
stdoutCache: make(map[string]*stdoutCacheItem),
}
}
func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([]DeliveryTarget, error) {
templates, err := s.awx.ListJobTemplates(ctx)
if err != nil {
return nil, err
}
component = strings.ToLower(strings.TrimSpace(component))
if component == "mysql" {
return s.listNamedDeliveryTargets(ctx, templates, s.cfg.MySQLDeliveryTemplateName)
}
var targets []DeliveryTarget
for _, template := range templates {
if template.Inventory == 0 {
continue
}
if component != "" && component != "all" {
text := strings.ToLower(template.Name + " " + template.Description)
if !strings.Contains(text, component) {
continue
}
if component == postgresqlServiceType && strings.Contains(text, "rollback") {
continue
}
}
target, err := s.awxDeliveryTarget(ctx, template)
if err != nil {
continue
}
if component == postgresqlServiceType {
target.TargetType = "host_pool"
meta := parseTargetMetadata(target.Metadata)
meta.MySQLPort = 0
if raw, marshalErr := json.Marshal(meta); marshalErr == nil {
target.Metadata = string(raw)
}
}
targets = append(targets, target)
}
return targets, nil
}
func (s *DeliveryService) listNamedDeliveryTargets(ctx context.Context, templates []AWXJobTemplate, name string) ([]DeliveryTarget, error) {
name = strings.TrimSpace(name)
if name == "" {
return nil, fmt.Errorf("DELIVERY_MYSQL_TEMPLATE_NAME must be configured")
}
var matched []AWXJobTemplate
for _, template := range templates {
if template.Name == name {
matched = append(matched, template)
}
}
if len(matched) == 0 {
return nil, fmt.Errorf("AWX job template named %q was not found", name)
}
if len(matched) > 1 {
return nil, fmt.Errorf("multiple AWX job templates named %q found", name)
}
template := matched[0]
if template.Inventory == 0 {
return nil, fmt.Errorf("AWX job template %q does not bind an inventory", name)
}
target, err := s.awxDeliveryTarget(ctx, template)
if err != nil {
return nil, err
}
return []DeliveryTarget{target}, nil
}
func (s *DeliveryService) getTarget(ctx context.Context, templateID uint64) (DeliveryTarget, error) {
template, err := s.awx.GetJobTemplate(ctx, templateID)
if err != nil {
return DeliveryTarget{}, fmt.Errorf("deployment target is unavailable: %w", err)
}
if expected := strings.TrimSpace(s.cfg.MySQLDeliveryTemplateName); expected != "" && template.Name != expected {
return DeliveryTarget{}, fmt.Errorf("deployment target must be AWX job template %q", expected)
}
return s.awxDeliveryTarget(ctx, *template)
}
func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint64, hostName string, prefix string) ([]DeliveryMountPath, error) {
if targetID == 0 {
return nil, fmt.Errorf("target_id is required")
}
if hostName == "" || len(hostName) > 253 || !hostNamePattern.MatchString(hostName) {
return nil, fmt.Errorf("host must be a valid inventory host name")
}
prefix = strings.TrimSpace(prefix)
if prefix != "" {
if err := validateDirectoryLookupPrefix(prefix); err != nil {
return nil, err
}
}
template, err := s.awx.GetJobTemplate(ctx, targetID)
if err != nil {
return nil, fmt.Errorf("deployment target is unavailable: %w", err)
}
hosts, err := s.awx.ListInventoryHosts(ctx, template.Inventory)
if err != nil {
return nil, err
}
var matched *AWXInventoryHost
for i := range hosts {
if hosts[i].Enabled && hosts[i].Name == hostName {
matched = &hosts[i]
break
}
}
if matched == nil {
return nil, fmt.Errorf("host %q is not in the deployment target inventory", hostName)
}
if s.cfg.AWXFactsTemplateID != 0 {
stdout, err := s.refreshHostFacts(ctx, hostName, prefix)
if err != nil {
return nil, err
}
if prefix != "" {
return directoryPathsFromAWXStdout(stdout), nil
}
}
if prefix != "" {
return []DeliveryMountPath{}, nil
}
facts, err := s.awx.GetHostFacts(ctx, matched.ID)
if err != nil {
return nil, err
}
items := mountPathsFromFacts(facts)
if items == nil {
items = []DeliveryMountPath{}
}
return items, nil
}
func validateDirectoryLookupPrefix(prefix string) error {
if prefix == "" {
return nil
}
if len(prefix) > 512 {
return fmt.Errorf("path prefix is too long")
}
if !strings.HasPrefix(prefix, "/") {
return fmt.Errorf("path prefix must be an absolute path")
}
if strings.ContainsRune(prefix, 0) {
return fmt.Errorf("path prefix contains invalid characters")
}
return nil
}
func directoryPathsFromAWXStdout(stdout string) []DeliveryMountPath {
const marker = "XINFRA_PATH_COMPLETIONS_JSON="
for _, line := range strings.Split(stdout, "\n") {
line = strings.TrimSpace(line)
idx := strings.Index(line, marker)
if idx < 0 {
continue
}
if items, ok := parseDirectoryCompletionJSON(line[idx+len(marker):]); ok {
sort.Slice(items, func(i, j int) bool {
return items[i].Path < items[j].Path
})
return items
}
}
return []DeliveryMountPath{}
}
func parseDirectoryCompletionJSON(raw string) ([]DeliveryMountPath, bool) {
raw = strings.TrimSpace(raw)
candidates := []string{raw}
if strings.Contains(raw, `\"`) {
candidates = append(candidates, strings.ReplaceAll(raw, `\"`, `"`))
}
for _, candidate := range candidates {
start := strings.Index(candidate, "[")
end := strings.LastIndex(candidate, "]")
if start < 0 || end < start {
continue
}
var items []DeliveryMountPath
if err := json.Unmarshal([]byte(candidate[start:end+1]), &items); err == nil {
return items, true
}
}
return nil, false
}
func (s *DeliveryService) refreshHostFacts(ctx context.Context, hostName string, lookupPath string) (string, error) {
extraVars := map[string]any{
"target_hosts": hostName,
}
if lookupPath != "" {
extraVars["lookup_path"] = lookupPath
}
job, err := s.awx.Launch(ctx, s.cfg.AWXFactsTemplateID, AWXLaunchRequest{
Limit: hostName,
ExtraVars: extraVars,
})
if err != nil {
return "", fmt.Errorf("launch AWX facts job: %w", err)
}
done, err := s.awx.WaitJob(ctx, strconv.FormatUint(job.ID, 10), time.Duration(s.cfg.AWXFactsTimeoutSeconds)*time.Second)
if err != nil {
return "", err
}
if done.Status != "successful" || done.Failed {
return "", fmt.Errorf("AWX facts job %d finished with status %s", done.ID, done.Status)
}
stdout, err := s.awx.JobStdout(ctx, strconv.FormatUint(job.ID, 10))
if err != nil {
return "", err
}
return stdout, nil
}
func (s *DeliveryService) awxDeliveryTarget(ctx context.Context, template AWXJobTemplate) (DeliveryTarget, error) {
hosts, err := s.awx.ListInventoryHosts(ctx, template.Inventory)
if err != nil {
return DeliveryTarget{}, err
}
meta := targetMetadata{Topology: "standalone", MySQLPort: 3307}
for _, host := range hosts {
if !host.Enabled {
continue
}
meta.Hosts = append(meta.Hosts, targetHost{Name: host.Name, IP: AWXHostIP(host)})
}
raw, err := json.Marshal(meta)
if err != nil {
return DeliveryTarget{}, err
}
return DeliveryTarget{
ID: template.ID,
Name: template.Name,
TargetType: "k8s",
AWXInventoryID: template.Inventory,
AWXTemplateID: template.ID,
Enabled: true,
Metadata: string(raw),
}, nil
}
func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin bool, idempotencyKey string, input MySQLDeliveryInput) (*model.DeliveryTask, bool, error) {
idempotencyKey = strings.TrimSpace(idempotencyKey)
if idempotencyKey == "" || len(idempotencyKey) > 128 {
return nil, false, fmt.Errorf("Idempotency-Key header is required and must not exceed 128 characters")
}
normalizeMySQLDeliveryInput(&input)
if err := validateDeliveryInput(input, s.cfg.DeliveryDataDisks); err != nil {
return nil, false, err
}
credentialInput := map[string]string{
"root@%": strings.TrimSpace(input.MySQLRootPassword),
}
hasCredentialInput := credentialInput["root@%"] != ""
if hasCredentialInput {
if len(credentialInput["root@%"]) < 16 {
return nil, false, fmt.Errorf("mysql root password must be at least 16 characters")
}
if !mysqlPasswordPattern.MatchString(credentialInput["root@%"]) {
return nil, false, fmt.Errorf("mysql passwords may only contain letters and digits")
}
}
input.MySQLRootPassword = ""
var existing model.DeliveryTask
if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil {
if existing.RequestedBy != userID {
return nil, false, fmt.Errorf("idempotency key is already in use by another user")
}
return &existing, true, nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, false, err
}
target, err := s.getTarget(ctx, input.TargetID)
if err != nil {
return nil, false, err
}
if target.TargetType != "k8s" {
return nil, false, fmt.Errorf("target type %q is not supported in the MVP", target.TargetType)
}
if !isAdmin {
var count int64
if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}).
Where("business_line_id = ? AND user_id = ?", input.BusinessLineID, userID).Count(&count).Error; err != nil {
return nil, false, err
}
if count == 0 {
return nil, false, fmt.Errorf("user is not authorized for this business line")
}
}
if input.MySQLVersion == "" {
input.MySQLVersion = "8.0"
}
if input.Topology == "" {
input.Topology = "standalone"
}
if input.DataDisk == "" {
if len(s.cfg.DeliveryDataDisks) > 0 {
input.DataDisk = s.cfg.DeliveryDataDisks[0]
} else {
input.DataDisk = "/data"
}
}
payload := deliveryPayload{MySQLDeliveryInput: input, TargetType: target.TargetType}
raw, err := json.Marshal(payload)
if err != nil {
return nil, false, err
}
digest := sha256.Sum256(raw)
task := model.DeliveryTask{
ID: randomUUID(),
BusinessLineID: input.BusinessLineID,
RequestedBy: userID,
Component: "mysql",
TargetType: target.TargetType,
ServiceType: "mysql",
TargetID: target.ID,
Namespace: input.Namespace,
InstanceName: input.InstanceName,
Status: model.TaskPending,
ImmutablePayload: string(raw),
PayloadHash: hex.EncodeToString(digest[:]),
IdempotencyKey: idempotencyKey,
}
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
// Serialize name allocation per business line so concurrent requests cannot
// pass the availability check and create the same instance identity.
var businessLine model.BusinessLine
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&businessLine, input.BusinessLineID).Error; err != nil {
return err
}
if err := ensureInstanceNameAvailable(tx, input.BusinessLineID, task.Component, input.InstanceName); err != nil {
return err
}
return tx.Create(&task).Error
}); err != nil {
if lookupErr := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; lookupErr == nil {
return &existing, true, nil
}
return nil, false, err
}
if hasCredentialInput {
if err := s.storeDeploymentCredentials(ctx, &task, credentialInput, "pending"); err != nil {
return nil, false, err
}
}
_ = s.db.WithContext(ctx).Create(&model.TaskEvent{TaskID: task.ID, ToState: model.TaskPending, Message: "delivery task created"}).Error
return &task, false, nil
}
// ensureInstanceNameAvailable treats in-flight tasks, retained failed deployments,
// and active results as name reservations. A name is reusable only after rollback
// or explicit cleanup has released the corresponding task/result.
func ensureInstanceNameAvailable(tx *gorm.DB, businessLineID uint64, component, instanceName string) error {
occupiedTaskStatuses := []string{
model.TaskPending,
model.TaskValidating,
model.TaskDispatching,
model.TaskRunning,
model.TaskRegistering,
model.TaskFinished,
model.TaskRegisterFailed,
model.TaskCanceling,
model.TaskRollbackPending,
model.TaskRollingBack,
model.TaskRollbackFailed,
}
var taskCount int64
if err := tx.Model(&model.DeliveryTask{}).
Where("business_line_id = ? AND component = ? AND instance_name = ? AND status IN ?", businessLineID, component, instanceName, occupiedTaskStatuses).
Count(&taskCount).Error; err != nil {
return err
}
if taskCount > 0 {
return fmt.Errorf("instance_name %q is already in use by an active or pending delivery", instanceName)
}
var resultCount int64
if err := tx.Model(&model.DeploymentResult{}).
Where("business_line_id = ? AND component = ? AND instance_name = ? AND status IN ?", businessLineID, component, instanceName, occupiedDeploymentStatuses()).
Count(&resultCount).Error; err != nil {
return err
}
if resultCount > 0 {
return fmt.Errorf("instance_name %q is already in use by a delivered instance", instanceName)
}
return nil
}
// 版本白名单与 playbook 的 mysql_package_map 保持同步:
// 8.0 走 Ubuntu 自带源,8.4 走 MySQL 官方 APT 源;5.6/5.7 已 EOL 且无 noble 包,不支持。
var supportedMySQLVersions = map[string]bool{"8.0": true, "8.4": true}
// 拓扑白名单:支持 standalone 和一主多从 primary_replica;MGR 暂未开放。
var supportedTopologies = map[string]bool{"standalone": true, "primary_replica": true}
const rollbackLaunchTimeout = 2 * time.Minute
func occupiedDeploymentStatuses() []string {
return []string{"active", "running", "stopped", "moved", "unknown"}
}
var supportedCharsets = map[string]bool{"utf8mb4": true, "utf8": true, "gbk": true, "latin1": true}
// 高级参数档位白名单(与 docs/mysql-parameter-selection.md 保持一致)
var (
supportedMaxConnections = map[string]bool{"auto": true, "200": true, "500": true, "1000": true, "2000": true, "4000": true, "8000": true, "16000": true}
supportedLogSizes = map[string]bool{"128M": true, "256M": true, "512M": true, "1G": true}
supportedRedoLogSizes = map[string]bool{"auto": true, "128M": true, "256M": true, "512M": true, "1G": true}
supportedIOCapacities = map[int]bool{200: true, 2000: true, 5000: true}
supportedLongQueryTimes = map[float64]bool{0.5: true, 1: true, 2: true, 5: true, 10: true}
supportedBinlogExpireSecs = map[int64]bool{86400: true, 259200: true, 604800: true, 1209600: true}
)
// timezone 仅接受偏移量(±HH:MM)、SYSTEM 或命名时区(如 Asia/Shanghai)。
var timezonePattern = regexp.MustCompile(`^([+-](0\d|1[0-4]):[0-5]\d|SYSTEM|[A-Za-z]+(?:/[A-Za-z0-9_+-]+)+)$`)
var mysqlPasswordPattern = regexp.MustCompile(`^[A-Za-z0-9]+$`)
func validateDeliveryInput(input MySQLDeliveryInput, _ []string) error {
if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) {
return fmt.Errorf("namespace must be a valid Kubernetes DNS label")
}
if len(input.InstanceName) > 63 || !dnsLabelPattern.MatchString(input.InstanceName) {
return fmt.Errorf("instance_name must be a valid Kubernetes DNS label")
}
// 与文档目标态一致(memory 2048-65536 MiB / storage 20-2000 GiB),playbook assert 同步。
if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 2048 || input.MemoryMi > 65536 || input.StorageGi < 20 || input.StorageGi > 2000 {
return fmt.Errorf("requested resources are outside the supported range (memory: 2048-65536 MiB, storage: 20-2000 GiB)")
}
if input.MySQLVersion != "" && !supportedMySQLVersions[input.MySQLVersion] {
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", 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)
}
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")
}
if input.LowerCaseTableNames != nil && *input.LowerCaseTableNames != 0 && *input.LowerCaseTableNames != 1 {
return fmt.Errorf("lower_case_table_names must be 0 or 1")
}
if input.CharacterSet != "" && !supportedCharsets[input.CharacterSet] {
return fmt.Errorf("unsupported character_set %q, supported: utf8mb4, utf8, gbk, latin1", input.CharacterSet)
}
if input.Collation != "" {
charset := input.CharacterSet
if charset == "" {
charset = "utf8mb4"
}
if !strings.HasPrefix(input.Collation, charset+"_") {
return fmt.Errorf("collation %q does not match character_set %q", input.Collation, charset)
}
}
if input.MaxConnections != "" && !supportedMaxConnections[input.MaxConnections] {
return fmt.Errorf("max_connections must be one of auto, 200, 500, 1000, 2000, 4000, 8000, 16000")
}
if input.InnodbRedoLogCapacity != "" && !supportedRedoLogSizes[input.InnodbRedoLogCapacity] {
return fmt.Errorf("innodb_redo_log_capacity must be one of auto, 128M, 256M, 512M, 1G")
}
if input.InnodbFlushLogAtTrxCommit != nil && (*input.InnodbFlushLogAtTrxCommit < 0 || *input.InnodbFlushLogAtTrxCommit > 2) {
return fmt.Errorf("innodb_flush_log_at_trx_commit must be 0, 1 or 2")
}
if input.SyncBinlog != nil && *input.SyncBinlog != 0 && *input.SyncBinlog != 1 {
return fmt.Errorf("sync_binlog must be 0 or 1")
}
if input.InnodbIOCapacity != 0 && !supportedIOCapacities[input.InnodbIOCapacity] {
return fmt.Errorf("innodb_io_capacity must be one of 200, 2000, 5000")
}
if input.LongQueryTime != 0 && !supportedLongQueryTimes[input.LongQueryTime] {
return fmt.Errorf("long_query_time must be one of 0.5, 1, 2, 5, 10")
}
if input.BinlogExpireLogsSeconds != 0 && !supportedBinlogExpireSecs[input.BinlogExpireLogsSeconds] {
return fmt.Errorf("binlog_expire_logs_seconds must be one of 86400, 259200, 604800, 1209600")
}
if input.MaxBinlogSize != "" && !supportedLogSizes[input.MaxBinlogSize] {
return fmt.Errorf("max_binlog_size must be one of 128M, 256M, 512M, 1G")
}
return nil
}
func normalizeMySQLDeliveryInput(input *MySQLDeliveryInput) {
input.Namespace = normalizeDNSLabel(input.Namespace)
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"
}
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
}
if input.MemoryMi == 0 && input.MemoryGB != 0 {
input.MemoryMi = input.MemoryGB * 1024
}
if input.StorageGi == 0 && input.StorageGB != 0 {
input.StorageGi = input.StorageGB
}
}
func normalizeDNSLabel(value string) string {
value = strings.ToLower(strings.TrimSpace(value))
value = regexp.MustCompile(`[^a-z0-9-]+`).ReplaceAllString(value, "-")
value = regexp.MustCompile(`-+`).ReplaceAllString(value, "-")
value = strings.Trim(value, "-")
if len(value) > 63 {
value = strings.Trim(value[:63], "-")
}
return value
}
func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin bool, filter DeliveryTaskListFilter) ([]model.DeliveryTask, error) {
query := s.db.WithContext(ctx).Order("created_at DESC")
if filter.BusinessLineID != 0 {
query = query.Where("business_line_id = ?", filter.BusinessLineID)
}
if filter.Component != "" {
query = query.Where("component = ?", filter.Component)
}
if filter.ActiveOnly {
query = query.Where("status NOT IN ?", terminalTaskStatuses())
}
if !isAdmin {
query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID))
}
var tasks []model.DeliveryTask
if err := query.Find(&tasks).Error; err != nil {
return nil, err
}
if err := s.fillCredentialAvailability(ctx, tasks); err != nil {
return nil, err
}
return tasks, nil
}
func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]MySQLServiceLedgerItem, error) {
if businessLineID == 0 {
return nil, fmt.Errorf("business_line_id is required")
}
if err := s.ensureBusinessLineAccess(ctx, userID, isAdmin, businessLineID); err != nil {
return nil, err
}
var businessLine model.BusinessLine
if err := s.db.WithContext(ctx).First(&businessLine, "id = ?", businessLineID).Error; err != nil {
return nil, err
}
var instances []model.DeploymentResult
if err := s.db.WithContext(ctx).
Where("business_line_id = ? AND component = ? AND service_type = ? AND status IN ?", businessLineID, "mysql", "database", []string{"active", "running", "stopped", "moved", "unknown"}).
Order("created_at DESC").
Find(&instances).Error; err != nil {
return nil, err
}
taskIDs := make([]string, 0, len(instances))
for _, instance := range instances {
taskIDs = append(taskIDs, instance.TaskID)
}
taskStatuses := map[string]string{}
if len(taskIDs) > 0 {
var tasks []model.DeliveryTask
if err := s.db.WithContext(ctx).Select("id", "status").Where("id IN ?", taskIDs).Find(&tasks).Error; err != nil {
return nil, err
}
for _, task := range tasks {
taskStatuses[task.ID] = task.Status
}
}
items := make([]MySQLServiceLedgerItem, 0, len(instances))
for _, instance := range instances {
s.ensureMySQLDeploymentMetadata(ctx, &instance)
status := "健康"
statusClass := "ok"
healthy := 1
switch instance.Status {
case "stopped":
status = "已关闭"
statusClass = "warn"
healthy = 0
case "moved":
status = "路径变更"
statusClass = "warn"
healthy = 0
case "unknown":
status = "未知"
statusClass = "warn"
healthy = 0
}
if taskStatuses[instance.TaskID] == model.TaskRegisterFailed && healthy == 1 {
status = "注册异常"
statusClass = "warn"
}
if instance.Host == "" || instance.Port == 0 {
status = "部分异常"
statusClass = "warn"
healthy = 0
}
items = append(items, MySQLServiceLedgerItem{
Name: instance.InstanceName,
Datacenter: instance.NodeName,
BusinessTag: businessLine.Name,
Instances: 1,
Healthy: healthy,
Address: fmt.Sprintf("%s:%d", instance.Host, instance.Port),
Status: status,
StatusClass: statusClass,
Version: instance.Version,
Namespace: instance.Namespace,
})
}
return items, nil
}
func (s *DeliveryService) SyncMySQLInstanceStatuses(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]MySQLInspectResult, error) {
if businessLineID == 0 {
return nil, fmt.Errorf("business_line_id is required")
}
if err := s.ensureBusinessLineAccess(ctx, userID, isAdmin, businessLineID); err != nil {
return nil, err
}
inspectTemplateName := strings.TrimSpace(s.cfg.MySQLInspectTemplateName)
if inspectTemplateName == "" {
return nil, fmt.Errorf("DELIVERY_MYSQL_INSPECT_TEMPLATE_NAME must be configured")
}
inspectTemplate, err := s.awxJobTemplateByName(ctx, inspectTemplateName)
if err != nil {
return nil, err
}
var instances []model.DeploymentResult
if err := s.db.WithContext(ctx).
Where("business_line_id = ? AND component = ? AND service_type = ? AND status IN ?", businessLineID, "mysql", "database", occupiedDeploymentStatuses()).
Order("target_id ASC, created_at DESC").
Find(&instances).Error; err != nil {
return nil, err
}
if len(instances) == 0 {
return []MySQLInspectResult{}, nil
}
byTarget := map[uint64][]model.DeploymentResult{}
for i := range instances {
s.ensureMySQLDeploymentMetadata(ctx, &instances[i])
byTarget[instances[i].TargetID] = append(byTarget[instances[i].TargetID], instances[i])
}
out := make([]MySQLInspectResult, 0, len(instances))
for targetID, group := range byTarget {
targetTemplate, err := s.awx.GetJobTemplate(ctx, targetID)
if err != nil {
return nil, fmt.Errorf("load AWX target template %d: %w", targetID, err)
}
extraInstances := make([]map[string]any, 0, len(group))
hosts := make([]string, 0, len(group))
seenHosts := map[string]struct{}{}
for _, instance := range group {
meta := metadataMap(instance.Metadata)
host := strings.TrimSpace(instance.NodeName)
if host == "" {
host = strings.TrimSpace(instance.Host)
}
if host != "" {
if _, ok := seenHosts[host]; !ok {
hosts = append(hosts, host)
seenHosts[host] = struct{}{}
}
}
extraInstances = append(extraInstances, map[string]any{
"id": instance.ID,
"task_id": instance.TaskID,
"instance_name": instance.InstanceName,
"target_host": host,
"host": instance.Host,
"port": instance.Port,
"data_disk": stringValue(meta["data_disk"]),
"base_dir": stringValue(meta["base_dir"]),
"install_dir": stringValue(meta["install_dir"]),
"data_dir": stringValue(meta["data_dir"]),
"run_dir": stringValue(meta["run_dir"]),
"config_file": stringValue(meta["config_file"]),
"service_name": stringValue(meta["service_name"]),
})
}
sort.Strings(hosts)
job, err := s.awx.Launch(ctx, inspectTemplate.ID, AWXLaunchRequest{
InventoryID: targetTemplate.Inventory,
Limit: strings.Join(hosts, ","),
ExtraVars: map[string]any{
"target_hosts": strings.Join(hosts, ","),
"mysql_instances": extraInstances,
},
})
if err != nil {
return nil, fmt.Errorf("launch AWX MySQL inspect job: %w", err)
}
done, err := s.awx.WaitJob(ctx, strconv.FormatUint(job.ID, 10), time.Duration(s.cfg.MySQLInspectTimeoutSeconds)*time.Second)
if err != nil {
return nil, err
}
stdout, err := s.awx.JobStdout(ctx, strconv.FormatUint(job.ID, 10))
if err != nil {
return nil, err
}
if done.Status != "successful" {
return nil, fmt.Errorf("AWX MySQL inspect job %d finished with status %s: %s", done.ID, done.Status, truncateForEvent(stdout, 1000))
}
results := parseMySQLInspectResults(stdout)
if len(results) == 0 {
return nil, fmt.Errorf("AWX MySQL inspect job %d returned no machine-readable results", job.ID)
}
if err := s.applyMySQLInspectResults(ctx, results); err != nil {
return nil, err
}
out = append(out, results...)
}
return out, nil
}
func (s *DeliveryService) awxJobTemplateByName(ctx context.Context, name string) (*AWXJobTemplate, error) {
name = strings.TrimSpace(name)
if name == "" {
return nil, fmt.Errorf("AWX job template name is required")
}
templates, err := s.awx.ListJobTemplates(ctx)
if err != nil {
return nil, err
}
var matched *AWXJobTemplate
for i := range templates {
if templates[i].Name != name {
continue
}
if matched != nil {
return nil, fmt.Errorf("multiple AWX job templates named %q found", name)
}
item := templates[i]
matched = &item
}
if matched == nil {
return nil, fmt.Errorf("AWX job template named %q was not found", name)
}
return matched, nil
}
func (s *DeliveryService) ensureBusinessLineAccess(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) error {
if isAdmin {
return nil
}
var count int64
if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}).
Where("business_line_id = ? AND user_id = ?", businessLineID, userID).
Count(&count).Error; err != nil {
return err
}
if count == 0 {
return fmt.Errorf("user is not authorized for this business line")
}
return nil
}
func parseMySQLInspectResults(stdout string) []MySQLInspectResult {
const marker = "XINFRA_MYSQL_INSPECT_RESULT_B64="
var out []MySQLInspectResult
for _, line := range strings.Split(stdout, "\n") {
idx := strings.Index(line, marker)
if idx < 0 {
continue
}
token := strings.Trim(strings.TrimSpace(line[idx+len(marker):]), "\\\",")
end := 0
for end < len(token) {
ch := token[end]
if (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '+' || ch == '/' || ch == '=' {
end++
continue
}
break
}
token = token[:end]
rawBytes, err := base64.StdEncoding.DecodeString(token)
if err != nil {
continue
}
var items []MySQLInspectResult
if err := json.Unmarshal(rawBytes, &items); err == nil {
out = append(out, items...)
}
}
return out
}
func (s *DeliveryService) applyMySQLInspectResults(ctx context.Context, results []MySQLInspectResult) error {
now := time.Now()
for _, result := range results {
status := normalizeMySQLInspectStatus(result.Status)
if result.ID == 0 || status == "" {
continue
}
var current model.DeploymentResult
if err := s.db.WithContext(ctx).First(&current, "id = ?", result.ID).Error; err != nil {
return err
}
meta := metadataMap(current.Metadata)
meta["last_probe"] = map[string]any{
"checked_at": now.Format(time.RFC3339),
"status": status,
"host": result.Host,
"service_state": result.ServiceState,
"port_listening": result.PortListening,
"config_exists": result.ConfigExists,
"install_exists": result.InstallExists,
"base_exists": result.BaseExists,
"data_exists": result.DataExists,
"run_exists": result.RunExists,
"expected_data_dir": result.ExpectedDataDir,
"actual_data_dir": result.ActualDataDir,
}
if status == "moved" && result.ActualDataDir != "" {
meta["actual_data_dir"] = result.ActualDataDir
}
result.Metadata = meta
if err := s.db.WithContext(ctx).Model(&model.DeploymentResult{}).
Where("id = ?", result.ID).
Updates(map[string]any{"status": status, "metadata": string(mustJSON(meta)), "updated_at": now}).Error; err != nil {
return err
}
if status == "deleted" {
if err := s.db.WithContext(ctx).Model(&model.ResourceUsage{}).
Where("instance_id = ? AND status = ?", result.ID, "active").
Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
return err
}
dataSourceID, deleteStatus := cloudDMDeleteInfo(meta)
if dataSourceID > 0 && deleteStatus != "deleted" {
if err := s.deleteCloudDMDataSource(ctx, current); err != nil {
meta = updateCloudDMDeleteMetadata(meta, "failed", err.Error(), "")
if updateErr := s.db.WithContext(ctx).Model(&model.DeploymentResult{}).
Where("id = ?", result.ID).
Update("metadata", string(mustJSON(meta))).Error; updateErr != nil {
return updateErr
}
return err
}
meta = updateCloudDMDeleteMetadata(meta, "deleted", "", now.Format(time.RFC3339))
if err := s.db.WithContext(ctx).Model(&model.DeploymentResult{}).
Where("id = ?", result.ID).
Update("metadata", string(mustJSON(meta))).Error; err != nil {
return err
}
}
}
}
return nil
}
func normalizeMySQLInspectStatus(status string) string {
switch strings.ToLower(strings.TrimSpace(status)) {
case "running", "stopped", "deleted", "moved", "unknown":
return strings.ToLower(strings.TrimSpace(status))
default:
return "unknown"
}
}
func truncateForEvent(value string, limit int) string {
value = strings.TrimSpace(value)
if limit <= 0 || len(value) <= limit {
return value
}
return value[:limit] + "..."
}
func (s *DeliveryService) storeDeploymentCredentials(ctx context.Context, task *model.DeliveryTask, credentials map[string]string, status string) error {
for key, password := range credentials {
username, host, ok := strings.Cut(key, "@")
if !ok || strings.TrimSpace(username) == "" {
return fmt.Errorf("invalid credential account %q", key)
}
ciphertext, nonce, err := encryptCredential(s.cfg.DeliveryCredentialSecret, password)
if err != nil {
return err
}
item := model.DeploymentCredential{
TaskID: task.ID,
BusinessLineID: task.BusinessLineID,
Component: task.Component,
InstanceName: task.InstanceName,
Username: strings.TrimSpace(username),
AccountHost: strings.TrimSpace(host),
Ciphertext: ciphertext,
Nonce: nonce,
Status: status,
Source: "user_input",
}
if item.Component == "" {
item.Component = "mysql"
}
if err := s.db.WithContext(ctx).Create(&item).Error; err != nil {
return err
}
}
return nil
}
func (s *DeliveryService) deploymentCredentialVars(ctx context.Context, taskID string) (map[string]string, error) {
var items []model.DeploymentCredential
if err := s.db.WithContext(ctx).
Where("task_id = ? AND status IN ?", taskID, []string{"pending", "available"}).
Find(&items).Error; err != nil {
return nil, err
}
if len(items) == 0 {
return map[string]string{}, nil
}
values := map[string]string{}
for _, item := range items {
password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, item.Ciphertext, item.Nonce)
if err != nil {
return nil, err
}
switch item.Username + "@" + item.AccountHost {
case "root@%", "root@localhost":
values["mysql_root_password"] = password
}
}
if values["mysql_root_password"] == "" {
return nil, fmt.Errorf("root deployment credential is missing for task %s", taskID)
}
return values, nil
}
func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uint64, isAdmin bool) (*model.DeliveryTask, []model.TaskEvent, error) {
query := s.db.WithContext(ctx).Where("id = ?", taskID)
if !isAdmin {
query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID))
}
var task model.DeliveryTask
if err := query.First(&task).Error; err != nil {
return nil, nil, err
}
taskItems := []model.DeliveryTask{task}
if err := s.fillCredentialAvailability(ctx, taskItems); err != nil {
return nil, nil, err
}
task = taskItems[0]
var events []model.TaskEvent
if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).Order("id ASC").Find(&events).Error; err != nil {
return nil, nil, err
}
return &task, events, nil
}
func (s *DeliveryService) fillCredentialAvailability(ctx context.Context, tasks []model.DeliveryTask) error {
if len(tasks) == 0 {
return nil
}
taskIDs := make([]string, 0, len(tasks))
for i := range tasks {
if tasks[i].Status == model.TaskFinished || tasks[i].Status == model.TaskRegisterFailed {
taskIDs = append(taskIDs, tasks[i].ID)
}
}
if len(taskIDs) == 0 {
return nil
}
var rows []struct {
TaskID string
Count int64
}
if err := s.db.WithContext(ctx).
Model(&model.DeploymentCredential{}).
Select("task_id, count(*) as count").
Where("task_id IN ? AND status = ?", taskIDs, "available").
Group("task_id").
Scan(&rows).Error; err != nil {
return err
}
available := make(map[string]bool, len(rows))
for _, row := range rows {
available[row.TaskID] = row.Count > 0
}
for i := range tasks {
tasks[i].CredentialAvailable = available[tasks[i].ID]
}
return nil
}
func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskID string, userID uint64, isAdmin bool) ([]DeploymentCredentialView, error) {
task, _, err := s.GetTask(ctx, taskID, userID, isAdmin)
if err != nil {
return nil, err
}
if !isMySQLServiceType(task.ServiceType) {
return nil, fmt.Errorf("task service type %q does not provide MySQL credentials", task.ServiceType)
}
if task.Status != model.TaskFinished && task.Status != model.TaskRegisterFailed {
return nil, fmt.Errorf("task credentials are available only after a successful deployment")
}
var out []DeploymentCredentialView
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var result model.DeploymentResult
if err := tx.Where("task_id = ?", taskID).First(&result).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
var credentials []model.DeploymentCredential
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
Where("task_id = ? AND status = ?", taskID, "available").
Order("id ASC").
Find(&credentials).Error; err != nil {
return err
}
if len(credentials) == 0 {
return gorm.ErrRecordNotFound
}
for _, credential := range credentials {
password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, credential.Ciphertext, credential.Nonce)
if err != nil {
return err
}
out = append(out, DeploymentCredentialView{
Service: firstNonEmptyDeliveryValue(result.Component, credential.Component, task.Component, "mysql"),
InstanceName: firstNonEmptyDeliveryValue(result.InstanceName, credential.InstanceName, task.InstanceName),
Host: firstNonEmptyDeliveryValue(result.Host, task.TargetHostIP),
Port: firstNonZero(result.Port, task.MySQLPort),
Username: credential.Username,
AccountHost: credential.AccountHost,
Password: password,
})
}
ids := make([]uint64, 0, len(credentials))
for _, credential := range credentials {
ids = append(ids, credential.ID)
}
return tx.Unscoped().Where("id IN ?", ids).Delete(&model.DeploymentCredential{}).Error
})
if err != nil {
return nil, err
}
return out, nil
}
func firstNonEmptyDeliveryValue(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return strings.TrimSpace(value)
}
}
return ""
}
func firstNonZero(values ...int) int {
for _, value := range values {
if value != 0 {
return value
}
}
return 0
}
func (s *DeliveryService) SubscribeTask(taskID string) (<-chan DeliveryTaskSnapshot, func()) {
ch := make(chan DeliveryTaskSnapshot, 8)
s.streamMu.Lock()
if s.streams[taskID] == nil {
s.streams[taskID] = make(map[chan DeliveryTaskSnapshot]struct{})
}
s.streams[taskID][ch] = struct{}{}
s.streamMu.Unlock()
cancel := func() {
s.streamMu.Lock()
if subscribers := s.streams[taskID]; subscribers != nil {
delete(subscribers, ch)
if len(subscribers) == 0 {
delete(s.streams, taskID)
}
}
s.streamMu.Unlock()
close(ch)
}
return ch, cancel
}
func (s *DeliveryService) taskSnapshot(ctx context.Context, taskID string) (DeliveryTaskSnapshot, error) {
task, events, err := s.GetTask(ctx, taskID, 0, true)
if err != nil {
return DeliveryTaskSnapshot{}, err
}
return DeliveryTaskSnapshot{Task: task, Events: events}, nil
}
func (s *DeliveryService) broadcastTask(ctx context.Context, taskID string) {
snapshot, err := s.taskSnapshot(ctx, taskID)
if err != nil {
return
}
// 任务状态变化时清除 stdout 缓存
s.invalidateStdoutCacheForTask(taskID)
s.streamMu.Lock()
defer s.streamMu.Unlock()
// 收集需要清理的 closed channel
var closedChannels []chan DeliveryTaskSnapshot
for ch := range s.streams[taskID] {
// 使用 recover 捕获 send on closed channel 的错误
func() {
defer func() {
if r := recover(); r != nil {
// channel 已关闭,标记需要清理
closedChannels = append(closedChannels, ch)
}
}()
select {
case ch <- snapshot:
default:
}
}()
}
// 清理 closed channels
for _, ch := range closedChannels {
delete(s.streams[taskID], ch)
}
if len(s.streams[taskID]) == 0 {
delete(s.streams, taskID)
}
}
// invalidateStdoutCacheForTask 清除与任务相关的 stdout 缓存
func (s *DeliveryService) invalidateStdoutCacheForTask(taskID string) {
var execution model.ExecutionJob
var rollback model.RollbackJob
s.cacheMu.Lock()
defer s.cacheMu.Unlock()
// 清除 execution job 的缓存
if err := s.db.Where("task_id = ?", taskID).First(&execution).Error; err == nil && execution.ExecutorJobID != "" {
delete(s.stdoutCache, execution.ExecutorJobID)
}
// 清除 rollback job 的缓存
if err := s.db.Where("task_id = ?", taskID).First(&rollback).Error; err == nil && rollback.ExecutorJobID != "" {
delete(s.stdoutCache, rollback.ExecutorJobID)
}
}
const stdoutCacheTTL = 30 * time.Second
func (s *DeliveryService) AWXJobStdout(ctx context.Context, jobID string) (string, error) {
// 检查缓存
s.cacheMu.RLock()
if item, ok := s.stdoutCache[jobID]; ok {
if time.Since(item.createdAt) < stdoutCacheTTL {
s.cacheMu.RUnlock()
return item.stdout, nil
}
}
s.cacheMu.RUnlock()
// 缓存未命中或已过期,重新获取
stdout, err := s.awx.JobStdout(ctx, jobID)
if err != nil {
return "", err
}
// 更新缓存
s.cacheMu.Lock()
s.stdoutCache[jobID] = &stdoutCacheItem{
stdout: stdout,
createdAt: time.Now(),
}
s.cacheMu.Unlock()
return stdout, nil
}
// InvalidateStdoutCache 清除指定 jobID 的 stdout 缓存
func (s *DeliveryService) InvalidateStdoutCache(jobID string) {
s.cacheMu.Lock()
delete(s.stdoutCache, jobID)
s.cacheMu.Unlock()
}
func (s *DeliveryService) Cancel(ctx context.Context, taskID string, userID uint64, isAdmin bool) error {
task, _, err := s.GetTask(ctx, taskID, userID, isAdmin)
if err != nil {
return err
}
if task.Status == model.TaskPending {
return s.transition(ctx, task, model.TaskCanceled, "canceled before dispatch", "")
}
if task.Status != model.TaskRunning && task.Status != model.TaskDispatching {
return fmt.Errorf("task in state %q cannot be canceled", task.Status)
}
var job model.ExecutionJob
if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&job).Error; err != nil {
return err
}
if err := s.awx.Cancel(ctx, job.ExecutorJobID); err != nil {
return err
}
return s.transition(ctx, task, model.TaskCanceling, "cancel requested in AWX", "")
}
func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryTask, error) {
var task model.DeliveryTask
var payload deliveryPayload
var target DeliveryTarget
dispatchable := false
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ? AND (service_type = '' OR service_type = 'mysql')", model.TaskPending).Order("created_at ASC").First(&task).Error; err != nil {
return err
}
var targetErr error
target, targetErr = s.getTarget(ctx, task.TargetID)
if targetErr != nil {
return s.failInTransaction(tx, &task, model.TaskValidationFailed, targetErr.Error())
}
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return s.failInTransaction(tx, &task, model.TaskValidationFailed, "stored deployment payload is invalid")
}
activeStates := []string{model.TaskValidating, model.TaskDispatching, model.TaskRunning, model.TaskCanceling}
checks := []struct {
query string
args []any
limit int
message string
}{
{"status IN ?", []any{activeStates}, s.cfg.DeliveryGlobalLimit, "global concurrency limit reached"},
{"status IN ? AND target_id = ?", []any{activeStates, task.TargetID}, s.cfg.DeliveryTargetLimit, "target concurrency limit reached"},
{"status IN ? AND target_id = ? AND namespace = ?", []any{activeStates, task.TargetID, task.Namespace}, 1, "namespace already has an active MySQL delivery"},
}
for _, check := range checks {
var count int64
if check.limit > 0 {
if err := tx.Model(&model.DeliveryTask{}).Where(check.query, check.args...).Count(&count).Error; err != nil {
return err
}
if count >= int64(check.limit) {
return fmt.Errorf("defer: %s", check.message)
}
}
}
nodeCount := topologyNodeCount(payload.MySQLDeliveryInput)
quotaOK, err := checkResourceQuota(tx, task.BusinessLineID, task.TargetID, payload, nodeCount)
if err != nil {
return err
}
if !quotaOK {
return s.failInTransaction(tx, &task, model.TaskValidationFailed, "resource quota is insufficient")
}
meta := parseTargetMetadata(target.Metadata)
if len(meta.Hosts) == 0 {
return s.failInTransaction(tx, &task, model.TaskValidationFailed, "deployment target has no candidate hosts")
}
var occupied []string
occupiedExclude := allocationReleasedTaskStatuses()
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 != "" {
// 点名主机:仅校验池内存在性与单机实例上限,端口/配额/实机守卫照常执行。
for i := range meta.Hosts {
if meta.Hosts[i].Name == payload.TargetHost {
host = &meta.Hosts[i]
break
}
}
if host == nil {
return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_host %q is not in the candidate host pool", payload.TargetHost))
}
if effective := max(limit, 1); hostTaskCount(occupied, host.Name) >= effective {
return fmt.Errorf("defer: pinned host %s reached the per-host instance limit %d", host.Name, effective)
}
} else {
host = firstFreeHost(meta.Hosts, occupied, limit)
}
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
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
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())
}
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, "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
}
dispatchable = true
return nil
})
if err == nil && !dispatchable {
err = gorm.ErrRecordNotFound
}
return &task, err
}
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) {
return true, nil
}
return false, err
}
type totals struct{ CPU, Memory, Storage, Instances int64 }
var used, reserved totals
if err := tx.Model(&model.ResourceUsage{}).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 = ?", businessLineID, targetID, "active").Scan(&used).Error; err != nil {
return false, err
}
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*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 {
if err := s.transitionTx(tx, task, status, message, message); err != nil {
return err
}
return nil
}
func (s *DeliveryService) transition(ctx context.Context, task *model.DeliveryTask, status, message, errorMessage string) error {
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return s.transitionTx(tx, task, status, message, errorMessage) })
}
func (s *DeliveryService) transitionTx(tx *gorm.DB, task *model.DeliveryTask, status, message, errorMessage string) error {
from := task.Status
updates := map[string]any{"status": status, "error_message": errorMessage}
now := time.Now()
if status == model.TaskRunning {
updates["started_at"] = now
}
if status == model.TaskFinished || status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled || status == model.TaskRegisterFailed || status == model.TaskRolledBack || status == model.TaskRollbackFailed || status == model.TaskRollbackAck {
updates["finished_at"] = now
}
result := tx.Model(&model.DeliveryTask{}).Where("id = ? AND status = ?", task.ID, from).Updates(updates)
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return fmt.Errorf("task %s changed concurrently", task.ID)
}
task.Status = status
task.ErrorMessage = errorMessage
return tx.Create(&model.TaskEvent{TaskID: task.ID, FromState: from, ToState: status, Message: message}).Error
}
func (s *DeliveryService) releaseReservation(tx *gorm.DB, taskID string) error {
return tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "reserved").Update("status", "released").Error
}
func randomUUID() string {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
panic(err)
}
b[6] = (b[6] & 0x0f) | 0x40
b[8] = (b[8] & 0x3f) | 0x80
return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16])
}
func mustJSON(value any) []byte {
raw, err := json.Marshal(value)
if err != nil {
return []byte(`{}`)
}
return raw
}
func metadataMap(raw string) map[string]any {
out := map[string]any{}
if strings.TrimSpace(raw) != "" {
_ = json.Unmarshal([]byte(raw), &out)
}
return out
}
func mysqlDeploymentMetadata(task model.DeliveryTask, payload deliveryPayload) map[string]any {
dataDisk := strings.TrimRight(strings.TrimSpace(payload.DataDisk), "/")
if dataDisk == "" {
dataDisk = "/data"
}
instance := payload.InstanceName
if instance == "" {
instance = task.InstanceName
}
mysqlVersion := payload.MySQLVersion
if mysqlVersion == "" {
mysqlVersion = "8.0"
}
topology := payload.Topology
if topology == "" {
topology = "standalone"
}
timezone := payload.Timezone
if timezone == "" {
timezone = "+08:00"
}
lowerCaseTableNames := 1
if payload.LowerCaseTableNames != nil {
lowerCaseTableNames = *payload.LowerCaseTableNames
}
characterSet := payload.CharacterSet
if characterSet == "" {
characterSet = "utf8mb4"
}
collation := payload.Collation
if collation == "" {
collation = "utf8mb4_general_ci"
}
baseDir := dataDisk + "/mysql-delivery/" + instance
return map[string]any{
"metadata_version": 2,
"component": "mysql",
"service_type": "database",
"business_line_id": task.BusinessLineID,
"target_id": task.TargetID,
"target_type": task.TargetType,
"namespace": payload.Namespace,
"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,
"data_disk": dataDisk,
"base_dir": baseDir,
"install_dir": "/opt/mysql-delivery/" + instance,
"data_dir": baseDir + "/data",
"log_dir": baseDir + "/logs",
"binlog_dir": baseDir + "/logs/binlog",
"redo_dir": baseDir + "/logs/redo",
"tmp_dir": baseDir + "/tmp",
"run_dir": "/run/mysql-delivery-" + instance,
"config_file": "/etc/mysql/mysql-delivery/" + instance + ".cnf",
"service_name": "mysql-delivery@" + instance + ".service",
"timezone": timezone,
"lower_case_table_names": lowerCaseTableNames,
"character_set": characterSet,
"collation": collation,
"max_connections": payload.MaxConnections,
"innodb_redo_log_capacity": payload.InnodbRedoLogCapacity,
"innodb_flush_log_at_trx_commit": payload.InnodbFlushLogAtTrxCommit,
"sync_binlog": payload.SyncBinlog,
"innodb_io_capacity": payload.InnodbIOCapacity,
"long_query_time": payload.LongQueryTime,
"binlog_expire_logs_seconds": payload.BinlogExpireLogsSeconds,
"max_binlog_size": payload.MaxBinlogSize,
}
}
func (s *DeliveryService) ensureMySQLDeploymentMetadata(ctx context.Context, result *model.DeploymentResult) map[string]any {
meta := metadataMap(result.Metadata)
if fmt.Sprint(meta["metadata_version"]) == "2" && strings.TrimSpace(stringValue(meta["data_dir"])) != "" && strings.TrimSpace(stringValue(meta["config_file"])) != "" {
return meta
}
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", result.TaskID).Error; err != nil {
return meta
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return meta
}
enriched := mysqlDeploymentMetadata(task, payload)
for key, value := range meta {
enriched[key] = value
}
if raw := string(mustJSON(enriched)); raw != result.Metadata {
_ = s.db.WithContext(ctx).Model(result).Update("metadata", raw).Error
result.Metadata = raw
}
return enriched
}
func mysqlReady(ctx context.Context, address string) error {
dialer := net.Dialer{Timeout: 5 * time.Second}
conn, err := dialer.DialContext(ctx, "tcp", address)
if err != nil {
return err
}
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 {
if errors.Is(err, gorm.ErrRecordNotFound) || strings.HasPrefix(err.Error(), "defer:") {
return nil
}
return err
}
_, _, err = s.CreateExecution(ctx, task.ID, task.PayloadHash, task.IdempotencyKey)
if err != nil {
failErr := s.beginRollback(ctx, task.ID, "deployment could not be started: "+err.Error())
s.broadcastTask(ctx, task.ID)
return failErr
}
s.broadcastTask(ctx, task.ID)
return nil
}
func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHash, idempotencyKey string) (*model.ExecutionJob, bool, error) {
s.executionMu.Lock()
defer s.executionMu.Unlock()
var existing model.ExecutionJob
if err := s.db.WithContext(ctx).Where("task_id = ? OR idempotency_key = ?", taskID, idempotencyKey).First(&existing).Error; err == nil {
if existing.TaskID != taskID || existing.IdempotencyKey != idempotencyKey {
return nil, false, fmt.Errorf("idempotency key is already bound to another task")
}
return &existing, true, nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, false, err
}
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return nil, false, err
}
if task.ServiceType != "" && task.ServiceType != "mysql" {
return nil, false, fmt.Errorf("task service type %q is not handled by the MySQL executor", task.ServiceType)
}
if task.PayloadHash != payloadHash || task.IdempotencyKey != idempotencyKey {
return nil, false, fmt.Errorf("execution request does not match the immutable task payload")
}
if task.Status != model.TaskDispatching {
return nil, false, fmt.Errorf("task in state %q is not ready for execution", task.Status)
}
target, err := s.getTarget(ctx, task.TargetID)
if err != nil {
return nil, false, err
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return nil, false, err
}
meta := parseTargetMetadata(target.Metadata)
// Persist execution record BEFORE launching AWX to ensure crash recovery.
now := time.Now()
execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: "pending-" + task.ID, Status: "launching", StartedAt: &now}
if err := s.db.WithContext(ctx).Create(&execution).Error; err != nil {
return nil, false, err
}
extraVars := deliveryExtraVars(&task, payload, meta)
credentialVars, err := s.deploymentCredentialVars(ctx, task.ID)
if err != nil {
_ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()})
return nil, false, err
}
for key, value := range credentialVars {
extraVars[key] = value
}
extraVars["delivery_callback_url"] = s.deliveryCallbackURL(task.ID)
extraVars["delivery_callback_token"] = s.cfg.AWXWebhookToken
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
}
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Model(&execution).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
return err
}
return s.transitionTx(tx, &task, model.TaskRunning, "AWX job started", "")
}); err != nil {
return nil, false, err
}
return &execution, false, nil
}
func (s *DeliveryService) deliveryCallbackURL(taskID string) string {
base := strings.TrimRight(strings.TrimSpace(s.cfg.DeliveryCallbackBaseURL), "/")
if base == "" {
return ""
}
return base + "/auth/internal/delivery/tasks/" + taskID + "/events"
}
// deliveryExtraVars 组装传给 playbook 的变量:必传项固定注入,
// 选填项仅在用户显式设置时下发,未设置时由 playbook 默认基线兜底。
func deliveryExtraVars(task *model.DeliveryTask, payload deliveryPayload, meta targetMetadata) map[string]any {
topology := payload.Topology
if topology == "" {
// 存量任务的 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": 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
}
if payload.Timezone != "" {
vars["timezone"] = payload.Timezone
}
if payload.LowerCaseTableNames != nil {
vars["lower_case_table_names"] = *payload.LowerCaseTableNames
}
if payload.CharacterSet != "" {
vars["character_set"] = payload.CharacterSet
}
if payload.Collation != "" {
vars["collation"] = payload.Collation
}
if payload.MaxConnections != "" {
vars["max_connections"] = payload.MaxConnections
}
if payload.InnodbRedoLogCapacity != "" {
vars["innodb_redo_log_capacity"] = payload.InnodbRedoLogCapacity
}
if payload.InnodbFlushLogAtTrxCommit != nil {
vars["innodb_flush_log_at_trx_commit"] = *payload.InnodbFlushLogAtTrxCommit
}
if payload.SyncBinlog != nil {
vars["sync_binlog"] = *payload.SyncBinlog
}
if payload.InnodbIOCapacity != 0 {
vars["innodb_io_capacity"] = payload.InnodbIOCapacity
}
if payload.LongQueryTime != 0 {
vars["long_query_time"] = payload.LongQueryTime
}
if payload.BinlogExpireLogsSeconds != 0 {
vars["binlog_expire_logs_seconds"] = payload.BinlogExpireLogsSeconds
}
if payload.MaxBinlogSize != "" {
vars["max_binlog_size"] = payload.MaxBinlogSize
}
return vars
}
func (s *DeliveryService) HandleStageEvent(ctx context.Context, taskID string, input DeliveryStageEventInput) error {
stage := strings.ToLower(strings.TrimSpace(input.Stage))
status := strings.ToLower(strings.TrimSpace(input.Status))
if !validDeliveryStage(stage) {
return fmt.Errorf("invalid delivery stage %q", input.Stage)
}
if !validDeliveryStageStatus(status) {
return fmt.Errorf("invalid delivery stage status %q", input.Status)
}
message := strings.TrimSpace(input.Message)
if message == "" {
message = stage + " " + status
}
eventState := "stage_" + stage + "_" + status
var failedTaskID string
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var task model.DeliveryTask
if err := tx.First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q is not handled by MySQL stage callbacks", task.ServiceType)
}
if input.AWXJobID != "" {
var execution model.ExecutionJob
if err := tx.Where("task_id = ?", task.ID).First(&execution).Error; err != nil {
return err
}
if execution.ExecutorJobID != strings.TrimSpace(input.AWXJobID) {
return fmt.Errorf("AWX job %s does not match task %s", input.AWXJobID, task.ID)
}
}
if isTerminalTaskStatus(task.Status) {
return nil
}
if err := tx.Create(&model.TaskEvent{
TaskID: task.ID,
FromState: task.Status,
ToState: eventState,
Stage: stage,
EventStatus: status,
Message: message,
}).Error; err != nil {
return err
}
if status == "failed" {
now := time.Now()
if err := tx.Model(&model.ExecutionJob{}).Where("task_id = ?", task.ID).Updates(map[string]any{"status": "failed", "finished_at": now}).Error; err != nil {
return err
}
failedTaskID = task.ID
}
return nil
})
if err != nil {
return err
}
if failedTaskID != "" {
if rollbackErr := s.beginRollback(ctx, failedTaskID, message); rollbackErr != nil {
return rollbackErr
}
}
s.broadcastTask(ctx, taskID)
return nil
}
func (s *DeliveryService) HandleAWXJobNotification(ctx context.Context, input AWXJobNotificationInput) error {
status := strings.ToLower(strings.TrimSpace(input.Status))
if status == "" && input.ID == 0 && awxNotificationTaskID(input.ExtraVars) == "" {
return nil
}
if status == "" {
return fmt.Errorf("missing AWX job status")
}
execution, err := s.findExecutionForAWXNotification(ctx, input)
if err != nil {
return err
}
var task model.DeliveryTask
if err := s.db.WithContext(ctx).Select("id", "service_type").First(&task, "id = ?", execution.TaskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
// PostgreSQL jobs are finalized by PostgreSQLDeliveryService.PollOnce.
return nil
}
message := awxNotificationMessage(input)
defer s.broadcastTask(ctx, execution.TaskID)
switch status {
case "pending", "waiting", "running", "new":
return s.recordAWXEvent(ctx, execution.TaskID, status, message)
case "successful":
s.finishExecution(ctx, execution, "successful")
if err := s.completeTask(ctx, execution.TaskID); err != nil {
return s.beginRollback(ctx, execution.TaskID, err.Error())
}
return nil
case "canceled", "cancelled":
s.finishExecution(ctx, execution, "canceled")
return s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, message)
case "failed", "error":
s.finishExecution(ctx, execution, "failed")
return s.beginRollback(ctx, execution.TaskID, message)
default:
s.finishExecution(ctx, execution, "failed")
return s.beginRollback(ctx, execution.TaskID, "AWX job finished with status "+status)
}
}
func (s *DeliveryService) findExecutionForAWXNotification(ctx context.Context, input AWXJobNotificationInput) (*model.ExecutionJob, error) {
var execution model.ExecutionJob
if input.ID != 0 {
if err := s.db.WithContext(ctx).Where("executor_job_id = ?", fmt.Sprint(input.ID)).First(&execution).Error; err == nil {
return &execution, nil
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
}
taskID := awxNotificationTaskID(input.ExtraVars)
if taskID == "" {
return nil, fmt.Errorf("AWX notification did not include a known job id or task_id")
}
if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).First(&execution).Error; err != nil {
return nil, err
}
return &execution, nil
}
func (s *DeliveryService) recordAWXEvent(ctx context.Context, taskID, status, message string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if isTerminalTaskStatus(task.Status) {
return nil
}
return s.db.WithContext(ctx).Create(&model.TaskEvent{
TaskID: task.ID,
FromState: task.Status,
ToState: "awx_" + status,
Stage: "awx",
EventStatus: status,
Message: message,
}).Error
}
func validDeliveryStage(stage string) bool {
switch stage {
case "precheck", "install", "configure", "healthcheck", "register":
return true
default:
return false
}
}
func validDeliveryStageStatus(status string) bool {
switch status {
case "running", "success", "failed":
return true
default:
return false
}
}
func isTerminalTaskStatus(status string) bool {
for _, item := range terminalTaskStatuses() {
if status == item {
return true
}
}
return false
}
func terminalTaskStatuses() []string {
return []string{
model.TaskFinished,
model.TaskExecutionFailed,
model.TaskValidationFailed,
model.TaskRegisterFailed,
model.TaskCanceled,
model.TaskRolledBack,
model.TaskRollbackFailed,
model.TaskRollbackAck,
}
}
func allocationReleasedTaskStatuses() []string {
return []string{
model.TaskFinished,
model.TaskExecutionFailed,
model.TaskValidationFailed,
model.TaskRegisterFailed,
model.TaskCanceled,
model.TaskRolledBack,
model.TaskRollbackAck,
}
}
func awxNotificationMessage(input AWXJobNotificationInput) string {
status := strings.TrimSpace(input.Status)
name := strings.TrimSpace(input.Name)
if strings.TrimSpace(input.Traceback) != "" {
return strings.TrimSpace(input.Traceback)
}
if name == "" {
return "AWX job " + status
}
return "AWX job " + name + " " + status
}
func awxNotificationTaskID(extraVars any) string {
var values map[string]any
switch v := extraVars.(type) {
case map[string]any:
values = v
case string:
if strings.TrimSpace(v) == "" {
return ""
}
_ = json.Unmarshal([]byte(v), &values)
default:
raw, err := json.Marshal(v)
if err != nil {
return ""
}
_ = json.Unmarshal(raw, &values)
}
if values == nil {
return ""
}
if taskID, ok := values["task_id"].(string); ok {
return strings.TrimSpace(taskID)
}
return ""
}
func (s *DeliveryService) finishExecution(ctx context.Context, execution *model.ExecutionJob, status string) {
now := time.Now()
_ = s.db.WithContext(ctx).Model(execution).Updates(map[string]any{"status": status, "finished_at": now})
}
func (s *DeliveryService) PollOnce(ctx context.Context) error {
var jobs []model.ExecutionJob
if err := s.db.WithContext(ctx).
Joins("JOIN delivery_tasks ON delivery_tasks.id = execution_jobs.task_id").
Where("execution_jobs.status = ? AND (delivery_tasks.service_type = '' OR delivery_tasks.service_type = ?)", "running", "mysql").
Find(&jobs).Error; err != nil {
return err
}
for _, execution := range jobs {
job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
if err != nil {
// A transient AWX/API failure is not evidence that deployment failed.
// Keep the job running and retry on the next scheduler tick.
continue
}
switch strings.ToLower(job.Status) {
case "pending", "waiting", "running", "new":
continue
case "successful":
s.finishExecution(ctx, &execution, "successful")
if err := s.completeTask(ctx, execution.TaskID); err != nil {
_ = s.beginRollback(ctx, execution.TaskID, err.Error())
}
case "canceled":
s.finishExecution(ctx, &execution, "canceled")
_ = s.beginRollback(ctx, execution.TaskID, "AWX deployment job was canceled")
default:
s.finishExecution(ctx, &execution, "failed")
_ = s.beginRollback(ctx, execution.TaskID, "AWX job finished with status "+job.Status)
}
}
return nil
}
func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot be completed by the MySQL delivery service", task.ServiceType)
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return 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 {
return err
}
metadata := mysqlDeploymentMetadata(task, payload)
result := model.DeploymentResult{
TaskID: task.ID,
BusinessLineID: task.BusinessLineID,
Component: task.Component,
ServiceType: "database",
InstanceName: payload.InstanceName,
Namespace: payload.Namespace,
TargetID: task.TargetID,
NodeName: task.TargetHost,
Host: task.TargetHostIP,
TargetHosts: task.TargetHosts,
TargetHostIPs: task.TargetHostIPs,
Port: task.MySQLPort,
Version: payload.MySQLVersion,
Status: "active",
Metadata: string(mustJSON(metadata)),
}
if result.Component == "" {
result.Component = "mysql"
}
if err := tx.Create(&result).Error; err != nil {
return err
}
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 * 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 {
return err
}
if err := tx.Model(&model.ExecutionJob{}).Where("task_id = ?", task.ID).Updates(map[string]any{"status": "successful", "finished_at": now}).Error; err != nil {
return err
}
return nil
}); err != nil {
return err
}
if err := s.RegisterCloudDM(ctx, task.ID); err != nil {
message := fmt.Sprintf("CloudDM registration failed: %v", err)
// The database is already healthy and accounted for at this point. A
// registration outage must not trigger destructive instance rollback.
if transitionErr := s.transition(ctx, &task, model.TaskRegisterFailed,
"MySQL delivered; CloudDM registration failed and can be retried", message); transitionErr != nil {
return fmt.Errorf("%s; cannot record register_failed: %w", message, transitionErr)
}
return nil
}
return s.transition(ctx, &task, model.TaskFinished, "MySQL delivery completed", "")
}
// RetryCloudDMRegistration retries only the external registration step. The
// instance and its resource usage remain active throughout the retry.
func (s *DeliveryService) RetryCloudDMRegistration(ctx context.Context, taskID string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot use the MySQL CloudDM registration flow", task.ServiceType)
}
if task.Status != model.TaskRegisterFailed {
return fmt.Errorf("task %s is in state %q and cannot retry CloudDM registration", taskID, task.Status)
}
if err := s.transition(ctx, &task, model.TaskRegistering, "CloudDM registration retry started", ""); err != nil {
return err
}
if err := s.RegisterCloudDM(ctx, taskID); err != nil {
message := fmt.Sprintf("CloudDM registration failed: %v", err)
if transitionErr := s.transition(ctx, &task, model.TaskRegisterFailed, "CloudDM registration retry failed; instance remains active", message); transitionErr != nil {
return fmt.Errorf("%s; cannot restore register_failed: %w", message, transitionErr)
}
return err
}
return s.transition(ctx, &task, model.TaskFinished, "CloudDM registration completed", "")
}
type cloudDMRegisterRequest struct {
SourceSystem string `json:"sourceSystem"`
ResourceType string `json:"resourceType"`
ExternalResourceID string `json:"externalResourceId"`
DataSource cloudDMDataSource `json:"dataSource"`
}
type cloudDMDeleteRequest struct {
SourceSystem string `json:"sourceSystem"`
ResourceType string `json:"resourceType"`
ExternalResourceID string `json:"externalResourceId"`
}
type cloudDMDataSource struct {
InstanceName string `json:"instanceName"`
InstanceDesc string `json:"instanceDesc"`
DSType string `json:"dsType"`
Host string `json:"host"`
ClusterID *uint64 `json:"cluster_id"`
MySQLVersion string `json:"mysql_version"`
SecurityType string `json:"securityType"`
UserName string `json:"userName"`
Password string `json:"password"`
DefaultSchema any `json:"defaultSchema"`
ClientTimeZone string `json:"clientTimeZone"`
ConnectTimeoutMs int `json:"connectTimeoutMs"`
SocketTimeoutSecs int `json:"socketTimeoutSeconds"`
ConnectionCharset string `json:"connectionCharset"`
}
func cloudDMExternalResourceID(instanceID uint64) string {
return "mysql-instance:" + strconv.FormatUint(instanceID, 10)
}
func cloudDMClientTimeZone(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "Asia/Shanghai"
}
return value
}
func cloudDMDataSourceIDFromResponse(raw []byte) (uint64, bool) {
var value any
if len(bytes.TrimSpace(raw)) == 0 {
return 0, false
}
if err := json.Unmarshal(raw, &value); err != nil {
return 0, false
}
return cloudDMDataSourceIDFromValue(value)
}
func cloudDMDataSourceIDFromValue(value any) (uint64, bool) {
switch typed := value.(type) {
case float64:
if typed <= 0 || typed != float64(uint64(typed)) {
return 0, false
}
return uint64(typed), true
case string:
id, err := strconv.ParseUint(strings.TrimSpace(typed), 10, 64)
if err != nil || id == 0 {
return 0, false
}
return id, true
case map[string]any:
for _, key := range []string{"dataSourceId", "datasourceId", "dsId", "id", "data"} {
if id, ok := cloudDMDataSourceIDFromValue(typed[key]); ok {
return id, true
}
}
}
return 0, false
}
func cloudDMMetadata(existing map[string]any, instance model.DeploymentResult, dataSourceID uint64) map[string]any {
meta := map[string]any{}
for key, value := range existing {
meta[key] = value
}
meta["clouddm"] = map[string]any{
"data_source_id": dataSourceID,
"external_resource_id": cloudDMExternalResourceID(instance.ID),
"registered_at": time.Now().Format(time.RFC3339),
"delete_status": "",
"delete_error": "",
"deleted_at": "",
}
return meta
}
func buildCloudDMDeleteRequest(instance model.DeploymentResult) cloudDMDeleteRequest {
return cloudDMDeleteRequest{
SourceSystem: "xinfra",
ResourceType: "MYSQL_INSTANCE",
ExternalResourceID: cloudDMExternalResourceID(instance.ID),
}
}
func cloudDMDeleteInfo(meta map[string]any) (uint64, string) {
clouddm, ok := meta["clouddm"].(map[string]any)
if !ok {
return 0, ""
}
dataSourceID, _ := cloudDMDataSourceIDFromValue(clouddm["data_source_id"])
deleteStatus := strings.ToLower(strings.TrimSpace(stringValue(clouddm["delete_status"])))
return dataSourceID, deleteStatus
}
func updateCloudDMDeleteMetadata(meta map[string]any, deleteStatus, deleteError, deletedAt string) map[string]any {
clouddm, ok := meta["clouddm"].(map[string]any)
if !ok {
clouddm = map[string]any{}
}
clouddm["delete_status"] = deleteStatus
clouddm["delete_error"] = deleteError
clouddm["deleted_at"] = deletedAt
meta["clouddm"] = clouddm
return meta
}
func buildCloudDMRegisterRequest(instance model.DeploymentResult, payload deliveryPayload, password string) cloudDMRegisterRequest {
description := strings.TrimSpace(payload.InstanceDesc)
if description == "" {
description = instance.InstanceName
}
mysqlVersion := strings.TrimSpace(instance.Version)
if mysqlVersion == "" {
mysqlVersion = strings.TrimSpace(payload.MySQLVersion)
}
if mysqlVersion == "" {
mysqlVersion = "8.0"
}
return cloudDMRegisterRequest{
SourceSystem: "xinfra",
ResourceType: "MYSQL_INSTANCE",
ExternalResourceID: cloudDMExternalResourceID(instance.ID),
DataSource: cloudDMDataSource{
InstanceName: instance.InstanceName,
InstanceDesc: description,
DSType: "MySQL",
Host: net.JoinHostPort(instance.Host, strconv.Itoa(instance.Port)),
ClusterID: nil,
MySQLVersion: mysqlVersion,
SecurityType: "USER_PASSWD",
UserName: "root",
Password: password,
DefaultSchema: nil,
ClientTimeZone: cloudDMClientTimeZone(payload.Timezone),
ConnectTimeoutMs: 5000,
SocketTimeoutSecs: 10,
ConnectionCharset: "utf8",
},
}
}
func (s *DeliveryService) deleteCloudDMDataSource(ctx context.Context, instance model.DeploymentResult) error {
if strings.TrimSpace(s.cfg.CloudDMDeleteURL) == "" {
return fmt.Errorf("CloudDM deletion requires CLOUDDM_DELETE_URL to be configured")
}
if strings.TrimSpace(s.cfg.CloudDMAPIToken) == "" {
return fmt.Errorf("CloudDM deletion requires CLOUDDM_API_TOKEN to be configured")
}
body := buildCloudDMDeleteRequest(instance)
raw, err := json.Marshal(body)
if err != nil {
return fmt.Errorf("encode CloudDM deletion request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMDeleteURL, bytes.NewReader(raw))
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(s.cfg.CloudDMAPIToken))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Request-ID", randomUUID())
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("CloudDM deletion returned %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
}
return nil
}
func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error {
if strings.TrimSpace(s.cfg.CloudDMRegisterURL) == "" {
return nil
}
if strings.TrimSpace(s.cfg.CloudDMAPIToken) == "" {
return fmt.Errorf("CloudDM registration is enabled but CLOUDDM_API_TOKEN is not configured")
}
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return fmt.Errorf("decode delivery payload: %w", err)
}
var instance model.DeploymentResult
if err := s.db.WithContext(ctx).Where("task_id = ? AND component = ? AND service_type = ?", taskID, "mysql", "database").First(&instance).Error; err != nil {
return err
}
var credential model.DeploymentCredential
if err := s.db.WithContext(ctx).
Where("task_id = ? AND username = ? AND account_host IN ? AND status IN ?", taskID, "root", []string{"%", "localhost"}, []string{"pending", "available"}).
First(&credential).Error; err != nil {
return fmt.Errorf("CloudDM registration requires the MySQL root credential: %w", err)
}
password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, credential.Ciphertext, credential.Nonce)
if err != nil {
return fmt.Errorf("decrypt MySQL root credential for CloudDM registration: %w", err)
}
body := buildCloudDMRegisterRequest(instance, payload, password)
raw, err := json.Marshal(body)
if err != nil {
return fmt.Errorf("encode CloudDM registration request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMRegisterURL, bytes.NewReader(raw))
if err != nil {
return err
}
// X-Request-ID is a trace identifier, so generate a fresh UUID for every
// HTTP call. externalResourceId above is the stable business idempotency key.
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(s.cfg.CloudDMAPIToken))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Request-ID", randomUUID())
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("CloudDM returned %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
}
dataSourceID, ok := cloudDMDataSourceIDFromResponse(respBody)
if !ok {
return fmt.Errorf("CloudDM registration response did not include dataSourceId: %s", strings.TrimSpace(string(respBody)))
}
meta := metadataMap(instance.Metadata)
meta = cloudDMMetadata(meta, instance, dataSourceID)
if err := s.db.WithContext(ctx).Model(&instance).Update("metadata", string(mustJSON(meta))).Error; err != nil {
return fmt.Errorf("save CloudDM dataSourceId: %w", err)
}
return nil
}
// beginRollback changes the task into the compensating workflow and launches
// the dedicated AWX rollback template. The deploy reservation is deliberately
// kept until rollback succeeds so a failed cleanup cannot be silently
// overwritten by a later delivery.
func (s *DeliveryService) beginRollback(ctx context.Context, taskID, reason string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot use the MySQL rollback flow", task.ServiceType)
}
if rollbackProtectedStatus(task.Status) {
return nil
}
if s.cfg.RollbackTemplateID == 0 {
if err := s.transition(ctx, &task, model.TaskRollbackFailed,
"automatic rollback is not configured; manual cleanup is required", reason); err != nil {
return err
}
return s.holdReservationForRollback(ctx, task.ID)
}
if err := s.transition(ctx, &task, model.TaskRollbackPending, "automatic rollback queued", reason); err != nil {
return err
}
if err := s.holdReservationForRollback(ctx, task.ID); err != nil {
return s.markRollbackFailed(ctx, task.ID, "cannot hold resources during rollback: "+err.Error())
}
return s.launchRollbackJob(ctx, &task, reason)
}
func rollbackProtectedStatus(status string) bool {
switch status {
case model.TaskRollbackPending, model.TaskRollingBack, model.TaskRolledBack, model.TaskRollbackFailed,
model.TaskRollbackAck, model.TaskRegisterFailed, model.TaskFinished, model.TaskCanceled:
return true
default:
return false
}
}
func (s *DeliveryService) launchRollbackJob(ctx context.Context, task *model.DeliveryTask, reason string) error {
target, err := s.getTarget(ctx, task.TargetID)
if err != nil {
return s.markRollbackFailed(ctx, task.ID, "rollback target is unavailable: "+err.Error())
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return s.markRollbackFailed(ctx, task.ID, "rollback payload is invalid: "+err.Error())
}
now := time.Now()
var rollback model.RollbackJob
if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&rollback).Error; errors.Is(err, gorm.ErrRecordNotFound) {
rollback = model.RollbackJob{TaskID: task.ID}
} else if err != nil {
return s.markRollbackFailed(ctx, task.ID, "cannot load rollback job: "+err.Error())
}
rollback.ExecutorJobID = "pending-rollback-" + task.ID
rollback.Status = "launching"
rollback.Reason = reason
rollback.StartedAt = &now
rollback.FinishedAt = nil
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: rollbackLimit,
ExtraVars: rollbackExtraVars(task, payload),
})
if err != nil {
_ = s.db.WithContext(ctx).Model(&rollback).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}).Error
return s.markRollbackFailed(ctx, task.ID, "cannot launch rollback AWX job: "+err.Error())
}
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current model.DeliveryTask
if err := tx.First(&current, "id = ?", task.ID).Error; err != nil {
return err
}
if current.Status != model.TaskRollbackPending {
return fmt.Errorf("task %s is no longer pending rollback (state %q)", task.ID, current.Status)
}
if err := tx.Model(&rollback).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
return err
}
return s.transitionTx(tx, &current, model.TaskRollingBack, "rollback AWX job started", reason)
}); err != nil {
return s.markRollbackFailed(ctx, task.ID, "cannot record rollback AWX job: "+err.Error())
}
return nil
}
func (s *DeliveryService) RetryRollback(ctx context.Context, taskID string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot use the MySQL rollback flow", task.ServiceType)
}
if task.Status != model.TaskRollbackFailed {
return fmt.Errorf("task %s is in state %q and cannot retry rollback", taskID, task.Status)
}
if s.cfg.RollbackTemplateID == 0 {
return fmt.Errorf("automatic rollback is not configured")
}
reason := "manual rollback retry requested"
if err := s.transition(ctx, &task, model.TaskRollbackPending, "manual rollback retry queued", reason); err != nil {
return err
}
if err := s.holdReservationForRollback(ctx, task.ID); err != nil {
_ = s.markRollbackFailed(ctx, task.ID, "cannot hold resources during rollback retry: "+err.Error())
return err
}
return s.launchRollbackJob(ctx, &task, reason)
}
// AcknowledgeRollbackRelease is an explicit platform-admin escape hatch after
// the operator has verified the target host. It releases bookkeeping only; it
// does not claim that the automatic rollback succeeded.
func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID string) error {
now := time.Now()
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var task model.DeliveryTask
if err := tx.First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot use the MySQL rollback flow", task.ServiceType)
}
if task.Status != model.TaskRollbackFailed {
return fmt.Errorf("task %s is in state %q and cannot acknowledge rollback release", taskID, task.Status)
}
if err := tx.Model(&model.DeploymentResult{}).Where("task_id = ? AND component = ? AND service_type = ? AND status IN ?", taskID, "mysql", "database", occupiedDeploymentStatuses()).Updates(map[string]any{"status": "rollback_acknowledged", "updated_at": now}).Error; err != nil {
return err
}
if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
return err
}
if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "rollback").Update("status", "released").Error; err != nil {
return err
}
return s.transitionTx(tx, &task, model.TaskRollbackAck, "rollback release acknowledged by platform admin", "")
})
}
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": hosts,
"instance_name": payload.InstanceName,
"data_disk": payload.DataDisk,
"task_id": task.ID,
"rollback": true,
}
}
func (s *DeliveryService) holdReservationForRollback(ctx context.Context, taskID string) error {
return s.db.WithContext(ctx).Model(&model.ResourceReservation{}).
Where("task_id = ? AND status IN ?", taskID, []string{"reserved", "consumed"}).
Update("status", "rollback").Error
}
func (s *DeliveryService) markRollbackFailed(ctx context.Context, taskID, message string) error {
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if task.Status == model.TaskRolledBack || task.Status == model.TaskFinished || task.Status == model.TaskCanceled ||
task.Status == model.TaskRollbackFailed || task.Status == model.TaskRollbackAck || task.Status == model.TaskRegisterFailed {
return nil
}
return s.transition(ctx, &task, model.TaskRollbackFailed, "automatic rollback failed; manual cleanup is required", message)
}
func (s *DeliveryService) PollRollbackOnce(ctx context.Context) error {
var jobs []model.RollbackJob
if err := s.db.WithContext(ctx).
Joins("JOIN delivery_tasks ON delivery_tasks.id = rollback_jobs.task_id").
Where("rollback_jobs.status IN ? AND (delivery_tasks.service_type = '' OR delivery_tasks.service_type = ?)", []string{"launching", "running"}, "mysql").
Find(&jobs).Error; err != nil {
return err
}
for _, rollback := range jobs {
if rollback.Status == "launching" {
if !rollbackLaunchExpired(rollback, time.Now()) {
continue
}
// The AWX launch result is unknown. Do not blindly launch a second
// job; surface a recoverable failure for an explicit admin retry.
expired, err := s.expireRollbackLaunch(ctx, rollback.ID)
if err != nil {
return err
}
if !expired {
continue
}
_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback launch timed out before its AWX job ID was recorded; manual retry is required")
continue
}
job, err := s.awx.GetJob(ctx, rollback.ExecutorJobID)
if err != nil {
// Do not declare cleanup failed because AWX is temporarily unreachable.
continue
}
switch strings.ToLower(job.Status) {
case "pending", "waiting", "running", "new":
continue
case "successful":
_ = s.finishRollbackJob(ctx, &rollback, "successful")
if err := s.completeRollback(ctx, rollback.TaskID); err != nil {
_ = s.markRollbackFailed(ctx, rollback.TaskID, err.Error())
}
case "canceled":
_ = s.finishRollbackJob(ctx, &rollback, "canceled")
_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback AWX job was canceled")
default:
_ = s.finishRollbackJob(ctx, &rollback, "failed")
_ = s.markRollbackFailed(ctx, rollback.TaskID, "rollback AWX job finished with status "+job.Status)
}
}
return nil
}
func rollbackLaunchExpired(rollback model.RollbackJob, now time.Time) bool {
if rollback.Status != "launching" {
return false
}
if rollback.StartedAt == nil {
return true
}
return !now.Before(rollback.StartedAt.Add(rollbackLaunchTimeout))
}
func (s *DeliveryService) expireRollbackLaunch(ctx context.Context, rollbackID uint64) (bool, error) {
result := s.db.WithContext(ctx).Model(&model.RollbackJob{}).
Where("id = ? AND status = ?", rollbackID, "launching").
Updates(map[string]any{"status": "launch_timeout", "finished_at": time.Now()})
return result.RowsAffected == 1, result.Error
}
func (s *DeliveryService) finishRollbackJob(ctx context.Context, rollback *model.RollbackJob, status string) error {
return s.db.WithContext(ctx).Model(rollback).Updates(map[string]any{"status": status, "finished_at": time.Now()}).Error
}
func (s *DeliveryService) completeRollback(ctx context.Context, taskID string) error {
now := time.Now()
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var task model.DeliveryTask
if err := tx.First(&task, "id = ?", taskID).Error; err != nil {
return err
}
if !isMySQLServiceType(task.ServiceType) {
return fmt.Errorf("task service type %q cannot use the MySQL rollback flow", task.ServiceType)
}
if task.Status != model.TaskRollingBack {
return fmt.Errorf("task %s is in state %q, cannot complete rollback", taskID, task.Status)
}
if err := tx.Model(&model.DeploymentResult{}).Where("task_id = ?", taskID).Updates(map[string]any{"status": "rolled_back", "updated_at": now}).Error; err != nil {
return err
}
if err := tx.Unscoped().Where("task_id = ?", taskID).Delete(&model.DeploymentCredential{}).Error; err != nil {
return err
}
if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil {
return err
}
if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status IN ?", taskID, []string{"reserved", "consumed", "rollback"}).Update("status", "released").Error; err != nil {
return err
}
return s.transitionTx(tx, &task, model.TaskRolledBack, "MySQL instance rollback completed", "")
})
}
func isMySQLServiceType(serviceType string) bool {
return serviceType == "" || serviceType == "mysql"
}
func (s *DeliveryService) failTask(ctx context.Context, task *model.DeliveryTask, status, message string) error {
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var current model.DeliveryTask
if err := tx.First(&current, "id = ?", task.ID).Error; err != nil {
return err
}
if current.Status != model.TaskPending && current.Status != model.TaskDispatching && current.Status != model.TaskRunning && current.Status != model.TaskRegistering && current.Status != model.TaskCanceling {
return nil
}
if err := s.transitionTx(tx, &current, status, message, message); err != nil {
return err
}
if status == model.TaskExecutionFailed || status == model.TaskValidationFailed || status == model.TaskCanceled {
return s.releaseReservation(tx, current.ID)
}
return nil
})
}
func (s *DeliveryService) Run(ctx context.Context) {
interval := time.Duration(s.cfg.DeliveryDispatchSeconds) * time.Second
if interval < time.Second {
interval = time.Second
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
_ = s.DispatchOnce(ctx)
_ = s.PollOnce(ctx)
_ = s.PollRollbackOnce(ctx)
}
}
}