A few bugfixes in the "standby cluster" workflow (#823)

* Always run `pg_rewind` against the remote master
* Always use the remote master as the source when "recovering" stopped standby leader
* Use remote master as the source when "recovering" the node in the unhealthy cluster
* Use the local dbname as the fallback when doing `pg_rewind` from the remote master
*  `no_replication_slot` is the allowed key in the `RemoteMember` object
* Make it possible to "bootstrap" the new `standby_cluster` with existing (and valid) data directory. There is one prerequisite though, there should be no `patroni.dynamic.json` file in it!

patroni/api.py

@@ -87,7 +87,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         replica_status_code = 200 if not patroni.noloadbalance and response.get('role') == 'replica' else 503
         status_code = 503
 
-        if patroni.config.is_standby_cluster and ('standby_leader' in path or 'standby-leader' in path):
+        if patroni.ha.is_standby_cluster() and ('standby_leader' in path or 'standby-leader' in path):
             status_code = 200 if patroni.ha.is_leader() else 503
         elif 'master' in path or 'leader' in path or 'primary' in path:
             status_code = 200 if patroni.ha.is_leader() else 503

patroni/config.py

@@ -9,7 +9,7 @@ import yaml
 
 from collections import defaultdict
 from copy import deepcopy
-from patroni.dcs import ClusterConfig, is_standby_cluster
+from patroni.dcs import ClusterConfig
 from patroni.postgresql import Postgresql
 from patroni.utils import deep_compare, parse_bool, parse_int, patch_config
 from requests.structures import CaseInsensitiveDict
@@ -99,10 +99,6 @@ class Config(object):
     def dynamic_configuration(self):
         return deepcopy(self._dynamic_configuration)
 
-    @property
-    def is_standby_cluster(self):
-        return is_standby_cluster(self._dynamic_configuration.get('standby_cluster'))
-
     def check_mode(self, mode):
         return bool(parse_bool(self._dynamic_configuration.get(mode)))
 

patroni/dcs/__init__.py

@@ -189,13 +189,12 @@ class RemoteMember(Member):
                 'create_replica_methods',
                 'restore_command',
                 'archive_cleanup_command',
-                'recovery_min_apply_delay')
+                'recovery_min_apply_delay',
+                'no_replication_slot')
 
     def __getattr__(self, name):
-        if name not in RemoteMember.allowed_keys():
-            return
-
-        return self.data.get(name)
+        if name in RemoteMember.allowed_keys():
+            return self.data.get(name)
 
 
 class Leader(namedtuple('Leader', 'index,session,member')):
@@ -398,9 +397,6 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
     def is_synchronous_mode(self):
         return self.check_mode('synchronous_mode')
 
-    def is_standby_cluster(self):
-        return is_standby_cluster(self.config and self.config.data.get('standby_cluster'))
-
 
 @six.add_metaclass(abc.ABCMeta)
 class AbstractDCS(object):
@@ -654,14 +650,3 @@ class AbstractDCS(object):
 
         self.event.wait(timeout)
         return self.event.isSet()
-
-
-def is_standby_cluster(config):
-    """ Check whether or not provided configuration describes a standby cluster.
-        Config can be both patroni config or cluster.config.data
-    """
-    return isinstance(config, dict) and (
-        config.get('host') or
-        config.get('port') or
-        config.get('restore_command')
-    )

patroni/ha.py

@@ -91,6 +91,18 @@ class Ha(object):
     def is_paused(self):
         return self.check_mode('pause')
 
+    def get_standby_cluster_config(self):
+        if self.cluster and self.cluster.config and self.cluster.config.modify_index:
+            config = self.cluster.config.data
+        else:
+            config = self.patroni.config.dynamic_configuration
+        return config.get('standby_cluster')
+
+    def is_standby_cluster(self):
+        config = self.get_standby_cluster_config()
+        # Check whether or not provided configuration describes a standby cluster
+        return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command'))
+
     def is_leader(self):
         with self._is_leader_lock:
             return self._is_leader
@@ -202,7 +214,7 @@ class Ha(object):
                 self.state_handler.bootstrapping = True
                 self._post_bootstrap_task = CriticalTask()
 
-                if self.patroni.config.is_standby_cluster:
+                if self.is_standby_cluster():
                     self._async_executor.schedule('bootstrap_standby_leader')
                     self._async_executor.run_async(self.bootstrap_standby_leader)
                     return 'trying to bootstrap a new standby leader'
@@ -228,8 +240,7 @@ class Ha(object):
             not a real master, but a 'standby leader', that will take base backup
             from a remote master and start follow it.
         """
-        patroni_config = self.patroni.config.dynamic_configuration
-        clone_source = self.get_remote_master(patroni_config)
+        clone_source = self.get_remote_master()
         msg = 'clone from remote master {0}'.format(clone_source.conn_url)
         result = self.clone(clone_source, msg)
         self._post_bootstrap_task.complete(result)
@@ -239,9 +250,10 @@ class Ha(object):
         return result
 
     def _handle_rewind(self):
-        if self.state_handler.rewind_needed_and_possible(self.cluster.leader):
-            self._async_executor.schedule('running pg_rewind from ' + self.cluster.leader.name)
-            self._async_executor.run_async(self.state_handler.rewind, (self.cluster.leader,))
+        leader = self.get_remote_master() if self.is_standby_cluster() else self.cluster.leader
+        if self.state_handler.rewind_needed_and_possible(leader):
+            self._async_executor.schedule('running pg_rewind from ' + leader.name)
+            self._async_executor.run_async(self.state_handler.rewind, (leader,))
             return True
 
     def recover(self):
@@ -273,16 +285,24 @@ class Ha(object):
 
         self.load_cluster_from_dcs()
 
-        if self.has_lock():
-            msg = "starting as readonly because i had the session lock"
-            node_to_follow = None
-        else:
+        if self.is_standby_cluster() or not self.has_lock():
             if not self.state_handler.rewind_executed:
                 self.state_handler.trigger_check_diverged_lsn()
             if self._handle_rewind():
                 return self._async_executor.scheduled_action
-            msg = "starting as a secondary"
-            node_to_follow = self._get_node_to_follow(self.cluster)
+
+            if self.has_lock():  # in standby cluster
+                msg = "starting as a standby leader because i had the session lock"
+                node_to_follow = self._get_node_to_follow(self.cluster)
+            elif self.is_standby_cluster() and self.cluster.is_unlocked():
+                msg = "trying to follow a remote master because standby cluster is unhealthy"
+                node_to_follow = self.get_remote_master()
+            else:
+                msg = "starting as a secondary"
+                node_to_follow = self._get_node_to_follow(self.cluster)
+        elif self.has_lock():
+            msg = "starting as readonly because i had the session lock"
+            node_to_follow = None
 
         self.recovering = True
 
@@ -295,8 +315,8 @@ class Ha(object):
         # try to follow the node mentioned there, otherwise, follow the leader.
         is_leader = self.cluster.leader and self.state_handler.name == self.cluster.leader.name
 
-        if self.cluster.is_standby_cluster() and is_leader:
-            node_to_follow = self.get_remote_master(cluster.config.data)
+        if self.is_standby_cluster() and (is_leader or self.cluster.is_unlocked()):
+            node_to_follow = self.get_remote_master()
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
@@ -780,7 +800,7 @@ class Ha(object):
                         self.dcs.manual_failover('', '')
                 self.load_cluster_from_dcs()
 
-                if self.cluster.is_standby_cluster():
+                if self.is_standby_cluster():
                     # standby leader disappeared, and this is a healthiest
                     # replica, so it should become a new standby leader.
                     # This imply that we need to start following a remote master
@@ -822,7 +842,7 @@ class Ha(object):
                 if msg is not None:
                     return msg
 
-                if self.cluster.is_standby_cluster():
+                if self.is_standby_cluster():
                     # in case of standby cluster we don't really need to
                     # enforce anything, since the leader is not a master.
                     # So just remind the role.
@@ -1276,15 +1296,14 @@ class Ha(object):
         This usually happens on the master or if the node is running async action"""
         self.dcs.event.set()
 
-    def get_remote_master(self, config):
+    def get_remote_master(self):
         """ In case of standby cluster this will tel us from which remote
             master to stream. Config can be both patroni config or
             cluster.config.data
         """
-        config = config or (self.config is not None and self.config.data)
+        cluster_params = self.get_standby_cluster_config()
 
-        if config and config.get('standby_cluster'):
-            cluster_params = config.get('standby_cluster')
+        if cluster_params:
             unique_name = 'remote_master:{}'.format(uuid.uuid1())
             data = {
                 'conn_kwargs': {

patroni/postgresql.py

@@ -1214,7 +1214,7 @@ class Postgresql(object):
             ('user', r.get('user')),
             ('host', r.get('host')),
             ('port', r.get('port')),
-            ('dbname', r.get('database')),
+            ('dbname', r.get('database') or self._database),
             ('sslmode', 'prefer'),
             ('sslcompression', '1'),
         ]

tests/test_api.py

@@ -95,6 +95,10 @@ class MockHa(object):
     def is_paused():
         return True
 
+    @staticmethod
+    def is_standby_cluster():
+        return False
+
 
 class MockPatroni(object):
 
@@ -167,8 +171,8 @@ class TestRestApiHandler(unittest.TestCase):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /master'))
         with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
-        MockPatroni.config.is_standby_cluster = PropertyMock(return_value=True)
-        MockRestApiServer(RestApiHandler, 'GET /standby_leader')
+        with patch.object(MockHa, 'is_standby_cluster', Mock(return_value=True)):
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader')
 
     def test_do_OPTIONS(self):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0'))

tests/test_ha.py

@@ -5,7 +5,6 @@ import unittest
 import sys
 
 from mock import Mock, MagicMock, PropertyMock, patch
-from patroni.async_executor import CriticalTask
 from patroni.config import Config
 from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory
 from patroni.dcs.etcd import Client
@@ -62,17 +61,6 @@ def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None)
     return get_cluster(True, leader, [leader], failover, None, cluster_config)
 
 
-def get_cluster_not_initialized_standby(failover=None, sync=None):
-    return get_cluster_not_initialized_without_leader(
-        cluster_config=ClusterConfig(1, {
-            "standby_cluster": {
-                "host": "localhost",
-                "port": 5432,
-                "primary_slot_name": "",
-            }}, 1)
-    )
-
-
 def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
     return get_cluster_initialized_with_only_leader(
         cluster_config=ClusterConfig(1, {
@@ -214,17 +202,10 @@ class TestHa(unittest.TestCase):
     @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True)
     def test_start_as_standby_leader(self, initialize):
         self.p.data_directory_empty = true
-        self.ha.cluster = get_cluster_not_initialized_standby()
+        self.ha.cluster = get_cluster_not_initialized_without_leader(cluster_config=ClusterConfig(0, {}, 0))
         self.ha.cluster.is_unlocked = true
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
-        self.assertEqual(
-            self.ha.run_cycle(),
-            'trying to bootstrap a new standby leader'
-        )
+        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {"port": 5432}}
+        self.assertEqual(self.ha.run_cycle(), 'trying to bootstrap a new standby leader')
 
     @patch.object(Cluster, 'get_clone_member',
                   Mock(return_value=Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})))
@@ -233,27 +214,7 @@ class TestHa(unittest.TestCase):
         self.p.data_directory_empty = true
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.ha.cluster.is_unlocked = false
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
-        self.assertEqual(
-            self.ha.run_cycle(),
-            "trying to bootstrap from replica 'test'"
-        )
-
-    @patch.object(Postgresql, 'create_replica', Mock(return_value=0))
-    def test_bootstrap_standby_leader(self):
-        self.ha.cluster = get_cluster_not_initialized_standby()
-        self.ha.cluster.is_unlocked = true
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
-        self.ha._post_bootstrap_task = CriticalTask()
-        self.assertEqual(self.ha.bootstrap_standby_leader(), True)
+        self.assertEqual(self.ha.run_cycle(), "trying to bootstrap from replica 'test'")
 
     def test_recover_replica_failed(self):
         self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID}
@@ -692,11 +653,6 @@ class TestHa(unittest.TestCase):
     def test_process_healthy_standby_cluster_as_standby_leader(self):
         self.p.is_leader = false
         self.p.name = 'leader'
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         msg = 'no action.  i am the standby leader with the lock'
         self.assertEqual(self.ha.run_cycle(), msg)
@@ -704,24 +660,13 @@ class TestHa(unittest.TestCase):
     def test_process_healthy_standby_cluster_as_cascade_replica(self):
         self.p.is_leader = false
         self.p.name = 'replica'
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         msg = 'no action.  i am a secondary and i am following a leader'
         self.assertEqual(self.ha.run_cycle(), msg)
 
-    @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True)
-    def test_process_unhealthy_standby_cluster_as_standby_leader(self, initialize):
+    def test_process_unhealthy_standby_cluster_as_standby_leader(self):
         self.p.is_leader = false
         self.p.name = 'leader'
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.ha.cluster.is_unlocked = true
         self.ha.sysid_valid = true
@@ -730,19 +675,30 @@ class TestHa(unittest.TestCase):
         self.assertEqual(self.ha.run_cycle(), msg)
 
     @patch.object(Postgresql, 'rewind_needed_and_possible', Mock(return_value=True))
-    @patch('patroni.dcs.etcd.Etcd.initialize', return_value=True)
-    def test_process_unhealthy_standby_cluster_as_cascade_replica(self, initialize):
+    def test_process_unhealthy_standby_cluster_as_cascade_replica(self):
         self.p.is_leader = false
         self.p.name = 'replica'
-        self.ha.patroni.config._dynamic_configuration = {"standby_cluster": {
-            "host": "localhost",
-            "port": 5432,
-            "primary_slot_name": "",
-        }}
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.ha.is_unlocked = true
-        msg = 'running pg_rewind from leader'
-        self.assertEqual(self.ha.run_cycle(), msg)
+        self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_master:'))
+
+    def test_recover_unhealthy_leader_in_standby_cluster(self):
+        self.p.is_leader = false
+        self.p.name = 'leader'
+        self.p.is_running = false
+        self.p.follow = false
+        self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
+        self.assertEqual(self.ha.run_cycle(), 'starting as a standby leader because i had the session lock')
+
+    def test_recover_unhealthy_unlocked_standby_cluster(self):
+        self.p.is_leader = false
+        self.p.name = 'leader'
+        self.p.is_running = false
+        self.p.follow = false
+        self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
+        self.ha.cluster.is_unlocked = true
+        self.ha.has_lock = false
+        self.assertEqual(self.ha.run_cycle(), 'trying to follow a remote master because standby cluster is unhealthy')
 
     def test_failed_to_update_lock_in_pause(self):
         self.ha.update_lock = false