Set citus.local_hostname (#2903)

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.

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

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']:

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'],

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')

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