l x
603 lines
26 KiB
Go
603 lines
26 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/1024XEngineer/xinfra/server/internal/config"
|
|
"github.com/1024XEngineer/xinfra/server/internal/model"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
const (
|
|
postgresqlPortPoolStart = 15432
|
|
postgresqlPortPoolEnd = 15999
|
|
postgresqlServiceType = "postgresql"
|
|
)
|
|
|
|
var supportedPostgreSQLVersions = map[string]bool{"15": true, "16": true}
|
|
var supportedPostgreSQLTopologies = map[string]bool{"standalone": true, "primary_replica": true}
|
|
var postgresqlNamePattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`)
|
|
var errPostgreSQLTaskHandled = errors.New("postgresql task validation completed without dispatch")
|
|
|
|
type PostgreSQLDeliveryInput struct {
|
|
BusinessLineID uint64 `json:"business_line_id" binding:"required"`
|
|
TargetID uint64 `json:"target_id" binding:"required"`
|
|
Namespace string `json:"namespace" binding:"required"`
|
|
ClusterName string `json:"cluster_name" binding:"required"`
|
|
VersionMajor string `json:"version_major"`
|
|
PostgreSQLVersion string `json:"postgresql_version,omitempty"`
|
|
Topology string `json:"topology"`
|
|
ReplicaCount int `json:"replica_count"`
|
|
TargetHosts []string `json:"target_hosts,omitempty"`
|
|
CPUMilli int64 `json:"cpu_milli" binding:"required"`
|
|
MemoryMi int64 `json:"memory_mi" binding:"required"`
|
|
StorageGi int64 `json:"storage_gi" binding:"required"`
|
|
DataRoot string `json:"data_root"`
|
|
MaxConnections int `json:"max_connections"`
|
|
}
|
|
|
|
type postgresqlDeliveryPayload struct {
|
|
PostgreSQLDeliveryInput
|
|
TargetType string `json:"target_type"`
|
|
}
|
|
|
|
func validatePostgreSQLDeliveryInput(input PostgreSQLDeliveryInput, dataDisks []string) error {
|
|
_ = dataDisks
|
|
if input.VersionMajor == "" {
|
|
input.VersionMajor = input.PostgreSQLVersion
|
|
}
|
|
if input.PostgreSQLVersion != "" && input.VersionMajor != "" && input.PostgreSQLVersion != input.VersionMajor {
|
|
return fmt.Errorf("version_major and postgresql_version must match")
|
|
}
|
|
if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) {
|
|
return fmt.Errorf("namespace must be a valid Kubernetes DNS label")
|
|
}
|
|
if len(input.ClusterName) > 63 || !postgresqlNamePattern.MatchString(input.ClusterName) {
|
|
return fmt.Errorf("cluster_name must be a valid DNS label")
|
|
}
|
|
if !supportedPostgreSQLVersions[input.VersionMajor] {
|
|
return fmt.Errorf("unsupported version_major %q, supported: 15, 16", input.VersionMajor)
|
|
}
|
|
topology := input.Topology
|
|
if topology == "" {
|
|
topology = "standalone"
|
|
}
|
|
if !supportedPostgreSQLTopologies[topology] {
|
|
return fmt.Errorf("unsupported topology %q, supported: standalone, primary_replica", topology)
|
|
}
|
|
if topology == "standalone" && input.ReplicaCount != 0 {
|
|
return fmt.Errorf("standalone topology cannot have replicas")
|
|
}
|
|
if topology == "primary_replica" && (input.ReplicaCount < 1 || input.ReplicaCount > 7) {
|
|
return fmt.Errorf("replica_count must be between 1 and 7 for primary_replica")
|
|
}
|
|
if input.CPUMilli < 100 || input.CPUMilli > 64000 || input.MemoryMi < 2048 || input.MemoryMi > 65536 || input.StorageGi < 20 || input.StorageGi > 2000 {
|
|
return fmt.Errorf("requested resources are outside the supported range (memory: 2048-65536 MiB, storage: 20-2000 GiB)")
|
|
}
|
|
nodes := 1
|
|
if topology == "primary_replica" {
|
|
nodes += input.ReplicaCount
|
|
}
|
|
if len(input.TargetHosts) > 0 && len(input.TargetHosts) != nodes {
|
|
return fmt.Errorf("target_hosts must contain exactly %d distinct hosts", nodes)
|
|
}
|
|
seen := map[string]bool{}
|
|
for _, host := range input.TargetHosts {
|
|
if len(host) > 253 || !hostNamePattern.MatchString(host) || seen[host] {
|
|
return fmt.Errorf("target_hosts must contain unique valid inventory host names")
|
|
}
|
|
seen[host] = true
|
|
}
|
|
if input.DataRoot != "" {
|
|
if input.DataRoot != "/data/postgresql" {
|
|
return fmt.Errorf("data_root is fixed to /data/postgresql in the first release")
|
|
}
|
|
}
|
|
if input.MaxConnections < 0 || input.MaxConnections > 10000 {
|
|
return fmt.Errorf("max_connections must be between 0 and 10000")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func allocatePostgreSQLPort(requested int, used []int) (int, error) {
|
|
taken := make(map[int]bool, len(used))
|
|
for _, port := range used {
|
|
taken[port] = true
|
|
}
|
|
if requested != 0 {
|
|
if requested < postgresqlPortPoolStart || requested > postgresqlPortPoolEnd {
|
|
return 0, fmt.Errorf("postgresql_port must be within %d-%d", postgresqlPortPoolStart, postgresqlPortPoolEnd)
|
|
}
|
|
if taken[requested] {
|
|
return 0, fmt.Errorf("postgresql port %d is already allocated on the target host", requested)
|
|
}
|
|
return requested, nil
|
|
}
|
|
for port := postgresqlPortPoolStart; port <= postgresqlPortPoolEnd; port++ {
|
|
if !taken[port] {
|
|
return port, nil
|
|
}
|
|
}
|
|
return 0, fmt.Errorf("postgresql port pool %d-%d is exhausted on the target host", postgresqlPortPoolStart, postgresqlPortPoolEnd)
|
|
}
|
|
|
|
type postgresqlPortProbe func(context.Context, string, int) bool
|
|
|
|
func allocateReachablePostgreSQLPort(ctx context.Context, host string, used []int, inUse postgresqlPortProbe) (int, error) {
|
|
occupied := append([]int(nil), used...)
|
|
for {
|
|
port, err := allocatePostgreSQLPort(0, occupied)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if !inUse(ctx, host, port) {
|
|
return port, nil
|
|
}
|
|
occupied = append(occupied, port)
|
|
}
|
|
}
|
|
|
|
func postgresqlPortInUse(ctx context.Context, host string, port int) bool {
|
|
probeCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
|
|
defer cancel()
|
|
conn, err := (&net.Dialer{}).DialContext(probeCtx, "tcp", net.JoinHostPort(host, strconv.Itoa(port)))
|
|
if err != nil {
|
|
return false
|
|
}
|
|
_ = conn.Close()
|
|
return true
|
|
}
|
|
|
|
type PostgreSQLDeliveryService struct {
|
|
db *gorm.DB
|
|
cfg config.Config
|
|
awx *AWXClient
|
|
common *DeliveryService
|
|
}
|
|
|
|
func NewPostgreSQLDeliveryService(cfg config.Config, db *gorm.DB, common *DeliveryService) *PostgreSQLDeliveryService {
|
|
return &PostgreSQLDeliveryService{db: db, cfg: cfg, awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword), common: common}
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin bool, idempotencyKey string, input PostgreSQLDeliveryInput) (*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 := validatePostgreSQLDeliveryInput(input, s.cfg.DeliveryDataDisks); err != nil {
|
|
return nil, false, err
|
|
}
|
|
if input.VersionMajor == "" {
|
|
input.VersionMajor = input.PostgreSQLVersion
|
|
}
|
|
var existing model.DeliveryTask
|
|
if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil {
|
|
if existing.RequestedBy != userID || existing.ServiceType != postgresqlServiceType {
|
|
return nil, false, fmt.Errorf("idempotency key is already in use")
|
|
}
|
|
return &existing, true, nil
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, false, err
|
|
}
|
|
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.Topology == "" {
|
|
input.Topology = "standalone"
|
|
}
|
|
if input.DataRoot == "" {
|
|
input.DataRoot = "/data/postgresql"
|
|
}
|
|
target, err := getPostgreSQLTarget(ctx, s.awx, input.TargetID)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
payload := postgresqlDeliveryPayload{PostgreSQLDeliveryInput: 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, Component: postgresqlServiceType, TargetType: target.TargetType, ServiceType: postgresqlServiceType, TargetID: input.TargetID, Namespace: input.Namespace, InstanceName: input.ClusterName, 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: "postgresql delivery task created"}).Error
|
|
return &task, false, nil
|
|
}
|
|
|
|
func getPostgreSQLTarget(ctx context.Context, awx *AWXClient, templateID uint64) (DeliveryTarget, error) {
|
|
template, err := awx.GetJobTemplate(ctx, templateID)
|
|
if err != nil {
|
|
return DeliveryTarget{}, fmt.Errorf("deployment target is unavailable: %w", err)
|
|
}
|
|
templateText := strings.ToLower(template.Name + " " + template.Description)
|
|
if !strings.Contains(templateText, "postgresql") || strings.Contains(templateText, "rollback") {
|
|
return DeliveryTarget{}, fmt.Errorf("AWX job template %d is not a PostgreSQL target", templateID)
|
|
}
|
|
hosts, err := awx.ListInventoryHosts(ctx, template.Inventory)
|
|
if err != nil {
|
|
return DeliveryTarget{}, err
|
|
}
|
|
meta := targetMetadata{Topology: "standalone"}
|
|
for _, host := range hosts {
|
|
if host.Enabled {
|
|
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: "host_pool", AWXInventoryID: template.Inventory, AWXTemplateID: template.ID, Enabled: true, Metadata: string(raw)}, nil
|
|
}
|
|
|
|
func selectPostgreSQLHosts(hosts []targetHost, requested []string, count int) ([]targetHost, error) {
|
|
if len(requested) == 0 {
|
|
if len(hosts) < count {
|
|
return nil, fmt.Errorf("deployment target has %d hosts, but %d PostgreSQL instances are required", len(hosts), count)
|
|
}
|
|
return append([]targetHost(nil), hosts[:count]...), nil
|
|
}
|
|
byName := map[string]targetHost{}
|
|
for _, host := range hosts {
|
|
byName[host.Name] = host
|
|
}
|
|
selected := make([]targetHost, 0, len(requested))
|
|
for _, name := range requested {
|
|
host, ok := byName[name]
|
|
if !ok {
|
|
return nil, fmt.Errorf("target host %q is not in the candidate host pool", name)
|
|
}
|
|
selected = append(selected, host)
|
|
}
|
|
return selected, nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryTask, error) {
|
|
var task model.DeliveryTask
|
|
var payload postgresqlDeliveryPayload
|
|
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).Where("status = ? AND service_type = ?", model.TaskPending, postgresqlServiceType).Order("created_at ASC").First(&task).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "stored PostgreSQL deployment payload is invalid", "stored PostgreSQL deployment payload is invalid")
|
|
}
|
|
activeStates := []string{model.TaskValidating, model.TaskDispatching, model.TaskRunning, model.TaskCanceling}
|
|
checks := []struct {
|
|
query string
|
|
args []any
|
|
limit int
|
|
}{
|
|
{"status IN ?", []any{activeStates}, s.cfg.DeliveryGlobalLimit},
|
|
{"status IN ? AND target_id = ?", []any{activeStates, task.TargetID}, s.cfg.DeliveryTargetLimit},
|
|
{"status IN ? AND business_line_id = ?", []any{activeStates, task.BusinessLineID}, s.cfg.DeliveryBusinessLimit},
|
|
}
|
|
for _, check := range checks {
|
|
if check.limit <= 0 {
|
|
continue
|
|
}
|
|
var count int64
|
|
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: delivery concurrency limit reached")
|
|
}
|
|
}
|
|
target, err := getPostgreSQLTarget(ctx, s.awx, task.TargetID)
|
|
if err != nil {
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
|
|
}
|
|
meta := parseTargetMetadata(target.Metadata)
|
|
nodes := 1
|
|
if payload.Topology == "primary_replica" {
|
|
nodes += payload.ReplicaCount
|
|
}
|
|
selected, err := selectPostgreSQLHosts(meta.Hosts, payload.TargetHosts, nodes)
|
|
if err != nil {
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
|
|
}
|
|
quotaOK, err := checkPostgreSQLResourceQuota(tx, task.BusinessLineID, task.TargetID, payload, int64(nodes))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !quotaOK {
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "resource quota is insufficient", "resource quota is insufficient")
|
|
}
|
|
var existingCluster model.PostgreSQLCluster
|
|
if err := tx.Where("name = ?", payload.ClusterName).First(&existingCluster).Error; err == nil {
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, "PostgreSQL cluster name is already in use", "PostgreSQL cluster name is already in use")
|
|
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return err
|
|
}
|
|
cluster := model.PostgreSQLCluster{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, Name: payload.ClusterName, VersionMajor: payload.VersionMajor, Topology: payload.Topology, ReplicationMode: "async", FailoverMode: "manual", Status: "provisioning", BackupStatus: "not_configured", MonitoringStatus: "not_configured"}
|
|
if err := tx.Create(&cluster).Error; err != nil {
|
|
return err
|
|
}
|
|
primaryPort := 0
|
|
for index, host := range selected {
|
|
var usedPorts []int
|
|
if err := tx.Model(&model.PostgreSQLInstance{}).Where("hostname = ? AND status IN ?", host.Name, []string{"provisioning", "active", "quarantined"}).Pluck("port", &usedPorts).Error; err != nil {
|
|
return err
|
|
}
|
|
port, err := allocateReachablePostgreSQLPort(ctx, host.IP, usedPorts, postgresqlPortInUse)
|
|
if err != nil {
|
|
if cleanupErr := tx.Where("task_id = ?", task.ID).Delete(&model.PostgreSQLInstance{}).Error; cleanupErr != nil {
|
|
return cleanupErr
|
|
}
|
|
if cleanupErr := tx.Delete(&cluster).Error; cleanupErr != nil {
|
|
return cleanupErr
|
|
}
|
|
return s.common.transitionTx(tx, &task, model.TaskValidationFailed, err.Error(), err.Error())
|
|
}
|
|
role := "replica"
|
|
instanceID := fmt.Sprintf("%s-replica-%d", payload.ClusterName, index)
|
|
upstream := payload.ClusterName + "-primary"
|
|
slot := fmt.Sprintf("xinfra_%s_replica_%d", strings.ReplaceAll(payload.ClusterName, "-", "_"), index)
|
|
if index == 0 {
|
|
role = "primary"
|
|
instanceID = payload.ClusterName + "-primary"
|
|
upstream = ""
|
|
slot = ""
|
|
primaryPort = port
|
|
}
|
|
if payload.Topology == "standalone" {
|
|
role = "standalone"
|
|
instanceID = payload.ClusterName
|
|
upstream = ""
|
|
slot = ""
|
|
}
|
|
root := strings.TrimRight(payload.DataRoot, "/") + "/" + instanceID
|
|
instance := model.PostgreSQLInstance{TaskID: task.ID, ClusterID: cluster.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, InstanceID: instanceID, Hostname: host.Name, HostIP: host.IP, Port: port, DataDir: root + "/data", ConfigDir: root + "/conf", LogDir: root + "/log", SystemdUnit: "postgresql-xinfra@" + instanceID + ".service", VersionMajor: payload.VersionMajor, Role: role, HAComponentRole: "database", UpstreamInstanceID: upstream, ReplicationSlotName: slot, Status: "provisioning", BackupStatus: "not_configured", MonitoringStatus: "not_configured"}
|
|
if err := tx.Create(&instance).Error; err != nil {
|
|
return err
|
|
}
|
|
if index == 0 {
|
|
cluster.PrimaryInstanceID = instance.ID
|
|
}
|
|
}
|
|
if err := tx.Save(&cluster).Error; err != nil {
|
|
return err
|
|
}
|
|
reservation := model.ResourceReservation{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli * int64(nodes), MemoryMi: payload.MemoryMi * int64(nodes), StorageGi: payload.StorageGi * int64(nodes), InstanceCount: int64(nodes), Status: "reserved", ExpiresAt: time.Now().Add(time.Duration(s.cfg.ReservationTTLMinutes) * time.Minute)}
|
|
if err := tx.Create(&reservation).Error; err != nil {
|
|
return err
|
|
}
|
|
hostNames := make([]string, 0, len(selected))
|
|
for _, host := range selected {
|
|
hostNames = append(hostNames, host.Name)
|
|
}
|
|
task.TargetHost = strings.Join(hostNames, ",")
|
|
task.TargetHostIP = selected[0].IP
|
|
task.PostgreSQLPort = primaryPort
|
|
if err := tx.Model(&model.DeliveryTask{}).Where("id = ?", task.ID).Updates(map[string]any{"target_host": task.TargetHost, "target_host_ip": task.TargetHostIP, "postgresql_port": primaryPort}).Error; err != nil {
|
|
return err
|
|
}
|
|
return s.common.transitionTx(tx, &task, model.TaskDispatching, "PostgreSQL resources, ports and directories reserved", "")
|
|
})
|
|
if err == nil && task.Status != model.TaskDispatching {
|
|
return nil, errPostgreSQLTaskHandled
|
|
}
|
|
return &task, err
|
|
}
|
|
|
|
func checkPostgreSQLResourceQuota(tx *gorm.DB, businessLineID, targetID uint64, payload postgresqlDeliveryPayload, nodes int64) (bool, error) {
|
|
var quota model.ResourceQuota
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("business_line_id = ? AND target_id = ?", businessLineID, targetID).First("a).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return true, nil
|
|
}
|
|
return false, err
|
|
}
|
|
type totals struct{ CPU, Memory, Storage, Instances int64 }
|
|
var mysqlUsed, postgresUsed, 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(&mysqlUsed).Error; err != nil {
|
|
return false, err
|
|
}
|
|
if err := tx.Model(&model.PostgreSQLResourceUsage{}).Select("COALESCE(SUM(cpu_milli),0) cpu, COALESCE(SUM(memory_mi),0) memory, COALESCE(SUM(storage_gi),0) storage, COUNT(*) instances").Where("business_line_id = ? AND target_id = ? AND status = ?", businessLineID, targetID, "active").Scan(&postgresUsed).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
|
|
}
|
|
requestedCPU := payload.CPUMilli * nodes
|
|
requestedMemory := payload.MemoryMi * nodes
|
|
requestedStorage := payload.StorageGi * nodes
|
|
return mysqlUsed.CPU+postgresUsed.CPU+reserved.CPU+requestedCPU <= quota.CPUMilli &&
|
|
mysqlUsed.Memory+postgresUsed.Memory+reserved.Memory+requestedMemory <= quota.MemoryMi &&
|
|
mysqlUsed.Storage+postgresUsed.Storage+reserved.Storage+requestedStorage <= quota.StorageGi &&
|
|
mysqlUsed.Instances+postgresUsed.Instances+reserved.Instances+nodes <= quota.InstanceLimit, nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) CreateExecution(ctx context.Context, task *model.DeliveryTask) error {
|
|
var payload postgresqlDeliveryPayload
|
|
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
|
|
return err
|
|
}
|
|
var instances []model.PostgreSQLInstance
|
|
if err := s.db.WithContext(ctx).Where("task_id = ?", task.ID).Order("id ASC").Find(&instances).Error; err != nil {
|
|
return err
|
|
}
|
|
instanceVars := make(map[string]any, len(instances))
|
|
for _, instance := range instances {
|
|
instanceVars[instance.Hostname] = map[string]any{"instance_id": instance.InstanceID, "port": instance.Port, "data_dir": instance.DataDir, "config_dir": instance.ConfigDir, "log_dir": instance.LogDir, "role": instance.Role, "replication_slot": instance.ReplicationSlotName}
|
|
}
|
|
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 err
|
|
}
|
|
target, err := getPostgreSQLTarget(ctx, s.awx, task.TargetID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
extraVars := map[string]any{"task_id": task.ID, "payload_hash": task.PayloadHash, "target_hosts": task.TargetHost, "cluster_name": payload.ClusterName, "postgresql_version": payload.VersionMajor, "postgresql_port": task.PostgreSQLPort, "topology": payload.Topology, "replica_count": payload.ReplicaCount, "postgresql_instances": instanceVars, "memory_mb": payload.MemoryMi, "storage_gb": payload.StorageGi}
|
|
if payload.MaxConnections > 0 {
|
|
extraVars["max_connections"] = payload.MaxConnections
|
|
}
|
|
job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: extraVars})
|
|
if err != nil {
|
|
_ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()})
|
|
return err
|
|
}
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
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.common.transitionTx(tx, task, model.TaskRunning, "PostgreSQL AWX job started", "")
|
|
})
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) DispatchOnce(ctx context.Context) error {
|
|
task, err := s.claimAndReserve(ctx)
|
|
if err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) || errors.Is(err, errPostgreSQLTaskHandled) || strings.HasPrefix(err.Error(), "defer:") {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
if err := s.CreateExecution(ctx, task); err != nil {
|
|
return s.beginRollback(ctx, task.ID, "PostgreSQL deployment could not be started: "+err.Error())
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) PollOnce(ctx context.Context) error {
|
|
var jobs []model.ExecutionJob
|
|
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 = ?", "running", postgresqlServiceType).Find(&jobs).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, execution := range jobs {
|
|
job, err := s.awx.GetJob(ctx, execution.ExecutorJobID)
|
|
if err != nil {
|
|
// A temporary AWX failure does not mean the deployment failed. Retry on
|
|
// the next scheduler tick while the execution remains running.
|
|
continue
|
|
}
|
|
switch strings.ToLower(job.Status) {
|
|
case "pending", "waiting", "running", "new":
|
|
continue
|
|
case "canceled":
|
|
_ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job was canceled")
|
|
case "successful":
|
|
if err := s.complete(ctx, execution.TaskID); err != nil {
|
|
_ = s.beginRollback(ctx, execution.TaskID, err.Error())
|
|
}
|
|
default:
|
|
_ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job finished with status "+job.Status)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PostgreSQLDeliveryService) complete(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
|
|
}
|
|
var payload postgresqlDeliveryPayload
|
|
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); 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 planned instances")
|
|
}
|
|
for _, instance := range instances {
|
|
if err := postgresReady(ctx, net.JoinHostPort(instance.HostIP, fmt.Sprint(instance.Port))); err != nil {
|
|
return fmt.Errorf("postgresql health check failed for %s: %w", instance.InstanceID, err)
|
|
}
|
|
}
|
|
now := time.Now()
|
|
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if err := s.common.transitionTx(tx, &task, model.TaskRegistering, "PostgreSQL health checks passed", ""); err != nil {
|
|
return err
|
|
}
|
|
for _, instance := range instances {
|
|
if err := tx.Model(&instance).Updates(map[string]any{"status": "active", "version_full": instance.VersionMajor}).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Create(&model.PostgreSQLResourceUsage{TaskID: task.ID, InstanceID: instance.ID, ClusterID: instance.ClusterID, BusinessLineID: instance.BusinessLineID, TargetID: instance.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, Port: instance.Port, DataDir: instance.DataDir, 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.PostgreSQLCluster{}).Where("task_id = ?", task.ID).Update("status", "active").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 {
|
|
message := fmt.Sprintf("PostgreSQL delivered; CloudDM registration failed: %v", err)
|
|
if transitionErr := s.common.transition(ctx, &task, model.TaskRegisterFailed, "PostgreSQL delivered; CloudDM registration failed and can be retried", message); transitionErr != nil {
|
|
return fmt.Errorf("%s; cannot record register_failed: %w", message, transitionErr)
|
|
}
|
|
return nil
|
|
}
|
|
message := "PostgreSQL delivery completed; CloudDM registration was skipped because no PostgreSQL endpoint is configured"
|
|
if strings.TrimSpace(s.cfg.CloudDMPostgreSQLRegisterURL) != "" {
|
|
message = "PostgreSQL delivery completed, registered and recorded in the resource ledger"
|
|
}
|
|
return s.common.transition(ctx, &task, model.TaskFinished, message, "")
|
|
}
|
|
|
|
func postgresReady(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 *PostgreSQLDeliveryService) 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)
|
|
_ = s.PollRollbackOnce(ctx)
|
|
}
|
|
}
|
|
}
|