Repository: zalando/patroni (python)

Issue:
Split-brain possible on simple configuration mistake
Currently Patroni will happily launch a 2nd primary if the "member" name is by accident a duplicate of the current leader.

Sure it's an operator error...but I think would be relatively easy to avoid it by first checking the current leader's (if exists) role directly over the Patroni API.

This situation won't happen on new setups as there's seems to be a check against that (although the message "released leader key voluntarily as data dir empty and currently leader" hints that the new duplicate node still grabbed the key), but rather with a "takeover" of some old Postgres clusters where the PGDATA exists.

**To Reproduce**

1. Start the 1st node as leader.
2. Start a 2nd node that will be bootstrapped as replica.
3. Change the replica config so that the member name (`name`) matches the leader's name.
4. Restart Patroni on 2nd node



Pull request description:
Prevent splitbrain from duplicate names in configuration
Fixes #1804 by ensuring that there is not already another node with the same name that is running in the cluster when joining. Will return an error during initialization if there is already a node with the same name. Added a behave test to ensure functionality.

Patch:
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 4b8b2686..4a0eda68 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -83,3 +83,7 @@ Feature: basic replication
     Then postgres0 role is the secondary after 20 seconds
     When I add the table buz to postgres1
     Then table buz is present on postgres0 after 20 seconds
+
+  Scenario: check graceful rejection when two nodes have the same name
+    Given I start duplicate postgres0 on port 8011
+    Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log
diff --git a/features/environment.py b/features/environment.py
index a4d22015..f7690ad2 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -52,10 +52,9 @@ class AbstractController(abc.ABC):
         self._log = open(os.path.join(self._output_dir, self._name + '.log'), 'a')
         self._handle = self._start()
 
-        assert self._has_started(), "Process {0} is not running after being started".format(self._name)
-
         max_wait_limit *= self._context.timeout_multiplier
         for _ in range(max_wait_limit):
+            assert self._has_started(), "Process {0} is not running after being started".format(self._name)
             if self._is_accessible():
                 break
             time.sleep(1)
@@ -344,6 +343,13 @@ class PatroniController(AbstractController):
                         '--datadir=' + os.path.join(self._work_directory, dest),
                         '--dbname=' + self.backup_source])
 
+    def read_patroni_log(self, level):
+        try:
+            with open(str(os.path.join(self._output_dir or '', self._name + ".log"))) as f:
+                return [line for line in f.readlines() if line[24:24 + len(level)] == level]
+        except IOError:
+            return []
+
 
 class ProcessHang(object):
 
@@ -827,7 +833,7 @@ class PatroniPoolController(object):
 
     def __getattr__(self, func):
         if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to',
-                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']:
+                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup', 'read_patroni_log']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 
         def wrapper(name, *args, **kwargs):
diff --git a/features/patroni_api.feature b/features/patroni_api.feature
index 0c47fe6e..624d3271 100644
--- a/features/patroni_api.feature
+++ b/features/patroni_api.feature
@@ -35,21 +35,21 @@ Scenario: check local configuration reload
 	Then I receive a response code 202
 
 Scenario: check dynamic configuration change via DCS
-	Given I run patronictl.py edit-config -s 'ttl=10' -p 'max_connections=101' --force batman
-	Then I receive a response returncode 0
-	And I receive a response output "+ttl: 10"
+	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "postgresql": {"parameters": {"max_connections": "101"}}}
+	Then I receive a response code 200
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
 	When I issue a GET request to http://127.0.0.1:8008/config
 	Then I receive a response code 200
-	And I receive a response ttl 10
+	And I receive a response ttl 20
 	When I issue a GET request to http://127.0.0.1:8008/patroni
 	Then I receive a response code 200
 	And I receive a response tags {'new_tag': 'new_value'}
 	And I sleep for 4 seconds
 
 Scenario: check the scheduled restart
-	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
-	Then I receive a response code 200
+	Given I run patronictl.py edit-config -p 'superuser_reserved_connections=6' --force batman
+	Then I receive a response returncode 0
+	And I receive a response output "+    superuser_reserved_connections: 6"
 	And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
 	Given I issue a scheduled restart at http://127.0.0.1:8008 in 5 seconds with {"role": "replica"}
 	Then I receive a response code 202
diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py
index 9b51a859..6499a03f 100644
--- a/features/steps/basic_replication.py
+++ b/features/steps/basic_replication.py
@@ -9,6 +9,22 @@ def start_patroni(context, name):
     return context.pctl.start(name)
 
 
+@step('I start duplicate {name:w} on port {port:d}')
+def start_duplicate_patroni(context, name, port):
+    config = {
+        "name": name,
+        "restapi": {
+            "listen": "127.0.0.1:{0}".format(port)
+        }
+    }
+    try:
+        context.pctl.start('dup-' + name, custom_config=config)
+        assert False, "Process was expected to fail"
+    except AssertionError as e:
+        assert 'is not running after being started' in str(e),\
+            "No error was raised by duplicate start of {0} ".format(name)
+
+
 @step('I shut down {name:w}')
 def stop_patroni(context, name):
     return context.pctl.stop(name, timeout=60)
@@ -90,3 +106,10 @@ def replication_works(context, primary, replica, time_limit):
         When I add the table test_{0} to {1}
         Then table test_{0} is present on {2} after {3} seconds
     """.format(int(time()), primary, replica, time_limit))
+
+
+@then('there is a "{message}" {level:w} in the {node} patroni log')
+def check_patroni_log(context, message, level, node):
+    messsages_of_level = context.pctl.read_patroni_log(node, level)
+    assert any(message in line for line in messsages_of_level),\
+        "There was no {0} {1} in the {2} patroni log".format(message, level, node)
diff --git a/patroni/__main__.py b/patroni/__main__.py
index 2a669d3f..b365bfe2 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -30,12 +30,15 @@ class Patroni(AbstractPatroniDaemon):
 
         self.version = __version__
         self.dcs = get_dcs(self.config)
+        self.request = PatroniRequest(self.config, True)
+
+        self.ensure_unique_name()
+
         self.watchdog = Watchdog(self.config)
         self.load_dynamic_configuration()
 
         self.postgresql = Postgresql(self.config['postgresql'])
         self.api = RestApiServer(self, self.config['restapi'])
-        self.request = PatroniRequest(self.config, True)
         self.ha = Ha(self)
 
         self.tags = self.get_tags()
@@ -60,6 +63,23 @@ class Patroni(AbstractPatroniDaemon):
                 logger.warning('Can not get cluster from dcs')
                 time.sleep(5)
 
+    def ensure_unique_name(self) -> None:
+        """A helper method to prevent splitbrain from operator naming error."""
+        from patroni.dcs import Member
+
+        cluster = self.dcs.get_cluster()
+        if not cluster:
+            return
+        member = cluster.get_member(self.config['name'], False)
+        if not isinstance(member, Member):
+            return
+        try:
+            _ = self.request(member, endpoint="/liveness")
+            logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
+            sys.exit(1)
+        except Exception:
+            return
+
     def get_tags(self) -> Dict[str, Any]:
         return {tag: value for tag, value in self.config.get('tags', {}).items()
                 if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index a62e7747..73755a41 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -400,8 +400,12 @@ class Consul(AbstractDCS):
 
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
+    @property
+    def _consistency(self) -> str:
+        return 'consistent' if self._ctl else self._client.consistency
+
     def _cluster_loader(self, path: str) -> Cluster:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         if results is None:
             raise NotFound
         nodes = {}
@@ -412,7 +416,7 @@ class Consul(AbstractDCS):
         return self._cluster_from_nodes(nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
-        _, results = self.retry(self._client.kv.get, path, recurse=True)
+        _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
         clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 3e7f0681..94e546ac 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -99,7 +99,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         self._dns_resolver = dns_resolver
         self.set_machines_cache_ttl(cache_ttl)
         self._machines_cache_updated = 0
-        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies',
+        kwargs = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'version_prefix',
                                              'username', 'password', 'cert', 'ca_cert') if config.get(p)}
         super(AbstractEtcdClientWithFailover, self).__init__(read_timeout=config['retry_timeout'], **kwargs)
         # For some reason python3-etcd on debian and ubuntu are not based on the latest version
@@ -443,6 +443,9 @@ class EtcdClient(AbstractEtcdClientWithFailover):
 
     ERROR_CLS = EtcdError
 
+    def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
+        super(EtcdClient, self).__init__({**config, 'version_prefix': None}, dns_resolver, cache_ttl)
+
     def __del__(self) -> None:
         try:
             self.http.clear()
@@ -722,13 +725,13 @@ class Etcd(AbstractEtcd):
         return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
     def _cluster_loader(self, path: str) -> Cluster:
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
         return self._cluster_from_nodes(result.etcd_index, nodes)
 
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
         clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
-        result = self.retry(self._client.read, path, recursive=True)
+        result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl)
         for node in result.leaves:
             key = node.key[len(result.key):].lstrip('/').split('/', 1)
             if len(key) == 2 and citus_group_re.match(key[0]):
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index b89e36a5..32d00359 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -206,8 +206,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
         self._token = None
         self._cluster_version: Tuple[int] = tuple()
-        self.version_prefix = '/v3beta'
-        super(Etcd3Client, self).__init__(config, dns_resolver, cache_ttl)
+        super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
 
         try:
             self.authenticate()
@@ -327,14 +326,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
             return retry(e)
 
     @_handle_auth_errors
-    def range(self, key: str, range_end: Union[bytes, str, None] = None,
+    def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True,
               retry: Optional[Retry] = None) -> Dict[str, Any]:
         params = build_range_request(key, range_end)
-        params['serializable'] = True  # For better performance. We can tolerate stale reads.
+        params['serializable'] = serializable  # For better performance. We can tolerate stale reads
         return self.call_rpc('/kv/range', params, retry)
 
-    def prefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]:
-        return self.range(key, prefix_range_end(key), retry)
+    def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        return self.range(key, prefix_range_end(key), serializable, retry)
 
     @_handle_auth_errors
     def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str:
@@ -595,7 +594,8 @@ class PatroniEtcd3Client(Etcd3Client):
                 self._wait_cache(self.read_timeout)
                 ret = self._kv_cache.copy()
         else:
-            ret = self._etcd3.retry(self.prefix, path).get('kvs', [])
+            serializable = not getattr(self._etcd3, '_ctl')  # use linearizable for patronictl
+            ret = self._etcd3.retry(self.prefix, path, serializable).get('kvs', [])
         for node in ret:
             node.update({'key': base64_decode(node['key']),
                          'value': base64_decode(node.get('value', '')),
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 498cfd0f..f770ac64 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -766,7 +766,7 @@ class Kubernetes(AbstractDCS):
             k8s_config.load_kube_config(context=config.get('context', 'kind-kind'))
 
         pod_ip = config.get('pod_ip')
-        self.__ips: List[str] = [] if config.get('patronictl') or not isinstance(pod_ip, str) else [pod_ip]
+        self.__ips: List[str] = [] if self._ctl or not isinstance(pod_ip, str) else [pod_ip]
         self.__ports: List[K8sObject] = []
         ports: List[Dict[str, Any]] = config.get('ports', [{}])
         for p in ports:
@@ -774,7 +774,7 @@ class Kubernetes(AbstractDCS):
             port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)})
             self.__ports.append(k8s_client.V1EndpointPort(**port))
 
-        bypass_api_service = not config.get('patronictl') and config.get('bypass_api_service')
+        bypass_api_service = not self._ctl and config.get('bypass_api_service')
         self._api = CoreV1ApiProxy(config.get('use_endpoints'), bypass_api_service)
         self._should_create_config_service = self._api.use_endpoints
         self.reload_config(config)
diff --git a/patroni/utils.py b/patroni/utils.py
index eb02c561..ee9a3d21 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -326,6 +326,21 @@ def parse_int(value: Any, base_unit: Optional[str] = None) -> Optional[int]:
         >>> parse_int('1TB', 'GB') is None
         True
 
+        >>> parse_int(50, None) == 50
+        True
+
+        >>> parse_int("51", None) == 51
+        True
+
+        >>> parse_int("nonsense", None) == None
+        True
+
+        >>> parse_int("nonsense", "kB") == None
+        True
+
+        >>> parse_int("nonsense") == None
+        True
+
         >>> parse_int(0) == 0
         True
 
diff --git a/patroni/validator.py b/patroni/validator.py
index f780f9e0..fc6cdde5 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -792,8 +792,7 @@ class IntValidator(object):
         :param value: value to be checked against the rules defined for this :class:`IntValidator` instance.
         :returns: ``True`` if *value* is valid and within the expected range.
         """
-        if self.base_unit:
-            value = parse_int(value, self.base_unit) or ""
+        value = parse_int(value, self.base_unit) or ""
         ret = isinstance(value, int)\
             and (self.min is None or value >= self.min)\
             and (self.max is None or value <= self.max)


Test patch:
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index d5c76c8b..e5785231 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -10,6 +10,7 @@ from http.server import HTTPServer
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
+from patroni.dcs import Cluster, Member
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
@@ -202,3 +203,36 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, check_psycopg)
         with patch('builtins.__import__', mock_import):
             self.assertRaises(SystemExit, check_psycopg)
+
+    def test_ensure_unique_name(self):
+        # None/empty cluster implies unique name
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=None)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        empty_cluster = Cluster.empty()
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=empty_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+        without_members = empty_cluster._asdict()
+        del without_members['members']
+
+        # Cluster with members with different names implies unique name
+        okay_cluster = Cluster(
+            members=[Member(version=1, name="distinct", session=1, data={})],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=okay_cluster)):
+            self.assertIsNone(self.p.ensure_unique_name())
+
+        # Cluster with a member with the same name that is running
+        bad_cluster = Cluster(
+            members=[Member(version=1, name="postgresql0", session=1, data={
+                "api_url": "https://127.0.0.1:8008",
+            })],
+            **without_members
+        )
+        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)):
+            # If the api of the running node cannot be reached, this implies unique name
+            with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)):
+                self.assertIsNone(self.p.ensure_unique_name())
+            # Only if the api of the running node is reachable do we throw an error
+            with patch.object(self.p, 'request', Mock()):
+                self.assertRaises(SystemExit, self.p.ensure_unique_name)