Fix callbacks behavior (mostly for standby cluster) (#998)

First of all, this patch changes the behavior of `on_start`/`on_restart` callbacks, they will be called only when postgres is started or restarted without role changes. In case if the member is promoted or demoted only the `on_role_change` callback will be executed. `on_role_change` was never called for standby leader, only `on_start`/`on_restart` and with a wrong role argument.
Before that `on_role_change` was never called for standby leader, only `on_start`/`on_restart` and with a wrong role argument.

In addition to that, the REST API will return standby_leader role for the leader of the standby cluster.

Closes https://github.com/zalando/patroni/issues/988

docs/SETTINGS.rst

@@ -139,10 +139,10 @@ PostgreSQL
         -  **password**: replication password; the user will be created during initialization.
 -  **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
         -  **on\_reload**: run this script when configuration reload is triggered.
-        -  **on\_restart**: run this script when the cluster restarts.
-        -  **on\_role\_change**: run this script when the cluster is being promoted or demoted.
-        -  **on\_start**: run this script when the cluster starts.
-        -  **on\_stop**: run this script when the cluster stops.
+        -  **on\_restart**: run this script when the postgres restarts (without changing role).
+        -  **on\_role\_change**: run this script when the postgres is being promoted or demoted.
+        -  **on\_start**: run this script when the postgres starts.
+        -  **on\_stop**: run this script when the postgres stops.
 -  **connect\_address**: IP address + port through which Postgres is accessible from other nodes and applications.
 -  **create\_replica\_methods**: an ordered list of the create methods for turning a Patroni node into a new replica.
    "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its

features/environment.py

@@ -111,9 +111,9 @@ class PatroniController(AbstractController):
         with open(os.path.join(self._data_dir, 'label'), 'w') as f:
             f.write(content)
 
-    def read_label(self):
+    def read_label(self, label):
         try:
-            with open(os.path.join(self._data_dir, 'label'), 'r') as f:
+            with open(os.path.join(self._data_dir, label), 'r') as f:
                 return f.read().strip()
         except IOError:
             return None

features/standby_cluster.feature

@@ -2,11 +2,11 @@ Feature: standby cluster
   Scenario: check permanent logical slots are preserved on failover/switchover
     Given I start postgres1
     Then postgres1 is a leader after 10 seconds
-    And I sleep for 2 seconds
+    And I sleep for 3 seconds
     When I issue a PATCH request to http://127.0.0.1:8009/config with {"loop_wait": 2, "slots": {"pm_1": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
     Then I receive a response code 200
     And Response on GET http://127.0.0.1:8009/config contains slots after 10 seconds
-    And I sleep for 2 seconds
+    And I sleep for 3 seconds
     When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
     Then I receive a response code 200
     When I start postgres0 with callback configured
@@ -14,7 +14,7 @@ Feature: standby cluster
     And replication works from postgres1 to postgres0 after 15 seconds
     When I shut down postgres1
     Then postgres0 is a leader after 10 seconds
-    And I sleep for 2 seconds
+    And "members/postgres0" key in DCS has role=master after 3 seconds
     When I issue a GET request to http://127.0.0.1:8008/
     Then I receive a response code 200
     And there is a label with "test_logical" in postgres0 data directory
@@ -24,6 +24,12 @@ Feature: standby cluster
     Then postgres1 is a leader of batman1 after 10 seconds
     When I add the table foo to postgres0
     Then table foo is present on postgres1 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8009/master
+    Then I receive a response code 200
+    When I issue a GET request to http://127.0.0.1:8009/standby_leader
+    Then I receive a response code 200
+    And I receive a response role standby_leader
+    And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
     And table foo is present on postgres2 after 20 seconds
@@ -31,4 +37,11 @@ Feature: standby cluster
   Scenario: check failover
     When I kill postgres1
     And I kill postmaster on postgres1
-    Then postgres2 is replicating from postgres0 after 20 seconds
+    Then postgres2 is replicating from postgres0 after 32 seconds
+    When I issue a GET request to http://127.0.0.1:8010/master
+    Then I receive a response code 200
+    When I issue a GET request to http://127.0.0.1:8010/standby_leader
+    Then I receive a response code 200
+    And I receive a response role standby_leader
+    And replication works from postgres0 to postgres2 after 15 seconds
+    And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory

features/steps/cascading_replication.py

@@ -9,9 +9,10 @@ def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value):
     return context.pctl.start(name, custom_config={'tags': {tag_name: tag_value}})
 
 
-@then('There is a label with "{content:w}" in {name:w} data directory')
-def check_label(context, content, name):
-    label = context.pctl.read_label(name)
+@then('There is a {label} with "{content}" in {name:w} data directory')
+def check_label(context, label, content, name):
+    label = context.pctl.read_label(name, label)
+    label = label.replace('\n', '\\n')
     assert label == content, "{0} is not equal to {1}".format(label, content)
 
 

features/steps/standby_cluster.py

@@ -9,6 +9,8 @@ SELECT * FROM pg_catalog.pg_stat_replication
 WHERE application_name = '{0}'
 """
 
+callback = "bash -c 'echo \"${*: -3:1} ${*: -2:1} ${*: -1:1}\" >> data/$1/$1_cb.log' -- "
+
 
 @step('I start {name:w} with callback configured')
 def start_patroni_with_callbacks(context, name):
@@ -24,12 +26,15 @@ def start_patroni_with_callbacks(context, name):
 @step('I start {name:w} in a cluster {cluster_name:w}')
 def start_patroni(context, name, cluster_name):
     return context.pctl.start(name, custom_config={
-        "scope": cluster_name
+        "scope": cluster_name,
+        "postgresql": {
+            "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')}
+        }
     })
 
 
 @step('I start {name:w} in a standby cluster {cluster_name:w} as a clone of {name2:w}')
-def start_patroni_stanby_cluster(context, name, cluster_name, name2):
+def start_patroni_standby_cluster(context, name, cluster_name, name2):
     # we need to remove patroni.dynamic.json in order to "bootstrap" standby cluster with existing PGDATA
     os.unlink(os.path.join(context.pctl._processes[name]._data_dir, 'patroni.dynamic.json'))
     port = context.pctl._processes[name2]._connkwargs.get('port')
@@ -37,12 +42,18 @@ def start_patroni_stanby_cluster(context, name, cluster_name, name2):
         "scope": cluster_name,
         "bootstrap": {
             "dcs": {
+                "ttl": 20,
+                "loop_wait": 2,
+                "retry_timeout": 5,
                 "standby_cluster": {
                     "host": "localhost",
                     "port": port,
                     "primary_slot_name": "pm_1",
                 }
             }
+        },
+        "postgresql": {
+            "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')}
         }
     })
     return context.pctl.start(name)
@@ -60,8 +71,8 @@ def check_replication_status(context, pg_name1, pg_name2, timeout):
         )
 
         if cur and len(cur.fetchall()) != 0:
-            return True
+            break
 
         time.sleep(1)
-
-    return False
+    else:
+        assert False, "{0} is not replicating from {1} after {2} seconds".format(pg_name1, pg_name2, timeout)

patroni/api.py

@@ -446,6 +446,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 })
             }
 
+            if result['role'] == 'replica' and self.server.patroni.ha.is_standby_cluster():
+                result['role'] = self.server.patroni.postgresql.role
+
             if row[1] > 0:
                 result['timeline'] = row[1]
             else:

patroni/ha.py

@@ -130,7 +130,7 @@ class Ha(object):
             self.old_cluster = cluster
         self.cluster = cluster
 
-        if self.cluster.is_unlocked() or self.cluster.leader.name != self.state_handler.name:
+        if not self.has_lock(False):
             self.set_is_leader(False)
 
         self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline
@@ -154,9 +154,10 @@ class Ha(object):
             self.watchdog.keepalive()
         return ret
 
-    def has_lock(self):
+    def has_lock(self, info=True):
         lock_owner = self.cluster.leader and self.cluster.leader.name
-        logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name)
+        if info:
+            logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name)
         return lock_owner == self.state_handler.name
 
     def get_effective_tags(self):
@@ -313,6 +314,7 @@ class Ha(object):
 
         self.load_cluster_from_dcs()
 
+        role = 'replica'
         if self.is_standby_cluster() or not self.has_lock():
             if not self.state_handler.rewind_executed:
                 self.state_handler.trigger_check_diverged_lsn()
@@ -321,6 +323,7 @@ class Ha(object):
 
             if self.has_lock():  # in standby cluster
                 msg = "starting as a standby leader because i had the session lock"
+                role = 'standby_leader'
                 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"
@@ -335,15 +338,13 @@ class Ha(object):
         self.recovering = True
 
         self._async_executor.schedule('restarting after failure')
-        self._async_executor.run_async(self.state_handler.follow, (node_to_follow, timeout))
+        self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role, timeout))
         return msg
 
     def _get_node_to_follow(self, cluster):
         # determine the node to follow. If replicatefrom tag is set,
         # 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.is_standby_cluster() and (is_leader or self.cluster.is_unlocked()):
+        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
             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)
@@ -375,9 +376,17 @@ class Ha(object):
         if self._handle_rewind_or_reinitialize():
             return self._async_executor.scheduled_action
 
-        if not self.state_handler.check_recovery_conf(node_to_follow):
+        role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
+        # It might happen that leader key in the standby cluster references non-exiting member.
+        # In this case it is safe to continue running without changing recovery.conf
+        if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url):
+            return 'continue following the old known standby leader'
+        elif not self.state_handler.check_recovery_conf(node_to_follow):
             self._async_executor.schedule('changing primary_conninfo and restarting')
-            self._async_executor.run_async(self.state_handler.follow, (node_to_follow,))
+            self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role))
+        elif role == 'standby_leader' and self.state_handler.role != role:
+            self.state_handler.set_role(role)
+            self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
 
         return follow_reason
 
@@ -495,9 +504,7 @@ class Ha(object):
                 self.dcs.set_history_value(json.dumps(history, separators=(',', ':')))
 
     def enforce_follow_remote_master(self, message):
-        self.state_handler.set_role('standby_leader')
         demote_reason = 'cannot be a real master in standby cluster'
-
         return self.follow(demote_reason, message)
 
     def enforce_master_role(self, message, promote_message):
@@ -854,7 +861,7 @@ class Ha(object):
                     # 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
-                    msg = 'promoted self to a standby leader because i had the session lock'
+                    msg = 'promoted self to a standby leader by acquiring session lock'
                     return self.enforce_follow_remote_master(msg)
                 else:
                     return self.enforce_master_role(
@@ -900,7 +907,9 @@ class Ha(object):
                     # 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.
-                    msg = 'no action.  i am the standby leader with the lock'
+                    msg = 'no action.  i am the standby leader with the lock' \
+                          if self.state_handler.role == 'standby_leader' else \
+                          'promoted self to a standby leader because i had the session lock'
                     return self.enforce_follow_remote_master(msg)
                 else:
                     return self.enforce_master_role(
@@ -919,6 +928,9 @@ class Ha(object):
                     return 'not promoting because failed to update leader lock in DCS'
         else:
             logger.info('does not have lock')
+        if self.is_standby_cluster():
+            return self.follow('cannot be a real master in standby cluster',
+                               'no action.  i am a secondary and i am following a standby leader', refresh=False)
         return self.follow('demoting self because i do not have the lock and i was a leader',
                            'no action.  i am a secondary and i am following a leader', refresh=False)
 
@@ -1050,7 +1062,7 @@ class Ha(object):
             if self.cluster.is_unlocked():
                 return 'Cluster has no leader, can not reinitialize'
 
-            if self.cluster.leader.name == self.state_handler.name:
+            if self.has_lock(False):
                 return 'I am the leader, can not reinitialize'
 
         if force:
@@ -1333,13 +1345,11 @@ class Ha(object):
                              (" Leaving watchdog running." if self.watchdog.is_running else ""))
 
     def watch(self, timeout):
-        cluster = self.cluster
         # watch on leader key changes if the postgres is running and leader is known and current node is not lock owner
-        if not self._async_executor.busy and cluster and cluster.leader \
-                and cluster.leader.name != self.state_handler.name:
-            leader_index = cluster.leader.index
-        else:
+        if self._async_executor.busy or self.cluster.is_unlocked() or self.has_lock(False):
             leader_index = None
+        else:
+            leader_index = self.cluster.leader.index
 
         return self.dcs.watch(leader_index, timeout)
 

patroni/postgresql.py

@@ -30,6 +30,7 @@ ACTION_ON_STOP = "on_stop"
 ACTION_ON_RESTART = "on_restart"
 ACTION_ON_RELOAD = "on_reload"
 ACTION_ON_ROLE_CHANGE = "on_role_change"
+ACTION_NOOP = "noop"
 
 STATE_RUNNING = 'running'
 STATE_REJECT = 'rejecting connections'
@@ -697,10 +698,11 @@ class Postgresql(object):
                     # if basebackup succeeds, exit with success
                     break
             else:
-                if not self.data_directory_empty() and not self.config.get(replica_method, {}).get('keep_data', False):
-                    self.remove_data_directory()
-                else:
-                    logger.info('Leaving data directory uncleaned')
+                if not self.data_directory_empty():
+                    if self.config.get(replica_method, {}).get('keep_data', False):
+                        logger.info('Leaving data directory uncleaned')
+                    else:
+                        self.remove_data_directory()
 
                 cmd = replica_method
                 method_config = {}
@@ -870,7 +872,7 @@ class Postgresql(object):
                 self._pending_restart = True
         return effective_configuration
 
-    def start(self, timeout=None, block_callbacks=False, task=None):
+    def start(self, timeout=None, task=None, block_callbacks=False, role=None):
         """Start PostgreSQL
 
         Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
@@ -890,7 +892,7 @@ class Postgresql(object):
         if not block_callbacks:
             self.__cb_pending = ACTION_ON_START
 
-        self.set_role(self.get_postgres_role_from_data_directory())
+        self.set_role(role or self.get_postgres_role_from_data_directory())
 
         self.set_state('starting')
         self._pending_restart = False
@@ -1091,7 +1093,7 @@ class Postgresql(object):
 
         return self.state == 'running'
 
-    def restart(self, timeout=None, task=None):
+    def restart(self, timeout=None, task=None, block_callbacks=False, role=None):
         """Restarts PostgreSQL.
 
         When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or
@@ -1100,8 +1102,9 @@ class Postgresql(object):
         :returns: True when restart was successful and timeout did not expire when waiting.
         """
         self.set_state('restarting')
-        self.__cb_pending = ACTION_ON_RESTART
-        ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True, task=task)
+        if not block_callbacks:
+            self.__cb_pending = ACTION_ON_RESTART
+        ret = self.stop(block_callbacks=True) and self.start(timeout, task, True, role)
         if not ret and not self.is_starting():
             self.set_state('restart failed ({0})'.format(self.state))
         return ret
@@ -1436,7 +1439,7 @@ class Postgresql(object):
     def rewind_failed(self):
         return self._rewind_state == REWIND_STATUS.FAILED
 
-    def follow(self, member, timeout=None):
+    def follow(self, member, role='replica', timeout=None):
         is_remote_master = isinstance(member, RemoteMember)
         no_replication_slot = is_remote_master and member.no_replication_slot
         restore_command = is_remote_master and member.restore_command
@@ -1444,7 +1447,8 @@ class Postgresql(object):
         archive_cleanup = is_remote_master and member.archive_cleanup_command
 
         primary_conninfo = self.primary_conninfo(member)
-        change_role = self.role in ('master', 'demoted')
+        change_role = self.cb_called and (self.role in ('master', 'demoted') or
+                                          not {'standby_leader', 'replica'} - {self.role, role})
 
         recovery_params = self.config.get('recovery_conf', {}).copy()
         recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
@@ -1463,11 +1467,17 @@ class Postgresql(object):
 
         self.write_recovery_conf(recovery_params)
 
+        # When we demoting the master or standby_leader to replica or promoting replica to a standby_leader
+        # and we know for sure that postgres was already running before, we will only execute on_role_change
+        # callback and prevent execution of on_restart/on_start callback.
+        # If the role remains the same (replica or standby_leader), we will execute on_start or on_restart
+        if change_role:
+            self.__cb_pending = ACTION_NOOP
+
         if self.is_running():
-            self.restart()
+            self.restart(block_callbacks=change_role, role=role)
         else:
-            self.start(timeout=timeout)
-        self.set_role('replica')
+            self.start(timeout=timeout, block_callbacks=change_role, role=role)
 
         if change_role:
             # TODO: postpone this until start completes, or maybe do even earlier
@@ -1506,7 +1516,7 @@ class Postgresql(object):
             logger.exception('unable to restore configuration files from backup')
 
     def _wait_promote(self, wait_seconds):
-        for _ in polling_loop(wait_seconds - 1):
+        for _ in polling_loop(wait_seconds):
             data = self.controldata()
             if data.get('Database cluster state') == 'in production':
                 return True

tests/test_ha.py

@@ -674,15 +674,17 @@ class TestHa(unittest.TestCase):
         self.p.is_leader = false
         self.p.name = 'leader'
         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)
+        self.p.check_recovery_conf = true
+        self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the standby leader with the lock')
 
     def test_process_healthy_standby_cluster_as_cascade_replica(self):
         self.p.is_leader = false
         self.p.name = 'replica'
         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)
+        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a standby leader')
+        with patch.object(Leader, 'conn_url', PropertyMock(return_value='')):
+            self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader')
 
     @patch('requests.get', requests_get)
     def test_process_unhealthy_standby_cluster_as_standby_leader(self):
@@ -692,8 +694,7 @@ class TestHa(unittest.TestCase):
         self.ha.cluster.is_unlocked = true
         self.ha.sysid_valid = true
         self.p._sysid = True
-        msg = 'promoted self to a standby leader because i had the session lock'
-        self.assertEqual(self.ha.run_cycle(), msg)
+        self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader by acquiring session lock')
 
     @patch.object(Postgresql, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True))

tests/test_patroni.py

@@ -118,6 +118,7 @@ class TestPatroni(unittest.TestCase):
         self.assertRaises(SystemExit, self.p.sigterm_handler)
 
     def test_schedule_next_run(self):
+        self.p.ha.cluster = Mock()
         self.p.ha.dcs.watch = Mock(return_value=True)
         self.p.schedule_next_run()
         self.p.next_run = time.time() - self.p.dcs.loop_wait - 1

tests/test_postgresql.py

@@ -403,6 +403,7 @@ class TestPostgresql(unittest.TestCase):
     @patch.object(Postgresql, 'is_running', Mock(return_value=False))
     @patch.object(Postgresql, 'start', Mock())
     def test_follow(self):
+        self.p.call_nowait('on_start')
         m = RemoteMember('1', {'restore_command': '2', 'recovery_min_apply_delay': 3, 'archive_cleanup_command': '4'})
         self.p.follow(m)