Add support for custom pg_rewind options (#3484)

barthisrael · web-flow · commit b3462eb4e918 · 2026-01-02T11:53:41.000+01:00
### Summary

This PR introduces the `postgresql.rewind` configuration parameter,
allowing users to pass custom command-line options to `pg_rewind`.

This provides flexibility for the PostgreSQL users, and is also a
necessary enhancement for supporting TDE (Transparent Data Encryption)
on proprietary Postgres, like EPAS (EDB Postgres Advanced Server), which
requires passing custom flags during the rewind process.

### Key Changes

* **Refactor:** The `process_user_options` function was first moved from
`bootstrap.py` to `utils.py` to make it a generic helper. Its unit tests
were moved, and a minor bug in its error reporting was fixed.
* **Feature:** `patroni/postgresql/rewind.py` now reads the new
`postgresql.rewind` configuration parameter, parses it using the
`process_user_options` helper, and appends the resulting flags to the
`pg_rewind` command.
* **Testing:** New unit tests have been added to `test_rewind.py` to
validate the new feature, including checking for disallowed options.
* **Docs:** The new `postgresql.rewind` parameter is documented in
`yaml_configuration.rst` with a usage example.
* **Validation:** The configuration schema in `validator.py` is updated
to include the new parameter.

---------

Signed-off-by: Israel Barth Rubio &lt;israel.barth@enterprisedb.com&gt;
diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst
@@ -320,6 +320,16 @@ PostgreSQL
       -  **- mapname1 systemname2 pguser2**
    -  **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds.
    -  **use\_pg\_rewind**: try to use pg\_rewind on the former leader when it joins cluster as a replica. Either the cluster must be initialized with ``data page checksums`` (``--data-checksums`` option for ``initdb``) and/or ``wal_log_hints`` must be set to ``on``, or ``pg_rewind`` will not work.
+   -  **rewind**: (optional) custom options to pass to the ``pg_rewind`` command. Can be specified as a list of strings and/or single key-value dictionaries. Not allowed options include: ``target-pgdata``, ``source-pgdata``, ``source-server``, ``write-recovery-conf``, ``dry-run``, ``restore-target-wal``, ``config-file``, ``no-ensure-shutdown``, ``version``, and ``help``. Example usage:
+
+      .. code:: YAML
+
+         postgresql:
+           rewind:
+             - debug
+             - progress
+             - sync-method: fsync
+
    -  **remove\_data\_directory\_on\_rewind\_failure**: If this option is enabled, Patroni will remove the PostgreSQL data directory and recreate the replica. Otherwise it will try to follow the new leader. Default value is **false**.
    -  **remove\_data\_directory\_on\_diverged\_timelines**: Patroni will remove the PostgreSQL data directory and recreate the replica if it notices that timelines are diverging and the former primary can not start streaming from the new primary. This option is useful when ``pg_rewind`` can not be used. While performing timelines divergence check on PostgreSQL v10 and older Patroni will try to connect with replication credential to the "postgres" database. Hence, such access should be allowed in the pg_hba.conf. Default value is **false**.
    -  **replica\_method**: for each create_replica_methods other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value".
diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py
@@ -4,13 +4,13 @@
 import tempfile
 import time
 
-from typing import Any, Callable, cast, Dict, List, Optional, Tuple, TYPE_CHECKING, Union
+from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 
 from ..async_executor import CriticalTask
 from ..collections import EMPTY_DICT
 from ..dcs import Leader, Member, RemoteMember
 from ..psycopg import quote_ident, quote_literal
-from ..utils import deep_compare, unquote
+from ..utils import deep_compare, process_user_options
 from .misc import PostgresqlState
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -33,95 +33,14 @@ def running_custom_bootstrap(self) -> bool:
     def keep_existing_recovery_conf(self) -> bool:
         return self._running_custom_bootstrap and self._keep_existing_recovery_conf
 
-    @staticmethod
-    def process_user_options(tool: str, options: Any,
-                             not_allowed_options: Tuple[str, ...],
-                             error_handler: Callable[[str], None]) -> List[str]:
-        """Format *options* in a list or dictionary format into command line long form arguments.
-
-        .. note::
-            The format of the output of this method is to prepare arguments for use in the ``initdb``
-            method of `self._postgres`.
-
-        :Example:
-
-            The *options* can be defined as a dictionary of key, values to be converted into arguments:
-            >>> Bootstrap.process_user_options('foo', {'foo': 'bar'}, (), print)
-            ['--foo=bar']
-
-            Or as a list of single string arguments
-            >>> Bootstrap.process_user_options('foo', ['yes'], (), print)
-            ['--yes']
-
-            Or as a list of key, value options
-            >>> Bootstrap.process_user_options('foo', [{'foo': 'bar'}], (), print)
-            ['--foo=bar']
-
-            Or a combination of single and key, values
-            >>> Bootstrap.process_user_options('foo', ['yes', {'foo': 'bar'}], (), print)
-            ['--yes', '--foo=bar']
-
-            Options that contain spaces are passed as is to ``subprocess.call``
-            >>> Bootstrap.process_user_options('foo', [{'foo': 'bar baz'}], (), print)
-            ['--foo=bar baz']
-
-            Options that are quoted will be unquoted, so the quotes aren't interpreted
-            literally by the postgres command
-            >>> Bootstrap.process_user_options('foo', [{'foo': '"bar baz"'}], (), print)
-            ['--foo=bar baz']
-
-        .. note::
-            The *error_handler* is called when any of these conditions are met:
-
-            * Key, value dictionaries in the list form contains multiple keys.
-            * If a key is listed in *not_allowed_options*.
-            * If the options list is not in the required structure.
-
-        :param tool: The name of the tool used in error reports to *error_handler*
-        :param options: Options to parse as a list of key, values or single values, or a dictionary
-        :param not_allowed_options: List of keys that cannot be used in the list of key, value formatted options
-        :param error_handler: A function which will be called when an error condition is encountered
-        :returns: List of long form arguments to pass to the named tool
-        """
-        user_options: List[str] = []
-
-        def option_is_allowed(name: str) -> bool:
-            ret = name not in not_allowed_options
-            if not ret:
-                error_handler('{0} option for {1} is not allowed'.format(name, tool))
-            return ret
-
-        if isinstance(options, dict):
-            for key, val in cast(Dict[str, str], options).items():
-                if key and val:
-                    user_options.append('--{0}={1}'.format(key, unquote(val)))
-        elif isinstance(options, list):
-            for opt in cast(List[Any], options):
-                if isinstance(opt, str) and option_is_allowed(opt):
-                    user_options.append('--{0}'.format(opt))
-                elif isinstance(opt, dict):
-                    args = cast(Dict[str, Any], opt)
-                    keys = list(args.keys())
-                    if len(keys) == 1 and isinstance(args[keys[0]], str) and option_is_allowed(keys[0]):
-                        user_options.append('--{0}={1}'.format(keys[0], unquote(args[keys[0]])))
-                    else:
-                        error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed'
-                                      ' and value should be a string'.format(tool, args[keys[0]]))
-                else:
-                    error_handler('Error when parsing {0} option {1}: value should be string value'
-                                  ' or a single key-value pair'.format(tool, opt))
-        else:
-            error_handler('{0} options must be list or dict'.format(tool))
-        return user_options
-
     def _initdb(self, config: Any) -> bool:
         self._postgresql.set_state(PostgresqlState.INITDB)
         not_allowed_options = ('pgdata', 'nosync', 'pwfile', 'sync-only', 'version')
 
         def error_handler(e: str) -> None:
             raise Exception(e)
 
-        options = self.process_user_options('initdb', config or [], not_allowed_options, error_handler)
+        options = process_user_options('initdb', config or [], not_allowed_options, error_handler)
         pwfile = None
 
         if self._postgresql.config.superuser:
@@ -341,7 +260,7 @@ def basebackup(self, conn_url: str, env: Dict[str, str], options: Dict[str, Any]
         ret = 1
         not_allowed_options = ('pgdata', 'format', 'wal-method', 'xlog-method', 'gzip',
                                'version', 'compress', 'dbname', 'host', 'port', 'username', 'password')
-        user_options = self.process_user_options('basebackup', options, not_allowed_options, logger.error)
+        user_options = process_user_options('basebackup', options, not_allowed_options, logger.error)
         cmd = [
             self._postgresql.pgcommand("pg_basebackup"),
             "--pgdata=" + self._postgresql.data_dir,
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
@@ -12,6 +12,7 @@
 from ..async_executor import CriticalTask
 from ..collections import EMPTY_DICT
 from ..dcs import Leader, RemoteMember
+from ..utils import process_user_options
 from . import Postgresql
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir, parse_history, parse_lsn, PostgresqlRole
@@ -443,6 +444,10 @@ def pg_rewind(self, conn_kwargs: Dict[str, Any]) -> bool:
 
         :returns: ``True`` if ``pg_rewind`` finished successfully, ``False`` otherwise.
         """
+        options = self._postgresql.config.get('rewind', [])
+        not_allowed_options = ('target-pgdata', 'source-pgdata', 'source-server', 'write-recovery-conf', 'dry-run',
+                               'restore-target-wal', 'config-file', 'no-ensure-shutdown', 'version', 'help')
+        user_options = process_user_options('rewind', options, not_allowed_options, logger.error)
         # prepare pg_rewind connection string
         env = self._postgresql.config.write_pgpass(conn_kwargs)
         env.update(LANG='C', LC_ALL='C', PGOPTIONS='-c statement_timeout=0')
@@ -466,6 +471,7 @@ def pg_rewind(self, conn_kwargs: Dict[str, Any]) -> bool:
                 cmd.append('--config-file={0}'.format(self._postgresql.config.postgresql_conf))
 
         cmd.extend(['-D', self._postgresql.data_dir, '--source-server', dsn])
+        cmd.extend(user_options)
 
         while True:
             results: Dict[str, bytes] = {}
diff --git a/patroni/utils.py b/patroni/utils.py
@@ -1280,3 +1280,86 @@ def get_major_version(bin_dir: Optional[str] = None, bin_name: str = 'postgres')
     """
     full_version = get_postgres_version(bin_dir, bin_name)
     return re.sub(r'\.\d+$', '', full_version)
+
+
+def process_user_options(tool: str, options: Any,
+                         not_allowed_options: Tuple[str, ...],
+                         error_handler: Callable[[str], None]) -> List[str]:
+    """Format *options* in a list or dictionary format into command line long form arguments.
+
+    :Example:
+
+        The *options* can be defined as a dictionary of key, values to be converted into arguments:
+        >>> process_user_options('foo', {'foo': 'bar'}, (), print)
+        ['--foo=bar']
+
+        Or as a list of single string arguments
+        >>> process_user_options('foo', ['yes'], (), print)
+        ['--yes']
+
+        Or as a list of key, value options
+        >>> process_user_options('foo', [{'foo': 'bar'}], (), print)
+        ['--foo=bar']
+
+        Or a combination of single and key, values
+        >>> process_user_options('foo', ['yes', {'foo': 'bar'}], (), print)
+        ['--yes', '--foo=bar']
+
+        Options that contain spaces are passed as is to ``subprocess.call``
+        >>> process_user_options('foo', [{'foo': 'bar baz'}], (), print)
+        ['--foo=bar baz']
+
+        Options that are quoted will be unquoted, so the quotes aren't interpreted
+        literally by the postgres command
+        >>> process_user_options('foo', [{'foo': '"bar baz"'}], (), print)
+        ['--foo=bar baz']
+
+    .. note::
+        The *error_handler* is called when any of these conditions are met:
+
+        * Key, value dictionaries in the list form contains multiple keys.
+        * If a key is listed in *not_allowed_options*.
+        * If the options list is not in the required structure.
+
+    :param tool: The name of the tool used in error reports to *error_handler*
+    :param options: Options to parse as a list of key, values or single values, or a dictionary
+    :param not_allowed_options: List of keys that cannot be used in the list of key, value formatted options
+    :param error_handler: A function which will be called when an error condition is encountered
+    :returns: List of long form arguments to pass to the named tool
+    """
+    user_options: List[str] = []
+
+    def option_is_allowed(name: str) -> bool:
+        ret = name not in not_allowed_options
+        if not ret:
+            error_handler('{0} option for {1} is not allowed'.format(name, tool))
+        return ret
+
+    if isinstance(options, dict):
+        for key, val in cast(Dict[str, str], options).items():
+            if key and val:
+                user_options.append('--{0}={1}'.format(key, unquote(val)))
+    elif isinstance(options, list):
+        for opt in cast(List[Any], options):
+            if isinstance(opt, str):
+                # This if needs to be nested, otherwise we confuse the user by logging two errors -- one issued by
+                # option_is_allowed and another by the else clause below.
+                if option_is_allowed(opt):
+                    user_options.append('--{0}'.format(opt))
+            elif isinstance(opt, dict):
+                args = cast(Dict[str, Any], opt)
+                keys = list(args.keys())
+                if len(keys) == 1 and isinstance(args[keys[0]], str):
+                    # This if needs to be nested, otherwise we confuse the user by logging two errors -- one issued by
+                    # option_is_allowed and another by the else clause below.
+                    if option_is_allowed(keys[0]):
+                        user_options.append('--{0}={1}'.format(keys[0], unquote(args[keys[0]])))
+                else:
+                    error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed'
+                                  ' and value should be a string'.format(tool, args[keys[0]]))
+            else:
+                error_handler('Error when parsing {0} option {1}: value should be string value'
+                              ' or a single key-value pair'.format(tool, opt))
+    else:
+        error_handler('{0} options must be list or dict'.format(tool))
+    return user_options
diff --git a/patroni/validator.py b/patroni/validator.py
@@ -1058,6 +1058,8 @@ def validate_watchdog_mode(value: Any) -> None:
                     Optional("max_worker_processes"): IntValidator(0, 262143, raise_assert=True),
                 },
                 Optional("use_pg_rewind"): bool,
+                Optional("rewind"): [Or(str, dict)],
+                Optional("basebackup"): [Or(str, dict)],
                 Optional("pg_hba"): [str],
                 Optional("pg_ident"): [str],
                 Optional("pg_ctl_timeout"): IntValidator(min=0, raise_assert=True),
@@ -1178,7 +1180,9 @@ def validate_watchdog_mode(value: Any) -> None:
         Optional("pg_hba"): [str],
         Optional("pg_ident"): [str],
         Optional("pg_ctl_timeout"): IntValidator(min=0, raise_assert=True),
-        Optional("use_pg_rewind"): bool
+        Optional("use_pg_rewind"): bool,
+        Optional("rewind"): [Or(str, dict)],
+        Optional("basebackup"): [Or(str, dict)],
     },
     Optional("watchdog"): {
         Optional("mode"): validate_watchdog_mode,
diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
@@ -1,5 +1,4 @@
 import os
-import sys
 
 from unittest.mock import Mock, patch, PropertyMock
 
@@ -115,58 +114,6 @@ def test__initdb(self):
         self.assertRaises(Exception, self.b.bootstrap, {'initdb': [1]})
         self.assertRaises(Exception, self.b.bootstrap, {'initdb': 1})
 
-    def test__process_user_options(self):
-        def error_handler(msg):
-            raise Exception(msg)
-
-        self.assertEqual(self.b.process_user_options('initdb', ['string'], (), error_handler), ['--string'])
-        self.assertEqual(
-            self.b.process_user_options(
-                'initdb',
-                [{'key': 'value'}],
-                (), error_handler
-            ),
-            ['--key=value'])
-        if sys.platform != 'win32':
-            self.assertEqual(
-                self.b.process_user_options(
-                    'initdb',
-                    [{'key': 'value with spaces'}],
-                    (), error_handler
-                ),
-                ["--key=value with spaces"])
-            self.assertEqual(
-                self.b.process_user_options(
-                    'initdb',
-                    [{'key': "'value with spaces'"}],
-                    (), error_handler
-                ),
-                ["--key=value with spaces"])
-            self.assertEqual(
-                self.b.process_user_options(
-                    'initdb',
-                    {'key': 'value with spaces'},
-                    (), error_handler
-                ),
-                ["--key=value with spaces"])
-            self.assertEqual(
-                self.b.process_user_options(
-                    'initdb',
-                    {'key': "'value with spaces'"},
-                    (), error_handler
-                ),
-                ["--key=value with spaces"])
-            # not allowed options in list of dicts/strs are filtered out
-            self.assertEqual(
-                self.b.process_user_options(
-                    'pg_basebackup',
-                    [{'checkpoint': 'fast'}, {'dbname': 'dbname=postgres'}, 'gzip', {'label': 'standby'}, 'verbose'],
-                    ('dbname', 'verbose'),
-                    print
-                ),
-                ['--checkpoint=fast', '--gzip', '--label=standby'],
-            )
-
     @patch.object(CancellableSubprocess, 'call', Mock())
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
diff --git a/tests/test_utils.py b/tests/test_utils.py
