mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-27 16:10:10 +00:00
Alexander KukushkinandGitHub
e684ca66e5
use replication=1 instead of replication='database' when opening replication connection, 9.3 doesn't understand it
1968 lines
87 KiB
Python
1968 lines
87 KiB
Python
import logging
|
|
import os
|
|
import psycopg2
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
|
|
from collections import defaultdict
|
|
from contextlib import contextmanager
|
|
from patroni.callback_executor import CallbackExecutor
|
|
from patroni.exceptions import PostgresConnectionException, PostgresException
|
|
from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop, split_host_port
|
|
from patroni.postmaster import PostmasterProcess
|
|
from patroni.dcs import slot_name_from_member_name, RemoteMember
|
|
from requests.structures import CaseInsensitiveDict
|
|
from six import string_types
|
|
from six.moves.urllib.parse import quote_plus
|
|
from threading import current_thread, Lock
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ACTION_ON_START = "on_start"
|
|
ACTION_ON_STOP = "on_stop"
|
|
ACTION_ON_RESTART = "on_restart"
|
|
ACTION_ON_RELOAD = "on_reload"
|
|
ACTION_ON_ROLE_CHANGE = "on_role_change"
|
|
|
|
STATE_RUNNING = 'running'
|
|
STATE_REJECT = 'rejecting connections'
|
|
STATE_NO_RESPONSE = 'not responding'
|
|
STATE_UNKNOWN = 'unknown'
|
|
|
|
STOP_POLLING_INTERVAL = 1
|
|
REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5})
|
|
sync_standby_name_re = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$')
|
|
|
|
cluster_info_query = ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
|
|
"ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name("
|
|
"pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, "
|
|
"CASE WHEN pg_catalog.pg_is_in_recovery() THEN GREATEST("
|
|
" pg_catalog.pg_{0}_{1}_diff(COALESCE("
|
|
"pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint,"
|
|
" pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint)"
|
|
"ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END")
|
|
|
|
|
|
def quote_ident(value):
|
|
"""Very simplified version of quote_ident"""
|
|
return value if sync_standby_name_re.match(value) else '"' + value + '"'
|
|
|
|
|
|
@contextmanager
|
|
def null_context():
|
|
yield
|
|
|
|
|
|
class Postgresql(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) >= 100, 90100),
|
|
'max_wal_senders': (10, lambda v: int(v) >= 10, 90100),
|
|
'wal_keep_segments': (8, lambda v: int(v) >= 8, 90100),
|
|
'max_prepared_transactions': (0, lambda v: int(v) >= 0, 90100),
|
|
'max_locks_per_transaction': (64, lambda v: int(v) >= 64, 90100),
|
|
'track_commit_timestamp': ('off', lambda v: parse_bool(v) is not None, 90500),
|
|
'max_replication_slots': (10, lambda v: int(v) >= 10, 90400),
|
|
'max_worker_processes': (8, lambda v: int(v) >= 8, 90400),
|
|
'wal_log_hints': ('on', lambda _: False, 90400)
|
|
})
|
|
|
|
_CONFIG_WARNING_HEADER = '# Do not edit this file manually!\n# It will be overwritten by Patroni!\n'
|
|
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.name = config['name']
|
|
self.scope = config['scope']
|
|
self._bin_dir = config.get('bin_dir') or ''
|
|
self._database = config.get('database', 'postgres')
|
|
self._data_dir = config['data_dir']
|
|
self._config_dir = os.path.abspath(config.get('config_dir') or self._data_dir)
|
|
self._pending_restart = False
|
|
self.bootstrapping = False
|
|
self._running_custom_bootstrap = False
|
|
self.__thread_ident = current_thread().ident
|
|
|
|
self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
|
|
self._synchronous_standby_names = None
|
|
self._configure_server_parameters()
|
|
|
|
self._connect_address = config.get('connect_address')
|
|
self._superuser = config['authentication'].get('superuser', {})
|
|
self.resolve_connection_addresses()
|
|
|
|
self._rewind_state = REWIND_STATUS.INITIAL
|
|
self._use_slots = config.get('use_slots', True)
|
|
self._schedule_load_slots = self.use_slots
|
|
|
|
self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
|
|
self._callback_executor = CallbackExecutor()
|
|
self.__cb_called = False
|
|
self.__cb_pending = None
|
|
config_base_name = config.get('config_base_name', 'postgresql')
|
|
self._postgresql_conf = os.path.join(self._config_dir, config_base_name + '.conf')
|
|
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._recovery_conf = os.path.join(self._data_dir, 'recovery.conf')
|
|
self._trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote'
|
|
self._trigger_file = os.path.abspath(os.path.join(self._data_dir, self._trigger_file))
|
|
|
|
self._is_cancelled = False
|
|
self._cancellable = None
|
|
self._cancellable_lock = Lock()
|
|
|
|
self._connection_lock = Lock()
|
|
self._connection = None
|
|
self._cursor_holder = None
|
|
self._sysid = None
|
|
self._replication_slots = {} # already existing replication slots
|
|
self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1,
|
|
retry_exceptions=PostgresConnectionException)
|
|
|
|
# Retry 'pg_is_in_recovery()' only once
|
|
self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout']/2.0, max_delay=1,
|
|
retry_exceptions=PostgresConnectionException)
|
|
|
|
self._state_lock = Lock()
|
|
self.set_state('stopped')
|
|
self._role_lock = Lock()
|
|
self.set_role(self.get_postgres_role_from_data_directory())
|
|
|
|
self._state_entry_timestamp = None
|
|
|
|
self._cluster_info_state = {}
|
|
self._cached_replica_timeline = None
|
|
|
|
# Last known running process
|
|
self._postmaster_proc = None
|
|
|
|
if self.is_running():
|
|
self.set_state('running')
|
|
self.set_role('master' if self.is_leader() else 'replica')
|
|
self._write_postgresql_conf() # we are "joining" already running postgres
|
|
if self._replace_pg_hba():
|
|
self.reload()
|
|
elif self.role == 'master':
|
|
self.set_role('demoted')
|
|
|
|
@property
|
|
def _create_replica_methods(self):
|
|
return (self.config.get('create_replica_methods', []) or
|
|
self.config.get('create_replica_method', []))
|
|
|
|
@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))
|
|
if not self._server_parameters.get('hba_file'):
|
|
configuration.append('pg_hba.conf')
|
|
if not self._server_parameters.get('ident_file'):
|
|
configuration.append('pg_ident.conf')
|
|
return configuration
|
|
|
|
@property
|
|
def use_slots(self):
|
|
return self._use_slots and self._major_version >= 90400
|
|
|
|
@property
|
|
def _replication(self):
|
|
return self.config['authentication']['replication']
|
|
|
|
@property
|
|
def callback(self):
|
|
return self.config.get('callbacks') or {}
|
|
|
|
@staticmethod
|
|
def _wal_name(version):
|
|
return 'wal' if version >= 100000 else 'xlog'
|
|
|
|
@property
|
|
def wal_name(self):
|
|
return self._wal_name(self._major_version)
|
|
|
|
@property
|
|
def lsn_name(self):
|
|
return 'lsn' if self._major_version >= 100000 else 'location'
|
|
|
|
def _version_file_exists(self):
|
|
return not self.data_directory_empty() and os.path.isfile(self._version_file)
|
|
|
|
def get_major_version(self):
|
|
if self._version_file_exists():
|
|
try:
|
|
with open(self._version_file) as f:
|
|
return self.postgres_major_version_to_int(f.read().strip())
|
|
except Exception:
|
|
logger.exception('Failed to read PG_VERSION from %s', self._data_dir)
|
|
return 0
|
|
|
|
def get_server_parameters(self, config):
|
|
parameters = config['parameters'].copy()
|
|
listen_addresses, port = split_host_port(config['listen'], 5432)
|
|
parameters.update({'cluster_name': self.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
|
|
if self._major_version >= 90600 and parameters['wal_level'] == 'hot_standby':
|
|
parameters['wal_level'] = 'replica'
|
|
ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._major_version or
|
|
self._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
|
|
|
|
def resolve_connection_addresses(self):
|
|
port = self._server_parameters['port']
|
|
tcp_local_address = self._get_tcp_local_address()
|
|
|
|
local_address = {'port': port}
|
|
if self.config.get('use_unix_socket'):
|
|
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 sutable values
|
|
local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address
|
|
|
|
# if unix_socket_directories is not specified, but use_unix_socket is set to true - do our best
|
|
# to use default value, i.e. don't specify a host neither in connection url nor arguments
|
|
else:
|
|
local_address['host'] = tcp_local_address
|
|
|
|
self._local_address = local_address
|
|
self._local_replication_address = {'host': tcp_local_address, 'port': port}
|
|
|
|
self.connection_string = 'postgres://{0}/{1}'.format(
|
|
self._connect_address or tcp_local_address + ':' + port, self._database)
|
|
|
|
def _pgcommand(self, cmd):
|
|
"""Returns path to the specified PostgreSQL command"""
|
|
return os.path.join(self._bin_dir, cmd)
|
|
|
|
def pg_ctl(self, cmd, *args, **kwargs):
|
|
"""Builds and executes pg_ctl command
|
|
|
|
:returns: `!True` when return_code == 0, otherwise `!False`"""
|
|
|
|
pg_ctl = [self._pgcommand('pg_ctl'), cmd]
|
|
return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0
|
|
|
|
def pg_isready(self):
|
|
"""Runs pg_isready to see if PostgreSQL is accepting connections.
|
|
|
|
:returns: 'ok' if PostgreSQL is up, 'reject' if starting up, 'no_resopnse' if not up."""
|
|
|
|
cmd = [self._pgcommand('pg_isready'), '-p', self._local_address['port'], '-d', self._database]
|
|
|
|
# Host is not set if we are connecting via default unix socket
|
|
if 'host' in self._local_address:
|
|
cmd.extend(['-h', self._local_address['host']])
|
|
|
|
# We only need the username because pg_isready does not try to authenticate
|
|
if 'username' in self._superuser:
|
|
cmd.extend(['-U', self._superuser['username']])
|
|
|
|
ret = subprocess.call(cmd)
|
|
return_codes = {0: STATE_RUNNING,
|
|
1: STATE_REJECT,
|
|
2: STATE_NO_RESPONSE,
|
|
3: STATE_UNKNOWN}
|
|
return return_codes.get(ret, STATE_UNKNOWN)
|
|
|
|
def reload_config(self, config):
|
|
self._superuser = config['authentication'].get('superuser', {})
|
|
server_parameters = self.get_server_parameters(config)
|
|
|
|
conf_changed = hba_changed = local_connection_address_changed = pending_restart = False
|
|
if self.state == 'running':
|
|
changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items() if '.' not in p})
|
|
changes.update({p: None for p in self._server_parameters.keys() if not ('.' in p or p in changes)})
|
|
if changes:
|
|
if 'wal_segment_size' not in changes:
|
|
changes['wal_segment_size'] = '16384kB'
|
|
# XXX: query can raise an exception
|
|
for r in self.query("""SELECT name, setting, unit, vartype, context
|
|
FROM pg_catalog.pg_settings
|
|
WHERE pg_catalog.lower(name) IN (""" + ', '.join(['%s'] * len(changes)) + """)
|
|
ORDER BY 1 DESC""", *(k.lower() for k in changes.keys())):
|
|
if r[4] == 'internal':
|
|
if r[0] == 'wal_segment_size':
|
|
server_parameters.pop(r[0], None)
|
|
wal_segment_size = parse_int(r[2], 'kB')
|
|
if wal_segment_size is not None:
|
|
changes['wal_segment_size'] = '{0}kB'.format(int(r[1]) * wal_segment_size)
|
|
elif r[0] in changes:
|
|
unit = changes['wal_segment_size'] if r[0] in ('min_wal_size', 'max_wal_size') else r[2]
|
|
new_value = changes.pop(r[0])
|
|
if new_value is None or not compare_values(r[3], unit, r[1], new_value):
|
|
if r[4] == 'postmaster':
|
|
pending_restart = True
|
|
logger.info('Changed %s from %s to %s (restart 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)
|
|
conf_changed = True
|
|
for param in changes:
|
|
if param in server_parameters:
|
|
logger.warning('Removing invalid parameter `%s` from postgresql.parameters', param)
|
|
server_parameters.pop(param)
|
|
|
|
# Check that user-defined-paramters have changed (parameters with period in name)
|
|
if not conf_changed:
|
|
for p, v in server_parameters.items():
|
|
if '.' in p and (p not in self._server_parameters or str(v) != str(self._server_parameters[p])):
|
|
logger.info('Changed %s from %s to %s', p, self._server_parameters.get(p), v)
|
|
conf_changed = True
|
|
break
|
|
if not conf_changed:
|
|
for p, v in self._server_parameters.items():
|
|
if '.' in p and (p not in server_parameters or str(v) != str(server_parameters[p])):
|
|
logger.info('Changed %s from %s to %s', p, v, server_parameters.get(p))
|
|
conf_changed = True
|
|
break
|
|
|
|
if not server_parameters.get('hba_file') and config.get('pg_hba'):
|
|
hba_changed = self.config.get('pg_hba', []) != config['pg_hba']
|
|
|
|
self.config = config
|
|
self._pending_restart = pending_restart
|
|
self._server_parameters = server_parameters
|
|
self._connect_address = config.get('connect_address')
|
|
|
|
if not local_connection_address_changed:
|
|
self.resolve_connection_addresses()
|
|
|
|
if conf_changed:
|
|
self._write_postgresql_conf()
|
|
|
|
if hba_changed:
|
|
self._replace_pg_hba()
|
|
|
|
if conf_changed or hba_changed:
|
|
logger.info('PostgreSQL configuration items changed, reloading configuration.')
|
|
self.reload()
|
|
elif not pending_restart:
|
|
logger.info('No PostgreSQL configuration items changed, nothing to reload.')
|
|
|
|
self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0
|
|
|
|
@property
|
|
def pending_restart(self):
|
|
return self._pending_restart
|
|
|
|
@staticmethod
|
|
def configuration_allows_rewind(data):
|
|
return data.get('wal_log_hints setting', 'off') == 'on' \
|
|
or data.get('Data page checksum version', '0') != '0'
|
|
|
|
@property
|
|
def can_rewind(self):
|
|
""" check if pg_rewind executable is there and that pg_controldata indicates
|
|
we have either wal_log_hints or checksums turned on
|
|
"""
|
|
# low-hanging fruit: check if pg_rewind configuration is there
|
|
if not (self.config.get('use_pg_rewind') and all(self._superuser.get(n) for n in ('username', 'password'))):
|
|
return False
|
|
|
|
cmd = [self._pgcommand('pg_rewind'), '--help']
|
|
try:
|
|
ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
|
|
if ret != 0: # pg_rewind is not there, close up the shop and go home
|
|
return False
|
|
except OSError:
|
|
return False
|
|
return self.configuration_allows_rewind(self.controldata())
|
|
|
|
@property
|
|
def sysid(self):
|
|
if not self._sysid and not self.bootstrapping:
|
|
data = self.controldata()
|
|
self._sysid = data.get('Database system identifier', "")
|
|
return self._sysid
|
|
|
|
@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
|
|
|
|
def get_postgres_role_from_data_directory(self):
|
|
if self.data_directory_empty():
|
|
return 'uninitialized'
|
|
elif os.path.exists(self._recovery_conf):
|
|
return 'replica'
|
|
else:
|
|
return 'master'
|
|
|
|
@property
|
|
def _local_connect_kwargs(self):
|
|
ret = self._local_address.copy()
|
|
ret.update({'database': self._database,
|
|
'fallback_application_name': 'Patroni',
|
|
'connect_timeout': 3,
|
|
'options': '-c statement_timeout=2000'})
|
|
if 'username' in self._superuser:
|
|
ret['user'] = self._superuser['username']
|
|
if 'password' in self._superuser:
|
|
ret['password'] = self._superuser['password']
|
|
return ret
|
|
|
|
def connection(self):
|
|
with self._connection_lock:
|
|
if not self._connection or self._connection.closed != 0:
|
|
self._connection = psycopg2.connect(**self._local_connect_kwargs)
|
|
self._connection.autocommit = True
|
|
self.server_version = self._connection.server_version
|
|
return self._connection
|
|
|
|
def _cursor(self):
|
|
if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0:
|
|
logger.info("establishing a new patroni connection to the postgres cluster")
|
|
self._cursor_holder = self.connection().cursor()
|
|
return self._cursor_holder
|
|
|
|
def close_connection(self):
|
|
if self._connection and self._connection.closed == 0:
|
|
self._connection.close()
|
|
logger.info("closed patroni connection to the postgresql cluster")
|
|
self._cursor_holder = self._connection = None
|
|
|
|
def _query(self, sql, *params):
|
|
"""We are always using the same cursor, therefore this method is not thread-safe!!!
|
|
You can call it from different threads only if you are holding explicit `AsyncExecutor` lock,
|
|
because the main thread is always holding this lock when running HA cycle."""
|
|
cursor = None
|
|
try:
|
|
cursor = self._cursor()
|
|
cursor.execute(sql, params)
|
|
return cursor
|
|
except psycopg2.Error as e:
|
|
if cursor and cursor.connection.closed == 0:
|
|
# When connected via unix socket, psycopg2 can't recoginze 'connection lost'
|
|
# and leaves `_cursor_holder.connection.closed == 0`, but psycopg2.OperationalError
|
|
# is still raised (what is correct). It doesn't make sense to continiue with existing
|
|
# connection and we will close it, to avoid its reuse by the `_cursor` method.
|
|
if isinstance(e, psycopg2.OperationalError):
|
|
self.close_connection()
|
|
else:
|
|
raise e
|
|
if self.state == 'restarting':
|
|
raise RetryFailedError('cluster is being restarted')
|
|
raise PostgresConnectionException('connection problems')
|
|
|
|
def query(self, sql, *params):
|
|
try:
|
|
return self.retry(self._query, sql, *params)
|
|
except RetryFailedError as e:
|
|
raise PostgresConnectionException(str(e))
|
|
|
|
def data_directory_empty(self):
|
|
return not os.path.exists(self._data_dir) or os.listdir(self._data_dir) == []
|
|
|
|
@staticmethod
|
|
def process_user_options(tool, options, not_allowed_options, error_handler):
|
|
user_options = []
|
|
|
|
def option_is_allowed(name):
|
|
ret = name not in not_allowed_options
|
|
if not ret:
|
|
error_handler('{0} option for {1} is not allowed'.format(name, tool))
|
|
return ret
|
|
|
|
if isinstance(options, dict):
|
|
for k, v in options.items():
|
|
if k and v:
|
|
user_options.append('--{0}={1}'.format(k, v))
|
|
elif isinstance(options, list):
|
|
for opt in options:
|
|
if isinstance(opt, string_types) and option_is_allowed(opt):
|
|
user_options.append('--{0}'.format(opt))
|
|
elif isinstance(opt, dict):
|
|
keys = list(opt.keys())
|
|
if len(keys) != 1 or not isinstance(opt[keys[0]], string_types) or not option_is_allowed(keys[0]):
|
|
error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed'
|
|
' and value should be a string'.format(tool, opt[keys[0]]))
|
|
user_options.append('--{0}={1}'.format(keys[0], opt[keys[0]]))
|
|
else:
|
|
error_handler('Error when parsing {0} option {1}: value should be string value'
|
|
' or a single key-value pair'.format(tool, opt))
|
|
else:
|
|
error_handler('{0} options must be list ot dict'.format(tool))
|
|
return user_options
|
|
|
|
def _initdb(self, config):
|
|
self.set_state('initalizing new cluster')
|
|
not_allowed_options = ('pgdata', 'nosync', 'pwfile', 'sync-only', 'version')
|
|
|
|
def error_handler(e):
|
|
raise Exception(e)
|
|
|
|
options = self.process_user_options('initdb', config.get('initdb') or [], not_allowed_options, error_handler)
|
|
pwfile = None
|
|
|
|
if self._superuser:
|
|
if 'username' in self._superuser:
|
|
options.append('--username={0}'.format(self._superuser['username']))
|
|
if 'password' in self._superuser:
|
|
(fd, pwfile) = tempfile.mkstemp()
|
|
os.write(fd, self._superuser['password'].encode('utf-8'))
|
|
os.close(fd)
|
|
options.append('--pwfile={0}'.format(pwfile))
|
|
options = ['-o', ' '.join(options)] if options else []
|
|
|
|
ret = self.pg_ctl('initdb', *options)
|
|
if pwfile:
|
|
os.remove(pwfile)
|
|
if not ret:
|
|
self.set_state('initdb failed')
|
|
return ret
|
|
|
|
def _custom_bootstrap(self, config):
|
|
self.set_state('running custom bootstrap script')
|
|
params = ['--scope=' + self.scope, '--datadir=' + self._data_dir]
|
|
try:
|
|
logger.info('Running custom bootstrap script: %s', config['command'])
|
|
if self.cancellable_subprocess_call(shlex.split(config['command']) + params) != 0:
|
|
self.set_state('custom bootstrap failed')
|
|
return False
|
|
except Exception:
|
|
logger.exception('Exception during custom bootstrap')
|
|
return False
|
|
self._post_restore()
|
|
|
|
if 'recovery_conf' in config:
|
|
self.write_recovery_conf(config['recovery_conf'])
|
|
elif (os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf)) and \
|
|
not config.get('keep_existing_recovery_conf'):
|
|
os.unlink(self._recovery_conf)
|
|
return True
|
|
|
|
def run_bootstrap_post_init(self, config):
|
|
"""
|
|
runs a script after initdb or custom bootstrap script is called and waits until completion.
|
|
"""
|
|
cmd = config.get('post_bootstrap') or config.get('post_init')
|
|
if cmd:
|
|
r = self._local_connect_kwargs
|
|
|
|
if 'host' in r:
|
|
# '/tmp' => '%2Ftmp' for unix socket path
|
|
host = quote_plus(r['host']) if r['host'].startswith('/') else r['host']
|
|
else:
|
|
host = ''
|
|
|
|
# https://www.postgresql.org/docs/current/static/libpq-pgpass.html
|
|
# A host name of localhost matches both TCP (host name localhost) and Unix domain socket
|
|
# (pghost empty or the default socket directory) connections coming from the local machine.
|
|
r['host'] = 'localhost' # set it to localhost to write into pgpass
|
|
|
|
if 'user' in r:
|
|
user = r['user'] + '@'
|
|
else:
|
|
user = ''
|
|
if 'password' in r:
|
|
import getpass
|
|
r.setdefault('user', os.environ.get('PGUSER', getpass.getuser()))
|
|
|
|
connstring = 'postgres://{0}{1}:{2}/{3}'.format(user, host, r['port'], r['database'])
|
|
env = self.write_pgpass(r) if 'password' in r else None
|
|
|
|
try:
|
|
ret = self.cancellable_subprocess_call(shlex.split(cmd) + [connstring], env=env)
|
|
except OSError:
|
|
logger.error('post_init script %s failed', cmd)
|
|
return False
|
|
if ret != 0:
|
|
logger.error('post_init script %s returned non-zero code %d', cmd, ret)
|
|
return False
|
|
return True
|
|
|
|
def delete_trigger_file(self):
|
|
if os.path.exists(self._trigger_file):
|
|
os.unlink(self._trigger_file)
|
|
|
|
def write_pgpass(self, record):
|
|
if 'user' not in record or 'password' not in record:
|
|
return os.environ.copy()
|
|
|
|
with open(self._pgpass, 'w') as f:
|
|
if os.name != 'nt':
|
|
os.fchmod(f.fileno(), 0o600)
|
|
f.write('{host}:{port}:*:{user}:{password}\n'.format(**record))
|
|
|
|
env = os.environ.copy()
|
|
env['PGPASSFILE'] = self._pgpass
|
|
return env
|
|
|
|
def replica_method_can_work_without_replication_connection(self, method):
|
|
return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master')
|
|
|
|
def can_create_replica_without_replication_connection(self):
|
|
""" go through the replication methods to see if there are ones
|
|
that does not require a working replication connection.
|
|
"""
|
|
replica_methods = self._create_replica_methods
|
|
return any(self.replica_method_can_work_without_replication_connection(method) for method in replica_methods)
|
|
|
|
def create_replica(self, clone_member):
|
|
"""
|
|
create the replica according to the replica_method
|
|
defined by the user. this is a list, so we need to
|
|
loop through all methods the user supplies
|
|
"""
|
|
|
|
self.set_state('creating replica')
|
|
self._sysid = None
|
|
|
|
is_remote_master = isinstance(clone_member, RemoteMember)
|
|
create_replica_methods = is_remote_master and clone_member.create_replica_methods
|
|
|
|
# get list of replica methods either from clone member or from
|
|
# the config. If there is no configuration key, or no value is
|
|
# specified, use basebackup
|
|
replica_methods = (
|
|
create_replica_methods
|
|
or self._create_replica_methods
|
|
or ['basebackup']
|
|
)
|
|
|
|
if clone_member and clone_member.conn_url:
|
|
r = clone_member.conn_kwargs(self._replication)
|
|
connstring = 'postgres://{user}@{host}:{port}/{database}'.format(**r)
|
|
# add the credentials to connect to the replica origin to pgpass.
|
|
env = self.write_pgpass(r)
|
|
else:
|
|
connstring = ''
|
|
env = os.environ.copy()
|
|
# if we don't have any source, leave only replica methods that work without it
|
|
replica_methods = \
|
|
[r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)]
|
|
|
|
# go through them in priority order
|
|
ret = 1
|
|
for replica_method in replica_methods:
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
break
|
|
# if the method is basebackup, then use the built-in
|
|
if replica_method == "basebackup":
|
|
ret = self.basebackup(connstring, env, self.config.get(replica_method, {}))
|
|
if ret == 0:
|
|
logger.info("replica has been created using basebackup")
|
|
# if basebackup succeeds, exit with success
|
|
break
|
|
else:
|
|
if not self.data_directory_empty() and not self.config.get(replica_method, {}).get('keep_data', False):
|
|
self.remove_data_directory()
|
|
else:
|
|
logger.info('Leaving data directory uncleaned')
|
|
|
|
cmd = replica_method
|
|
method_config = {}
|
|
# user-defined method; check for configuration
|
|
# not required, actually
|
|
if self.config.get(replica_method, {}):
|
|
method_config = self.config[replica_method].copy()
|
|
# look to see if the user has supplied a full command path
|
|
# if not, use the method name as the command
|
|
cmd = method_config.pop('command', cmd)
|
|
|
|
# add the default parameters
|
|
if not method_config.get('no_params', False):
|
|
method_config.update({"scope": self.scope,
|
|
"role": "replica",
|
|
"datadir": self._data_dir,
|
|
"connstring": connstring})
|
|
else:
|
|
for param in ('no_params', 'no_master', 'keep_data'):
|
|
method_config.pop(param, None)
|
|
params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()]
|
|
try:
|
|
# call script with the full set of parameters
|
|
ret = self.cancellable_subprocess_call(shlex.split(cmd) + params, env=env)
|
|
# if we succeeded, stop
|
|
if ret == 0:
|
|
logger.info('replica has been created using %s', replica_method)
|
|
break
|
|
else:
|
|
logger.error('Error creating replica using method %s: %s exited with code=%s',
|
|
replica_method, cmd, ret)
|
|
except Exception:
|
|
logger.exception('Error creating replica using method %s', replica_method)
|
|
ret = 1
|
|
|
|
self.set_state('stopped')
|
|
return ret
|
|
|
|
def reset_cluster_info_state(self):
|
|
self._cluster_info_state = {}
|
|
|
|
def _cluster_info_state_get(self, name):
|
|
if not self._cluster_info_state:
|
|
stmt = cluster_info_query.format(self.wal_name, self.lsn_name)
|
|
try:
|
|
result = self._is_leader_retry(self._query, stmt).fetchone()
|
|
self._cluster_info_state = dict(zip(['timeline', 'wal_position'], result))
|
|
except RetryFailedError as e: # SELECT failed two times
|
|
self._cluster_info_state = {'error': str(e)}
|
|
if not self.is_starting() and self.pg_isready() == STATE_REJECT:
|
|
self.set_state('starting')
|
|
|
|
if 'error' in self._cluster_info_state:
|
|
raise PostgresConnectionException(self._cluster_info_state['error'])
|
|
|
|
return self._cluster_info_state.get(name)
|
|
|
|
def is_leader(self):
|
|
return bool(self._cluster_info_state_get('timeline'))
|
|
|
|
def is_running(self):
|
|
"""Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
|
|
is running updates the cached process based on pid file."""
|
|
if self._postmaster_proc:
|
|
if self._postmaster_proc.is_running():
|
|
return self._postmaster_proc
|
|
self._postmaster_proc = None
|
|
|
|
# we noticed that postgres was restarted, force syncing of replication
|
|
self._schedule_load_slots = self.use_slots
|
|
|
|
self._postmaster_proc = PostmasterProcess.from_pidfile(self._data_dir)
|
|
return self._postmaster_proc
|
|
|
|
@property
|
|
def cb_called(self):
|
|
return self.__cb_called
|
|
|
|
def call_nowait(self, cb_name):
|
|
""" pick a callback command and call it without waiting for it to finish """
|
|
if self.bootstrapping:
|
|
return
|
|
if cb_name in (ACTION_ON_START, ACTION_ON_STOP, ACTION_ON_RESTART, ACTION_ON_ROLE_CHANGE):
|
|
self.__cb_called = True
|
|
|
|
if self.callback and cb_name in self.callback:
|
|
cmd = self.callback[cb_name]
|
|
try:
|
|
cmd = shlex.split(self.callback[cb_name]) + [cb_name, self.role, self.scope]
|
|
self._callback_executor.call(cmd)
|
|
except Exception:
|
|
logger.exception('callback %s %s %s %s failed', cmd, cb_name, self.role, self.scope)
|
|
|
|
@property
|
|
def role(self):
|
|
with self._role_lock:
|
|
return self._role
|
|
|
|
def set_role(self, value):
|
|
with self._role_lock:
|
|
self._role = value
|
|
|
|
@property
|
|
def state(self):
|
|
with self._state_lock:
|
|
return self._state
|
|
|
|
def set_state(self, value):
|
|
with self._state_lock:
|
|
self._state = value
|
|
self._state_entry_timestamp = time.time()
|
|
|
|
def time_in_state(self):
|
|
return time.time() - self._state_entry_timestamp
|
|
|
|
def is_starting(self):
|
|
return self.state == 'starting'
|
|
|
|
def wait_for_port_open(self, postmaster, timeout):
|
|
"""Waits until PostgreSQL opens ports."""
|
|
for _ in polling_loop(timeout):
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
return False
|
|
|
|
if not postmaster.is_running():
|
|
logger.error('postmaster is not running')
|
|
self.set_state('start failed')
|
|
return False
|
|
|
|
isready = self.pg_isready()
|
|
if isready != STATE_NO_RESPONSE:
|
|
if isready not in [STATE_REJECT, STATE_RUNNING]:
|
|
logger.warning("Can't determine PostgreSQL startup status, assuming running")
|
|
return True
|
|
|
|
logger.warning("Timed out waiting for PostgreSQL to start")
|
|
return False
|
|
|
|
def _build_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."""
|
|
|
|
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._major_version >= 90400:
|
|
OPTIONS_MAPPING['max_worker_processes'] = 'max_worker_processes setting'
|
|
|
|
data = self.controldata()
|
|
effective_configuration = self._server_parameters.copy()
|
|
|
|
for name, cname in OPTIONS_MAPPING.items():
|
|
value = parse_int(effective_configuration[name])
|
|
cvalue = parse_int(data[cname])
|
|
if cvalue > value:
|
|
effective_configuration[name] = cvalue
|
|
self._pending_restart = True
|
|
return effective_configuration
|
|
|
|
def start(self, timeout=None, block_callbacks=False, task=None):
|
|
"""Start PostgreSQL
|
|
|
|
Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
|
|
or failure.
|
|
|
|
:returns: True if start was initiated and postmaster ports are open, False if start failed"""
|
|
# make sure we close all connections established against
|
|
# the former node, otherwise, we might get a stalled one
|
|
# after kill -9, which would report incorrect data to
|
|
# patroni.
|
|
self.close_connection()
|
|
|
|
if self.is_running():
|
|
logger.error('Cannot start PostgreSQL because one is already running.')
|
|
return True
|
|
|
|
if not block_callbacks:
|
|
self.__cb_pending = ACTION_ON_START
|
|
|
|
self.set_role(self.get_postgres_role_from_data_directory())
|
|
|
|
self.set_state('starting')
|
|
self._pending_restart = False
|
|
|
|
configuration = self._server_parameters if self.role == 'master' else self._build_effective_configuration()
|
|
self._write_postgresql_conf(configuration)
|
|
self.resolve_connection_addresses()
|
|
self._replace_pg_hba()
|
|
|
|
options = ['--{0}={1}'.format(p, configuration[p]) for p in self.CMDLINE_OPTIONS
|
|
if p in configuration and p != 'wal_keep_segments']
|
|
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
return False
|
|
|
|
with task or null_context():
|
|
if task and task.is_cancelled:
|
|
logger.info("PostgreSQL start cancelled.")
|
|
return False
|
|
|
|
self._postmaster_proc = PostmasterProcess.start(self._pgcommand('postgres'),
|
|
self._data_dir,
|
|
self._postgresql_conf,
|
|
options)
|
|
|
|
if task:
|
|
task.complete(self._postmaster_proc)
|
|
|
|
start_timeout = timeout
|
|
if not start_timeout:
|
|
try:
|
|
start_timeout = float(self.config.get('pg_ctl_timeout', 60))
|
|
except ValueError:
|
|
start_timeout = 60
|
|
|
|
# We want postmaster to open ports before we continue
|
|
if not self._postmaster_proc or not self.wait_for_port_open(self._postmaster_proc, start_timeout):
|
|
return False
|
|
|
|
ret = self.wait_for_startup(start_timeout)
|
|
if ret is not None:
|
|
return ret
|
|
elif timeout is not None:
|
|
return False
|
|
else:
|
|
return None
|
|
|
|
def checkpoint(self, connect_kwargs=None):
|
|
check_not_is_in_recovery = connect_kwargs is not None
|
|
connect_kwargs = connect_kwargs or self._local_connect_kwargs
|
|
for p in ['connect_timeout', 'options']:
|
|
connect_kwargs.pop(p, None)
|
|
try:
|
|
with self._get_connection_cursor(**connect_kwargs) as cur:
|
|
cur.execute("SET statement_timeout = 0")
|
|
if check_not_is_in_recovery:
|
|
cur.execute('SELECT pg_catalog.pg_is_in_recovery()')
|
|
if cur.fetchone()[0]:
|
|
return 'is_in_recovery=true'
|
|
return cur.execute('CHECKPOINT')
|
|
except psycopg2.Error:
|
|
logger.exception('Exception during CHECKPOINT')
|
|
return 'not accessible or not healty'
|
|
|
|
def stop(self, mode='fast', block_callbacks=False, checkpoint=None, on_safepoint=None):
|
|
"""Stop PostgreSQL
|
|
|
|
Supports a callback when a safepoint is reached. A safepoint is when no user backend can return a successful
|
|
commit to users. Currently this means we wait for user backends to close. But in the future alternate mechanisms
|
|
could be added.
|
|
|
|
:param on_safepoint: This callback is called when no user backends are running.
|
|
"""
|
|
if checkpoint is None:
|
|
checkpoint = False if mode == 'immediate' else True
|
|
|
|
success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint)
|
|
if success:
|
|
# block_callbacks is used during restart to avoid
|
|
# running start/stop callbacks in addition to restart ones
|
|
if not block_callbacks:
|
|
self.set_state('stopped')
|
|
if pg_signaled:
|
|
self.call_nowait(ACTION_ON_STOP)
|
|
else:
|
|
logger.warning('pg_ctl stop failed')
|
|
self.set_state('stop failed')
|
|
return success
|
|
|
|
def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint):
|
|
postmaster = self.is_running()
|
|
if not postmaster:
|
|
if on_safepoint:
|
|
on_safepoint()
|
|
return True, False
|
|
|
|
if checkpoint and not self.is_starting():
|
|
self.checkpoint()
|
|
|
|
if not block_callbacks:
|
|
self.set_state('stopping')
|
|
|
|
# Send signal to postmaster to stop
|
|
success = postmaster.signal_stop(mode)
|
|
if success is not None:
|
|
if success and on_safepoint:
|
|
on_safepoint()
|
|
return success, True
|
|
|
|
# We can skip safepoint detection if we don't have a callback
|
|
if on_safepoint:
|
|
# Wait for our connection to terminate so we can be sure that no new connections are being initiated
|
|
self._wait_for_connection_close(postmaster)
|
|
postmaster.wait_for_user_backends_to_close()
|
|
on_safepoint()
|
|
|
|
postmaster.wait()
|
|
|
|
return True, True
|
|
|
|
@staticmethod
|
|
def terminate_starting_postmaster(postmaster):
|
|
"""Terminates a postmaster that has not yet opened ports or possibly even written a pid file. Blocks
|
|
until the process goes away."""
|
|
postmaster.signal_stop('immediate')
|
|
postmaster.wait()
|
|
|
|
def _wait_for_connection_close(self, postmaster):
|
|
try:
|
|
with self.connection().cursor() as cur:
|
|
while postmaster.is_running(): # Need a timeout here?
|
|
cur.execute("SELECT 1")
|
|
time.sleep(STOP_POLLING_INTERVAL)
|
|
except psycopg2.Error:
|
|
pass
|
|
|
|
def reload(self):
|
|
ret = self.pg_ctl('reload')
|
|
if ret:
|
|
self.call_nowait(ACTION_ON_RELOAD)
|
|
return ret
|
|
|
|
def check_for_startup(self):
|
|
"""Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup."""
|
|
return self.is_starting() and not self.check_startup_state_changed()
|
|
|
|
def check_startup_state_changed(self):
|
|
"""Checks if PostgreSQL has completed starting up or failed or still starting.
|
|
|
|
Should only be called when state == 'starting'
|
|
|
|
:returns: True if state was changed from 'starting'
|
|
"""
|
|
ready = self.pg_isready()
|
|
|
|
if ready == STATE_REJECT:
|
|
return False
|
|
elif ready == STATE_NO_RESPONSE:
|
|
self.set_state('start failed')
|
|
self._schedule_load_slots = False # TODO: can remove this?
|
|
if not self._running_custom_bootstrap:
|
|
self.save_configuration_files() # TODO: maybe remove this?
|
|
return True
|
|
else:
|
|
if ready != STATE_RUNNING:
|
|
# Bad configuration or unexpected OS error. No idea of PostgreSQL status.
|
|
# Let the main loop of run cycle clean up the mess.
|
|
logger.warning("%s status returned from pg_isready",
|
|
"Unknown" if ready == STATE_UNKNOWN else "Invalid")
|
|
self.set_state('running')
|
|
self._schedule_load_slots = self.use_slots
|
|
if not self._running_custom_bootstrap:
|
|
self.save_configuration_files()
|
|
# TODO: __cb_pending can be None here after PostgreSQL restarts on its own. Do we want to call the callback?
|
|
# Previously we didn't even notice.
|
|
action = self.__cb_pending or ACTION_ON_START
|
|
self.call_nowait(action)
|
|
self.__cb_pending = None
|
|
|
|
return True
|
|
|
|
def wait_for_startup(self, timeout=None):
|
|
"""Waits for PostgreSQL startup to complete or fail.
|
|
|
|
:returns: True if start was successful, False otherwise"""
|
|
if not self.is_starting():
|
|
# Should not happen
|
|
logger.warning("wait_for_startup() called when not in starting state")
|
|
|
|
while not self.check_startup_state_changed():
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
return None
|
|
if timeout and self.time_in_state() > timeout:
|
|
return None
|
|
time.sleep(1)
|
|
|
|
return self.state == 'running'
|
|
|
|
def restart(self, timeout=None, task=None):
|
|
"""Restarts PostgreSQL.
|
|
|
|
When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or
|
|
timeout arrives.
|
|
|
|
:returns: True when restart was successful and timeout did not expire when waiting.
|
|
"""
|
|
self.set_state('restarting')
|
|
self.__cb_pending = ACTION_ON_RESTART
|
|
ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True, task=task)
|
|
if not ret and not self.is_starting():
|
|
self.set_state('restart failed ({0})'.format(self.state))
|
|
return ret
|
|
|
|
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)
|
|
|
|
with open(self._postgresql_conf, 'w') as f:
|
|
f.write(self._CONFIG_WARNING_HEADER)
|
|
f.write("include '{0}'\n\n".format(self.config.get('custom_conf') or self._postgresql_base_conf_name))
|
|
for name, value in sorted((configuration or self._server_parameters).items()):
|
|
if not self._running_custom_bootstrap or name != 'hba_file':
|
|
f.write("{0} = '{1}'\n".format(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 overriden
|
|
# after changing superuser password we will "revert" all these "changes"
|
|
if self._running_custom_bootstrap or 'hba_file' not in self._server_parameters:
|
|
f.write("hba_file = '{0}'\n".format(self._pg_hba_conf.replace('\\', '\\\\')))
|
|
if 'ident_file' not in self._server_parameters:
|
|
s = "ident_file = '{0}'\n".format(os.path.join(self._config_dir, 'pg_ident.conf').replace('\\', '\\\\'))
|
|
f.write(s)
|
|
|
|
def is_healthy(self):
|
|
if not self.is_running():
|
|
logger.warning('Postgresql is not running.')
|
|
return False
|
|
return True
|
|
|
|
def write_pg_hba(self, config):
|
|
if not self._server_parameters.get('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._running_custom_bootstrap:
|
|
addresses = {'': 'local'}
|
|
if 'host' in self._local_address and not self._local_address['host'].startswith('/'):
|
|
for _, _, _, _, sa in socket.getaddrinfo(self._local_address['host'], self._local_address['port'],
|
|
0, socket.SOCK_STREAM, socket.IPPROTO_TCP):
|
|
addresses[sa[0] + '/32'] = 'host'
|
|
|
|
with open(self._pg_hba_conf, 'w') as f:
|
|
f.write(self._CONFIG_WARNING_HEADER)
|
|
for address, t in addresses.items():
|
|
f.write('{0}\t{1}\t{2}\t{3}\ttrust\n'.format(t, 'all',
|
|
self._superuser.get('username') or 'all', address))
|
|
elif not self._server_parameters.get('hba_file') and self.config.get('pg_hba'):
|
|
with open(self._pg_hba_conf, 'w') as f:
|
|
f.write(self._CONFIG_WARNING_HEADER)
|
|
for line in self.config['pg_hba']:
|
|
f.write('{0}\n'.format(line))
|
|
return True
|
|
|
|
def primary_conninfo(self, member):
|
|
if not (member and member.conn_url) or member.name == self.name:
|
|
return None
|
|
r = member.conn_kwargs(self._replication)
|
|
r.update({'application_name': self.name, 'sslmode': 'prefer', 'sslcompression': '1'})
|
|
keywords = 'user password host port sslmode sslcompression application_name'.split()
|
|
return ' '.join('{0}={{{0}}}'.format(kw) for kw in keywords).format(**r)
|
|
|
|
def check_recovery_conf(self, member):
|
|
# TODO: recovery.conf could be stale, would be nice to detect that.
|
|
primary_conninfo = self.primary_conninfo(member)
|
|
|
|
if not os.path.isfile(self._recovery_conf):
|
|
return False
|
|
|
|
with open(self._recovery_conf, 'r') as f:
|
|
for line in f:
|
|
if line.startswith('primary_conninfo'):
|
|
return primary_conninfo and (primary_conninfo in line)
|
|
return not primary_conninfo
|
|
|
|
def write_recovery_conf(self, recovery_params):
|
|
with open(self._recovery_conf, 'w') as f:
|
|
for name, value in recovery_params.items():
|
|
f.write("{0} = '{1}'\n".format(name, value))
|
|
|
|
def pg_rewind(self, r):
|
|
# prepare pg_rewind connection
|
|
env = self.write_pgpass(r)
|
|
dsn_attrs = [
|
|
('user', r.get('user')),
|
|
('host', r.get('host')),
|
|
('port', r.get('port')),
|
|
('dbname', r.get('database') or self._database),
|
|
('sslmode', 'prefer'),
|
|
('sslcompression', '1'),
|
|
]
|
|
dsn = " ".join("{0}={1}".format(k, v) for k, v in dsn_attrs if v is not None)
|
|
logger.info('running pg_rewind from %s', dsn)
|
|
try:
|
|
return self.cancellable_subprocess_call([self._pgcommand('pg_rewind'),
|
|
'-D', self._data_dir,
|
|
'--source-server', dsn], env=env) == 0
|
|
except OSError:
|
|
return False
|
|
|
|
def controldata(self):
|
|
""" return the contents of pg_controldata, or non-True value if pg_controldata call failed """
|
|
result = {}
|
|
# Don't try to call pg_controldata during backup restore
|
|
if self._version_file_exists() and self.state != 'creating replica':
|
|
try:
|
|
env = {'LANG': 'C', 'LC_ALL': 'C', 'PATH': os.getenv('PATH')}
|
|
if os.getenv('SYSTEMROOT') is not None:
|
|
env['SYSTEMROOT'] = os.getenv('SYSTEMROOT')
|
|
data = subprocess.check_output([self._pgcommand('pg_controldata'), self._data_dir], env=env)
|
|
if data:
|
|
data = data.decode('utf-8').splitlines()
|
|
# pg_controldata output depends on major verion. Some of parameters are prefixed by 'Current '
|
|
result = {l.split(':')[0].replace('Current ', '', 1): l.split(':', 1)[1].strip() for l in data
|
|
if l and ':' in l}
|
|
except subprocess.CalledProcessError:
|
|
logger.exception("Error when calling pg_controldata")
|
|
return result
|
|
|
|
@property
|
|
def need_rewind(self):
|
|
return self._rewind_state in (REWIND_STATUS.CHECK, REWIND_STATUS.NEED)
|
|
|
|
@staticmethod
|
|
@contextmanager
|
|
def _get_connection_cursor(**kwargs):
|
|
with psycopg2.connect(**kwargs) as conn:
|
|
conn.autocommit = True
|
|
with conn.cursor() as cur:
|
|
yield cur
|
|
|
|
@contextmanager
|
|
def _get_replication_connection_cursor(self, host='localhost', port=5432, database=None, **kwargs):
|
|
database = database or self._database
|
|
replication = 'database' if self._major_version >= 90400 else 1
|
|
with self._get_connection_cursor(host=host, port=int(port), database=database, replication=replication,
|
|
user=self._replication['username'], password=self._replication['password'],
|
|
connect_timeout=3, options='-c statement_timeout=2000') as cur:
|
|
yield cur
|
|
|
|
def check_leader_is_not_in_recovery(self, **kwargs):
|
|
try:
|
|
with self._get_connection_cursor(connect_timeout=3, options='-c statement_timeout=2000', **kwargs) as cur:
|
|
cur.execute('SELECT pg_catalog.pg_is_in_recovery()')
|
|
if not cur.fetchone()[0]:
|
|
return True
|
|
logger.info('Leader is still in_recovery and therefore can\'t be used for rewind')
|
|
except Exception:
|
|
return logger.exception('Exception when working with leader')
|
|
|
|
def _get_local_timeline_lsn_from_replication_connection(self):
|
|
timeline = lsn = None
|
|
try:
|
|
with self._get_replication_connection_cursor(**self._local_replication_address) as cur:
|
|
cur.execute('IDENTIFY_SYSTEM')
|
|
timeline, lsn = cur.fetchone()[1:3]
|
|
except Exception:
|
|
logger.exception('Can not fetch local timeline and lsn from replication connection')
|
|
return timeline, lsn
|
|
|
|
def _get_local_timeline_lsn_from_controldata(self):
|
|
timeline = lsn = None
|
|
data = self.controldata()
|
|
try:
|
|
if data.get('Database cluster state') == 'shut down in recovery':
|
|
lsn = data.get('Minimum recovery ending location')
|
|
timeline = int(data.get("Min recovery ending loc's timeline"))
|
|
if lsn == '0/0' or timeline == 0: # it was a master when it crashed
|
|
data['Database cluster state'] = 'shut down'
|
|
if data.get('Database cluster state') == 'shut down':
|
|
lsn = data.get('Latest checkpoint location')
|
|
timeline = int(data.get("Latest checkpoint's TimeLineID"))
|
|
except (TypeError, ValueError):
|
|
logger.exception('Failed to get local timeline and lsn from pg_controldata output')
|
|
return timeline, lsn
|
|
|
|
def _get_local_timeline_lsn(self):
|
|
if self.is_running(): # if postgres is running - get timeline and lsn from replication connection
|
|
timeline, lsn = self._get_local_timeline_lsn_from_replication_connection()
|
|
else: # otherwise analyze pg_controldata output
|
|
timeline, lsn = self._get_local_timeline_lsn_from_controldata()
|
|
logger.info('Local timeline=%s lsn=%s', timeline, lsn)
|
|
return timeline, lsn
|
|
|
|
@staticmethod
|
|
def parse_lsn(lsn):
|
|
t = lsn.split('/')
|
|
return int(t[0], 16) * 0x100000000 + int(t[1], 16)
|
|
|
|
@staticmethod
|
|
def parse_history(data):
|
|
for line in data.split('\n'):
|
|
values = line.strip().split('\t')
|
|
if len(values) == 3:
|
|
try:
|
|
values[0] = int(values[0])
|
|
values[1] = Postgresql.parse_lsn(values[1])
|
|
yield values
|
|
except (IndexError, ValueError):
|
|
logger.exception('Exception when parsing timeline history line "%s"', values)
|
|
|
|
def _check_timeline_and_lsn(self, leader):
|
|
local_timeline, local_lsn = self._get_local_timeline_lsn()
|
|
if local_timeline is None or local_lsn is None:
|
|
return
|
|
|
|
if not self.check_leader_is_not_in_recovery(**leader.conn_kwargs(self._superuser)):
|
|
return
|
|
|
|
history = need_rewind = None
|
|
try:
|
|
with self._get_replication_connection_cursor(**leader.conn_kwargs()) as cur:
|
|
cur.execute('IDENTIFY_SYSTEM')
|
|
master_timeline = cur.fetchone()[1]
|
|
logger.info('master_timeline=%s', master_timeline)
|
|
if local_timeline > master_timeline: # Not always supported by pg_rewind
|
|
need_rewind = True
|
|
elif master_timeline > 1:
|
|
cur.execute('TIMELINE_HISTORY %s', (master_timeline,))
|
|
history = bytes(cur.fetchone()[1]).decode('utf-8')
|
|
logger.info('master: history=%s', history)
|
|
else: # local_timeline == master_timeline == 1
|
|
need_rewind = False
|
|
except Exception:
|
|
return logger.exception('Exception when working with master via replication connection')
|
|
|
|
if history is not None:
|
|
for parent_timeline, switchpoint, _ in self.parse_history(history):
|
|
if parent_timeline == local_timeline:
|
|
try:
|
|
need_rewind = self.parse_lsn(local_lsn) >= switchpoint
|
|
except (IndexError, ValueError):
|
|
logger.exception('Exception when parsing lsn')
|
|
break
|
|
elif parent_timeline > local_timeline:
|
|
break
|
|
|
|
self._rewind_state = need_rewind and REWIND_STATUS.NEED or REWIND_STATUS.NOT_NEED
|
|
|
|
def get_replica_timeline(self):
|
|
return self._get_local_timeline_lsn_from_replication_connection()[0]
|
|
|
|
def replica_cached_timeline(self, master_timeline):
|
|
if not self._cached_replica_timeline or not master_timeline or self._cached_replica_timeline != master_timeline:
|
|
self._cached_replica_timeline = self.get_replica_timeline()
|
|
return self._cached_replica_timeline
|
|
|
|
def get_master_timeline(self):
|
|
return self._cluster_info_state_get('timeline')
|
|
|
|
def get_history(self, timeline):
|
|
history_path = 'pg_{0}/{1:08X}.history'.format(self.wal_name, timeline)
|
|
try:
|
|
cursor = self._cursor()
|
|
cursor.execute('SELECT isdir, modification FROM pg_catalog.pg_stat_file(%s)', (history_path,))
|
|
isdir, modification = cursor.fetchone()
|
|
if not isdir:
|
|
cursor.execute('SELECT pg_catalog.pg_read_file(%s)', (history_path,))
|
|
history = list(self.parse_history(cursor.fetchone()[0]))
|
|
if history[-1][0] == timeline - 1:
|
|
history[-1].append(modification.isoformat())
|
|
return history
|
|
except Exception:
|
|
logger.exception('Failed to read and parse %s', (history_path,))
|
|
|
|
def rewind(self, leader):
|
|
if self.is_running() and not self.stop(checkpoint=False):
|
|
return logger.warning('Can not run pg_rewind because postgres is still running')
|
|
|
|
# prepare pg_rewind connection
|
|
r = leader.conn_kwargs(self._superuser)
|
|
|
|
# first make sure that we are really trying to rewind
|
|
# from the master and run a checkpoint on it in order to
|
|
# make it store the new timeline ([email protected])
|
|
leader_status = self.checkpoint(r)
|
|
if leader_status:
|
|
return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
|
|
|
|
if self.pg_rewind(r):
|
|
self._rewind_state = REWIND_STATUS.SUCCESS
|
|
elif not self.check_leader_is_not_in_recovery(**r):
|
|
logger.warning('Failed to rewind because master %s become unreachable', leader.name)
|
|
else:
|
|
logger.error('Failed to rewind from healty master: %s', leader.name)
|
|
|
|
if self.config.get('remove_data_directory_on_rewind_failure', False):
|
|
logger.warning('remove_data_directory_on_rewind_failure is set. removing...')
|
|
self.remove_data_directory()
|
|
self._rewind_state = REWIND_STATUS.INITIAL
|
|
else:
|
|
self._rewind_state = REWIND_STATUS.FAILED
|
|
return False
|
|
|
|
def trigger_check_diverged_lsn(self):
|
|
if self.can_rewind and self._rewind_state != REWIND_STATUS.NEED:
|
|
self._rewind_state = REWIND_STATUS.CHECK
|
|
|
|
def rewind_needed_and_possible(self, leader):
|
|
if leader and leader.name != self.name and leader.conn_url and self._rewind_state == REWIND_STATUS.CHECK:
|
|
self._check_timeline_and_lsn(leader)
|
|
return leader and leader.conn_url and self._rewind_state == REWIND_STATUS.NEED
|
|
|
|
@property
|
|
def rewind_executed(self):
|
|
return self._rewind_state > REWIND_STATUS.NOT_NEED
|
|
|
|
@property
|
|
def rewind_failed(self):
|
|
return self._rewind_state == REWIND_STATUS.FAILED
|
|
|
|
def follow(self, member, timeout=None):
|
|
is_remote_master = isinstance(member, RemoteMember)
|
|
no_replication_slot = is_remote_master and member.no_replication_slot
|
|
restore_command = is_remote_master and member.restore_command
|
|
min_apply_delay = is_remote_master and member.recovery_min_apply_delay
|
|
archive_cleanup = is_remote_master and member.archive_cleanup_command
|
|
|
|
primary_conninfo = self.primary_conninfo(member)
|
|
change_role = self.role in ('master', 'demoted')
|
|
|
|
recovery_params = self.config.get('recovery_conf', {}).copy()
|
|
recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
|
|
if primary_conninfo:
|
|
recovery_params['primary_conninfo'] = primary_conninfo
|
|
if self.use_slots and not no_replication_slot:
|
|
required_name = is_remote_master and member.data.get('primary_slot_name')
|
|
name = required_name or slot_name_from_member_name(self.name)
|
|
recovery_params['primary_slot_name'] = name
|
|
if restore_command:
|
|
recovery_params['restore_command'] = restore_command
|
|
if min_apply_delay:
|
|
recovery_params['recovery_min_apply_delay'] = min_apply_delay
|
|
if archive_cleanup:
|
|
recovery_params['archive_cleanup_command'] = archive_cleanup
|
|
|
|
self.write_recovery_conf(recovery_params)
|
|
|
|
if self.is_running():
|
|
self.restart()
|
|
else:
|
|
self.start(timeout=timeout)
|
|
self.set_role('replica')
|
|
|
|
if change_role:
|
|
# TODO: postpone this until start completes, or maybe do even earlier
|
|
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
|
return True
|
|
|
|
def save_configuration_files(self):
|
|
"""
|
|
copy postgresql.conf to postgresql.conf.backup to be able to retrive 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)
|
|
"""
|
|
try:
|
|
for f in self._configuration_to_save:
|
|
config_file = os.path.join(self._config_dir, f)
|
|
backup_file = os.path.join(self._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._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 _wait_promote(self, wait_seconds):
|
|
for _ in polling_loop(wait_seconds - 1):
|
|
data = self.controldata()
|
|
if data.get('Database cluster state') == 'in production':
|
|
return True
|
|
|
|
def promote(self, wait_seconds, access_is_restricted=False):
|
|
if self.role == 'master':
|
|
return True
|
|
ret = self.pg_ctl('promote', '-W')
|
|
if ret:
|
|
self.set_role('master')
|
|
logger.info("cleared rewind state after becoming the leader")
|
|
self._rewind_state = REWIND_STATUS.INITIAL
|
|
if not access_is_restricted:
|
|
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
|
ret = self._wait_promote(wait_seconds)
|
|
return ret
|
|
|
|
def create_or_update_role(self, name, password, options):
|
|
options = list(map(str.upper, options))
|
|
if 'NOLOGIN' not in options and 'LOGIN' not in options:
|
|
options.append('LOGIN')
|
|
|
|
self.query("""DO $$
|
|
BEGIN
|
|
SET local synchronous_commit = 'local';
|
|
PERFORM * FROM pg_authid WHERE rolname = %s;
|
|
IF FOUND THEN
|
|
ALTER ROLE "{0}" WITH {1} PASSWORD %s;
|
|
ELSE
|
|
CREATE ROLE "{0}" WITH {1} PASSWORD %s;
|
|
END IF;
|
|
END;
|
|
$$""".format(name, ' '.join(options)), name, password, password)
|
|
|
|
def timeline_wal_position(self):
|
|
# This method could be called from different threads (simultaneously with some other `_query` calls).
|
|
# If it is called not from main thread we will create a new cursor to execute statement.
|
|
if current_thread().ident == self.__thread_ident:
|
|
return self._cluster_info_state_get('timeline'), self._cluster_info_state_get('wal_position')
|
|
|
|
with self.connection().cursor() as cursor:
|
|
cursor.execute(cluster_info_query.format(self.wal_name, self.lsn_name))
|
|
return cursor.fetchone()[:2]
|
|
|
|
def load_replication_slots(self):
|
|
if self.use_slots and self._schedule_load_slots:
|
|
replication_slots = {}
|
|
cursor = self._query('SELECT slot_name, slot_type, plugin, database FROM pg_catalog.pg_replication_slots')
|
|
for r in cursor:
|
|
value = {'type': r[1]}
|
|
if r[1] == 'logical':
|
|
value.update({'plugin': r[2], 'database': r[3]})
|
|
replication_slots[r[0]] = value
|
|
self._replication_slots = replication_slots
|
|
self._schedule_load_slots = False
|
|
|
|
def postmaster_start_time(self):
|
|
try:
|
|
cursor = self.query("SELECT pg_catalog.to_char(pg_catalog.pg_postmaster_start_time(),"
|
|
" 'YYYY-MM-DD HH24:MI:SS.MS TZ')")
|
|
return cursor.fetchone()[0]
|
|
except psycopg2.Error:
|
|
return None
|
|
|
|
def drop_replication_slot(self, name):
|
|
cursor = self._query(('SELECT pg_catalog.pg_drop_replication_slot(%s) WHERE EXISTS (SELECT 1 ' +
|
|
'FROM pg_catalog.pg_replication_slots WHERE slot_name = %s AND NOT active)'), name, name)
|
|
# In normal situation rowcount should be 1, otherwise either slot doesn't exists or it is still active
|
|
return cursor.rowcount == 1
|
|
|
|
@staticmethod
|
|
def compare_slots(s1, s2):
|
|
return s1['type'] == s2['type'] and\
|
|
(s1['type'] == 'physical' or s1['database'] == s2['database'] and s1['plugin'] == s2['plugin'])
|
|
|
|
def sync_replication_slots(self, cluster):
|
|
if self.use_slots:
|
|
try:
|
|
self.load_replication_slots()
|
|
|
|
slots = cluster.get_replication_slots(self.name, self.role)
|
|
|
|
# drop old replication slots which are not presented in desired slots
|
|
for name in set(self._replication_slots) - set(slots):
|
|
if not self.drop_replication_slot(name):
|
|
logger.error("Failed to drop replication slot '%s'", name)
|
|
self._schedule_load_slots = True
|
|
|
|
immediately_reserve = ', true' if self._major_version >= 90600 else ''
|
|
|
|
logical_slots = defaultdict(dict)
|
|
for name, value in slots.items():
|
|
if name in self._replication_slots and not self.compare_slots(value, self._replication_slots[name]):
|
|
logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s",
|
|
name, self._replication_slots[name], value)
|
|
if not self.drop_replication_slot(name):
|
|
logger.error("Failed to drop replication slot '%s'", name)
|
|
self._schedule_load_slots = True
|
|
continue
|
|
self._replication_slots.pop(name)
|
|
if name not in self._replication_slots:
|
|
if value['type'] == 'physical':
|
|
try:
|
|
self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" +
|
|
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
|
|
" WHERE slot_type = 'physical' AND slot_name = %s)").format(
|
|
immediately_reserve), name, name)
|
|
except Exception:
|
|
logger.exception("Failed to create physical replication slot '%s'", name)
|
|
self._schedule_load_slots = True
|
|
elif value['type'] == 'logical' and name not in self._replication_slots:
|
|
logical_slots[value['database']][name] = value
|
|
|
|
# create new logical slots
|
|
for database, values in logical_slots.items():
|
|
conn_kwargs = self._local_connect_kwargs
|
|
conn_kwargs['database'] = database
|
|
with self._get_connection_cursor(**conn_kwargs) as cur:
|
|
for name, value in values.items():
|
|
try:
|
|
cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
|
|
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
|
|
" WHERE slot_type = 'logical' AND slot_name = %s)",
|
|
(name, value['plugin'], name))
|
|
except Exception:
|
|
logger.exception("Failed to create logical replication slot '%s' plugin='%s'",
|
|
name, value['plugin'])
|
|
self._schedule_load_slots = True
|
|
self._replication_slots = slots
|
|
except Exception:
|
|
logger.exception('Exception when changing replication slots')
|
|
self._schedule_load_slots = True
|
|
|
|
def last_operation(self):
|
|
return str(self._cluster_info_state_get('wal_position'))
|
|
|
|
def _post_restore(self):
|
|
self.delete_trigger_file()
|
|
self.restore_configuration_files()
|
|
|
|
def _configure_server_parameters(self):
|
|
self._major_version = self.get_major_version()
|
|
self._server_parameters = self.get_server_parameters(self.config)
|
|
return True
|
|
|
|
def clone(self, clone_member):
|
|
"""
|
|
- initialize the replica from an existing member (master or replica)
|
|
- initialize the replica using the replica creation method that
|
|
works without the replication connection (i.e. restore from on-disk
|
|
base backup)
|
|
"""
|
|
|
|
ret = self.create_replica(clone_member) == 0
|
|
if ret:
|
|
self._post_restore()
|
|
self._configure_server_parameters()
|
|
return ret
|
|
|
|
def bootstrap(self, config):
|
|
""" Initialize a new node from scratch and start it. """
|
|
pg_hba = config.get('pg_hba', [])
|
|
method = config.get('method') or 'initdb'
|
|
self._running_custom_bootstrap = method != 'initdb' and method in config and 'command' in config[method]
|
|
if self._running_custom_bootstrap:
|
|
do_initialize = self._custom_bootstrap
|
|
config = config[method]
|
|
else:
|
|
do_initialize = self._initdb
|
|
return do_initialize(config) and self.write_pg_hba(pg_hba) and self.save_configuration_files() \
|
|
and self._configure_server_parameters() and self.start()
|
|
|
|
def post_bootstrap(self, config, task):
|
|
try:
|
|
self.create_or_update_role(self._superuser['username'], self._superuser['password'], ['SUPERUSER'])
|
|
|
|
task.complete(self.run_bootstrap_post_init(config))
|
|
if task.result:
|
|
self.create_or_update_role(self._replication['username'],
|
|
self._replication['password'], ['REPLICATION'])
|
|
for name, value in (config.get('users') or {}).items():
|
|
if name not in (self._superuser.get('username'), self._replication['username']):
|
|
self.create_or_update_role(name, value['password'], value.get('options', []))
|
|
|
|
# We were doing a custom bootstrap instead of running initdb, therefore we opened trust
|
|
# access from certain addresses to be able to reach cluster and change password
|
|
if self._running_custom_bootstrap:
|
|
self._running_custom_bootstrap = False
|
|
# If we don't have custom configuration for pg_hba.conf we need to restore original file
|
|
if not self.config.get('pg_hba'):
|
|
os.unlink(self._pg_hba_conf)
|
|
self.restore_configuration_files()
|
|
self._write_postgresql_conf()
|
|
if self._server_parameters.get('hba_file') and \
|
|
self._server_parameters['hba_file'] != self._pg_hba_conf:
|
|
self.restart()
|
|
else:
|
|
self._replace_pg_hba()
|
|
if self.pending_restart:
|
|
self.restart()
|
|
else:
|
|
self.reload()
|
|
time.sleep(1) # give a time to postgres to "reload" configuration files
|
|
self.close_connection() # close connection to reconnect with a new password
|
|
except Exception:
|
|
logger.exception('post_bootstrap')
|
|
task.complete(False)
|
|
return task.result
|
|
|
|
def move_data_directory(self):
|
|
if os.path.isdir(self._data_dir) and not self.is_running():
|
|
try:
|
|
new_name = '{0}_{1}'.format(self._data_dir, time.strftime('%Y-%m-%d-%H-%M-%S'))
|
|
logger.info('renaming data directory to %s', new_name)
|
|
os.rename(self._data_dir, new_name)
|
|
except OSError:
|
|
logger.exception("Could not rename data directory %s", self._data_dir)
|
|
|
|
def remove_data_directory(self):
|
|
self.set_role('uninitialized')
|
|
logger.info('Removing data directory: %s', self._data_dir)
|
|
try:
|
|
if os.path.islink(self._data_dir):
|
|
os.unlink(self._data_dir)
|
|
elif not os.path.exists(self._data_dir):
|
|
return
|
|
elif os.path.isfile(self._data_dir):
|
|
os.remove(self._data_dir)
|
|
elif os.path.isdir(self._data_dir):
|
|
shutil.rmtree(self._data_dir)
|
|
except (IOError, OSError):
|
|
logger.exception('Could not remove data directory %s', self._data_dir)
|
|
self.move_data_directory()
|
|
|
|
def basebackup(self, conn_url, env, options):
|
|
# creates a replica data dir using pg_basebackup.
|
|
# this is the default, built-in create_replica_methods
|
|
# tries twice, then returns failure (as 1)
|
|
# uses "stream" as the xlog-method to avoid sync issues
|
|
# supports additional user-supplied options, those are not validated
|
|
maxfailures = 2
|
|
ret = 1
|
|
not_allowed_options = ('pgdata', 'format', 'wal-method', 'xlog-method', 'gzip',
|
|
'version', 'compress', 'dbname', 'host', 'port', 'username', 'password')
|
|
user_options = self.process_user_options('basebackup', options, not_allowed_options, logger.error)
|
|
|
|
for bbfailures in range(0, maxfailures):
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
break
|
|
if not self.data_directory_empty():
|
|
self.remove_data_directory()
|
|
try:
|
|
ret = self.cancellable_subprocess_call([self._pgcommand('pg_basebackup'), '--pgdata=' + self._data_dir,
|
|
'-X', 'stream', '--dbname=' + conn_url] + user_options, env=env)
|
|
if ret == 0:
|
|
break
|
|
else:
|
|
logger.error('Error when fetching backup: pg_basebackup exited with code=%s', ret)
|
|
|
|
except Exception as e:
|
|
logger.error('Error when fetching backup with pg_basebackup: %s', e)
|
|
|
|
if bbfailures < maxfailures - 1:
|
|
logger.warning('Trying again in 5 seconds')
|
|
time.sleep(5)
|
|
|
|
return ret
|
|
|
|
def pick_synchronous_standby(self, cluster):
|
|
"""Finds the best candidate to be the synchronous standby.
|
|
|
|
Current synchronous standby is always preferred, unless it has disconnected or does not want to be a
|
|
synchronous standby any longer.
|
|
|
|
:returns tuple of candidate name or None, and bool showing if the member is the active synchronous standby.
|
|
"""
|
|
current = cluster.sync.sync_standby
|
|
current = current.lower() if current else current
|
|
members = {m.name.lower(): m for m in cluster.members}
|
|
candidates = []
|
|
# Pick candidates based on who has flushed WAL farthest.
|
|
# TODO: for synchronous_commit = remote_write we actually want to order on write_location
|
|
for app_name, state, sync_state in self.query(
|
|
"SELECT pg_catalog.lower(application_name), state, sync_state"
|
|
" FROM pg_catalog.pg_stat_replication"
|
|
" ORDER BY flush_{0} DESC".format(self.lsn_name)):
|
|
member = members.get(app_name)
|
|
if state != 'streaming' or not member or member.tags.get('nosync', False):
|
|
continue
|
|
if sync_state == 'sync':
|
|
return app_name, True
|
|
if sync_state == 'potential' and app_name == current:
|
|
# Prefer current even if not the best one any more to avoid indecisivness and spurious swaps.
|
|
return current, False
|
|
if sync_state in ('async', 'potential'):
|
|
candidates.append(app_name)
|
|
|
|
if candidates:
|
|
return candidates[0], False
|
|
return None, False
|
|
|
|
def set_synchronous_standby(self, name):
|
|
"""Sets a node to be synchronous standby and if changed does a reload for PostgreSQL."""
|
|
if name and name != '*':
|
|
name = quote_ident(name)
|
|
if name != self._synchronous_standby_names:
|
|
if name is None:
|
|
self._server_parameters.pop('synchronous_standby_names', None)
|
|
else:
|
|
self._server_parameters['synchronous_standby_names'] = name
|
|
self._synchronous_standby_names = name
|
|
if self.state == 'running':
|
|
self._write_postgresql_conf()
|
|
self.reload()
|
|
|
|
@staticmethod
|
|
def postgres_version_to_int(pg_version):
|
|
"""Convert the server_version to integer
|
|
|
|
>>> Postgresql.postgres_version_to_int('9.5.3')
|
|
90503
|
|
>>> Postgresql.postgres_version_to_int('9.3.13')
|
|
90313
|
|
>>> Postgresql.postgres_version_to_int('10.1')
|
|
100001
|
|
>>> Postgresql.postgres_version_to_int('10') # doctest: +IGNORE_EXCEPTION_DETAIL
|
|
Traceback (most recent call last):
|
|
...
|
|
PostgresException: 'Invalid PostgreSQL version format: X.Y or X.Y.Z is accepted: 10'
|
|
>>> Postgresql.postgres_version_to_int('9.6') # doctest: +IGNORE_EXCEPTION_DETAIL
|
|
Traceback (most recent call last):
|
|
...
|
|
PostgresException: 'Invalid PostgreSQL version format: X.Y or X.Y.Z is accepted: 9.6'
|
|
>>> Postgresql.postgres_version_to_int('a.b.c') # doctest: +IGNORE_EXCEPTION_DETAIL
|
|
Traceback (most recent call last):
|
|
...
|
|
PostgresException: 'Invalid PostgreSQL version: a.b.c'
|
|
"""
|
|
|
|
try:
|
|
components = list(map(int, pg_version.split('.')))
|
|
except ValueError:
|
|
raise PostgresException('Invalid PostgreSQL version: {0}'.format(pg_version))
|
|
|
|
if len(components) < 2 or len(components) == 2 and components[0] < 10 or len(components) > 3:
|
|
raise PostgresException('Invalid PostgreSQL version format: X.Y or X.Y.Z is accepted: {0}'
|
|
.format(pg_version))
|
|
|
|
if len(components) == 2:
|
|
# new style verion numbers, i.e. 10.1 becomes 100001
|
|
components.insert(1, 0)
|
|
|
|
return int(''.join('{0:02d}'.format(c) for c in components))
|
|
|
|
@staticmethod
|
|
def postgres_major_version_to_int(pg_version):
|
|
"""
|
|
>>> Postgresql.postgres_major_version_to_int('10')
|
|
100000
|
|
>>> Postgresql.postgres_major_version_to_int('9.6')
|
|
90600
|
|
"""
|
|
return Postgresql.postgres_version_to_int(pg_version + '.0')
|
|
|
|
def read_postmaster_opts(self):
|
|
"""returns the list of option names/values from postgres.opts, Empty dict if read failed or no file"""
|
|
result = {}
|
|
try:
|
|
with open(os.path.join(self._data_dir, 'postmaster.opts')) as f:
|
|
data = f.read()
|
|
for opt in data.split('" "'):
|
|
if '=' in opt and opt.startswith('--'):
|
|
name, val = opt.split('=', 1)
|
|
result[name.strip('-')] = val.rstrip('"\n')
|
|
except IOError:
|
|
logger.exception('Error when reading postmaster.opts')
|
|
return result
|
|
|
|
def single_user_mode(self, command=None, options=None):
|
|
"""run a given command in a single-user mode. If the command is empty - then just start and stop"""
|
|
cmd = [self._pgcommand('postgres'), '--single', '-D', self._data_dir]
|
|
for opt, val in sorted((options or {}).items()):
|
|
cmd.extend(['-c', '{0}={1}'.format(opt, val)])
|
|
# need a database name to connect
|
|
cmd.append(self._database)
|
|
return self.cancellable_subprocess_call(cmd, communicate_input=command)
|
|
|
|
def cleanup_archive_status(self):
|
|
status_dir = os.path.join(self._data_dir, 'pg_' + self.wal_name, 'archive_status')
|
|
try:
|
|
for f in os.listdir(status_dir):
|
|
path = os.path.join(status_dir, f)
|
|
try:
|
|
if os.path.islink(path):
|
|
os.unlink(path)
|
|
elif os.path.isfile(path):
|
|
os.remove(path)
|
|
except OSError:
|
|
logger.exception('Unable to remove %s', path)
|
|
except OSError:
|
|
logger.exception('Unable to list %s', status_dir)
|
|
|
|
def fix_cluster_state(self):
|
|
self.cleanup_archive_status()
|
|
|
|
# Start in a single user mode and stop to produce a clean shutdown
|
|
opts = self.read_postmaster_opts()
|
|
opts.update({'archive_mode': 'on', 'archive_command': 'false'})
|
|
if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf):
|
|
os.unlink(self._recovery_conf)
|
|
return self.single_user_mode(options=opts) == 0 or None
|
|
|
|
def cancellable_subprocess_call(self, *args, **kwargs):
|
|
for s in ('stdin', 'stdout', 'stderr'):
|
|
kwargs.pop(s, None)
|
|
|
|
communicate_input = 'communicate_input' in kwargs
|
|
if communicate_input:
|
|
input_data = kwargs.pop('communicate_input', None)
|
|
if not isinstance(input_data, string_types):
|
|
input_data = ''
|
|
if input_data and input_data[-1] != '\n':
|
|
input_data += '\n'
|
|
kwargs['stdin'] = subprocess.PIPE
|
|
kwargs['stdout'] = open(os.devnull, 'w')
|
|
kwargs['stderr'] = subprocess.STDOUT
|
|
|
|
try:
|
|
with self._cancellable_lock:
|
|
if self._is_cancelled:
|
|
raise PostgresException('cancelled')
|
|
|
|
self._is_cancelled = False
|
|
self._cancellable = subprocess.Popen(*args, **kwargs)
|
|
|
|
if communicate_input:
|
|
if input_data:
|
|
self._cancellable.communicate(input_data)
|
|
self._cancellable.stdin.close()
|
|
|
|
return self._cancellable.wait()
|
|
finally:
|
|
with self._cancellable_lock:
|
|
self._cancellable = None
|
|
|
|
def reset_is_cancelled(self):
|
|
with self._cancellable_lock:
|
|
self._is_cancelled = False
|
|
|
|
def cancel(self):
|
|
with self._cancellable_lock:
|
|
self._is_cancelled = True
|
|
if self._cancellable is None or self._cancellable.returncode is not None:
|
|
return
|
|
self._cancellable.terminate()
|
|
|
|
for _ in polling_loop(10):
|
|
with self._cancellable_lock:
|
|
if self._cancellable is None or self._cancellable.returncode is not None:
|
|
return
|
|
|
|
with self._cancellable_lock:
|
|
if self._cancellable is not None and self._cancellable.returncode is None:
|
|
self._cancellable.kill()
|
|
|
|
def schedule_sanity_checks_after_pause(self):
|
|
"""
|
|
After coming out of pause we have to:
|
|
1. sync replication slots, because it might happen that slots were removed
|
|
2. get new 'Database system identifier' to make sure that it wasn't changed
|
|
"""
|
|
self._schedule_load_slots = self.use_slots
|
|
self._sysid = None
|