feat: switch base delivery to awx task logs

2026-07-23 18:36:24 +08:00
parent c2a3bbbdc0
commit a934380c9c
20 changed files with 1638 additions and 1243 deletions
@@ -40,7 +40,17 @@ type deliveryPayload struct {
 	TargetType string `json:"target_type"`
 }

-// targetMetadata describes the native VM候选节点池以及部署形态，存储在 DeploymentTarget.Metadata (JSON)。
+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"`
+}
+
+// targetMetadata describes the native VM候选节点池以及部署形态，由 AWX inventory hosts 动态组装。
 type targetMetadata struct {
 	Topology  string       `json:"topology"`
 	MySQLPort int          `json:"mysql_port"`
@@ -91,7 +101,68 @@ type DeliveryService struct {
 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), audit: audit}
+	return &DeliveryService{db: db, cfg: cfg, awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword), audit: audit}
+}
+
+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))
+	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
+			}
+		}
+		target, err := s.awxDeliveryTarget(ctx, template)
+		if err != nil {
+			continue
+		}
+		targets = append(targets, target)
+	}
+	return targets, 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)
+	}
+	return s.awxDeliveryTarget(ctx, *template)
+}
+
+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) {
@@ -113,9 +184,9 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin
 		return nil, false, err
 	}

-	var target model.DeploymentTarget
-	if err := s.db.WithContext(ctx).First(&target, "id = ? AND enabled = ?", input.TargetID, true).Error; err != nil {
-		return nil, false, fmt.Errorf("deployment target is unavailable: %w", 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)
@@ -208,6 +279,10 @@ func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uin
 	return &task, events, nil
 }

+func (s *DeliveryService) AWXJobStdout(ctx context.Context, jobID string) (string, error) {
+	return s.awx.JobStdout(ctx, jobID)
+}
+
 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 {
@@ -232,14 +307,16 @@ func (s *DeliveryService) Cancel(ctx context.Context, taskID string, userID uint
 func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryTask, error) {
 	var task model.DeliveryTask
 	var payload deliveryPayload
-	var target model.DeploymentTarget
+	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 {
 			return err
 		}
-		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&target, "id = ? AND enabled = ?", task.TargetID, true).Error; err != nil {
-			return s.failInTransaction(tx, &task, model.TaskValidationFailed, "deployment target is unavailable")
+		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")
@@ -267,19 +344,11 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
 				}
 			}
 		}
-		var quota model.ResourceQuota
-		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", task.BusinessLineID, task.TargetID).First(&quota).Error; err != nil {
-			return s.failInTransaction(tx, &task, model.TaskValidationFailed, "resource quota is not configured")
-		}
-		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 = ?", task.BusinessLineID, task.TargetID, "active").Scan(&used).Error; err != nil {
+		quotaOK, err := checkResourceQuota(tx, task.BusinessLineID, task.TargetID, payload)
+		if err != nil {
 			return 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 = ? AND expires_at > ?", task.BusinessLineID, task.TargetID, "reserved", time.Now()).Scan(&reserved).Error; err != nil {
-			return err
-		}
-		if used.CPU+reserved.CPU+payload.CPUMilli > quota.CPUMilli || used.Memory+reserved.Memory+payload.MemoryMi > quota.MemoryMi || used.Storage+reserved.Storage+payload.StorageGi > quota.StorageGi || used.Instances+reserved.Instances+1 > quota.InstanceLimit {
+		if !quotaOK {
 			return s.failInTransaction(tx, &task, model.TaskValidationFailed, "resource quota is insufficient")
 		}
 		meta := parseTargetMetadata(target.Metadata)
@@ -317,6 +386,28 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
 	return &task, err
 }

+func checkResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload deliveryPayload) (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 = ? AND expires_at > ?", businessLineID, targetID, "reserved", time.Now()).Scan(&reserved).Error; err != nil {
+		return false, err
+	}
+	return used.CPU+reserved.CPU+payload.CPUMilli <= quota.CPUMilli &&
+		used.Memory+reserved.Memory+payload.MemoryMi <= quota.MemoryMi &&
+		used.Storage+reserved.Storage+payload.StorageGi <= quota.StorageGi &&
+		used.Instances+reserved.Instances+1 <= quota.InstanceLimit, nil
+}
+
 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
@@ -410,8 +501,8 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa
 	if task.Status != model.TaskDispatching {
 		return nil, false, fmt.Errorf("task in state %q is not ready for execution", task.Status)
 	}
-	var target model.DeploymentTarget
-	if err := s.db.WithContext(ctx).First(&target, "id = ? AND enabled = ?", task.TargetID, true).Error; err != nil {
+	target, err := s.getTarget(ctx, task.TargetID)
+	if err != nil {
 		return nil, false, err
 	}
 	var payload deliveryPayload
@@ -455,6 +546,8 @@ func (s *DeliveryService) PollOnce(ctx context.Context) error {
 	for _, execution := range jobs {
 		job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
 		if err != nil {
+			s.finishExecution(ctx, &execution, "failed")
+			_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "poll AWX job "+execution.ExecutorJobID+": "+err.Error())
 			continue
 		}
 		switch strings.ToLower(job.Status) {