From 95ba8b9e596ecaf0898f08d42fcbf6e073c93830 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 13 Mar 2023 13:30:39 +0100 Subject: [PATCH] Fix bug with metadata after coordinator failover (#2597) We made incorrect assumption that `citus_set_coordinator_host()` will trigger `pg_dist_node` sync. Instead we should also use `citus_update_node()` and call `citus_set_coordinator_host()` only during the bootstrap. Adjust behave tests to verify that coordinator failover is visible on workers. --- features/citus.feature | 16 ++++++++-------- features/steps/citus.py | 2 +- patroni/postgresql/citus.py | 18 +++++++++--------- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/features/citus.feature b/features/citus.feature index a7f53ef0..2cf295eb 100644 --- a/features/citus.feature +++ b/features/citus.feature @@ -10,20 +10,20 @@ Feature: citus And I start postgres3 in citus group 1 Then replication works from postgres0 to postgres1 after 15 seconds Then replication works from postgres2 to postgres3 after 15 seconds - And postgres0 is registered in the coordinator postgres0 as the worker in group 0 - And postgres2 is registered in the coordinator postgres0 as the worker in group 1 + And postgres0 is registered in the postgres0 as the worker in group 0 + And postgres2 is registered in the postgres0 as the worker in group 1 Scenario: coordinator failover updates pg_dist_node Given I run patronictl.py failover batman --group 0 --candidate postgres1 --force Then postgres1 role is the primary after 10 seconds And replication works from postgres1 to postgres0 after 15 seconds And "sync" key in a group 0 in DCS has sync_standby=postgres0 after 15 seconds - And postgres1 is registered in the coordinator postgres1 as the worker in group 0 + And postgres1 is registered in the postgres2 as the worker in group 0 When I run patronictl.py failover batman --group 0 --candidate postgres0 --force Then postgres0 role is the primary after 10 seconds And replication works from postgres0 to postgres1 after 15 seconds And "sync" key in a group 0 in DCS has sync_standby=postgres1 after 15 seconds - And postgres0 is registered in the coordinator postgres0 as the worker in group 0 + And postgres0 is registered in the postgres2 as the worker in group 0 Scenario: worker switchover doesn't break client queries on the coordinator Given I create a distributed table on postgres0 @@ -33,14 +33,14 @@ Feature: citus And postgres3 role is the primary after 10 seconds And replication works from postgres3 to postgres2 after 15 seconds And "sync" key in a group 1 in DCS has sync_standby=postgres2 after 15 seconds - And postgres3 is registered in the coordinator postgres0 as the worker in group 1 + And postgres3 is registered in the postgres0 as the worker in group 1 And a thread is still alive When I run patronictl.py switchover batman --group 1 --force Then I receive a response returncode 0 And postgres2 role is the primary after 10 seconds And replication works from postgres2 to postgres3 after 15 seconds And "sync" key in a group 1 in DCS has sync_standby=postgres3 after 15 seconds - And postgres2 is registered in the coordinator postgres0 as the worker in group 1 + And postgres2 is registered in the postgres0 as the worker in group 1 And a thread is still alive When I stop a thread Then a distributed table on postgres0 has expected rows @@ -52,7 +52,7 @@ Feature: citus Then I receive a response returncode 0 And postgres2 role is the primary after 10 seconds And replication works from postgres2 to postgres3 after 15 seconds - And postgres2 is registered in the coordinator postgres0 as the worker in group 1 + And postgres2 is registered in the postgres0 as the worker in group 1 And a thread is still alive When I stop a thread Then a distributed table on postgres0 has expected rows @@ -65,7 +65,7 @@ Feature: citus Then I receive a response returncode 0 And I receive a response output "+ttl: 20" When I sleep for 2 seconds - Then postgres4 is registered in the coordinator postgres0 as the worker in group 2 + Then postgres4 is registered in the postgres2 as the worker in group 2 When I shut down postgres4 Then There is a transaction in progress on postgres0 changing pg_dist_node When I run patronictl.py restart batman postgres2 --group 1 --force diff --git a/features/steps/citus.py b/features/steps/citus.py index 5144f466..01f5ef2f 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -44,7 +44,7 @@ def start_citus(context, name, group): return context.pctl.start(name, custom_config={"citus": {"database": "postgres", "group": int(group)}}) -@step('{name1:w} is registered in the coordinator {name2:w} as the worker in group {group:d}') +@step('{name1:w} is registered in the {name2:w} as the worker in group {group:d}') def check_registration(context, name1, name2, group): worker_port = int(context.pctl.query(name1, "SHOW port").fetchone()[0]) r = context.pctl.query(name2, "SELECT nodeport FROM pg_catalog.pg_dist_node WHERE groupid = {0}".format(group)) diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index 0866eb9d..69fa96a7 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -196,17 +196,12 @@ class CitusHandler(Thread): return i, task def update_node(self, task): - if task.group == CITUS_COORDINATOR_GROUP_ID: - return self.query("SELECT pg_catalog.citus_set_coordinator_host(%s, %s, 'primary', 'default')", - task.host, task.port) - - if task.nodeid is None and task.event != 'before_demote': - task.nodeid = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, 'primary', 'default')", - task.host, task.port, task.group).fetchone()[0] - elif task.nodeid is not None: - # XXX: statement_timeout? + if task.nodeid is not None: self.query('SELECT pg_catalog.citus_update_node(%s, %s, %s, true, %s)', task.nodeid, task.host, task.port, task.cooldown) + elif task.event != 'before_demote': + task.nodeid = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, 'primary', 'default')", + task.host, task.port, task.group).fetchone()[0] def process_task(self, task): """Updates a single row in `pg_dist_node` table, optionally in a transaction. @@ -361,6 +356,11 @@ class CitusHandler(Thread): cur.execute("INSERT INTO pg_catalog.pg_dist_authinfo VALUES" "(0, pg_catalog.current_user(), %s)", (self._postgresql.config.format_dsn(params),)) + + if self.is_coordinator(): + r = urlparse(self._postgresql.connection_string) + cur.execute("SELECT pg_catalog.citus_set_coordinator_host(%s, %s, 'primary', 'default')", + (r.hostname, r.port or 5432)) finally: conn.close()