From efaba9f1839d29aaeaf6d84df064700abc7ebd76 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 9 Aug 2023 14:47:53 +0200 Subject: [PATCH 1/6] Rename Postgresql.is_leader() to is_primary() (#2809) It'll help to avoid confusion with the Ha.is_leader() method. --- patroni/ha.py | 24 +++++------ patroni/postgresql/__init__.py | 20 ++++----- patroni/postgresql/citus.py | 2 +- patroni/postgresql/rewind.py | 2 +- patroni/postgresql/slots.py | 2 +- patroni/postgresql/sync.py | 2 +- tests/test_ha.py | 77 ++++++++++++++++------------------ tests/test_postgresql.py | 8 ++-- tests/test_slots.py | 10 ++--- 9 files changed, 72 insertions(+), 75 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index fae3535e..886dcc8e 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -572,7 +572,7 @@ class Ha(object): if refresh: self.load_cluster_from_dcs() - is_leader = self.state_handler.is_leader() + is_leader = self.state_handler.is_primary() node_to_follow = self._get_node_to_follow(self.cluster) @@ -751,7 +751,7 @@ class Ha(object): """ if not self.is_paused(): if not self.watchdog.is_running and not self.watchdog.activate(): - if self.state_handler.is_leader(): + if self.state_handler.is_primary(): self.demote('immediate') return 'Demoting self because watchdog could not be activated' else: @@ -767,7 +767,7 @@ class Ha(object): self._async_response.reset() return 'Promotion cancelled because the pre-promote script failed' - if self.state_handler.is_leader(): + if self.state_handler.is_primary(): # Inform the state handler about its primary role. # It may be unaware of it if postgres is promoted manually. self.state_handler.set_role('master') @@ -959,7 +959,7 @@ class Ha(object): # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely # In order to avoid attempts to delete this key from all nodes only the primary is allowed to do it. if not self.cluster.get_member(failover.candidate, fallback_to_leader=False)\ - and self.state_handler.is_leader(): + and self.state_handler.is_primary(): logger.warning("manual failover: removing failover key because failover candidate is not running") self.dcs.manual_failover('', '', version=failover.version) return None @@ -1017,7 +1017,7 @@ class Ha(object): if ret is not None: # continue if we just deleted the stale failover key as a leader return ret - if self.state_handler.is_leader(): + if self.state_handler.is_primary(): if self.is_paused(): # in pause leader is the healthiest only when no initialize or sysid matches with initialize! return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize @@ -1206,7 +1206,7 @@ class Ha(object): :returns: action message if demote was initiated, None if no action was taken""" failover = self.cluster.failover - if not failover or (self.is_paused() and not self.state_handler.is_leader()): + if not failover or (self.is_paused() and not self.state_handler.is_primary()): return if (failover.scheduled_at and not @@ -1276,7 +1276,7 @@ class Ha(object): def process_healthy_cluster(self) -> str: if self.has_lock(): - if self.is_paused() and not self.state_handler.is_leader(): + if self.is_paused() and not self.state_handler.is_primary(): if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name: return 'waiting to become primary after promote...' @@ -1308,7 +1308,7 @@ class Ha(object): else: # Either there is no connection to DCS or someone else acquired the lock logger.error('failed to update leader lock') - if self.state_handler.is_leader(): + if self.state_handler.is_primary(): if self.is_paused(): return 'continue to run as primary after failing to update leader lock in DCS' self.demote('immediate-nolock') @@ -1547,7 +1547,7 @@ class Ha(object): self.cancel_initialization() if result is None: - if not self.state_handler.is_leader(): + if not self.state_handler.is_primary(): return 'waiting for end of recovery after bootstrap' self.state_handler.set_role('master') @@ -1731,7 +1731,7 @@ class Ha(object): elif self.cluster.is_unlocked() and not self.is_paused(): # "bootstrap", but data directory is not empty if not self.state_handler.cb_called and self.state_handler.is_running() \ - and not self.state_handler.is_leader(): + and not self.state_handler.is_primary(): self._join_aborted = True logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start') logger.error('Please first start Patroni on the node running as primary') @@ -1774,7 +1774,7 @@ class Ha(object): create_slots = self._sync_replication_slots(False) if not self.state_handler.cb_called: - if not is_promoting and not self.state_handler.is_leader(): + if not is_promoting and not self.state_handler.is_primary(): self._rewind.trigger_check_diverged_lsn() self.state_handler.call_nowait(CallbackAction.ON_START) @@ -1799,7 +1799,7 @@ class Ha(object): def _handle_dcs_error(self) -> str: if not self.is_paused() and self.state_handler.is_running(): - if self.state_handler.is_leader(): + if self.state_handler.is_primary(): if self.is_failsafe_mode() and self.check_failsafe_topology(): self.set_is_leader(True) self._failsafe.set_is_active(time.time()) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index e9844710..172e575e 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -120,9 +120,9 @@ class Postgresql(object): if self.is_running(): # we are "joining" already running postgres self.set_state('running') - self.set_role('master' if self.is_leader() else 'replica') + self.set_role('master' if self.is_primary() else 'replica') # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known - if self.major_version < 120000 or self.is_leader(): + if self.major_version < 120000 or self.is_primary(): self.config.write_postgresql_conf() hba_saved = self.config.replace_pg_hba() ident_saved = self.config.replace_pg_ident() @@ -474,14 +474,14 @@ class Postgresql(object): """:returns: a result set of 'SELECT * FROM pg_stat_replication'.""" return self._cluster_info_state_get('pg_stat_replication') or [] - def replication_state_from_parameters(self, is_leader: bool, receiver_state: Optional[str], + def replication_state_from_parameters(self, is_primary: bool, receiver_state: Optional[str], restore_command: Optional[str]) -> Optional[str]: """Figure out the replication state from input parameters. .. note:: This method could be only called when Postgres is up, running and queries are successfuly executed. - :is_leader: `True` is postgres is not running in recovery + :is_primary: `True` is postgres is not running in recovery :receiver_state: value from `pg_stat_get_wal_receiver.state` or None if Postgres is older than 9.6 :restore_command: value of ``restore_command`` GUC for PostgreSQL 12+ or `postgresql.recovery_conf.restore_command` if it is set in Patroni configuration @@ -490,7 +490,7 @@ class Postgresql(object): - 'streaming' if replica is streaming according to the `pg_stat_wal_receiver` view; - 'in archive recovery' if replica isn't streaming and there is a `restore_command` """ - if self._major_version >= 90600 and not is_leader: + if self._major_version >= 90600 and not is_primary: if receiver_state == 'streaming': return 'streaming' # For Postgres older than 12 we get `restore_command` from Patroni config, otherwise we check GUC @@ -505,11 +505,11 @@ class Postgresql(object): :returns: ``streaming``, ``in archive recovery``, or ``None`` """ - return self.replication_state_from_parameters(self.is_leader(), + return self.replication_state_from_parameters(self.is_primary(), self._cluster_info_state_get('receiver_state'), self._cluster_info_state_get('restore_command')) - def is_leader(self) -> bool: + def is_primary(self) -> bool: try: return bool(self._cluster_info_state_get('timeline')) except PostgresConnectionException: @@ -1157,9 +1157,9 @@ class Postgresql(object): return ret @staticmethod - def _wal_position(is_leader: bool, wal_position: int, + def _wal_position(is_primary: bool, wal_position: int, received_location: Optional[int], replayed_location: Optional[int]) -> int: - return wal_position if is_leader else max(received_location or 0, replayed_location or 0) + return wal_position if is_primary else max(received_location or 0, replayed_location or 0) def timeline_wal_position(self) -> Tuple[int, int, Optional[int]]: # This method could be called from different threads (simultaneously with some other `_query` calls). @@ -1195,7 +1195,7 @@ class Postgresql(object): return None def last_operation(self) -> int: - return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position') or 0, + return self._wal_position(self.is_primary(), self._cluster_info_state_get('wal_position') or 0, self.received_location(), self.replayed_location()) def configure_server_parameters(self) -> None: diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index e11c206c..b5c90cf5 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -412,7 +412,7 @@ class CitusHandler(Thread): parameters['wal_level'] = 'logical' def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: - if isinstance(self._config, dict) and self._postgresql.is_leader() and\ + if isinstance(self._config, dict) and self._postgresql.is_primary() and\ slot['type'] == 'logical' and slot['database'] == self._config['database']: m = CITUS_SLOT_NAME_RE.match(slot['name']) return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin']) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 270d629c..73bccb44 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -280,7 +280,7 @@ class Rewind(object): """After promote issue a CHECKPOINT from a new thread and asynchronously check the result. In case if CHECKPOINT failed, just check that timeline in pg_control was updated.""" - if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_leader(): + if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_primary(): with self._checkpoint_task_lock: if self._checkpoint_task: with self._checkpoint_task: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 4abda4b1..4df27eaf 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -508,7 +508,7 @@ class SlotsHandler: self._ensure_physical_slots(slots) - if self._postgresql.is_leader(): + if self._postgresql.is_primary(): self._logical_slots_processing_queue.clear() self._ensure_logical_slots_primary(slots) elif cluster.slots and slots: diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 2280a68c..3193a7e7 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -338,7 +338,7 @@ END;$$""") sync_param = next(iter(sync), None) if not (self._postgresql.config.set_synchronous_standby_names(sync_param) - and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk: + and self._postgresql.state == 'running' and self._postgresql.is_primary()) or has_asterisk: return time.sleep(0.1) # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms diff --git a/tests/test_ha.py b/tests/test_ha.py index 6eb38204..f7378b43 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -162,7 +162,7 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster())) -@patch.object(Postgresql, 'is_leader', Mock(return_value=True)) +@patch.object(Postgresql, 'is_primary', Mock(return_value=True)) @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1))) @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @@ -224,7 +224,7 @@ class TestHa(PostgresInit): @patch.object(Postgresql, 'received_timeline', Mock(return_value=None)) def test_touch_member(self): self.p._major_version = 110000 - self.p.is_leader = false + self.p.is_primary = false self.p.timeline_wal_position = Mock(return_value=(0, 1, 0)) self.p.replica_cached_timeline = Mock(side_effect=Exception) with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')): @@ -320,7 +320,7 @@ class TestHa(PostgresInit): @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)) @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True)) def test_crash_recovery_before_rewind(self): - self.p.is_leader = false + self.p.is_primary = false self.p.is_running = false self.p.controldata = lambda: {'Database cluster state': 'in archive recovery', 'Database system identifier': SYSID} @@ -365,7 +365,7 @@ class TestHa(PostgresInit): @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_start_as_readonly(self): - self.p.is_leader = false + self.p.is_primary = false self.p.is_healthy = true self.ha.has_lock = true self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID} @@ -383,11 +383,11 @@ class TestHa(PostgresInit): def test_promoted_by_acquiring_lock(self): self.ha.is_healthiest_node = true - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') def test_promotion_cancelled_after_pre_promote_failed(self): - self.p.is_leader = false + self.p.is_primary = false self.p._pre_promote = false self.ha._is_healthiest_node = true self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') @@ -402,7 +402,7 @@ class TestHa(PostgresInit): @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_long_promote(self): self.ha.has_lock = true - self.p.is_leader = false + self.p.is_primary = false self.p.set_role('primary') self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') @@ -413,7 +413,7 @@ class TestHa(PostgresInit): def test_follow_new_leader_after_failing_to_obtain_lock(self): self.ha.is_healthiest_node = true self.ha.acquire_lock = false - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock') def test_demote_because_not_healthiest(self): @@ -422,21 +422,20 @@ class TestHa(PostgresInit): def test_follow_new_leader_because_not_healthiest(self): self.ha.is_healthiest_node = false - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_promote_because_have_lock(self): self.ha.has_lock = true - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because I had the session lock') def test_promote_without_watchdog(self): self.ha.has_lock = true - self.p.is_leader = true with patch.object(Watchdog, 'activate', Mock(return_value=False)): self.assertEqual(self.ha.run_cycle(), 'Demoting self because watchdog could not be activated') - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'Not promoting self because watchdog could not be activated') def test_leader_with_lock(self): @@ -462,12 +461,12 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS') with patch.object(Ha, '_get_node_to_follow', Mock(side_effect=DCSError('foo'))): self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS') - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS') @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_follow(self): - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()') self.ha.patroni.replicatefrom = "foo" self.p.config.check_recovery_conf = Mock(return_value=(True, False)) @@ -484,13 +483,13 @@ class TestHa(PostgresInit): def test_follow_in_pause(self): self.ha.is_paused = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock') - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0)') @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)) @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True)) def test_follow_triggers_rewind(self): - self.p.is_leader = false + self.p.is_primary = false self.ha._rewind.trigger_check_diverged_lsn() self.ha.cluster = get_cluster_initialized_with_leader() self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader') @@ -544,7 +543,7 @@ class TestHa(PostgresInit): self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}}) - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible') def test_no_dcs_connection_replica_failsafe_not_enabled_but_active(self): @@ -552,7 +551,7 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_initialized_with_leader() self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}}) - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible') def test_update_failsafe(self): @@ -591,9 +590,9 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_not_initialized_without_leader() self.e.initialize = true self.assertEqual(self.ha.bootstrap(), 'trying to bootstrap a new cluster') - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap') - self.p.is_leader = true + self.p.is_primary = true self.ha.is_synchronous_mode = true self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap') self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster') @@ -613,7 +612,6 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_not_initialized_without_leader() self.e.initialize = true self.ha.bootstrap() - self.p.is_leader = true with patch.object(Watchdog, 'activate', Mock(return_value=False)), \ patch('patroni.ha.logger.error') as mock_logger: self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap') @@ -745,7 +743,6 @@ class TestHa(PostgresInit): self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle()) def test_manual_failover_from_leader_in_synchronous_mode(self): - self.p.is_leader = true self.ha.has_lock = true self.ha.is_synchronous_mode = true self.ha.process_sync_replication = Mock() @@ -756,7 +753,7 @@ class TestHa(PostgresInit): self.assertEqual('manual failover: demoting myself', self.ha.run_cycle()) def test_manual_failover_process_no_leader(self): - self.p.is_leader = false + self.p.is_primary = false self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None)) self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None)) self.p.set_role('replica') @@ -780,7 +777,7 @@ class TestHa(PostgresInit): def test_manual_failover_process_no_leader_in_synchronous_mode(self): self.ha.is_synchronous_mode = true - self.p.is_leader = false + self.p.is_primary = false # switchover to a specific node, which name doesn't match our name (postgresql0) self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'other', None)) @@ -840,14 +837,14 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None)) self.assertEqual('PAUSE: acquired session lock as a leader', self.ha.run_cycle()) - self.p.is_leader = false + self.p.is_primary = false self.p.set_role('replica') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None)) self.assertEqual(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock') def test_is_healthiest_node(self): self.ha.is_failsafe_mode = true - self.ha.state_handler.is_leader = false + self.p.is_primary = false self.ha.patroni.nofailover = False self.ha.fetch_node_status = get_node_status() self.ha.dcs._last_failsafe = {'foo': ''} @@ -861,7 +858,7 @@ class TestHa(PostgresInit): self.assertFalse(self.ha.is_healthiest_node()) def test__is_healthiest_node(self): - self.p.is_leader = false + self.p.is_primary = false self.ha.cluster = get_cluster_initialized_without_leader(sync=('postgresql1', self.p.name)) self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) @@ -960,7 +957,7 @@ class TestHa(PostgresInit): self.assertTrue(self.ha.restart_matches("replica", "9.5.2", False)) def test_process_healthy_cluster_in_pause(self): - self.p.is_leader = false + self.p.is_primary = false self.ha.is_paused = true self.p.name = 'leader' self.ha.cluster = get_cluster_initialized_with_leader() @@ -971,7 +968,7 @@ class TestHa(PostgresInit): @patch('patroni.postgresql.mtime', Mock(return_value=1588316884)) @patch('builtins.open', mock_open(read_data='1\t0/40159C0\tno recovery target specified\n')) def test_process_healthy_standby_cluster_as_standby_leader(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'leader' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.p.config.check_recovery_conf = Mock(return_value=(False, False)) @@ -983,7 +980,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock') def test_process_healthy_standby_cluster_as_cascade_replica(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'replica' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.assertEqual(self.ha.run_cycle(), @@ -993,7 +990,7 @@ class TestHa(PostgresInit): @patch.object(Cluster, 'is_unlocked', Mock(return_value=True)) def test_process_unhealthy_standby_cluster_as_standby_leader(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'leader' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.ha.sysid_valid = true @@ -1003,13 +1000,13 @@ class TestHa(PostgresInit): @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)) @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True)) def test_process_unhealthy_standby_cluster_as_cascade_replica(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'replica' self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_member:')) def test_recover_unhealthy_leader_in_standby_cluster(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'leader' self.p.is_running = false self.p.follow = false @@ -1018,7 +1015,7 @@ class TestHa(PostgresInit): @patch.object(Cluster, 'is_unlocked', Mock(return_value=True)) def test_recover_unhealthy_unlocked_standby_cluster(self): - self.p.is_leader = false + self.p.is_primary = false self.p.name = 'leader' self.p.is_running = false self.p.follow = false @@ -1078,7 +1075,7 @@ class TestHa(PostgresInit): check_calls([(update_lock, True), (demote, True)]) self.ha.has_lock = false - self.p.is_leader = false + self.p.is_primary = false self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader (leader)') check_calls([(update_lock, False), (demote, False)]) @@ -1212,7 +1209,7 @@ class TestHa(PostgresInit): self.ha.is_synchronous_mode = true mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock() - self.p.is_leader = false + self.p.is_primary = false self.p.set_role('replica') self.ha.has_lock = true mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) @@ -1235,7 +1232,7 @@ class TestHa(PostgresInit): def test_unhealthy_sync_mode(self): self.ha.is_synchronous_mode = true - self.p.is_leader = false + self.p.is_primary = false self.p.set_role('replica') self.p.name = 'other' self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'other2')) @@ -1266,7 +1263,7 @@ class TestHa(PostgresInit): self.ha.is_synchronous_mode = true self.p.name = 'other' - self.p.is_leader = false + self.p.is_primary = false self.p.set_role('replica') mock_restart = self.p.restart = Mock(return_value=True) self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) @@ -1369,7 +1366,7 @@ class TestHa(PostgresInit): @patch('sys.exit', return_value=1) def test_abort_join(self, exit_mock): self.ha.cluster = get_cluster_not_initialized_without_leader() - self.p.is_leader = false + self.p.is_primary = false self.ha.run_cycle() exit_mock.assert_called_once_with(1) @@ -1429,7 +1426,7 @@ class TestHa(PostgresInit): @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['ls'])) def test_follow_copy(self): self.ha.cluster.config.data['slots'] = {'ls': {'database': 'a', 'plugin': 'b'}} - self.p.is_leader = false + self.p.is_primary = false self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots')) def test_acquire_lock(self): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a3475ef5..a703f235 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -363,11 +363,11 @@ class TestPostgresql(BaseTestPostgresql): self.assertRaises(psycopg.ProgrammingError, self.p.query, 'blabla') @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT)) - def test_is_leader(self): - self.assertTrue(self.p.is_leader()) + def test_is_primary(self): + self.assertTrue(self.p.is_primary()) self.p.reset_cluster_info_state(None) with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))): - self.assertFalse(self.p.is_leader()) + self.assertFalse(self.p.is_primary()) @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down', 'Latest checkpoint location': '0/1ADBC18', @@ -461,7 +461,7 @@ class TestPostgresql(BaseTestPostgresql): self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START)) @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster())) - def test_is_leader_exception(self): + def test_is_primary_exception(self): self.p.start() self.p.query = Mock(side_effect=psycopg.OperationalError("not supported")) self.assertTrue(self.p.stop()) diff --git a/tests/test_slots.py b/tests/test_slots.py index a5be0d1d..0210d701 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -48,7 +48,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.s.sync_replication_slots(cluster, False) mock_debug.assert_called_once() self.p.set_role('replica') - with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \ + with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \ patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop: self.s.sync_replication_slots(cluster, False, paused=True) mock_drop.assert_not_called() @@ -79,7 +79,7 @@ class TestSlotsHandler(BaseTestPostgresql): None, SyncState.empty(), None, {'ls': 10}, None) self.p.set_role('replica') with patch.object(Postgresql, '_query') as mock_query, \ - patch.object(Postgresql, 'is_leader', Mock(return_value=False)): + patch.object(Postgresql, 'is_primary', Mock(return_value=False)): mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)] ret = self.s.sync_replication_slots(cluster, False) self.assertEqual(ret, []) @@ -106,7 +106,7 @@ class TestSlotsHandler(BaseTestPostgresql): "confirmed_flush_lsn": 12345, "catalog_xmin": 105}]) self.assertEqual(self.p.slots(), {}) - @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) def test__ensure_logical_slots_replica(self): self.p.set_role('replica') self.cluster.slots['ls'] = 12346 @@ -133,7 +133,7 @@ class TestSlotsHandler(BaseTestPostgresql): @patch.object(Postgresql, 'stop', Mock(return_value=True)) @patch.object(Postgresql, 'start', Mock(return_value=True)) - @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) def test_check_logical_slots_readiness(self): self.s.copy_logical_slots(self.cluster, ['ls']) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \ @@ -147,7 +147,7 @@ class TestSlotsHandler(BaseTestPostgresql): @patch.object(Postgresql, 'stop', Mock(return_value=True)) @patch.object(Postgresql, 'start', Mock(return_value=True)) - @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) def test_on_promote(self): self.s.schedule_advance_slots({'foo': {'bar': 100}}) self.s.copy_logical_slots(self.cluster, ['ls']) From 713244975c4c888f200c09ac25021e7ce743e067 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 9 Aug 2023 14:48:18 +0200 Subject: [PATCH 2/6] Silence useless warnings in patronictl (#2808) Close https://github.com/zalando/patroni/issues/2805 --- patroni/config.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/patroni/config.py b/patroni/config.py index d2770517..78523dfc 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -224,7 +224,8 @@ class Config(object): self.__effective_configuration = self._build_effective_configuration({}, self._local_configuration) self._data_dir = self.__effective_configuration.get('postgresql', {}).get('data_dir', "") self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME) - self._load_cache() + if validator: # patronictl uses validator=None and we don't want to load anything from local cache in this case + self._load_cache() self._cache_needs_saving = False @property @@ -341,8 +342,8 @@ class Config(object): if ConfigHandler.CMDLINE_OPTIONS[name][1](value): pg_params[name] = value else: - logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s", - name, value, ConfigHandler.CMDLINE_OPTIONS[name][0]) + logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s", + name, value, ConfigHandler.CMDLINE_OPTIONS[name][0]) return pg_params From 3734ecc8510ba6a1f4903cdcd95f3855df0b6d7d Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Wed, 9 Aug 2023 17:46:53 +0200 Subject: [PATCH 3/6] Implement generate-config (#2786) New patroni.py option that allows to * generate patroni.yml configuration file with the values from a running cluster * generate a sample patroni.yml configuration file --- patroni/__main__.py | 19 +- patroni/collections.py | 9 +- patroni/config.py | 21 +- patroni/config_generator.py | 463 +++++++++++++++++++++++++++++++++ patroni/utils.py | 44 ++++ patroni/validator.py | 34 +-- tests/__init__.py | 24 ++ tests/test_config_generator.py | 334 ++++++++++++++++++++++++ 8 files changed, 912 insertions(+), 36 deletions(-) create mode 100644 patroni/config_generator.py create mode 100644 tests/test_config_generator.py diff --git a/patroni/__main__.py b/patroni/__main__.py index 42f539d1..e9dcd28a 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -163,11 +163,26 @@ def patroni_main(configfile: str) -> None: def process_arguments() -> Namespace: + from patroni.config_generator import generate_config + parser = get_base_arg_parser() - parser.add_argument('--validate-config', action='store_true', help='Run config validator and exit') + group = parser.add_mutually_exclusive_group() + group.add_argument('--validate-config', action='store_true', help='Run config validator and exit') + group.add_argument('--generate-sample-config', action='store_true', + help='Generate a sample Patroni yaml configuration file') + group.add_argument('--generate-config', action='store_true', + help='Generate a Patroni yaml configuration file for a running instance') + parser.add_argument('--dsn', help='Optional DSN string of the instance to be used as a source \ + for config generation. Superuser connection is required.') args = parser.parse_args() - if args.validate_config: + if args.generate_sample_config: + generate_config(args.configfile, True, None) + sys.exit(0) + elif args.generate_config: + generate_config(args.configfile, False, args.dsn) + sys.exit(0) + elif args.validate_config: from patroni.validator import schema from patroni.config import Config, ConfigParseError diff --git a/patroni/collections.py b/patroni/collections.py index c6d8f5d6..2d38e8d7 100644 --- a/patroni/collections.py +++ b/patroni/collections.py @@ -3,7 +3,7 @@ Provides a case insensitive :class:`dict` and :class:`set` object types. """ from collections import OrderedDict -from typing import Any, Collection, Dict, Iterator, MutableMapping, MutableSet, Optional +from typing import Any, Collection, Dict, Iterator, KeysView, MutableMapping, MutableSet, Optional class CaseInsensitiveSet(MutableSet[str]): @@ -187,6 +187,13 @@ class CaseInsensitiveDict(MutableMapping[str, Any]): """ return CaseInsensitiveDict({v[0]: v[1] for v in self._values.values()}) + def keys(self) -> KeysView[str]: + """Return a new view of the dict's keys. + + :returns: a set-like object providing a view on the dict's keys + """ + return self._values.keys() + def __repr__(self) -> str: """Get a string representation of the dict. diff --git a/patroni/config.py b/patroni/config.py index 78523dfc..4e6da10f 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -194,10 +194,9 @@ class Config(object): 'recovery_min_apply_delay': '' }, 'postgresql': { - 'bin_dir': '', 'use_slots': True, 'parameters': CaseInsensitiveDict({p: v[0] for p, v in ConfigHandler.CMDLINE_OPTIONS.items() - if p not in ('wal_keep_segments', 'wal_keep_size')}) + if v[0] is not None and p not in ('wal_keep_segments', 'wal_keep_size')}) } } @@ -236,6 +235,22 @@ class Config(object): def dynamic_configuration(self) -> Dict[str, Any]: return deepcopy(self._dynamic_configuration) + @property + def local_configuration(self) -> Dict[str, Any]: + """Deep copy of cached Patroni local configuration. + + :returns: copy of :attr:`~Config._local_configuration` + """ + return deepcopy(dict(self._local_configuration)) + + @classmethod + def get_default_config(cls) -> Dict[str, Any]: + """Deep copy default configuration. + + :returns: copy of :attr:`~Config.__DEFAULT_CONFIG` + """ + return deepcopy(cls.__DEFAULT_CONFIG) + def _load_config_path(self, path: str) -> Dict[str, Any]: """ If path is a file, loads the yml file pointed to by path. @@ -348,7 +363,7 @@ class Config(object): return pg_params def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]: - config = deepcopy(self.__DEFAULT_CONFIG) + config = self.get_default_config() for name, value in dynamic_configuration.items(): if name == 'postgresql': diff --git a/patroni/config_generator.py b/patroni/config_generator.py new file mode 100644 index 00000000..956e18d2 --- /dev/null +++ b/patroni/config_generator.py @@ -0,0 +1,463 @@ +"""patroni ``--generate-config`` machinery.""" +import abc +import logging +import os +import psutil +import socket +import sys +import yaml + +from getpass import getuser, getpass +from contextlib import contextmanager +from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union +if TYPE_CHECKING: # pragma: no cover + from psycopg import Cursor + from psycopg2 import cursor + +from . import psycopg +from .config import Config +from .exceptions import PatroniException +from .postgresql.config import ConfigHandler, parse_dsn +from .postgresql.misc import postgres_major_version_to_int +from .utils import get_major_version, parse_bool, patch_config, read_stripped + + +# Mapping between the libpq connection parameters and the environment variables. +# This dict should be kept in sync with `patroni.utils._AUTH_ALLOWED_PARAMETERS` +# (we use "username" in the Patroni config for some reason, other parameter names are the same). +_AUTH_ALLOWED_PARAMETERS_MAPPING = { + 'user': 'PGUSER', + 'password': 'PGPASSWORD', + 'sslmode': 'PGSSLMODE', + 'sslcert': 'PGSSLCERT', + 'sslkey': 'PGSSLKEY', + 'sslpassword': '', + 'sslrootcert': 'PGSSLROOTCERT', + 'sslcrl': 'PGSSLCRL', + 'sslcrldir': 'PGSSLCRLDIR', + 'gssencmode': 'PGGSSENCMODE', + 'channel_binding': 'PGCHANNELBINDING' +} +_NO_VALUE_MSG = '#FIXME' + + +def get_address() -> Tuple[str, str]: + """Try to get hostname and the ip address for it returned by :func:`~socket.gethostname`. + + .. note:: + Can also return local ip. + + :returns: tuple consisting of the hostname returned by :func:`~socket.gethostname` + and the first element in the sorted list of the addresses returned by :func:`~socket.getaddrinfo`. + Sorting guarantees it will prefer IPv4. + If an exception occured, hostname and ip values are equal to :data:`~patroni.config_generator._NO_VALUE_MSG`. + """ + hostname = None + try: + hostname = socket.gethostname() + return hostname, sorted(socket.getaddrinfo(hostname, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0), + key=lambda x: x[0])[0][4][0] + except Exception as err: + logging.warning('Failed to obtain address: %r', err) + return _NO_VALUE_MSG, _NO_VALUE_MSG + + +class AbstractConfigGenerator(abc.ABC): + """Object representing the generated Patroni config. + + :ivar output_file: full path to the output file to be used. + :ivar pg_major: integer representation of the major PostgreSQL version. + :ivar config: dictionary used for the generated configuration storage. + """ + + _HOSTNAME, _IP = get_address() + + def __init__(self, output_file: Optional[str]) -> None: + """Set up the output file (if passed), helper vars and the minimal config structure. + + :param output_file: full path to the output file to be used. + """ + self.output_file = output_file + self.pg_major = 0 + self.config = self.get_template_config() + + self.generate() + + @classmethod + def get_template_config(cls) -> Dict[str, Any]: + """Generate a template config for further extension (e.g. in the inherited classes). + + :returns: dictionary with the values gathered from Patroni env, hopefully defined hostname and ip address + (otherwise set to :data:`~patroni.config_generator._NO_VALUE_MSG`), and some sane defaults. + """ + template_config: Dict[str, Any] = { + 'scope': _NO_VALUE_MSG, + 'name': cls._HOSTNAME, + 'postgresql': { + 'data_dir': _NO_VALUE_MSG, + 'connect_address': _NO_VALUE_MSG + ':5432', + 'listen': _NO_VALUE_MSG + ':5432', + 'bin_dir': '', + 'authentication': { + 'superuser': { + 'username': 'postgres', + 'password': _NO_VALUE_MSG + }, + 'replication': { + 'username': 'replicator', + 'password': _NO_VALUE_MSG + } + } + }, + 'restapi': { + 'connect_address': cls._IP + ':8008', + 'listen': cls._IP + ':8008' + } + } + + dynamic_config = Config.get_default_config() + # to properly dump CaseInsensitiveDict as YAML later + dynamic_config['postgresql']['parameters'] = dict(dynamic_config['postgresql']['parameters']) + config = Config('', None).local_configuration # Get values from env + config.setdefault('bootstrap', {})['dcs'] = dynamic_config + config.setdefault('postgresql', {}) + del config['bootstrap']['dcs']['standby_cluster'] + + patch_config(template_config, config) + return template_config + + @abc.abstractmethod + def generate(self) -> None: + """Generate config and store in :attr:`~AbstractConfigGenerator.config`.""" + + def write_config(self) -> None: + """Write current :attr:`~AbstractConfigGenerator.config` to the output file if provided, to stdout otherwise.""" + if self.output_file: + dir_path = os.path.dirname(self.output_file) + if dir_path and not os.path.isdir(dir_path): + os.makedirs(dir_path) + with open(self.output_file, 'w', encoding='UTF-8') as output_file: + yaml.safe_dump(self.config, output_file, default_flow_style=False, allow_unicode=True) + else: + yaml.safe_dump(self.config, sys.stdout, default_flow_style=False, allow_unicode=True) + + +class SampleConfigGenerator(AbstractConfigGenerator): + """Object representing the generated sample Patroni config. + + Sane defults are used based on the gathered PG version. + """ + + @property + def get_auth_method(self) -> str: + """Return the preferred authentication method for a specific PG version if provided or the default ``md5``. + + :returns: :class:`str` value for the preferred authentication method. + """ + return 'scram-sha-256' if self.pg_major and self.pg_major >= 100000 else 'md5' + + def _get_int_major_version(self) -> int: + """Get major PostgreSQL version from the binary as an integer. + + :returns: an integer PostgreSQL major version representation gathered from the PostgreSQL binary. + See :func:`~patroni.postgresql.misc.postgres_major_version_to_int` and + :func:`~patroni.utils.get_major_version`. + """ + postgres_bin = ((self.config.get('postgresql') or {}).get('bin_name') or {}).get('postgres', 'postgres') + return postgres_major_version_to_int(get_major_version(self.config['postgresql'].get('bin_dir'), postgres_bin)) + + def generate(self) -> None: + """Generate sample config using some sane defaults and update :attr:`~AbstractConfigGenerator.config`.""" + self.pg_major = self._get_int_major_version() + + self.config['postgresql']['parameters'] = {'password_encryption': self.get_auth_method} + username = self.config["postgresql"]["authentication"]["replication"]["username"] + self.config['postgresql']['pg_hba'] = [ + f'host all all all {self.get_auth_method}', + f'host replication {username} all {self.get_auth_method}' + ] + + # add version-specific configuration + wal_keep_param = 'wal_keep_segments' if self.pg_major < 130000 else 'wal_keep_size' + self.config['bootstrap']['dcs']['postgresql']['parameters'][wal_keep_param] = \ + ConfigHandler.CMDLINE_OPTIONS[wal_keep_param][0] + + self.config['bootstrap']['dcs']['postgresql']['use_pg_rewind'] = True + if self.pg_major >= 110000: + self.config['postgresql']['authentication'].setdefault( + 'rewind', {'username': 'rewind_user'}).setdefault('password', _NO_VALUE_MSG) + + +class RunningClusterConfigGenerator(AbstractConfigGenerator): + """Object representing the Patroni config generated using information gathered from the running instance. + + :ivar dsn: DSN string for the local instance to get GUC values from (if provided). + :ivar parsed_dsn: DSN string parsed into a dictionary (see :func:`~patroni.postgresql.config.parse_dsn`). + """ + + def __init__(self, output_file: Optional[str] = None, dsn: Optional[str] = None) -> None: + """Additionally store the passed dsn (if any) in both original and parsed version and run config generation. + + :param output_file: full path to the output file to be used. + :param dsn: DSN string for the local instance to get GUC values from. + + :raises: + :exc:`~patroni.exceptions.PatroniException`: if DSN parsing failed. + """ + self.dsn = dsn + self.parsed_dsn = {} + + super().__init__(output_file) + + @property + def _get_hba_conn_types(self) -> Tuple[str, ...]: + """Return the connection types allowed. + + If :attr:`~RunningClusterConfigGenerator.pg_major` is defined, adds additional parameters + for PostgreSQL version >=16. + + :returns: tuple of the connection methods allowed. + """ + allowed_types = ('local', 'host', 'hostssl', 'hostnossl', 'hostgssenc', 'hostnogssenc') + if self.pg_major and self.pg_major >= 160000: + allowed_types += ('include', 'include_if_exists', 'include_dir') + return allowed_types + + @property + def _required_pg_params(self) -> List[str]: + """PG configuration prameters that have to be always present in the generated config. + + :returns: list of the parameter names. + """ + return ['hba_file', 'ident_file', 'config_file', 'data_directory'] + \ + list(ConfigHandler.CMDLINE_OPTIONS.keys()) + + def _get_bin_dir_from_running_instance(self) -> str: + """Define the directory postgres binaries reside using postmaster's pid executable. + + :returns: path to the PostgreSQL binaries directory. + + :raises: + :exc:`~patroni.exceptions.PatroniException`: if: + + * pid could not be obtained from the ``postmaster.pid`` file; or + * :exc:`OSError` occured during ``postmaster.pid`` file handling; or + * the obtained postmaster pid doesn't exist. + """ + postmaster_pid = None + data_dir = self.config['postgresql']['data_dir'] + try: + with open(f"{data_dir}/postmaster.pid", 'r') as pid_file: + postmaster_pid = pid_file.readline() + if not postmaster_pid: + raise PatroniException('Failed to obtain postmaster pid from postmaster.pid file') + postmaster_pid = int(postmaster_pid.strip()) + except OSError as err: + raise PatroniException(f'Error while reading postmaster.pid file: {err}') + try: + return os.path.dirname(psutil.Process(postmaster_pid).exe()) + except psutil.NoSuchProcess: + raise PatroniException("Obtained postmaster pid doesn't exist.") + + @contextmanager + def _get_connection_cursor(self) -> Iterator[Union['cursor', 'Cursor[Any]']]: + """Get cursor for the PG connection established based on the stored information. + + :raises: + :exc:`~patroni.exceptions.PatroniException`: if :exc:`psycopg.Error` occured. + """ + try: + conn = psycopg.connect(dsn=self.dsn, + password=self.config['postgresql']['authentication']['superuser']['password']) + with conn.cursor() as cur: + yield cur + conn.close() + except psycopg.Error as e: + raise PatroniException(f'Failed to establish PostgreSQL connection: {e}') + + def _set_pg_params(self, cur: Union['cursor', 'Cursor[Any]']) -> None: + """Extend :attr:`~RunningClusterConfigGenerator.config` with the actual PG GUCs values. + + THe following GUC values are set: + + * Non-internal having configuration file, postmaster command line or environment variable + as a source. + + * List of the always required parameters (see :meth:`~RunningClusterConfigGenerator._required_pg_params`). + + :param cur: connection cursor to use. + """ + cur.execute("SELECT name, current_setting(name) FROM pg_settings " + "WHERE context <> 'internal' " + "AND source IN ('configuration file', 'command line', 'environment variable') " + "AND category <> 'Write-Ahead Log / Recovery Target' " + "AND setting <> '(disabled)' " + "OR name = ANY(%s)", (self._required_pg_params,)) + + helper_dict = dict.fromkeys(['port', 'listen_addresses']) + self.config['postgresql'].setdefault('parameters', {}) + for param, value in cur.fetchall(): + if param == 'data_directory': + self.config['postgresql']['data_dir'] = value + elif param == 'cluster_name' and value: + self.config['scope'] = value + elif param in ('archive_command', 'restore_command', + 'archive_cleanup_command', 'recovery_end_command', + 'ssl_passphrase_command', 'hba_file', + 'ident_file', 'config_file'): + # write commands to the local config due to security implications + # write hba/ident/config_file to local config to ensure they are not removed later + self.config['postgresql']['parameters'][param] = value + elif param in helper_dict: + helper_dict[param] = value + else: + self.config['bootstrap']['dcs']['postgresql']['parameters'][param] = value + + connect_port = self.parsed_dsn.get('port', os.getenv('PGPORT', helper_dict['port'])) + self.config['postgresql']['connect_address'] = f'{self._IP}:{connect_port}' + self.config['postgresql']['listen'] = f'{helper_dict["listen_addresses"]}:{helper_dict["port"]}' + + def _set_su_params(self) -> None: + """Extend :attr:`~RunningClusterConfigGenerator.config` with the superuser auth information. + + Information set is based on the options used for connection. + """ + su_params: Dict[str, str] = {} + for conn_param, env_var in _AUTH_ALLOWED_PARAMETERS_MAPPING.items(): + val = self.parsed_dsn.get(conn_param, os.getenv(env_var)) + if val: + su_params[conn_param] = val + patroni_env_su_username = ((self.config.get('authentication') or {}).get('superuser') or {}).get('username') + patroni_env_su_pwd = ((self.config.get('authentication') or {}).get('superuser') or {}).get('password') + # because we use "username" in the config for some reason + su_params['username'] = su_params.pop('user', patroni_env_su_username) or getuser() + su_params['password'] = su_params.get('password', patroni_env_su_pwd) or \ + getpass('Please enter the user password:') + self.config['postgresql']['authentication'] = { + 'superuser': su_params, + 'replication': {'username': _NO_VALUE_MSG, 'password': _NO_VALUE_MSG} + } + + def _set_conf_files(self) -> None: + """Extend :attr:`~RunningClusterConfigGenerator.config` with ``pg_hba.conf`` and ``pg_ident.conf`` content. + + .. note:: + This function only defines ``postgresql.pg_hba`` and ``postgresql.pg_ident`` when + ``hba_file`` and ``ident_file`` are set to the defaults. It may happen these files + are located outside of ``PGDATA`` and Patroni doesn't have write permissions for them. + + :raises: + :exc:`~patroni.exceptions.PatroniException`: if :exc:`OSError` occured during the conf files handling. + """ + default_hba_path = os.path.join(self.config['postgresql']['data_dir'], 'pg_hba.conf') + if self.config['postgresql']['parameters']['hba_file'] == default_hba_path: + try: + self.config['postgresql']['pg_hba'] = list( + filter(lambda i: i and i.split()[0] in self._get_hba_conn_types, read_stripped(default_hba_path))) + except OSError as err: + raise PatroniException(f'Failed to read pg_hba.conf: {err}') + + default_ident_path = os.path.join(self.config['postgresql']['data_dir'], 'pg_ident.conf') + if self.config['postgresql']['parameters']['ident_file'] == default_ident_path: + try: + self.config['postgresql']['pg_ident'] = [i for i in read_stripped(default_ident_path) + if i and not i.startswith('#')] + except OSError as err: + raise PatroniException(f'Failed to read pg_ident.conf: {err}') + if not self.config['postgresql']['pg_ident']: + del self.config['postgresql']['pg_ident'] + + def _enrich_config_from_running_instance(self) -> None: + """Extend :attr:`~RunningClusterConfigGenerator.config` with the values gathered from the running instance. + + Retrieve the following information from the running PostgreSQL instance: + + * superuser auth parameters (see :meth:`~RunningClusterConfigGenerator._set_su_params`); + * some GUC values (see :meth:`~RunningClusterConfigGenerator._set_pg_params`); + * ``postgresql.connect_address``, ``postgresql.listen``; + * ``postgresql.pg_hba`` and ``postgresql.pg_ident`` (see :meth:`~RunningClusterConfigGenerator._set_conf_files`) + + And redefine ``scope`` with the ``cluster_name`` GUC value if set. + + :raises: + :exc:`~patroni.exceptions.PatroniException`: if the provided user doesn't have superuser privileges. + """ + self._set_su_params() + + with self._get_connection_cursor() as cur: + self.pg_major = getattr(cur.connection, 'server_version', 0) + + if not parse_bool(cur.connection.info.parameter_status('is_superuser')): + raise PatroniException('The provided user does not have superuser privilege') + + self._set_pg_params(cur) + + self._set_conf_files() + + def generate(self) -> None: + """Generate config using the info gathered from the specified running PG instance. + + Result is written to :attr:`~RunningClusterConfigGenerator.config`. + """ + if self.dsn: + self.parsed_dsn = parse_dsn(self.dsn) or {} + if not self.parsed_dsn: + raise PatroniException('Failed to parse DSN string') + + self._enrich_config_from_running_instance() + self.config['postgresql']['bin_dir'] = self._get_bin_dir_from_running_instance() + + +def generate_config(output_file: str, sample: bool, dsn: Optional[str]) -> None: + """Generate Patroni configuration file. + + Gather all the available non-internal GUC values having configuration file, postmaster command line or environment + variable as a source and store them in the appropriate part of Patroni configuration (``postgresql.parameters`` or + ``bootstrap.dcs.postgresql.parameters``). Either the provided DSN (takes precedence) or PG ENV vars will be used + for the connection. If password is not provided, it should be entered via prompt. + + The created configuration contains: + * ``scope``: ``cluster_name`` GUC value or ``PATRONI_SCOPE ENV`` variable value if available. + * ``name``: ``PATRONI_NAME`` ENV variable value if set, otherwise hostname. + + * ``bootstrap.dcs``: section with all the parameters (incl. the majority of PG GUCs) set to their default values + defined by Patroni and adjusted by the source instances's configuration values. + + * ``postgresql.parameters``: the source instance's ``archive_command``, ``restore_command``, + ``archive_cleanup_command``, ``recovery_end_command``, ``ssl_passphrase_command``, ``hba_file``, ``ident_file``, + ``config_file`` GUC values. + + * ``postgresql.bin_dir``: path to Postgres binaries gathered from the running instance or, if not available, + the value of ``PATRONI_POSTGRESQL_BIN_DIR`` ENV variable. Otherwise, an empty string. + + * ``postgresql.datadir``: the value gathered from the corresponding PG GUC. + * ``postgresql.listen``: source instance's ``listen_addresses`` and port GUC values. + * ``postgresql.connect_address``: if possible, generated from the connection params. + * ``postgresql.authentication``: + + * superuser and replication users defined (if possible, usernames are set from the respective Patroni ENV vars, + otherwise the default ``postgres`` and ``replicator`` values are used). + If not a sample config, either DSN or PG ENV vars are used to define superuser authentication parameters. + + * rewind user is defined only for sample config, if PG version can be defined and PG version is >=11 + (if possible, username is set from the respective Patroni ENV var). + + * ``bootstrap.dcs.postgresql.use_pg_rewind`` set to ``True`` for a sample config only. + * ``postgresql.pg_hba`` defaults or the lines gathered from the source instance's ``hba_file``. + * ``postgresql.pg_ident`` the lines gathered from the source instance's ``ident_file``. + + :param output_file: Full path to the configuration file to be used. If not provided, result is sent to ``stdout``. + :param sample: Optional flag. If set, no source instance will be used - generate config with some sane defaults. + :param dsn: Optional DSN string for the local instance to get GUC values from. + """ + try: + if sample: + config_generator = SampleConfigGenerator(output_file) + else: + config_generator = RunningClusterConfigGenerator(output_file, dsn) + + config_generator.write_config() + except PatroniException as e: + sys.exit(str(e)) + except Exception as e: + sys.exit(f'Unexpected exception: {e}') diff --git a/patroni/utils.py b/patroni/utils.py index b7694d17..0e12eaf2 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -16,6 +16,7 @@ import platform import random import re import socket +import subprocess import sys import tempfile import time @@ -467,6 +468,18 @@ def _sleep(interval: Union[int, float]) -> None: time.sleep(interval) +def read_stripped(file_path: str) -> Iterator[str]: + """Iterate over stripped lines in the given file. + + :param file_path: path to the file to read from + + :yields: each line from the given file stripped + """ + with open(file_path) as f: + for line in f: + yield line.strip() + + class RetryFailedError(PatroniException): """Maximum number of attempts exhausted in retry operation.""" @@ -977,3 +990,34 @@ def unquote(string: str) -> str: except ValueError: ret = string return ret + + +def get_major_version(bin_dir: Optional[str] = None, bin_name: str = 'postgres') -> str: + """Get the major version of PostgreSQL. + + It is based on the output of ``postgres --version``. + + :param bin_dir: path to the PostgreSQL binaries directory. If ``None`` or an empty string, it will use the first + *bin_name* binary that is found by the subprocess in the ``PATH``. + :param bin_name: name of the postgres binary to call (``postgres`` by default) + :returns: the PostgreSQL major version. + :raises: + :exc:`~patroni.exceptions.PatroniException`: if the postgres binary call failed due to :exc:`OSError`. + + :Example: + + * Returns `9.6` for PostgreSQL 9.6.24 + * Returns `15` for PostgreSQL 15.2 + """ + if not bin_dir: + binary = bin_name + else: + binary = os.path.join(bin_dir, bin_name) + try: + version = subprocess.check_output([binary, '--version']).decode() + except OSError as e: + raise PatroniException(f'Failed to get postgres version: {e}') + version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version) + if TYPE_CHECKING: # pragma: no cover + assert version is not None + return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1) diff --git a/patroni/validator.py b/patroni/validator.py index a8ada1f5..2f70e334 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -6,17 +6,16 @@ This module contains facilities for validating configuration of Patroni processe :var schema: configuration schema of the daemon launched by `patroni` command. """ import os -import re import shutil import socket -import subprocess -from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING +from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple from .collections import CaseInsensitiveSet + from .dcs import dcs_modules from .exceptions import ConfigParseError -from .utils import parse_int, split_host_port, data_directory_is_empty +from .utils import parse_int, split_host_port, data_directory_is_empty, get_major_version def data_directory_empty(data_dir: str) -> bool: @@ -187,31 +186,6 @@ def get_bin_name(bin_name: str) -> str: return (schema.data.get('postgresql', {}).get('bin_name', {}) or {}).get(bin_name, bin_name) -def get_major_version(bin_dir: OptionalType[str] = None) -> str: - """Get the major version of PostgreSQL. - - It is based on the output of ``postgres --version``. - - :param bin_dir: path to PostgreSQL binaries directory. If ``None`` it will use the first ``postgres`` binary that - is found by subprocess in the ``PATH``. - :returns: the PostgreSQL major version. - - :Example: - - * Returns `9.6` for PostgreSQL 9.6.24 - * Returns `15` for PostgreSQL 15.2 - """ - if not bin_dir: - binary = get_bin_name('postgres') - else: - binary = os.path.join(bin_dir, get_bin_name('postgres')) - version = subprocess.check_output([binary, '--version']).decode() - version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version) - if TYPE_CHECKING: # pragma: no cover - assert version is not None - return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1) - - def validate_data_dir(data_dir: str) -> bool: """Validate the value of ``postgresql.data_dir`` configuration option. @@ -246,7 +220,7 @@ def validate_data_dir(data_dir: str) -> bool: raise ConfigParseError("data dir for the cluster is not empty, but doesn't contain" " \"{}\" directory".format(waldir)) bin_dir = schema.data.get("postgresql", {}).get("bin_dir", None) - major_version = get_major_version(bin_dir) + major_version = get_major_version(bin_dir, get_bin_name('postgres')) if pgversion != major_version: raise ConfigParseError("data_dir directory postgresql version ({}) doesn't match with " "'postgres --version' output ({})".format(pgversion, major_version)) diff --git a/tests/__init__.py b/tests/__init__.py index dcf9ed9e..01d74616 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -118,6 +118,21 @@ class MockCursor(object): '"state":"streaming","sync_state":"async","sync_priority":0}]' now = datetime.datetime.now(tzutc) self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)] + elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'): + self.results = [('data_directory', 'data'), + ('hba_file', os.path.join('data', 'pg_hba.conf')), + ('ident_file', os.path.join('data', 'pg_ident.conf')), + ('max_connections', 42), + ('max_locks_per_transaction', 73), + ('max_prepared_transactions', 0), + ('max_replication_slots', 21), + ('max_wal_senders', 37), + ('track_commit_timestamp', 'off'), + ('wal_level', 'replica'), + ('listen_addresses', '6.6.6.6'), + ('port', 1984), + ('archive_command', 'my archive command'), + ('cluster_name', 'my_cluster')] elif sql.startswith('SELECT name, setting'): self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'), ('wal_block_size', '8192', None, 'integer', 'internal'), @@ -159,11 +174,20 @@ class MockCursor(object): pass +class MockConnectionInfo(object): + + def parameter_status(self, param_name): + if param_name == 'is_superuser': + return 'on' + return '0' + + class MockConnect(object): server_version = 99999 autocommit = False closed = 0 + info = MockConnectionInfo() def cursor(self): return MockCursor(self) diff --git a/tests/test_config_generator.py b/tests/test_config_generator.py new file mode 100644 index 00000000..49799c01 --- /dev/null +++ b/tests/test_config_generator.py @@ -0,0 +1,334 @@ +import os +import psutil +import socket +import unittest + +from . import MockConnect, MockCursor, MockConnectionInfo +from copy import deepcopy +from mock import MagicMock, Mock, PropertyMock, mock_open, patch + +from patroni.__main__ import main as _main +from patroni.config import Config +from patroni.config_generator import AbstractConfigGenerator, get_address + +from patroni.utils import patch_config + +from . import psycopg_connect + + +@patch('patroni.psycopg.connect', psycopg_connect) +@patch('socket.getaddrinfo', Mock(return_value=[(0, 0, 0, 0, ('1.9.8.4', 1984))])) +@patch('builtins.open', MagicMock()) +@patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 16.2")) +@patch('psutil.Process.exe', Mock(return_value='/bin/dir/from/running/postgres')) +@patch('psutil.Process.__init__', Mock(return_value=None)) +class TestGenerateConfig(unittest.TestCase): + + no_value_msg = '#FIXME' + _HOSTNAME = socket.gethostname() + _IP = sorted(socket.getaddrinfo(_HOSTNAME, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0), key=lambda x: x[0])[0][4][0] + + def setUp(self): + self.maxDiff = None + + os.environ['PATRONI_SCOPE'] = 'scope_from_env' + os.environ['PATRONI_POSTGRESQL_BIN_DIR'] = '/bin/from/env' + os.environ['PATRONI_SUPERUSER_USERNAME'] = 'su_user_from_env' + os.environ['PATRONI_SUPERUSER_PASSWORD'] = 'su_pwd_from_env' + os.environ['PATRONI_REPLICATION_USERNAME'] = 'repl_user_from_env' + os.environ['PATRONI_REPLICATION_PASSWORD'] = 'repl_pwd_from_env' + os.environ['PATRONI_REWIND_USERNAME'] = 'rewind_user_from_env' + os.environ['PGUSER'] = 'pguser_from_env' + os.environ['PGPASSWORD'] = 'pguser_pwd_from_env' + os.environ['PATRONI_RESTAPI_CONNECT_ADDRESS'] = 'localhost:8080' + os.environ['PATRONI_RESTAPI_LISTEN'] = 'localhost:8080' + os.environ['PATRONI_POSTGRESQL_BIN_POSTGRES'] = 'custom_postgres_bin_from_env' + + self.environ = deepcopy(os.environ) + + dynamic_config = Config.get_default_config() + dynamic_config['postgresql']['parameters'] = dict(dynamic_config['postgresql']['parameters']) + del dynamic_config['standby_cluster'] + dynamic_config['postgresql']['parameters']['wal_keep_segments'] = 8 + dynamic_config['postgresql']['use_pg_rewind'] = True + + self.config = { + 'scope': self.environ['PATRONI_SCOPE'], + 'name': self._HOSTNAME, + 'bootstrap': { + 'dcs': dynamic_config + }, + 'postgresql': { + 'connect_address': self.no_value_msg + ':5432', + 'data_dir': self.no_value_msg, + 'listen': self.no_value_msg + ':5432', + 'pg_hba': ['host all all all md5', + f'host replication {self.environ["PATRONI_REPLICATION_USERNAME"]} all md5'], + 'authentication': {'superuser': {'username': self.environ['PATRONI_SUPERUSER_USERNAME'], + 'password': self.environ['PATRONI_SUPERUSER_PASSWORD']}, + 'replication': {'username': self.environ['PATRONI_REPLICATION_USERNAME'], + 'password': self.environ['PATRONI_REPLICATION_PASSWORD']}, + 'rewind': {'username': self.environ['PATRONI_REWIND_USERNAME']}}, + 'bin_dir': self.environ['PATRONI_POSTGRESQL_BIN_DIR'], + 'bin_name': {'postgres': self.environ['PATRONI_POSTGRESQL_BIN_POSTGRES']}, + 'parameters': {'password_encryption': 'md5'} + }, + 'restapi': { + 'connect_address': self.environ['PATRONI_RESTAPI_CONNECT_ADDRESS'], + 'listen': self.environ['PATRONI_RESTAPI_LISTEN'] + } + } + + def _set_running_instance_config_vals(self): + # values are taken from tests/__init__.py + conf = { + 'scope': 'my_cluster', + 'bootstrap': { + 'dcs': { + 'postgresql': { + 'parameters': { + 'max_connections': 42, + 'max_locks_per_transaction': 73, + 'max_replication_slots': 21, + 'max_wal_senders': 37, + 'wal_level': 'replica', + 'wal_keep_segments': None + }, + 'use_pg_rewind': None + } + } + }, + 'postgresql': { + 'connect_address': f'{self._IP}:bar', + 'listen': '6.6.6.6:1984', + 'data_dir': 'data', + 'bin_dir': '/bin/dir/from/running', + 'parameters': { + 'archive_command': 'my archive command', + 'hba_file': os.path.join('data', 'pg_hba.conf'), + 'ident_file': os.path.join('data', 'pg_ident.conf'), + 'password_encryption': None + }, + 'authentication': { + 'superuser': { + 'username': 'foobar', + 'password': 'qwerty', + 'channel_binding': 'prefer', + 'gssencmode': 'prefer', + 'sslmode': 'prefer' + }, + 'replication': { + 'username': self.no_value_msg, + 'password': self.no_value_msg + }, + 'rewind': None + }, + } + } + patch_config(self.config, conf) + + def _get_running_instance_open_res(self): + hba_content = '\n'.join(self.config['postgresql']['pg_hba'] + ['#host all all all md5', + ' host all all all md5', + '', + 'hostall all all md5']) + ident_content = '\n'.join(['# something very interesting', ' ']) + + self.config['postgresql']['pg_hba'] += ['host all all all md5'] + return [ + mock_open(read_data=hba_content)(), + mock_open(read_data=ident_content)(), + mock_open(read_data='1984')(), + mock_open()() + ] + + @patch('os.makedirs') + @patch('yaml.safe_dump') + def test_generate_sample_config_pre_13_dir_creation(self, mock_config_dump, mock_makedir): + with patch('sys.argv', ['patroni.py', '--generate-sample-config', '/foo/bar.yml']), \ + patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 9.4.3")) as pg_bin_mock, \ + self.assertRaises(SystemExit) as e: + _main() + self.assertEqual(e.exception.code, 0) + self.assertEqual(self.config, mock_config_dump.call_args[0][0]) + mock_makedir.assert_called_once() + pg_bin_mock.assert_called_once_with([os.path.join(self.environ['PATRONI_POSTGRESQL_BIN_DIR'], + self.environ['PATRONI_POSTGRESQL_BIN_POSTGRES']), + '--version']) + + @patch('os.makedirs', Mock()) + @patch('yaml.safe_dump') + def test_generate_sample_config_16(self, mock_config_dump): + conf = { + 'bootstrap': { + 'dcs': { + 'postgresql': { + 'parameters': { + 'wal_keep_size': '128MB', + 'wal_keep_segments': None + }, + } + } + }, + 'postgresql': { + 'parameters': { + 'password_encryption': 'scram-sha-256' + }, + 'pg_hba': ['host all all all scram-sha-256', + f'host replication {self.environ["PATRONI_REPLICATION_USERNAME"]} all scram-sha-256'], + 'authentication': { + 'rewind': { + 'username': self.environ['PATRONI_REWIND_USERNAME'], + 'password': self.no_value_msg} + }, + } + } + patch_config(self.config, conf) + + with patch('sys.argv', ['patroni.py', '--generate-sample-config', '/foo/bar.yml']), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertEqual(e.exception.code, 0) + self.assertEqual(self.config, mock_config_dump.call_args[0][0]) + + @patch('os.makedirs', Mock()) + @patch('yaml.safe_dump') + def test_generate_config_running_instance_16(self, mock_config_dump): + self._set_running_instance_config_vals() + + with patch('builtins.open', Mock(side_effect=self._get_running_instance_open_res())), \ + patch('sys.argv', ['patroni.py', '--generate-config', + '--dsn', 'host=foo port=bar user=foobar password=qwerty']), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertEqual(e.exception.code, 0) + self.assertEqual(self.config, mock_config_dump.call_args[0][0]) + + @patch('os.makedirs', Mock()) + @patch('yaml.safe_dump') + def test_generate_config_running_instance_16_connect_from_env(self, mock_config_dump): + self._set_running_instance_config_vals() + # su auth params and connect host from env + os.environ['PGCHANNELBINDING'] = \ + self.config['postgresql']['authentication']['superuser']['channel_binding'] = 'disable' + + conf = { + 'scope': 'my_cluster', + 'bootstrap': { + 'dcs': { + 'postgresql': { + 'parameters': { + 'max_connections': 42, + 'max_locks_per_transaction': 73, + 'max_replication_slots': 21, + 'max_wal_senders': 37, + 'wal_level': 'replica', + 'wal_keep_segments': None + }, + 'use_pg_rewind': None + } + } + }, + 'postgresql': { + 'connect_address': f'{self._IP}:1984', + 'authentication': { + 'superuser': { + 'username': self.environ['PGUSER'], + 'password': self.environ['PGPASSWORD'], + 'gssencmode': None, + 'sslmode': None + }, + }, + } + } + patch_config(self.config, conf) + + with patch('builtins.open', Mock(side_effect=self._get_running_instance_open_res())), \ + patch('sys.argv', ['patroni.py', '--generate-config']), \ + patch.object(MockConnect, 'server_version', PropertyMock(return_value=160000)), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertEqual(e.exception.code, 0) + self.assertEqual(self.config, mock_config_dump.call_args[0][0]) + + def test_generate_config_running_instance_errors(self): + # 1. Wrong DSN format + with patch('sys.argv', ['patroni.py', '--generate-config', '--dsn', 'host:foo port:bar user:foobar']), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to parse DSN string', e.exception.code) + + # 2. User is not a superuser + with patch('sys.argv', ['patroni.py', + '--generate-config', '--dsn', 'host=foo port=bar user=foobar password=pwd_from_dsn']), \ + patch.object(MockCursor, 'rowcount', PropertyMock(return_value=0), create=True), \ + patch.object(MockConnectionInfo, 'parameter_status', Mock(return_value='off')), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('The provided user does not have superuser privilege', e.exception.code) + + # 3. Error while calling postgres --version + with patch('subprocess.check_output', Mock(side_effect=OSError)), \ + patch('sys.argv', ['patroni.py', '--generate-sample-config']), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to get postgres version:', e.exception.code) + + with patch('sys.argv', ['patroni.py', '--generate-config']): + + # 4. empty postmaster.pid + with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(), + mock_open(read_data='ident_content')(), + mock_open(read_data='')()])), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to obtain postmaster pid from postmaster.pid file', e.exception.code) + + # 5. Failed to open postmaster.pid + with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(), + mock_open(read_data='ident_content')(), + OSError])), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Error while reading postmaster.pid file', e.exception.code) + + # 6. Invalid postmaster pid + with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(), + mock_open(read_data='ident_content')(), + mock_open(read_data='1984')()])), \ + patch('psutil.Process.__init__', Mock(return_value=None)), \ + patch('psutil.Process.exe', Mock(side_effect=psutil.NoSuchProcess(1984))), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn("Obtained postmaster pid doesn't exist", e.exception.code) + + # 7. Failed to open pg_hba + with patch('builtins.open', Mock(side_effect=OSError)), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to read pg_hba.conf', e.exception.code) + + # 8. Failed to open pg_ident + with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(), OSError])), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to read pg_ident.conf', e.exception.code) + + # 9. Failed PG connecttion + from . import psycopg + with patch('patroni.psycopg.connect', side_effect=psycopg.Error), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Failed to establish PostgreSQL connection', e.exception.code) + + # 10. An unexpected error + with patch.object(AbstractConfigGenerator, '__init__', side_effect=psycopg.Error), \ + self.assertRaises(SystemExit) as e: + _main() + self.assertIn('Unexpected exception', e.exception.code) + + def test_get_address(self): + with patch('socket.getaddrinfo', Mock(side_effect=Exception)), \ + patch('logging.warning') as mock_warning: + self.assertEqual(get_address(), (self.no_value_msg, self.no_value_msg)) + self.assertIn('Failed to obtain address: %r', mock_warning.call_args_list[0][0]) From 9209a5a133da1cbaf6cfcb61dcc228af46369181 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 11 Aug 2023 10:19:29 +0200 Subject: [PATCH 4/6] Refactor delete_leader interface (#2810) similar to https://github.com/zalando/patroni/pull/2690, but it helps mostly Consul implementation. --- patroni/dcs/__init__.py | 9 ++++++--- patroni/dcs/consul.py | 8 ++------ patroni/dcs/etcd.py | 2 +- patroni/dcs/etcd3.py | 9 ++++----- patroni/dcs/kubernetes.py | 4 ++-- patroni/dcs/raft.py | 2 +- patroni/dcs/zookeeper.py | 2 +- patroni/ha.py | 6 +++--- tests/test_consul.py | 5 +++-- tests/test_etcd.py | 2 +- tests/test_etcd3.py | 7 ++++--- tests/test_kubernetes.py | 2 +- tests/test_raft.py | 8 ++++---- tests/test_zookeeper.py | 2 +- 14 files changed, 34 insertions(+), 34 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index e536e9dc..279ce265 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1754,26 +1754,29 @@ class AbstractDCS(abc.ABC): """ @abc.abstractmethod - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: """Remove leader key from DCS. This method should remove leader key if current instance is the leader. + :param leader: :class:`Leader` object with information about the leader. + :returns: ``True`` if successfully committed to DCS. """ - def delete_leader(self, last_lsn: Optional[int] = None) -> bool: + def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool: """Update ``optime/leader`` and voluntarily remove leader key from DCS. This method should remove leader key if current instance is the leader. + :param leader: :class:`Leader` object with information about the leader. :param last_lsn: latest checkpoint location in bytes. :returns: boolean result of called abstract :meth:`~AbstractDCS._delete_leader`. """ if last_lsn: self.write_status({self._OPTIME: last_lsn}) - return self._delete_leader() + return bool(leader) and self._delete_leader(leader) @abc.abstractmethod def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 7b310d15..5dd69c86 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -643,12 +643,8 @@ class Consul(AbstractDCS): return self._client.kv.put(self.history_path, value) @catch_consul_errors - def _delete_leader(self) -> bool: - cluster = self.cluster - if cluster and isinstance(cluster.leader, Leader) and\ - cluster.leader.name == self._name and isinstance(cluster.leader.version, int): - return self._client.kv.delete(self.leader_path, cas=cluster.leader.version) - return True + def _delete_leader(self, leader: Leader) -> bool: + return self._client.kv.delete(self.leader_path, cas=int(leader.version)) @catch_consul_errors def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 447c7dab..335cf7a7 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -809,7 +809,7 @@ class Etcd(AbstractEtcd): return bool(self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new))) @catch_etcd_errors - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: return bool(self._client.delete(self.leader_path, prevValue=self._name)) @catch_etcd_errors diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index e5e069c5..5b1acac3 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -912,11 +912,10 @@ class Etcd3(AbstractEtcd): return self.retry(self._client.put, self.initialize_path, sysid, create_revision='0' if create_new else None) @catch_etcd_errors - def _delete_leader(self) -> bool: - cluster = self.cluster - if cluster and isinstance(cluster.leader, Leader) and cluster.leader.name == self._name: - return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.version) - return True + def _delete_leader(self, leader: Leader) -> bool: + fields = build_range_request(self.leader_path) + compare = {'key': fields['key'], 'target': 'VALUE', 'value': base64_encode(self._name)} + return bool(self._client.txn(compare, {'request_delete_range': fields})) @catch_etcd_errors def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 4a66df71..fec544b0 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -1308,11 +1308,11 @@ class Kubernetes(AbstractDCS): if cluster and cluster.config and cluster.config.version else None return self.patch_or_create_config({self._INITIALIZE: sysid}, resource_version) - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: """Unused""" raise NotImplementedError # pragma: no cover - def delete_leader(self, last_lsn: Optional[int] = None) -> bool: + def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool: ret = False kind = self._kinds.get(self.leader_path) if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name: diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index c4286f1e..3f9337cb 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -446,7 +446,7 @@ class Raft(AbstractDCS): def initialize(self, create_new: bool = True, sysid: str = '') -> bool: return self._sync_obj.set(self.initialize_path, sysid, prevExist=(not create_new)) is not False - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: return self._sync_obj.delete(self.leader_path, prevValue=self._name, timeout=1) def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index ed2b71ed..29d159e6 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -466,7 +466,7 @@ class ZooKeeper(AbstractDCS): return False return True - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: self._client.restart() return True diff --git a/patroni/ha.py b/patroni/ha.py index 886dcc8e..0aa9b9d0 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1078,7 +1078,7 @@ class Ha(object): def _delete_leader(self, last_lsn: Optional[int] = None) -> None: self.set_is_leader(False) - self.dcs.delete_leader(last_lsn) + self.dcs.delete_leader(self.cluster.leader, last_lsn) self.dcs.reset_cluster() def release_leader_key_voluntarily(self, last_lsn: Optional[int] = None) -> None: @@ -1874,7 +1874,7 @@ class Ha(object): # location, we can remove the leader key and allow them to start leader race. if self.is_failover_possible(cluster_lsn=checkpoint_location): - self.dcs.delete_leader(checkpoint_location) + self.dcs.delete_leader(self.cluster.leader, checkpoint_location) status['deleted'] = True else: self.dcs.write_leader_optime(checkpoint_location) @@ -1891,7 +1891,7 @@ class Ha(object): if not self.state_handler.is_running(): if self.is_leader() and not status['deleted']: checkpoint_location = self.state_handler.latest_checkpoint_location() - self.dcs.delete_leader(checkpoint_location) + self.dcs.delete_leader(self.cluster.leader, checkpoint_location) self.touch_member() else: # XXX: what about when Patroni is started as the wrong user that has access to the watchdog device diff --git a/tests/test_consul.py b/tests/test_consul.py index ebbbb90e..83ee67d8 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -197,9 +197,10 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)) def test_delete_leader(self): - self.c.delete_leader() + leader = self.c.get_cluster().leader + self.c.delete_leader(leader) self.c._name = 'other' - self.c.delete_leader() + self.c.delete_leader(leader) @patch.object(consul.Consul.KV, 'put', Mock(return_value=True)) def test_initialize(self): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index f9e07144..90402b5f 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -313,7 +313,7 @@ class TestEtcd(unittest.TestCase): self.assertFalse(self.etcd.cancel_initialization()) def test_delete_leader(self): - self.assertFalse(self.etcd.delete_leader()) + self.assertFalse(self.etcd.delete_leader(self.etcd.get_cluster().leader)) def test_delete_cluster(self): self.assertFalse(self.etcd.delete_cluster()) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 2e3ed59d..9aed7eb1 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -298,9 +298,10 @@ class TestEtcd3(BaseTestEtcd3): self.etcd3.cancel_initialization() def test_delete_leader(self): - self.etcd3.delete_leader() + leader = self.etcd3.get_cluster().leader + self.etcd3.delete_leader(leader) self.etcd3._name = 'other' - self.etcd3.delete_leader() + self.etcd3.delete_leader(leader) def test_delete_cluster(self): self.etcd3.delete_cluster() @@ -312,7 +313,7 @@ class TestEtcd3(BaseTestEtcd3): self.etcd3.set_sync_state_value('', 1) def test_delete_sync_state(self): - self.etcd3.delete_sync_state() + self.etcd3.delete_sync_state('1') def test_watch(self): self.etcd3.set_ttl(10) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 9eaa173a..694cd505 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -324,7 +324,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.initialize() def test_delete_leader(self): - self.k.delete_leader(1) + self.k.delete_leader(self.k.get_cluster().leader, 1) def test_cancel_initialization(self): self.k.cancel_initialization() diff --git a/tests/test_raft.py b/tests/test_raft.py index 1fe8733c..9bb109e9 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -142,25 +142,25 @@ class TestRaft(unittest.TestCase): raft._citus_group = '1' self.assertTrue(raft.manual_failover('foo', 'bar')) raft._citus_group = '0' + self.assertTrue(raft.take_leader()) cluster = raft.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) + leader = cluster.leader + self.assertTrue(raft.delete_leader(leader)) self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}')) - leader = raft.get_cluster().leader + raft.get_cluster() self.assertTrue(raft.update_leader(leader, '1', failsafe={'foo': 'bat'})) self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}')) self.assertTrue(raft._sync_obj.set(raft.status_path, '{')) raft.get_citus_coordinator() self.assertTrue(raft.delete_sync_state()) - self.assertTrue(raft.delete_leader()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) raft._citus_group = '1' self.assertTrue(raft.delete_cluster()) raft._citus_group = None raft.get_cluster() - self.assertTrue(raft.take_leader()) - raft.get_cluster() raft.watch(None, 0.001) raft._sync_obj.destroy() diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index c72fefe9..45aeac10 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -202,7 +202,7 @@ class TestZooKeeper(unittest.TestCase): mock_logger.assert_called_once() def test_delete_leader(self): - self.assertTrue(self.zk.delete_leader()) + self.assertTrue(self.zk.delete_leader(self.zk.get_cluster().leader)) def test_set_failover_value(self): self.zk.set_failover_value('') From b83f1c0f440eb748f103f3e282102a85ef90e49c Mon Sep 17 00:00:00 2001 From: Mark Pekala Date: Fri, 11 Aug 2023 01:30:20 -0700 Subject: [PATCH 5/6] [Refactor] Rename _is_leader to _leader_expiry (#2807) --- patroni/ha.py | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 0aa9b9d0..e4b68fcf 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -147,8 +147,8 @@ class Ha(object): self.cluster = Cluster.empty() self.global_config = self.patroni.config.get_global_config(None) self.old_cluster = Cluster.empty() - self._is_leader = False - self._is_leader_lock = RLock() + self._leader_expiry = 0 + self._leader_expiry_lock = RLock() self._failsafe = Failsafe(patroni.dcs) self._was_paused = False self._leader_timeline = None @@ -193,12 +193,20 @@ class Ha(object): return self.global_config.is_standby_cluster def is_leader(self) -> bool: - with self._is_leader_lock: - return self._is_leader > time.time() + """:returns: `True` if the current node is the leader, based on expiration set when it last held the key.""" + with self._leader_expiry_lock: + return self._leader_expiry > time.time() def set_is_leader(self, value: bool) -> None: - with self._is_leader_lock: - self._is_leader = time.time() + self.dcs.ttl if value else 0 + """Update the current node's view of it's own leadership status. + + Will update the expiry timestamp to match the dcs ttl if setting leadership to true, + otherwise will set the expiry to the past to immediately invalidate. + + :param value: is the current node the leader. + """ + with self._leader_expiry_lock: + self._leader_expiry = time.time() + self.dcs.ttl if value else 0 def sync_mode_is_active(self) -> bool: """Check whether synchronous replication is requested and already active. From 82d2ef4878cd327e63fdb25e6faaf47dca3cebbe Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Fri, 11 Aug 2023 09:31:31 +0100 Subject: [PATCH 6/6] Make docs more clear on changes to the `bootstrap.dcs` section of YAML config (#2811) It seems that a common pitfall for new users of Patroni is that the `bootstrap.dcs` section is only used to initialize the configuration in DCS. This moves the comment about this to an info block so it is more visible to the reader. --- docs/yaml_configuration.rst | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index ab5b5587..41776f87 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -30,9 +30,15 @@ Log Bootstrap configuration ----------------------- + +.. note:: + Once Patroni has initialized the cluster for the first time and settings have been stored in the DCS, all future + changes to the ``bootstrap.dcs`` section of the YAML configuration will not take any effect! If you want to change + them please use either ``patronictl edit-config`` or the Patroni :ref:`REST API `. + - **bootstrap**: - - **dcs**: This section will be written into `///config` of the given configuration store after initializing of new cluster. The global dynamic configuration for the cluster. Under the ``bootstrap.dcs`` you can put any of the parameters described in the :ref:`Dynamic Configuration settings ` and after Patroni initialized (bootstrapped) the new cluster, it will write this section into `///config` of the configuration store. All later changes of ``bootstrap.dcs`` will not take any effect! If you want to change them please use either ``patronictl edit-config`` or Patroni :ref:`REST API `. + - **dcs**: This section will be written into `///config` of the given configuration store after initializing the new cluster. The global dynamic configuration for the cluster. You can put any of the parameters described in the :ref:`Dynamic Configuration settings ` under ``bootstrap.dcs`` and after Patroni has initialized (bootstrapped) the new cluster, it will write this section into `///config` of the configuration store. - **method**: custom script to use for bootstrapping this cluster. See :ref:`custom bootstrap methods documentation ` for details.