feat(delivery): implement delivery scheduler and state machine
Core delivery service with: task creation (idempotent, validated), background scheduler (claimAndReserve with SELECT FOR UPDATE SKIP LOCKED), 4-level concurrency control, resource quota enforcement, host allocation from candidate pool, AWX job dispatch, polling, TCP health check, instance registration, CloudDM integration (optional), and cancel support. 11-state machine with optimistic locking and full event audit trail. Relates-to: #97
This commit is contained in:
mac
@@ -0,0 +1,565 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/1024XEngineer/xinfra/server/internal/config"
|
||||
"github.com/1024XEngineer/xinfra/server/internal/model"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
var dnsLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-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"`
|
||||
MySQLVersion string `json:"mysql_version"`
|
||||
CPUMilli int64 `json:"cpu_milli" binding:"required"`
|
||||
MemoryMi int64 `json:"memory_mi" binding:"required"`
|
||||
StorageGi int64 `json:"storage_gi" binding:"required"`
|
||||
}
|
||||
|
||||
type deliveryPayload struct {
|
||||
MySQLDeliveryInput
|
||||
TargetType string `json:"target_type"`
|
||||
}
|
||||
|
||||
// targetMetadata describes the native VM候选节点池以及部署形态，存储在 DeploymentTarget.Metadata (JSON)。
|
||||
type targetMetadata struct {
|
||||
Topology string `json:"topology"`
|
||||
MySQLPort int `json:"mysql_port"`
|
||||
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
|
||||
}
|
||||
|
||||
// firstFreeHost 返回候选池中第一个未被占用的节点。
|
||||
func firstFreeHost(hosts []targetHost, occupied []string) *targetHost {
|
||||
taken := make(map[string]bool, len(occupied))
|
||||
for _, h := range occupied {
|
||||
taken[h] = true
|
||||
}
|
||||
for i := range hosts {
|
||||
if !taken[hosts[i].Name] {
|
||||
return &hosts[i]
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type DeliveryService struct {
|
||||
db *gorm.DB
|
||||
cfg config.Config
|
||||
awx *AWXClient
|
||||
audit *AuditService
|
||||
executionMu sync.Mutex
|
||||
}
|
||||
|
||||
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}
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
if err := validateDeliveryInput(input); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
var existing model.DeliveryTask
|
||||
if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil {
|
||||
return &existing, true, nil
|
||||
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
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)
|
||||
}
|
||||
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"
|
||||
}
|
||||
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,
|
||||
TargetType: target.TargetType,
|
||||
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).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
|
||||
}
|
||||
_ = s.db.WithContext(ctx).Create(&model.TaskEvent{TaskID: task.ID, ToState: model.TaskPending, Message: "delivery task created"}).Error
|
||||
return &task, false, nil
|
||||
}
|
||||
|
||||
func validateDeliveryInput(input MySQLDeliveryInput) 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")
|
||||
}
|
||||
if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 256 || input.MemoryMi > 262144 || input.StorageGi < 1 || input.StorageGi > 16384 {
|
||||
return fmt.Errorf("requested resources are outside the supported range")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]model.DeliveryTask, error) {
|
||||
query := s.db.WithContext(ctx).Order("created_at DESC")
|
||||
if businessLineID != 0 {
|
||||
query = query.Where("business_line_id = ?", businessLineID)
|
||||
}
|
||||
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
|
||||
return tasks, query.Find(&tasks).Error
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
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) 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 model.DeploymentTarget
|
||||
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")
|
||||
}
|
||||
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 business_line_id = ?", []any{activeStates, task.BusinessLineID}, s.cfg.DeliveryBusinessLimit, "business line 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
var quota model.ResourceQuota
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", task.BusinessLineID, task.TargetID).First("a).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 {
|
||||
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 = ?", task.BusinessLineID, task.TargetID, "reserved").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 {
|
||||
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 := []string{model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskCanceled}
|
||||
if err := tx.Model(&model.DeliveryTask{}).Where("target_id = ? AND target_host <> ? AND status NOT IN ?", task.TargetID, "", occupiedExclude).Pluck("target_host", &occupied).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
host := firstFreeHost(meta.Hosts, occupied)
|
||||
if host == nil {
|
||||
return fmt.Errorf("defer: no free host available on target")
|
||||
}
|
||||
reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, 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, "mysql_port": meta.MySQLPort}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
task.TargetHost = host.Name
|
||||
task.TargetHostIP = host.IP
|
||||
task.MySQLPort = meta.MySQLPort
|
||||
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 (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 {
|
||||
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 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()
|
||||
}
|
||||
|
||||
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 {
|
||||
return s.failTask(ctx, task, model.TaskExecutionFailed, err.Error())
|
||||
}
|
||||
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.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)
|
||||
}
|
||||
var target model.DeploymentTarget
|
||||
if err := s.db.WithContext(ctx).First(&target, "id = ? AND enabled = ?", task.TargetID, true).Error; 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)
|
||||
job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: map[string]any{
|
||||
"task_id": task.ID, "payload_hash": task.PayloadHash,
|
||||
"target_hosts": task.TargetHost, "topology": meta.Topology,
|
||||
"instance_name": payload.InstanceName, "mysql_port": task.MySQLPort,
|
||||
"memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi,
|
||||
}})
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
now := time.Now()
|
||||
execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: fmt.Sprint(job.ID), Status: "running", StartedAt: &now}
|
||||
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(&execution).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) PollOnce(ctx context.Context) error {
|
||||
var jobs []model.ExecutionJob
|
||||
if err := s.db.WithContext(ctx).Where("status = ?", "running").Find(&jobs).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
for _, execution := range jobs {
|
||||
job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
switch strings.ToLower(job.Status) {
|
||||
case "pending", "waiting", "running", "new":
|
||||
continue
|
||||
case "successful":
|
||||
if err := s.completeTask(ctx, execution.TaskID); err != nil {
|
||||
_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskValidationFailed, err.Error())
|
||||
}
|
||||
default:
|
||||
_ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskExecutionFailed, "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
|
||||
}
|
||||
var payload deliveryPayload
|
||||
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
|
||||
return err
|
||||
}
|
||||
addr := fmt.Sprintf("%s:%d", task.TargetHostIP, task.MySQLPort)
|
||||
if err := mysqlReady(ctx, addr); err != nil {
|
||||
return fmt.Errorf("MySQL health check failed: %w", err)
|
||||
}
|
||||
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
|
||||
}
|
||||
instance := model.MySQLInstance{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, Namespace: payload.Namespace, Name: payload.InstanceName, NodeName: task.TargetHost, Host: task.TargetHostIP, Port: task.MySQLPort, Version: payload.MySQLVersion, Status: "active"}
|
||||
if err := tx.Create(&instance).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: instance.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, 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 {
|
||||
return s.transition(ctx, &task, model.TaskRegisterFailed, "CloudDM registration failed", err.Error())
|
||||
}
|
||||
return s.transition(ctx, &task, model.TaskFinished, "MySQL delivery completed", "")
|
||||
}
|
||||
|
||||
func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error {
|
||||
if s.cfg.CloudDMRegisterURL == "" {
|
||||
return nil
|
||||
}
|
||||
var instance model.MySQLInstance
|
||||
if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).First(&instance).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
body := map[string]any{"name": instance.Name, "host": instance.Host, "port": instance.Port, "username": "root", "database_type": "mysql"}
|
||||
raw, _ := json.Marshal(body)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMRegisterURL, bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if s.cfg.CloudDMAPIToken != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+s.cfg.CloudDMAPIToken)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("CloudDM returned %s", resp.Status)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
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(¤t, "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 {
|
||||
return nil
|
||||
}
|
||||
if err := s.transitionTx(tx, ¤t, 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.DeliveryPollSeconds) * 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package service
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestValidateDeliveryInput(t *testing.T) {
|
||||
valid := MySQLDeliveryInput{BusinessLineID: 1, TargetID: 1, Namespace: "team-a", InstanceName: "mysql-01", CPUMilli: 500, MemoryMi: 1024, StorageGi: 10}
|
||||
if err := validateDeliveryInput(valid); err != nil {
|
||||
t.Fatalf("valid input rejected: %v", err)
|
||||
}
|
||||
for name, input := range map[string]MySQLDeliveryInput{
|
||||
"uppercase namespace": valid,
|
||||
"bad instance": valid,
|
||||
"too little memory": valid,
|
||||
} {
|
||||
switch name {
|
||||
case "uppercase namespace":
|
||||
input.Namespace = "Team-A"
|
||||
case "bad instance":
|
||||
input.InstanceName = "mysql_01"
|
||||
case "too little memory":
|
||||
input.MemoryMi = 128
|
||||
}
|
||||
if err := validateDeliveryInput(input); err == nil {
|
||||
t.Errorf("%s was accepted", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user