From c957627ebbeb640fc39bbdd7f1f79f729ea7cddc Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Fri, 18 Aug 2023 11:57:37 +0000 Subject: [PATCH v28 3/3] pg_upgrade: Add check function for logical replication slots To prevent data loss, pg_upgrade will fail if the old node has slots with the status 'lost', or with unconsumed WAL records. Author: Hayato Kuroda Reviewed-by: Wang Wei, Vignesh C, Peter Smith, Hou Zhijie --- src/bin/pg_upgrade/check.c | 87 ++++++++++++++++++ src/bin/pg_upgrade/controldata.c | 38 ++++++++ src/bin/pg_upgrade/pg_upgrade.h | 2 + .../t/003_logical_replication_slots.pl | 91 +++++++++++++++++-- 4 files changed, 210 insertions(+), 8 deletions(-) diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c index 4056b7a7a9..a013366280 100644 --- a/src/bin/pg_upgrade/check.c +++ b/src/bin/pg_upgrade/check.c @@ -9,6 +9,7 @@ #include "postgres_fe.h" +#include "access/xlogdefs.h" #include "catalog/pg_authid_d.h" #include "catalog/pg_collation.h" #include "fe_utils/string_utils.h" @@ -31,6 +32,7 @@ static void check_for_pg_role_prefix(ClusterInfo *cluster); static void check_for_new_tablespace_dir(void); static void check_for_user_defined_encoding_conversions(ClusterInfo *cluster); static void check_new_cluster_logical_replication_slots(void); +static void check_old_cluster_for_valid_slots(bool live_check); /* @@ -108,6 +110,13 @@ check_and_dump_old_cluster(bool live_check) check_for_reg_data_type_usage(&old_cluster); check_for_isn_and_int8_passing_mismatch(&old_cluster); + /* + * Logical replication slots can be migrated since PG17. See comments atop + * get_old_cluster_logical_slot_infos(). + */ + if (GET_MAJOR_VERSION(old_cluster.major_version) >= 1700) + check_old_cluster_for_valid_slots(live_check); + /* * PG 16 increased the size of the 'aclitem' type, which breaks the * on-disk format for existing data. @@ -1471,3 +1480,81 @@ check_new_cluster_logical_replication_slots(void) check_ok(); } + +/* + * check_old_cluster_for_valid_slots() + * + * Make sure logical replication slots can be migrated to new cluster. + * Following points are checked: + * + * - All logical replication slots are usable. + * - All logical replication slots consumed all WALs, except a + * CHECKPOINT_SHUTDOWN record. + */ +static void +check_old_cluster_for_valid_slots(bool live_check) +{ + int i, + ntups, + i_slotname; + PGresult *res; + DbInfo *active_db = &old_cluster.dbarr.dbs[0]; + PGconn *conn; + + /* Quick exit if the cluster does not have logical slots. */ + if (count_old_cluster_logical_slots() == 0) + return; + + conn = connectToServer(&old_cluster, active_db->db_name); + + prep_status("Checking for logical replication slots"); + + /* Check there are no logical replication slots with a 'lost' state. */ + res = executeQueryOrDie(conn, + "SELECT slot_name FROM pg_catalog.pg_replication_slots " + "WHERE wal_status = 'lost' AND " + "temporary IS FALSE;"); + + ntups = PQntuples(res); + i_slotname = PQfnumber(res, "slot_name"); + + for (i = 0; i < ntups; i++) + pg_log(PG_WARNING, + "\nWARNING: logical replication slot \"%s\" is in 'lost' state.", + PQgetvalue(res, i, i_slotname)); + + PQclear(res); + + if (ntups) + pg_fatal("One or more logical replication slots with a state of 'lost' were detected."); + + /* + * Do additional checks if a live check is not required. This requires + * that confirmed_flush_lsn of all the slots is the same as the latest + * checkpoint location, but it would be satisfied only when the server has + * been shut down. + */ + if (!live_check) + { + res = executeQueryOrDie(conn, + "SELECT slot_name FROM pg_catalog.pg_replication_slots " + "WHERE confirmed_flush_lsn != '%X/%X' AND temporary IS FALSE;", + LSN_FORMAT_ARGS(old_cluster.controldata.chkpnt_latest)); + + ntups = PQntuples(res); + i_slotname = PQfnumber(res, "slot_name"); + + for (i = 0; i < ntups; i++) + pg_log(PG_WARNING, + "\nWARNING: logical replication slot \"%s\" has not consumed WALs yet", + PQgetvalue(res, i, i_slotname)); + + PQclear(res); + + if (ntups) + pg_fatal("One or more logical replication slots still have unconsumed WAL records."); + } + + PQfinish(conn); + check_ok(); +} diff --git a/src/bin/pg_upgrade/controldata.c b/src/bin/pg_upgrade/controldata.c index 4beb65ab22..808156ec09 100644 --- a/src/bin/pg_upgrade/controldata.c +++ b/src/bin/pg_upgrade/controldata.c @@ -169,6 +169,44 @@ get_control_data(ClusterInfo *cluster, bool live_check) } got_cluster_state = true; } + + else if ((p = strstr(bufin, "Latest checkpoint location:")) != NULL) + { + /* + * Read the latest checkpoint location if the cluster is PG17 + * or later. This is used for upgrading logical replication + * slots. + */ + if (GET_MAJOR_VERSION(cluster->major_version) >= 1700) + { + char *slash = NULL; + uint32 upper_lsn, + lower_lsn; + + p = strchr(p, ':'); + + if (p == NULL || strlen(p) <= 1) + pg_fatal("%d: controldata retrieval problem", __LINE__); + + p++; /* remove ':' char */ + + p = strpbrk(p, "01234567890ABCDEF"); + + if (p == NULL || strlen(p) <= 1) + pg_fatal("%d: controldata retrieval problem", __LINE__); + + /* + * The upper and lower part of LSN must be read separately + * because it is stored as in %X/%X format. + */ + upper_lsn = strtoul(p, &slash, 16); + lower_lsn = strtoul(++slash, NULL, 16); + + /* And combine them */ + cluster->controldata.chkpnt_latest = + ((uint64) upper_lsn << 32) | lower_lsn; + } + } } rc = pclose(output); diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h index dae92ef6c0..e72318f500 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -10,6 +10,7 @@ #include #include +#include "access/xlogdefs.h" #include "common/relpath.h" #include "libpq-fe.h" @@ -242,6 +243,7 @@ typedef struct bool date_is_int; bool float8_pass_by_value; uint32 data_checksum_version; + XLogRecPtr chkpnt_latest; } ControlData; /* diff --git a/src/bin/pg_upgrade/t/003_logical_replication_slots.pl b/src/bin/pg_upgrade/t/003_logical_replication_slots.pl index ae87c33708..640964c4e1 100644 --- a/src/bin/pg_upgrade/t/003_logical_replication_slots.pl +++ b/src/bin/pg_upgrade/t/003_logical_replication_slots.pl @@ -22,6 +22,10 @@ $old_publisher->init(allows_streaming => 'logical'); my $new_publisher = PostgreSQL::Test::Cluster->new('new_publisher'); $new_publisher->init(allows_streaming => 'replica'); +# Initialize subscriber cluster +my $subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$subscriber->init(allows_streaming => 'logical'); + my $bindir = $new_publisher->config_data('--bindir'); # ------------------------------ @@ -65,13 +69,19 @@ $old_publisher->start; $old_publisher->safe_psql('postgres', "SELECT pg_create_logical_replication_slot('test_slot2', 'test_decoding', false, true);" ); + +# 2. Consume WAL records to avoid another type of upgrade failure. It will be +# tested in subsequent cases. +$old_publisher->safe_psql('postgres', + "SELECT count(*) FROM pg_logical_slot_get_changes('test_slot1', NULL, NULL);" +); $old_publisher->stop; -# 2. max_replication_slots is set to smaller than the number of slots (2) +# 3. max_replication_slots is set to smaller than the number of slots (2) # present on the old cluster $new_publisher->append_conf('postgresql.conf', "max_replication_slots = 1"); -# 3. wal_level is set correctly on the new cluster +# 4. wal_level is set correctly on the new cluster $new_publisher->append_conf('postgresql.conf', "wal_level = 'logical'"); # pg_upgrade will fail because the new cluster has insufficient max_replication_slots @@ -95,7 +105,7 @@ ok( -d $new_publisher->data_dir . "/pg_upgrade_output.d", rmtree($new_publisher->data_dir . "/pg_upgrade_output.d"); # ------------------------------ -# TEST: Successful upgrade +# TEST: Confirm pg_upgrade fails when the slot still has unconsumed WAL records # Preparations for the subsequent test: # 1. Remove the slot 'test_slot2', leaving only 1 slot remaining on the old @@ -106,10 +116,57 @@ $old_publisher->safe_psql('postgres', "SELECT * FROM pg_drop_replication_slot('test_slot2');" ); -# 2. Consume WAL records +# 2. Generate extra WAL records. Because these WAL records do not get consumed +# it will cause the upcoming pg_upgrade test to fail. +$old_publisher->safe_psql('postgres', + "CREATE TABLE tbl AS SELECT generate_series(1, 10) AS a;" +); +$old_publisher->stop; + +# pg_upgrade will fail because the slot still has unconsumed WAL records +command_fails( + [ + 'pg_upgrade', '--no-sync', + '-d', $old_publisher->data_dir, + '-D', $new_publisher->data_dir, + '-b', $bindir, + '-B', $bindir, + '-s', $new_publisher->host, + '-p', $old_publisher->port, + '-P', $new_publisher->port, + $mode, + ], + 'run of pg_upgrade of old cluster with idle replication slots'); +ok( -d $new_publisher->data_dir . "/pg_upgrade_output.d", + "pg_upgrade_output.d/ not removed after pg_upgrade failure"); + +# Clean up +rmtree($new_publisher->data_dir . "/pg_upgrade_output.d"); +# Remove the remained slot +$old_publisher->start; +$old_publisher->safe_psql('postgres', + "SELECT * FROM pg_drop_replication_slot('test_slot1');" +); + +# ------------------------------ +# TEST: Successful upgrade + +# Preparations for the subsequent test: +# 1. Setup logical replication +my $old_connstr = $old_publisher->connstr . ' dbname=postgres'; $old_publisher->safe_psql('postgres', - "SELECT count(*) FROM pg_logical_slot_get_changes('test_slot1', NULL, NULL)" + "CREATE PUBLICATION pub FOR ALL TABLES;" ); +$subscriber->start; +$subscriber->safe_psql( + 'postgres', qq[ + CREATE TABLE tbl (a int); + CREATE SUBSCRIPTION sub CONNECTION '$old_connstr' PUBLICATION pub WITH (two_phase = 'true') +]); +$subscriber->wait_for_subscription_sync($old_publisher, 'sub'); + +# 2. Temporarily disable the subscription +$subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub DISABLE"); $old_publisher->stop; # Actual run, successful upgrade is expected @@ -129,11 +186,29 @@ command_ok( ok( !-d $new_publisher->data_dir . "/pg_upgrade_output.d", "pg_upgrade_output.d/ removed after pg_upgrade success"); -# Check that the slot 'test_slot1' has migrated to the new cluster +# Check that the slot 'sub' has migrated to the new cluster $new_publisher->start; my $result = $new_publisher->safe_psql('postgres', "SELECT slot_name, two_phase FROM pg_replication_slots"); -is($result, qq(test_slot1|t), 'check the slot exists on new cluster'); -$new_publisher->stop; +is($result, qq(sub|t), 'check the slot exists on new cluster'); + +# Update the connection +my $new_connstr = $new_publisher->connstr . ' dbname=postgres'; +$subscriber->safe_psql( + 'postgres', qq[ + ALTER SUBSCRIPTION sub CONNECTION '$new_connstr'; + ALTER SUBSCRIPTION sub ENABLE; +]); + +# Check whether changes on the new publisher get replicated to the subscriber +$new_publisher->safe_psql('postgres', + "INSERT INTO tbl VALUES (generate_series(11, 20))"); +$new_publisher->wait_for_catchup('sub'); +$result = $subscriber->safe_psql('postgres', "SELECT count(*) FROM tbl"); +is($result, qq(20), 'check changes are replicated to the subscriber'); + +# Clean up +$subscriber->stop(); +$new_publisher->stop(); done_testing(); -- 2.27.0