diff --git a/patroni/ha.py b/patroni/ha.py index e4f84b8f..e5058579 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -72,6 +72,10 @@ class Ha(object): # standby. Changes protected by _member_state_lock. self._disable_sync = 0 + # We need following property to avoid shutdown of postgres when join of Patroni to the postgres + # already running as replica was aborted due to cluster not beeing initialized in DCS. + self._join_aborted = False + def is_paused(self): return self.cluster and self.cluster.is_paused() @@ -1079,6 +1083,12 @@ class Ha(object): return self.bootstrap() # new node # "bootstrap", but data directory is not empty elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked() and not self.is_paused(): + if not self.state_handler.cb_called and self.state_handler.is_running() \ + and not self.state_handler.is_leader(): + 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 master') + sys.exit(1) self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) else: # check if we are allowed to join @@ -1138,7 +1148,7 @@ class Ha(object): if self.is_paused(): logger.info('Leader key is not deleted and Postgresql is not stopped due paused state') self.watchdog.disable() - else: + elif not self._join_aborted: # FIXME: If stop doesn't reach safepoint quickly enough keepalive is triggered. If shutdown checkpoint # takes longer than ttl, then leader key is lost and replication might not have sent out all xlog. # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data. diff --git a/tests/test_ha.py b/tests/test_ha.py index 2312ddbe..bdd2ff9f 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -16,6 +16,9 @@ from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get from test_postgresql import psycopg2_connect, MockPostmaster +SYSID = '12345678901' + + def true(*args, **kwargs): return True @@ -45,7 +48,7 @@ def get_cluster_initialized_without_leader(leader=False, failover=None, sync=Non 'scheduled_restart': {'schedule': "2100-01-01 10:53:07.560445+00:00", 'postgres_version': '99.0.0'}}) syncstate = SyncState(0 if sync else None, sync and sync[0], sync and sync[1]) - return get_cluster(True, leader, [m1, m2], failover, syncstate) + return get_cluster(SYSID, leader, [m1, m2], failover, syncstate) def get_cluster_initialized_with_leader(failover=None, sync=None): @@ -120,7 +123,7 @@ def run_async(self, func, args=()): @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3)) @patch.object(Postgresql, 'call_nowait', Mock(return_value=True)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) -@patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'})) +@patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': SYSID})) @patch.object(Postgresql, 'sync_replication_slots', Mock()) @patch.object(Postgresql, 'write_pg_hba', Mock()) @patch.object(Postgresql, 'write_pgpass', Mock(return_value={})) @@ -161,7 +164,7 @@ class TestHa(unittest.TestCase): 'name': 'foo', 'retry_timeout': 10}}) self.ha = Ha(MockPatroni(self.p, self.e)) self.ha.old_cluster = self.e.get_cluster() - self.ha.cluster = get_cluster_not_initialized_without_leader() + self.ha.cluster = get_cluster_initialized_without_leader() self.ha.load_cluster_from_dcs = Mock() def test_update_lock(self): @@ -178,7 +181,7 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'starting as a secondary') def test_recover_replica_failed(self): - self.p.controldata = lambda: {'Database cluster state': 'in recovery'} + self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID} self.p.is_running = false self.p.follow = false self.assertEquals(self.ha.run_cycle(), 'starting as a secondary') @@ -189,14 +192,14 @@ class TestHa(unittest.TestCase): self.p.is_running = false self.p.name = 'leader' self.p.set_role('master') - self.p.controldata = lambda: {'Database cluster state': 'shut down'} + self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID} self.ha.cluster = get_cluster_initialized_with_leader() self.assertEquals(self.ha.run_cycle(), 'starting as readonly because i had the session lock') @patch.object(Postgresql, 'fix_cluster_state', Mock()) def test_crash_recovery(self): self.p.is_running = false - self.p.controldata = lambda: {'Database cluster state': 'in production'} + self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID} self.assertEquals(self.ha.run_cycle(), 'doing crash recovery in a single user mode') @patch.object(Postgresql, 'rewind_needed_and_possible', Mock(return_value=True)) @@ -208,7 +211,7 @@ class TestHa(unittest.TestCase): @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True)) @patch.object(Postgresql, 'fix_cluster_state', Mock()) def test_single_user_after_recover_failed(self): - self.p.controldata = lambda: {'Database cluster state': 'in recovery'} + self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID} self.p.is_running = false self.p.follow = false self.assertEquals(self.ha.run_cycle(), 'starting as a secondary') @@ -217,6 +220,7 @@ class TestHa(unittest.TestCase): @patch('sys.exit', return_value=1) @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True)) def test_sysid_no_match(self, exit_mock): + self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': '123'} self.ha.run_cycle() exit_mock.assert_called_once_with(1) @@ -225,7 +229,7 @@ class TestHa(unittest.TestCase): self.p.is_leader = false self.p.is_healthy = true self.ha.has_lock = true - self.p.controldata = lambda: {'Database cluster state': 'in production'} + self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID} self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') @patch('psycopg2.connect', psycopg2_connect) @@ -279,6 +283,7 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'Not promoting self because watchdog could not be activated') def test_leader_with_lock(self): + self.ha.cluster = get_cluster_not_initialized_without_leader() self.ha.cluster.is_unlocked = false self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') @@ -896,3 +901,10 @@ class TestHa(unittest.TestCase): self.ha.has_lock = true self.ha.cluster.is_unlocked = false self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + + @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.ha.run_cycle() + exit_mock.assert_called_once_with(1)