mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Alexander KukushkinandGitHub
4c3af2d1a0
keep as much backward compatibility as possible.
Following changes were made:
1. All internal checks are performed as `role in ('master', 'primary')`
2. All internal variables/functions/methods are renamed
3. `GET /metrics` endpoint returns `patroni_primary` in addition to `patroni_master`.
4. Logs are changed to use leader/primary/member/remote depending on the context
5. Unit-tests are using only role = 'primary' instead of 'master' to verify that 1 works.
6. patronictl still supports old syntax, but also accepts `--leader` and `--primary`.
7. `master_(start|stop)_timeout` is automatically translated to `primary_(start|stop)_timeout` if the last one is not set.
8. updated the documentation and some examples
Future plan: in the next major release switch role name from `master` to `primary` and maybe drop `master` altogether.
The Kubernetes implementation will require more work and keep two labels in parallel. Label values should probably be configurable as described in https://github.com/zalando/patroni/issues/2495.
1139 lines
53 KiB
Python
1139 lines
53 KiB
Python
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import stat
|
|
import time
|
|
|
|
from six.moves.urllib_parse import urlparse, parse_qsl, unquote
|
|
|
|
from .validator import CaseInsensitiveDict, recovery_parameters,\
|
|
transform_postgresql_parameter_value, transform_recovery_parameter_value
|
|
from ..dcs import slot_name_from_member_name, RemoteMember
|
|
from ..exceptions import PatroniFatalException
|
|
from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \
|
|
validate_directory, is_subpath
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PARAMETER_RE = re.compile(r'([a-z_]+)\s*=\s*')
|
|
|
|
|
|
def conninfo_uri_parse(dsn):
|
|
ret = {}
|
|
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 = []
|
|
ports = []
|
|
for netloc in r.netloc.split('@')[-1].split(','):
|
|
host = port = None
|
|
if '[' in netloc and ']' in netloc:
|
|
host = netloc.split(']')[0][1:]
|
|
tmp = netloc.split(':', 1)
|
|
if host is None:
|
|
host = tmp[0]
|
|
if len(tmp) == 2:
|
|
host, port = tmp
|
|
if host is not None:
|
|
hosts.append(host)
|
|
if port is not None:
|
|
ports.append(port)
|
|
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):
|
|
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):
|
|
ret = {}
|
|
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:
|
|
return
|
|
i += end
|
|
ret[param] = value
|
|
return ret
|
|
|
|
|
|
def parse_dsn(value):
|
|
"""
|
|
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,[%2Fhost2]/db%2Fsdf?application_name=mya%2Fpp&ssl=true')
|
|
>>> r == {'application_name': 'mya/pp', 'host': ',/host2', '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):
|
|
i = value.find('#')
|
|
if i > -1:
|
|
value = value[:i].strip()
|
|
return value
|
|
|
|
|
|
def read_recovery_param_value(value):
|
|
"""
|
|
>>> 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):
|
|
try:
|
|
return os.stat(filename).st_mtime
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
class ConfigWriter(object):
|
|
|
|
def __init__(self, filename):
|
|
self._filename = filename
|
|
self._fd = None
|
|
|
|
def __enter__(self):
|
|
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, exc_val, exc_tb):
|
|
if self._fd:
|
|
self._fd.close()
|
|
|
|
def writeline(self, line):
|
|
self._fd.write(line)
|
|
self._fd.write('\n')
|
|
|
|
def writelines(self, lines):
|
|
for line in lines:
|
|
self.writeline(line)
|
|
|
|
@staticmethod
|
|
def escape(value): # 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, value):
|
|
self.writeline("{0} = '{1}'".format(param, self.escape(value)))
|
|
|
|
|
|
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, lambda _: False, 90100),
|
|
'port': (None, lambda _: False, 90100),
|
|
'cluster_name': (None, lambda _: False, 90500),
|
|
'wal_level': ('hot_standby', lambda v: v.lower() in ('hot_standby', 'replica', 'logical'), 90100),
|
|
'hot_standby': ('on', lambda _: False, 90100),
|
|
'max_connections': (100, lambda v: int(v) >= 25, 90100),
|
|
'max_wal_senders': (10, lambda v: int(v) >= 3, 90100),
|
|
'wal_keep_segments': (8, lambda v: int(v) >= 1, 90100),
|
|
'wal_keep_size': ('128MB', lambda v: parse_int(v, 'MB') >= 16, 130000),
|
|
'max_prepared_transactions': (0, lambda v: int(v) >= 0, 90100),
|
|
'max_locks_per_transaction': (64, lambda v: int(v) >= 32, 90100),
|
|
'track_commit_timestamp': ('off', lambda v: parse_bool(v) is not None, 90500),
|
|
'max_replication_slots': (10, lambda v: int(v) >= 4, 90400),
|
|
'max_worker_processes': (8, lambda v: int(v) >= 2, 90400),
|
|
'wal_log_hints': ('on', lambda _: False, 90400)
|
|
})
|
|
|
|
_RECOVERY_PARAMETERS = set(recovery_parameters.keys())
|
|
|
|
def __init__(self, postgresql, config):
|
|
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._synchronous_standby_names = None
|
|
self._postmaster_ctime = None
|
|
self._current_recovery_params = None
|
|
self._config = {}
|
|
self._recovery_params = {}
|
|
self.reload_config(config)
|
|
|
|
def setup_server_parameters(self):
|
|
self._server_parameters = self.get_server_parameters(self._config)
|
|
self._adjust_recovery_parameters()
|
|
|
|
def try_to_create_dir(self, d, msg):
|
|
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):
|
|
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 _configuration_to_save(self):
|
|
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 save_configuration_files(self, check_custom_bootstrap=False):
|
|
"""
|
|
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)
|
|
except IOError:
|
|
logger.exception('unable to create backup copies of configuration files')
|
|
return True
|
|
|
|
def restore_configuration_files(self):
|
|
""" 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)
|
|
# 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()
|
|
except IOError:
|
|
logger.exception('unable to restore configuration files from backup')
|
|
|
|
def write_postgresql_conf(self, configuration=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 ConfigWriter(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)
|
|
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):
|
|
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)))
|
|
return True
|
|
|
|
def replace_pg_hba(self):
|
|
"""
|
|
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 ConfigWriter(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 ConfigWriter(self._pg_hba_conf) as f:
|
|
f.writelines(self._config['pg_hba'])
|
|
return True
|
|
|
|
def replace_pg_ident(self):
|
|
"""
|
|
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 ConfigWriter(self._pg_ident_conf) as f:
|
|
f.writelines(self._config['pg_ident'])
|
|
return True
|
|
|
|
def primary_conninfo_params(self, member):
|
|
if not (member and 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, include_dbname=False):
|
|
# 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):
|
|
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, recovery_params):
|
|
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)
|
|
if value is None:
|
|
continue
|
|
fd.write_param(name, value)
|
|
|
|
def build_recovery_params(self, member):
|
|
recovery_params = CaseInsensitiveDict({p: v for p, v in self.get('recovery_conf', {}).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 = self.get('use_slots', True) 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):
|
|
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):
|
|
return 'trigger_file' if self._postgresql.major_version < 120000 else 'promote_trigger_file'
|
|
|
|
@property
|
|
def _triggerfile_wrong_name(self):
|
|
return 'trigger_file' if self._postgresql.major_version >= 120000 else 'promote_trigger_file'
|
|
|
|
@property
|
|
def _recovery_parameters_to_compare(self):
|
|
skip_params = {'pause_at_recovery_target', 'recovery_target_inclusive',
|
|
'recovery_target_action', 'standby_mode', self._triggerfile_wrong_name}
|
|
return self._RECOVERY_PARAMETERS - skip_params
|
|
|
|
def _read_recovery_params(self):
|
|
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 = {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:
|
|
values = None
|
|
return values, True
|
|
|
|
def _read_recovery_params_pre_v12(self):
|
|
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 = {}
|
|
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 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, wanted_primary_conninfo):
|
|
# 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).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, wanted_primary_conninfo):
|
|
# 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):
|
|
"""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():
|
|
# 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):
|
|
required['restart' if mtype else 'reload'] += 1
|
|
|
|
wanted_recovery_params = self.build_recovery_params(member)
|
|
for param, value in self._current_recovery_params.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):
|
|
if os.path.isfile(name) or os.path.islink(name):
|
|
os.unlink(name)
|
|
|
|
@staticmethod
|
|
def _pgpass_line(record):
|
|
if 'password' in record:
|
|
def escape(value):
|
|
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.get('host').split(','):
|
|
line += hostname + ':{port}:*:{user}:{password}'.format(**record) + '\n'
|
|
return line.rstrip()
|
|
|
|
def write_pgpass(self, record):
|
|
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)
|
|
|
|
env = os.environ.copy()
|
|
env['PGPASSFILE'] = self._pgpass
|
|
return env
|
|
|
|
def write_recovery_conf(self, recovery_params):
|
|
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()
|
|
else:
|
|
self._remove_file_if_exists(self._standby_signal)
|
|
open(self._recovery_signal, 'w').close()
|
|
|
|
def restart_required(name):
|
|
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 = {n: [v, restart_required(n), self._postgresql_conf]
|
|
for n, v in recovery_params.items()}
|
|
else:
|
|
with ConfigWriter(self._recovery_conf) as f:
|
|
os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
|
|
self._write_recovery_params(f, recovery_params)
|
|
|
|
def remove_recovery_conf(self):
|
|
for name in (self._recovery_conf, self._standby_signal, self._recovery_signal):
|
|
self._remove_file_if_exists(name)
|
|
self._recovery_params = {}
|
|
self._current_recovery_params = None
|
|
|
|
def _sanitize_auto_conf(self):
|
|
overwrite = False
|
|
lines = []
|
|
|
|
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:
|
|
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):
|
|
# 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):
|
|
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 config.get('synchronous_mode', False):
|
|
if self._synchronous_standby_names is None:
|
|
if config.get('synchronous_mode_strict', False):
|
|
parameters['synchronous_standby_names'] = '*'
|
|
else:
|
|
parameters.pop('synchronous_standby_names', None)
|
|
else:
|
|
parameters['synchronous_standby_names'] = self._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 + 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):
|
|
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):
|
|
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
|
|
|
|
@property
|
|
def local_connect_kwargs(self):
|
|
ret = self._local_address.copy()
|
|
# add all of the other connection settings that are available
|
|
ret.update(self._superuser)
|
|
# if the "username" parameter is present, it actually needs to be "user"
|
|
# for connecting to PostgreSQL
|
|
if 'username' in self._superuser:
|
|
ret['user'] = self._superuser['username']
|
|
del ret['username']
|
|
# ensure certain Patroni configurations are available
|
|
ret.update({'dbname': self._postgresql.database,
|
|
'fallback_application_name': 'Patroni',
|
|
'connect_timeout': 3,
|
|
'options': '-c statement_timeout=2000'})
|
|
return ret
|
|
|
|
def resolve_connection_addresses(self):
|
|
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_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address
|
|
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)
|
|
self._postgresql.set_connection_kwargs(self.local_connect_kwargs)
|
|
|
|
def _get_pg_settings(self, names):
|
|
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, changes):
|
|
wal_block_size = parse_int(old_values['wal_block_size'][1])
|
|
wal_segment_size = old_values['wal_segment_size']
|
|
wal_segment_unit = parse_int(wal_segment_size[2], 'B') if wal_segment_size[2][0].isdigit() else 1
|
|
wal_segment_size = parse_int(wal_segment_size[1]) * wal_segment_unit / wal_block_size
|
|
default_wal_buffers = min(max(parse_int(old_values['shared_buffers'][1]) / 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, sighup=False):
|
|
self._superuser = config['authentication'].get('superuser', {})
|
|
server_parameters = self.get_server_parameters(config)
|
|
|
|
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.lower() not in self._RECOVERY_PARAMETERS})
|
|
changes.update({p: None for p in self._server_parameters.keys()
|
|
if not (p in changes or p.lower() in self._RECOVERY_PARAMETERS)})
|
|
if changes:
|
|
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)
|
|
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 self._RECOVERY_PARAMETERS]).fetchone()[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):
|
|
"""Updates synchronous_standby_names and reloads if necessary.
|
|
:returns: True if value was updated."""
|
|
if value != self._synchronous_standby_names:
|
|
if value is None:
|
|
self._server_parameters.pop('synchronous_standby_names', None)
|
|
else:
|
|
self._server_parameters['synchronous_standby_names'] = value
|
|
self._synchronous_standby_names = value
|
|
if self._postgresql.state == 'running':
|
|
self.write_postgresql_conf()
|
|
self._postgresql.reload()
|
|
return True
|
|
|
|
@property
|
|
def effective_configuration(self):
|
|
"""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 > 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 self._postgresql.bootstrap.running_custom_bootstrap and \
|
|
(self._postgresql.bootstrap.keep_existing_recovery_conf or self._recovery_conf):
|
|
effective_configuration['hot_standby'] = 'off'
|
|
self._postgresql.set_pending_restart(True)
|
|
|
|
return effective_configuration
|
|
|
|
@property
|
|
def replication(self):
|
|
return self._config['authentication']['replication']
|
|
|
|
@property
|
|
def superuser(self):
|
|
return self._superuser
|
|
|
|
@property
|
|
def rewind_credentials(self):
|
|
return self._config['authentication'].get('rewind', self._superuser) \
|
|
if self._postgresql.major_version >= 110000 else self._superuser
|
|
|
|
@property
|
|
def ident_file(self):
|
|
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):
|
|
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):
|
|
return self._pg_hba_conf
|
|
|
|
@property
|
|
def postgresql_conf(self):
|
|
return self._postgresql_conf
|
|
|
|
def get(self, key, default=None):
|
|
return self._config.get(key, default)
|