Clean cluster demotion (patroni#3361)

Make sure introduction of the "standby_cluster" section in the dynamic configuration leads to a clean cluster demotion

Author: hughcapet

features/environment.py

@@ -123,6 +123,12 @@ def read_label(self, label):
         except IOError:
             return None
 
+    def read_config(self):
+        config = dict()
+        with open(self._config) as r:
+            config = yaml.safe_load(r)
+        return config
+
     @staticmethod
     def recursive_update(dst, src):
         for k, v in src.items():
@@ -132,11 +138,10 @@ def recursive_update(dst, src):
                 dst[k] = v
 
     def update_config(self, custom_config):
-        with open(self._config) as r:
-            config = yaml.safe_load(r)
-            self.recursive_update(config, custom_config)
-            with open(self._config, 'w') as w:
-                yaml.safe_dump(config, w, default_flow_style=False)
+        config = self.read_config()
+        self.recursive_update(config, custom_config)
+        with open(self._config, 'w') as w:
+            yaml.safe_dump(config, w, default_flow_style=False)
         self._scope = config.get('scope', 'batman')
 
     def add_tag_to_config(self, tag, value):
@@ -857,7 +862,7 @@ def start(self, name, max_wait_limit=40, custom_config=None):
         self._processes[name].start(max_wait_limit)
 
     def __getattr__(self, func):
-        if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to',
+        if func not in ['stop', 'query', 'write_label', 'read_label', 'read_config', 'check_role_has_changed_to',
                         'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup', 'read_patroni_log']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 

features/standby_cluster.feature

@@ -64,3 +64,17 @@ Feature: standby cluster
     And I receive a response role standby_leader
     And replication works from postgres-0 to postgres-1 after 15 seconds
     And there is a postgres-1_cb.log with "on_role_change replica batman1\non_role_change standby_leader batman1" in postgres-1 data directory
+
+  Scenario: demote cluster
+    When I switch standby cluster batman1 to archive recovery
+    Then Response on GET http://127.0.0.1:8009/patroni contains replication_state=in archive recovery after 30 seconds
+    When I demote cluster batman
+    And "members/postgres-0" key in DCS has role=standby_leader after 20 seconds
+    And "members/postgres-0" key in DCS has state=running after 10 seconds
+
+  Scenario: promote cluster
+    When I issue a PATCH request to http://127.0.0.1:8009/config with {"standby_cluster": null}
+    Then I receive a response code 200
+    And postgres-1 role is the primary after 10 seconds
+    When I add the table foo2 to postgres-1
+    Then table foo2 is present on postgres-0 after 20 seconds

features/steps/standby_cluster.py

@@ -1,7 +1,9 @@
+import json
 import os
 import time
 
 from behave import step
+from patroni_api import check_response, do_request
 
 
 def callbacks(context, name):
@@ -66,3 +68,29 @@ def check_replication_status(context, pg_name1, pg_name2, timeout):
         time.sleep(1)
     else:
         assert False, "{0} is not replicating from {1} after {2} seconds".format(pg_name1, pg_name2, timeout)
+
+
+@step('I switch standby cluster {scope:name} to archive recovery')
+def standby_cluster_archive(context, scope, demote=False):
+    for name, proc in context.pctl._processes.items():
+        if proc._scope != scope or not proc._is_running:
+            continue
+
+        config = context.pctl.read_config(name)
+        url = f'http://{config["restapi"]["connect_address"]}/config'
+        data = {
+            "standby_cluster": {
+                "restore_command": config['bootstrap']['dcs']['postgresql']['parameters']['restore_command']
+            }
+        }
+        if not demote:
+            data['standby_cluster']['primary_slot_name'] = data['standby_cluster']['host'] =\
+                data['standby_cluster']['port'] = None
+        do_request(context, 'PATCH', url, json.dumps(data))
+        check_response(context, 'code', 200)
+        break
+
+
+@step('I demote cluster {scope:name}')
+def demote_cluster(context, scope):
+    standby_cluster_archive(context, scope, True)

patroni/ha.py

@@ -8,7 +8,7 @@
 
 from multiprocessing.pool import ThreadPool
 from threading import RLock
-from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Tuple, TYPE_CHECKING, Union
+from typing import Any, Callable, cast, Collection, Dict, List, NamedTuple, Optional, Tuple, TYPE_CHECKING, Union
 
 from . import global_config, psycopg
 from .__main__ import Patroni
@@ -746,7 +746,10 @@ def follow(self, demote_reason: str, follow_reason: str, refresh: bool = True) -
                 if not node_to_follow:
                     return 'no action. I am ({0})'.format(self.state_handler.name)
         elif is_leader:
-            self.demote('immediate-nolock')
+            if self.is_standby_cluster():
+                self._async_executor.try_run_async('demoting to a standby cluster', self.demote, ('demote-cluster',))
+            else:
+                self.demote('immediate-nolock')
             return demote_reason
 
         if self.is_standby_cluster() and self._leader_timeline and \
@@ -1563,6 +1566,7 @@ def demote(self, mode: str) -> Optional[bool]:
             'graceful':         dict(stop='fast',      checkpoint=True,  release=True,  offline=False, async_req=False),  # noqa: E241,E501
             'immediate':        dict(stop='immediate', checkpoint=False, release=True,  offline=False, async_req=True),  # noqa: E241,E501
             'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True),  # noqa: E241,E501
+            'demote-cluster':   dict(stop='fast',      checkpoint=False, release=True,  offline=False,  async_req=False),  # noqa: E241,E501
 
         }[mode]
 
@@ -1572,6 +1576,16 @@ def demote(self, mode: str) -> Optional[bool]:
 
         status = {'released': False}
 
+        demote_cluster_with_archive = False
+        archive_cmd = self._rewind.get_archive_command()
+        if mode == 'demote-cluster' and archive_cmd is not None:
+            # We need to send the shutdown checkpoint WAL file to archive to eliminate the need of rewind
+            # from a promoted instance that was previously replicating from archive
+            # When doing this, we disable stop timeout, do not run on_shutdown callback and do not release
+            # leader key.
+            demote_cluster_with_archive = True
+            mode_control['release'] = False
+
         def on_shutdown(checkpoint_location: int, prev_location: int) -> None:
             # Postmaster is still running, but pg_control already reports clean "shut down".
             # It could happen if Postgres is still archiving the backlog of WAL files.
@@ -1580,8 +1594,11 @@ def on_shutdown(checkpoint_location: int, prev_location: int) -> None:
             time.sleep(1)  # give replicas some more time to catch up
             if self.is_failover_possible(cluster_lsn=checkpoint_location):
                 self.state_handler.set_role(PostgresqlRole.DEMOTED)
+                # for demotion to a standby cluster we need shutdown checkpoint lsn to be written to optime,
+                # not the prev one
+                last_lsn = checkpoint_location if mode == 'demote-cluster' else prev_location
                 with self._async_executor:
-                    self.release_leader_key_voluntarily(prev_location)
+                    self.release_leader_key_voluntarily(last_lsn)
                     status['released'] = True
 
         def before_shutdown() -> None:
@@ -1594,16 +1611,33 @@ def before_shutdown() -> None:
                                 on_safepoint=self.watchdog.disable if self.watchdog.is_running else None,
                                 on_shutdown=on_shutdown if mode_control['release'] else None,
                                 before_shutdown=before_shutdown if mode == 'graceful' else None,
-                                stop_timeout=self.primary_stop_timeout())
+                                stop_timeout=None if demote_cluster_with_archive else self.primary_stop_timeout())
         self.state_handler.set_role(PostgresqlRole.DEMOTED)
-        self.set_is_leader(False)
+
+        # for demotion to a standby cluster we need shutdown checkpoint lsn to be written to optime, not the prev one
+        checkpoint_lsn, prev_lsn = self.state_handler.latest_checkpoint_locations() \
+            if mode == 'graceful' else (None, None)
+
+        is_standby_leader = mode == 'demote-cluster' and not status['released']
+        if is_standby_leader:
+            with self._async_executor:
+                self.dcs.update_leader(self.cluster, checkpoint_lsn, None, self._failsafe_config())
+            mode_control['release'] = False
+        else:
+            self.set_is_leader(False)
 
         if mode_control['release']:
             if not status['released']:
-                checkpoint_location = self.state_handler.latest_checkpoint_location() if mode == 'graceful' else None
                 with self._async_executor:
-                    self.release_leader_key_voluntarily(checkpoint_location)
+                    self.release_leader_key_voluntarily(prev_lsn)
             time.sleep(2)  # Give a time to somebody to take the leader lock
+
+        if mode == 'demote-cluster':
+            if demote_cluster_with_archive:
+                self._rewind.archive_shutdown_checkpoint_wal(cast(str, archive_cmd))
+            else:
+                logger.info('Not archiving latest checkpoint WAL file. Archiving is not configured.')
+
         if mode_control['offline']:
             node_to_follow, leader = None, None
         else:
@@ -1616,15 +1650,17 @@ def before_shutdown() -> None:
         if self.is_synchronous_mode():
             self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet())
 
+        role = PostgresqlRole.STANDBY_LEADER if is_standby_leader else PostgresqlRole.REPLICA
         # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress
         # there could be an async action already running, calling follow from here will lead
         # to racy state handler state updates.
         if mode_control['async_req']:
-            self._async_executor.try_run_async('starting after demotion', self.state_handler.follow, (node_to_follow,))
+            self._async_executor.try_run_async('starting after demotion', self.state_handler.follow,
+                                               (node_to_follow, role,))
         else:
             if self._rewind.rewind_or_reinitialize_needed_and_possible(leader):
                 return False  # do not start postgres, but run pg_rewind on the next iteration
-            self.state_handler.follow(node_to_follow)
+            self.state_handler.follow(node_to_follow, role)
 
     def should_run_scheduled_action(self, action_name: str, scheduled_at: Optional[datetime.datetime],
                                     cleanup_fn: Callable[..., Any]) -> bool:
@@ -2363,8 +2399,8 @@ def _before_shutdown() -> None:
                                                                         stop_timeout=self.primary_stop_timeout()))
             if not self.state_handler.is_running():
                 if self.is_leader() and not status['deleted']:
-                    checkpoint_location = self.state_handler.latest_checkpoint_location()
-                    self.dcs.delete_leader(self.cluster.leader, checkpoint_location)
+                    _, prev_location = self.state_handler.latest_checkpoint_locations()
+                    self.dcs.delete_leader(self.cluster.leader, prev_location)
                 self.touch_member()
             else:
                 # XXX: what about when Patroni is started as the wrong user that has access to the watchdog device

patroni/postgresql/__init__.py

@@ -623,14 +623,16 @@ def parse_wal_record(self, timeline: str,
                 return match.group(1), match.group(2), match.group(3), match.group(4)
         return None, None, None, None
 
-    def _checkpoint_locations_from_controldata(self, data: Dict[str, str]) -> Optional[Tuple[int, int]]:
+    def latest_checkpoint_locations(self, data: Optional[Dict[str, str]] = None) -> Tuple[Optional[int], Optional[int]]:
         """Get shutdown checkpoint location.
 
         :param data: :class:`dict` object with values returned by `pg_controldata` tool.
 
         :returns: a tuple of checkpoint LSN for the cleanly shut down primary, and LSN of prev wal record (SWITCH)
                   if we know that the checkpoint was written to the new WAL file due to the archive_mode=on.
         """
+        if data is None:
+            data = self.controldata()
         timeline = data.get("Latest checkpoint's TimeLineID")
         lsn = checkpoint_lsn = data.get('Latest checkpoint location')
         prev_lsn = None
@@ -651,19 +653,7 @@ def _checkpoint_locations_from_controldata(self, data: Dict[str, str]) -> Option
                 logger.error('Exception when parsing WAL pg_%sdump output: %r', self.wal_name, e)
             if isinstance(checkpoint_lsn, int):
                 return checkpoint_lsn, (prev_lsn or checkpoint_lsn)
-
-    def latest_checkpoint_location(self) -> Optional[int]:
-        """Get shutdown checkpoint location.
-
-        .. note::
-            In case if checkpoint was written to the new WAL file due to the archive_mode=on
-            we return LSN of the previous wal record (SWITCH).
-
-        :returns: checkpoint LSN for the cleanly shut down primary.
-        """
-        checkpoint_locations = self._checkpoint_locations_from_controldata(self.controldata())
-        if checkpoint_locations:
-            return checkpoint_locations[1]
+        return None, None
 
     def is_running(self) -> Optional[PostmasterProcess]:
         """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
@@ -922,9 +912,9 @@ def _do_stop(self, mode: str, block_callbacks: bool, checkpoint: bool,
             while postmaster.is_running():
                 data = self.controldata()
                 if data.get('Database cluster state', '') == 'shut down':
-                    checkpoint_locations = self._checkpoint_locations_from_controldata(data)
-                    if checkpoint_locations:
-                        on_shutdown(*checkpoint_locations)
+                    checkpoint_lsn, prev_lsn = self.latest_checkpoint_locations(data)
+                    if checkpoint_lsn is not None and prev_lsn is not None:
+                        on_shutdown(checkpoint_lsn, prev_lsn)
                     break
                 elif data.get('Database cluster state', '').startswith('shut down'):  # shut down in recovery
                     break

patroni/postgresql/rewind.py

@@ -327,6 +327,16 @@ def ensure_checkpoint_after_promote(self, wakeup: Callable[..., Any]) -> None:
     def checkpoint_after_promote(self) -> bool:
         return self._state == REWIND_STATUS.CHECKPOINT
 
+    def get_archive_command(self) -> Optional[str]:
+        """Get ``archive_command`` GUC value if defined and archiving is enabled.
+
+        :returns: ``archive_command`` defined in the Postgres configuration or None.
+        """
+        archive_mode = self._postgresql.get_guc_value('archive_mode')
+        archive_cmd = self._postgresql.get_guc_value('archive_command')
+        if archive_mode in ('on', 'always') and archive_cmd:
+            return archive_cmd
+
     def _build_archiver_command(self, command: str, wal_filename: str) -> str:
         """Replace placeholders in the given archiver command's template.
         Applicable for archive_command and restore_command.
@@ -380,9 +390,8 @@ def _archive_ready_wals(self) -> None:
         after it the WALs were recycled on the promoted replica.
         With this we prevent the entire loss of such WALs and the
         consequent old leader's start failure."""
-        archive_mode = self._postgresql.get_guc_value('archive_mode')
-        archive_cmd = self._postgresql.get_guc_value('archive_command')
-        if archive_mode not in ('on', 'always') or not archive_cmd:
+        archive_cmd = self.get_archive_command()
+        if not archive_cmd:
             return
 
         walseg_regex = re.compile(r'^[0-9A-F]{24}(\.partial){0,1}\.ready$')
@@ -602,3 +611,17 @@ def ensure_clean_shutdown(self) -> Optional[bool]:
             logger.info(' stdout=%s', output['stdout'].decode('utf-8'))
             logger.info(' stderr=%s', output['stderr'].decode('utf-8'))
         return ret == 0 or None
+
+    def archive_shutdown_checkpoint_wal(self, archive_cmd: str) -> None:
+        """Archive WAL file with the shutdown checkpoint.
+
+        :param archive_cmd: archiver command to use
+        """
+        data = self._postgresql.controldata()
+        wal_file = data.get("Latest checkpoint's REDO WAL file", '')
+        if not wal_file:
+            logger.error("Cannot extract latest checkpoint's WAL file name")
+            return
+        cmd = self._build_archiver_command(archive_cmd, wal_file)
+        if self._postgresql.cancellable.call([cmd], shell=True):
+            logger.error("Failed to archive WAL file with the shutdown checkpoint")