Make it possible to run pg_rewind without superuser on pg11+ (#1035)

* expose the current patroni version in DCS
* expose `checkpoint_after_promote` flag in DCS as an indicator that pg_rewind could be safely executed
* other nodes will wait until this flag is set instead of connecting as superuser and issuing the CHECKPOINT
* define `postgresql.authention.rewind` with credentials for pg_rewind in patroni configuration files.
* create user for pg_rewind if postgres is 11+
* grant execute on functions required for pg_rewind to rewind user

docs/SETTINGS.rst

@@ -137,6 +137,9 @@ PostgreSQL
     -  **replication**:
         -  **username**: replication username; the user will be created during initialization. Replicas will use this user to access master via streaming replication
         -  **password**: replication password; the user will be created during initialization.
+    -  **rewind**:
+        -  **username**: name for the user for ``pg_rewind``; the user will be created during initialization of postgres 11+ and all necessary `permissions <https://paquier.xyz/postgresql-2/postgres-11-superuser-rewind/>`__ will be granted.
+        -  **password**: password for the user for ``pg_rewind``; 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 postgres restarts (without changing role).

patroni/dcs/__init__.py

@@ -239,6 +239,19 @@ class Leader(namedtuple('Leader', 'index,session,member')):
     def timeline(self):
         return self.member.data.get('timeline')
 
+    @property
+    def checkpoint_after_promote(self):
+        """
+        >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote
+        """
+        version = self.member.data.get('version')
+        if version:
+            try:
+                if tuple(map(int, version.split('.'))) >= (1, 5, 6):
+                    return bool(self.member.data.get('checkpoint_after_promote'))
+            except Exception:
+                logger.debug('Failed to parse Patroni version %s', version)
+
 
 class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')):
 

patroni/dcs/zookeeper.py

@@ -274,7 +274,10 @@ class ZooKeeper(AbstractDCS):
         cluster = self.cluster
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
-        if member and self._client.client_id is not None and member.session != self._client.client_id[0]:
+        if member and (self._client.client_id is not None and member.session != self._client.client_id[0] or
+                       not (deep_compare(member.data.get('tags', {}), data.get('tags', {})) and
+                            member.data.get('version') == data.get('version') and
+                            member.data.get('checkpoint_after_promote') == data.get('checkpoint_after_promote'))):
             try:
                 self._client.delete_async(self.member_path).get(timeout=1)
             except NoNodeError:

patroni/ha.py

@@ -174,12 +174,15 @@ class Ha(object):
                 'conn_url': self.state_handler.connection_string,
                 'api_url': self.patroni.api.connection_string,
                 'state': self.state_handler.state,
-                'role': self.state_handler.role
+                'role': self.state_handler.role,
+                'version': self.patroni.version
             }
 
             # following two lines are mainly necessary for consul, to avoid creation of master service
             if data['role'] == 'master' and not self.is_leader():
                 data['role'] = 'promoted'
+            if self.is_leader():
+                data['checkpoint_after_promote'] = self.state_handler.checkpoint_after_promote()
             tags = self.get_effective_tags()
             if tags:
                 data['tags'] = tags
@@ -904,6 +907,9 @@ class Ha(object):
                 if msg is not None:
                     return msg
 
+                # check if the node is ready to be used by pg_rewind
+                self.state_handler.check_for_checkpoint_after_promote()
+
                 if self.is_standby_cluster():
                     # in case of standby cluster we don't really need to
                     # enforce anything, since the leader is not a master.

patroni/postgresql.py

@@ -38,7 +38,8 @@ STATE_NO_RESPONSE = 'not responding'
 STATE_UNKNOWN = 'unknown'
 
 STOP_POLLING_INTERVAL = 1
-REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5})
+REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECKPOINT': 1, 'CHECK': 2, 'NEED': 3,
+                                  'NOT_NEED': 4, 'SUCCESS': 5, 'FAILED': 6})
 sync_standby_name_re = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$')
 
 cluster_info_query = ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
@@ -202,6 +203,11 @@ class Postgresql(object):
     def _replication(self):
         return self.config['authentication']['replication']
 
+    @property
+    def _rewind_credentials(self):
+        return self.config['authentication'].get('rewind', self._superuser) \
+                if self._major_version >= 110000 else self._superuser
+
     @property
     def callback(self):
         return self.config.get('callbacks') or {}
@@ -1055,6 +1061,18 @@ class Postgresql(object):
             self.call_nowait(ACTION_ON_RELOAD)
         return ret
 
+    def check_for_checkpoint_after_promote(self):
+        if self._rewind_state == REWIND_STATUS.INITIAL and self.is_leader():
+            try:
+                timeline = int(self.controldata().get("Latest checkpoint's TimeLineID"))
+                if self.get_master_timeline() == timeline:
+                    self._rewind_state = REWIND_STATUS.CHECKPOINT
+            except (TypeError, ValueError):
+                logger.exception('Failed to parse timeline from pg_controldata output')
+
+    def checkpoint_after_promote(self):
+        return self._rewind_state == REWIND_STATUS.CHECKPOINT
+
     def check_for_startup(self):
         """Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup."""
         return self.is_starting() and not self.check_startup_state_changed()
@@ -1361,7 +1379,7 @@ class Postgresql(object):
             if leader.member.data.get('role') != 'master':
                 return
         # standby cluster
-        elif not self.check_leader_is_not_in_recovery(**leader.conn_kwargs(self._superuser)):
+        elif not self.check_leader_is_not_in_recovery(**leader.conn_kwargs(self._replication)):
             return
 
         history = need_rewind = None
@@ -1425,14 +1443,21 @@ class Postgresql(object):
             return logger.warning('Can not run pg_rewind because postgres is still running')
 
         # prepare pg_rewind connection
-        r = leader.conn_kwargs(self._superuser)
+        r = leader.conn_kwargs(self._rewind_credentials)
 
-        # first make sure that we are really trying to rewind
+        # 1. make sure that we are really trying to rewind from the master
-        # from the master and run a checkpoint on it in order to
+        # 2. make sure that pg_control contains the new timeline by:
-        # make it store the new timeline ([email protected])
+        #   running a checkpoint or
-        leader_status = self.checkpoint(r)
+        #   waiting until Patroni on the master will expose checkpoint_after_promote=True
-        if leader_status:
+        checkpoint_status = leader.checkpoint_after_promote if isinstance(leader, Leader) else None
-            return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
+        if checkpoint_status is None:  # master still runs the old Patroni
+            leader_status = self.checkpoint(leader.conn_kwargs(self._superuser))
+            if leader_status:
+                return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
+        elif not checkpoint_status:
+            return logger.info('Waiting for checkpoint on %s before rewind', leader.name)
+        elif not self.check_leader_is_not_in_recovery(**r):
+            return
 
         if self.pg_rewind(r):
             self._rewind_state = REWIND_STATUS.SUCCESS
@@ -1733,6 +1758,14 @@ END;$$""".format(name, ' '.join(options))
             if task.result:
                 self.create_or_update_role(self._replication['username'],
                                            self._replication.get('password'), ['REPLICATION'])
+
+                if self._major_version >= 110000 and 'rewind' in self.config['authentication']:
+                    rewind = self.config['authentication']['rewind']
+                    self.create_or_update_role(rewind['username'], rewind.get('password'), [])
+                    for f in ('pg_ls_dir(text, boolean, boolean)', 'pg_stat_file(text, boolean)',
+                              'pg_read_binary_file(text)', 'pg_read_binary_file(text, bigint, bigint, boolean)'):
+                        self.query('GRANT EXECUTE ON function pg_catalog.{0} TO "{1}"'.format(f, rewind['username']))
+
                 for name, value in (config.get('users') or {}).items():
                     if name not in (self._superuser.get('username'), self._replication['username']):
                         self.create_or_update_role(name, value.get('password'), value.get('options', []))

postgres0.yml

@@ -74,12 +74,6 @@ bootstrap:
         - createdb
 
 postgresql:
-  # Fully qualified kerberos ticket file for the running user
-  # same as KRB5CCNAME used by the GSS
-#  krb_server_keyfile: /var/spool/keytabs/postgres
-
-  # Server side kerberos spn
-#  krbsrvname: postgres
   listen: 127.0.0.1:5432
   connect_address: 127.0.0.1:5432
   data_dir: data/postgresql0
@@ -93,7 +87,15 @@ postgresql:
     superuser:
       username: postgres
       password: zalando
+    rewind:  # Has no effect on postgres 10 and lower
+      username: rewind_user
+      password: rewind_password
+  # Server side kerberos spn
+#  krbsrvname: postgres
   parameters:
+    # Fully qualified kerberos ticket file for the running user
+    # same as KRB5CCNAME used by the GSS
+#   krb_server_keyfile: /var/spool/keytabs/postgres
     unix_socket_directories: '.'
 
 #watchdog:

postgres1.yml

@@ -68,12 +68,6 @@ bootstrap:
         - createdb
 
 postgresql:
-  # Fully qualified kerberos ticket file for the running user
-  # same as KRB5CCNAME used by the GSS
-#  krb_server_keyfile: /var/spool/keytabs/postgres
-
-  # Server side kerberos spn
-#  krbsrvname: postgres
   listen: 127.0.0.1:5433
   connect_address: 127.0.0.1:5433
   data_dir: data/postgresql1
@@ -87,7 +81,15 @@ postgresql:
     superuser:
       username: postgres
       password: zalando
+    rewind:  # Has no effect on postgres 10 and lower
+      username: rewind_user
+      password: rewind_password
+  # Server side kerberos spn
+#  krbsrvname: postgres
   parameters:
+    # Fully qualified kerberos ticket file for the running user
+    # same as KRB5CCNAME used by the GSS
+#    krb_server_keyfile: /var/spool/keytabs/postgres
     unix_socket_directories: '.'
   basebackup:
       - verbose

postgres2.yml

@@ -65,12 +65,6 @@ bootstrap:
         - createdb
 
 postgresql:
-  # Fully qualified kerberos ticket file for the running user
-  # same as KRB5CCNAME used by the GSS
-#  krb_server_keyfile: /var/spool/keytabs/postgres
-
-  # Server side kerberos spn
-#  krbsrvname: postgres
   listen: 127.0.0.1:5434
   connect_address: 127.0.0.1:5434
   data_dir: data/postgresql2
@@ -84,7 +78,15 @@ postgresql:
     superuser:
       username: postgres
       password: zalando
+    rewind:  # Has no effect on postgres 10 and lower
+      username: rewind_user
+      password: rewind_password
+  # Server side kerberos spn
+#  krbsrvname: postgres
   parameters:
+    # Fully qualified kerberos ticket file for the running user
+    # same as KRB5CCNAME used by the GSS
+#    krb_server_keyfile: /var/spool/keytabs/postgres
     unix_socket_directories: '.'
 tags:
     nofailover: false

tests/test_ha.py

@@ -119,6 +119,7 @@ zookeeper:
 
         self.config = Config()
         self.config.set_dynamic_configuration({'maximum_lag_on_failover': 5})
+        self.version = '1.5.7'
         self.postgresql = p
         self.dcs = d
         self.api = Mock()

tests/test_postgresql.py

@@ -191,7 +191,7 @@ class TestPostgresql(unittest.TestCase):
             os.makedirs(self.data_dir)
         self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir,
                              'config_dir': self.config_dir, 'retry_timeout': 10,
-                             'pgpass': os.path.join(gettempdir(), 'pgpass0'),
+                             'krbsrvname': 'postgres', 'pgpass': os.path.join(gettempdir(), 'pgpass0'),
                              'listen': '127.0.0.2, 127.0.0.3:5432', 'connect_address': '127.0.0.2:5432',
                              'authentication': {'superuser': {'username': 'test', 'password': 'test'},
                                                 'replication': {'username': 'replicator', 'password': 'rep-pass'}},
@@ -398,6 +398,14 @@ class TestPostgresql(unittest.TestCase):
             self.p.config['remove_data_directory_on_rewind_failure'] = False
             self.p.trigger_check_diverged_lsn()
             self.p.rewind(self.leader)
+
+        self.leader.member.data.update(version='1.5.7', checkpoint_after_promote=False)
+        self.assertIsNone(self.p.rewind(self.leader))
+
+        self.leader.member.data['checkpoint_after_promote'] = True
+        with patch.object(Postgresql, 'check_leader_is_not_in_recovery', Mock(return_value=False)):
+            self.assertIsNone(self.p.rewind(self.leader))
+
         with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
             self.p.rewind(self.leader)
             self.p.is_leader = Mock(return_value=False)
@@ -663,7 +671,7 @@ class TestPostgresql(unittest.TestCase):
     @patch.object(Postgresql, 'run_bootstrap_post_init', Mock(return_value=True))
     @patch.object(Postgresql, '_custom_bootstrap', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
-    @patch.object(Postgresql, 'get_major_version', Mock(return_value=90600))
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=110000))
     def test_post_bootstrap(self):
         config = {'method': 'foo', 'foo': {'command': 'bar'}}
         self.p.bootstrap(config)
@@ -686,7 +694,8 @@ class TestPostgresql(unittest.TestCase):
         self.p.bootstrap(config)
         self.p.set_state('stopped')
         self.p.reload_config({'authentication': {'superuser': {'username': 'p', 'password': 'p'},
-                                                 'replication': {'username': 'r', 'password': 'r'}},
+                                                 'replication': {'username': 'r', 'password': 'r'},
+                                                 'rewind': {'username': 'rw', 'password': 'rw'}},
                               'listen': '*', 'retry_timeout': 10, 'parameters': {'wal_level': '', 'hba_file': 'foo'}})
         with patch.object(Postgresql, 'restart', Mock()) as mock_restart:
             self.p.post_bootstrap({}, task)
@@ -787,7 +796,7 @@ class TestPostgresql(unittest.TestCase):
         parameters = self._PARAMETERS.copy()
         parameters.pop('f.oo')
         config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'authentication': {},
-                  'retry_timeout': 10, 'listen': '*', 'parameters': parameters}
+                  'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters}
         self.p.reload_config(config)
         parameters['b.ar'] = 'bar'
         self.p.reload_config(config)
@@ -1040,3 +1049,7 @@ class TestPostgresql(unittest.TestCase):
             self.p.cancel()
             self.assertFalse(self.p.start())
             self.assertTrue(self.p.pending_restart)
+
+    @patch.object(Postgresql, 'controldata', Mock(return_value={"Latest checkpoint's TimeLineID": 1}))
+    def test_check_for_checkpoint_after_promote(self):
+        self.p.check_for_checkpoint_after_promote()