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

feat(delivery): integrate PostgreSQL delivery

Browse Source
This commit is contained in:
l x
2026-07-31 10:30:03 +08:00
parent 048ac6507e
commit 66fde73792
25 changed files with 2110 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/service/delivery.go
+54 -3
View File
@@ -160,7 +160,7 @@ type DeploymentCredentialView struct {
// targetMetadata describes the native VM候选节点池以及部署形态，由 AWX inventory hosts 动态组装。
type targetMetadata struct {
Topology string `json:"topology"`
MySQLPort int `json:"mysql_port"`
MySQLPort int `json:"mysql_port,omitempty"`
Hosts []targetHost `json:"hosts"`
}
@@ -410,6 +410,14 @@ func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([]
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
@@ -692,6 +700,7 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin
RequestedBy: userID,
Component: "mysql",
TargetType: target.TargetType,
ServiceType: "mysql",
TargetID: target.ID,
Namespace: input.Namespace,
InstanceName: input.InstanceName,
@@ -1376,6 +1385,9 @@ func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskI
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")
}
@@ -1593,7 +1605,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
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 = ?", model.TaskPending).Order("created_at ASC").First(&task).Error; err != nil {
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
@@ -1930,6 +1942,9 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa
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")
}
@@ -2063,6 +2078,9 @@ func (s *DeliveryService) HandleStageEvent(ctx context.Context, taskID string, i
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 {
@@ -2118,6 +2136,14 @@ func (s *DeliveryService) HandleAWXJobNotification(ctx context.Context, input AW
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 {
@@ -2275,7 +2301,10 @@ func (s *DeliveryService) finishExecution(ctx context.Context, execution *model.
func (s *DeliveryService) PollOnce(ctx context.Context) error {
var jobs []model.ExecutionJob
if err := s.db.WithContext(ctx).Where("status = ?", "running").Find(&jobs).Error; err != nil {
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 {
@@ -2309,6 +2338,9 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error
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
@@ -2380,6 +2412,9 @@ func (s *DeliveryService) RetryCloudDMRegistration(ctx context.Context, taskID s
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)
}
@@ -2657,6 +2692,9 @@ func (s *DeliveryService) beginRollback(ctx context.Context, taskID, reason stri
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
}
@@ -2743,6 +2781,9 @@ func (s *DeliveryService) RetryRollback(ctx context.Context, taskID string) erro
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)
}
@@ -2770,6 +2811,9 @@ func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID
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)
}
@@ -2888,6 +2932,9 @@ func (s *DeliveryService) completeRollback(ctx context.Context, taskID string) e
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)
}
@@ -2907,6 +2954,10 @@ func (s *DeliveryService) completeRollback(ctx context.Context, taskID string) e
})
}
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