l x
427 lines
18 KiB
Go
427 lines
18 KiB
Go
package service
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/1024XEngineer/xinfra/server/internal/model"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type postgresqlCloudDMRegisterRequest struct {
|
|
SourceSystem string `json:"sourceSystem"`
|
|
ResourceType string `json:"resourceType"`
|
|
ExternalResourceID string `json:"externalResourceId"`
|
|
DataSource postgresqlCloudDMDataSource `json:"dataSource"`
|
|
}
|
|
|
|
type postgresqlCloudDMDataSource struct {
|
|
InstanceName string `json:"instanceName"`
|
|
InstanceDesc string `json:"instanceDesc"`
|
|
DSType string `json:"dsType"`
|
|
Host string `json:"host"`
|
|
ClusterID *uint64 `json:"cluster_id"`
|
|
PostgreSQLVersion string `json:"postgresql_version"`
|
|
SecurityType string `json:"securityType"`
|
|
UserName string `json:"userName"`
|
|
Password string `json:"password"`
|
|
DefaultSchema string `json:"defaultSchema"`
|
|
ClientTimeZone string `json:"clientTimeZone"`
|
|
ConnectTimeoutMs int `json:"connectTimeoutMs"`
|
|
SocketTimeoutSecs int `json:"socketTimeoutSeconds"`
|
|
ConnectionCharset string `json:"connectionCharset"`
|
|
}
|
|
|
|
func postgresqlCloudDMExternalResourceID(instanceID uint64) string {
|
|
return "postgresql-instance:" + strconv.FormatUint(instanceID, 10)
|
|
}
|
|
|
|
func buildPostgreSQLCloudDMRegisterRequest(instance model.PostgreSQLInstance, cluster model.PostgreSQLCluster, username, password string) postgresqlCloudDMRegisterRequest {
|
|
description := fmt.Sprintf("%s PostgreSQL %s", cluster.Name, instance.Role)
|
|
return postgresqlCloudDMRegisterRequest{
|
|
SourceSystem: "xinfra",
|
|
ResourceType: "POSTGRESQL_INSTANCE",
|
|
ExternalResourceID: postgresqlCloudDMExternalResourceID(instance.ID),
|
|
DataSource: postgresqlCloudDMDataSource{
|
|
InstanceName: instance.InstanceID,
|
|
InstanceDesc: description,
|
|
DSType: "PostgreSQL",
|
|
Host: net.JoinHostPort(instance.HostIP, strconv.Itoa(instance.Port)),
|
|
ClusterID: nil,
|
|
PostgreSQLVersion: instance.VersionMajor,
|
|
SecurityType: "USER_PASSWD",
|
|
UserName: username,
|
|
Password: password,
|
|
DefaultSchema: "postgres",
|
|
ClientTimeZone: "Asia/Shanghai",
|
|
ConnectTimeoutMs: 5000,
|
|
SocketTimeoutSecs: 10,
|
|
ConnectionCharset: "utf8",
|
|
},
|
|
}
|
|
}
|
|
|
|
func postgresqlCloudDMMetadata(existing map[string]any, instance model.PostgreSQLInstance, dataSourceID uint64) map[string]any {
|
|
meta := map[string]any{}
|
|
for key, value := range existing {
|
|
meta[key] = value
|
|
}
|
|
meta["clouddm"] = map[string]any{
|
|
"data_source_id": dataSourceID,
|
|
"external_resource_id": postgresqlCloudDMExternalResourceID(instance.ID),
|
|
"registered_at": time.Now().Format(time.RFC3339),
|
|
}
|
|
return meta
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error {
|
|
registerURL := strings.TrimSpace(s.cfg.CloudDMPostgreSQLRegisterURL)
|
|
if registerURL == "" {
|
|
return nil
|
|
}
|
|
if strings.TrimSpace(s.cfg.CloudDMAPIToken) == "" {
|
|
return fmt.Errorf("PostgreSQL CloudDM registration is enabled but CLOUDDM_API_TOKEN is not configured")
|
|
}
|
|
username := strings.TrimSpace(s.cfg.CloudDMPostgreSQLUsername)
|
|
password := s.cfg.CloudDMPostgreSQLPassword
|
|
if username == "" || password == "" {
|
|
return fmt.Errorf("PostgreSQL CloudDM registration requires CLOUDDM_POSTGRESQL_USERNAME and CLOUDDM_POSTGRESQL_PASSWORD")
|
|
}
|
|
var cluster model.PostgreSQLCluster
|
|
if err := s.db.WithContext(ctx).First(&cluster, "task_id = ?", taskID).Error; err != nil {
|
|
return err
|
|
}
|
|
var instances []model.PostgreSQLInstance
|
|
if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).Order("id ASC").Find(&instances).Error; err != nil {
|
|
return err
|
|
}
|
|
if len(instances) == 0 {
|
|
return fmt.Errorf("PostgreSQL task has no instances to register")
|
|
}
|
|
client := &http.Client{Timeout: 30 * time.Second}
|
|
for _, instance := range instances {
|
|
body := buildPostgreSQLCloudDMRegisterRequest(instance, cluster, username, password)
|
|
raw, err := json.Marshal(body)
|
|
if err != nil {
|
|
return fmt.Errorf("encode PostgreSQL CloudDM registration request: %w", err)
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, registerURL, bytes.NewReader(raw))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(s.cfg.CloudDMAPIToken))
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("X-Request-ID", randomUUID())
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
respBody, readErr := io.ReadAll(resp.Body)
|
|
closeErr := resp.Body.Close()
|
|
if readErr != nil {
|
|
return readErr
|
|
}
|
|
if closeErr != nil {
|
|
return closeErr
|
|
}
|
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
|
return fmt.Errorf("PostgreSQL CloudDM registration returned %s: %s", resp.Status, strings.TrimSpace(string(respBody)))
|
|
}
|
|
dataSourceID, ok := cloudDMDataSourceIDFromResponse(respBody)
|
|
if !ok {
|
|
return fmt.Errorf("PostgreSQL CloudDM registration response did not include dataSourceId: %s", strings.TrimSpace(string(respBody)))
|
|
}
|
|
meta := postgresqlCloudDMMetadata(metadataMap(instance.Metadata), instance, dataSourceID)
|
|
if err := s.db.WithContext(ctx).Model(&instance).Update("metadata", string(mustJSON(meta))).Error; err != nil {
|
|
return fmt.Errorf("save PostgreSQL CloudDM dataSourceId: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) RetryCloudDMRegistration(ctx context.Context, taskID string) error {
|
|
var task model.DeliveryTask
|
|
if err := s.db.WithContext(ctx).First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if task.Status != model.TaskRegisterFailed {
|
|
return fmt.Errorf("task %s is in state %q and cannot retry CloudDM registration", taskID, task.Status)
|
|
}
|
|
if err := s.common.transition(ctx, &task, model.TaskRegistering, "PostgreSQL CloudDM registration retry started", ""); err != nil {
|
|
return err
|
|
}
|
|
if err := s.RegisterCloudDM(ctx, taskID); err != nil {
|
|
message := fmt.Sprintf("PostgreSQL CloudDM registration failed: %v", err)
|
|
if transitionErr := s.common.transition(ctx, &task, model.TaskRegisterFailed, "PostgreSQL CloudDM registration retry failed; instances remain active", message); transitionErr != nil {
|
|
return fmt.Errorf("%s; cannot restore register_failed: %w", message, transitionErr)
|
|
}
|
|
return err
|
|
}
|
|
return s.common.transition(ctx, &task, model.TaskFinished, "PostgreSQL CloudDM registration completed", "")
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) beginRollback(ctx context.Context, taskID, reason string) error {
|
|
var task model.DeliveryTask
|
|
if err := s.db.WithContext(ctx).First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if rollbackProtectedStatus(task.Status) {
|
|
return nil
|
|
}
|
|
if err := s.common.holdReservationForRollback(ctx, task.ID); err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot hold PostgreSQL resources during rollback: "+err.Error())
|
|
}
|
|
if s.cfg.PostgreSQLRollbackTemplateID == 0 {
|
|
return s.markRollbackFailed(ctx, task.ID, "PostgreSQL automatic rollback is not configured; manual cleanup is required: "+reason)
|
|
}
|
|
if err := s.common.transition(ctx, &task, model.TaskRollbackPending, "PostgreSQL automatic rollback queued", reason); err != nil {
|
|
return err
|
|
}
|
|
if err := s.setRollbackResourceStatus(ctx, task.ID, "rollback_pending"); err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot mark PostgreSQL resources for rollback: "+err.Error())
|
|
}
|
|
return s.launchRollbackJob(ctx, &task, reason)
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) launchRollbackJob(ctx context.Context, task *model.DeliveryTask, reason string) error {
|
|
target, err := getPostgreSQLTarget(ctx, s.awx, task.TargetID)
|
|
if err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "PostgreSQL rollback target is unavailable: "+err.Error())
|
|
}
|
|
var instances []model.PostgreSQLInstance
|
|
if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).Order("id ASC").Find(&instances).Error; err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot load PostgreSQL rollback plan: "+err.Error())
|
|
}
|
|
if len(instances) == 0 {
|
|
return s.markRollbackFailed(ctx, task.ID, "PostgreSQL rollback plan has no instances")
|
|
}
|
|
instanceVars := make(map[string]any, len(instances))
|
|
for _, instance := range instances {
|
|
instanceVars[instance.Hostname] = map[string]any{
|
|
"instance_id": instance.InstanceID,
|
|
"data_dir": instance.DataDir,
|
|
"config_dir": instance.ConfigDir,
|
|
"log_dir": instance.LogDir,
|
|
"systemd_unit": instance.SystemdUnit,
|
|
}
|
|
}
|
|
now := time.Now()
|
|
var rollback model.RollbackJob
|
|
if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&rollback).Error; errors.Is(err, gorm.ErrRecordNotFound) {
|
|
rollback = model.RollbackJob{TaskID: task.ID}
|
|
} else if err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot load PostgreSQL rollback job: "+err.Error())
|
|
}
|
|
rollback.ExecutorJobID = "pending-postgresql-rollback-" + task.ID
|
|
rollback.Status = "launching"
|
|
rollback.Reason = reason
|
|
rollback.StartedAt = &now
|
|
rollback.FinishedAt = nil
|
|
if err := s.db.WithContext(ctx).Save(&rollback).Error; err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot persist PostgreSQL rollback job: "+err.Error())
|
|
}
|
|
job, err := s.awx.Launch(ctx, s.cfg.PostgreSQLRollbackTemplateID, AWXLaunchRequest{
|
|
InventoryID: target.AWXInventoryID,
|
|
Limit: task.TargetHost,
|
|
ExtraVars: map[string]any{
|
|
"target_hosts": task.TargetHost,
|
|
"postgresql_instances": instanceVars,
|
|
"task_id": task.ID,
|
|
"rollback": true,
|
|
},
|
|
})
|
|
if err != nil {
|
|
_ = s.db.WithContext(ctx).Model(&rollback).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}).Error
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot launch PostgreSQL rollback AWX job: "+err.Error())
|
|
}
|
|
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var current model.DeliveryTask
|
|
if err := tx.First(¤t, "id = ? AND service_type = ?", task.ID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if current.Status != model.TaskRollbackPending {
|
|
return fmt.Errorf("task %s is no longer pending rollback (state %q)", task.ID, current.Status)
|
|
}
|
|
if err := tx.Model(&rollback).Updates(map[string]any{"executor_job_id": fmt.Sprint(job.ID), "status": "running"}).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ?", task.ID).Update("status", "rolling_back").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ?", task.ID).Update("status", "rolling_back").Error; err != nil {
|
|
return err
|
|
}
|
|
return s.common.transitionTx(tx, ¤t, model.TaskRollingBack, "PostgreSQL rollback AWX job started", reason)
|
|
}); err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, "cannot record PostgreSQL rollback AWX job: "+err.Error())
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) RetryRollback(ctx context.Context, taskID string) error {
|
|
var task model.DeliveryTask
|
|
if err := s.db.WithContext(ctx).First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if task.Status != model.TaskRollbackFailed {
|
|
return fmt.Errorf("task %s is in state %q and cannot retry rollback", taskID, task.Status)
|
|
}
|
|
if s.cfg.PostgreSQLRollbackTemplateID == 0 {
|
|
return fmt.Errorf("PostgreSQL automatic rollback is not configured")
|
|
}
|
|
reason := "manual PostgreSQL rollback retry requested"
|
|
if err := s.common.transition(ctx, &task, model.TaskRollbackPending, "manual PostgreSQL rollback retry queued", reason); err != nil {
|
|
return err
|
|
}
|
|
if err := s.common.holdReservationForRollback(ctx, task.ID); err != nil {
|
|
_ = s.markRollbackFailed(ctx, task.ID, err.Error())
|
|
return err
|
|
}
|
|
if err := s.setRollbackResourceStatus(ctx, task.ID, "rollback_pending"); err != nil {
|
|
return s.markRollbackFailed(ctx, task.ID, err.Error())
|
|
}
|
|
return s.launchRollbackJob(ctx, &task, reason)
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID string) error {
|
|
now := time.Now()
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var task model.DeliveryTask
|
|
if err := tx.First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if task.Status != model.TaskRollbackFailed {
|
|
return fmt.Errorf("task %s is in state %q and cannot acknowledge rollback release", taskID, task.Status)
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ?", taskID).Update("status", "rollback_acknowledged").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ?", taskID).Update("status", "rollback_acknowledged").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now}).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "rollback").Update("status", "released").Error; err != nil {
|
|
return err
|
|
}
|
|
return s.common.transitionTx(tx, &task, model.TaskRollbackAck, "PostgreSQL rollback release acknowledged by platform admin", "")
|
|
})
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) setRollbackResourceStatus(ctx context.Context, taskID, status string) error {
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ?", taskID).Update("status", status).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ?", taskID).Update("status", status).Error
|
|
})
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) markRollbackFailed(ctx context.Context, taskID, message string) error {
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var task model.DeliveryTask
|
|
if err := tx.First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if task.Status == model.TaskRolledBack || task.Status == model.TaskFinished || task.Status == model.TaskRegisterFailed || task.Status == model.TaskRollbackAck {
|
|
return nil
|
|
}
|
|
if task.Status != model.TaskRollbackFailed {
|
|
if err := s.common.transitionTx(tx, &task, model.TaskRollbackFailed, "PostgreSQL automatic rollback failed; manual cleanup is required", message); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ?", taskID).Update("status", "rollback_failed").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ?", taskID).Update("status", "rollback_failed").Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Model(&model.ExecutionJob{}).Where("task_id = ? AND status IN ?", taskID, []string{"launching", "running"}).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}).Error
|
|
})
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) PollRollbackOnce(ctx context.Context) error {
|
|
var jobs []model.RollbackJob
|
|
if err := s.db.WithContext(ctx).
|
|
Joins("JOIN delivery_tasks ON delivery_tasks.id = rollback_jobs.task_id").
|
|
Where("rollback_jobs.status IN ? AND delivery_tasks.service_type = ?", []string{"launching", "running"}, postgresqlServiceType).
|
|
Find(&jobs).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, rollback := range jobs {
|
|
if rollback.Status == "launching" {
|
|
if !rollbackLaunchExpired(rollback, time.Now()) {
|
|
continue
|
|
}
|
|
expired, err := s.common.expireRollbackLaunch(ctx, rollback.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if expired {
|
|
_ = s.markRollbackFailed(ctx, rollback.TaskID, "PostgreSQL rollback launch timed out before its AWX job ID was recorded; manual retry is required")
|
|
}
|
|
continue
|
|
}
|
|
job, err := s.awx.GetJob(ctx, rollback.ExecutorJobID)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
switch strings.ToLower(job.Status) {
|
|
case "pending", "waiting", "running", "new":
|
|
continue
|
|
case "successful":
|
|
_ = s.common.finishRollbackJob(ctx, &rollback, "successful")
|
|
if err := s.completeRollback(ctx, rollback.TaskID); err != nil {
|
|
_ = s.markRollbackFailed(ctx, rollback.TaskID, err.Error())
|
|
}
|
|
case "canceled":
|
|
_ = s.common.finishRollbackJob(ctx, &rollback, "canceled")
|
|
_ = s.markRollbackFailed(ctx, rollback.TaskID, "PostgreSQL rollback AWX job was canceled")
|
|
default:
|
|
_ = s.common.finishRollbackJob(ctx, &rollback, "failed")
|
|
_ = s.markRollbackFailed(ctx, rollback.TaskID, "PostgreSQL rollback AWX job finished with status "+job.Status)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) completeRollback(ctx context.Context, taskID string) error {
|
|
now := time.Now()
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var task model.DeliveryTask
|
|
if err := tx.First(&task, "id = ? AND service_type = ?", taskID, postgresqlServiceType).Error; err != nil {
|
|
return err
|
|
}
|
|
if task.Status != model.TaskRollingBack {
|
|
return fmt.Errorf("task %s is in state %q, cannot complete PostgreSQL rollback", taskID, task.Status)
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ?", taskID).Update("status", "rolled_back").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ?", taskID).Update("status", "rolled_back").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now}).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status IN ?", taskID, []string{"reserved", "consumed", "rollback"}).Update("status", "released").Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.ExecutionJob{}).Where("task_id = ? AND status IN ?", taskID, []string{"launching", "running"}).Updates(map[string]any{"status": "failed", "finished_at": now}).Error; err != nil {
|
|
return err
|
|
}
|
|
return s.common.transitionTx(tx, &task, model.TaskRolledBack, "PostgreSQL instance rollback completed", "")
|
|
})
|
|
}
|