Feature: failover priority (#2780)

The priority is configured with `failover_priority` tag. Possible values are from `0` till infinity, where `0` means that the node will never become the leader, which is the same as `nofailover` tag set to `true`. As a result, in the configuration file one should set only one of `failover_priority` or `nofailover` tags.

The failover priority kicks in only when there are more than one node have the same receive/replay LSN and are ahead of other nodes in the cluster. In this case the node with higher value of `failover_priority` is preferred. If there is a node with higher values of receive/replay LSN, it will become the new leader even if it has lower value of `failover_priority` (except when priority is set to 0).

Close https://github.com/zalando/patroni/issues/2759

docs/yaml_configuration.rst

@@ -368,11 +368,15 @@ Watchdog
 
 Tags
 ----
--  **nofailover**: ``true`` or ``false``, controls whether this node is allowed to participate in the leader race and become a leader. Defaults to ``false``
 -  **clonefrom**: ``true`` or ``false``. If set to ``true`` other nodes might prefer to use this node for bootstrap (take ``pg_basebackup`` from). If there are several nodes with ``clonefrom`` tag set to ``true`` the node to bootstrap from will be chosen randomly. The default value is ``false``.
 -  **noloadbalance**: ``true`` or ``false``. If set to ``true`` the node will return HTTP Status Code 503 for the ``GET /replica`` REST API health-check and therefore will be excluded from the load-balancing. Defaults to ``false``.
 -  **replicatefrom**: The IP address/hostname of another replica. Used to support cascading replication.
 -  **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
+-  **nofailover**: ``true`` or ``false``, controls whether this node is allowed to participate in the leader race and become a leader. Defaults to ``false``, meaning this node _can_ participate in leader races. 
+-  **failover_priority**: integer, controls the priority that this node should have during failover. Nodes with higher priority will be preferred over lower priority nodes if they received/replayed the same amount of WAL. However, nodes with higher values of receive/replay LSN are preferred regardless of their priority. If the ``failover_priority`` is 0 or negative - such node is not allowed to participate in the leader race and to become a leader (similar to ``nofailover: true``).
+
+.. warning::
+   Provide only one of ``nofailover`` or ``failover_priority``. Providing ``nofailover: true`` is the same as ``failover_priority: 0``, and providing ``nofailover: false`` will give the node priority 1. 
 
 In addition to these predefined tags, you can also add your own ones:
 

features/basic_replication.feature

@@ -82,4 +82,4 @@ Feature: basic replication
   @reject-duplicate-name
   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
+    Then there is one of ["Can't start; there is already a node named 'postgres0' running"] CRITICAL in the dup-postgres0 patroni log after 5 seconds

features/priority_failover.feature

@@ -0,0 +1,23 @@
+Feature: priority replication
+  We should check that we can give nodes priority during failover
+
+  Scenario: check failover priority 0 prevents leaderships
+    Given I configure and start postgres0 with a tag failover_priority 1
+    And I configure and start postgres1 with a tag failover_priority 0
+    Then replication works from postgres0 to postgres1 after 20 seconds
+    When I shut down postgres0
+    And I sleep for 5 seconds
+    Then postgres1 role is the secondary after 10 seconds
+    And there is one of ["following a different leader because I am not allowed to promote"] INFO in the postgres1 patroni log after 5 seconds
+    Given I start postgres0
+    Then postgres0 role is the primary after 10 seconds
+
+  Scenario: check higher failover priority is respected
+    Given I configure and start postgres2 with a tag failover_priority 1
+    And I configure and start postgres3 with a tag failover_priority 2
+    Then replication works from postgres0 to postgres2 after 20 seconds
+    And replication works from postgres0 to postgres3 after 20 seconds
+    When I shut down postgres0
+    And I sleep for 5 seconds
+    Then postgres3 role is the primary after 10 seconds
+    And there is one of ["postgres3 has equally tolerable WAL position and priority 2, while this node has priority 1","Wal position of postgres3 is ahead of my wal position"] INFO in the postgres2 patroni log after 5 seconds

features/steps/basic_replication.py

@@ -1,3 +1,4 @@
+import json
 import patroni.psycopg as pg
 
 from behave import step, then
@@ -113,8 +114,15 @@ def replication_works(context, primary, replica, time_limit):
     """.format(str(time()).replace('.', '_').replace(',', '_'), 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)
+@then('there is one of {message_list} {level:w} in the {node} patroni log after {timeout:d} seconds')
+def check_patroni_log(context, message_list, level, node, timeout):
+    timeout *= context.timeout_multiplier
+    message_list = json.loads(message_list)
+
+    for _ in range(int(timeout)):
+        messsages_of_level = context.pctl.read_patroni_log(node, level)
+        if any(any(message in line for line in messsages_of_level) for message in message_list):
+            break
+        time.sleep(1)
+    else:
+        assert False, f"There were none of {message_list} {level} in the {node} patroni log after {timeout} seconds"

patroni/config.py

@@ -293,6 +293,7 @@ class Config(object):
         if validator:  # patronictl uses validator=None and we don't want to load anything from local cache in this case
             self._load_cache()
         self._cache_needs_saving = False
+        self._validate_failover_tags()
 
     @property
     def config_file(self) -> Optional[str]:
@@ -959,3 +960,24 @@ class Config(object):
         :returns: :class:`GlobalConfig` object.
         """
         return get_global_config(cluster, self._dynamic_configuration)
+
+    def _validate_failover_tags(self) -> None:
+        """Check ``nofailover``/``failover_priority`` config and warn user if it's contradictory.
+
+        .. note::
+          To preserve sanity (and backwards compatibility) the ``nofailover`` tag will still exist. A contradictory
+          configuration is one where ``nofailover`` is ``True`` but ``failover_priority > 0``, or where
+          ``nofailover`` is ``False``, but ``failover_priority <= 0``. Essentially, ``nofailover`` and
+          ``failover_priority`` are communicating different things.
+          This checks for this edge case (which is a misconfiguration on the part of the user) and warns them.
+          The behaviour is as if ``failover_priority`` were not provided (i.e ``nofailover`` is the
+          bedrock source of truth)
+        """
+        tags = self.get('tags', {})
+        nofailover_tag = tags.get('nofailover')
+        failover_priority_tag = parse_int(tags.get('failover_priority'))
+        if failover_priority_tag is not None \
+                and (nofailover_tag is True and failover_priority_tag > 0
+                     or nofailover_tag is False and failover_priority_tag <= 0):
+            logger.warning('Conflicting configuration between nofailover: %s and failover_priority: %s. '
+                           'Defaulting to nofailover: %s', nofailover_tag, failover_priority_tag, nofailover_tag)

patroni/ha.py

@@ -1009,6 +1009,15 @@ class Ha(object):
                     if not self.sync_mode_is_active() or not self.cluster.sync.leader_matches(st.member.name):
                         return False
                     logger.info('Ignoring the former leader being ahead of us')
+                if my_wal_position == st.wal_position and self.patroni.failover_priority < st.failover_priority:
+                    # There's a higher priority non-lagging replica
+                    logger.info(
+                        '%s has equally tolerable WAL position and priority %s, while this node has priority %s',
+                        st.member.name,
+                        st.failover_priority,
+                        self.patroni.failover_priority,
+                    )
+                    return False
         return True
 
     def is_failover_possible(self, *, cluster_lsn: int = 0, exclude_failover_candidate: bool = False) -> bool:

patroni/tags.py

@@ -3,6 +3,8 @@ import abc
 
 from typing import Any, Dict, Optional
 
+from patroni.utils import parse_int
+
 
 class Tags(abc.ABC):
     """An abstract class that encapsulates all the ``tags`` logic.
@@ -45,8 +47,29 @@ class Tags(abc.ABC):
 
     @property
     def nofailover(self) -> bool:
-        """``True`` if ``nofailover`` is ``True``, else ``False``."""
-        return bool(self.tags.get('nofailover', False))
+        """Common logic for obtaining the value of ``nofailover`` from ``tags`` if defined.
+
+        If ``nofailover`` is not defined, this methods returns ``True`` if ``failover_priority`` is non-positive,
+        ``False`` otherwise.
+        """
+        from_tags = self.tags.get('nofailover')
+        if from_tags is not None:
+            # Value of `nofailover` takes precedence over `failover_priority`
+            return bool(from_tags)
+        failover_priority = parse_int(self.tags.get('failover_priority'))
+        return failover_priority is not None and failover_priority <= 0
+
+    @property
+    def failover_priority(self) -> int:
+        """Common logic for obtaining the value of ``failover_priority`` from ``tags`` if defined.
+
+        If ``nofailover`` is defined as ``True``, this will return ``0``. Otherwise, it will return the value of
+        ``failover_priority``, defaulting to ``1`` if it's not defined or invalid.
+        """
+        from_tags = self.tags.get('nofailover')
+        failover_priority = parse_int(self.tags.get('failover_priority'))
+        failover_priority = 1 if failover_priority is None else failover_priority
+        return 0 if from_tags else failover_priority
 
     @property
     def noloadbalance(self) -> bool:

patroni/validator.py

@@ -379,6 +379,37 @@ class Or(object):
         self.args = args
 
 
+class AtMostOne(object):
+    """Mark that at most one option from a :class:`Case` can be suplied.
+
+    Represents a list of possible configuration options in a given scope, where at most one can actually
+    be provided.
+
+    .. note::
+
+        It should be used together with a :class:`Case` object.
+    """
+
+    def __init__(self, *args: str) -> None:
+        """Create a :class`AtMostOne` object.
+
+        :param `*args`: any arguments that the caller wants to be stored in this :class:`Or` object.
+
+        :Example:
+
+            .. code-block:: python
+
+                AtMostOne("nofailover", "failover_priority"): Case({
+                    "nofailover": bool,
+                    "failover_priority": IntValidator(min=0, raise_assert=True),
+                })
+
+        The :class`AtMostOne` object is used to define that at most one of ``nofailover`` and
+        ``failover_priority`` can be provided.
+        """
+        self.args = args
+
+
 class Optional(object):
     """Mark a configuration option as optional.
 
@@ -671,6 +702,9 @@ class Schema(object):
         # One key in `validator` attribute (`key` variable) can be mapped to one or more keys in `data` attribute (`d`
         # variable), depending on the `key` type.
         for key in self.validator.keys():
+            if isinstance(key, AtMostOne) and len(list(self._data_key(key))) > 1:
+                yield Result(False, f"Multiple of {key.args} provided")
+                continue
             for d in self._data_key(key):
                 if d not in self.data and not isinstance(key, Optional):
                     yield Result(False, "is not defined.", path=d)
@@ -680,7 +714,7 @@ class Schema(object):
                     if d not in self.data and isinstance(key, Optional):
                         self.data[d] = key.default
                     validator = self.validator[key]
-                    if isinstance(key, Or) and isinstance(self.validator[key], Case):
+                    if isinstance(key, (Or, AtMostOne)) and isinstance(self.validator[key], Case):
                         validator = self.validator[key]._schema[d]
                     # In this loop we may be calling a new `Schema` either over an intermediate node in the tree, or
                     # over a leaf node. In the latter case the recursive calls in the given path will finish.
@@ -715,7 +749,7 @@ class Schema(object):
                 max_level = v.level
                 yield Result(v.status, v.error, path=v.path, level=v.level, data=v.data)
 
-    def _data_key(self, key: Union[str, Optional, Or]) -> Iterator[str]:
+    def _data_key(self, key: Union[str, Optional, Or, AtMostOne]) -> Iterator[str]:
         """Map a key from the ``validator`` dictionary to the corresponding key(s) in the ``data`` dictionary.
 
         :param key: key from the ``validator`` attribute.
@@ -735,15 +769,23 @@ class Schema(object):
         elif isinstance(key, Or):
             # At least one of the `Or` entries should be available in the `data` dictionary. If we find at least one of
             # them in `data`, then we return all found entries so the caller method can validate them all.
-            if any([i in self.data for i in key.args]):
-                for i in key.args:
-                    if i in self.data:
-                        yield i
+            if any([item in self.data for item in key.args]):
+                for item in key.args:
+                    if item in self.data:
+                        yield item
             # If none of the `Or` entries is available in the `data` dictionary, then we return all entries so the
             # caller method will issue errors that they are all absent.
             else:
-                for i in key.args:
-                    yield i
+                for item in key.args:
+                    yield item
+        # If the key was defined as a `AtMostOne` object in `validator` attribute, then each of its values
+        # are the keys to access the `data` dictionary.
+        elif isinstance(key, AtMostOne):
+            # Yield back all of the entries from the `data` dictionary, each will be validated and then counted
+            # to inform us if we've provided too many
+            for item in key.args:
+                if item in self.data:
+                    yield item
 
 
 def _get_type_name(python_type: Any) -> str:
@@ -1056,7 +1098,10 @@ schema = Schema({
         Optional("safety_margin"): int
     },
     Optional("tags"): {
-        Optional("nofailover"): bool,
+        AtMostOne("nofailover", "failover_priority"): Case({
+            "nofailover": bool,
+            "failover_priority": IntValidator(min=0, raise_assert=True),
+        }),
         Optional("clonefrom"): bool,
         Optional("noloadbalance"): bool,
         Optional("replicatefrom"): str,

tests/test_config.py

@@ -151,6 +151,53 @@ class TestConfig(unittest.TestCase):
     def test_invalid_path(self):
         self.assertRaises(ConfigParseError, Config, 'postgres0')
 
+    @patch.object(Config, 'get')
+    @patch('patroni.config.logger')
+    def test__validate_failover_tags(self, mock_logger, mock_get):
+        """Ensures that only one of `nofailover` or `failover_priority` can be provided"""
+        mock_logger.warning.reset_mock()
+        config = Config("postgres0.yml")
+        # Providing one of `nofailover` or `failover_priority` is fine
+        just_nofailover = {"nofailover": True}
+        mock_get.side_effect = [just_nofailover] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_not_called()
+        just_failover_priority = {"failover_priority": 1}
+        mock_get.side_effect = [just_failover_priority] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_not_called()
+        # Providing both `nofailover` and `failover_priority` is fine if consistent
+        consistent_false = {"nofailover": False, "failover_priority": 1}
+        mock_get.side_effect = [consistent_false] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_not_called()
+        consistent_true = {"nofailover": True, "failover_priority": 0}
+        mock_get.side_effect = [consistent_true] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_not_called()
+        # Providing both inconsistently should log a warning
+        inconsistent_false = {"nofailover": False, "failover_priority": 0}
+        mock_get.side_effect = [inconsistent_false] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_called_once_with(
+            'Conflicting configuration between nofailover: %s and failover_priority: %s.'
+            + ' Defaulting to nofailover: %s',
+            False,
+            0,
+            False
+        )
+        mock_logger.warning.reset_mock()
+        inconsistent_true = {"nofailover": True, "failover_priority": 1}
+        mock_get.side_effect = [inconsistent_true] * 2
+        self.assertIsNone(config._validate_failover_tags())
+        mock_logger.warning.assert_called_once_with(
+            'Conflicting configuration between nofailover: %s and failover_priority: %s.'
+            + ' Defaulting to nofailover: %s',
+            True,
+            1,
+            True
+        )
+
     def test__process_postgresql_parameters(self):
         expected_params = {
             'f.oo': 'bar',  # not in ConfigHandler.CMDLINE_OPTIONS

tests/test_ha.py

@@ -94,11 +94,12 @@ def get_cluster_initialized_with_leader_and_failsafe():
 
 def get_node_status(reachable=True, in_recovery=True, dcs_last_seen=0,
                     timeline=2, wal_position=10, nofailover=False,
-                    watchdog_failed=False):
+                    watchdog_failed=False, failover_priority=1):
     def fetch_node_status(e):
         tags = {}
         if nofailover:
             tags['nofailover'] = True
+        tags['failover_priority'] = failover_priority
         return _MemberStatus(e, reachable, in_recovery, wal_position,
                              {'tags': tags, 'watchdog_failed': watchdog_failed,
                               'dcs_last_seen': dcs_last_seen, 'timeline': timeline})
@@ -153,6 +154,7 @@ zookeeper:
                                   'postmaster_start_time': str(postmaster_start_time)}
         self.watchdog = Watchdog(self.config)
         self.request = lambda *args, **kwargs: requests_get(args[0].api_url, *args[1:], **kwargs)
+        self.failover_priority = 1
 
 
 def run_async(self, func, args=()):
@@ -1036,6 +1038,11 @@ class TestHa(PostgresInit):
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.fetch_node_status = get_node_status(in_recovery=False)  # accessible, not in_recovery
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.ha.fetch_node_status = get_node_status(failover_priority=2)  # accessible, in_recovery, higher priority
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        # if there is a higher-priority node but it has a lower WAL position then this node should race
+        self.ha.fetch_node_status = get_node_status(failover_priority=6, wal_position=9)
+        self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.fetch_node_status = get_node_status(wal_position=11)  # accessible, in_recovery, wal position ahead
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us
@@ -1051,7 +1058,9 @@ class TestHa(PostgresInit):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.patroni.nofailover = True
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        self.ha.patroni.nofailover = False
+        self.ha.patroni.nofailover = None
+        self.ha.patroni.failover_priority = 0
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
 
     def test_fetch_node_status(self):
         member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})

tests/test_patroni.py

@@ -179,10 +179,42 @@ class TestPatroni(unittest.TestCase):
         self.assertTrue(self.p.noloadbalance)
 
     def test_nofailover(self):
-        self.p.tags['nofailover'] = True
-        self.assertTrue(self.p.nofailover)
-        self.p.tags['nofailover'] = None
-        self.assertFalse(self.p.nofailover)
+        for (nofailover, failover_priority, expected) in [
+            # Without any tags, default is False
+            (None, None, False),
+            # Setting `nofailover: True` has precedence
+            (True, 0, True),
+            (True, 1, True),
+            # Similarly, setting `nofailover: False` has precedence
+            (False, 0, False),
+            (False, 1, False),
+            # Only when we have `nofailover: None` should we got based on priority
+            (None, 0, True),
+            (None, 1, False),
+        ]:
+            with self.subTest(nofailover=nofailover, failover_priority=failover_priority, expected=expected):
+                self.p.tags['nofailover'] = nofailover
+                self.p.tags['failover_priority'] = failover_priority
+                self.assertEqual(self.p.nofailover, expected)
+
+    def test_failover_priority(self):
+        for (nofailover, failover_priority, expected) in [
+            # Without any tags, default is 1
+            (None, None, 1),
+            # Setting `nofailover: True` has precedence (value 0)
+            (True, 0, 0),
+            (True, 1, 0),
+            # Setting `nofailover: False` and `failover_priority: None` gives 1
+            (False, None, 1),
+            # Normal function of failover_priority
+            (None, 0, 0),
+            (None, 1, 1),
+            (None, 2, 2),
+        ]:
+            with self.subTest(nofailover=nofailover, failover_priority=failover_priority, expected=expected):
+                self.p.tags['nofailover'] = nofailover
+                self.p.tags['failover_priority'] = failover_priority
+                self.assertEqual(self.p.failover_priority, expected)
 
     def test_replicatefrom(self):
         self.assertIsNone(self.p.replicatefrom)

tests/test_validator.py

@@ -325,3 +325,35 @@ class TestValidator(unittest.TestCase):
         output = "\n".join(errors)
         self.assertEqual(['postgresql.bin_dir', 'postgresql.bin_name.postgres', 'raft.bind_addr', 'raft.self_addr'],
                          parse_output(output))
+
+    def test_one_of(self, _, __):
+        c = copy.deepcopy(config)
+        # Providing neither is fine
+        del c["tags"]["nofailover"]
+        errors = schema(c)
+        self.assertNotIn("tags  Multiple of ('nofailover', 'failover_priority') provided", errors)
+        # Just nofailover is fine
+        c["tags"]["nofailover"] = False
+        errors = schema(c)
+        self.assertNotIn("tags  Multiple of ('nofailover', 'failover_priority') provided", errors)
+        # Just failover_priority is fine
+        del c["tags"]["nofailover"]
+        c["tags"]["failover_priority"] = 1
+        errors = schema(c)
+        self.assertNotIn("tags  Multiple of ('nofailover', 'failover_priority') provided", errors)
+        # Providing both is not fine
+        c["tags"]["nofailover"] = False
+        errors = schema(c)
+        self.assertIn("tags  Multiple of ('nofailover', 'failover_priority') provided", errors)
+
+    def test_failover_priority_int(self, *args):
+        c = copy.deepcopy(config)
+        del c["tags"]["nofailover"]
+        c["tags"]["failover_priority"] = 'a string'
+        errors = schema(c)
+        self.assertIn('tags.failover_priority a string is not an integer', errors)
+        c = copy.deepcopy(config)
+        del c["tags"]["nofailover"]
+        c["tags"]["failover_priority"] = -6
+        errors = schema(c)
+        self.assertIn('tags.failover_priority -6 didn\'t pass validation: Wrong value', errors)