diff --git a/ansible/mysql-deploy.yml b/ansible/mysql-deploy.yml index 96b90e0..24d2cf2 100644 --- a/ansible/mysql-deploy.yml +++ b/ansible/mysql-deploy.yml @@ -6,6 +6,10 @@ any_errors_fatal: true vars: mysql_instance: "{{ instance_name }}" + mysql_version_value: "{{ mysql_version | default('8.0') }}" + mysql_package_map: + "8.0": "mysql-server=8.0.46-0ubuntu0.24.04.3" + mysql_package_name: "{{ mysql_package_map[mysql_version_value] }}" mysql_port_value: "{{ mysql_port | default(3307) | int }}" mysql_memory_mb_value: "{{ memory_mb | default(2048) | int }}" mysql_storage_gb_value: "{{ storage_gb | default(20) | int }}" @@ -22,6 +26,7 @@ that: - topology == 'standalone' - mysql_instance is match('^[a-z0-9][a-z0-9-]{0,62}$') + - mysql_version_value in mysql_package_map - (mysql_port_value | int) >= 1024 - (mysql_port_value | int) <= 65535 - (mysql_memory_mb_value | int) >= 1024 @@ -73,7 +78,7 @@ - name: Install Ubuntu MySQL package ansible.builtin.apt: - name: mysql-server=8.0.46-0ubuntu0.24.04.3 + name: "{{ mysql_package_name }}" state: present update_cache: true cache_valid_time: 3600 @@ -191,25 +196,28 @@ ansible.builtin.shell: | set -euo pipefail client_file="$(mktemp)" - trap 'rm -f "$client_file"' EXIT - chmod 600 "$client_file" - cat >"$client_file" <"$client_file" <<'EOF' [client] user=root password={{ mysql_root_password_value }} socket={{ mysql_run_dir }}/mysql.sock EOF if ! /usr/bin/mysql --defaults-extra-file="$client_file" -e 'SELECT 1' >/dev/null 2>&1; then - /usr/bin/mysql --protocol=socket --socket={{ mysql_run_dir }}/mysql.sock -uroot <<'SQL' - ALTER USER 'root'@'localhost' IDENTIFIED BY '{{ mysql_root_password_value }}'; - SQL + cat >"$sql_file" <<'EOF' + ALTER USER 'root'@'localhost' IDENTIFIED BY '{{ mysql_root_password_value | replace("'", "''") }}'; + EOF + /usr/bin/mysql --protocol=socket --socket={{ mysql_run_dir }}/mysql.sock -uroot <"$sql_file" fi - /usr/bin/mysql --defaults-extra-file="$client_file" <<'SQL' - CREATE USER IF NOT EXISTS 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value }}'; - ALTER USER 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value }}'; + cat >"$sql_file" <<'EOF' + CREATE USER IF NOT EXISTS 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value | replace("'", "''") }}'; + ALTER USER 'xinfra_admin'@'%' IDENTIFIED BY '{{ mysql_admin_password_value | replace("'", "''") }}'; GRANT ALL PRIVILEGES ON *.* TO 'xinfra_admin'@'%' WITH GRANT OPTION; FLUSH PRIVILEGES; - SQL + EOF + /usr/bin/mysql --defaults-extra-file="$client_file" <"$sql_file" args: executable: /bin/bash changed_when: false diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 5ecdb9f..3135a42 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -105,6 +105,9 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin var existing model.DeliveryTask if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil { + if existing.RequestedBy != userID { + return nil, false, fmt.Errorf("idempotency key is already in use by another user") + } return &existing, true, nil } else if !errors.Is(err, gorm.ErrRecordNotFound) { return nil, false, err @@ -159,6 +162,8 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin return &task, false, nil } +var supportedMySQLVersions = map[string]bool{"8.0": true} + 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") @@ -166,8 +171,11 @@ func validateDeliveryInput(input MySQLDeliveryInput) error { 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") + if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 1024 || input.MemoryMi > 4096 || input.StorageGi < 10 || input.StorageGi > 100 { + return fmt.Errorf("requested resources are outside the supported range (memory: 1024-4096 MiB, storage: 10-100 GiB)") + } + if input.MySQLVersion != "" && !supportedMySQLVersions[input.MySQLVersion] { + return fmt.Errorf("unsupported mysql_version %q, supported: 8.0", input.MySQLVersion) } return nil } @@ -268,7 +276,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT 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 { + 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 { @@ -411,19 +419,25 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa return nil, false, err } meta := parseTargetMetadata(target.Metadata) + // Persist execution record BEFORE launching AWX to ensure crash recovery. + now := time.Now() + execution := model.ExecutionJob{TaskID: task.ID, IdempotencyKey: task.IdempotencyKey, ExecutorJobID: "pending", Status: "launching", StartedAt: &now} + if err := s.db.WithContext(ctx).Create(&execution).Error; err != nil { + return nil, false, err + } 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, + "mysql_version": payload.MySQLVersion, }}) if err != nil { + _ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}) 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 { + if err := tx.Model(&execution).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil { return err } return s.transitionTx(tx, &task, model.TaskRunning, "AWX job started", "") @@ -447,16 +461,26 @@ func (s *DeliveryService) PollOnce(ctx context.Context) error { case "pending", "waiting", "running", "new": continue case "successful": + s.finishExecution(ctx, &execution, "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()) } + case "canceled": + s.finishExecution(ctx, &execution, "canceled") + _ = s.failTask(ctx, &model.DeliveryTask{ID: execution.TaskID, Status: model.TaskRunning}, model.TaskCanceled, "AWX job was canceled") default: + s.finishExecution(ctx, &execution, "failed") _ = 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) finishExecution(ctx context.Context, execution *model.ExecutionJob, status string) { + now := time.Now() + _ = s.db.WithContext(ctx).Model(execution).Updates(map[string]any{"status": status, "finished_at": now}) +} + 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 { @@ -533,7 +557,7 @@ func (s *DeliveryService) failTask(ctx context.Context, task *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 { + if current.Status != model.TaskPending && current.Status != model.TaskDispatching && current.Status != model.TaskRunning && current.Status != model.TaskRegistering && current.Status != model.TaskCanceling { return nil } if err := s.transitionTx(tx, ¤t, status, message, message); err != nil { diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index 6d2a88b..bc18b82 100644 --- a/server/internal/service/delivery_test.go +++ b/server/internal/service/delivery_test.go @@ -7,10 +7,19 @@ func TestValidateDeliveryInput(t *testing.T) { if err := validateDeliveryInput(valid); err != nil { t.Fatalf("valid input rejected: %v", err) } + withVersion := valid + withVersion.MySQLVersion = "8.0" + if err := validateDeliveryInput(withVersion); err != nil { + t.Fatalf("valid input with version 8.0 rejected: %v", err) + } for name, input := range map[string]MySQLDeliveryInput{ - "uppercase namespace": valid, - "bad instance": valid, - "too little memory": valid, + "uppercase namespace": valid, + "bad instance": valid, + "too little memory": valid, + "too much memory": valid, + "too little storage": valid, + "too much storage": valid, + "unsupported version": valid, } { switch name { case "uppercase namespace": @@ -18,7 +27,15 @@ func TestValidateDeliveryInput(t *testing.T) { case "bad instance": input.InstanceName = "mysql_01" case "too little memory": - input.MemoryMi = 128 + input.MemoryMi = 512 + case "too much memory": + input.MemoryMi = 8192 + case "too little storage": + input.StorageGi = 5 + case "too much storage": + input.StorageGi = 200 + case "unsupported version": + input.MySQLVersion = "5.7" } if err := validateDeliveryInput(input); err == nil { t.Errorf("%s was accepted", name)