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

Merge remote-tracking branch 'upstream/main' into feat/base-service-delivery

Browse Source 
# Conflicts:
#	ansible/mysql-deploy.yml
#	frontend/src/api/delivery.ts
#	frontend/src/views/service/Catalog.vue
#	server/.env.example
#	server/internal/config/config.go
#	server/internal/handler/task_log.go
#	server/internal/service/delivery.go
This commit is contained in:
mac
2026-07-28 15:08:37 +08:00
parent 8e09c30d21 7d6efeb3dd
commit 897670b721
15 changed files with 1145 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/config/config.go
+4
View File
@@ -66,6 +66,8 @@ type Config struct {
AWXWebhookToken string
AWXFactsTemplateID uint64
AWXFactsTimeoutSeconds int
RollbackTemplateID uint64
DeliveryServiceToken string
DeliverySchedulerEnabled bool
DeliveryDispatchSeconds int
DeliveryCallbackBaseURL string
@@ -138,6 +140,8 @@ func Load() Config {
AWXWebhookToken: env("AWX_WEBHOOK_TOKEN", ""),
AWXFactsTemplateID: envUint64("AWX_FACTS_TEMPLATE_ID", 0),
AWXFactsTimeoutSeconds: envInt("AWX_FACTS_TIMEOUT_SECONDS", 45),
RollbackTemplateID: uint64(envInt("DELIVERY_ROLLBACK_TEMPLATE_ID", 0)),
DeliveryServiceToken: env("DELIVERY_SERVICE_TOKEN", ""),
DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false),
DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5),
DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)),
server/internal/database/database.go
+1
View File
@@ -27,6 +27,7 @@ func AutoMigrate(db *gorm.DB) error {
&model.MySQLInstance{},
&model.ResourceUsage{},
&model.ExecutionJob{},
&model.RollbackJob{},
&model.TaskEvent{},
)
}
server/internal/handler/delivery.go
+49
View File
@@ -242,6 +242,55 @@ func (h *DeliveryHandler) Cancel(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"ok": true})
}
// RetryRollback retries the compensating AWX job after a previous cleanup failure.
func (h *DeliveryHandler) RetryRollback(c *gin.Context) {
if !requirePlatformAdmin(c) {
return
}
if err := h.service.RetryRollback(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusAccepted, gin.H{"ok": true, "status": model.TaskRollbackPending})
}
// AcknowledgeRollbackRelease releases bookkeeping after an administrator has
// independently verified that no instance artifacts remain on the target host.
func (h *DeliveryHandler) AcknowledgeRollbackRelease(c *gin.Context) {
if !requirePlatformAdmin(c) {
return
}
if err := h.service.AcknowledgeRollbackRelease(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"ok": true, "status": model.TaskRollbackAck})
}
// RetryCloudDMRegistration retries only the CloudDM registration step for an
// already healthy and accounted-for MySQL instance.
func (h *DeliveryHandler) RetryCloudDMRegistration(c *gin.Context) {
claims, ok := CurrentClaims(c)
if !ok {
c.JSON(http.StatusUnauthorized, gin.H{"error": "missing current user"})
return
}
taskID := c.Param("id")
if _, _, err := h.service.GetTask(c.Request.Context(), taskID, claims.UserID, claims.IsAdmin); err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
if err := h.service.RetryCloudDMRegistration(c.Request.Context(), taskID); err != nil {
c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusAccepted, gin.H{"ok": true, "status": model.TaskFinished})
}
// Targets 获取可用部署目标
// @Summary 获取可用部署目标
// @Description 从 AWX 动态返回可用 Job Template 及其 Inventory hosts
server/internal/handler/task_log.go
+26 -3
View File
@@ -138,7 +138,7 @@ func (h *TaskLogHandler) getAWXTask(c *gin.Context, taskID string, userID uint64
lines = append(lines, taskLogLine{Time: formatTaskLogTime(event.CreatedAt), Message: "[" + event.ToState + "] " + event.Message, Class: classForTaskStatus(event.ToState)})
}
var execution model.ExecutionJob
if err := h.db.WithContext(c.Request.Context()).Where("task_id = ?", task.ID).First(&execution).Error; err == nil && execution.ExecutorJobID != "" && execution.ExecutorJobID != "pending" && !strings.HasPrefix(execution.ExecutorJobID, "pending:") {
if err := h.db.WithContext(c.Request.Context()).Where("task_id = ?", task.ID).First(&execution).Error; err == nil && execution.ExecutorJobID != "" && execution.ExecutorJobID != "pending" && !strings.HasPrefix(execution.ExecutorJobID, "pending:") && !strings.HasPrefix(execution.ExecutorJobID, "pending-") {
stdout, stdoutErr := h.delivery.AWXJobStdout(c.Request.Context(), execution.ExecutorJobID)
if stdoutErr != nil {
lines = append(lines, taskLogLine{Time: formatTaskLogTime(time.Now()), Message: "[awx] stdout fetch failed: " + stdoutErr.Error(), Class: "err"})
@@ -146,6 +146,15 @@ func (h *TaskLogHandler) getAWXTask(c *gin.Context, taskID string, userID uint64
lines = append(lines, splitStdoutLines(stdout)...)
}
}
var rollback model.RollbackJob
if err := h.db.WithContext(c.Request.Context()).Where("task_id = ?", task.ID).First(&rollback).Error; err == nil && rollback.ExecutorJobID != "" && rollback.ExecutorJobID != "pending" && !strings.HasPrefix(rollback.ExecutorJobID, "pending-") {
stdout, stdoutErr := h.delivery.AWXJobStdout(c.Request.Context(), rollback.ExecutorJobID)
if stdoutErr != nil {
lines = append(lines, taskLogLine{Time: formatTaskLogTime(time.Now()), Message: "[rollback awx] stdout fetch failed: " + stdoutErr.Error(), Class: "err"})
} else {
lines = append(lines, splitStdoutLines(stdout)...)
}
}
c.JSON(http.StatusOK, gin.H{"task": awxTaskSummary(*task), "lines": lines})
}
@@ -257,8 +266,18 @@ func textForTaskStatus(status string) string {
return "等待"
case model.TaskRunning, model.TaskRegistering, model.TaskCanceling:
return "执行中"
case model.TaskRollbackPending, model.TaskRollingBack:
return "回退中"
case model.TaskFinished:
return "成功"
case model.TaskRolledBack:
return "已回退"
case model.TaskRollbackFailed:
return "回退失败"
case model.TaskRollbackAck:
return "已确认释放"
case model.TaskRegisterFailed:
return "注册失败（实例保留）"
case model.TaskCanceled:
return "已取消"
default:
@@ -270,10 +289,14 @@ func classForTaskStatus(status string) string {
switch status {
case model.TaskFinished:
return "ok"
case model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskRegisterFailed, model.TaskCanceled:
case model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskCanceled, model.TaskRollbackFailed:
return "err"
case model.TaskRunning, model.TaskDispatching, model.TaskRegistering, model.TaskCanceling:
case model.TaskRollbackAck, model.TaskRegisterFailed:
return "warn"
case model.TaskRunning, model.TaskDispatching, model.TaskRegistering, model.TaskCanceling, model.TaskRollbackPending, model.TaskRollingBack:
return "warn"
case model.TaskRolledBack:
return "ok"
default:
return ""
}
server/internal/model/delivery.go
+19
View File
@@ -14,6 +14,11 @@ const (
TaskValidationFailed = "validation_failed"
TaskCanceling = "canceling"
TaskCanceled = "canceled"
TaskRollbackPending = "rollback_pending"
TaskRollingBack = "rolling_back"
TaskRolledBack = "rolled_back"
TaskRollbackFailed = "rollback_failed"
TaskRollbackAck = "rollback_acknowledged"
)
type ResourceQuota struct {
@@ -111,6 +116,20 @@ type ExecutionJob struct {
UpdatedAt time.Time `json:"updated_at"`
}
// RollbackJob tracks the compensating AWX run independently from the deploy run.
// Keeping a separate record preserves both job IDs for audit and retry tooling.
type RollbackJob struct {
ID uint64 `gorm:"primaryKey" json:"id"`
TaskID string `gorm:"size:36;not null;uniqueIndex" json:"task_id"`
ExecutorJobID string `gorm:"size:128;not null" json:"executor_job_id"`
Status string `gorm:"size:32;not null;index" json:"status"`
Reason string `gorm:"type:text" json:"reason"`
StartedAt *time.Time `json:"started_at,omitempty"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type TaskEvent struct {
ID uint64 `gorm:"primaryKey" json:"id"`
TaskID string `gorm:"size:36;not null;index" json:"task_id"`
server/internal/router/router.go
+3
View File
@@ -152,6 +152,9 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) {
protected.GET("/delivery/tasks/:id", deliveryHandler.Get)
protected.GET("/delivery/tasks/:id/stream", deliveryHandler.Stream)
protected.POST("/delivery/tasks/:id/cancel", deliveryHandler.Cancel)
protected.POST("/delivery/tasks/:id/rollback/retry", deliveryHandler.RetryRollback)
protected.POST("/delivery/tasks/:id/rollback/release", deliveryHandler.AcknowledgeRollbackRelease)
protected.POST("/delivery/tasks/:id/clouddm/retry", deliveryHandler.RetryCloudDMRegistration)
protected.GET("/task-logs", taskLogHandler.List)
protected.GET("/task-logs/:id", taskLogHandler.Get)
}
server/internal/service/delivery.go
+572 -269
View File
File diff suppressed because it is too large Load Diff
server/internal/service/delivery_test.go
+59 -14
View File
@@ -1,6 +1,11 @@
package service
import "testing"
import (
"testing"
"time"
"github.com/1024XEngineer/xinfra/server/internal/model"
)
func TestValidateDeliveryInput(t *testing.T) {
dataDisks := []string{"/data", "/disk1"}
@@ -18,15 +23,18 @@ func TestValidateDeliveryInput(t *testing.T) {
full.CPUMilli = 2000
full.MemoryMi = 8192
full.StorageGi = 2000
full.TimeZone = "+08:00"
full.LowerCaseTableNames = 0
full.Timezone = "+08:00"
lowerCaseZero := 0
full.LowerCaseTableNames = &lowerCaseZero
full.CharacterSet = "utf8mb4"
full.Collation = "utf8mb4_general_ci"
full.MaxConnections = "auto"
full.InnoDBRedoLogCapacity = "256M"
full.InnoDBFlushLogAtTrxCommit = 2
full.SyncBinlog = 0
full.InnoDBIOCapacity = 2000
full.InnodbRedoLogCapacity = "256M"
flushLog := 2
full.InnodbFlushLogAtTrxCommit = &flushLog
syncBinlog := 0
full.SyncBinlog = &syncBinlog
full.InnodbIOCapacity = 2000
full.LongQueryTime = 0.5
full.BinlogExpireLogsSeconds = 604800
full.MaxBinlogSize = "512M"
@@ -34,7 +42,7 @@ func TestValidateDeliveryInput(t *testing.T) {
t.Fatalf("valid full input rejected: %v", err)
}
namedZone := valid
namedZone.TimeZone = "Asia/Shanghai"
namedZone.Timezone = "Asia/Shanghai"
if err := validateDeliveryInput(namedZone, dataDisks); err != nil {
t.Fatalf("named timezone rejected: %v", err)
}
@@ -58,15 +66,15 @@ func TestValidateDeliveryInput(t *testing.T) {
"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 timezone": func(in *MySQLDeliveryInput) { in.Timezone = "UTC+8" },
"bad lower case": func(in *MySQLDeliveryInput) { v := 2; in.LowerCaseTableNames = &v },
"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 redo capacity": func(in *MySQLDeliveryInput) { in.InnodbRedoLogCapacity = "2G" },
"bad flush log": func(in *MySQLDeliveryInput) { v := 3; in.InnodbFlushLogAtTrxCommit = &v },
"bad sync binlog": func(in *MySQLDeliveryInput) { v := 2; in.SyncBinlog = &v },
"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" },
@@ -122,3 +130,40 @@ func TestAllocatePort(t *testing.T) {
t.Fatal("exhausted pool still allocated a port")
}
}
func TestRollbackExtraVarsTargetsOnlyTheAllocatedInstance(t *testing.T) {
task := &model.DeliveryTask{ID: "task-1", TargetHost: "db-01"}
payload := deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceName: "mysql-a", DataDisk: "/disk1"}}
vars := rollbackExtraVars(task, payload)
if vars["target_hosts"] != "db-01" || vars["instance_name"] != "mysql-a" || vars["data_disk"] != "/disk1" {
t.Fatalf("rollback vars target the wrong instance: %#v", vars)
}
if vars["rollback"] != true {
t.Fatalf("rollback marker missing: %#v", vars)
}
}
func TestRegisterFailedIsProtectedFromRollback(t *testing.T) {
if !rollbackProtectedStatus(model.TaskRegisterFailed) {
t.Fatal("register_failed must preserve the healthy instance and resource usage")
}
if rollbackProtectedStatus(model.TaskValidationFailed) {
t.Fatal("validation_failed must still be eligible for cleanup rollback")
}
}
func TestRollbackLaunchExpired(t *testing.T) {
now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)
started := now.Add(-rollbackLaunchTimeout - time.Second)
if !rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: &started}, now) {
t.Fatal("stale launching rollback job must be recoverable")
}
if rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: ptrTime(now.Add(-rollbackLaunchTimeout + time.Second))}, now) {
t.Fatal("recent launching rollback job must remain pending")
}
if rollbackLaunchExpired(model.RollbackJob{Status: "running", StartedAt: &started}, now) {
t.Fatal("running rollback job is not a launch timeout")
}
}
func ptrTime(v time.Time) *time.Time { return &v }