From d78790b194eb0397b5e82636b10291e31812ea36 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 16 Apr 2018 17:32:26 +0200 Subject: [PATCH] Abort start if attaching to running postgres and cluster not initiazlied (#661) Patroni can attach itself to an already running PostgreSQL instance. If that is the first instance "seen" in the given cluster, Patroni for that instance will create the initialize key, grab the leader key and, if the instance is running a replica, promote. Because of this behavior, when a cluster with a master and one or more replicas gets Patroni for each node, it is imperative to start running Patroni on the master node before getting to the replicas. This commit changes such weird behavior and will abort Patroni start if there is no initialize key in DCS and postgres is running as a replica. Closes https://github.com/zalando/patroni/issues/655 --- patroni/ha.py | 12 +++++++++++- tests/test_ha.py | 28 ++++++++++++++++++++-------- 2 files changed, 31 insertions(+), 9 deletions(-) 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)