wonder/xinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: wire mysql delivery to awx callbacks

Browse Source
This commit is contained in:
mac
2026-07-28 14:45:17 +08:00
parent 5a260c443f
commit 8e09c30d21
13 changed files with 676 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/.env.example
+3 -1
View File
@@ -25,12 +25,14 @@ DELIVERY_TARGET_LIMIT=2
# 单机 MySQL 实例数上限（同机多实例，容量由配额 + playbook 实机守卫兜底）
DELIVERY_HOST_INSTANCE_LIMIT=4
# 数据盘挂载点白名单（逗号分隔，第一项为默认值）
DELIVERY_DATA_DISKS=/data
DELIVERY_DATA_DISKS=/data,/disk1,/mnt,/opt/mysql-delivery
AWX_BASE_URL=
AWX_TOKEN=
AWX_USERNAME=
AWX_PASSWORD=
AWX_WEBHOOK_TOKEN=
AWX_FACTS_TEMPLATE_ID=
AWX_FACTS_TIMEOUT_SECONDS=45
CLOUDDM_REGISTER_URL=
CLOUDDM_API_TOKEN=
server/internal/config/config.go
+17 -1
View File
@@ -64,6 +64,8 @@ type Config struct {
AWXUsername string
AWXPassword string
AWXWebhookToken string
AWXFactsTemplateID uint64
AWXFactsTimeoutSeconds int
DeliverySchedulerEnabled bool
DeliveryDispatchSeconds int
DeliveryCallbackBaseURL string
@@ -134,6 +136,8 @@ func Load() Config {
AWXUsername: env("AWX_USERNAME", ""),
AWXPassword: env("AWX_PASSWORD", ""),
AWXWebhookToken: env("AWX_WEBHOOK_TOKEN", ""),
AWXFactsTemplateID: envUint64("AWX_FACTS_TEMPLATE_ID", 0),
AWXFactsTimeoutSeconds: envInt("AWX_FACTS_TIMEOUT_SECONDS", 45),
DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false),
DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5),
DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)),
@@ -141,7 +145,7 @@ func Load() Config {
DeliveryGlobalLimit: envInt("DELIVERY_GLOBAL_LIMIT", 2),
DeliveryTargetLimit: envInt("DELIVERY_TARGET_LIMIT", 2),
DeliveryHostInstanceLimit: envInt("DELIVERY_HOST_INSTANCE_LIMIT", 4),
DeliveryDataDisks: splitCSV(env("DELIVERY_DATA_DISKS", "/data")),
DeliveryDataDisks: splitCSV(env("DELIVERY_DATA_DISKS", "/data,/disk1,/mnt,/opt/mysql-delivery")),
}
}
@@ -245,6 +249,18 @@ func envInt(key string, fallback int) int {
return parsed
}
func envUint64(key string, fallback uint64) uint64 {
value := os.Getenv(key)
if value == "" {
return fallback
}
parsed, err := strconv.ParseUint(value, 10, 64)
if err != nil {
return fallback
}
return parsed
}
func defaultPublicBaseURL(httpAddr string) string {
addr := strings.TrimSpace(httpAddr)
if addr == "" {
server/internal/handler/delivery.go
+14
View File
@@ -261,6 +261,20 @@ func (h *DeliveryHandler) Targets(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"items": items})
}
func (h *DeliveryHandler) TargetHostMountPaths(c *gin.Context) {
targetID, err := strconv.ParseUint(c.Param("target_id"), 10, 64)
if err != nil || targetID == 0 {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid target_id"})
return
}
items, err := h.service.ListHostMountPaths(c.Request.Context(), targetID, c.Param("host"))
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"items": items})
}
type quotaPayload struct {
BusinessLineID uint64 `json:"business_line_id" binding:"required"`
TargetID uint64 `json:"target_id" binding:"required"`
server/internal/router/router.go
+1
View File
@@ -145,6 +145,7 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) {
protected.GET("/container-services/business-lines/:id/workloads", containerServiceHandler.Workloads)
protected.GET("/clouddm/login", clouddmHandler.Login)
protected.GET("/delivery/targets", deliveryHandler.Targets)
protected.GET("/delivery/targets/:target_id/hosts/:host/mount-paths", deliveryHandler.TargetHostMountPaths)
protected.PUT("/delivery/quotas", deliveryHandler.UpsertQuota)
protected.POST("/delivery/mysql", deliveryHandler.CreateMySQL)
protected.GET("/delivery/tasks", deliveryHandler.List)
server/internal/service/awx.go
+46 -1
View File
@@ -87,7 +87,13 @@ func (c *AWXClient) Configured() bool {
}
func (c *AWXClient) Launch(ctx context.Context, templateID uint64, input AWXLaunchRequest) (*AWXJob, error) {
body := map[string]any{"inventory": input.InventoryID, "extra_vars": input.ExtraVars}
body := map[string]any{}
if input.InventoryID != 0 {
body["inventory"] = input.InventoryID
}
if input.ExtraVars != nil {
body["extra_vars"] = input.ExtraVars
}
if input.Limit != "" {
body["limit"] = input.Limit
}
@@ -141,6 +147,31 @@ func (c *AWXClient) Cancel(ctx context.Context, jobID string) error {
return c.request(ctx, http.MethodPost, "/api/v2/jobs/"+jobID+"/cancel/", map[string]any{}, nil)
}
func (c *AWXClient) WaitJob(ctx context.Context, jobID string, timeout time.Duration) (*AWXJob, error) {
if timeout <= 0 {
timeout = 45 * time.Second
}
deadlineCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
job, err := c.GetJob(deadlineCtx, jobID)
if err != nil {
return nil, err
}
switch job.Status {
case "successful", "failed", "error", "canceled":
return job, nil
}
select {
case <-deadlineCtx.Done():
return nil, fmt.Errorf("AWX job %s did not finish within %s", jobID, timeout)
case <-ticker.C:
}
}
}
func (c *AWXClient) ListJobTemplates(ctx context.Context) ([]AWXJobTemplate, error) {
var out []AWXJobTemplate
path := "/api/v2/job_templates/?page_size=200"
@@ -186,6 +217,20 @@ func (c *AWXClient) ListInventoryHosts(ctx context.Context, inventoryID uint64)
return out, nil
}
func (c *AWXClient) GetHostFacts(ctx context.Context, hostID uint64) (map[string]any, error) {
if hostID == 0 {
return nil, fmt.Errorf("invalid AWX host id")
}
var facts map[string]any
if err := c.request(ctx, http.MethodGet, fmt.Sprintf("/api/v2/hosts/%d/ansible_facts/", hostID), nil, &facts); err != nil {
return nil, err
}
if nested, ok := facts["ansible_facts"].(map[string]any); ok {
return nested, nil
}
return facts, nil
}
func (c *AWXClient) request(ctx context.Context, method, path string, payload any, output any) error {
raw, err := c.requestRaw(ctx, method, path, payload, "application/json")
if err != nil {
server/internal/service/delivery.go
+142 -17
View File
@@ -13,6 +13,8 @@ import (
"net"
"net/http"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
@@ -72,6 +74,12 @@ type DeliveryTarget struct {
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"`
@@ -125,6 +133,81 @@ func parseTargetMetadata(raw string) targetMetadata {
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 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
}
// firstFreeHost 返回候选池中非失败任务数未达单机实例上限的第一个节点。
func firstFreeHost(hosts []targetHost, occupied []string, limit int) *targetHost {
if limit < 1 {
@@ -222,6 +305,64 @@ func (s *DeliveryService) getTarget(ctx context.Context, templateID uint64) (Del
return s.awxDeliveryTarget(ctx, *template)
}
func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint64, hostName 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")
}
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 {
if err := s.refreshHostFacts(ctx, hostName); err != nil {
return nil, err
}
}
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 (s *DeliveryService) refreshHostFacts(ctx context.Context, hostName string) error {
job, err := s.awx.Launch(ctx, s.cfg.AWXFactsTemplateID, AWXLaunchRequest{
Limit: hostName,
})
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)
}
return nil
}
func (s *DeliveryService) awxDeliveryTarget(ctx context.Context, template AWXJobTemplate) (DeliveryTarget, error) {
if !template.AskVariablesOnLaunch || !template.AskLimitOnLaunch {
return DeliveryTarget{}, fmt.Errorf("AWX job template %d must enable Prompt on launch for Variables and Limit", template.ID)
@@ -424,22 +565,6 @@ func validateDeliveryInput(input MySQLDeliveryInput, dataDisks []string) error {
if input.MySQLPort != 0 && (input.MySQLPort < mysqlPortPoolStart || input.MySQLPort > mysqlPortPoolEnd) {
return fmt.Errorf("mysql_port must be empty for auto assignment or between %d and %d", mysqlPortPoolStart, mysqlPortPoolEnd)
}
if input.DataDisk != "" {
allowed := dataDisks
if len(allowed) == 0 {
allowed = []string{"/data"}
}
found := false
for _, disk := range allowed {
if input.DataDisk == disk {
found = true
break
}
}
if !found {
return fmt.Errorf("data_disk %q is not in the allowed mount point list %v", input.DataDisk, allowed)
}
}
if input.TargetHost != "" && (len(input.TargetHost) > 253 || !hostNamePattern.MatchString(input.TargetHost)) {
return fmt.Errorf("target_host must be a valid inventory host name")
}
@@ -889,7 +1014,7 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa
"target_hosts": task.TargetHost, "topology": topology,
"instance_name": payload.InstanceName, "mysql_port": task.MySQLPort,
"data_disk": payload.DataDisk, "cpu_cores": payload.CPUCores,
"memory_gb": payload.MemoryGB, "storage_gb": payload.StorageGB,
"memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi,
"cpu_milli": payload.CPUMilli, "memory_mi": payload.MemoryMi, "storage_gi": payload.StorageGi,
"mysql_version": payload.MySQLVersion, "param_template": payload.ParamTemplate,
"target_host": payload.TargetHost,
server/internal/service/delivery_test.go
+27 -28
View File
@@ -13,7 +13,7 @@ func TestValidateDeliveryInput(t *testing.T) {
full.MySQLVersion = "8.0"
full.Topology = "standalone"
full.MySQLPort = 13306
full.DataDisk = "/disk1"
full.DataDisk = "/var/lib/mysql01"
full.TargetHost = "k8s-server-03"
full.CPUMilli = 2000
full.MemoryMi = 8192
@@ -44,33 +44,32 @@ func TestValidateDeliveryInput(t *testing.T) {
t.Fatalf("8.4 LTS rejected: %v", err)
}
for name, mutate := range map[string]func(*MySQLDeliveryInput){
"uppercase namespace": func(in *MySQLDeliveryInput) { in.Namespace = "Team-A" },
"bad instance": func(in *MySQLDeliveryInput) { in.InstanceName = "mysql_01" },
"too little cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 50 },
"too much cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 65000 },
"too little memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 1024 },
"too much memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 131072 },
"too little storage": func(in *MySQLDeliveryInput) { in.StorageGi = 10 },
"too much storage": func(in *MySQLDeliveryInput) { in.StorageGi = 4000 },
"unsupported version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.7" },
"eol version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.6" },
"unsupported topology": func(in *MySQLDeliveryInput) { in.Topology = "mgr_3" },
"port below pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 3307 },
"port above pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 14000 },
"data disk not in list": func(in *MySQLDeliveryInput) { in.DataDisk = "/mnt/other" },
"bad target host": func(in *MySQLDeliveryInput) { in.TargetHost = "-bad-host" },
"bad timezone": func(in *MySQLDeliveryInput) { in.TimeZone = "UTC+8" },
"bad lower case": func(in *MySQLDeliveryInput) { in.LowerCaseTableNames = 2 },
"bad charset": func(in *MySQLDeliveryInput) { in.CharacterSet = "big5" },
"collation mismatch": func(in *MySQLDeliveryInput) { in.CharacterSet = "gbk"; in.Collation = "utf8mb4_general_ci" },
"bad max connections": func(in *MySQLDeliveryInput) { in.MaxConnections = "300" },
"bad redo capacity": func(in *MySQLDeliveryInput) { in.InnoDBRedoLogCapacity = "2G" },
"bad flush log": func(in *MySQLDeliveryInput) { in.InnoDBFlushLogAtTrxCommit = 3 },
"bad sync binlog": func(in *MySQLDeliveryInput) { in.SyncBinlog = 2 },
"bad io capacity": func(in *MySQLDeliveryInput) { in.InnoDBIOCapacity = 500 },
"bad long query time": func(in *MySQLDeliveryInput) { in.LongQueryTime = 3 },
"bad binlog expire": func(in *MySQLDeliveryInput) { in.BinlogExpireLogsSeconds = 3600 },
"bad max binlog size": func(in *MySQLDeliveryInput) { in.MaxBinlogSize = "64M" },
"uppercase namespace": func(in *MySQLDeliveryInput) { in.Namespace = "Team-A" },
"bad instance": func(in *MySQLDeliveryInput) { in.InstanceName = "mysql_01" },
"too little cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 50 },
"too much cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 65000 },
"too little memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 1024 },
"too much memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 131072 },
"too little storage": func(in *MySQLDeliveryInput) { in.StorageGi = 10 },
"too much storage": func(in *MySQLDeliveryInput) { in.StorageGi = 4000 },
"unsupported version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.7" },
"eol version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.6" },
"unsupported topology": func(in *MySQLDeliveryInput) { in.Topology = "mgr_3" },
"port below pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 3307 },
"port above pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 14000 },
"bad target host": func(in *MySQLDeliveryInput) { in.TargetHost = "-bad-host" },
"bad timezone": func(in *MySQLDeliveryInput) { in.TimeZone = "UTC+8" },
"bad lower case": func(in *MySQLDeliveryInput) { in.LowerCaseTableNames = 2 },
"bad charset": func(in *MySQLDeliveryInput) { in.CharacterSet = "big5" },
"collation mismatch": func(in *MySQLDeliveryInput) { in.CharacterSet = "gbk"; in.Collation = "utf8mb4_general_ci" },
"bad max connections": func(in *MySQLDeliveryInput) { in.MaxConnections = "300" },
"bad redo capacity": func(in *MySQLDeliveryInput) { in.InnoDBRedoLogCapacity = "2G" },
"bad flush log": func(in *MySQLDeliveryInput) { in.InnoDBFlushLogAtTrxCommit = 3 },
"bad sync binlog": func(in *MySQLDeliveryInput) { in.SyncBinlog = 2 },
"bad io capacity": func(in *MySQLDeliveryInput) { in.InnoDBIOCapacity = 500 },
"bad long query time": func(in *MySQLDeliveryInput) { in.LongQueryTime = 3 },
"bad binlog expire": func(in *MySQLDeliveryInput) { in.BinlogExpireLogsSeconds = 3600 },
"bad max binlog size": func(in *MySQLDeliveryInput) { in.MaxBinlogSize = "64M" },
} {
input := valid
mutate(&input)