From 683c4fc3e767a068eca0c85561186c1322287d69 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Fri, 21 Jun 2024 10:55:46 +0000 Subject: [PATCH] pg_createsubscriber: Drop pre-existing subscriptions from the converted instance Previously, we did nothing for pre-existing subscriptions on the streaming replication cluster. However, after the conversion, the downstream node will try to connect to another publisher node specified by the pre-existing subscriptions, which will cause an ERROR. To avoid failure, drop such subscriptions from the converted node at the end of this command. --- src/bin/pg_basebackup/pg_createsubscriber.c | 167 +++++++++++++++++- .../t/041_pg_createsubscriber_added.pl | 118 +++++++++++++ 2 files changed, 283 insertions(+), 2 deletions(-) create mode 100644 src/bin/pg_basebackup/t/041_pg_createsubscriber_added.pl diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c index 1138c20e56..b5f62c7c54 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -57,6 +57,9 @@ struct LogicalRepInfo bool made_replslot; /* replication slot was created */ bool made_publication; /* publication was created */ + int num_subscriptions; /* number of pre-existing subscriptions */ + char **pre_subnames; /* subscription name */ + char **pre_slotnames; /* used replication slots name */ }; static void cleanup_objects_atexit(void); @@ -78,7 +81,7 @@ static bool server_is_in_recovery(PGconn *conn); static char *generate_object_name(PGconn *conn); static void check_publisher(const struct LogicalRepInfo *dbinfo); static char *setup_publisher(struct LogicalRepInfo *dbinfo); -static void check_subscriber(const struct LogicalRepInfo *dbinfo); +static void check_subscriber(struct LogicalRepInfo *dbinfo); static void setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn); static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir, @@ -103,6 +106,11 @@ static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo * const char *lsn); static void enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo); +static void obtain_and_disable_pre_existing_subscriptions(struct LogicalRepInfo *dbinfo); +static void disable_subscription(PGconn *conn, const char *subname); +static void enable_subscirptions_on_publisher(struct LogicalRepInfo *dbinfo); +static void drop_pre_existing_subscriptions(struct LogicalRepInfo *dbinfo); + #define USEC_PER_SEC 1000000 #define WAIT_INTERVAL 1 /* 1 second */ @@ -912,6 +920,152 @@ check_publisher(const struct LogicalRepInfo *dbinfo) exit(1); } +/* + * Connect to the primary to obtain a list of subscriptions and disable them. + * They will be enabled again on the primary and dropped on the converted node. + */ +static void +obtain_and_disable_pre_existing_subscriptions(struct LogicalRepInfo *dbinfo) +{ + PQExpBuffer query = createPQExpBuffer(); + + for (int i = 0; i < num_dbs; i++) + { + PGconn *conn; + PGresult *res; + int ntups; + + /* Connect to publisher */ + conn = connect_database(dbinfo[i].pubconninfo, true); + + appendPQExpBuffer(query, + "SELECT s.subname, s.subslotname FROM pg_catalog.pg_subscription s " + "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) " + "WHERE d.datname = '%s'", + dbinfo[i].dbname); + + res = PQexec(conn, query->data); + ntups = PQntuples(res); + + dbinfo[i].num_subscriptions = ntups; + + if (ntups > 0) + { + dbinfo[i].pre_subnames = pg_malloc_array(char *, ntups); + dbinfo[i].pre_slotnames = pg_malloc_array(char *, ntups); + + for (int j = 0; j < ntups; j++) + { + /* + * Store name of subscriptions and specified slots. They are + * used to enable them again on the primary node. + */ + dbinfo[i].pre_subnames[j] = pg_strdup(PQgetvalue(res, j, 0)); + dbinfo[i].pre_slotnames[j] = pg_strdup(PQgetvalue(res, j, 1)); + + disable_subscription(conn, dbinfo[i].pre_subnames[j]); + } + } + + resetPQExpBuffer(query); + PQclear(res); + disconnect_database(conn, false); + } + + destroyPQExpBuffer(query); +} + +/* + * Disable the given subscription. After that, the subscription will be altered + * to set slot_name = NONE. This is needed for the standby server to allow + * subscription drops. + */ +static void +disable_subscription(PGconn *conn, const char *subname) +{ + PQExpBuffer query = createPQExpBuffer(); + + Assert(conn != NULL); + + /* Disable the given subscription */ + appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s DISABLE", subname); + PQclear(PQexec(conn, query->data)); + resetPQExpBuffer(query); + + /* ...and alter the slot_name to NONE */ + appendPQExpBuffer(query, + "ALTER SUBSCRIPTION %s SET (slot_name = NONE)", + subname); + PQclear(PQexec(conn, query->data)); + + destroyPQExpBuffer(query); +} + +/* + * Enable pre-existing subscriptions on the primary + */ +static void +enable_subscirptions_on_publisher(struct LogicalRepInfo *dbinfo) +{ + PQExpBuffer query = createPQExpBuffer(); + + for (int i = 0; i < num_dbs; i++) + { + PGconn *conn; + struct LogicalRepInfo info = dbinfo[i]; + + /* Connect to publisher */ + conn = connect_database(info.pubconninfo, false); + + for (int j = 0; j < info.num_subscriptions; j++) + { + /* Restore the slot_name parameter */ + appendPQExpBuffer(query, + "ALTER SUBSCRIPTION %s SET (slot_name = %s)", + info.pre_subnames[j], info.pre_slotnames[j]); + PQclear(PQexec(conn, query->data)); + resetPQExpBuffer(query); + + /* ...and them enable the subscription */ + appendPQExpBuffer(query, + "ALTER SUBSCRIPTION %s ENABLE", + info.pre_subnames[j]); + PQclear(PQexec(conn, query->data)); + resetPQExpBuffer(query); + } + } + + destroyPQExpBuffer(query); +} + +/* + * Drop pre-existing subscriptions on the standby + */ +static void +drop_pre_existing_subscriptions(struct LogicalRepInfo *dbinfo) +{ + PQExpBuffer query = createPQExpBuffer(); + + for (int i = 0; i < num_dbs; i++) + { + PGconn *conn; + struct LogicalRepInfo info = dbinfo[i]; + + /* Connect to subscriber */ + conn = connect_database(info.subconninfo, false); + + for (int j = 0; j < info.num_subscriptions; j++) + { + appendPQExpBuffer(query, + "DROP SUBSCRIPTION %s;", info.pre_subnames[j]); + PQexec(conn, query->data); + resetPQExpBuffer(query); + } + } + + destroyPQExpBuffer(query); +} + /* * Is the standby server ready for logical replication? * @@ -924,7 +1078,7 @@ check_publisher(const struct LogicalRepInfo *dbinfo) * will be broken at the end of this process (due to pg_resetwal). */ static void -check_subscriber(const struct LogicalRepInfo *dbinfo) +check_subscriber(struct LogicalRepInfo *dbinfo) { PGconn *conn; PGresult *res; @@ -2067,6 +2221,9 @@ main(int argc, char **argv) /* Check if the primary server is ready for logical replication */ check_publisher(dbinfo); + /* Check the existence of pre-existing subscriptions, and disable once */ + obtain_and_disable_pre_existing_subscriptions(dbinfo); + /* * Stop the target server. The recovery process requires that the server * reaches a consistent state before targeting the recovery stop point. @@ -2109,9 +2266,15 @@ main(int argc, char **argv) /* Remove primary_slot_name if it exists on primary */ drop_primary_replication_slot(dbinfo, primary_slot_name); + /* Enable pre-subscriptions on primary */ + enable_subscirptions_on_publisher(dbinfo); + /* Remove failover replication slots if they exist on subscriber */ drop_failover_replication_slots(dbinfo); + /* Drop pre-subscriptions on standby */ + drop_pre_existing_subscriptions(dbinfo); + /* Stop the subscriber */ pg_log_info("stopping the subscriber"); stop_standby_server(subscriber_dir); diff --git a/src/bin/pg_basebackup/t/041_pg_createsubscriber_added.pl b/src/bin/pg_basebackup/t/041_pg_createsubscriber_added.pl new file mode 100644 index 0000000000..59cddb2fc1 --- /dev/null +++ b/src/bin/pg_basebackup/t/041_pg_createsubscriber_added.pl @@ -0,0 +1,118 @@ +# Copyright (c) 2024, PostgreSQL Global Development Group + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Construct a cascading replication system like: +# +# node_a --(logical replication)--> node_b --(streaming replication)--> node_c +# + +# Set up node A as publisher +my $node_a = PostgreSQL::Test::Cluster->new('node_a'); +my $aconnstr = $node_a->connstr; +$node_a->init(allows_streaming => 'logical'); +$node_a->start; + +# On node A +# - create databases +# - create test tables +# - insert a row +# - create publications +$node_a->safe_psql( + 'postgres', q( + CREATE DATABASE pg1; + CREATE DATABASE pg2; +)); +$node_a->safe_psql('pg1', 'CREATE TABLE tbl1 (a text)'); +$node_a->safe_psql('pg1', "INSERT INTO tbl1 VALUES('first row')"); +$node_a->safe_psql('pg2', 'CREATE TABLE tbl2 (a text)'); +$node_a->safe_psql('pg1', 'CREATE PUBLICATION pub1_pg1 FOR ALL TABLES'); +$node_a->safe_psql('pg1', 'CREATE PUBLICATION pub2_pg1'); +$node_a->safe_psql('pg2', 'CREATE PUBLICATION pub1_pg2 FOR ALL TABLES'); +$node_a->safe_psql('pg2', 'CREATE PUBLICATION pub2_pg2'); + +# Set up node B as subscriber/primary +my $node_b = PostgreSQL::Test::Cluster->new('node_b'); +my $bconnstr = $node_b->connstr; +$node_b->init(allows_streaming => 'logical'); +$node_b->start; + +# On node B +# - create databases +# - create subscriptions +$node_b->safe_psql( + 'postgres', q( + CREATE DATABASE pg1; + CREATE DATABASE pg2; +)); +$node_b->safe_psql('pg1', 'CREATE TABLE tbl1 (a text)'); +$node_b->safe_psql('pg2', 'CREATE TABLE tbl2 (a text)'); +$node_b->safe_psql('pg1', + "CREATE SUBSCRIPTION sub1_pg1 CONNECTION '$aconnstr dbname=pg1' PUBLICATION pub1_pg1"); +$node_b->safe_psql('pg1', + "CREATE SUBSCRIPTION sub2_pg1 CONNECTION '$aconnstr dbname=pg1' PUBLICATION pub2_pg1"); +$node_b->safe_psql('pg2', + "CREATE SUBSCRIPTION sub1_pg2 CONNECTION '$aconnstr dbname=pg2' PUBLICATION pub1_pg2"); +$node_b->safe_psql('pg2', + "CREATE SUBSCRIPTION sub2_pg2 CONNECTION '$aconnstr dbname=pg2' PUBLICATION pub2_pg2"); + +$node_b->wait_for_subscription_sync($node_a, 'sub1_pg1'); +$node_b->wait_for_subscription_sync($node_a, 'sub1_pg2'); + +# Confirms logical replication works well +my $result = $node_b->safe_psql('pg1', 'SELECT * FROM tbl1;'); +is($result, 'first row', 'check logical replication works well'); + +# Set up node C as standby +$node_b->backup('backup_1'); +my $node_c = PostgreSQL::Test::Cluster->new('node_c'); +$node_c->init_from_backup($node_b, 'backup_1', has_streaming => 1); +$node_c->append_conf( + 'postgresql.conf', qq[ +primary_conninfo = '$bconnstr' +]); +$node_c->set_standby_mode(); +$node_c->start; + +$node_b->wait_for_replay_catchup($node_c); + +# Confirms streaming replication works well +$result = $node_c->safe_psql('pg1', 'SELECT * FROM tbl1;'); +is($result, 'first row', 'check streaming replication works well'); + +$node_c->stop; + +# Run pg_createsubscriber +command_ok( + [ + 'pg_createsubscriber', '--verbose', + '--recovery-timeout', "$PostgreSQL::Test::Utils::timeout_default", + '--verbose', + '--pgdata', $node_c->data_dir, + '--publisher-server', $bconnstr, + '--socket-directory', $node_c->host, + '--subscriber-port', $node_c->port, + '--database', 'pg1', + '--database', 'pg2' + ], + 'run pg_createsubscriber on node S'); + +# Confirms pre-existing subscriptions are removed from the converted node +$node_c->start; +$result = $node_c->safe_psql('pg1', + "SELECT subname FROM pg_subscription WHERE subname NOT LIKE 'pg_createsubscriber%';"); +is($result, '', 'check subscriptions are removed'); + +# Confirms pre-existing subscriptions still exist on the primary node +$result = $node_b->safe_psql('pg1', + "SELECT subname, subenabled, subslotname FROM pg_subscription WHERE subname NOT LIKE 'pg_createsubscriber%' ORDER BY subname;"); +is($result, 'sub1_pg1|t|sub1_pg1 +sub1_pg2|t|sub1_pg2 +sub2_pg1|t|sub2_pg1 +sub2_pg2|t|sub2_pg2', 'check subscriptions still exist'); + +done_testing(); -- 2.43.0