Repository: zalando/patroni
Pull request #2903 (2023-10-10 07:21:03)

Set citus.local_hostname
There are cases when Citus wants to have a connection to the local postgres. By default it uses `localhost` for that, which is not alwasy available. To solve it we will set `citus.local_hostname` GUC to custom value, which is the same as Patroni uses to connect to Postgres.

coveralls: ## Pull Request Test Coverage Report for [Build 6466025019](https://coveralls.io/builds/63187082)

* **1** of **1**   **(100.0%)**  changed or added relevant line in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.863%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/63187082/badge)](https://coveralls.io/builds/63187082) |
| :-- | --: |
| Change from base [Build 6454323774](https://coveralls.io/builds/63164371): |  0.0% |
| Covered Lines: | 13144 |
| Relevant Lines: | 13162 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/docs/citus.rst b/docs/citus.rst
index 084931df..62cbdd1b 100644
--- a/docs/citus.rst
+++ b/docs/citus.rst
@@ -38,14 +38,18 @@ After that you just need to start Patroni and it will handle the rest:
 2. If ``max_prepared_transactions`` isn't explicitly set in the global
    :ref:`dynamic configuration <dynamic_configuration>` Patroni will
    automatically set it to ``2*max_connections``.
-3. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``.
-4. Current superuser :ref:`credentials <postgresql_settings>` will be added to the ``pg_dist_authinfo``
+3. The ``citus.local_hostname`` GUC value will be adjusted from ``localhost`` to the
+   value that Patroni is using in order to connect to the local PostgreSQL
+   instance. The value sometimes should be different from the ``localhost``
+   because PostgreSQL might be not listening on it.
+4. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``.
+5. Current superuser :ref:`credentials <postgresql_settings>` will be added to the ``pg_dist_authinfo``
    table to allow cross-node communication. Don't forget to update them if
    later you decide to change superuser username/password/sslcert/sslkey!
-5. The coordinator primary node will automatically discover worker primary
+6. The coordinator primary node will automatically discover worker primary
    nodes and add them to the ``pg_dist_node`` table using the
    ``citus_add_node()`` function.
-6. Patroni will also maintain ``pg_dist_node`` in case failover/switchover
+7. Patroni will also maintain ``pg_dist_node`` in case failover/switchover
    on the coordinator or worker clusters occurs.
 
 patronictl
diff --git a/docs/patroni_configuration.rst b/docs/patroni_configuration.rst
index 3978e48d..e18c118e 100644
--- a/docs/patroni_configuration.rst
+++ b/docs/patroni_configuration.rst
@@ -70,15 +70,16 @@ There also are some parameters like **postgresql.listen**, **postgresql.data_dir
 
 When applying the local or dynamic configuration options, the following actions are taken:
 
-- The node first checks if there is a `postgresql.base.conf` or if the ``custom_conf`` parameter is set.
-- If the ``custom_conf`` parameter is set, it will take the file specified on it as a base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`.
-- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and it will be used as a base configuration.
-- If there is no ``custom_conf``` nor `postgresql.base.conf`, the original `postgresql.conf`` is taken and renamed to postgresql.base.conf.
-- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf`` and an include is set in
-  postgresql.conf to the used base configuration (either `postgresql.base.conf` or what is on ``custom_conf``). Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present not.
+- The node first checks if there is a `postgresql.base.conf` file or if the ``custom_conf`` parameter is set.
+- If the ``custom_conf`` parameter is set, the file it specifies is used as the base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`.
+- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and is used as the base configuration.
+- If there is no ``custom_conf`` nor `postgresql.base.conf`, the original `postgresql.conf` is renamed to `postgresql.base.conf` and used as the base configuration.
+- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf` and an include is set in
+  `postgresql.conf` to the base configuration (either `postgresql.base.conf` or the file at ``custom_conf``).
+  Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present or not.
 - Some parameters that are essential for Patroni to manage the cluster are overridden using the command line.
-- If some of the options that require restart are changed (we should look at the context in pg_settings and at the actual
-  values of those options), a pending_restart flag of a given node is set. This flag is reset on any restart.
+- If an option that requires restart is changed (we should look at the context in pg_settings and at the actual
+  values of those options), a pending_restart flag is set on that node. This flag is reset on any restart.
 
 The parameters would be applied in the following order (run-time are given the highest priority):
 
diff --git a/features/environment.py b/features/environment.py
index 79db5d16..1f36eb0c 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -245,6 +245,10 @@ class PatroniController(AbstractController):
             self.recursive_update(config, custom_config)
 
         self.recursive_update(config, {
+            'log': {
+                'format': '%(asctime)s %(levelname)s [%(pathname)s:%(lineno)d - %(funcName)s]: %(message)s',
+                'loggers': {'patroni.postgresql.callback_executor': 'DEBUG'}
+            },
             'bootstrap': {
                 'dcs': {
                     'loop_wait': 2,
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index ac6bbb85..a9f00c01 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -54,21 +54,18 @@ Feature: standby cluster
     And postgres1 does not have a logical replication slot named test_logical
 
   Scenario: check switchover
-    When I run patronictl.py switchover batman1 --force
-    And I issue a GET request to http://127.0.0.1:8010/standby_leader
-    Then I receive a response code 200
-    And I receive a response role standby_leader
+    Given I run patronictl.py switchover batman1 --force
+    Then Status code on GET http://127.0.0.1:8010/standby_leader is 200 after 10 seconds
     And postgres1 is replicating from postgres2 after 32 seconds
+    And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory
 
   Scenario: check failover
     When I kill postgres2
     And I kill postmaster on postgres2
     Then postgres1 is replicating from postgres0 after 32 seconds
+    And Status code on GET http://127.0.0.1:8009/standby_leader is 200 after 10 seconds
     When I issue a GET request to http://127.0.0.1:8009/primary
     Then I receive a response code 503
-    And I sleep for 3 seconds
-    When I issue a GET request to http://127.0.0.1:8009/standby_leader
-    Then I receive a response code 200
     And I receive a response role standby_leader
     And replication works from postgres0 to postgres1 after 15 seconds
-    And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
+    And there is a postgres1_cb.log with "on_role_change replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 08112651..d71d41ad 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -158,7 +158,7 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
     """Attempt to load a Distributed Configuration Store from known available implementations.
 
     .. note::
-        Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and
+        Using the list of available DCS classes returned by :func:`iter_dcs_classes` attempt to dynamically
         instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`.
 
         Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 863d7b40..649c1ac5 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -334,6 +334,7 @@ class ZooKeeper(AbstractDCS):
         try:
             self._client.retry(self._client.create, self.leader_path, self._name.encode('utf-8'),
                                makepath=True, ephemeral=True)
+            self.cluster_watcher(None)  # the next _load_cluster() call must read from ZooKeeper.
             return True
         except (ConnectionClosedError, RetryFailedError) as e:
             raise ZooKeeperError(e)
diff --git a/patroni/ha.py b/patroni/ha.py
index 3b936e4b..877e24ef 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -584,7 +584,9 @@ class Ha(object):
         """
         # The standby leader or when there is no standby leader we want to follow
         # the remote member, except when there is no standby leader in pause.
-        if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
+        if self.is_standby_cluster() \
+                and (cluster.leader and cluster.leader.name and cluster.leader.name == self.state_handler.name
+                     or cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()
         # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
diff --git a/patroni/postgresql/callback_executor.py b/patroni/postgresql/callback_executor.py
index 3ae073fd..06b9f353 100644
--- a/patroni/postgresql/callback_executor.py
+++ b/patroni/postgresql/callback_executor.py
@@ -1,8 +1,9 @@
 import logging
+import sys
 
 from enum import Enum
 from threading import Condition, Thread
-from typing import List
+from typing import Any, Dict, List
 
 from .cancellable import CancellableExecutor, CancellableSubprocess
 
@@ -30,7 +31,9 @@ class OnReloadExecutor(CancellableSubprocess):
         self.cancel(kill=True)
         self._kill_children()
         with self._lock:
-            self._start_process(cmd, close_fds=True)
+            started = self._start_process(cmd, close_fds=True)
+        if started and self._process is not None:
+            Thread(target=self._process.wait).start()
 
 
 class CallbackExecutor(CancellableExecutor, Thread):
@@ -51,6 +54,8 @@ class CallbackExecutor(CancellableExecutor, Thread):
         If it couldn't be killed we wait until it finishes.
 
         :param cmd: command to be executed"""
+        kwargs: Dict[str, Any] = {'stacklevel': 3} if sys.version_info >= (3, 8) else {}
+        logger.debug('CallbackExecutor.call(%s)', cmd, **kwargs)
 
         if cmd[-3] == CallbackAction.ON_RELOAD:
             return self._on_reload_executor.call_nowait(cmd)
diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py
index 8ca2790e..26923f37 100644
--- a/patroni/postgresql/citus.py
+++ b/patroni/postgresql/citus.py
@@ -403,6 +403,9 @@ class CitusHandler(Thread):
         # Resharding in Citus implemented using logical replication
         parameters['wal_level'] = 'logical'
 
+        # Sometimes Citus needs to connect to the local postgres. We will do it the same way as Patroni does.
+        parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost')
+
     def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
         if isinstance(self._config, dict) and self._postgresql.is_primary() and\
                 slot['type'] == 'logical' and slot['database'] == self._config['database']:


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index 32172926..bd70ba3d 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -241,7 +241,7 @@ class PostgresInit(unittest.TestCase):
                                                 'replication': {'username': '', 'password': 'rep-pass'},
                                                 'rewind': {'username': 'rewind', 'password': 'test'}},
                              'remove_data_directory_on_rewind_failure': True,
-                             'use_pg_rewind': True, 'pg_ctl_timeout': 'bla',
+                             'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True,
                              'parameters': self._PARAMETERS,
                              'recovery_conf': {'foo': 'bar'},
                              'pg_hba': ['host all all 0.0.0.0/0 md5'],
diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
index 9ac6aa68..4c2d1c98 100644
--- a/tests/test_bootstrap.py
+++ b/tests/test_bootstrap.py
@@ -256,7 +256,7 @@ class TestBootstrap(BaseTestPostgresql):
         mock_cancellable_subprocess_call.assert_called()
         args, kwargs = mock_cancellable_subprocess_call.call_args
         self.assertTrue('PGPASSFILE' in kwargs['env'])
-        self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=127.0.0.2 port=5432'])
+        self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=/tmp port=5432'])
 
         mock_cancellable_subprocess_call.reset_mock()
         self.p.connection_pool._conn_kwargs.pop('host')
diff --git a/tests/test_callback_executor.py b/tests/test_callback_executor.py
index df2556b9..51c915d9 100644
--- a/tests/test_callback_executor.py
+++ b/tests/test_callback_executor.py
@@ -35,5 +35,6 @@ class TestCallbackExecutor(unittest.TestCase):
         ce._invoke_excepthook = Mock()
         self.assertIsNone(ce.call(callback))
 
+        mock_popen.side_effect = [Mock()]
         self.assertIsNone(ce.call(['test.sh', 'on_reload', 'replica', 'foo']))
         ce.join()
diff --git a/tests/test_citus.py b/tests/test_citus.py
index 9849a069..7279893e 100644
--- a/tests/test_citus.py
+++ b/tests/test_citus.py
@@ -13,7 +13,6 @@ class TestCitus(BaseTestPostgresql):
     def setUp(self):
         super(TestCitus, self).setUp()
         self.c = self.p.citus_handler
-        self.p.connection_pool.conn_kwargs = {'host': 'localhost', 'dbname': 'postgres'}
         self.cluster = get_cluster_initialized_with_leader()
         self.cluster.workers[1] = self.cluster
 
@@ -139,6 +138,7 @@ class TestCitus(BaseTestPostgresql):
         self.assertEqual(parameters['max_prepared_transactions'], 202)
         self.assertEqual(parameters['shared_preload_libraries'], 'citus,foo,bar')
         self.assertEqual(parameters['wal_level'], 'logical')
+        self.assertEqual(parameters['citus.local_hostname'], '/tmp')
 
     def test_bootstrap(self):
         self.c._config = None