patroni/patroni/postgresql/config.py

import logging
import os
import re
import shutil
import socket
import stat
import time

from contextlib import contextmanager
from urllib.parse import urlparse, parse_qsl, unquote
from types import TracebackType
from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING

from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
from .. import global_config
from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
from ..exceptions import PatroniFatalException, PostgresConnectionException
from ..file_perm import pg_perm
from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
from ..validator import IntValidator, EnumValidator

if TYPE_CHECKING:  # pragma: no cover
    from . import Postgresql

logger = logging.getLogger(__name__)

PARAMETER_RE = re.compile(r'([a-z_]+)\s*=\s*')


def conninfo_uri_parse(dsn: str) -> Dict[str, str]:
    ret: Dict[str, str] = {}
    r = urlparse(dsn)
    if r.username:
        ret['user'] = r.username
    if r.password:
        ret['password'] = r.password
    if r.path[1:]:
        ret['dbname'] = r.path[1:]
    hosts: List[str] = []
    ports: List[str] = []
    for netloc in r.netloc.split('@')[-1].split(','):
        host = None
        if '[' in netloc and ']' in netloc:
            tmp = netloc.split(']') + ['']
            host = tmp[0][1:]
            netloc = ':'.join(tmp[:2])
        tmp = netloc.rsplit(':', 1)
        if host is None:
            host = tmp[0]
        hosts.append(host)
        ports.append(tmp[1] if len(tmp) == 2 else '')
    if hosts:
        ret['host'] = ','.join(hosts)
    if ports:
        ret['port'] = ','.join(ports)
    ret = {name: unquote(value) for name, value in ret.items()}
    ret.update({name: value for name, value in parse_qsl(r.query)})
    if ret.get('ssl') == 'true':
        del ret['ssl']
        ret['sslmode'] = 'require'
    return ret


def read_param_value(value: str) -> Union[Tuple[None, None], Tuple[str, int]]:
    length = len(value)
    ret = ''
    is_quoted = value[0] == "'"
    i = int(is_quoted)
    while i < length:
        if is_quoted:
            if value[i] == "'":
                return ret, i + 1
        elif value[i].isspace():
            break
        if value[i] == '\\':
            i += 1
            if i >= length:
                break
        ret += value[i]
        i += 1
    return (None, None) if is_quoted else (ret, i)


def conninfo_parse(dsn: str) -> Optional[Dict[str, str]]:
    ret: Dict[str, str] = {}
    length = len(dsn)
    i = 0
    while i < length:
        if dsn[i].isspace():
            i += 1
            continue

        param_match = PARAMETER_RE.match(dsn[i:])
        if not param_match:
            return

        param = param_match.group(1)
        i += param_match.end()

        if i >= length:
            return

        value, end = read_param_value(dsn[i:])
        if value is None or end is None:
            return
        i += end
        ret[param] = value
    return ret


def parse_dsn(value: str) -> Optional[Dict[str, str]]:
    """
    Very simple equivalent of `psycopg2.extensions.parse_dsn` introduced in 2.7.0.
    We are not using psycopg2 function in order to remain compatible with 2.5.4+.
    There is one minor difference though, this function removes `dbname` from the result
    and sets the `sslmode`, 'gssencmode', and `channel_binding` to `prefer` if it is not present in
    the connection string. This is necessary to simplify comparison of the old and the new values.

    >>> r = parse_dsn('postgresql://u%2Fse:pass@:%2f123,[::1]/db%2Fsdf?application_name=mya%2Fpp&ssl=true')
    >>> r == {'application_name': 'mya/pp', 'host': ',::1', 'sslmode': 'require',\
              'password': 'pass', 'port': '/123,', 'user': 'u/se', 'gssencmode': 'prefer', 'channel_binding': 'prefer'}
    True
    >>> r = parse_dsn(" host = 'host' dbname = db\\\\ name requiressl=1 ")
    >>> r == {'host': 'host', 'sslmode': 'require', 'gssencmode': 'prefer', 'channel_binding': 'prefer'}
    True
    >>> parse_dsn('requiressl = 0\\\\') == {'sslmode': 'prefer', 'gssencmode': 'prefer', 'channel_binding': 'prefer'}
    True
    >>> parse_dsn("host=a foo = '") is None
    True
    >>> parse_dsn("host=a foo = ") is None
    True
    >>> parse_dsn("1") is None
    True
    """
    if value.startswith('postgres://') or value.startswith('postgresql://'):
        ret = conninfo_uri_parse(value)
    else:
        ret = conninfo_parse(value)

    if ret:
        if 'sslmode' not in ret:  # allow sslmode to take precedence over requiressl
            requiressl = ret.pop('requiressl', None)
            if requiressl == '1':
                ret['sslmode'] = 'require'
            elif requiressl is not None:
                ret['sslmode'] = 'prefer'
            ret.setdefault('sslmode', 'prefer')
        if 'dbname' in ret:
            del ret['dbname']
        ret.setdefault('gssencmode', 'prefer')
        ret.setdefault('channel_binding', 'prefer')
    return ret


def strip_comment(value: str) -> str:
    i = value.find('#')
    if i > -1:
        value = value[:i].strip()
    return value


def read_recovery_param_value(value: str) -> Optional[str]:
    """
    >>> read_recovery_param_value('') is None
    True
    >>> read_recovery_param_value("'") is None
    True
    >>> read_recovery_param_value("''a") is None
    True
    >>> read_recovery_param_value('a b') is None
    True
    >>> read_recovery_param_value("'''") is None
    True
    >>> read_recovery_param_value("'\\\\") is None
    True
    >>> read_recovery_param_value("'a' s#") is None
    True
    >>> read_recovery_param_value("'\\\\'''' #a")
    "''"
    >>> read_recovery_param_value('asd')
    'asd'
    """
    value = value.strip()
    length = len(value)
    if length == 0:
        return None
    elif value[0] == "'":
        if length == 1:
            return None
        ret = ''
        i = 1
        while i < length:
            if value[i] == '\\':
                i += 1
                if i >= length:
                    return None
            elif value[i] == "'":
                i += 1
                if i >= length:
                    break
                if value[i] in ('#', ' '):
                    if strip_comment(value[i:]):
                        return None
                    break
                if value[i] != "'":
                    return None
            ret += value[i]
            i += 1
        else:
            return None
        return ret
    else:
        value = strip_comment(value)
        if not value or ' ' in value or '\\' in value:
            return None
    return value


def mtime(filename: str) -> Optional[float]:
    try:
        return os.stat(filename).st_mtime
    except OSError:
        return None


class ConfigWriter(object):

    def __init__(self, filename: str) -> None:
        self._filename = filename
        self._fd = None

    def __enter__(self) -> 'ConfigWriter':
        self._fd = open(self._filename, 'w')
        self.writeline('# Do not edit this file manually!\n# It will be overwritten by Patroni!')
        return self

    def __exit__(self, exc_type: Optional[Type[BaseException]],
                 exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> None:
        if self._fd:
            self._fd.close()

    def writeline(self, line: str) -> None:
        if self._fd:
            self._fd.write(line)
            self._fd.write('\n')

    def writelines(self, lines: List[Optional[str]]) -> None:
        for line in lines:
            if isinstance(line, str):
                self.writeline(line)

    @staticmethod
    def escape(value: Any) -> str:  # Escape (by doubling) any single quotes or backslashes in given string
        return re.sub(r'([\'\\])', r'\1\1', str(value))

    def write_param(self, param: str, value: Any) -> None:
        self.writeline("{0} = '{1}'".format(param, self.escape(value)))


def _false_validator(value: Any) -> bool:
    return False


def _bool_validator(value: Any) -> bool:
    return parse_bool(value) is not None


def _bool_is_true_validator(value: Any) -> bool:
    return parse_bool(value) is True


class ConfigHandler(object):

    # List of parameters which must be always passed to postmaster as command line options
    # to make it not possible to change them with 'ALTER SYSTEM'.
    # Some of these parameters have sane default value assigned and Patroni doesn't allow
    # to decrease this value. E.g. 'wal_level' can't be lower then 'hot_standby' and so on.
    # These parameters could be changed only globally, i.e. via DCS.
    # P.S. 'listen_addresses' and 'port' are added here just for convenience, to mark them
    # as a parameters which should always be passed through command line.
    #
    # Format:
    #  key - parameter name
    #  value - tuple(default_value, check_function, min_version)
    #    default_value -- some sane default value
    #    check_function -- if the new value is not correct must return `!False`
    #    min_version -- major version of PostgreSQL when parameter was introduced
    CMDLINE_OPTIONS = CaseInsensitiveDict({
        'listen_addresses': (None, _false_validator, 90100),
        'port': (None, _false_validator, 90100),
        'cluster_name': (None, _false_validator, 90500),
        'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100),
        'hot_standby': ('on', _bool_is_true_validator, 90100),
        'max_connections': (100, IntValidator(min=25), 90100),
        'max_wal_senders': (10, IntValidator(min=3), 90100),
        'wal_keep_segments': (8, IntValidator(min=1), 90100),
        'wal_keep_size': ('128MB', IntValidator(min=16, base_unit='MB'), 130000),
        'max_prepared_transactions': (0, IntValidator(min=0), 90100),
        'max_locks_per_transaction': (64, IntValidator(min=32), 90100),
        'track_commit_timestamp': ('off', _bool_validator, 90500),
        'max_replication_slots': (10, IntValidator(min=4), 90400),
        'max_worker_processes': (8, IntValidator(min=2), 90400),
        'wal_log_hints': ('on', _bool_is_true_validator, 90400)
    })

    _RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys())

    def __init__(self, postgresql: 'Postgresql', config: Dict[str, Any]) -> None:
        self._postgresql = postgresql
        self._config_dir = os.path.abspath(config.get('config_dir', '') or postgresql.data_dir)
        config_base_name = config.get('config_base_name', 'postgresql')
        self._postgresql_conf = os.path.join(self._config_dir, config_base_name + '.conf')
        self._postgresql_conf_mtime = None
        self._postgresql_base_conf_name = config_base_name + '.base.conf'
        self._postgresql_base_conf = os.path.join(self._config_dir, self._postgresql_base_conf_name)
        self._pg_hba_conf = os.path.join(self._config_dir, 'pg_hba.conf')
        self._pg_ident_conf = os.path.join(self._config_dir, 'pg_ident.conf')
        self._recovery_conf = os.path.join(postgresql.data_dir, 'recovery.conf')
        self._recovery_conf_mtime = None
        self._recovery_signal = os.path.join(postgresql.data_dir, 'recovery.signal')
        self._standby_signal = os.path.join(postgresql.data_dir, 'standby.signal')
        self._auto_conf = os.path.join(postgresql.data_dir, 'postgresql.auto.conf')
        self._auto_conf_mtime = None
        self._pgpass = os.path.abspath(config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass'))
        if os.path.exists(self._pgpass) and not os.path.isfile(self._pgpass):
            raise PatroniFatalException("'{0}' exists and it's not a file, check your `postgresql.pgpass` configuration"
                                        .format(self._pgpass))
        self._passfile = None
        self._passfile_mtime = None
        self._postmaster_ctime = None
        self._current_recovery_params: Optional[CaseInsensitiveDict] = None
        self._config = {}
        self._recovery_params = CaseInsensitiveDict()
        self._server_parameters: CaseInsensitiveDict = CaseInsensitiveDict()
        self.reload_config(config)

    def load_current_server_parameters(self) -> None:
        """Read GUC's values from ``pg_settings`` when Patroni is joining the the postgres that is already running."""
        exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator]
        keep_values = {k: self._server_parameters[k] for k in exclude}
        server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query(
            "SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings"
            " WHERE (source IN ('command line', 'environment variable') OR sourcefile = %s)"
            " AND pg_catalog.lower(name) != ALL(%s)", self._postgresql_conf, exclude)})
        recovery_params = CaseInsensitiveDict({k: server_parameters.pop(k) for k in self._RECOVERY_PARAMETERS
                                               if k in server_parameters})
        # We also want to load current settings of recovery parameters, including primary_conninfo
        # and primary_slot_name, otherwise patronictl restart will update postgresql.conf
        # and remove them, what in the worst case will cause another restart.
        # We are doing it only for PostgresSQL v12 onwards, because older version still have recovery.conf
        if not self._postgresql.is_primary() and self._postgresql.major_version >= 120000:
            # primary_conninfo is expected to be a dict, therefore we need to parse it
            recovery_params['primary_conninfo'] = parse_dsn(recovery_params.pop('primary_conninfo', '')) or {}
            self._recovery_params = recovery_params

        self._server_parameters = CaseInsensitiveDict({**server_parameters, **keep_values})

    def setup_server_parameters(self) -> None:
        self._server_parameters = self.get_server_parameters(self._config)
        self._adjust_recovery_parameters()

    def try_to_create_dir(self, d: str, msg: str) -> None:
        d = os.path.join(self._postgresql.data_dir, d)
        if (not is_subpath(self._postgresql.data_dir, d) or not self._postgresql.data_directory_empty()):
            validate_directory(d, msg)

    def check_directories(self) -> None:
        if "unix_socket_directories" in self._server_parameters:
            for d in self._server_parameters["unix_socket_directories"].split(","):
                self.try_to_create_dir(d.strip(), "'{}' is defined in unix_socket_directories, {}")
        if "stats_temp_directory" in self._server_parameters:
            self.try_to_create_dir(self._server_parameters["stats_temp_directory"],
                                   "'{}' is defined in stats_temp_directory, {}")
        if not self._krbsrvname:
            self.try_to_create_dir(os.path.dirname(self._pgpass),
                                   "'{}' is defined in `postgresql.pgpass`, {}")

    @property
    def config_dir(self) -> str:
        return self._config_dir

    @property
    def _configuration_to_save(self) -> List[str]:
        configuration = [os.path.basename(self._postgresql_conf)]
        if 'custom_conf' not in self._config:
            configuration.append(os.path.basename(self._postgresql_base_conf_name))
        if not self.hba_file:
            configuration.append('pg_hba.conf')
        if not self.ident_file:
            configuration.append('pg_ident.conf')
        return configuration

    def set_file_permissions(self, filename: str) -> None:
        """Set permissions of file *filename* according to the expected permissions if it resides under PGDATA.

        .. note::
            Do nothing if the file is not under PGDATA.

        :param filename: path to a file which permissions might need to be adjusted.
        """
        if is_subpath(self._postgresql.data_dir, filename):
            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
            os.chmod(filename, pg_perm.file_create_mode)

    @contextmanager
    def config_writer(self, filename: str) -> Iterator[ConfigWriter]:
        """Create :class:`ConfigWriter` object and set permissions on a *filename*.

        :param filename: path to a config file.

        :yields: :class:`ConfigWriter` object.
        """
        with ConfigWriter(filename) as writer:
            yield writer
        self.set_file_permissions(filename)

    def save_configuration_files(self, check_custom_bootstrap: bool = False) -> bool:
        """
            copy postgresql.conf to postgresql.conf.backup to be able to retrieve configuration files
            - originally stored as symlinks, those are normally skipped by pg_basebackup
            - in case of WAL-E basebackup (see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239)
        """
        if not (check_custom_bootstrap and self._postgresql.bootstrap.running_custom_bootstrap):
            try:
                for f in self._configuration_to_save:
                    config_file = os.path.join(self._config_dir, f)
                    backup_file = os.path.join(self._postgresql.data_dir, f + '.backup')
                    if os.path.isfile(config_file):
                        shutil.copy(config_file, backup_file)
                        self.set_file_permissions(backup_file)
            except IOError:
                logger.exception('unable to create backup copies of configuration files')
        return True

    def restore_configuration_files(self) -> None:
        """ restore a previously saved postgresql.conf """
        try:
            for f in self._configuration_to_save:
                config_file = os.path.join(self._config_dir, f)
                backup_file = os.path.join(self._postgresql.data_dir, f + '.backup')
                if not os.path.isfile(config_file):
                    if os.path.isfile(backup_file):
                        shutil.copy(backup_file, config_file)
                        self.set_file_permissions(config_file)
                    # Previously we didn't backup pg_ident.conf, if file is missing just create empty
                    elif f == 'pg_ident.conf':
                        open(config_file, 'w').close()
                        self.set_file_permissions(config_file)
        except IOError:
            logger.exception('unable to restore configuration files from backup')

    def write_postgresql_conf(self, configuration: Optional[CaseInsensitiveDict] = None) -> None:
        # rename the original configuration if it is necessary
        if 'custom_conf' not in self._config and not os.path.exists(self._postgresql_base_conf):
            os.rename(self._postgresql_conf, self._postgresql_base_conf)

        configuration = configuration or self._server_parameters.copy()
        # Due to the permanent logical replication slots configured we have to enable hot_standby_feedback
        if self._postgresql.enforce_hot_standby_feedback:
            configuration['hot_standby_feedback'] = 'on'

        with self.config_writer(self._postgresql_conf) as f:
            include = self._config.get('custom_conf') or self._postgresql_base_conf_name
            f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
            for name, value in sorted((configuration).items()):
                value = transform_postgresql_parameter_value(self._postgresql.major_version, name, value,
                                                             self._postgresql.available_gucs)
                if value is not None and\
                        (name != 'hba_file' or not self._postgresql.bootstrap.running_custom_bootstrap):
                    f.write_param(name, value)
            # when we are doing custom bootstrap we assume that we don't know superuser password
            # and in order to be able to change it, we are opening trust access from a certain address
            # therefore we need to make sure that hba_file is not overridden
            # after changing superuser password we will "revert" all these "changes"
            if self._postgresql.bootstrap.running_custom_bootstrap or 'hba_file' not in self._server_parameters:
                f.write_param('hba_file', self._pg_hba_conf)
            if 'ident_file' not in self._server_parameters:
                f.write_param('ident_file', self._pg_ident_conf)

            if self._postgresql.major_version >= 120000:
                if self._recovery_params:
                    f.writeline('\n# recovery.conf')
                    self._write_recovery_params(f, self._recovery_params)

                if not self._postgresql.bootstrap.keep_existing_recovery_conf:
                    self._sanitize_auto_conf()

    def append_pg_hba(self, config: List[str]) -> bool:
        if not self.hba_file and not self._config.get('pg_hba'):
            with open(self._pg_hba_conf, 'a') as f:
                f.write('\n{}\n'.format('\n'.join(config)))
            self.set_file_permissions(self._pg_hba_conf)
        return True

    def replace_pg_hba(self) -> Optional[bool]:
        """
        Replace pg_hba.conf content in the PGDATA if hba_file is not defined in the
        `postgresql.parameters` and pg_hba is defined in `postgresql` configuration section.

        :returns: True if pg_hba.conf was rewritten.
        """

        # when we are doing custom bootstrap we assume that we don't know superuser password
        # and in order to be able to change it, we are opening trust access from a certain address
        if self._postgresql.bootstrap.running_custom_bootstrap:
            addresses = {} if os.name == 'nt' else {'': 'local'}  # windows doesn't yet support unix-domain sockets
            if 'host' in self.local_replication_address and not self.local_replication_address['host'].startswith('/'):
                addresses.update({sa[0] + '/32': 'host' for _, _, _, _, sa in socket.getaddrinfo(
                                  self.local_replication_address['host'], self.local_replication_address['port'],
                                  0, socket.SOCK_STREAM, socket.IPPROTO_TCP)})

            with self.config_writer(self._pg_hba_conf) as f:
                for address, t in addresses.items():
                    f.writeline((
                        '{0}\treplication\t{1}\t{3}\ttrust\n'
                        '{0}\tall\t{2}\t{3}\ttrust'
                    ).format(t, self.replication['username'], self._superuser.get('username') or 'all', address))
        elif not self.hba_file and self._config.get('pg_hba'):
            with self.config_writer(self._pg_hba_conf) as f:
                f.writelines(self._config['pg_hba'])
            return True

    def replace_pg_ident(self) -> Optional[bool]:
        """
        Replace pg_ident.conf content in the PGDATA if ident_file is not defined in the
        `postgresql.parameters` and pg_ident is defined in the `postgresql` section.

        :returns: True if pg_ident.conf was rewritten.
        """

        if not self.ident_file and self._config.get('pg_ident'):
            with self.config_writer(self._pg_ident_conf) as f:
                f.writelines(self._config['pg_ident'])
            return True

    def primary_conninfo_params(self, member: Union[Leader, Member, None]) -> Optional[Dict[str, Any]]:
        if not member or not member.conn_url or member.name == self._postgresql.name:
            return None
        ret = member.conn_kwargs(self.replication)
        ret['application_name'] = self._postgresql.name
        ret.setdefault('sslmode', 'prefer')
        if self._postgresql.major_version >= 120000:
            ret.setdefault('gssencmode', 'prefer')
        if self._postgresql.major_version >= 130000:
            ret.setdefault('channel_binding', 'prefer')
        if self._krbsrvname:
            ret['krbsrvname'] = self._krbsrvname
        if 'dbname' in ret:
            del ret['dbname']
        return ret

    def format_dsn(self, params: Dict[str, Any], include_dbname: bool = False) -> str:
        # A list of keywords that can be found in a conninfo string. Follows what is acceptable by libpq
        keywords = ('dbname', 'user', 'passfile' if params.get('passfile') else 'password', 'host', 'port',
                    'sslmode', 'sslcompression', 'sslcert', 'sslkey', 'sslpassword', 'sslrootcert', 'sslcrl',
                    'sslcrldir', 'application_name', 'krbsrvname', 'gssencmode', 'channel_binding',
                    'target_session_attrs')
        if include_dbname:
            params = params.copy()
            if 'dbname' not in params:
                params['dbname'] = self._postgresql.database
            # we are abusing information about the necessity of dbname
            # dsn should contain passfile or password only if there is no dbname in it (it is used in recovery.conf)
            skip = {'passfile', 'password'}
        else:
            skip = {'dbname'}

        def escape(value: Any) -> str:
            return re.sub(r'([\'\\ ])', r'\\\1', str(value))

        return ' '.join('{0}={1}'.format(kw, escape(params[kw])) for kw in keywords
                        if kw not in skip and params.get(kw) is not None)

    def _write_recovery_params(self, fd: ConfigWriter, recovery_params: CaseInsensitiveDict) -> None:
        if self._postgresql.major_version >= 90500:
            pause_at_recovery_target = parse_bool(recovery_params.pop('pause_at_recovery_target', None))
            if pause_at_recovery_target is not None:
                recovery_params.setdefault('recovery_target_action', 'pause' if pause_at_recovery_target else 'promote')
        else:
            if str(recovery_params.pop('recovery_target_action', None)).lower() == 'promote':
                recovery_params.setdefault('pause_at_recovery_target', 'false')
        for name, value in sorted(recovery_params.items()):
            if name == 'primary_conninfo':
                if 'password' in value and self._postgresql.major_version >= 100000:
                    self.write_pgpass(value)
                    value['passfile'] = self._passfile = self._pgpass
                    self._passfile_mtime = mtime(self._pgpass)
                value = self.format_dsn(value)
            else:
                value = transform_recovery_parameter_value(self._postgresql.major_version, name, value,
                                                           self._postgresql.available_gucs)
                if value is None:
                    continue
            fd.write_param(name, value)

    def build_recovery_params(self, member: Union[Leader, Member, None]) -> CaseInsensitiveDict:
        recovery_params = CaseInsensitiveDict({p: v for p, v in (self.get('recovery_conf') or {}).items()
                                               if not p.lower().startswith('recovery_target')
                                               and p.lower() not in ('primary_conninfo', 'primary_slot_name')})
        recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
        if self._postgresql.major_version >= 120000:
            # on pg12 we want to protect from following params being set in one of included files
            # not doing so might result in a standby being paused, promoted or shutted down.
            recovery_params.update({'recovery_target': '', 'recovery_target_name': '', 'recovery_target_time': '',
                                    'recovery_target_xid': '', 'recovery_target_lsn': ''})

        is_remote_member = isinstance(member, RemoteMember)
        primary_conninfo = self.primary_conninfo_params(member)
        if primary_conninfo:
            use_slots = global_config.use_slots and self._postgresql.major_version >= 90400
            if use_slots and not (is_remote_member and member.no_replication_slot):
                primary_slot_name = member.primary_slot_name if is_remote_member else self._postgresql.name
                recovery_params['primary_slot_name'] = slot_name_from_member_name(primary_slot_name)
                # We are a standby leader and are using a replication slot. Make sure we connect to
                # the leader of the main cluster (in case more than one host is specified in the
                # connstr) by adding 'target_session_attrs=read-write' to primary_conninfo.
                if is_remote_member and 'target_sesions_attrs' not in primary_conninfo and\
                        self._postgresql.major_version >= 100000:
                    primary_conninfo['target_session_attrs'] = 'read-write'
            recovery_params['primary_conninfo'] = primary_conninfo

        # standby_cluster config might have different parameters, we want to override them
        standby_cluster_params = ['restore_command', 'archive_cleanup_command']\
            + (['recovery_min_apply_delay'] if is_remote_member else [])
        recovery_params.update({p: member.data.get(p) for p in standby_cluster_params if member and member.data.get(p)})
        return recovery_params

    def recovery_conf_exists(self) -> bool:
        if self._postgresql.major_version >= 120000:
            return os.path.exists(self._standby_signal) or os.path.exists(self._recovery_signal)
        return os.path.exists(self._recovery_conf)

    @property
    def triggerfile_good_name(self) -> str:
        return 'trigger_file' if self._postgresql.major_version < 120000 else 'promote_trigger_file'

    @property
    def _triggerfile_wrong_name(self) -> str:
        return 'trigger_file' if self._postgresql.major_version >= 120000 else 'promote_trigger_file'

    @property
    def _recovery_parameters_to_compare(self) -> CaseInsensitiveSet:
        skip_params = CaseInsensitiveSet({'pause_at_recovery_target', 'recovery_target_inclusive',
                                          'recovery_target_action', 'standby_mode', self._triggerfile_wrong_name})
        return CaseInsensitiveSet(self._RECOVERY_PARAMETERS - skip_params)

    def _read_recovery_params(self) -> Tuple[Optional[CaseInsensitiveDict], bool]:
        """Read current recovery parameters values.

        .. note::
            We query Postgres only if we detected that Postgresql was restarted
            or when at least one of the following files was updated:

                * ``postgresql.conf``;
                * ``postgresql.auto.conf``;
                * ``passfile`` that is used in the ``primary_conninfo``.

        :returns: a tuple with two elements:

            * :class:`CaseInsensitiveDict` object with current values of recovery parameters,
              or ``None`` if no configuration files were updated;

            * ``True`` if new values of recovery parameters were queried, ``False`` otherwise.
        """
        if self._postgresql.is_starting():
            return None, False

        pg_conf_mtime = mtime(self._postgresql_conf)
        auto_conf_mtime = mtime(self._auto_conf)
        passfile_mtime = mtime(self._passfile) if self._passfile else False
        postmaster_ctime = self._postgresql.is_running()
        if postmaster_ctime:
            postmaster_ctime = postmaster_ctime.create_time()

        if self._postgresql_conf_mtime == pg_conf_mtime and self._auto_conf_mtime == auto_conf_mtime \
                and self._passfile_mtime == passfile_mtime and self._postmaster_ctime == postmaster_ctime:
            return None, False

        try:
            values = self._get_pg_settings(self._recovery_parameters_to_compare).values()
            values = CaseInsensitiveDict({p[0]: [p[1], p[4] == 'postmaster', p[5]] for p in values})
            self._postgresql_conf_mtime = pg_conf_mtime
            self._auto_conf_mtime = auto_conf_mtime
            self._postmaster_ctime = postmaster_ctime
        except Exception as exc:
            if all((isinstance(exc, PostgresConnectionException),
                    self._postgresql_conf_mtime == pg_conf_mtime,
                    self._auto_conf_mtime == auto_conf_mtime,
                    self._passfile_mtime == passfile_mtime,
                    self._postmaster_ctime != postmaster_ctime)):
                # We detected that the connection to postgres fails, but the process creation time of the postmaster
                # doesn't match the old value. It is an indicator that Postgres crashed and either doing crash
                # recovery or down. In this case we return values like nothing changed in the config.
                return None, False
            values = None
        return values, True

    def _read_recovery_params_pre_v12(self) -> Tuple[Optional[CaseInsensitiveDict], bool]:
        recovery_conf_mtime = mtime(self._recovery_conf)
        passfile_mtime = mtime(self._passfile) if self._passfile else False
        if recovery_conf_mtime == self._recovery_conf_mtime and passfile_mtime == self._passfile_mtime:
            return None, False

        values = CaseInsensitiveDict()
        with open(self._recovery_conf, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                value = None
                match = PARAMETER_RE.match(line)
                if match:
                    value = read_recovery_param_value(line[match.end():])
                if match is None or value is None:
                    return None, True
                values[match.group(1)] = [value, True]
            self._recovery_conf_mtime = recovery_conf_mtime
        values.setdefault('recovery_min_apply_delay', ['0', True])
        values['recovery_min_apply_delay'][0] = parse_int(values['recovery_min_apply_delay'][0], 'ms')
        values.update({param: ['', True] for param in self._recovery_parameters_to_compare if param not in values})
        return values, True

    def _check_passfile(self, passfile: str, wanted_primary_conninfo: Dict[str, Any]) -> bool:
        # If there is a passfile in the primary_conninfo try to figure out that
        # the passfile contains the line(s) allowing connection to the given node.
        # We assume that the passfile was created by Patroni and therefore doing
        # the full match and not covering cases when host, port or user are set to '*'
        passfile_mtime = mtime(passfile)
        if passfile_mtime:
            try:
                with open(passfile) as f:
                    wanted_lines = (self._pgpass_line(wanted_primary_conninfo) or '').splitlines()
                    file_lines = f.read().splitlines()
                    if set(wanted_lines) == set(file_lines):
                        self._passfile = passfile
                        self._passfile_mtime = passfile_mtime
                        return True
            except Exception:
                logger.info('Failed to read %s', passfile)
        return False

    def _check_primary_conninfo(self, primary_conninfo: Dict[str, Any],
                                wanted_primary_conninfo: Dict[str, Any]) -> bool:
        # first we will cover corner cases, when we are replicating from somewhere while shouldn't
        # or there is no primary_conninfo but we should replicate from some specific node.
        if not wanted_primary_conninfo:
            return not primary_conninfo
        elif not primary_conninfo:
            return False

        if not self._postgresql.is_starting():
            wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
            if wal_receiver_primary_conninfo:
                wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo)
                # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison
                if wal_receiver_primary_conninfo:
                    primary_conninfo = wal_receiver_primary_conninfo
                    # There could be no password in the primary_conninfo or it is masked.
                    # Just copy the "desired" value in order to make comparison succeed.
                    if 'password' in wanted_primary_conninfo:
                        primary_conninfo['password'] = wanted_primary_conninfo['password']

        if 'passfile' in primary_conninfo and 'password' not in primary_conninfo \
                and 'password' in wanted_primary_conninfo:
            if self._check_passfile(primary_conninfo['passfile'], wanted_primary_conninfo):
                primary_conninfo['password'] = wanted_primary_conninfo['password']
            else:
                return False

        return all(str(primary_conninfo.get(p)) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None)

    def check_recovery_conf(self, member: Union[Leader, Member, None]) -> Tuple[bool, bool]:
        """Returns a tuple. The first boolean element indicates that recovery params don't match
           and the second is set to `True` if the restart is required in order to apply new values"""

        # TODO: recovery.conf could be stale, would be nice to detect that.
        if self._postgresql.major_version >= 120000:
            if not os.path.exists(self._standby_signal):
                return True, True

            _read_recovery_params = self._read_recovery_params
        else:
            if not self.recovery_conf_exists():
                return True, True

            _read_recovery_params = self._read_recovery_params_pre_v12

        params, updated = _read_recovery_params()
        # updated indicates that mtime of postgresql.conf, postgresql.auto.conf, or recovery.conf
        # was changed and params were read either from the config or from the database connection.
        if updated:
            if params is None:  # exception or unparsable config
                return True, True

            # We will cache parsed value until the next config change.
            self._current_recovery_params = params
            primary_conninfo = params['primary_conninfo']
            if primary_conninfo[0]:
                primary_conninfo[0] = parse_dsn(params['primary_conninfo'][0])
                # If we failed to parse non-empty connection string this indicates that config if broken.
                if not primary_conninfo[0]:
                    return True, True
            else:  # empty string, primary_conninfo is not in the config
                primary_conninfo[0] = {}

        if not self._postgresql.is_starting() and self._current_recovery_params:
            # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver
            wal_receiver_primary_slot_name = self._postgresql.primary_slot_name()
            if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo():
                wal_receiver_primary_slot_name = ''
            if wal_receiver_primary_slot_name is not None:
                self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name

        # Increment the 'reload' to enforce write of postgresql.conf when joining the running postgres
        required = {'restart': 0,
                    'reload': int(self._postgresql.major_version >= 120000
                                  and not self._postgresql.cb_called
                                  and not self._postgresql.is_starting())}

        def record_missmatch(mtype: bool) -> None:
            required['restart' if mtype else 'reload'] += 1

        wanted_recovery_params = self.build_recovery_params(member)
        for param, value in (self._current_recovery_params or {}).items():
            # Skip certain parameters defined in the included postgres config files
            # if we know that they are not specified in the patroni configuration.
            if len(value) > 2 and value[2] not in (self._postgresql_conf, self._auto_conf) and \
                    param in ('archive_cleanup_command', 'promote_trigger_file', 'recovery_end_command',
                              'recovery_min_apply_delay', 'restore_command') and param not in wanted_recovery_params:
                continue
            if param == 'recovery_min_apply_delay':
                if not compare_values('integer', 'ms', value[0], wanted_recovery_params.get(param, 0)):
                    record_missmatch(value[1])
            elif param == 'standby_mode':
                if not compare_values('bool', None, value[0], wanted_recovery_params.get(param, 'on')):
                    record_missmatch(value[1])
            elif param == 'primary_conninfo':
                if not self._check_primary_conninfo(value[0], wanted_recovery_params.get('primary_conninfo', {})):
                    record_missmatch(value[1])
            elif (param != 'primary_slot_name' or wanted_recovery_params.get('primary_conninfo')) \
                    and str(value[0]) != str(wanted_recovery_params.get(param, '')):
                record_missmatch(value[1])
        return required['restart'] + required['reload'] > 0, required['restart'] > 0

    @staticmethod
    def _remove_file_if_exists(name: str) -> None:
        if os.path.isfile(name) or os.path.islink(name):
            os.unlink(name)

    @staticmethod
    def _pgpass_line(record: Dict[str, Any]) -> Optional[str]:
        if 'password' in record:
            def escape(value: Any) -> str:
                return re.sub(r'([:\\])', r'\\\1', str(value))

            record = {n: escape(record.get(n) or '*') for n in ('host', 'port', 'user', 'password')}
            # 'host' could be several comma-separated hostnames, in this case
            # we need to write on pgpass line per host
            line = ''
            for hostname in record['host'].split(','):
                line += hostname + ':{port}:*:{user}:{password}'.format(**record) + '\n'
            return line.rstrip()

    def write_pgpass(self, record: Dict[str, Any]) -> Dict[str, str]:
        line = self._pgpass_line(record)
        if not line:
            return os.environ.copy()

        with open(self._pgpass, 'w') as f:
            os.chmod(self._pgpass, stat.S_IWRITE | stat.S_IREAD)
            f.write(line)

        return {**os.environ, 'PGPASSFILE': self._pgpass}

    def write_recovery_conf(self, recovery_params: CaseInsensitiveDict) -> None:
        self._recovery_params = recovery_params
        if self._postgresql.major_version >= 120000:
            if parse_bool(recovery_params.pop('standby_mode', None)):
                open(self._standby_signal, 'w').close()
                self.set_file_permissions(self._standby_signal)
            else:
                self._remove_file_if_exists(self._standby_signal)
                open(self._recovery_signal, 'w').close()
                self.set_file_permissions(self._recovery_signal)

            def restart_required(name: str) -> bool:
                if self._postgresql.major_version >= 140000:
                    return False
                return name == 'restore_command' or (self._postgresql.major_version < 130000
                                                     and name in ('primary_conninfo', 'primary_slot_name'))

            self._current_recovery_params = CaseInsensitiveDict({n: [v, restart_required(n), self._postgresql_conf]
                                                                 for n, v in recovery_params.items()})
        else:
            with self.config_writer(self._recovery_conf) as f:
                self._write_recovery_params(f, recovery_params)

    def remove_recovery_conf(self) -> None:
        for name in (self._recovery_conf, self._standby_signal, self._recovery_signal):
            self._remove_file_if_exists(name)
        self._recovery_params = CaseInsensitiveDict()
        self._current_recovery_params = None

    def _sanitize_auto_conf(self) -> None:
        overwrite = False
        lines: List[str] = []

        if os.path.exists(self._auto_conf):
            try:
                with open(self._auto_conf) as f:
                    for raw_line in f:
                        line = raw_line.strip()
                        match = PARAMETER_RE.match(line)
                        if match and match.group(1).lower() in self._RECOVERY_PARAMETERS:
                            overwrite = True
                        else:
                            lines.append(raw_line)
            except Exception:
                logger.info('Failed to read %s', self._auto_conf)

        if overwrite:
            try:
                with open(self._auto_conf, 'w') as f:
                    self.set_file_permissions(self._auto_conf)
                    for raw_line in lines:
                        f.write(raw_line)
            except Exception:
                logger.exception('Failed to remove some unwanted parameters from %s', self._auto_conf)

    def _adjust_recovery_parameters(self) -> None:
        # It is not strictly necessary, but we can make patroni configs crossi-compatible with all postgres versions.
        recovery_conf = {n: v for n, v in self._server_parameters.items() if n.lower() in self._RECOVERY_PARAMETERS}
        if recovery_conf:
            self._config['recovery_conf'] = recovery_conf

        if self.get('recovery_conf'):
            value = self._config['recovery_conf'].pop(self._triggerfile_wrong_name, None)
            if self.triggerfile_good_name not in self._config['recovery_conf'] and value:
                self._config['recovery_conf'][self.triggerfile_good_name] = value

    def get_server_parameters(self, config: Dict[str, Any]) -> CaseInsensitiveDict:
        parameters = config['parameters'].copy()
        listen_addresses, port = split_host_port(config['listen'], 5432)
        parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port))
        if global_config.is_synchronous_mode:
            synchronous_standby_names = self._server_parameters.get('synchronous_standby_names')
            if synchronous_standby_names is None:
                if global_config.is_synchronous_mode_strict\
                        and self._postgresql.role in ('master', 'primary', 'promoted'):
                    parameters['synchronous_standby_names'] = '*'
                else:
                    parameters.pop('synchronous_standby_names', None)
            else:
                parameters['synchronous_standby_names'] = synchronous_standby_names

        # Handle hot_standby <-> replica rename
        if parameters.get('wal_level') == ('hot_standby' if self._postgresql.major_version >= 90600 else 'replica'):
            parameters['wal_level'] = 'replica' if self._postgresql.major_version >= 90600 else 'hot_standby'

        # Try to recalcualte wal_keep_segments <-> wal_keep_size assuming that typical wal_segment_size is 16MB.
        # The real segment size could be estimated from pg_control, but we don't really care, because the only goal of
        # this exercise is improving cross version compatibility and user must set the correct parameter in the config.
        if self._postgresql.major_version >= 130000:
            wal_keep_segments = parameters.pop('wal_keep_segments', self.CMDLINE_OPTIONS['wal_keep_segments'][0])
            parameters.setdefault('wal_keep_size', str(int(wal_keep_segments) * 16) + 'MB')
        elif self._postgresql.major_version:
            wal_keep_size = parse_int(parameters.pop('wal_keep_size', self.CMDLINE_OPTIONS['wal_keep_size'][0]), 'MB')
            parameters.setdefault('wal_keep_segments', int(((wal_keep_size or 0) + 8) / 16))

        self._postgresql.citus_handler.adjust_postgres_gucs(parameters)

        ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version
                                   or self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]})
        ret.update({k: os.path.join(self._config_dir, ret[k]) for k in ('hba_file', 'ident_file') if k in ret})
        return ret

    @staticmethod
    def _get_unix_local_address(unix_socket_directories: str) -> str:
        for d in unix_socket_directories.split(','):
            d = d.strip()
            if d.startswith('/'):  # Only absolute path can be used to connect via unix-socket
                return d
        return ''

    def _get_tcp_local_address(self) -> str:
        listen_addresses = self._server_parameters['listen_addresses'].split(',')

        for la in listen_addresses:
            if la.strip().lower() in ('*', '0.0.0.0', '127.0.0.1', 'localhost'):  # we are listening on '*' or localhost
                return 'localhost'  # connection via localhost is preferred
        return listen_addresses[0].strip()  # can't use localhost, take first address from listen_addresses

    def resolve_connection_addresses(self) -> None:
        """Calculates and sets local and remote connection urls and options.

        This method sets:
            * :attr:`Postgresql.connection_string <patroni.postgresql.Postgresql.connection_string>` attribute, which
              is later written to the member key in DCS as ``conn_url``.
            * :attr:`ConfigHandler.local_replication_address` attribute, which is used for replication connections to
              local postgres.
            * :attr:`ConnectionPool.conn_kwargs <patroni.postgresql.connection.ConnectionPool.conn_kwargs>` attribute,
              which is used for superuser connections to local postgres.

        .. note::
            If there is a valid directory in ``postgresql.parameters.unix_socket_directories`` in the Patroni
            configuration and ``postgresql.use_unix_socket`` and/or ``postgresql.use_unix_socket_repl``
            are set to ``True``, we respectively use unix sockets for superuser and replication connections
            to local postgres.

            If there is a requirement to use unix sockets, but nothing is set in the
            ``postgresql.parameters.unix_socket_directories``, we omit a ``host`` in connection parameters relying
            on the ability of ``libpq`` to connect via some default unix socket directory.

            If unix sockets are not requested we "switch" to TCP, prefering to use ``localhost`` if it is possible
            to deduce that Postgres is listening on a local interface address.

            Otherwise we just used the first address specified in the ``listen_addresses`` GUC.
        """
        port = self._server_parameters['port']
        tcp_local_address = self._get_tcp_local_address()
        netloc = self._config.get('connect_address') or tcp_local_address + ':' + port

        unix_local_address = {'port': port}
        unix_socket_directories = self._server_parameters.get('unix_socket_directories')
        if unix_socket_directories is not None:
            # fallback to tcp if unix_socket_directories is set, but there are no suitable values
            unix_local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address

        tcp_local_address = {'host': tcp_local_address, 'port': port}

        self.local_replication_address = unix_local_address\
            if self._config.get('use_unix_socket_repl') else tcp_local_address

        self._postgresql.connection_string = uri('postgres', netloc, self._postgresql.database)

        local_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address
        local_conn_kwargs = {
            **local_address,
            **self._superuser,
            'dbname': self._postgresql.database,
            'fallback_application_name': 'Patroni',
            'connect_timeout': 3,
            'options': '-c statement_timeout=2000'
        }
        # if the "username" parameter is present, it actually needs to be "user" for connecting to PostgreSQL
        if 'username' in local_conn_kwargs:
            local_conn_kwargs['user'] = local_conn_kwargs.pop('username')
        # "notify" connection_pool about the "new" local connection address
        self._postgresql.connection_pool.conn_kwargs = local_conn_kwargs

    def _get_pg_settings(self, names: Collection[str]) -> Dict[Any, Tuple[Any, ...]]:
        return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile'
                                                         + ' FROM pg_catalog.pg_settings '
                                                         + ' WHERE pg_catalog.lower(name) = ANY(%s)'),
                                                        [n.lower() for n in names])}

    @staticmethod
    def _handle_wal_buffers(old_values: Dict[Any, Tuple[Any, ...]], changes: CaseInsensitiveDict) -> None:
        wal_block_size = parse_int(old_values['wal_block_size'][1]) or 8192
        wal_segment_size = old_values['wal_segment_size']
        wal_segment_unit = parse_int(wal_segment_size[2], 'B') or 8192 \
            if wal_segment_size[2] is not None and wal_segment_size[2][0].isdigit() else 1
        wal_segment_size = parse_int(wal_segment_size[1]) or (16777216 if wal_segment_size[2] is None else 2048)
        wal_segment_size *= wal_segment_unit / wal_block_size
        default_wal_buffers = min(max((parse_int(old_values['shared_buffers'][1]) or 16384) / 32, 8), wal_segment_size)

        wal_buffers = old_values['wal_buffers']
        new_value = str(changes['wal_buffers'] or -1)

        new_value = default_wal_buffers if new_value == '-1' else parse_int(new_value, wal_buffers[2])
        old_value = default_wal_buffers if wal_buffers[1] == '-1' else parse_int(*wal_buffers[1:3])

        if new_value == old_value:
            del changes['wal_buffers']

    def reload_config(self, config: Dict[str, Any], sighup: bool = False) -> None:
        self._superuser = config['authentication'].get('superuser', {})
        server_parameters = self.get_server_parameters(config)
        params_skip_changes = CaseInsensitiveSet((*self._RECOVERY_PARAMETERS, 'hot_standby', 'wal_log_hints'))

        conf_changed = hba_changed = ident_changed = local_connection_address_changed = pending_restart = False
        if self._postgresql.state == 'running':
            changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items()
                                           if p not in params_skip_changes})
            changes.update({p: None for p in self._server_parameters.keys()
                            if not (p in changes or p in params_skip_changes)})
            if changes:
                undef = []
                if 'wal_buffers' in changes:  # we need to calculate the default value of wal_buffers
                    undef = [p for p in ('shared_buffers', 'wal_segment_size', 'wal_block_size') if p not in changes]
                    changes.update({p: None for p in undef})
                # XXX: query can raise an exception
                old_values = self._get_pg_settings(changes.keys())
                if 'wal_buffers' in changes:
                    self._handle_wal_buffers(old_values, changes)
                    for p in undef:
                        del changes[p]

                for r in old_values.values():
                    if r[4] != 'internal' and r[0] in changes:
                        new_value = changes.pop(r[0])
                        if new_value is None or not compare_values(r[3], r[2], r[1], new_value):
                            conf_changed = True
                            if r[4] == 'postmaster':
                                pending_restart = True
                                logger.info('Changed %s from %s to %s (restart might be required)',
                                            r[0], r[1], new_value)
                                if config.get('use_unix_socket') and r[0] == 'unix_socket_directories'\
                                        or r[0] in ('listen_addresses', 'port'):
                                    local_connection_address_changed = True
                            else:
                                logger.info('Changed %s from %s to %s', r[0], r[1], new_value)
                        elif r[0] in self._server_parameters \
                                and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]):
                            # Check if any parameter was set back to the current pg_settings value
                            # We can use pg_settings value here, as it is proved to be equal to new_value
                            logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1])
                            conf_changed = True
                for param, value in changes.items():
                    if '.' in param:
                        # Check that user-defined-paramters have changed (parameters with period in name)
                        if value is None or param not in self._server_parameters \
                                or str(value) != str(self._server_parameters[param]):
                            logger.info('Changed %s from %s to %s', param, self._server_parameters.get(param), value)
                            conf_changed = True
                    elif param in server_parameters:
                        logger.warning('Removing invalid parameter `%s` from postgresql.parameters', param)
                        server_parameters.pop(param)

            if (not server_parameters.get('hba_file') or server_parameters['hba_file'] == self._pg_hba_conf) \
                    and config.get('pg_hba'):
                hba_changed = self._config.get('pg_hba', []) != config['pg_hba']

            if (not server_parameters.get('ident_file') or server_parameters['ident_file'] == self._pg_hba_conf) \
                    and config.get('pg_ident'):
                ident_changed = self._config.get('pg_ident', []) != config['pg_ident']

        self._config = config
        self._postgresql.set_pending_restart(pending_restart)
        self._server_parameters = server_parameters
        self._adjust_recovery_parameters()
        self._krbsrvname = config.get('krbsrvname')

        # for not so obvious connection attempts that may happen outside of pyscopg2
        if self._krbsrvname:
            os.environ['PGKRBSRVNAME'] = self._krbsrvname

        if not local_connection_address_changed:
            self.resolve_connection_addresses()

        proxy_addr = config.get('proxy_address')
        self._postgresql.proxy_url = uri('postgres', proxy_addr, self._postgresql.database) if proxy_addr else None

        if conf_changed:
            self.write_postgresql_conf()

        if hba_changed:
            self.replace_pg_hba()

        if ident_changed:
            self.replace_pg_ident()

        if sighup or conf_changed or hba_changed or ident_changed:
            logger.info('Reloading PostgreSQL configuration.')
            self._postgresql.reload()
            if self._postgresql.major_version >= 90500:
                time.sleep(1)
                try:
                    pending_restart = self._postgresql.query(
                        'SELECT COUNT(*) FROM pg_catalog.pg_settings'
                        ' WHERE pg_catalog.lower(name) != ALL(%s) AND pending_restart',
                        [n.lower() for n in params_skip_changes])[0][0] > 0
                    self._postgresql.set_pending_restart(pending_restart)
                except Exception as e:
                    logger.warning('Exception %r when running query', e)
        else:
            logger.info('No PostgreSQL configuration items changed, nothing to reload.')

    def set_synchronous_standby_names(self, value: Optional[str]) -> Optional[bool]:
        """Updates synchronous_standby_names and reloads if necessary.
        :returns: True if value was updated."""
        if value != self._server_parameters.get('synchronous_standby_names'):
            if value is None:
                self._server_parameters.pop('synchronous_standby_names', None)
            else:
                self._server_parameters['synchronous_standby_names'] = value
            if self._postgresql.state == 'running':
                self.write_postgresql_conf()
                self._postgresql.reload()
            return True

    @property
    def effective_configuration(self) -> CaseInsensitiveDict:
        """It might happen that the current value of one (or more) below parameters stored in
        the controldata is higher than the value stored in the global cluster configuration.

        Example: max_connections in global configuration is 100, but in controldata
        `Current max_connections setting: 200`. If we try to start postgres with
        max_connections=100, it will immediately exit.
        As a workaround we will start it with the values from controldata and set `pending_restart`
        to true as an indicator that current values of parameters are not matching expectations."""

        if self._postgresql.role in ('master', 'primary'):
            return self._server_parameters

        options_mapping = {
            'max_connections': 'max_connections setting',
            'max_prepared_transactions': 'max_prepared_xacts setting',
            'max_locks_per_transaction': 'max_locks_per_xact setting'
        }

        if self._postgresql.major_version >= 90400:
            options_mapping['max_worker_processes'] = 'max_worker_processes setting'

        if self._postgresql.major_version >= 120000:
            options_mapping['max_wal_senders'] = 'max_wal_senders setting'

        data = self._postgresql.controldata()
        effective_configuration = self._server_parameters.copy()

        for name, cname in options_mapping.items():
            value = parse_int(effective_configuration[name])
            if cname not in data:
                logger.warning('%s is missing from pg_controldata output', cname)
                continue

            cvalue = parse_int(data[cname])
            if cvalue is not None and value is not None and cvalue > value:
                effective_configuration[name] = cvalue
                self._postgresql.set_pending_restart(True)

        # If we are using custom bootstrap with PITR it could fail when values like max_connections
        # are increased, therefore we disable hot_standby if recovery_target_action == 'promote'.
        if self._postgresql.bootstrap.running_custom_bootstrap:
            disable_hot_standby = False
            if self._postgresql.bootstrap.keep_existing_recovery_conf:
                disable_hot_standby = True  # trust that pgBackRest does the right thing
            # `pause_at_recovery_target` has no effect if hot_standby is not enabled, therefore we consider only 9.5+
            elif self._postgresql.major_version >= 90500 and self._recovery_params:
                pause_at_recovery_target = parse_bool(self._recovery_params.get('pause_at_recovery_target'))
                recovery_target_action = self._recovery_params.get(
                    'recovery_target_action', 'promote' if pause_at_recovery_target is False else 'pause')
                disable_hot_standby = recovery_target_action == 'promote'

            if disable_hot_standby:
                effective_configuration['hot_standby'] = 'off'

        return effective_configuration

    @property
    def replication(self) -> Dict[str, Any]:
        return self._config['authentication']['replication']

    @property
    def superuser(self) -> Dict[str, Any]:
        return self._superuser

    @property
    def rewind_credentials(self) -> Dict[str, Any]:
        return self._config['authentication'].get('rewind', self._superuser) \
            if self._postgresql.major_version >= 110000 else self._superuser

    @property
    def ident_file(self) -> Optional[str]:
        ident_file = self._server_parameters.get('ident_file')
        return None if ident_file == self._pg_ident_conf else ident_file

    @property
    def hba_file(self) -> Optional[str]:
        hba_file = self._server_parameters.get('hba_file')
        return None if hba_file == self._pg_hba_conf else hba_file

    @property
    def pg_hba_conf(self) -> str:
        return self._pg_hba_conf

    @property
    def postgresql_conf(self) -> str:
        return self._postgresql_conf

    def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
        return self._config.get(key, default)

    def restore_command(self) -> Optional[str]:
        return (self.get('recovery_conf') or {}).get('restore_command')