diff --git a/ansible/postgresql-rollback.yml b/ansible/postgresql-rollback.yml new file mode 100644 index 0000000..56dce7d --- /dev/null +++ b/ansible/postgresql-rollback.yml @@ -0,0 +1,86 @@ +--- +- name: Roll back native PostgreSQL delivery instances + hosts: "{{ target_hosts }}" + become: true + gather_facts: false + any_errors_fatal: true + vars: + postgresql_instance_plan: "{{ postgresql_instances[inventory_hostname] | default({}) }}" + postgresql_instance_id: "{{ postgresql_instance_plan.instance_id | default('') }}" + postgresql_data_root: /data/postgresql + postgresql_instance_root: "{{ postgresql_data_root }}/{{ postgresql_instance_id }}" + postgresql_data_dir: "{{ postgresql_instance_plan.data_dir | default('') }}" + postgresql_config_dir: "{{ postgresql_instance_plan.config_dir | default('') }}" + postgresql_log_dir: "{{ postgresql_instance_plan.log_dir | default('') }}" + postgresql_systemd_unit: "{{ postgresql_instance_plan.systemd_unit | default('') }}" + postgresql_dropin_dir: "/etc/systemd/system/{{ postgresql_systemd_unit }}.d" + postgresql_run_dir: "/run/postgresql-xinfra/{{ postgresql_instance_id }}" + + pre_tasks: + - name: Validate PostgreSQL rollback target + ansible.builtin.assert: + that: + - postgresql_instance_plan is mapping + - postgresql_instance_id is match('^[a-z0-9][a-z0-9-]{0,62}$') + - postgresql_instance_root == '/data/postgresql/' + postgresql_instance_id + - postgresql_data_dir == postgresql_instance_root + '/data' + - postgresql_config_dir == postgresql_instance_root + '/conf' + - postgresql_log_dir == postgresql_instance_root + '/log' + - postgresql_systemd_unit == 'postgresql-xinfra@' + postgresql_instance_id + '.service' + - postgresql_instance_root != '/data/postgresql' + - "'..' not in postgresql_instance_root" + fail_msg: "Rollback target is outside the native PostgreSQL delivery layout" + quiet: true + + tasks: + - name: Stop the PostgreSQL delivery instance when present + ansible.builtin.systemd_service: + name: "{{ postgresql_systemd_unit }}" + state: stopped + enabled: false + failed_when: false + + - name: Reset the PostgreSQL instance failure state + ansible.builtin.command: + argv: + - systemctl + - reset-failed + - "{{ postgresql_systemd_unit }}" + changed_when: false + failed_when: false + + - name: Remove the PostgreSQL instance systemd drop-in + ansible.builtin.file: + path: "{{ postgresql_dropin_dir }}" + state: absent + + - name: Remove the PostgreSQL instance data, configuration, and logs + ansible.builtin.file: + path: "{{ postgresql_instance_root }}" + state: absent + + - name: Remove the PostgreSQL instance runtime directory + ansible.builtin.file: + path: "{{ postgresql_run_dir }}" + state: absent + + - name: Reload systemd after removing the instance drop-in + ansible.builtin.systemd_service: + daemon_reload: true + + - name: Confirm PostgreSQL instance artifacts are gone + ansible.builtin.stat: + path: "{{ item }}" + loop: + - "{{ postgresql_dropin_dir }}" + - "{{ postgresql_instance_root }}" + - "{{ postgresql_run_dir }}" + register: postgresql_rollback_artifacts + + - name: Assert all PostgreSQL instance artifacts are gone + ansible.builtin.assert: + that: + - not item.stat.exists + fail_msg: "PostgreSQL rollback left an instance artifact on the target host" + quiet: true + loop: "{{ postgresql_rollback_artifacts.results }}" diff --git a/config/local/README.md b/config/local/README.md index 0ae7807..b8ee922 100644 --- a/config/local/README.md +++ b/config/local/README.md @@ -17,6 +17,8 @@ The local object mapping is: | Secret credential | `XINFRA PostgreSQL Runtime Secrets` | | Job Template | `XINFRA PostgreSQL Delivery` | | Playbook | `postgresql-deploy.yml` | +| Rollback Template | `XINFRA PostgreSQL Rollback` | +| Rollback Playbook | `postgresql-rollback.yml` | | Instance Group | `xinfra-e2e-execution` | Run the idempotent setup whenever the AWX task pod restarts or the local @@ -25,6 +27,13 @@ cannot use a Manual Project, so the Project is Git-backed with launch updates disabled. The script refreshes a local Git source inside the AWX task pod and runs a project update so AWX rebuilds its playbook index. +The setup output includes `DELIVERY_POSTGRESQL_ROLLBACK_TEMPLATE_ID`; copy that +value into `config/local/server-postgresql.env` before starting the scheduler. +PostgreSQL CloudDM registration is optional and requires the dedicated URL, +API token, username, and password documented in the server env example. The +CloudDM password must match the PostgreSQL administrator password configured +in the AWX runtime credential. + ```bash ./scripts/setup-postgresql-awx.sh ``` diff --git a/config/local/postgresql-awx.env.example b/config/local/postgresql-awx.env.example index 84cee5c..e7a48e2 100644 --- a/config/local/postgresql-awx.env.example +++ b/config/local/postgresql-awx.env.example @@ -40,4 +40,7 @@ XINFRA_POSTGRES_REPLICATION_PASSWORD= AWX_JOB_TEMPLATE_NAME="XINFRA PostgreSQL Delivery" AWX_JOB_TEMPLATE_DESCRIPTION="PostgreSQL host_pool delivery template for XINFRA" AWX_PLAYBOOK=postgresql-deploy.yml +AWX_ROLLBACK_JOB_TEMPLATE_NAME="XINFRA PostgreSQL Rollback" +AWX_ROLLBACK_JOB_TEMPLATE_DESCRIPTION="PostgreSQL compensating cleanup template for XINFRA" +AWX_ROLLBACK_PLAYBOOK=postgresql-rollback.yml AWX_INSTANCE_GROUP=xinfra-e2e-execution diff --git a/config/local/server-postgresql.env.example b/config/local/server-postgresql.env.example index 8224347..d2de363 100644 --- a/config/local/server-postgresql.env.example +++ b/config/local/server-postgresql.env.example @@ -14,3 +14,11 @@ DELIVERY_GLOBAL_LIMIT=2 DELIVERY_TARGET_LIMIT=2 DELIVERY_BUSINESS_LIMIT=1 DELIVERY_DATA_DISKS=/data +DELIVERY_POSTGRESQL_ROLLBACK_TEMPLATE_ID=0 + +# Optional PostgreSQL CloudDM registration. Keep the URL empty to disable it. +# The password must match the postgres administrator password delivered by AWX. +CLOUDDM_POSTGRESQL_REGISTER_URL= +CLOUDDM_API_TOKEN= +CLOUDDM_POSTGRESQL_USERNAME=postgres +CLOUDDM_POSTGRESQL_PASSWORD= diff --git a/frontend/src/views/service/Catalog.vue b/frontend/src/views/service/Catalog.vue index fe065b5..1b84470 100644 --- a/frontend/src/views/service/Catalog.vue +++ b/frontend/src/views/service/Catalog.vue @@ -732,7 +732,7 @@ const basicServices = ref([ charsets: ['UTF8'], defaultPort: 0, defaultPaths: { install: '/usr/lib/postgresql', data: '/data/postgresql', log: '/data/postgresql' }, - registerTo: 'PostgreSQL 专用资源台账', + registerTo: 'PostgreSQL 资源台账 / CloudDM(可选)', configId: 'PG-NATIVE-V1', assetId: '', healthText: 'PostgreSQL 就绪 · TCP 探测通过', @@ -835,7 +835,7 @@ const rollbackReleasing = ref(false) const cloudDMRetrying = ref(false) const rollbackActionBusy = computed(() => rollbackRetrying.value || rollbackReleasing.value) const canManageRollback = computed(() => authStore.isAdmin && lastDeliveryStatus.value === 'rollback_failed' && Boolean(deploymentId.value)) -const canRetryCloudDM = computed(() => !isPostgreSQL.value && deliveryRegisterFailed.value && Boolean(deploymentId.value)) +const canRetryCloudDM = computed(() => deliveryRegisterFailed.value && Boolean(deploymentId.value)) const deliveryForm = reactive({ instanceName: generateInstanceName(activeServiceKey.value, currentName.value), @@ -1318,7 +1318,7 @@ function defaultSteps(): DeliveryStep[] { ['应用实例配置', 'postgresql.conf、pg_hba.conf、systemd'], ['配置复制拓扑', '复制账号、slot、primary_conninfo'], ['数据库健康检查', '端口、角色、版本和数据目录'], - ['资源入账与交付归档', 'PostgreSQL 集群、实例与资源台账'], + ['资源入账与交付归档', 'PostgreSQL 资源台账与可选 CloudDM 注册'], ].map(([name, desc]) => ({ name, desc, state: 'pending' as StepState })) } const isMysql = activeServiceKey.value === 'mysql' diff --git a/scripts/setup-postgresql-awx.sh b/scripts/setup-postgresql-awx.sh index 076e1f1..dbb0d70 100755 --- a/scripts/setup-postgresql-awx.sh +++ b/scripts/setup-postgresql-awx.sh @@ -37,7 +37,8 @@ for variable_name in \ AWX_HOST_NAME AWX_HOST_ADDRESS \ AWX_MACHINE_CREDENTIAL_NAME AWX_SSH_PRIVATE_KEY_FILE \ AWX_POSTGRES_CREDENTIAL_TYPE_NAME AWX_POSTGRES_CREDENTIAL_NAME \ - AWX_JOB_TEMPLATE_NAME AWX_PLAYBOOK; do + AWX_JOB_TEMPLATE_NAME AWX_PLAYBOOK \ + AWX_ROLLBACK_JOB_TEMPLATE_NAME AWX_ROLLBACK_PLAYBOOK; do require_value "${variable_name}" done @@ -64,6 +65,7 @@ case ${PROJECT_SOURCE}/ in *) die "AWX_PROJECT_SOURCE_DIR must resolve inside the repository" ;; esac [[ -f ${PROJECT_SOURCE}/${AWX_PLAYBOOK} ]] || die "playbook not found: ${PROJECT_SOURCE}/${AWX_PLAYBOOK}" +[[ -f ${PROJECT_SOURCE}/${AWX_ROLLBACK_PLAYBOOK} ]] || die "rollback playbook not found: ${PROJECT_SOURCE}/${AWX_ROLLBACK_PLAYBOOK}" if [[ -z ${AWX_TOKEN:-} && -z ${AWX_PASSWORD:-} ]]; then AWX_PASSWORD=$(kubectl get secret \ @@ -186,6 +188,10 @@ if ! awx_request GET "/api/v2/projects/${project_id}/playbooks/" | \ jq -e --arg playbook "${AWX_PLAYBOOK}" 'index($playbook) != null' >/dev/null; then die "AWX project ${project_id} does not expose playbook ${AWX_PLAYBOOK}" fi +if ! awx_request GET "/api/v2/projects/${project_id}/playbooks/" | \ + jq -e --arg playbook "${AWX_ROLLBACK_PLAYBOOK}" 'index($playbook) != null' >/dev/null; then + die "AWX project ${project_id} does not expose rollback playbook ${AWX_ROLLBACK_PLAYBOOK}" +fi inventory_payload=$(jq -nc \ --arg name "${AWX_INVENTORY_NAME}" \ @@ -307,29 +313,57 @@ job_template_payload=$(jq -nc \ job_template_id=$(upsert_named_object '/api/v2/job_templates/' \ "${AWX_JOB_TEMPLATE_NAME}" "${job_template_payload}") -while IFS= read -r existing_group_id; do - [[ -z ${existing_group_id} || ${existing_group_id} == "${instance_group_id}" ]] && continue - awx_request POST "/api/v2/job_templates/${job_template_id}/instance_groups/" \ - "$(jq -nc --argjson id "${existing_group_id}" '{id:$id,disassociate:true}')" >/dev/null -done < <(awx_request GET "/api/v2/job_templates/${job_template_id}/instance_groups/" | jq -r '.results[].id') +rollback_template_payload=$(jq -nc \ + --arg name "${AWX_ROLLBACK_JOB_TEMPLATE_NAME}" \ + --arg description "${AWX_ROLLBACK_JOB_TEMPLATE_DESCRIPTION:-PostgreSQL rollback template}" \ + --arg playbook "${AWX_ROLLBACK_PLAYBOOK}" \ + --argjson organization "${organization_id}" \ + --argjson inventory "${inventory_id}" \ + --argjson project "${project_id}" \ + --argjson prevent_fallback "${prevent_fallback}" \ + '{ + name:$name, + description:$description, + organization:$organization, + inventory:$inventory, + project:$project, + playbook:$playbook, + job_type:"run", + ask_inventory_on_launch:true, + ask_variables_on_launch:true, + ask_limit_on_launch:true, + allow_simultaneous:true, + prevent_instance_group_fallback:$prevent_fallback + }') +rollback_template_id=$(upsert_named_object '/api/v2/job_templates/' \ + "${AWX_ROLLBACK_JOB_TEMPLATE_NAME}" "${rollback_template_payload}") + +for configured_template_id in "${job_template_id}" "${rollback_template_id}"; do + while IFS= read -r existing_group_id; do + [[ -z ${existing_group_id} || ${existing_group_id} == "${instance_group_id}" ]] && continue + awx_request POST "/api/v2/job_templates/${configured_template_id}/instance_groups/" \ + "$(jq -nc --argjson id "${existing_group_id}" '{id:$id,disassociate:true}')" >/dev/null + done < <(awx_request GET "/api/v2/job_templates/${configured_template_id}/instance_groups/" | jq -r '.results[].id') + + if ! awx_request GET "/api/v2/job_templates/${configured_template_id}/credentials/" | \ + jq -e --argjson id "${machine_credential_id}" '.results | any(.id == $id)' >/dev/null; then + awx_request POST "/api/v2/job_templates/${configured_template_id}/credentials/" \ + "$(jq -nc --argjson id "${machine_credential_id}" '{id:$id}')" >/dev/null + fi + if [[ -n ${instance_group_id} ]]; then + if ! awx_request GET "/api/v2/job_templates/${configured_template_id}/instance_groups/" | \ + jq -e --argjson id "${instance_group_id}" '.results | any(.id == $id)' >/dev/null; then + awx_request POST "/api/v2/job_templates/${configured_template_id}/instance_groups/" \ + "$(jq -nc --argjson id "${instance_group_id}" '{id:$id}')" >/dev/null + fi + fi +done -if ! awx_request GET "/api/v2/job_templates/${job_template_id}/credentials/" | \ - jq -e --argjson id "${machine_credential_id}" '.results | any(.id == $id)' >/dev/null; then - awx_request POST "/api/v2/job_templates/${job_template_id}/credentials/" \ - "$(jq -nc --argjson id "${machine_credential_id}" '{id:$id}')" >/dev/null -fi if ! awx_request GET "/api/v2/job_templates/${job_template_id}/credentials/" | \ jq -e --argjson id "${postgres_credential_id}" '.results | any(.id == $id)' >/dev/null; then awx_request POST "/api/v2/job_templates/${job_template_id}/credentials/" \ "$(jq -nc --argjson id "${postgres_credential_id}" '{id:$id}')" >/dev/null fi -if [[ -n ${instance_group_id} ]]; then - if ! awx_request GET "/api/v2/job_templates/${job_template_id}/instance_groups/" | \ - jq -e --argjson id "${instance_group_id}" '.results | any(.id == $id)' >/dev/null; then - awx_request POST "/api/v2/job_templates/${job_template_id}/instance_groups/" \ - "$(jq -nc --argjson id "${instance_group_id}" '{id:$id}')" >/dev/null - fi -fi printf '\nPostgreSQL AWX configuration is ready:\n' printf ' Project: %s (id=%s)\n' "${AWX_PROJECT_NAME}" "${project_id}" @@ -338,6 +372,8 @@ printf ' Host: %s -> %s (id=%s)\n' "${AWX_HOST_NAME}" "${AWX_HOST_ADD printf ' SSH credential: %s (id=%s)\n' "${AWX_MACHINE_CREDENTIAL_NAME}" "${machine_credential_id}" printf ' PG credential: %s (id=%s)\n' "${AWX_POSTGRES_CREDENTIAL_NAME}" "${postgres_credential_id}" printf ' Job Template: %s (id=%s)\n' "${AWX_JOB_TEMPLATE_NAME}" "${job_template_id}" +printf ' Rollback: %s (id=%s)\n' "${AWX_ROLLBACK_JOB_TEMPLATE_NAME}" "${rollback_template_id}" +printf ' Server env: DELIVERY_POSTGRESQL_ROLLBACK_TEMPLATE_ID=%s\n' "${rollback_template_id}" if [[ -n ${instance_group_id} ]]; then printf ' Instance Group: %s (id=%s)\n' "${AWX_INSTANCE_GROUP}" "${instance_group_id}" fi diff --git a/server/.env.example b/server/.env.example index 18a080e..fb8a290 100644 --- a/server/.env.example +++ b/server/.env.example @@ -68,16 +68,23 @@ DELIVERY_MYSQL_INSPECT_TEMPLATE_NAME=XINFRA MySQL Inspect DELIVERY_MYSQL_INSPECT_TIMEOUT_SECONDS=90 # AWX Job Template ID for ansible/mysql-rollback.yml; required for automatic cleanup DELIVERY_ROLLBACK_TEMPLATE_ID=0 +# AWX Job Template ID for ansible/postgresql-rollback.yml +DELIVERY_POSTGRESQL_ROLLBACK_TEMPLATE_ID=0 DELIVERY_SERVICE_TOKEN= -# PostgreSQL runtime secrets are injected by an AWX Credential and are never -# persisted in xinfra task payloads: +# PostgreSQL deployment secrets are injected by an AWX Credential and are never +# persisted in xinfra task payloads. CloudDM registration, when enabled, uses +# the separate CLOUDDM_POSTGRESQL_* credential below. # XINFRA_POSTGRES_ADMIN_PASSWORD # XINFRA_POSTGRES_REPLICATION_PASSWORD # XINFRA_POSTGRES_REPLICATION_USER (optional, defaults to xinfra_replication) CLOUDDM_REGISTER_URL= CLOUDDM_DELETE_URL= CLOUDDM_API_TOKEN= +# PostgreSQL uses an explicit contract and separate endpoint/credential. +CLOUDDM_POSTGRESQL_REGISTER_URL= +CLOUDDM_POSTGRESQL_USERNAME=postgres +CLOUDDM_POSTGRESQL_PASSWORD= JWT_SECRET=change-this-secret JWT_ISSUER=authserver diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 51c021b..f9c4893 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -14,82 +14,86 @@ type OAuthClient struct { } type Config struct { - AppEnv string - HTTPAddr string - PublicBaseURL string - MySQLDSN string - AutoMigrate bool - SSOEnabled bool - JWTSecret string - JWTIssuer string - JWTTTLMinutes int - SAMLEntityID string - SAMLACSURL string - SAMLSPCert string - SAMLSPKey string - SAMLIDPMetaURL string - SAMLLogoutURL string - WayenLoginURL string - WayenTargetURL string - WayenUsernameKey string - WayenPasswordKey string - WayenLoginFormat string - WayenLoginValue string - WayenOAuthRef string - WayenOAuthLoginURL string - WayneAPIBaseURL string - WayneAdminUsername string - WayneAdminPassword string - WayneTokenTTLMinutes int - WayneInternalAPIBaseURL string - WayneServiceName string - WayneServiceAPISecretKey string - OAuthClientID string - OAuthClientSecret string - OAuthRedirectURI string - OAuthCodeTTLSeconds int - OIDCIssuer string - OIDCAuthorizeURL string - OIDCTokenURL string - OIDCUserInfoURL string - OIDCJWKSURL string - CloudDMClientID string - CloudDMClientSecret string - CloudDMRedirectURI string - CloudDMTargetURL string - CloudDMPublicURL string - CloudDMLoginURL string - CloudDMAdminUsername string - CloudDMAdminPassword string - CloudDMRegisterURL string - CloudDMDeleteURL string - CloudDMAPIToken string - AWXBaseURL string - AWXToken string - AWXUsername string - AWXPassword string - AWXWebhookToken string - AWXFactsTemplateID uint64 - AWXFactsTimeoutSeconds int - MySQLDeliveryTemplateName string - MySQLInspectTemplateName string - MySQLInspectTimeoutSeconds int - RollbackTemplateID uint64 - DeliveryServiceToken string - DeliverySchedulerEnabled bool - DeliveryDispatchSeconds int - DeliveryPollSeconds int - DeliveryCallbackBaseURL string - DeliveryCredentialSecret string - ReservationTTLMinutes int - DeliveryGlobalLimit int - DeliveryTargetLimit int - DeliveryHostInstanceLimit int - DeliveryBusinessLimit int - DeliveryDataDisks []string - SINABaseURL string - SINAUsername string - SINAPassword string + AppEnv string + HTTPAddr string + PublicBaseURL string + MySQLDSN string + AutoMigrate bool + SSOEnabled bool + JWTSecret string + JWTIssuer string + JWTTTLMinutes int + SAMLEntityID string + SAMLACSURL string + SAMLSPCert string + SAMLSPKey string + SAMLIDPMetaURL string + SAMLLogoutURL string + WayenLoginURL string + WayenTargetURL string + WayenUsernameKey string + WayenPasswordKey string + WayenLoginFormat string + WayenLoginValue string + WayenOAuthRef string + WayenOAuthLoginURL string + WayneAPIBaseURL string + WayneAdminUsername string + WayneAdminPassword string + WayneTokenTTLMinutes int + WayneInternalAPIBaseURL string + WayneServiceName string + WayneServiceAPISecretKey string + OAuthClientID string + OAuthClientSecret string + OAuthRedirectURI string + OAuthCodeTTLSeconds int + OIDCIssuer string + OIDCAuthorizeURL string + OIDCTokenURL string + OIDCUserInfoURL string + OIDCJWKSURL string + CloudDMClientID string + CloudDMClientSecret string + CloudDMRedirectURI string + CloudDMTargetURL string + CloudDMPublicURL string + CloudDMLoginURL string + CloudDMAdminUsername string + CloudDMAdminPassword string + CloudDMRegisterURL string + CloudDMDeleteURL string + CloudDMAPIToken string + CloudDMPostgreSQLRegisterURL string + CloudDMPostgreSQLUsername string + CloudDMPostgreSQLPassword string + AWXBaseURL string + AWXToken string + AWXUsername string + AWXPassword string + AWXWebhookToken string + AWXFactsTemplateID uint64 + AWXFactsTimeoutSeconds int + MySQLDeliveryTemplateName string + MySQLInspectTemplateName string + MySQLInspectTimeoutSeconds int + RollbackTemplateID uint64 + PostgreSQLRollbackTemplateID uint64 + DeliveryServiceToken string + DeliverySchedulerEnabled bool + DeliveryDispatchSeconds int + DeliveryPollSeconds int + DeliveryCallbackBaseURL string + DeliveryCredentialSecret string + ReservationTTLMinutes int + DeliveryGlobalLimit int + DeliveryTargetLimit int + DeliveryHostInstanceLimit int + DeliveryBusinessLimit int + DeliveryDataDisks []string + SINABaseURL string + SINAUsername string + SINAPassword string } func Load() Config { @@ -102,82 +106,86 @@ func Load() Config { oidcIssuer = strings.TrimRight(oidcIssuer, "/") return Config{ - AppEnv: env("APP_ENV", "dev"), - HTTPAddr: httpAddr, - PublicBaseURL: publicBaseURL, - MySQLDSN: env("MYSQL_DSN", "auth:auth@tcp(127.0.0.1:3306)/authserver?charset=utf8mb4&parseTime=True&loc=Local"), - AutoMigrate: envBool("AUTO_MIGRATE", true), - SSOEnabled: envBool("SSO_ENABLED", true), - JWTSecret: env("JWT_SECRET", "change-this-secret"), - JWTIssuer: env("JWT_ISSUER", "authserver"), - JWTTTLMinutes: envInt("JWT_TTL_MINUTES", 120), - SAMLEntityID: samlEntityID, - SAMLACSURL: samlACSURL, - SAMLSPCert: env("SAML_SP_CERT_FILE", "certs/sp.crt"), - SAMLSPKey: env("SAML_SP_KEY_FILE", "certs/sp.key"), - SAMLIDPMetaURL: env("SAML_IDP_METADATA_URL", "http://sso-internal.dev.qiniu.io/saml2/meta"), - SAMLLogoutURL: trimURL(env("SAML_LOGOUT_URL", "")), - WayenLoginURL: env("WAYEN_LOGIN_URL", ""), - WayenTargetURL: env("WAYEN_TARGET_URL", ""), - WayenUsernameKey: env("WAYEN_USERNAME_KEY", "email"), - WayenPasswordKey: env("WAYEN_PASSWORD_KEY", "password"), - WayenLoginFormat: env("WAYEN_LOGIN_FORMAT", "form"), - WayenLoginValue: env("WAYEN_LOGIN_VALUE", "email"), - WayenOAuthRef: env("WAYEN_OAUTH_REF", "/portal/namespace/1/app"), - WayenOAuthLoginURL: trimURL(env("WAYEN_OAUTH_LOGIN_URL", "")), - WayneAPIBaseURL: trimURL(env("WAYNE_API_BASE_URL", env("WAYNE_INTERNAL_API_BASE_URL", ""))), - WayneAdminUsername: env("WAYNE_ADMIN_USERNAME", ""), - WayneAdminPassword: env("WAYNE_ADMIN_PASSWORD", ""), - WayneTokenTTLMinutes: envInt("WAYNE_TOKEN_TTL_MINUTES", 1440), - WayneInternalAPIBaseURL: trimURL(env("WAYNE_INTERNAL_API_BASE_URL", "")), - WayneServiceName: env("WAYNE_SERVICE_NAME", "xinfra"), - WayneServiceAPISecretKey: env("WAYNE_SERVICE_API_SECRET_KEY", ""), - OAuthClientID: env("OAUTH_WAYNE_CLIENT_ID", "wayne"), - OAuthClientSecret: env("OAUTH_WAYNE_CLIENT_SECRET", "wayne-secret"), - OAuthRedirectURI: env("OAUTH_WAYNE_REDIRECT_URI", ""), - OAuthCodeTTLSeconds: envInt("OAUTH_CODE_TTL_SECONDS", 120), - OIDCIssuer: oidcIssuer, - OIDCAuthorizeURL: trimURL(env("OIDC_AUTHORIZATION_ENDPOINT", oidcIssuer+"/oauth/authorize")), - OIDCTokenURL: trimURL(env("OIDC_TOKEN_ENDPOINT", oidcIssuer+"/oauth/token")), - OIDCUserInfoURL: trimURL(env("OIDC_USERINFO_ENDPOINT", oidcIssuer+"/oauth/userinfo")), - OIDCJWKSURL: trimURL(env("OIDC_JWKS_URI", oidcIssuer+"/oauth/jwks")), - CloudDMClientID: env("OIDC_CLOUDDM_CLIENT_ID", "clouddm"), - CloudDMClientSecret: env("OIDC_CLOUDDM_CLIENT_SECRET", ""), - CloudDMRedirectURI: env("OIDC_CLOUDDM_REDIRECT_URI", ""), - CloudDMTargetURL: env("CLOUDDM_TARGET_URL", ""), - CloudDMPublicURL: trimURL(env("CLOUDDM_PUBLIC_URL", "")), - CloudDMLoginURL: trimURL(env("CLOUDDM_LOGIN_URL", "")), - CloudDMAdminUsername: env("CLOUDDM_ADMIN_USERNAME", ""), - CloudDMAdminPassword: env("CLOUDDM_ADMIN_PASSWORD", ""), - CloudDMRegisterURL: trimURL(env("CLOUDDM_REGISTER_URL", "")), - CloudDMDeleteURL: trimURL(env("CLOUDDM_DELETE_URL", "")), - CloudDMAPIToken: env("CLOUDDM_API_TOKEN", ""), - AWXBaseURL: trimURL(env("AWX_BASE_URL", "")), - AWXToken: env("AWX_TOKEN", ""), - AWXUsername: env("AWX_USERNAME", ""), - AWXPassword: env("AWX_PASSWORD", ""), - AWXWebhookToken: env("AWX_WEBHOOK_TOKEN", ""), - AWXFactsTemplateID: envUint64("AWX_FACTS_TEMPLATE_ID", 0), - AWXFactsTimeoutSeconds: envInt("AWX_FACTS_TIMEOUT_SECONDS", 45), - MySQLDeliveryTemplateName: env("DELIVERY_MYSQL_TEMPLATE_NAME", "XINFRA MySQL Native Prototype"), - MySQLInspectTemplateName: env("DELIVERY_MYSQL_INSPECT_TEMPLATE_NAME", "XINFRA MySQL Inspect"), - MySQLInspectTimeoutSeconds: envInt("DELIVERY_MYSQL_INSPECT_TIMEOUT_SECONDS", 90), - RollbackTemplateID: uint64(envInt("DELIVERY_ROLLBACK_TEMPLATE_ID", 0)), - DeliveryServiceToken: env("DELIVERY_SERVICE_TOKEN", ""), - DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false), - DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5), - DeliveryPollSeconds: envInt("DELIVERY_POLL_SECONDS", 5), - DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)), - DeliveryCredentialSecret: env("DELIVERY_CREDENTIAL_SECRET", env("JWT_SECRET", "change-this-secret")), - ReservationTTLMinutes: envInt("DELIVERY_RESERVATION_TTL_MINUTES", 120), - DeliveryGlobalLimit: envInt("DELIVERY_GLOBAL_LIMIT", 2), - DeliveryTargetLimit: envInt("DELIVERY_TARGET_LIMIT", 2), - DeliveryHostInstanceLimit: envInt("DELIVERY_HOST_INSTANCE_LIMIT", 4), - DeliveryBusinessLimit: envInt("DELIVERY_BUSINESS_LIMIT", 1), - DeliveryDataDisks: splitCSV(env("DELIVERY_DATA_DISKS", "/data,/disk1,/mnt,/opt/mysql-delivery")), - SINABaseURL: trimURL(env("SINA_BASE_URL", "https://sinai.qiniu.io:443")), - SINAUsername: env("SINA_USERNAME", ""), - SINAPassword: env("SINA_PASSWORD", ""), + AppEnv: env("APP_ENV", "dev"), + HTTPAddr: httpAddr, + PublicBaseURL: publicBaseURL, + MySQLDSN: env("MYSQL_DSN", "auth:auth@tcp(127.0.0.1:3306)/authserver?charset=utf8mb4&parseTime=True&loc=Local"), + AutoMigrate: envBool("AUTO_MIGRATE", true), + SSOEnabled: envBool("SSO_ENABLED", true), + JWTSecret: env("JWT_SECRET", "change-this-secret"), + JWTIssuer: env("JWT_ISSUER", "authserver"), + JWTTTLMinutes: envInt("JWT_TTL_MINUTES", 120), + SAMLEntityID: samlEntityID, + SAMLACSURL: samlACSURL, + SAMLSPCert: env("SAML_SP_CERT_FILE", "certs/sp.crt"), + SAMLSPKey: env("SAML_SP_KEY_FILE", "certs/sp.key"), + SAMLIDPMetaURL: env("SAML_IDP_METADATA_URL", "http://sso-internal.dev.qiniu.io/saml2/meta"), + SAMLLogoutURL: trimURL(env("SAML_LOGOUT_URL", "")), + WayenLoginURL: env("WAYEN_LOGIN_URL", ""), + WayenTargetURL: env("WAYEN_TARGET_URL", ""), + WayenUsernameKey: env("WAYEN_USERNAME_KEY", "email"), + WayenPasswordKey: env("WAYEN_PASSWORD_KEY", "password"), + WayenLoginFormat: env("WAYEN_LOGIN_FORMAT", "form"), + WayenLoginValue: env("WAYEN_LOGIN_VALUE", "email"), + WayenOAuthRef: env("WAYEN_OAUTH_REF", "/portal/namespace/1/app"), + WayenOAuthLoginURL: trimURL(env("WAYEN_OAUTH_LOGIN_URL", "")), + WayneAPIBaseURL: trimURL(env("WAYNE_API_BASE_URL", env("WAYNE_INTERNAL_API_BASE_URL", ""))), + WayneAdminUsername: env("WAYNE_ADMIN_USERNAME", ""), + WayneAdminPassword: env("WAYNE_ADMIN_PASSWORD", ""), + WayneTokenTTLMinutes: envInt("WAYNE_TOKEN_TTL_MINUTES", 1440), + WayneInternalAPIBaseURL: trimURL(env("WAYNE_INTERNAL_API_BASE_URL", "")), + WayneServiceName: env("WAYNE_SERVICE_NAME", "xinfra"), + WayneServiceAPISecretKey: env("WAYNE_SERVICE_API_SECRET_KEY", ""), + OAuthClientID: env("OAUTH_WAYNE_CLIENT_ID", "wayne"), + OAuthClientSecret: env("OAUTH_WAYNE_CLIENT_SECRET", "wayne-secret"), + OAuthRedirectURI: env("OAUTH_WAYNE_REDIRECT_URI", ""), + OAuthCodeTTLSeconds: envInt("OAUTH_CODE_TTL_SECONDS", 120), + OIDCIssuer: oidcIssuer, + OIDCAuthorizeURL: trimURL(env("OIDC_AUTHORIZATION_ENDPOINT", oidcIssuer+"/oauth/authorize")), + OIDCTokenURL: trimURL(env("OIDC_TOKEN_ENDPOINT", oidcIssuer+"/oauth/token")), + OIDCUserInfoURL: trimURL(env("OIDC_USERINFO_ENDPOINT", oidcIssuer+"/oauth/userinfo")), + OIDCJWKSURL: trimURL(env("OIDC_JWKS_URI", oidcIssuer+"/oauth/jwks")), + CloudDMClientID: env("OIDC_CLOUDDM_CLIENT_ID", "clouddm"), + CloudDMClientSecret: env("OIDC_CLOUDDM_CLIENT_SECRET", ""), + CloudDMRedirectURI: env("OIDC_CLOUDDM_REDIRECT_URI", ""), + CloudDMTargetURL: env("CLOUDDM_TARGET_URL", ""), + CloudDMPublicURL: trimURL(env("CLOUDDM_PUBLIC_URL", "")), + CloudDMLoginURL: trimURL(env("CLOUDDM_LOGIN_URL", "")), + CloudDMAdminUsername: env("CLOUDDM_ADMIN_USERNAME", ""), + CloudDMAdminPassword: env("CLOUDDM_ADMIN_PASSWORD", ""), + CloudDMRegisterURL: trimURL(env("CLOUDDM_REGISTER_URL", "")), + CloudDMDeleteURL: trimURL(env("CLOUDDM_DELETE_URL", "")), + CloudDMAPIToken: env("CLOUDDM_API_TOKEN", ""), + CloudDMPostgreSQLRegisterURL: trimURL(env("CLOUDDM_POSTGRESQL_REGISTER_URL", "")), + CloudDMPostgreSQLUsername: env("CLOUDDM_POSTGRESQL_USERNAME", "postgres"), + CloudDMPostgreSQLPassword: env("CLOUDDM_POSTGRESQL_PASSWORD", ""), + AWXBaseURL: trimURL(env("AWX_BASE_URL", "")), + AWXToken: env("AWX_TOKEN", ""), + AWXUsername: env("AWX_USERNAME", ""), + AWXPassword: env("AWX_PASSWORD", ""), + AWXWebhookToken: env("AWX_WEBHOOK_TOKEN", ""), + AWXFactsTemplateID: envUint64("AWX_FACTS_TEMPLATE_ID", 0), + AWXFactsTimeoutSeconds: envInt("AWX_FACTS_TIMEOUT_SECONDS", 45), + MySQLDeliveryTemplateName: env("DELIVERY_MYSQL_TEMPLATE_NAME", "XINFRA MySQL Native Prototype"), + MySQLInspectTemplateName: env("DELIVERY_MYSQL_INSPECT_TEMPLATE_NAME", "XINFRA MySQL Inspect"), + MySQLInspectTimeoutSeconds: envInt("DELIVERY_MYSQL_INSPECT_TIMEOUT_SECONDS", 90), + RollbackTemplateID: uint64(envInt("DELIVERY_ROLLBACK_TEMPLATE_ID", 0)), + PostgreSQLRollbackTemplateID: uint64(envInt("DELIVERY_POSTGRESQL_ROLLBACK_TEMPLATE_ID", 0)), + DeliveryServiceToken: env("DELIVERY_SERVICE_TOKEN", ""), + DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false), + DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5), + DeliveryPollSeconds: envInt("DELIVERY_POLL_SECONDS", 5), + DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)), + DeliveryCredentialSecret: env("DELIVERY_CREDENTIAL_SECRET", env("JWT_SECRET", "change-this-secret")), + ReservationTTLMinutes: envInt("DELIVERY_RESERVATION_TTL_MINUTES", 120), + DeliveryGlobalLimit: envInt("DELIVERY_GLOBAL_LIMIT", 2), + DeliveryTargetLimit: envInt("DELIVERY_TARGET_LIMIT", 2), + DeliveryHostInstanceLimit: envInt("DELIVERY_HOST_INSTANCE_LIMIT", 4), + DeliveryBusinessLimit: envInt("DELIVERY_BUSINESS_LIMIT", 1), + DeliveryDataDisks: splitCSV(env("DELIVERY_DATA_DISKS", "/data,/disk1,/mnt,/opt/mysql-delivery")), + SINABaseURL: trimURL(env("SINA_BASE_URL", "https://sinai.qiniu.io:443")), + SINAUsername: env("SINA_USERNAME", ""), + SINAPassword: env("SINA_PASSWORD", ""), } } diff --git a/server/internal/handler/delivery.go b/server/internal/handler/delivery.go index eb4416f..5949824 100644 --- a/server/internal/handler/delivery.go +++ b/server/internal/handler/delivery.go @@ -16,15 +16,16 @@ import ( ) type DeliveryHandler struct { - service *service.DeliveryService + service *service.DeliveryService + postgresqlService *service.PostgreSQLDeliveryService } type PostgreSQLDeliveryHandler struct { service *service.PostgreSQLDeliveryService } -func NewDeliveryHandler(s *service.DeliveryService) *DeliveryHandler { - return &DeliveryHandler{service: s} +func NewDeliveryHandler(s *service.DeliveryService, postgresqlService *service.PostgreSQLDeliveryService) *DeliveryHandler { + return &DeliveryHandler{service: s, postgresqlService: postgresqlService} } func NewPostgreSQLDeliveryHandler(s *service.PostgreSQLDeliveryService) *PostgreSQLDeliveryHandler { @@ -345,7 +346,14 @@ func (h *DeliveryHandler) RetryRollback(c *gin.Context) { if !requirePlatformAdmin(c) { return } - if err := h.service.RetryRollback(c.Request.Context(), c.Param("id")); err != nil { + taskID := c.Param("id") + task, _, err := h.service.GetTask(c.Request.Context(), taskID, 0, true) + if err == nil && task.ServiceType == "postgresql" && h.postgresqlService != nil { + err = h.postgresqlService.RetryRollback(c.Request.Context(), taskID) + } else if err == nil { + err = h.service.RetryRollback(c.Request.Context(), taskID) + } + if err != nil { c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) return } @@ -358,7 +366,14 @@ func (h *DeliveryHandler) AcknowledgeRollbackRelease(c *gin.Context) { if !requirePlatformAdmin(c) { return } - if err := h.service.AcknowledgeRollbackRelease(c.Request.Context(), c.Param("id")); err != nil { + taskID := c.Param("id") + task, _, err := h.service.GetTask(c.Request.Context(), taskID, 0, true) + if err == nil && task.ServiceType == "postgresql" && h.postgresqlService != nil { + err = h.postgresqlService.AcknowledgeRollbackRelease(c.Request.Context(), taskID) + } else if err == nil { + err = h.service.AcknowledgeRollbackRelease(c.Request.Context(), taskID) + } + if err != nil { c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) return } @@ -374,7 +389,8 @@ func (h *DeliveryHandler) RetryCloudDMRegistration(c *gin.Context) { return } taskID := c.Param("id") - if _, _, err := h.service.GetTask(c.Request.Context(), taskID, claims.UserID, claims.IsAdmin); err != nil { + task, _, err := h.service.GetTask(c.Request.Context(), taskID, claims.UserID, claims.IsAdmin) + if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) return @@ -382,7 +398,12 @@ func (h *DeliveryHandler) RetryCloudDMRegistration(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } - if err := h.service.RetryCloudDMRegistration(c.Request.Context(), taskID); err != nil { + if task.ServiceType == "postgresql" && h.postgresqlService != nil { + err = h.postgresqlService.RetryCloudDMRegistration(c.Request.Context(), taskID) + } else { + err = h.service.RetryCloudDMRegistration(c.Request.Context(), taskID) + } + if err != nil { c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) return } diff --git a/server/internal/model/delivery.go b/server/internal/model/delivery.go index 4722d77..0565dcb 100644 --- a/server/internal/model/delivery.go +++ b/server/internal/model/delivery.go @@ -167,6 +167,7 @@ type PostgreSQLInstance struct { UpstreamInstanceID string `gorm:"size:63;not null;default:''" json:"upstream_instance_id"` ReplicationSlotName string `gorm:"size:63;not null;default:''" json:"replication_slot_name"` Status string `gorm:"size:32;not null;index" json:"status"` + Metadata string `gorm:"type:json" json:"metadata"` BackupStatus string `gorm:"size:32;not null;default:'not_configured'" json:"backup_status"` MonitoringStatus string `gorm:"size:32;not null;default:'not_configured'" json:"monitoring_status"` CreatedAt time.Time `json:"created_at"` diff --git a/server/internal/router/router.go b/server/internal/router/router.go index 34226c1..14b4463 100644 --- a/server/internal/router/router.go +++ b/server/internal/router/router.go @@ -90,7 +90,7 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { clouddmHandler := handler.NewCloudDMHandler(deps.Config, auditService) samlHandler := handler.NewSAMLHandler(deps.Config, authService) oauthHandler := handler.NewOAuthHandler(deps.Config, deps.DB, auditService) - deliveryHandler := handler.NewDeliveryHandler(deliveryService) + deliveryHandler := handler.NewDeliveryHandler(deliveryService, postgresqlDeliveryService) deliveryCallbackHandler := handler.NewDeliveryCallbackHandler(deliveryService, deps.Config.AWXWebhookToken) postgresqlDeliveryHandler := handler.NewPostgreSQLDeliveryHandler(postgresqlDeliveryService) containerServiceHandler := handler.NewContainerServiceHandler(deps.DB, wayneRoleBindingService) diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 931df17..840d177 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -405,6 +405,9 @@ func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([] if !strings.Contains(text, component) { continue } + if component == postgresqlServiceType && strings.Contains(text, "rollback") { + continue + } } target, err := s.awxDeliveryTarget(ctx, template) if err != nil { @@ -2860,7 +2863,10 @@ func (s *DeliveryService) markRollbackFailed(ctx context.Context, taskID, messag func (s *DeliveryService) PollRollbackOnce(ctx context.Context) error { var jobs []model.RollbackJob - if err := s.db.WithContext(ctx).Where("status IN ?", []string{"launching", "running"}).Find(&jobs).Error; err != nil { + 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 = '' OR delivery_tasks.service_type = ?)", []string{"launching", "running"}, "mysql"). + Find(&jobs).Error; err != nil { return err } for _, rollback := range jobs { diff --git a/server/internal/service/postgresql_delivery.go b/server/internal/service/postgresql_delivery.go index 9f8a14a..e5b151f 100644 --- a/server/internal/service/postgresql_delivery.go +++ b/server/internal/service/postgresql_delivery.go @@ -28,6 +28,7 @@ const ( 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"` @@ -230,7 +231,8 @@ func getPostgreSQLTarget(ctx context.Context, awx *AWXClient, templateID uint64) if err != nil { return DeliveryTarget{}, fmt.Errorf("deployment target is unavailable: %w", err) } - if !strings.Contains(strings.ToLower(template.Name+" "+template.Description), "postgresql") { + 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) @@ -342,7 +344,12 @@ func (s *PostgreSQLDeliveryService) claimAndReserve(ctx context.Context) (*model } port, err := allocateReachablePostgreSQLPort(ctx, host.IP, usedPorts, postgresqlPortInUse) if err != nil { - _ = tx.Delete(&cluster).Error + 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" @@ -390,6 +397,9 @@ func (s *PostgreSQLDeliveryService) claimAndReserve(ctx context.Context) (*model } 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 } @@ -463,13 +473,13 @@ func (s *PostgreSQLDeliveryService) CreateExecution(ctx context.Context, task *m func (s *PostgreSQLDeliveryService) DispatchOnce(ctx context.Context) error { task, err := s.claimAndReserve(ctx) if err != nil { - if errors.Is(err, gorm.ErrRecordNotFound) || strings.HasPrefix(err.Error(), "defer:") { + 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.fail(ctx, task.ID, model.TaskExecutionFailed, err.Error()) + return s.beginRollback(ctx, task.ID, "PostgreSQL deployment could not be started: "+err.Error()) } return nil } @@ -490,13 +500,13 @@ func (s *PostgreSQLDeliveryService) PollOnce(ctx context.Context) error { case "pending", "waiting", "running", "new": continue case "canceled": - _ = s.fail(ctx, execution.TaskID, model.TaskCanceled, "AWX job was canceled") + _ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job was canceled") case "successful": if err := s.complete(ctx, execution.TaskID); err != nil { - _ = s.fail(ctx, execution.TaskID, model.TaskValidationFailed, err.Error()) + _ = s.beginRollback(ctx, execution.TaskID, err.Error()) } default: - _ = s.fail(ctx, execution.TaskID, model.TaskExecutionFailed, "AWX job finished with status "+job.Status) + _ = s.beginRollback(ctx, execution.TaskID, "PostgreSQL AWX job finished with status "+job.Status) } } return nil @@ -524,7 +534,7 @@ func (s *PostgreSQLDeliveryService) complete(ctx context.Context, taskID string) } } now := time.Now() - return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + 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 } @@ -545,8 +555,22 @@ func (s *PostgreSQLDeliveryService) complete(ctx context.Context, taskID string) 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 s.common.transitionTx(tx, &task, model.TaskFinished, "PostgreSQL delivery completed and recorded in the PostgreSQL resource ledger", "") - }) + 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 { @@ -558,36 +582,6 @@ func postgresReady(ctx context.Context, address string) error { return conn.Close() } -func (s *PostgreSQLDeliveryService) fail(ctx context.Context, taskID, status, 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 err := s.common.transitionTx(tx, &task, status, message, message); err != nil { - return err - } - reservationStatus := "released" - if status == model.TaskExecutionFailed || (status == model.TaskValidationFailed && strings.Contains(strings.ToLower(message), "health")) { - reservationStatus = "quarantined" - } - if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", taskID, "reserved").Update("status", reservationStatus).Error; err != nil { - return err - } - instanceStatus := "failed" - if reservationStatus == "quarantined" { - instanceStatus = "quarantined" - } - if err := tx.Model(&model.PostgreSQLInstance{}).Where("task_id = ? AND status = ?", taskID, "provisioning").Update("status", instanceStatus).Error; err != nil { - return err - } - if err := tx.Model(&model.PostgreSQLCluster{}).Where("task_id = ? AND status = ?", taskID, "provisioning").Update("status", instanceStatus).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) Run(ctx context.Context) { interval := time.Duration(s.cfg.DeliveryPollSeconds) * time.Second if interval < time.Second { @@ -602,6 +596,7 @@ func (s *PostgreSQLDeliveryService) Run(ctx context.Context) { case <-ticker.C: _ = s.DispatchOnce(ctx) _ = s.PollOnce(ctx) + _ = s.PollRollbackOnce(ctx) } } } diff --git a/server/internal/service/postgresql_delivery_test.go b/server/internal/service/postgresql_delivery_test.go index f5c79a0..9f457ef 100644 --- a/server/internal/service/postgresql_delivery_test.go +++ b/server/internal/service/postgresql_delivery_test.go @@ -2,7 +2,10 @@ package service import ( "context" + "encoding/json" "testing" + + "github.com/1024XEngineer/xinfra/server/internal/model" ) func TestValidatePostgreSQLDeliveryInput(t *testing.T) { @@ -102,3 +105,42 @@ func TestSelectPostgreSQLHosts(t *testing.T) { t.Fatal("host outside the pool was accepted") } } + +func TestBuildPostgreSQLCloudDMRegisterRequest(t *testing.T) { + req := buildPostgreSQLCloudDMRegisterRequest( + model.PostgreSQLInstance{ID: 42, InstanceID: "orders-pg-primary", HostIP: "10.0.0.10", Port: 15432, VersionMajor: "16", Role: "primary"}, + model.PostgreSQLCluster{ID: 7, Name: "orders-pg"}, + "postgres", + "secret", + ) + if req.SourceSystem != "xinfra" || req.ResourceType != "POSTGRESQL_INSTANCE" { + t.Fatalf("unexpected request metadata: %#v", req) + } + if req.ExternalResourceID != "postgresql-instance:42" { + t.Fatalf("unexpected externalResourceId: %q", req.ExternalResourceID) + } + if req.DataSource.DSType != "PostgreSQL" || req.DataSource.Host != "10.0.0.10:15432" || req.DataSource.UserName != "postgres" || req.DataSource.Password != "secret" { + t.Fatalf("unexpected PostgreSQL data source: %#v", req.DataSource) + } + if req.DataSource.PostgreSQLVersion != "16" || req.DataSource.DefaultSchema != "postgres" { + t.Fatalf("unexpected PostgreSQL version/schema: %#v", req.DataSource) + } + raw, err := json.Marshal(req) + if err != nil { + t.Fatalf("marshal PostgreSQL CloudDM request: %v", err) + } + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatalf("decode PostgreSQL CloudDM request: %v", err) + } + dataSource := decoded["dataSource"].(map[string]any) + if dataSource["postgresql_version"] != "16" { + t.Fatalf("postgresql_version missing from contract: %#v", dataSource) + } +} + +func TestPostgreSQLCloudDMExternalResourceID(t *testing.T) { + if got := postgresqlCloudDMExternalResourceID(99); got != "postgresql-instance:99" { + t.Fatalf("external resource id = %q", got) + } +} diff --git a/server/internal/service/postgresql_integrations.go b/server/internal/service/postgresql_integrations.go new file mode 100644 index 0000000..a119ae5 --- /dev/null +++ b/server/internal/service/postgresql_integrations.go @@ -0,0 +1,426 @@ +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", "") + }) +}