Improve protection of DCS being accidentally wiped (#680)

We already have a lot of logic in place to prevent failover in such case and restore all keys, but an accidental removal of `/config` key was effectively switching off pause mode for 1 cycle of HA loop.

patroni/config.py

@@ -9,7 +9,7 @@ from collections import defaultdict
 from copy import deepcopy
 from patroni.dcs import ClusterConfig
 from patroni.postgresql import Postgresql
-from patroni.utils import deep_compare, parse_int, patch_config
+from patroni.utils import deep_compare, parse_bool, parse_int, patch_config
 from requests.structures import CaseInsensitiveDict
 
 logger = logging.getLogger(__name__)
@@ -88,6 +88,9 @@ class Config(object):
     def dynamic_configuration(self):
         return deepcopy(self._dynamic_configuration)
 
+    def check_mode(self, mode):
+        return bool(parse_bool(self._dynamic_configuration.get(mode)))
+
     def _load_config_file(self):
         """Loads config.yaml from filesystem and applies some values which were set via ENV"""
         with open(self._config_file) as f:

patroni/dcs/__init__.py

@@ -11,6 +11,7 @@ import sys
 
 from collections import namedtuple
 from patroni.exceptions import PatroniException
+from patroni.utils import parse_bool
 from random import randint
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
 from threading import Event, Lock
@@ -349,14 +350,14 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
         candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude]
         return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader
 
+    def check_mode(self, mode):
+        return bool(self.config and parse_bool(self.config.data.get(mode)))
+
     def is_paused(self):
-        return self.config and self.config.data.get('pause', False) or False
+        return self.check_mode('pause')
 
     def is_synchronous_mode(self):
-        return bool(self.config and self.config.data.get('synchronous_mode'))
-
-    def is_synchronous_mode_strict(self):
-        return bool(self.config and self.config.data.get('synchronous_mode_strict'))
+        return self.check_mode('synchronous_mode')
 
 
 @six.add_metaclass(abc.ABCMeta)

patroni/ha.py

@@ -76,8 +76,15 @@ class Ha(object):
         # already running as replica was aborted due to cluster not beeing initialized in DCS.
         self._join_aborted = False
 
+    def check_mode(self, mode):
+        # Try to protect from the case when DCS was wiped out during pause
+        if self.cluster and self.cluster.config and self.cluster.config.modify_index:
+            return self.cluster.check_mode(mode)
+        else:
+            return self.patroni.config.check_mode(mode)
+
     def is_paused(self):
-        return self.cluster and self.cluster.is_paused()
+        return self.check_mode('pause')
 
     def load_cluster_from_dcs(self):
         cluster = self.dcs.get_cluster()
@@ -288,10 +295,10 @@ class Ha(object):
         return follow_reason
 
     def is_synchronous_mode(self):
-        return bool(self.cluster and self.cluster.is_synchronous_mode())
+        return self.check_mode('synchronous_mode')
 
     def is_synchronous_mode_strict(self):
-        return bool(self.cluster and self.cluster.is_synchronous_mode_strict())
+        return self.check_mode('synchronous_mode_strict')
 
     def process_sync_replication(self):
         """Process synchronous standby beahvior.

tests/test_etcd.py

@@ -90,7 +90,7 @@ def etcd_read(self, key, **kwargs):
         raise etcd.EtcdKeyNotFound
 
     response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
-                {"key": "/service/batman5/config", "value": '{"foo": "bar"}',
+                {"key": "/service/batman5/config", "value": '{"synchronous_mode": 0}',
                  "modifiedIndex": 1582, "createdIndex": 1582},
                 {"key": "/service/batman5/failover", "value": "",
                  "modifiedIndex": 1582, "createdIndex": 1582},
@@ -276,7 +276,9 @@ class TestEtcd(unittest.TestCase):
                                   {'hosts': 'foo:4001,bar', 'retry_timeout': 10})
 
     def test_get_cluster(self):
-        self.assertIsInstance(self.etcd.get_cluster(), Cluster)
+        cluster = self.etcd.get_cluster()
+        self.assertIsInstance(cluster, Cluster)
+        self.assertFalse(cluster.is_synchronous_mode())
         self.etcd._base_path = '/service/nocluster'
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)