New to Kestra?
Use blueprints to kickstart your first workflows.
Upgrade PostgreSQL to a new major version with a verified cutover
Upgrade PostgreSQL to a new major version with logical replication. Freeze the old server, catch up, compare every table and cut over with a recorded result.
Categories
Data
# SPDX-License-Identifier: Apache-2.0
id: postgres-major-upgrade-logical-replication-cutover
namespace: company.team
description: Upgrade one PostgreSQL database to a newer major version with
logical replication, then cut over only after the new server has caught up and
every table matches.
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: mode
type: SELECT
values:
- PREPARE
- CUTOVER
- ROLLBACK
- FINALIZE
defaults: PREPARE
description: PREPARE copies the schema and starts replication. CUTOVER freezes
the old server and proves the match. ROLLBACK reopens the old server.
FINALIZE removes the replication objects after a completed cutover.
- id: source_host
type: STRING
defaults: pg-source
- id: source_port
type: STRING
defaults: "5432"
validator: ^[0-9]{2,5}$
- id: target_host
type: STRING
defaults: pg-target
- id: target_port
type: STRING
defaults: "5432"
validator: ^[0-9]{2,5}$
- id: database
type: STRING
defaults: appdb
validator: ^[a-zA-Z_][a-zA-Z0-9_]{0,62}$
- id: admin_user
type: STRING
defaults: postgres
validator: ^[a-zA-Z_][a-zA-Z0-9_]{0,62}$
- id: source_password
type: SECRET
defaults: "{{ secret('PG_SOURCE_PASSWORD') }}"
- id: target_password
type: SECRET
defaults: "{{ secret('PG_TARGET_PASSWORD') }}"
- id: replication_name
type: STRING
defaults: kestra_upgrade
validator: ^[a-z_][a-z0-9_]{0,40}$
description: Name of the publication, the subscription and the replication slot.
- id: tools_image
type: STRING
defaults: postgres:17.5
description: Image that provides pg_dump and psql. Use the target major version
or newer.
- id: docker_network
type: STRING
defaults: bridge
- id: lag_rounds
type: STRING
defaults: "12"
validator: ^[0-9]{1,2}$
description: Maximum catch-up checks during a cutover. They run 5 seconds apart.
- id: dry_run
type: BOOL
defaults: true
- id: auto_remediate
type: BOOL
defaults: false
variables:
source_url: "jdbc:postgresql://{{ inputs.source_host }}:{{ inputs.source_port
}}/{{ inputs.database }}"
target_url: "jdbc:postgresql://{{ inputs.target_host }}:{{ inputs.target_port
}}/{{ inputs.database }}"
live: "{{ inputs.dry_run == false and inputs.auto_remediate == true }}"
tasks:
- id: source_status
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: |
SELECT current_setting('wal_level') AS wal_level,
current_setting('server_version_num')::int AS version_num,
COALESCE((SELECT 'on' FROM pg_db_role_setting s WHERE s.setdatabase = (SELECT oid FROM pg_database WHERE datname = current_database()) AND s.setrole = 0 AND 'default_transaction_read_only=on' = ANY (s.setconfig)), 'off') AS read_only,
(SELECT count(*) FROM pg_replication_slots) AS slots_used,
current_setting('max_replication_slots')::int AS slots_max,
(SELECT count(*) FROM pg_replication_slots WHERE slot_name = '{{ inputs.replication_name }}') AS slot_exists,
(SELECT count(*) FROM pg_publication WHERE pubname = '{{ inputs.replication_name }}') AS publication_exists,
(SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind IN ('r', 'p') AND n.nspname NOT IN ('pg_catalog', 'information_schema') AND n.nspname NOT LIKE 'pg_toast%') AS tables_total,
(SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind IN ('r', 'p') AND n.nspname NOT IN ('pg_catalog', 'information_schema') AND n.nspname NOT LIKE 'pg_toast%'
AND c.relreplident NOT IN ('f', 'i')
AND NOT EXISTS (SELECT 1 FROM pg_index i WHERE i.indrelid = c.oid AND i.indisprimary)) AS tables_without_identity
- id: target_status
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: |
SELECT current_setting('server_version_num')::int AS version_num,
(SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind IN ('r', 'p') AND n.nspname NOT IN ('pg_catalog', 'information_schema', 'kestra_upgrade') AND n.nspname NOT LIKE 'pg_toast%') AS tables_total,
(SELECT count(*) FROM pg_subscription WHERE subname = '{{ inputs.replication_name }}') AS subscription_exists,
COALESCE(to_regclass('kestra_upgrade.upgrade_log')::text, '') AS log_table
- id: require_ready
type: io.kestra.plugin.core.execution.Fail
condition: >-
{{ outputs.source_status.row.wal_level != 'logical' or
outputs.source_status.row.tables_without_identity != 0 or
outputs.target_status.row.version_num <= outputs.source_status.row.version_num or
outputs.source_status.row.tables_total == 0 or
(inputs.mode == 'PREPARE' and outputs.target_status.row.subscription_exists == 0 and outputs.source_status.row.slots_used >= outputs.source_status.row.slots_max) }}
errorMessage: The source needs wal_level logical and at least one table. Every
table needs a primary key or a replica identity. A free replication slot
is needed to prepare. The target must run a newer major version.
- id: plan
type: io.kestra.plugin.core.log.Log
message: >-
Mode {{ inputs.mode }}. Source {{ outputs.source_status.row.version_num }}
with {{ outputs.source_status.row.tables_total }} tables, target {{
outputs.target_status.row.version_num }} with {{
outputs.target_status.row.tables_total }} tables. Publication present {{
outputs.source_status.row.publication_exists }}, subscription present {{
outputs.target_status.row.subscription_exists }}. {{ render(vars.live) ==
'true' ? 'Live run: changes will be applied.' : 'Dry run: nothing is
written on either server.' }}
- id: live_run
type: io.kestra.plugin.core.flow.If
condition: "{{ render(vars.live) == 'true' }}"
then:
- id: ensure_log
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: |
CREATE SCHEMA IF NOT EXISTS kestra_upgrade;
CREATE TABLE IF NOT EXISTS kestra_upgrade.upgrade_log (id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, execution_id TEXT NOT NULL, state TEXT NOT NULL, freeze_lsn TEXT, details JSONB, updated_at TIMESTAMPTZ NOT NULL DEFAULT now())
- id: upgrade_state
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: SELECT COALESCE((SELECT state FROM kestra_upgrade.upgrade_log ORDER BY id
DESC LIMIT 1), 'NONE') AS state
- id: require_allowed_state
type: io.kestra.plugin.core.execution.Fail
condition: >-
{{ (inputs.mode == 'PREPARE' and not (['NONE', 'PREPARING',
'PREPARED', 'ROLLED_BACK'] contains outputs.upgrade_state.row.state))
or
(inputs.mode == 'CUTOVER' and not (['PREPARED', 'ROLLED_BACK', 'FREEZING', 'FROZEN'] contains outputs.upgrade_state.row.state)) or
(inputs.mode == 'ROLLBACK' and not (['FREEZING', 'FROZEN', 'COMPLETED'] contains outputs.upgrade_state.row.state)) or
(inputs.mode == 'FINALIZE' and outputs.upgrade_state.row.state != 'COMPLETED') }}
errorMessage: "The recorded upgrade state does not allow this mode. PREPARE
needs no cutover in progress, CUTOVER needs a prepared upgrade,
ROLLBACK needs a freeze to undo and FINALIZE needs a completed
cutover."
- id: prepare
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'PREPARE' and (outputs.upgrade_state.row.state ==
'NONE' or outputs.upgrade_state.row.state == 'PREPARING') }}"
then:
- id: require_empty_target
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.upgrade_state.row.state == 'NONE' and
outputs.target_status.row.tables_total != 0 }}"
errorMessage: The target database already contains tables. PREPARE needs an
empty target so the copied schema matches the source.
- id: preparing_mark
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: >-
INSERT INTO kestra_upgrade.upgrade_log (execution_id, state)
SELECT '{{ execution.id }}', 'PREPARING' WHERE NOT EXISTS (SELECT
1 FROM kestra_upgrade.upgrade_log WHERE state = 'PREPARING' AND id
= (SELECT MAX(id) FROM kestra_upgrade.upgrade_log))
- id: copy_schema
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.target_status.row.tables_total == 0 }}"
then:
- id: dump_and_load_schema
type: io.kestra.plugin.scripts.shell.Commands
containerImage: "{{ inputs.tools_image }}"
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: "{{ inputs.docker_network }}"
env:
SRC_HOST: "{{ inputs.source_host }}"
SRC_PORT: "{{ inputs.source_port }}"
TGT_HOST: "{{ inputs.target_host }}"
TGT_PORT: "{{ inputs.target_port }}"
DB: "{{ inputs.database }}"
DB_USER: "{{ inputs.admin_user }}"
SRC_PW: "{{ inputs.source_password }}"
TGT_PW: "{{ inputs.target_password }}"
commands:
- set -o pipefail
- PGPASSWORD="$SRC_PW" pg_dump -h "$SRC_HOST" -p "$SRC_PORT"
-U "$DB_USER" -d "$DB" --schema-only --no-owner
--no-privileges --no-publications --no-subscriptions |
PGPASSWORD="$TGT_PW" psql -h "$TGT_HOST" -p "$TGT_PORT" -U
"$DB_USER" -d "$DB" --single-transaction -v ON_ERROR_STOP=1
-q
- id: create_publication
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
sql: |
DO $$ BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = '{{ inputs.replication_name }}') THEN
CREATE PUBLICATION {{ inputs.replication_name }} FOR ALL TABLES;
END IF;
END $$
- id: create_subscription
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.target_status.row.subscription_exists == 0 }}"
then:
- id: subscribe
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
transaction: false
sql: >-
CREATE SUBSCRIPTION {{ inputs.replication_name }} CONNECTION
'host={{ inputs.source_host }} port={{ inputs.source_port }}
dbname={{ inputs.database }} user={{ inputs.admin_user }}
password={{ inputs.source_password }}' PUBLICATION {{
inputs.replication_name }} WITH (copy_data = true, create_slot
= true)
- id: wait_initial_sync
type: io.kestra.plugin.core.flow.LoopUntil
condition: "{{ outputs.sync_progress.row.pending == 0 and
outputs.sync_progress.row.total ==
outputs.source_status.row.tables_total }}"
checkFrequency:
interval: PT5S
maxIterations: 60
failOnMaxReached: true
tasks:
- id: sync_progress
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: SELECT count(*) FILTER (WHERE srsubstate <> 'r') AS pending, count(*) AS
total FROM pg_subscription_rel
- id: prepared_mark
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: >-
INSERT INTO kestra_upgrade.upgrade_log (execution_id, state,
details) VALUES ('{{ execution.id }}', 'PREPARED',
jsonb_build_object('tables', {{ outputs.sync_progress.row.total
}}))
- id: cutover_approval
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'CUTOVER' and outputs.upgrade_state.row.state !=
'FROZEN' }}"
then:
- id: approve_cutover
type: io.kestra.plugin.core.flow.Pause
pauseDuration: PT1H
behavior: FAIL
onResume:
- id: approved
type: BOOL
defaults: false
description: Approve making the source database read only and cutting over to
the target.
- id: writers_stopped
type: BOOL
defaults: false
description: Confirm that applications stopped writing to the source or can
tolerate failed writes.
- id: cutover
type: io.kestra.plugin.core.flow.If
condition: >-
{{ inputs.mode == 'CUTOVER' and
(outputs.upgrade_state.row.state == 'FROZEN' or
(outputs.approve_cutover.onResume.approved == true and outputs.approve_cutover.onResume.writers_stopped == true)) }}
then:
- id: recheck_source
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: |
SELECT COALESCE((SELECT 'on' FROM pg_db_role_setting s WHERE s.setdatabase = (SELECT oid FROM pg_database WHERE datname = current_database()) AND s.setrole = 0 AND 'default_transaction_read_only=on' = ANY (s.setconfig)), 'off') AS read_only,
(SELECT count(*) FROM pg_replication_slots WHERE slot_name = '{{ inputs.replication_name }}' AND active) AS slot_active
- id: recheck_target
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: |
SELECT COALESCE((SELECT state FROM kestra_upgrade.upgrade_log ORDER BY id DESC LIMIT 1), 'NONE') AS state,
(SELECT count(*) FROM pg_subscription WHERE subname = '{{ inputs.replication_name }}' AND subenabled) AS subscription_enabled,
(SELECT count(*) FROM pg_subscription_rel WHERE srsubstate <> 'r') AS pending
- id: require_unchanged_plan
type: io.kestra.plugin.core.execution.Fail
condition: >-
{{ outputs.recheck_target.row.state !=
outputs.upgrade_state.row.state or
outputs.recheck_target.row.subscription_enabled != 1 or
outputs.recheck_target.row.pending != 0 or
outputs.recheck_source.row.slot_active != 1 or
(outputs.upgrade_state.row.state != 'FROZEN' and outputs.upgrade_state.row.state != 'FREEZING' and outputs.recheck_source.row.read_only != 'off') }}
errorMessage: The upgrade state, the subscription or the source changed while
the approval was pending. Nothing was frozen. Run again from the
current state.
- id: freeze_begin
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: >-
INSERT INTO kestra_upgrade.upgrade_log (execution_id, state)
SELECT '{{ execution.id }}', 'FREEZING' WHERE NOT EXISTS (SELECT 1
FROM kestra_upgrade.upgrade_log WHERE state IN ('FREEZING',
'FROZEN') AND id = (SELECT MAX(id) FROM
kestra_upgrade.upgrade_log))
- id: freeze_source
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
transaction: false
sql: |
DO $$ BEGIN EXECUTE format('ALTER DATABASE %I SET default_transaction_read_only = on', current_database()); END $$;
SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = current_database() AND pid <> pg_backend_pid() AND backend_type = 'client backend'
- id: verify_frozen
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: |
SELECT COALESCE((SELECT 'on' FROM pg_db_role_setting s WHERE s.setdatabase = (SELECT oid FROM pg_database WHERE datname = current_database()) AND s.setrole = 0 AND 'default_transaction_read_only=on' = ANY (s.setconfig)), 'off') AS read_only,
(SELECT count(*) FROM pg_db_role_setting s WHERE s.setrole <> 0 AND (s.setdatabase = 0 OR s.setdatabase = (SELECT oid FROM pg_database WHERE datname = current_database())) AND 'default_transaction_read_only=off' = ANY (s.setconfig)) AS role_overrides,
pg_current_wal_lsn()::text AS freeze_lsn
- id: require_frozen
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.verify_frozen.row.read_only != 'on' or
outputs.verify_frozen.row.role_overrides != 0 }}"
errorMessage: The source is not read only for every role after the freeze. A
role setting can override the database setting. Remove it or run
ROLLBACK to reopen the source.
- id: freeze_mark
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: >-
UPDATE kestra_upgrade.upgrade_log SET state = 'FROZEN', freeze_lsn
= '{{ outputs.verify_frozen.row.freeze_lsn }}', updated_at = now()
WHERE id = (SELECT MAX(id) FROM kestra_upgrade.upgrade_log) AND
state = 'FREEZING'
- id: frozen_state
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: SELECT freeze_lsn FROM kestra_upgrade.upgrade_log WHERE state = 'FROZEN'
ORDER BY id DESC LIMIT 1
- id: replicated_tables
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH
sql: |
SELECT format('%I.%I', schemaname, tablename) AS qname, replace(schemaname || '.' || tablename, '''', '''''') AS label
FROM pg_publication_tables WHERE pubname = '{{ inputs.replication_name }}' ORDER BY 1
- id: catch_up
type: io.kestra.plugin.core.flow.LoopUntil
condition: "{{ outputs.slot_position.row.caught_up == true and
outputs.parity.row.mismatches == 0 and outputs.parity.row.tables
== outputs.source_status.row.tables_total }}"
checkFrequency:
interval: PT5S
maxIterations: "{{ inputs.lag_rounds }}"
failOnMaxReached: true
tasks:
- id: slot_position
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: >-
SELECT COALESCE(confirmed_flush_lsn >= '{{
outputs.frozen_state.row.freeze_lsn }}'::pg_lsn, false) AS
caught_up FROM pg_replication_slots WHERE slot_name = '{{
inputs.replication_name }}'
- id: source_fingerprint
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH
sql: |
{% for r in outputs.replicated_tables.rows %}SELECT '{{ r.label }}' AS tbl, count(*)::text AS n, COALESCE(sum(('x' || substr(md5(x::text), 1, 12))::bit(48)::bigint::numeric), 0)::text AS h FROM {{ r.qname }} x{% if not loop.last %}
UNION ALL
{% endif %}{% endfor %}
- id: parity
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: |
WITH s(tbl, n, h) AS (VALUES {% for r in outputs.source_fingerprint.rows %}('{{ r.tbl | replace({"'": "''"}) }}', '{{ r.n }}', '{{ r.h }}'){% if not loop.last %}, {% endif %}{% endfor %}),
t AS (
{% for r in outputs.replicated_tables.rows %}SELECT '{{ r.label }}' AS tbl, count(*)::text AS n, COALESCE(sum(('x' || substr(md5(x::text), 1, 12))::bit(48)::bigint::numeric), 0)::text AS h FROM {{ r.qname }} x{% if not loop.last %}
UNION ALL
{% endif %}{% endfor %}
)
SELECT count(*) FILTER (WHERE s.n IS DISTINCT FROM t.n OR s.h IS DISTINCT FROM t.h) AS mismatches, count(*) AS tables
FROM s FULL JOIN t USING (tbl)
- id: source_sequences
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH
sql: |
SELECT format('%I.%I', schemaname, sequencename) AS qname, last_value::text AS last_value
FROM pg_sequences WHERE last_value IS NOT NULL AND schemaname NOT IN ('pg_catalog', 'information_schema') ORDER BY 1
- id: copy_sequences
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.source_sequences.size > 0 }}"
then:
- id: set_sequences
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: |
{% for r in outputs.source_sequences.rows %}SELECT setval('{{ r.qname | replace({"'": "''"}) }}', {{ r.last_value }}, true);
{% endfor %}
- id: verify_sequences
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: |
WITH s(q, v) AS (VALUES {% for r in outputs.source_sequences.rows %}('{{ r.qname | replace({"'": "''"}) }}', '{{ r.last_value }}'){% if not loop.last %}, {% endif %}{% endfor %})
SELECT count(*) FILTER (WHERE q.last_value IS DISTINCT FROM s.v::bigint) AS mismatches
FROM s LEFT JOIN (SELECT format('%I.%I', schemaname, sequencename) AS q, last_value FROM pg_sequences) q ON q.q = s.q
- id: require_sequences
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.verify_sequences.row.mismatches != 0 }}"
errorMessage: A sequence on the target does not match the source after the copy.
- id: record_cutover
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
fetchType: FETCH_ONE
sql: >-
UPDATE kestra_upgrade.upgrade_log SET state = 'COMPLETED', details
= jsonb_build_object('tables', {{ outputs.parity.row.tables }},
'sequences', {{ outputs.source_sequences.size }}, 'completed_by',
'{{ execution.id }}'), updated_at = now() WHERE id = (SELECT
MAX(id) FROM kestra_upgrade.upgrade_log) AND state = 'FROZEN'
RETURNING state, freeze_lsn
- id: require_recorded
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.record_cutover.row.state != 'COMPLETED' }}"
errorMessage: The cutover could not be recorded as completed.
- id: cutover_done
type: io.kestra.plugin.core.log.Log
message: >-
Cutover recorded at WAL position {{
outputs.record_cutover.row.freeze_lsn }}: {{
outputs.parity.row.tables }} tables and {{
outputs.source_sequences.size }} sequences match on the new
server. Repoint the applications now. The old server stays read
only.
- id: rollback_or_finalize
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'ROLLBACK' or inputs.mode == 'FINALIZE' }}"
then:
- id: approve_change
type: io.kestra.plugin.core.flow.Pause
pauseDuration: PT1H
behavior: FAIL
onResume:
- id: approved
type: BOOL
defaults: false
description: Approve ROLLBACK (reopen the source for writes) or FINALIZE (drop
the subscription, the slot and the publication).
- id: rollback
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'ROLLBACK' and
outputs.approve_change.onResume.approved == true }}"
then:
- id: reopen_source
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
transaction: false
sql: |
SET default_transaction_read_only = off;
DO $$ BEGIN EXECUTE format('ALTER DATABASE %I RESET default_transaction_read_only', current_database()); END $$
- id: verify_reopened
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: SELECT COALESCE((SELECT 'on' FROM pg_db_role_setting s WHERE s.setdatabase
= (SELECT oid FROM pg_database WHERE datname =
current_database()) AND s.setrole = 0 AND
'default_transaction_read_only=on' = ANY (s.setconfig)),
'off') AS read_only
- id: require_reopened
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.verify_reopened.row.read_only != 'off' }}"
errorMessage: The source database is still read only after the rollback.
- id: record_rollback
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: INSERT INTO kestra_upgrade.upgrade_log (execution_id, state) VALUES ('{{
execution.id }}', 'ROLLED_BACK')
- id: finalize
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.mode == 'FINALIZE' and
outputs.approve_change.onResume.approved == true }}"
then:
- id: drop_subscription
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
transaction: false
sql: DROP SUBSCRIPTION IF EXISTS {{ inputs.replication_name }}
- id: drop_publication
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
transaction: false
sql: DROP PUBLICATION IF EXISTS {{ inputs.replication_name }}
- id: verify_slot_gone
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.source_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.source_password }}"
fetchType: FETCH_ONE
sql: SELECT (SELECT count(*) FROM pg_replication_slots WHERE slot_name = '{{
inputs.replication_name }}') AS slots, (SELECT count(*) FROM
pg_publication WHERE pubname = '{{ inputs.replication_name
}}') AS publications
- id: require_slot_gone
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.verify_slot_gone.row.slots != 0 or
outputs.verify_slot_gone.row.publications != 0 }}"
errorMessage: The replication slot or the publication still exists on the
source.
- id: record_finalize
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ render(vars.target_url) }}"
username: "{{ inputs.admin_user }}"
password: "{{ inputs.target_password }}"
sql: INSERT INTO kestra_upgrade.upgrade_log (execution_id, state) VALUES ('{{
execution.id }}', 'FINALIZED')
errors:
- id: alert_upgrade_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: >-
{{ {'text': 'PostgreSQL upgrade ' ~ inputs.mode ~ ' failed. Execution ' ~
execution.id ~ '. If the source is read only after a failed CUTOVER, run
ROLLBACK to reopen it or fix the cause and run CUTOVER again.'} | toJson
}}
triggers:
- id: scheduled_readiness
type: io.kestra.plugin.core.trigger.Schedule
disabled: true
cron: "0 * * * *"
inputs:
mode: PREPARE
dry_run: true
auto_remediate: false
A dump and restore needs a long outage and an in-place upgrade has no way back. Logical replication lets a newer server catch up while the old one keeps serving, but the last steps are easy to get wrong: stopping writers, waiting for the exact catch-up point, copying sequences and proving that nothing is missing. This flow runs those steps in order and records the result.
How it works
PREPARE checks the source (wal_level logical, a key on every table), copies the schema with pg_dump in a single transaction, creates a publication and a subscription and waits until every table is in the ready state.
CUTOVER asks for approval through a Pause. It makes the source database read only, ends the other sessions and verifies that no role overrides the setting. It reads the WAL position and loops until the replication slot confirmed that position and a count plus an MD5 sum of every row match on both servers. Then it copies every sequence value and records COMPLETED.
ROLLBACK reopens the source. Replication keeps running, so CUTOVER can be tried again. FINALIZE drops the subscription, the slot and the publication after approval.
What you get
A target database that matches the source at a recorded WAL position. Sequences continue where the old ones stopped. A kestra_upgrade.upgrade_log table shows each state. A failed cutover leaves the source read only on purpose.
Who it's for
Platform teams that upgrade a PostgreSQL major version and want a short, reversible outage.
Why orchestrate this with Kestra
Pause holds the approval. LoopUntil waits for the catch-up and the comparison. The log table makes every mode safe to run again after a failure.
Tested end to end
Kestra 2.0.6 and 2.0.5 with PostgreSQL 16.15 as the source and 17.5 as the target. Writers ran during the initial copy and during the cutover. A prepare that failed halfway resumed without copying again. Declined and stale approvals, a role setting that kept one role writable, a table that no longer matched, rollback and finalize were exercised. Each result was compared with psql.
Prerequisites
wal_level logical on the source. A superuser for both servers. An empty target database. Docker workers for pg_dump. The same roles and extensions on the target. The flow copies tables and sequences only.
Secrets
PG_SOURCE_PASSWORD, PG_TARGET_PASSWORD and SLACK_WEBHOOK_URL.
Inputs
mode selects PREPARE, CUTOVER, ROLLBACK or FINALIZE. dry_run defaults to true and auto_remediate to false, so nothing changes until both are flipped. The subscription connection string contains the source password, so use one without spaces or quotes.
Quick start
- Set wal_level to logical on the source and restart it. Create the roles on the target.
- Run PREPARE as a dry run and read the plan. Run it again with dry_run false and auto_remediate true.
- Stop the applications. Run CUTOVER and approve it. Repoint the applications when it reports COMPLETED.
- Run FINALIZE when the old server is no longer needed.
How to extend
Replace the publication with a table list to upgrade part of a database. Add a smoke test of the new server before the approval. Large tables make the comparison slow, so sample them. Large objects and DDL during the window are not replicated.
Links
This blueprint was created by zkasuran.
Create a Blueprint
Go to the developer platformMore Data Blueprints
