Choose synchronous nodes based on replication lag (#1786)

This commit makes it possible to configure the maximum lag (`maximum_lag_on_syncnode`) after which Patroni will "demote" the node from synchronous and replace it with another node.

The previous implementation always tried to stick to the same synchronous nodes (even if they are not optimal ones).

docs/SETTINGS.rst

@@ -15,6 +15,7 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and
 -  **ttl**: the TTL to acquire the leader lock (in seconds). Think of it as the length of time before initiation of the automatic failover process. Default value: 30
 -  **retry\_timeout**: timeout for DCS and PostgreSQL operation retries (in seconds). DCS or network issues shorter than this will not cause Patroni to demote the leader. Default value: 10
 -  **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag to be able to participate in leader election.
+-  **maximum\_lag\_on\_syncnode**: the maximum bytes a synchronous follower may lag before it is considered as an unhealthy candidate and swapped by healthy asynchronous follower. Patroni utilize the max replica lsn if there is more than one follower, otherwise it will use leader's current wal lsn. Default is -1, Patroni will not take action to swap synchronous unhealthy follower when the value is set to 0 or below. Please set the value high enough so Patroni won't swap synchrounous follower fequently during high transaction volume.
 -  **max\_timelines\_history**: maximum number of timeline history items kept in DCS.  Default value: 0. When set to 0, it keeps the full history in DCS.
 -  **master\_start\_timeout**: the amount of time a master is allowed to recover from failures before failover is triggered (in seconds). Default is 300 seconds. When set to 0 failover is done immediately after a crash is detected if possible. When using asynchronous replication a failover can cause lost transactions. Worst case failover time for master failure is: loop\_wait + master\_start\_timeout + loop\_wait, unless master\_start\_timeout is zero, in which case it's just loop\_wait. Set the value according to your durability/availability tradeoff.
 - **master\_stop\_timeout**: The number of seconds Patroni is allowed to wait when stopping Postgres and effective only when synchronous_mode is enabled. When set to > 0 and the synchronous_mode is enabled, Patroni sends SIGKILL to the postmaster if the stop operation is running for more than the value set by master_stop_timeout. Set the value according to your durability/availability tradeoff. If the parameter is not set or set <= 0, master_stop_timeout does not apply.

features/basic_replication.feature

@@ -28,6 +28,25 @@ Feature: basic replication
     When I issue a GET request to http://127.0.0.1:8009/async
     Then I receive a response code 200
 
+  Scenario: check stuck sync replica
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
+    Then I receive a response code 200
+    And I create table on postgres0
+    And table mytest is present on postgres1 after 2 seconds
+    And table mytest is present on postgres2 after 2 seconds
+    When I pause wal replay on postgres2
+    And I load data on postgres0
+    Then "sync" key in DCS has sync_standby=postgres1 after 15 seconds
+    And I resume wal replay on postgres2
+    And I sleep for 2 seconds
+    And I issue a GET request to http://127.0.0.1:8009/sync
+    Then I receive a response code 200
+    When I issue a GET request to http://127.0.0.1:8010/async
+    Then I receive a response code 200
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
+    Then I receive a response code 200
+    And I drop table on postgres0
+
   Scenario: check multi sync replication
     Given I issue a PATCH request to http://127.0.0.1:8008/config with {"synchronous_node_count": 2}
     Then I receive a response code 200

features/environment.py

@@ -194,6 +194,8 @@ class PatroniController(AbstractController):
         if custom_config is not None:
             self.recursive_update(config, custom_config)
 
+        self.recursive_update(config, {
+            'bootstrap': {'dcs': {'postgresql': {'parameters': {'wal_keep_segments': 100}}}}})
         if config['postgresql'].get('callbacks', {}).get('on_role_change'):
             config['postgresql']['callbacks']['on_role_change'] += ' ' + str(self.__PORT)
 

features/steps/basic_replication.py

@@ -33,6 +33,37 @@ def add_table(context, table_name, pg_name):
         assert False, "Error creating table {0} on {1}: {2}".format(table_name, pg_name, e)
 
 
+@step('I {action:w} wal replay on {pg_name:w}')
+def toggle_wal_replay(context, action, pg_name):
+    # pause or resume the wal replay process
+    try:
+       version = context.pctl.query(pg_name, "select pg_catalog.pg_read_file('PG_VERSION', 0, 2)").fetchone()
+       wal = version and version[0] and int(version[0].split('.')[0]) < 10 and "xlog" or "wal"
+       context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal, action))
+    except pg.Error as e:
+        assert False, "Error during {0} wal recovery on {1}: {2}".format(action, pg_name, e)
+
+
+@step('I {action:w} table on {pg_name:w}')
+def crdr_mytest(context, action, pg_name):
+    try:
+       if (action == "create"):
+           context.pctl.query(pg_name, "create table if not exists mytest(id Numeric)")
+       else:
+           context.pctl.query(pg_name, "drop table if exists mytest")
+    except pg.Error as e:
+        assert False, "Error {0} table mytest on {1}: {2}".format(action, pg_name, e)
+
+
+@step('I load data on {pg_name:w}')
+def initiate_load(context, pg_name):
+    # perform dummy load
+    try:
+       context.pctl.query(pg_name, "begin; insert into mytest select r::numeric from generate_series(1, 350000) r; commit;")
+    except pg.Error as e:
+        assert False, "Error loading test data on {0}: {1}".format(pg_name, e)
+
+
 @then('Table {table_name:w} is present on {pg_name:w} after {max_replication_delay:d} seconds')
 def table_is_present_on(context, table_name, pg_name, max_replication_delay):
     max_replication_delay *= context.timeout_multiplier

patroni/api.py

@@ -197,7 +197,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def do_PATCH_config(self):
         request = self._read_json_content()
         if request:
-            cluster = self.server.patroni.dcs.get_cluster()
+            cluster = self.server.patroni.dcs.get_cluster(True)
             if not (cluster.config and cluster.config.modify_index):
                 return self.send_error(503)
             data = cluster.config.data.copy()

patroni/config.py

@@ -60,6 +60,7 @@ class Config(object):
     __DEFAULT_CONFIG = {
         'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10,
         'maximum_lag_on_failover': 1048576,
+        'maximum_lag_on_syncnode': -1,
         'check_timeline': False,
         'master_start_timeout': 300,
         'master_stop_timeout': 0,
@@ -413,6 +414,7 @@ class Config(object):
             'synchronous_mode',
             'synchronous_mode_strict',
             'synchronous_node_count',
+            'maximum_lag_on_syncnode'
         )
 
         pg_config.update({p: config[p] for p in updated_fields if p in config})

patroni/ha.py

@@ -460,7 +460,9 @@ class Ha(object):
         if self.is_synchronous_mode():
             sync_node_count = self.patroni.config['synchronous_node_count']
             current = self.cluster.sync.leader and self.cluster.sync.members or []
-            picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, sync_node_count)
+            picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, sync_node_count,
+                                                                                self.patroni.config[
+                                                                                    'maximum_lag_on_syncnode'])
             if set(picked) != set(current):
                 # update synchronous standby list in dcs temporarily to point to common nodes in current and picked
                 sync_common = list(set(current).intersection(set(allow_promote)))
@@ -484,7 +486,9 @@ class Ha(object):
                     # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
                     time.sleep(2)
                     _, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster,
-                                                                                   sync_node_count)
+                                                                                   sync_node_count,
+                                                                                   self.patroni.config[
+                                                                                       'maximum_lag_on_syncnode'])
                 if allow_promote and set(allow_promote) != set(sync_common):
                     try:
                         cluster = self.dcs.get_cluster()

patroni/postgresql/__init__.py

@@ -962,11 +962,15 @@ class Postgresql(object):
     def _get_synchronous_commit_param(self):
         return self.query("SHOW synchronous_commit").fetchone()[0]
 
-    def pick_synchronous_standby(self, cluster, sync_node_count=1):
+    def pick_synchronous_standby(self, cluster, sync_node_count=1, sync_node_maxlag=-1):
         """Finds the best candidate to be the synchronous standby.
 
         Current synchronous standby is always preferred, unless it has disconnected or does not want to be a
         synchronous standby any longer.
+        Parameter sync_node_maxlag(maximum_lag_on_syncnode) would help swapping unhealthy sync replica incase
+        if it stops responding (or hung). Please set the value high enough so it won't unncessarily swap sync
+        standbys during high loads. Any less or equal of 0 value keep the behavior backward compatible and
+        will not swap. Please note that it will not also swap sync standbys in case where all replicas are hung.
 
         :returns tuple of candidates list and synchronous standby list.
         """
@@ -975,6 +979,7 @@ class Postgresql(object):
         members = {m.name.lower(): m for m in cluster.members}
         candidates = []
         sync_nodes = []
+        replica_list = []
         # Pick candidates based on who has higher replay/remote_write/flush lsn.
         sync_commit_par = self._get_synchronous_commit_param()
         sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(sync_commit_par, 'flush')
@@ -984,17 +989,22 @@ class Postgresql(object):
         # receiving changes faster than the sync member (very rare but possible). Such cases would
         # trigger sync standby member swapping frequently and the sort on sync_state desc should
         # help in keeping the query result consistent.
-        for app_name, state, sync_state in self.query(
-                "SELECT pg_catalog.lower(application_name), state, sync_state"
+        for app_name, sync_state, replica_lsn in self.query(
+                "SELECT pg_catalog.lower(application_name), sync_state, pg_{2}_{1}_diff({0}_{1}, '0/0')::bigint"
                 " FROM pg_catalog.pg_stat_replication"
                 " WHERE state = 'streaming'"
-                " ORDER BY sync_state DESC, {0}_{1} DESC".format(sort_col, self.lsn_name)):
+                " ORDER BY sync_state DESC, {0}_{1} DESC".format(sort_col, self.lsn_name, self.wal_name)):
             member = members.get(app_name)
-            if not member or member.tags.get('nosync', False):
-                continue
-            candidates.append(member.name)
-            if sync_state == 'sync':
-                sync_nodes.append(member.name)
+            if member and not member.tags.get('nosync', False):
+                replica_list.append((member.name, sync_state, replica_lsn))
+
+        max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else int(str(self.last_operation()))
+
+        for app_name, sync_state, replica_lsn in replica_list:
+            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
+                candidates.append(app_name)
+                if sync_state == 'sync':
+                    sync_nodes.append(app_name)
             if len(candidates) >= sync_node_count:
                 break
 

tests/test_postgresql.py

@@ -615,32 +615,32 @@ class TestPostgresql(BaseTestPostgresql):
 
         with patch.object(Postgresql, "query", side_effect=[
                     mock_cursor,
-                    [(self.leadermem.name, 'streaming', 'sync'),
-                     (self.me.name, 'streaming', 'async'),
-                     (self.other.name, 'streaming', 'async')]
+                    [(self.leadermem.name, 'sync', 1),
+                     (self.me.name, 'async', 2),
+                     (self.other.name, 'async', 2)]
                 ]):
             self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [self.leadermem.name]))
 
         with patch.object(Postgresql, "query", side_effect=[
                     mock_cursor,
-                    [(self.leadermem.name, 'streaming', 'potential'),
-                     (self.me.name, 'streaming', 'async'),
-                     (self.other.name, 'streaming', 'async')]
+                    [(self.leadermem.name, 'potential', 1),
+                     (self.me.name, 'async', 2),
+                     (self.other.name, 'async', 2)]
                 ]):
             self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], []))
 
         with patch.object(Postgresql, "query", side_effect=[
                     mock_cursor,
-                    [(self.me.name, 'streaming', 'async'),
-                     (self.other.name, 'streaming', 'async')]
+                    [(self.me.name, 'async', 1),
+                     (self.other.name, 'async', 2)]
                 ]):
             self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))
 
         with patch.object(Postgresql, "query", side_effect=[
                     mock_cursor,
-                    [('missing', 'streaming', 'sync'),
-                     (self.me.name, 'streaming', 'async'),
-                     (self.other.name, 'streaming', 'async')]
+                    [('missing', 'sync', 1),
+                     (self.me.name, 'async', 2),
+                     (self.other.name, 'async', 3)]
                 ]):
             self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))