Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

docs/yaml_configuration.rst

@@ -30,9 +30,15 @@ Log
 
 Bootstrap configuration
 -----------------------
+
+.. note::
+    Once Patroni has initialized the cluster for the first time and settings have been stored in the DCS, all future
+    changes to the ``bootstrap.dcs`` section of the YAML configuration will not take any effect! If you want to change
+    them please use either ``patronictl edit-config`` or the Patroni :ref:`REST API <rest_api>`.
+
 -  **bootstrap**:
 
-   -  **dcs**: This section will be written into `/<namespace>/<scope>/config` of the given configuration store after initializing of new cluster. The global dynamic configuration for the cluster. Under the ``bootstrap.dcs`` you can put any of the parameters described in the :ref:`Dynamic Configuration settings <dynamic_configuration>` and after Patroni initialized (bootstrapped) the new cluster, it will write this section into `/<namespace>/<scope>/config` of the configuration store. All later changes of ``bootstrap.dcs`` will not take any effect! If you want to change them please use either ``patronictl edit-config`` or Patroni :ref:`REST API <rest_api>`.
+   -  **dcs**: This section will be written into `/<namespace>/<scope>/config` of the given configuration store after initializing the new cluster. The global dynamic configuration for the cluster. You can put any of the parameters described in the :ref:`Dynamic Configuration settings <dynamic_configuration>` under ``bootstrap.dcs`` and after Patroni has initialized (bootstrapped) the new cluster, it will write this section into `/<namespace>/<scope>/config` of the configuration store.
    -  **method**: custom script to use for bootstrapping this cluster.
 
       See :ref:`custom bootstrap methods documentation <custom_bootstrap>` for details.

patroni/__main__.py

@@ -163,11 +163,26 @@ def patroni_main(configfile: str) -> None:
 
 
 def process_arguments() -> Namespace:
+    from patroni.config_generator import generate_config
+
     parser = get_base_arg_parser()
-    parser.add_argument('--validate-config', action='store_true', help='Run config validator and exit')
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('--validate-config', action='store_true', help='Run config validator and exit')
+    group.add_argument('--generate-sample-config', action='store_true',
+                       help='Generate a sample Patroni yaml configuration file')
+    group.add_argument('--generate-config', action='store_true',
+                       help='Generate a Patroni yaml configuration file for a running instance')
+    parser.add_argument('--dsn', help='Optional DSN string of the instance to be used as a source \
+                                    for config generation. Superuser connection is required.')
     args = parser.parse_args()
 
-    if args.validate_config:
+    if args.generate_sample_config:
+        generate_config(args.configfile, True, None)
+        sys.exit(0)
+    elif args.generate_config:
+        generate_config(args.configfile, False, args.dsn)
+        sys.exit(0)
+    elif args.validate_config:
         from patroni.validator import schema
         from patroni.config import Config, ConfigParseError
 

patroni/collections.py

@@ -3,7 +3,7 @@
 Provides a case insensitive :class:`dict` and :class:`set` object types.
 """
 from collections import OrderedDict
-from typing import Any, Collection, Dict, Iterator, MutableMapping, MutableSet, Optional
+from typing import Any, Collection, Dict, Iterator, KeysView, MutableMapping, MutableSet, Optional
 
 
 class CaseInsensitiveSet(MutableSet[str]):
@@ -187,6 +187,13 @@ class CaseInsensitiveDict(MutableMapping[str, Any]):
         """
         return CaseInsensitiveDict({v[0]: v[1] for v in self._values.values()})
 
+    def keys(self) -> KeysView[str]:
+        """Return a new view of the dict's keys.
+
+        :returns: a set-like object providing a view on the dict's keys
+        """
+        return self._values.keys()
+
     def __repr__(self) -> str:
         """Get a string representation of the dict.
 

patroni/config.py

@@ -199,10 +199,9 @@ class Config(object):
             'recovery_min_apply_delay': ''
         },
         'postgresql': {
-            'bin_dir': '',
             'use_slots': True,
             'parameters': CaseInsensitiveDict({p: v[0] for p, v in ConfigHandler.CMDLINE_OPTIONS.items()
-                                               if p not in ('wal_keep_segments', 'wal_keep_size')})
+                                               if v[0] is not None and p not in ('wal_keep_segments', 'wal_keep_size')})
         }
     }
 
@@ -229,7 +228,8 @@ class Config(object):
         self.__effective_configuration = self._build_effective_configuration({}, self._local_configuration)
         self._data_dir = self.__effective_configuration.get('postgresql', {}).get('data_dir', "")
         self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME)
-        self._load_cache()
+        if validator:  # patronictl uses validator=None and we don't want to load anything from local cache in this case
+            self._load_cache()
         self._cache_needs_saving = False
 
     @property
@@ -240,6 +240,22 @@ class Config(object):
     def dynamic_configuration(self) -> Dict[str, Any]:
         return deepcopy(self._dynamic_configuration)
 
+    @property
+    def local_configuration(self) -> Dict[str, Any]:
+        """Deep copy of cached Patroni local configuration.
+
+        :returns: copy of :attr:`~Config._local_configuration`
+        """
+        return deepcopy(dict(self._local_configuration))
+
+    @classmethod
+    def get_default_config(cls) -> Dict[str, Any]:
+        """Deep copy default configuration.
+
+        :returns: copy of :attr:`~Config.__DEFAULT_CONFIG`
+        """
+        return deepcopy(cls.__DEFAULT_CONFIG)
+
     def _load_config_path(self, path: str) -> Dict[str, Any]:
         """
         If path is a file, loads the yml file pointed to by path.
@@ -346,13 +362,13 @@ class Config(object):
                 if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
                     pg_params[name] = value
                 else:
-                    logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
-                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
+                    logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
+                                   name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
 
         return pg_params
 
     def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]:
-        config = deepcopy(self.__DEFAULT_CONFIG)
+        config = self.get_default_config()
 
         for name, value in dynamic_configuration.items():
             if name == 'postgresql':

patroni/config_generator.py

@@ -0,0 +1,463 @@
+"""patroni ``--generate-config`` machinery."""
+import abc
+import logging
+import os
+import psutil
+import socket
+import sys
+import yaml
+
+from getpass import getuser, getpass
+from contextlib import contextmanager
+from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union
+if TYPE_CHECKING:  # pragma: no cover
+    from psycopg import Cursor
+    from psycopg2 import cursor
+
+from . import psycopg
+from .config import Config
+from .exceptions import PatroniException
+from .postgresql.config import ConfigHandler, parse_dsn
+from .postgresql.misc import postgres_major_version_to_int
+from .utils import get_major_version, parse_bool, patch_config, read_stripped
+
+
+# Mapping between the libpq connection parameters and the environment variables.
+# This dict should be kept in sync with `patroni.utils._AUTH_ALLOWED_PARAMETERS`
+# (we use "username" in the Patroni config for some reason, other parameter names are the same).
+_AUTH_ALLOWED_PARAMETERS_MAPPING = {
+    'user': 'PGUSER',
+    'password': 'PGPASSWORD',
+    'sslmode': 'PGSSLMODE',
+    'sslcert': 'PGSSLCERT',
+    'sslkey': 'PGSSLKEY',
+    'sslpassword': '',
+    'sslrootcert': 'PGSSLROOTCERT',
+    'sslcrl': 'PGSSLCRL',
+    'sslcrldir': 'PGSSLCRLDIR',
+    'gssencmode': 'PGGSSENCMODE',
+    'channel_binding': 'PGCHANNELBINDING'
+}
+_NO_VALUE_MSG = '#FIXME'
+
+
+def get_address() -> Tuple[str, str]:
+    """Try to get hostname and the ip address for it returned by :func:`~socket.gethostname`.
+
+    .. note::
+        Can also return local ip.
+
+    :returns: tuple consisting of the hostname returned by :func:`~socket.gethostname`
+        and the first element in the sorted list of the addresses returned by :func:`~socket.getaddrinfo`.
+        Sorting guarantees it will prefer IPv4.
+        If an exception occured, hostname and ip values are equal to :data:`~patroni.config_generator._NO_VALUE_MSG`.
+    """
+    hostname = None
+    try:
+        hostname = socket.gethostname()
+        return hostname, sorted(socket.getaddrinfo(hostname, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0),
+                                key=lambda x: x[0])[0][4][0]
+    except Exception as err:
+        logging.warning('Failed to obtain address: %r', err)
+        return _NO_VALUE_MSG, _NO_VALUE_MSG
+
+
+class AbstractConfigGenerator(abc.ABC):
+    """Object representing the generated Patroni config.
+
+    :ivar output_file: full path to the output file to be used.
+    :ivar pg_major: integer representation of the major PostgreSQL version.
+    :ivar config: dictionary used for the generated configuration storage.
+    """
+
+    _HOSTNAME, _IP = get_address()
+
+    def __init__(self, output_file: Optional[str]) -> None:
+        """Set up the output file (if passed), helper vars and the minimal config structure.
+
+        :param output_file: full path to the output file to be used.
+        """
+        self.output_file = output_file
+        self.pg_major = 0
+        self.config = self.get_template_config()
+
+        self.generate()
+
+    @classmethod
+    def get_template_config(cls) -> Dict[str, Any]:
+        """Generate a template config for further extension (e.g. in the inherited classes).
+
+        :returns: dictionary with the values gathered from Patroni env, hopefully defined hostname and ip address
+                  (otherwise set to :data:`~patroni.config_generator._NO_VALUE_MSG`), and some sane defaults.
+        """
+        template_config: Dict[str, Any] = {
+            'scope': _NO_VALUE_MSG,
+            'name': cls._HOSTNAME,
+            'postgresql': {
+                'data_dir': _NO_VALUE_MSG,
+                'connect_address': _NO_VALUE_MSG + ':5432',
+                'listen': _NO_VALUE_MSG + ':5432',
+                'bin_dir': '',
+                'authentication': {
+                    'superuser': {
+                        'username': 'postgres',
+                        'password': _NO_VALUE_MSG
+                    },
+                    'replication': {
+                        'username': 'replicator',
+                        'password': _NO_VALUE_MSG
+                    }
+                }
+            },
+            'restapi': {
+                'connect_address': cls._IP + ':8008',
+                'listen': cls._IP + ':8008'
+            }
+        }
+
+        dynamic_config = Config.get_default_config()
+        # to properly dump CaseInsensitiveDict as YAML later
+        dynamic_config['postgresql']['parameters'] = dict(dynamic_config['postgresql']['parameters'])
+        config = Config('', None).local_configuration  # Get values from env
+        config.setdefault('bootstrap', {})['dcs'] = dynamic_config
+        config.setdefault('postgresql', {})
+        del config['bootstrap']['dcs']['standby_cluster']
+
+        patch_config(template_config, config)
+        return template_config
+
+    @abc.abstractmethod
+    def generate(self) -> None:
+        """Generate config and store in :attr:`~AbstractConfigGenerator.config`."""
+
+    def write_config(self) -> None:
+        """Write current :attr:`~AbstractConfigGenerator.config` to the output file if provided, to stdout otherwise."""
+        if self.output_file:
+            dir_path = os.path.dirname(self.output_file)
+            if dir_path and not os.path.isdir(dir_path):
+                os.makedirs(dir_path)
+            with open(self.output_file, 'w', encoding='UTF-8') as output_file:
+                yaml.safe_dump(self.config, output_file, default_flow_style=False, allow_unicode=True)
+        else:
+            yaml.safe_dump(self.config, sys.stdout, default_flow_style=False, allow_unicode=True)
+
+
+class SampleConfigGenerator(AbstractConfigGenerator):
+    """Object representing the generated sample Patroni config.
+
+    Sane defults are used based on the gathered PG version.
+    """
+
+    @property
+    def get_auth_method(self) -> str:
+        """Return the preferred authentication method for a specific PG version if provided or the default ``md5``.
+
+        :returns: :class:`str` value for the preferred authentication method.
+        """
+        return 'scram-sha-256' if self.pg_major and self.pg_major >= 100000 else 'md5'
+
+    def _get_int_major_version(self) -> int:
+        """Get major PostgreSQL version from the binary as an integer.
+
+        :returns: an integer PostgreSQL major version representation gathered from the PostgreSQL binary.
+                  See :func:`~patroni.postgresql.misc.postgres_major_version_to_int` and
+                  :func:`~patroni.utils.get_major_version`.
+        """
+        postgres_bin = ((self.config.get('postgresql') or {}).get('bin_name') or {}).get('postgres', 'postgres')
+        return postgres_major_version_to_int(get_major_version(self.config['postgresql'].get('bin_dir'), postgres_bin))
+
+    def generate(self) -> None:
+        """Generate sample config using some sane defaults and update :attr:`~AbstractConfigGenerator.config`."""
+        self.pg_major = self._get_int_major_version()
+
+        self.config['postgresql']['parameters'] = {'password_encryption': self.get_auth_method}
+        username = self.config["postgresql"]["authentication"]["replication"]["username"]
+        self.config['postgresql']['pg_hba'] = [
+            f'host all all all {self.get_auth_method}',
+            f'host replication {username} all {self.get_auth_method}'
+        ]
+
+        # add version-specific configuration
+        wal_keep_param = 'wal_keep_segments' if self.pg_major < 130000 else 'wal_keep_size'
+        self.config['bootstrap']['dcs']['postgresql']['parameters'][wal_keep_param] = \
+            ConfigHandler.CMDLINE_OPTIONS[wal_keep_param][0]
+
+        self.config['bootstrap']['dcs']['postgresql']['use_pg_rewind'] = True
+        if self.pg_major >= 110000:
+            self.config['postgresql']['authentication'].setdefault(
+                'rewind', {'username': 'rewind_user'}).setdefault('password', _NO_VALUE_MSG)
+
+
+class RunningClusterConfigGenerator(AbstractConfigGenerator):
+    """Object representing the Patroni config generated using information gathered from the running instance.
+
+    :ivar dsn: DSN string for the local instance to get GUC values from (if provided).
+    :ivar parsed_dsn: DSN string parsed into a dictionary (see :func:`~patroni.postgresql.config.parse_dsn`).
+    """
+
+    def __init__(self, output_file: Optional[str] = None, dsn: Optional[str] = None) -> None:
+        """Additionally store the passed dsn (if any) in both original and parsed version and run config generation.
+
+        :param output_file: full path to the output file to be used.
+        :param dsn: DSN string for the local instance to get GUC values from.
+
+        :raises:
+            :exc:`~patroni.exceptions.PatroniException`: if DSN parsing failed.
+        """
+        self.dsn = dsn
+        self.parsed_dsn = {}
+
+        super().__init__(output_file)
+
+    @property
+    def _get_hba_conn_types(self) -> Tuple[str, ...]:
+        """Return the connection types allowed.
+
+        If :attr:`~RunningClusterConfigGenerator.pg_major` is defined, adds additional parameters
+        for PostgreSQL version >=16.
+
+        :returns: tuple of the connection methods allowed.
+        """
+        allowed_types = ('local', 'host', 'hostssl', 'hostnossl', 'hostgssenc', 'hostnogssenc')
+        if self.pg_major and self.pg_major >= 160000:
+            allowed_types += ('include', 'include_if_exists', 'include_dir')
+        return allowed_types
+
+    @property
+    def _required_pg_params(self) -> List[str]:
+        """PG configuration prameters that have to be always present in the generated config.
+
+        :returns: list of the parameter names.
+        """
+        return ['hba_file', 'ident_file', 'config_file', 'data_directory'] + \
+            list(ConfigHandler.CMDLINE_OPTIONS.keys())
+
+    def _get_bin_dir_from_running_instance(self) -> str:
+        """Define the directory postgres binaries reside using postmaster's pid executable.
+
+        :returns: path to the PostgreSQL binaries directory.
+
+        :raises:
+            :exc:`~patroni.exceptions.PatroniException`: if:
+
+                * pid could not be obtained from the ``postmaster.pid`` file; or
+                * :exc:`OSError` occured during ``postmaster.pid`` file handling; or
+                * the obtained postmaster pid doesn't exist.
+        """
+        postmaster_pid = None
+        data_dir = self.config['postgresql']['data_dir']
+        try:
+            with open(f"{data_dir}/postmaster.pid", 'r') as pid_file:
+                postmaster_pid = pid_file.readline()
+                if not postmaster_pid:
+                    raise PatroniException('Failed to obtain postmaster pid from postmaster.pid file')
+                postmaster_pid = int(postmaster_pid.strip())
+        except OSError as err:
+            raise PatroniException(f'Error while reading postmaster.pid file: {err}')
+        try:
+            return os.path.dirname(psutil.Process(postmaster_pid).exe())
+        except psutil.NoSuchProcess:
+            raise PatroniException("Obtained postmaster pid doesn't exist.")
+
+    @contextmanager
+    def _get_connection_cursor(self) -> Iterator[Union['cursor', 'Cursor[Any]']]:
+        """Get cursor for the PG connection established based on the stored information.
+
+        :raises:
+            :exc:`~patroni.exceptions.PatroniException`: if :exc:`psycopg.Error` occured.
+        """
+        try:
+            conn = psycopg.connect(dsn=self.dsn,
+                                   password=self.config['postgresql']['authentication']['superuser']['password'])
+            with conn.cursor() as cur:
+                yield cur
+            conn.close()
+        except psycopg.Error as e:
+            raise PatroniException(f'Failed to establish PostgreSQL connection: {e}')
+
+    def _set_pg_params(self, cur: Union['cursor', 'Cursor[Any]']) -> None:
+        """Extend :attr:`~RunningClusterConfigGenerator.config` with the actual PG GUCs values.
+
+        THe following GUC values are set:
+
+            * Non-internal having configuration file, postmaster command line or environment variable
+              as a source.
+
+            * List of the always required parameters (see :meth:`~RunningClusterConfigGenerator._required_pg_params`).
+
+        :param cur: connection cursor to use.
+        """
+        cur.execute("SELECT name, current_setting(name) FROM pg_settings "
+                    "WHERE context <> 'internal' "
+                    "AND source IN ('configuration file', 'command line', 'environment variable') "
+                    "AND category <> 'Write-Ahead Log / Recovery Target' "
+                    "AND setting <> '(disabled)' "
+                    "OR name = ANY(%s)", (self._required_pg_params,))
+
+        helper_dict = dict.fromkeys(['port', 'listen_addresses'])
+        self.config['postgresql'].setdefault('parameters', {})
+        for param, value in cur.fetchall():
+            if param == 'data_directory':
+                self.config['postgresql']['data_dir'] = value
+            elif param == 'cluster_name' and value:
+                self.config['scope'] = value
+            elif param in ('archive_command', 'restore_command',
+                           'archive_cleanup_command', 'recovery_end_command',
+                           'ssl_passphrase_command', 'hba_file',
+                           'ident_file', 'config_file'):
+                # write commands to the local config due to security implications
+                # write hba/ident/config_file to local config to ensure they are not removed later
+                self.config['postgresql']['parameters'][param] = value
+            elif param in helper_dict:
+                helper_dict[param] = value
+            else:
+                self.config['bootstrap']['dcs']['postgresql']['parameters'][param] = value
+
+        connect_port = self.parsed_dsn.get('port', os.getenv('PGPORT', helper_dict['port']))
+        self.config['postgresql']['connect_address'] = f'{self._IP}:{connect_port}'
+        self.config['postgresql']['listen'] = f'{helper_dict["listen_addresses"]}:{helper_dict["port"]}'
+
+    def _set_su_params(self) -> None:
+        """Extend :attr:`~RunningClusterConfigGenerator.config` with the superuser auth information.
+
+        Information set is based on the options used for connection.
+        """
+        su_params: Dict[str, str] = {}
+        for conn_param, env_var in _AUTH_ALLOWED_PARAMETERS_MAPPING.items():
+            val = self.parsed_dsn.get(conn_param, os.getenv(env_var))
+            if val:
+                su_params[conn_param] = val
+        patroni_env_su_username = ((self.config.get('authentication') or {}).get('superuser') or {}).get('username')
+        patroni_env_su_pwd = ((self.config.get('authentication') or {}).get('superuser') or {}).get('password')
+        # because we use "username" in the config for some reason
+        su_params['username'] = su_params.pop('user', patroni_env_su_username) or getuser()
+        su_params['password'] = su_params.get('password', patroni_env_su_pwd) or \
+            getpass('Please enter the user password:')
+        self.config['postgresql']['authentication'] = {
+            'superuser': su_params,
+            'replication': {'username': _NO_VALUE_MSG, 'password': _NO_VALUE_MSG}
+        }
+
+    def _set_conf_files(self) -> None:
+        """Extend :attr:`~RunningClusterConfigGenerator.config` with ``pg_hba.conf`` and ``pg_ident.conf`` content.
+
+        .. note::
+            This function only defines ``postgresql.pg_hba`` and ``postgresql.pg_ident`` when
+            ``hba_file`` and ``ident_file`` are set to the defaults. It may happen these files
+            are located outside of ``PGDATA`` and Patroni doesn't have write permissions for them.
+
+        :raises:
+            :exc:`~patroni.exceptions.PatroniException`: if :exc:`OSError` occured during the conf files handling.
+        """
+        default_hba_path = os.path.join(self.config['postgresql']['data_dir'], 'pg_hba.conf')
+        if self.config['postgresql']['parameters']['hba_file'] == default_hba_path:
+            try:
+                self.config['postgresql']['pg_hba'] = list(
+                    filter(lambda i: i and i.split()[0] in self._get_hba_conn_types, read_stripped(default_hba_path)))
+            except OSError as err:
+                raise PatroniException(f'Failed to read pg_hba.conf: {err}')
+
+        default_ident_path = os.path.join(self.config['postgresql']['data_dir'], 'pg_ident.conf')
+        if self.config['postgresql']['parameters']['ident_file'] == default_ident_path:
+            try:
+                self.config['postgresql']['pg_ident'] = [i for i in read_stripped(default_ident_path)
+                                                         if i and not i.startswith('#')]
+            except OSError as err:
+                raise PatroniException(f'Failed to read pg_ident.conf: {err}')
+            if not self.config['postgresql']['pg_ident']:
+                del self.config['postgresql']['pg_ident']
+
+    def _enrich_config_from_running_instance(self) -> None:
+        """Extend :attr:`~RunningClusterConfigGenerator.config` with the values gathered from the running instance.
+
+        Retrieve the following information from the running PostgreSQL instance:
+
+        * superuser auth parameters (see :meth:`~RunningClusterConfigGenerator._set_su_params`);
+        * some GUC values (see :meth:`~RunningClusterConfigGenerator._set_pg_params`);
+        * ``postgresql.connect_address``, ``postgresql.listen``;
+        * ``postgresql.pg_hba`` and ``postgresql.pg_ident`` (see :meth:`~RunningClusterConfigGenerator._set_conf_files`)
+
+        And redefine ``scope`` with the ``cluster_name`` GUC value if set.
+
+        :raises:
+            :exc:`~patroni.exceptions.PatroniException`: if the provided user doesn't have superuser privileges.
+        """
+        self._set_su_params()
+
+        with self._get_connection_cursor() as cur:
+            self.pg_major = getattr(cur.connection, 'server_version', 0)
+
+            if not parse_bool(cur.connection.info.parameter_status('is_superuser')):
+                raise PatroniException('The provided user does not have superuser privilege')
+
+            self._set_pg_params(cur)
+
+        self._set_conf_files()
+
+    def generate(self) -> None:
+        """Generate config using the info gathered from the specified running PG instance.
+
+        Result is written to :attr:`~RunningClusterConfigGenerator.config`.
+        """
+        if self.dsn:
+            self.parsed_dsn = parse_dsn(self.dsn) or {}
+            if not self.parsed_dsn:
+                raise PatroniException('Failed to parse DSN string')
+
+        self._enrich_config_from_running_instance()
+        self.config['postgresql']['bin_dir'] = self._get_bin_dir_from_running_instance()
+
+
+def generate_config(output_file: str, sample: bool, dsn: Optional[str]) -> None:
+    """Generate Patroni configuration file.
+
+    Gather all the available non-internal GUC values having configuration file, postmaster command line or environment
+    variable as a source and store them in the appropriate part of Patroni configuration (``postgresql.parameters`` or
+    ``bootstrap.dcs.postgresql.parameters``). Either the provided DSN (takes precedence) or PG ENV vars will be used
+    for the connection. If password is not provided, it should be entered via prompt.
+
+    The created configuration contains:
+    * ``scope``: ``cluster_name`` GUC value or ``PATRONI_SCOPE ENV`` variable value if available.
+    * ``name``: ``PATRONI_NAME`` ENV variable value if set, otherwise hostname.
+
+    * ``bootstrap.dcs``: section with all the parameters (incl. the majority of PG GUCs) set to their default values
+      defined by Patroni and adjusted by the source instances's configuration values.
+
+    * ``postgresql.parameters``: the source instance's ``archive_command``, ``restore_command``,
+      ``archive_cleanup_command``, ``recovery_end_command``, ``ssl_passphrase_command``, ``hba_file``, ``ident_file``,
+      ``config_file`` GUC values.
+
+    * ``postgresql.bin_dir``: path to Postgres binaries gathered from the running instance or, if not available,
+      the value of ``PATRONI_POSTGRESQL_BIN_DIR`` ENV variable. Otherwise, an empty string.
+
+    * ``postgresql.datadir``: the value gathered from the corresponding PG GUC.
+    * ``postgresql.listen``: source instance's ``listen_addresses`` and port GUC values.
+    * ``postgresql.connect_address``: if possible, generated from the connection params.
+    * ``postgresql.authentication``:
+
+        * superuser and replication users defined (if possible, usernames are set from the respective Patroni ENV vars,
+          otherwise the default ``postgres`` and ``replicator`` values are used).
+          If not a sample config, either DSN or PG ENV vars are used to define superuser authentication parameters.
+
+        * rewind user is defined only for sample config, if PG version can be defined and PG version is >=11
+          (if possible, username is set from the respective Patroni ENV var).
+
+    * ``bootstrap.dcs.postgresql.use_pg_rewind`` set to ``True`` for a sample config only.
+    * ``postgresql.pg_hba`` defaults or the lines gathered from the source instance's ``hba_file``.
+    * ``postgresql.pg_ident`` the lines gathered from the source instance's ``ident_file``.
+
+    :param output_file: Full path to the configuration file to be used. If not provided, result is sent to ``stdout``.
+    :param sample: Optional flag. If set, no source instance will be used - generate config with some sane defaults.
+    :param dsn: Optional DSN string for the local instance to get GUC values from.
+    """
+    try:
+        if sample:
+            config_generator = SampleConfigGenerator(output_file)
+        else:
+            config_generator = RunningClusterConfigGenerator(output_file, dsn)
+
+        config_generator.write_config()
+    except PatroniException as e:
+        sys.exit(str(e))
+    except Exception as e:
+        sys.exit(f'Unexpected exception: {e}')

patroni/dcs/__init__.py

@@ -1766,26 +1766,29 @@ class AbstractDCS(abc.ABC):
         """
 
     @abc.abstractmethod
-    def _delete_leader(self) -> bool:
+    def _delete_leader(self, leader: Leader) -> bool:
         """Remove leader key from DCS.
 
         This method should remove leader key if current instance is the leader.
 
+        :param leader: :class:`Leader` object with information about the leader.
+
         :returns: ``True`` if successfully committed to DCS.
         """
 
-    def delete_leader(self, last_lsn: Optional[int] = None) -> bool:
+    def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool:
         """Update ``optime/leader`` and voluntarily remove leader key from DCS.
 
         This method should remove leader key if current instance is the leader.
 
+        :param leader: :class:`Leader` object with information about the leader.
         :param last_lsn: latest checkpoint location in bytes.
 
         :returns: boolean result of called abstract :meth:`~AbstractDCS._delete_leader`.
         """
         if last_lsn:
             self.write_status({self._OPTIME: last_lsn})
-        return self._delete_leader()
+        return bool(leader) and self._delete_leader(leader)
 
     @abc.abstractmethod
     def cancel_initialization(self) -> bool:

patroni/dcs/consul.py

@@ -643,12 +643,8 @@ class Consul(AbstractDCS):
         return self._client.kv.put(self.history_path, value)
 
     @catch_consul_errors
-    def _delete_leader(self) -> bool:
-        cluster = self.cluster
-        if cluster and isinstance(cluster.leader, Leader) and\
-                cluster.leader.name == self._name and isinstance(cluster.leader.version, int):
-            return self._client.kv.delete(self.leader_path, cas=cluster.leader.version)
-        return True
+    def _delete_leader(self, leader: Leader) -> bool:
+        return self._client.kv.delete(self.leader_path, cas=int(leader.version))
 
     @catch_consul_errors
     def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]:

patroni/dcs/etcd.py

@@ -809,7 +809,7 @@ class Etcd(AbstractEtcd):
         return bool(self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new)))
 
     @catch_etcd_errors
-    def _delete_leader(self) -> bool:
+    def _delete_leader(self, leader: Leader) -> bool:
         return bool(self._client.delete(self.leader_path, prevValue=self._name))
 
     @catch_etcd_errors

patroni/dcs/etcd3.py

@@ -912,11 +912,10 @@ class Etcd3(AbstractEtcd):
         return self.retry(self._client.put, self.initialize_path, sysid, create_revision='0' if create_new else None)
 
     @catch_etcd_errors
-    def _delete_leader(self) -> bool:
-        cluster = self.cluster
-        if cluster and isinstance(cluster.leader, Leader) and cluster.leader.name == self._name:
-            return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.version)
-        return True
+    def _delete_leader(self, leader: Leader) -> bool:
+        fields = build_range_request(self.leader_path)
+        compare = {'key': fields['key'], 'target': 'VALUE', 'value': base64_encode(self._name)}
+        return bool(self._client.txn(compare, {'request_delete_range': fields}))
 
     @catch_etcd_errors
     def cancel_initialization(self) -> bool:

patroni/dcs/kubernetes.py

@@ -1308,11 +1308,11 @@ class Kubernetes(AbstractDCS):
             if cluster and cluster.config and cluster.config.version else None
         return self.patch_or_create_config({self._INITIALIZE: sysid}, resource_version)
 
-    def _delete_leader(self) -> bool:
+    def _delete_leader(self, leader: Leader) -> bool:
         """Unused"""
         raise NotImplementedError  # pragma: no cover
 
-    def delete_leader(self, last_lsn: Optional[int] = None) -> bool:
+    def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool:
         ret = False
         kind = self._kinds.get(self.leader_path)
         if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name:

patroni/dcs/raft.py

@@ -446,7 +446,7 @@ class Raft(AbstractDCS):
     def initialize(self, create_new: bool = True, sysid: str = '') -> bool:
         return self._sync_obj.set(self.initialize_path, sysid, prevExist=(not create_new)) is not False
 
-    def _delete_leader(self) -> bool:
+    def _delete_leader(self, leader: Leader) -> bool:
         return self._sync_obj.delete(self.leader_path, prevValue=self._name, timeout=1)
 
     def cancel_initialization(self) -> bool:

patroni/dcs/zookeeper.py

@@ -466,7 +466,7 @@ class ZooKeeper(AbstractDCS):
                 return False
         return True
 
-    def _delete_leader(self) -> bool:
+    def _delete_leader(self, leader: Leader) -> bool:
         self._client.restart()
         return True
 

patroni/ha.py

@@ -148,8 +148,8 @@ class Ha(object):
         self.cluster = Cluster.empty()
         self.global_config = self.patroni.config.get_global_config(None)
         self.old_cluster = Cluster.empty()
-        self._is_leader = False
-        self._is_leader_lock = RLock()
+        self._leader_expiry = 0
+        self._leader_expiry_lock = RLock()
         self._failsafe = Failsafe(patroni.dcs)
         self._was_paused = False
         self._promote_timestamp = 0
@@ -195,12 +195,20 @@ class Ha(object):
         return self.global_config.is_standby_cluster
 
     def is_leader(self) -> bool:
-        with self._is_leader_lock:
-            return self._is_leader > time.time()
+        """:returns: `True` if the current node is the leader, based on expiration set when it last held the key."""
+        with self._leader_expiry_lock:
+            return self._leader_expiry > time.time()
 
     def set_is_leader(self, value: bool) -> None:
-        with self._is_leader_lock:
-            self._is_leader = time.time() + self.dcs.ttl if value else 0
+        """Update the current node's view of it's own leadership status.
+
+        Will update the expiry timestamp to match the dcs ttl if setting leadership to true,
+        otherwise will set the expiry to the past to immediately invalidate.
+
+        :param value: is the current node the leader.
+        """
+        with self._leader_expiry_lock:
+            self._leader_expiry = time.time() + self.dcs.ttl if value else 0
             if not value:
                 self._promote_timestamp = 0
 
@@ -583,7 +591,7 @@ class Ha(object):
         if refresh:
             self.load_cluster_from_dcs()
 
-        is_leader = self.state_handler.is_leader()
+        is_leader = self.state_handler.is_primary()
 
         node_to_follow = self._get_node_to_follow(self.cluster)
 
@@ -772,7 +780,7 @@ class Ha(object):
             # in the /sync key. Further changes of synchronous_standby_names and /sync key should
             # be postponed for `loop_wait` seconds, to give a chance to some replicas to start streaming.
             # In opposite case the /sync key will end up without synchronous nodes.
-            if self.state_handler.is_leader():
+            if self.state_handler.is_primary():
                 if self._promote_timestamp == 0 or time.time() - self._promote_timestamp > self.dcs.loop_wait:
                     self._process_quorum_replication()
                 if self._promote_timestamp == 0:
@@ -892,7 +900,7 @@ class Ha(object):
         """
         if not self.is_paused():
             if not self.watchdog.is_running and not self.watchdog.activate():
-                if self.state_handler.is_leader():
+                if self.state_handler.is_primary():
                     self.demote('immediate')
                     return 'Demoting self because watchdog could not be activated'
                 else:
@@ -908,7 +916,7 @@ class Ha(object):
                     self._async_response.reset()
                     return 'Promotion cancelled because the pre-promote script failed'
 
-        if self.state_handler.is_leader():
+        if self.state_handler.is_primary():
             # Inform the state handler about its primary role.
             # It may be unaware of it if postgres is promoted manually.
             self.state_handler.set_role('master')
@@ -1135,7 +1143,7 @@ class Ha(object):
                 # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely
                 # In order to avoid attempts to delete this key from all nodes only the primary is allowed to do it.
                 if not self.cluster.get_member(failover.candidate, fallback_to_leader=False)\
-                        and self.state_handler.is_leader():
+                        and self.state_handler.is_primary():
                     logger.warning("manual failover: removing failover key because failover candidate is not running")
                     self.dcs.manual_failover('', '', version=failover.version)
                     return None
@@ -1194,7 +1202,7 @@ class Ha(object):
             if ret is not None:  # continue if we just deleted the stale failover key as a leader
                 return ret
 
-        if self.state_handler.is_leader():
+        if self.state_handler.is_primary():
             if self.is_paused():
                 # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
                 return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize
@@ -1257,7 +1265,7 @@ class Ha(object):
 
     def _delete_leader(self, last_lsn: Optional[int] = None) -> None:
         self.set_is_leader(False)
-        self.dcs.delete_leader(last_lsn)
+        self.dcs.delete_leader(self.cluster.leader, last_lsn)
         self.dcs.reset_cluster()
 
     def release_leader_key_voluntarily(self, last_lsn: Optional[int] = None) -> None:
@@ -1385,7 +1393,7 @@ class Ha(object):
 
         :returns: action message if demote was initiated, None if no action was taken"""
         failover = self.cluster.failover
-        if not failover or (self.is_paused() and not self.state_handler.is_leader()):
+        if not failover or (self.is_paused() and not self.state_handler.is_primary()):
             return
 
         if (failover.scheduled_at and not
@@ -1455,7 +1463,7 @@ class Ha(object):
 
     def process_healthy_cluster(self) -> str:
         if self.has_lock():
-            if self.is_paused() and not self.state_handler.is_leader():
+            if self.is_paused() and not self.state_handler.is_primary():
                 if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name:
                     return 'waiting to become primary after promote...'
 
@@ -1487,7 +1495,7 @@ class Ha(object):
             else:
                 # Either there is no connection to DCS or someone else acquired the lock
                 logger.error('failed to update leader lock')
-                if self.state_handler.is_leader():
+                if self.state_handler.is_primary():
                     if self.is_paused():
                         return 'continue to run as primary after failing to update leader lock in DCS'
                     self.demote('immediate-nolock')
@@ -1726,7 +1734,7 @@ class Ha(object):
             self.cancel_initialization()
 
         if result is None:
-            if not self.state_handler.is_leader():
+            if not self.state_handler.is_primary():
                 return 'waiting for end of recovery after bootstrap'
 
             self.state_handler.set_role('master')
@@ -1910,7 +1918,7 @@ class Ha(object):
                 elif self.cluster.is_unlocked() and not self.is_paused():
                     # "bootstrap", but data directory is not empty
                     if not self.state_handler.cb_called and self.state_handler.is_running() \
-                            and not self.state_handler.is_leader():
+                            and not self.state_handler.is_primary():
                         self._join_aborted = True
                         logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start')
                         logger.error('Please first start Patroni on the node running as primary')
@@ -1953,7 +1961,7 @@ class Ha(object):
                 create_slots = self._sync_replication_slots(False)
 
                 if not self.state_handler.cb_called:
-                    if not is_promoting and not self.state_handler.is_leader():
+                    if not is_promoting and not self.state_handler.is_primary():
                         self._rewind.trigger_check_diverged_lsn()
                     self.state_handler.call_nowait(CallbackAction.ON_START)
 
@@ -1978,7 +1986,7 @@ class Ha(object):
 
     def _handle_dcs_error(self) -> str:
         if not self.is_paused() and self.state_handler.is_running():
-            if self.state_handler.is_leader():
+            if self.state_handler.is_primary():
                 if self.is_failsafe_mode() and self.check_failsafe_topology():
                     self.set_is_leader(True)
                     self._failsafe.set_is_active(time.time())
@@ -2053,7 +2061,7 @@ class Ha(object):
                     # location, we can remove the leader key and allow them to start leader race.
 
                     if self.is_failover_possible(cluster_lsn=checkpoint_location):
-                        self.dcs.delete_leader(checkpoint_location)
+                        self.dcs.delete_leader(self.cluster.leader, checkpoint_location)
                         status['deleted'] = True
                     else:
                         self.dcs.write_leader_optime(checkpoint_location)
@@ -2070,7 +2078,7 @@ class Ha(object):
             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(checkpoint_location)
+                    self.dcs.delete_leader(self.cluster.leader, checkpoint_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

@@ -120,9 +120,9 @@ class Postgresql(object):
 
         if self.is_running():  # we are "joining" already running postgres
             self.set_state('running')
-            self.set_role('master' if self.is_leader() else 'replica')
+            self.set_role('master' if self.is_primary() else 'replica')
             # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known
-            if self.major_version < 120000 or self.is_leader():
+            if self.major_version < 120000 or self.is_primary():
                 self.config.write_postgresql_conf()
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
@@ -479,14 +479,14 @@ class Postgresql(object):
         """:returns: a result set of 'SELECT * FROM pg_stat_replication'."""
         return self._cluster_info_state_get('pg_stat_replication') or []
 
-    def replication_state_from_parameters(self, is_leader: bool, receiver_state: Optional[str],
+    def replication_state_from_parameters(self, is_primary: bool, receiver_state: Optional[str],
                                           restore_command: Optional[str]) -> Optional[str]:
         """Figure out the replication state from input parameters.
 
         .. note::
             This method could be only called when Postgres is up, running and queries are successfuly executed.
 
-        :is_leader: `True` is postgres is not running in recovery
+        :is_primary: `True` is postgres is not running in recovery
         :receiver_state: value from `pg_stat_get_wal_receiver.state` or None if Postgres is older than 9.6
         :restore_command: value of ``restore_command`` GUC for PostgreSQL 12+ or
                           `postgresql.recovery_conf.restore_command` if it is set in Patroni configuration
@@ -495,7 +495,7 @@ class Postgresql(object):
                   - 'streaming' if replica is streaming according to the `pg_stat_wal_receiver` view;
                   - 'in archive recovery' if replica isn't streaming and there is a `restore_command`
         """
-        if self._major_version >= 90600 and not is_leader:
+        if self._major_version >= 90600 and not is_primary:
             if receiver_state == 'streaming':
                 return 'streaming'
             # For Postgres older than 12 we get `restore_command` from Patroni config, otherwise we check GUC
@@ -510,11 +510,11 @@ class Postgresql(object):
 
         :returns: ``streaming``, ``in archive recovery``, or ``None``
         """
-        return self.replication_state_from_parameters(self.is_leader(),
+        return self.replication_state_from_parameters(self.is_primary(),
                                                       self._cluster_info_state_get('receiver_state'),
                                                       self._cluster_info_state_get('restore_command'))
 
-    def is_leader(self) -> bool:
+    def is_primary(self) -> bool:
         try:
             return bool(self._cluster_info_state_get('timeline'))
         except PostgresConnectionException:
@@ -1162,9 +1162,9 @@ class Postgresql(object):
         return ret
 
     @staticmethod
-    def _wal_position(is_leader: bool, wal_position: int,
+    def _wal_position(is_primary: bool, wal_position: int,
                       received_location: Optional[int], replayed_location: Optional[int]) -> int:
-        return wal_position if is_leader else max(received_location or 0, replayed_location or 0)
+        return wal_position if is_primary else max(received_location or 0, replayed_location or 0)
 
     def timeline_wal_position(self) -> Tuple[int, int, Optional[int]]:
         # This method could be called from different threads (simultaneously with some other `_query` calls).
@@ -1200,7 +1200,7 @@ class Postgresql(object):
             return None
 
     def last_operation(self) -> int:
-        return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position') or 0,
+        return self._wal_position(self.is_primary(), self._cluster_info_state_get('wal_position') or 0,
                                   self.received_location(), self.replayed_location())
 
     def configure_server_parameters(self) -> None:

patroni/postgresql/citus.py

@@ -412,7 +412,7 @@ class CitusHandler(Thread):
         parameters['wal_level'] = 'logical'
 
     def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
-        if isinstance(self._config, dict) and self._postgresql.is_leader() and\
+        if isinstance(self._config, dict) and self._postgresql.is_primary() and\
                 slot['type'] == 'logical' and slot['database'] == self._config['database']:
             m = CITUS_SLOT_NAME_RE.match(slot['name'])
             return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin'])

patroni/postgresql/rewind.py

@@ -280,7 +280,7 @@ class Rewind(object):
         """After promote issue a CHECKPOINT from a new thread and asynchronously check the result.
         In case if CHECKPOINT failed, just check that timeline in pg_control was updated."""
 
-        if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_leader():
+        if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_primary():
             with self._checkpoint_task_lock:
                 if self._checkpoint_task:
                     with self._checkpoint_task:

patroni/postgresql/slots.py

@@ -508,7 +508,7 @@ class SlotsHandler:
 
                 self._ensure_physical_slots(slots)
 
-                if self._postgresql.is_leader():
+                if self._postgresql.is_primary():
                     self._logical_slots_processing_queue.clear()
                     self._ensure_logical_slots_primary(slots)
                 elif cluster.slots and slots:

patroni/postgresql/sync.py

@@ -408,7 +408,7 @@ END;$$""")
             sync_param = f'{prefix}{num} ({sync_param})'
 
         if not (self._postgresql.config.set_synchronous_standby_names(sync_param)
-                and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk:
+                and self._postgresql.state == 'running' and self._postgresql.is_primary()) or has_asterisk:
             return
 
         time.sleep(0.1)  # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms

patroni/utils.py

@@ -16,6 +16,7 @@ import platform
 import random
 import re
 import socket
+import subprocess
 import sys
 import tempfile
 import time
@@ -467,6 +468,18 @@ def _sleep(interval: Union[int, float]) -> None:
     time.sleep(interval)
 
 
+def read_stripped(file_path: str) -> Iterator[str]:
+    """Iterate over stripped lines in the given file.
+
+    :param file_path: path to the file to read from
+
+    :yields: each line from the given file stripped
+    """
+    with open(file_path) as f:
+        for line in f:
+            yield line.strip()
+
+
 class RetryFailedError(PatroniException):
     """Maximum number of attempts exhausted in retry operation."""
 
@@ -978,3 +991,34 @@ def unquote(string: str) -> str:
     except ValueError:
         ret = string
     return ret
+
+
+def get_major_version(bin_dir: Optional[str] = None, bin_name: str = 'postgres') -> str:
+    """Get the major version of PostgreSQL.
+
+    It is based on the output of ``postgres --version``.
+
+    :param bin_dir: path to the PostgreSQL binaries directory. If ``None`` or an empty string, it will use the first
+    *bin_name* binary that is found by the subprocess in the ``PATH``.
+    :param bin_name: name of the postgres binary to call (``postgres`` by default)
+    :returns: the PostgreSQL major version.
+    :raises:
+        :exc:`~patroni.exceptions.PatroniException`: if the postgres binary call failed due to :exc:`OSError`.
+
+    :Example:
+
+        * Returns `9.6` for PostgreSQL 9.6.24
+        * Returns `15` for PostgreSQL 15.2
+    """
+    if not bin_dir:
+        binary = bin_name
+    else:
+        binary = os.path.join(bin_dir, bin_name)
+    try:
+        version = subprocess.check_output([binary, '--version']).decode()
+    except OSError as e:
+        raise PatroniException(f'Failed to get postgres version: {e}')
+    version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version)
+    if TYPE_CHECKING:  # pragma: no cover
+        assert version is not None
+    return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)

patroni/validator.py

@@ -6,17 +6,16 @@ This module contains facilities for validating configuration of Patroni processe
 :var schema: configuration schema of the daemon launched by `patroni` command.
 """
 import os
-import re
 import shutil
 import socket
-import subprocess
 
-from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple
 
 from .collections import CaseInsensitiveSet
+
 from .dcs import dcs_modules
 from .exceptions import ConfigParseError
-from .utils import parse_int, split_host_port, data_directory_is_empty
+from .utils import parse_int, split_host_port, data_directory_is_empty, get_major_version
 
 
 def data_directory_empty(data_dir: str) -> bool:
@@ -187,31 +186,6 @@ def get_bin_name(bin_name: str) -> str:
     return (schema.data.get('postgresql', {}).get('bin_name', {}) or {}).get(bin_name, bin_name)
 
 
-def get_major_version(bin_dir: OptionalType[str] = None) -> str:
-    """Get the major version of PostgreSQL.
-
-    It is based on the output of ``postgres --version``.
-
-    :param bin_dir: path to PostgreSQL binaries directory. If ``None`` it will use the first ``postgres`` binary that
-        is found by subprocess in the ``PATH``.
-    :returns: the PostgreSQL major version.
-
-    :Example:
-
-        * Returns `9.6` for PostgreSQL 9.6.24
-        * Returns `15` for PostgreSQL 15.2
-    """
-    if not bin_dir:
-        binary = get_bin_name('postgres')
-    else:
-        binary = os.path.join(bin_dir, get_bin_name('postgres'))
-    version = subprocess.check_output([binary, '--version']).decode()
-    version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version)
-    if TYPE_CHECKING:  # pragma: no cover
-        assert version is not None
-    return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)
-
-
 def validate_data_dir(data_dir: str) -> bool:
     """Validate the value of ``postgresql.data_dir`` configuration option.
 
@@ -246,7 +220,7 @@ def validate_data_dir(data_dir: str) -> bool:
                 raise ConfigParseError("data dir for the cluster is not empty, but doesn't contain"
                                        " \"{}\" directory".format(waldir))
             bin_dir = schema.data.get("postgresql", {}).get("bin_dir", None)
-            major_version = get_major_version(bin_dir)
+            major_version = get_major_version(bin_dir, get_bin_name('postgres'))
             if pgversion != major_version:
                 raise ConfigParseError("data_dir directory postgresql version ({}) doesn't match with "
                                        "'postgres --version' output ({})".format(pgversion, major_version))

tests/__init__.py

@@ -118,6 +118,21 @@ class MockCursor(object):
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
             now = datetime.datetime.now(tzutc)
             self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)]
+        elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'):
+            self.results = [('data_directory', 'data'),
+                            ('hba_file', os.path.join('data', 'pg_hba.conf')),
+                            ('ident_file', os.path.join('data', 'pg_ident.conf')),
+                            ('max_connections', 42),
+                            ('max_locks_per_transaction', 73),
+                            ('max_prepared_transactions', 0),
+                            ('max_replication_slots', 21),
+                            ('max_wal_senders', 37),
+                            ('track_commit_timestamp', 'off'),
+                            ('wal_level', 'replica'),
+                            ('listen_addresses', '6.6.6.6'),
+                            ('port', 1984),
+                            ('archive_command', 'my archive command'),
+                            ('cluster_name', 'my_cluster')]
         elif sql.startswith('SELECT name, setting'):
             self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
                             ('wal_block_size', '8192', None, 'integer', 'internal'),
@@ -159,11 +174,20 @@ class MockCursor(object):
         pass
 
 
+class MockConnectionInfo(object):
+
+    def parameter_status(self, param_name):
+        if param_name == 'is_superuser':
+            return 'on'
+        return '0'
+
+
 class MockConnect(object):
 
     server_version = 99999
     autocommit = False
     closed = 0
+    info = MockConnectionInfo()
 
     def cursor(self):
         return MockCursor(self)

tests/test_config_generator.py

@@ -0,0 +1,334 @@
+import os
+import psutil
+import socket
+import unittest
+
+from . import MockConnect, MockCursor, MockConnectionInfo
+from copy import deepcopy
+from mock import MagicMock, Mock, PropertyMock, mock_open, patch
+
+from patroni.__main__ import main as _main
+from patroni.config import Config
+from patroni.config_generator import AbstractConfigGenerator, get_address
+
+from patroni.utils import patch_config
+
+from . import psycopg_connect
+
+
+@patch('patroni.psycopg.connect', psycopg_connect)
+@patch('socket.getaddrinfo', Mock(return_value=[(0, 0, 0, 0, ('1.9.8.4', 1984))]))
+@patch('builtins.open', MagicMock())
+@patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 16.2"))
+@patch('psutil.Process.exe', Mock(return_value='/bin/dir/from/running/postgres'))
+@patch('psutil.Process.__init__', Mock(return_value=None))
+class TestGenerateConfig(unittest.TestCase):
+
+    no_value_msg = '#FIXME'
+    _HOSTNAME = socket.gethostname()
+    _IP = sorted(socket.getaddrinfo(_HOSTNAME, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0), key=lambda x: x[0])[0][4][0]
+
+    def setUp(self):
+        self.maxDiff = None
+
+        os.environ['PATRONI_SCOPE'] = 'scope_from_env'
+        os.environ['PATRONI_POSTGRESQL_BIN_DIR'] = '/bin/from/env'
+        os.environ['PATRONI_SUPERUSER_USERNAME'] = 'su_user_from_env'
+        os.environ['PATRONI_SUPERUSER_PASSWORD'] = 'su_pwd_from_env'
+        os.environ['PATRONI_REPLICATION_USERNAME'] = 'repl_user_from_env'
+        os.environ['PATRONI_REPLICATION_PASSWORD'] = 'repl_pwd_from_env'
+        os.environ['PATRONI_REWIND_USERNAME'] = 'rewind_user_from_env'
+        os.environ['PGUSER'] = 'pguser_from_env'
+        os.environ['PGPASSWORD'] = 'pguser_pwd_from_env'
+        os.environ['PATRONI_RESTAPI_CONNECT_ADDRESS'] = 'localhost:8080'
+        os.environ['PATRONI_RESTAPI_LISTEN'] = 'localhost:8080'
+        os.environ['PATRONI_POSTGRESQL_BIN_POSTGRES'] = 'custom_postgres_bin_from_env'
+
+        self.environ = deepcopy(os.environ)
+
+        dynamic_config = Config.get_default_config()
+        dynamic_config['postgresql']['parameters'] = dict(dynamic_config['postgresql']['parameters'])
+        del dynamic_config['standby_cluster']
+        dynamic_config['postgresql']['parameters']['wal_keep_segments'] = 8
+        dynamic_config['postgresql']['use_pg_rewind'] = True
+
+        self.config = {
+            'scope': self.environ['PATRONI_SCOPE'],
+            'name': self._HOSTNAME,
+            'bootstrap': {
+                'dcs': dynamic_config
+            },
+            'postgresql': {
+                'connect_address': self.no_value_msg + ':5432',
+                'data_dir': self.no_value_msg,
+                'listen': self.no_value_msg + ':5432',
+                'pg_hba': ['host all all all md5',
+                           f'host replication {self.environ["PATRONI_REPLICATION_USERNAME"]} all md5'],
+                'authentication': {'superuser': {'username': self.environ['PATRONI_SUPERUSER_USERNAME'],
+                                                 'password': self.environ['PATRONI_SUPERUSER_PASSWORD']},
+                                   'replication': {'username': self.environ['PATRONI_REPLICATION_USERNAME'],
+                                                   'password': self.environ['PATRONI_REPLICATION_PASSWORD']},
+                                   'rewind': {'username': self.environ['PATRONI_REWIND_USERNAME']}},
+                'bin_dir': self.environ['PATRONI_POSTGRESQL_BIN_DIR'],
+                'bin_name': {'postgres': self.environ['PATRONI_POSTGRESQL_BIN_POSTGRES']},
+                'parameters': {'password_encryption': 'md5'}
+            },
+            'restapi': {
+                'connect_address': self.environ['PATRONI_RESTAPI_CONNECT_ADDRESS'],
+                'listen': self.environ['PATRONI_RESTAPI_LISTEN']
+            }
+        }
+
+    def _set_running_instance_config_vals(self):
+        # values are taken from tests/__init__.py
+        conf = {
+            'scope': 'my_cluster',
+            'bootstrap': {
+                'dcs': {
+                    'postgresql': {
+                        'parameters': {
+                            'max_connections': 42,
+                            'max_locks_per_transaction': 73,
+                            'max_replication_slots': 21,
+                            'max_wal_senders': 37,
+                            'wal_level': 'replica',
+                            'wal_keep_segments': None
+                        },
+                        'use_pg_rewind': None
+                    }
+                }
+            },
+            'postgresql': {
+                'connect_address': f'{self._IP}:bar',
+                'listen': '6.6.6.6:1984',
+                'data_dir': 'data',
+                'bin_dir': '/bin/dir/from/running',
+                'parameters': {
+                    'archive_command': 'my archive command',
+                    'hba_file': os.path.join('data', 'pg_hba.conf'),
+                    'ident_file': os.path.join('data', 'pg_ident.conf'),
+                    'password_encryption': None
+                },
+                'authentication': {
+                    'superuser': {
+                        'username': 'foobar',
+                        'password': 'qwerty',
+                        'channel_binding': 'prefer',
+                        'gssencmode': 'prefer',
+                        'sslmode': 'prefer'
+                    },
+                    'replication': {
+                        'username': self.no_value_msg,
+                        'password': self.no_value_msg
+                    },
+                    'rewind': None
+                },
+            }
+        }
+        patch_config(self.config, conf)
+
+    def _get_running_instance_open_res(self):
+        hba_content = '\n'.join(self.config['postgresql']['pg_hba'] + ['#host all all all md5',
+                                                                       '  host all all all md5',
+                                                                       '',
+                                                                       'hostall all all md5'])
+        ident_content = '\n'.join(['# something very interesting', '  '])
+
+        self.config['postgresql']['pg_hba'] += ['host all all all md5']
+        return [
+            mock_open(read_data=hba_content)(),
+            mock_open(read_data=ident_content)(),
+            mock_open(read_data='1984')(),
+            mock_open()()
+        ]
+
+    @patch('os.makedirs')
+    @patch('yaml.safe_dump')
+    def test_generate_sample_config_pre_13_dir_creation(self, mock_config_dump, mock_makedir):
+        with patch('sys.argv', ['patroni.py', '--generate-sample-config', '/foo/bar.yml']), \
+             patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 9.4.3")) as pg_bin_mock, \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertEqual(e.exception.code, 0)
+        self.assertEqual(self.config, mock_config_dump.call_args[0][0])
+        mock_makedir.assert_called_once()
+        pg_bin_mock.assert_called_once_with([os.path.join(self.environ['PATRONI_POSTGRESQL_BIN_DIR'],
+                                                          self.environ['PATRONI_POSTGRESQL_BIN_POSTGRES']),
+                                             '--version'])
+
+    @patch('os.makedirs', Mock())
+    @patch('yaml.safe_dump')
+    def test_generate_sample_config_16(self, mock_config_dump):
+        conf = {
+            'bootstrap': {
+                'dcs': {
+                    'postgresql': {
+                        'parameters': {
+                            'wal_keep_size': '128MB',
+                            'wal_keep_segments': None
+                        },
+                    }
+                }
+            },
+            'postgresql': {
+                'parameters': {
+                    'password_encryption': 'scram-sha-256'
+                },
+                'pg_hba': ['host all all all scram-sha-256',
+                           f'host replication {self.environ["PATRONI_REPLICATION_USERNAME"]} all scram-sha-256'],
+                'authentication': {
+                    'rewind': {
+                        'username': self.environ['PATRONI_REWIND_USERNAME'],
+                        'password': self.no_value_msg}
+                },
+            }
+        }
+        patch_config(self.config, conf)
+
+        with patch('sys.argv', ['patroni.py', '--generate-sample-config', '/foo/bar.yml']), \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertEqual(e.exception.code, 0)
+        self.assertEqual(self.config, mock_config_dump.call_args[0][0])
+
+    @patch('os.makedirs', Mock())
+    @patch('yaml.safe_dump')
+    def test_generate_config_running_instance_16(self, mock_config_dump):
+        self._set_running_instance_config_vals()
+
+        with patch('builtins.open', Mock(side_effect=self._get_running_instance_open_res())), \
+             patch('sys.argv', ['patroni.py', '--generate-config',
+                                '--dsn', 'host=foo port=bar user=foobar password=qwerty']), \
+                self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertEqual(e.exception.code, 0)
+        self.assertEqual(self.config, mock_config_dump.call_args[0][0])
+
+    @patch('os.makedirs', Mock())
+    @patch('yaml.safe_dump')
+    def test_generate_config_running_instance_16_connect_from_env(self, mock_config_dump):
+        self._set_running_instance_config_vals()
+        # su auth params and connect host from env
+        os.environ['PGCHANNELBINDING'] = \
+            self.config['postgresql']['authentication']['superuser']['channel_binding'] = 'disable'
+
+        conf = {
+            'scope': 'my_cluster',
+            'bootstrap': {
+                'dcs': {
+                    'postgresql': {
+                        'parameters': {
+                            'max_connections': 42,
+                            'max_locks_per_transaction': 73,
+                            'max_replication_slots': 21,
+                            'max_wal_senders': 37,
+                            'wal_level': 'replica',
+                            'wal_keep_segments': None
+                        },
+                        'use_pg_rewind': None
+                    }
+                }
+            },
+            'postgresql': {
+                'connect_address': f'{self._IP}:1984',
+                'authentication': {
+                    'superuser': {
+                        'username': self.environ['PGUSER'],
+                        'password': self.environ['PGPASSWORD'],
+                        'gssencmode': None,
+                        'sslmode': None
+                    },
+                },
+            }
+        }
+        patch_config(self.config, conf)
+
+        with patch('builtins.open', Mock(side_effect=self._get_running_instance_open_res())), \
+             patch('sys.argv', ['patroni.py', '--generate-config']), \
+             patch.object(MockConnect, 'server_version', PropertyMock(return_value=160000)), \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertEqual(e.exception.code, 0)
+        self.assertEqual(self.config, mock_config_dump.call_args[0][0])
+
+    def test_generate_config_running_instance_errors(self):
+        # 1. Wrong DSN format
+        with patch('sys.argv', ['patroni.py', '--generate-config', '--dsn', 'host:foo port:bar user:foobar']), \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertIn('Failed to parse DSN string', e.exception.code)
+
+        # 2. User is not a superuser
+        with patch('sys.argv', ['patroni.py',
+                                '--generate-config', '--dsn', 'host=foo port=bar user=foobar password=pwd_from_dsn']), \
+             patch.object(MockCursor, 'rowcount', PropertyMock(return_value=0), create=True), \
+             patch.object(MockConnectionInfo, 'parameter_status', Mock(return_value='off')), \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertIn('The provided user does not have superuser privilege', e.exception.code)
+
+        # 3. Error while calling postgres --version
+        with patch('subprocess.check_output', Mock(side_effect=OSError)), \
+             patch('sys.argv', ['patroni.py', '--generate-sample-config']), \
+             self.assertRaises(SystemExit) as e:
+            _main()
+        self.assertIn('Failed to get postgres version:', e.exception.code)
+
+        with patch('sys.argv', ['patroni.py', '--generate-config']):
+
+            # 4. empty postmaster.pid
+            with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(),
+                                                          mock_open(read_data='ident_content')(),
+                                                          mock_open(read_data='')()])), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Failed to obtain postmaster pid from postmaster.pid file', e.exception.code)
+
+            # 5. Failed to open postmaster.pid
+            with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(),
+                                                          mock_open(read_data='ident_content')(),
+                                                          OSError])), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Error while reading postmaster.pid file', e.exception.code)
+
+            # 6. Invalid postmaster pid
+            with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(),
+                                                          mock_open(read_data='ident_content')(),
+                                                          mock_open(read_data='1984')()])), \
+                 patch('psutil.Process.__init__', Mock(return_value=None)), \
+                 patch('psutil.Process.exe', Mock(side_effect=psutil.NoSuchProcess(1984))), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn("Obtained postmaster pid doesn't exist", e.exception.code)
+
+            # 7. Failed to open pg_hba
+            with patch('builtins.open', Mock(side_effect=OSError)), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Failed to read pg_hba.conf', e.exception.code)
+
+            # 8. Failed to open pg_ident
+            with patch('builtins.open', Mock(side_effect=[mock_open(read_data='hba_content')(), OSError])), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Failed to read pg_ident.conf', e.exception.code)
+
+            # 9. Failed PG connecttion
+            from . import psycopg
+            with patch('patroni.psycopg.connect', side_effect=psycopg.Error), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Failed to establish PostgreSQL connection', e.exception.code)
+
+            # 10. An unexpected error
+            with patch.object(AbstractConfigGenerator, '__init__', side_effect=psycopg.Error), \
+                 self.assertRaises(SystemExit) as e:
+                _main()
+            self.assertIn('Unexpected exception', e.exception.code)
+
+    def test_get_address(self):
+        with patch('socket.getaddrinfo', Mock(side_effect=Exception)), \
+             patch('logging.warning') as mock_warning:
+            self.assertEqual(get_address(), (self.no_value_msg, self.no_value_msg))
+            self.assertIn('Failed to obtain address: %r', mock_warning.call_args_list[0][0])

tests/test_consul.py

@@ -197,9 +197,10 @@ class TestConsul(unittest.TestCase):
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):
-        self.c.delete_leader()
+        leader = self.c.get_cluster().leader
+        self.c.delete_leader(leader)
         self.c._name = 'other'
-        self.c.delete_leader()
+        self.c.delete_leader(leader)
 
     @patch.object(consul.Consul.KV, 'put', Mock(return_value=True))
     def test_initialize(self):

tests/test_etcd.py

@@ -313,7 +313,7 @@ class TestEtcd(unittest.TestCase):
         self.assertFalse(self.etcd.cancel_initialization())
 
     def test_delete_leader(self):
-        self.assertFalse(self.etcd.delete_leader())
+        self.assertFalse(self.etcd.delete_leader(self.etcd.get_cluster().leader))
 
     def test_delete_cluster(self):
         self.assertFalse(self.etcd.delete_cluster())

tests/test_etcd3.py

@@ -298,9 +298,10 @@ class TestEtcd3(BaseTestEtcd3):
         self.etcd3.cancel_initialization()
 
     def test_delete_leader(self):
-        self.etcd3.delete_leader()
+        leader = self.etcd3.get_cluster().leader
+        self.etcd3.delete_leader(leader)
         self.etcd3._name = 'other'
-        self.etcd3.delete_leader()
+        self.etcd3.delete_leader(leader)
 
     def test_delete_cluster(self):
         self.etcd3.delete_cluster()
@@ -312,7 +313,7 @@ class TestEtcd3(BaseTestEtcd3):
         self.etcd3.set_sync_state_value('', 1)
 
     def test_delete_sync_state(self):
-        self.etcd3.delete_sync_state()
+        self.etcd3.delete_sync_state('1')
 
     def test_watch(self):
         self.etcd3.set_ttl(10)

tests/test_ha.py

@@ -163,7 +163,7 @@ def run_async(self, func, args=()):
 
 
 @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
-@patch.object(Postgresql, 'is_leader', Mock(return_value=True))
+@patch.object(Postgresql, 'is_primary', Mock(return_value=True))
 @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1)))
 @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
@@ -226,7 +226,7 @@ class TestHa(PostgresInit):
     @patch.object(Postgresql, 'received_timeline', Mock(return_value=None))
     def test_touch_member(self):
         self.p._major_version = 110000
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 0))
         self.p.replica_cached_timeline = Mock(side_effect=Exception)
         with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')):
@@ -322,7 +322,7 @@ class TestHa(PostgresInit):
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
     def test_crash_recovery_before_rewind(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.is_running = false
         self.p.controldata = lambda: {'Database cluster state': 'in archive recovery',
                                       'Database system identifier': SYSID}
@@ -367,7 +367,7 @@ class TestHa(PostgresInit):
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_start_as_readonly(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.is_healthy = true
         self.ha.has_lock = true
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
@@ -385,11 +385,11 @@ class TestHa(PostgresInit):
 
     def test_promoted_by_acquiring_lock(self):
         self.ha.is_healthiest_node = true
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
     def test_promotion_cancelled_after_pre_promote_failed(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p._pre_promote = false
         self.ha._is_healthiest_node = true
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
@@ -404,7 +404,7 @@ class TestHa(PostgresInit):
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_long_promote(self):
         self.ha.has_lock = true
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('primary')
         self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
@@ -415,7 +415,7 @@ class TestHa(PostgresInit):
     def test_follow_new_leader_after_failing_to_obtain_lock(self):
         self.ha.is_healthiest_node = true
         self.ha.acquire_lock = false
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock')
 
     def test_demote_because_not_healthiest(self):
@@ -424,21 +424,20 @@ class TestHa(PostgresInit):
 
     def test_follow_new_leader_because_not_healthiest(self):
         self.ha.is_healthiest_node = false
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_promote_because_have_lock(self):
         self.ha.has_lock = true
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because I had the session lock')
 
     def test_promote_without_watchdog(self):
         self.ha.has_lock = true
-        self.p.is_leader = true
         with patch.object(Watchdog, 'activate', Mock(return_value=False)):
             self.assertEqual(self.ha.run_cycle(), 'Demoting self because watchdog could not be activated')
-            self.p.is_leader = false
+            self.p.is_primary = false
             self.assertEqual(self.ha.run_cycle(), 'Not promoting self because watchdog could not be activated')
 
     def test_leader_with_lock(self):
@@ -464,12 +463,12 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS')
         with patch.object(Ha, '_get_node_to_follow', Mock(side_effect=DCSError('foo'))):
             self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS')
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_follow(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
         self.ha.patroni.replicatefrom = "foo"
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
@@ -486,13 +485,13 @@ class TestHa(PostgresInit):
     def test_follow_in_pause(self):
         self.ha.is_paused = true
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock')
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0)')
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
     def test_follow_triggers_rewind(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.ha._rewind.trigger_check_diverged_lsn()
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
@@ -546,7 +545,7 @@ class TestHa(PostgresInit):
         self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster)
         self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni',
                                  'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}})
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
 
     def test_no_dcs_connection_replica_failsafe_not_enabled_but_active(self):
@@ -554,7 +553,7 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni',
                                  'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}})
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
 
     def test_update_failsafe(self):
@@ -593,9 +592,9 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
         self.e.initialize = true
         self.assertEqual(self.ha.bootstrap(), 'trying to bootstrap a new cluster')
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap')
-        self.p.is_leader = true
+        self.p.is_primary = true
         self.ha.is_synchronous_mode = true
         self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap')
         self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster')
@@ -615,7 +614,6 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
         self.e.initialize = true
         self.ha.bootstrap()
-        self.p.is_leader = true
         with patch.object(Watchdog, 'activate', Mock(return_value=False)), \
                 patch('patroni.ha.logger.error') as mock_logger:
             self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
@@ -747,7 +745,6 @@ class TestHa(PostgresInit):
         self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
     def test_manual_failover_from_leader_in_synchronous_mode(self):
-        self.p.is_leader = true
         self.ha.has_lock = true
         self.ha.is_synchronous_mode = true
         self.ha.process_sync_replication = Mock()
@@ -758,7 +755,7 @@ class TestHa(PostgresInit):
         self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
 
     def test_manual_failover_process_no_leader(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None))
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
         self.p.set_role('replica')
@@ -782,7 +779,7 @@ class TestHa(PostgresInit):
 
     def test_manual_failover_process_no_leader_in_synchronous_mode(self):
         self.ha.is_synchronous_mode = true
-        self.p.is_leader = false
+        self.p.is_primary = false
 
         # switchover to a specific node, which name doesn't match our name (postgresql0)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'other', None))
@@ -842,14 +839,14 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock')
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None))
         self.assertEqual('PAUSE: acquired session lock as a leader', self.ha.run_cycle())
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('replica')
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None))
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock')
 
     def test_is_healthiest_node(self):
         self.ha.is_failsafe_mode = true
-        self.ha.state_handler.is_leader = false
+        self.p.is_primary = false
         self.ha.patroni.nofailover = False
         self.ha.fetch_node_status = get_node_status()
         self.ha.dcs._last_failsafe = {'foo': ''}
@@ -863,7 +860,7 @@ class TestHa(PostgresInit):
         self.assertFalse(self.ha.is_healthiest_node())
 
     def test__is_healthiest_node(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('postgresql1', self.p.name))
         self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster)
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -962,7 +959,7 @@ class TestHa(PostgresInit):
         self.assertTrue(self.ha.restart_matches("replica", "9.5.2", False))
 
     def test_process_healthy_cluster_in_pause(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.ha.is_paused = true
         self.p.name = 'leader'
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -973,7 +970,7 @@ class TestHa(PostgresInit):
     @patch('patroni.postgresql.mtime', Mock(return_value=1588316884))
     @patch('builtins.open', mock_open(read_data='1\t0/40159C0\tno recovery target specified\n'))
     def test_process_healthy_standby_cluster_as_standby_leader(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'leader'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.p.config.check_recovery_conf = Mock(return_value=(False, False))
@@ -985,7 +982,7 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
 
     def test_process_healthy_standby_cluster_as_cascade_replica(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'replica'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.assertEqual(self.ha.run_cycle(),
@@ -995,7 +992,7 @@ class TestHa(PostgresInit):
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=True))
     def test_process_unhealthy_standby_cluster_as_standby_leader(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'leader'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.ha.sysid_valid = true
@@ -1005,13 +1002,13 @@ class TestHa(PostgresInit):
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
     def test_process_unhealthy_standby_cluster_as_cascade_replica(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'replica'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_member:'))
 
     def test_recover_unhealthy_leader_in_standby_cluster(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'leader'
         self.p.is_running = false
         self.p.follow = false
@@ -1020,7 +1017,7 @@ class TestHa(PostgresInit):
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=True))
     def test_recover_unhealthy_unlocked_standby_cluster(self):
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.name = 'leader'
         self.p.is_running = false
         self.p.follow = false
@@ -1080,7 +1077,7 @@ class TestHa(PostgresInit):
         check_calls([(update_lock, True), (demote, True)])
 
         self.ha.has_lock = false
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(),
                          'no action. I am (postgresql0), a secondary, and following a leader (leader)')
         check_calls([(update_lock, False), (demote, False)])
@@ -1219,7 +1216,7 @@ class TestHa(PostgresInit):
         self.ha.is_synchronous_mode = true
 
         mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('replica')
         self.ha.has_lock = true
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
@@ -1242,7 +1239,7 @@ class TestHa(PostgresInit):
     def test_unhealthy_sync_mode(self):
         self.ha.is_synchronous_mode = true
 
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('replica')
         self.p.name = 'other'
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'other2'))
@@ -1273,7 +1270,7 @@ class TestHa(PostgresInit):
         self.ha.is_synchronous_mode = true
 
         self.p.name = 'other'
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('replica')
         mock_restart = self.p.restart = Mock(return_value=True)
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
@@ -1376,7 +1373,7 @@ class TestHa(PostgresInit):
     @patch('sys.exit', return_value=1)
     def test_abort_join(self, exit_mock):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.ha.run_cycle()
         exit_mock.assert_called_once_with(1)
 
@@ -1436,7 +1433,7 @@ class TestHa(PostgresInit):
     @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['ls']))
     def test_follow_copy(self):
         self.ha.cluster.config.data['slots'] = {'ls': {'database': 'a', 'plugin': 'b'}}
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))
 
     def test_acquire_lock(self):
@@ -1462,7 +1459,7 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('other', self.p.name + ',foo'))
         self.ha.cluster.config.data.update({'synchronous_mode': 'quorum'})
         self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster)
-        self.p.is_leader = false
+        self.p.is_primary = false
         self.p.set_role('replica')
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=None)
         # Postgres 9.5, write_sync_state to DCS failed

tests/test_kubernetes.py

@@ -324,7 +324,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         self.k.initialize()
 
     def test_delete_leader(self):
-        self.k.delete_leader(1)
+        self.k.delete_leader(self.k.get_cluster().leader, 1)
 
     def test_cancel_initialization(self):
         self.k.cancel_initialization()

tests/test_postgresql.py

@@ -363,11 +363,11 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertRaises(psycopg.ProgrammingError, self.p.query, 'blabla')
 
     @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT))
-    def test_is_leader(self):
-        self.assertTrue(self.p.is_leader())
+    def test_is_primary(self):
+        self.assertTrue(self.p.is_primary())
         self.p.reset_cluster_info_state(None)
         with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
-            self.assertFalse(self.p.is_leader())
+            self.assertFalse(self.p.is_primary())
 
     @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down',
                                                                 'Latest checkpoint location': '0/1ADBC18',
@@ -461,7 +461,7 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START))
 
     @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
-    def test_is_leader_exception(self):
+    def test_is_primary_exception(self):
         self.p.start()
         self.p.query = Mock(side_effect=psycopg.OperationalError("not supported"))
         self.assertTrue(self.p.stop())

tests/test_raft.py

@@ -142,25 +142,25 @@ class TestRaft(unittest.TestCase):
         raft._citus_group = '1'
         self.assertTrue(raft.manual_failover('foo', 'bar'))
         raft._citus_group = '0'
+        self.assertTrue(raft.take_leader())
         cluster = raft.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
+        leader = cluster.leader
+        self.assertTrue(raft.delete_leader(leader))
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
-        leader = raft.get_cluster().leader
+        raft.get_cluster()
         self.assertTrue(raft.update_leader(leader, '1', failsafe={'foo': 'bat'}))
         self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}'))
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
         raft.get_citus_coordinator()
         self.assertTrue(raft.delete_sync_state())
-        self.assertTrue(raft.delete_leader())
         self.assertTrue(raft.set_history_value(''))
         self.assertTrue(raft.delete_cluster())
         raft._citus_group = '1'
         self.assertTrue(raft.delete_cluster())
         raft._citus_group = None
         raft.get_cluster()
-        self.assertTrue(raft.take_leader())
-        raft.get_cluster()
         raft.watch(None, 0.001)
         raft._sync_obj.destroy()
 

tests/test_slots.py

@@ -48,7 +48,7 @@ class TestSlotsHandler(BaseTestPostgresql):
             self.s.sync_replication_slots(cluster, False)
             mock_debug.assert_called_once()
         self.p.set_role('replica')
-        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \
+        with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
             self.s.sync_replication_slots(cluster, False, paused=True)
             mock_drop.assert_not_called()
@@ -79,7 +79,7 @@ class TestSlotsHandler(BaseTestPostgresql):
                           None, SyncState.empty(), None, {'ls': 10}, None)
         self.p.set_role('replica')
         with patch.object(Postgresql, '_query') as mock_query, \
-                patch.object(Postgresql, 'is_leader', Mock(return_value=False)):
+                patch.object(Postgresql, 'is_primary', Mock(return_value=False)):
             mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)]
             ret = self.s.sync_replication_slots(cluster, False)
         self.assertEqual(ret, [])
@@ -106,7 +106,7 @@ class TestSlotsHandler(BaseTestPostgresql):
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {})
 
-    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
         self.cluster.slots['ls'] = 12346
@@ -133,7 +133,7 @@ class TestSlotsHandler(BaseTestPostgresql):
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
-    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
         self.s.copy_logical_slots(self.cluster, ['ls'])
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
@@ -147,7 +147,7 @@ class TestSlotsHandler(BaseTestPostgresql):
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
-    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test_on_promote(self):
         self.s.schedule_advance_slots({'foo': {'bar': 100}})
         self.s.copy_logical_slots(self.cluster, ['ls'])

tests/test_zookeeper.py

@@ -202,7 +202,7 @@ class TestZooKeeper(unittest.TestCase):
         mock_logger.assert_called_once()
 
     def test_delete_leader(self):
-        self.assertTrue(self.zk.delete_leader())
+        self.assertTrue(self.zk.delete_leader(self.zk.get_cluster().leader))
 
     def test_set_failover_value(self):
         self.zk.set_failover_value('')