From 7827951c8cfe36fa32cabad6495673c88e4cf1b9 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 25 May 2016 14:17:05 +0200 Subject: [PATCH] Dynamic configuration --- features/environment.py | 6 +- patroni/__init__.py | 72 ++++++++----- patroni/config.py | 168 +++++++++++++++++++++++++++++ patroni/ctl.py | 3 +- patroni/dcs/__init__.py | 42 ++++++-- patroni/dcs/consul.py | 18 +++- patroni/dcs/etcd.py | 18 +++- patroni/dcs/zookeeper.py | 27 +++-- patroni/ha.py | 11 +- patroni/postgresql.py | 225 ++++++++++++++++++++++++--------------- postgres0.yml | 148 ++++++++++--------------- tests/test_api.py | 3 +- tests/test_config.py | 30 ++++++ tests/test_consul.py | 6 +- tests/test_ctl.py | 2 +- tests/test_etcd.py | 5 +- tests/test_ha.py | 26 ++++- tests/test_patroni.py | 16 ++- tests/test_postgresql.py | 87 +++++++++------ tests/test_zookeeper.py | 10 +- 20 files changed, 642 insertions(+), 281 deletions(-) create mode 100644 patroni/config.py create mode 100644 tests/test_config.py diff --git a/features/environment.py b/features/environment.py index bc1106f4..bd03df46 100644 --- a/features/environment.py +++ b/features/environment.py @@ -116,11 +116,12 @@ class PatroniController(AbstractController): config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT) - user = config['postgresql'].get('superuser', {}) + user = config['postgresql'].get('authentication', config['postgresql']).get('superuser', {}) self._connkwargs = {k: user[n] for n, k in [('username', 'user'), ('password', 'password')] if n in user} self._connkwargs.update({'host': host, 'port': self.__PORT, 'database': 'postgres'}) - config['postgresql'].update({'name': name, 'data_dir': self._data_dir}) + config['name'] = name + config['postgresql']['data_dir'] = self._data_dir config['postgresql']['parameters'].update({ 'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir, 'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1'}) @@ -135,7 +136,6 @@ class PatroniController(AbstractController): if dcs == 'consul': config[dcs] = dcs_config else: - dcs_config.update({'session_timeout': dcs_config.pop('ttl'), 'reconnect_timeout': config['loop_wait']}) if dcs == 'exhibitor': dcs_config['exhibitor'] = {'hosts': ['127.0.0.1'], 'port': 8181} else: diff --git a/patroni/__init__.py b/patroni/__init__.py index 1f6a2ab1..2addc264 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -3,10 +3,11 @@ import os import signal import sys import time -import yaml from patroni.api import RestApiServer +from patroni.config import Config from patroni.dcs import get_dcs +from patroni.exceptions import DCSError from patroni.ha import Ha from patroni.postgresql import Postgresql from patroni.utils import reap_children, set_ignore_sigterm, setup_signal_handlers @@ -19,43 +20,50 @@ class Patroni(object): PATRONI_CONFIG_VARIABLE = 'PATRONI_CONFIGURATION' def __init__(self, config_file=None, config_env=None): - self._config_file = config_file - config = yaml.load(config_env) if config_env else self._load_config() - - self.nap_time = config['loop_wait'] - self.tags = self.get_tags(config) - self.postgresql = Postgresql(config['postgresql']) - self.dcs = get_dcs(self.postgresql.name, config) self.version = __version__ - self.api = RestApiServer(self, config['restapi']) + self.config = Config(config_file=config_file, config_env=config_env) + self.dcs = get_dcs(self.config) + self.load_dynamic_configuration() + + self.postgresql = Postgresql(self.config['postgresql']) + self.api = RestApiServer(self, self.config['restapi']) self.ha = Ha(self) + + self.tags = self.get_tags() + self.nap_time = self.config['loop_wait'] self.next_run = time.time() self._reload_config_scheduled = False + self._received_sighup = False - @staticmethod - def get_tags(config): - return {tag: value for tag, value in config.get('tags', {}).items() + def load_dynamic_configuration(self): + while True: + try: + cluster = self.dcs.get_cluster() + if cluster and cluster.config: + self.config.set_dynamic_configuration(cluster.config.data) + elif not self.config.dynamic_configuration and 'bootstrap' in self.config: + self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']) + break + except DCSError: + logger.warning('Can not get cluster from dcs') + + def get_tags(self): + return {tag: value for tag, value in self.config.get('tags', {}).items() if tag not in ('clonefrom', 'nofailover', 'noloadbalance') or value} - def _load_config(self): - with open(self._config_file) as f: - return yaml.load(f) - def reload_config(self): try: - config = self._load_config() - self.tags = self.get_tags(config) - self.nap_time = config['loop_wait'] - self.dcs.set_ttl(config.get('ttl') or 30) - self.api.reload_config(config['restapi']) - self.postgresql.reload_config(config['postgresql']) + self.tags = self.get_tags() + self.nap_time = self.config['loop_wait'] + self.dcs.set_ttl(self.config.get('ttl') or 30) + self.api.reload_config(self.config['restapi']) + self.postgresql.reload_config(self.config['postgresql']) except Exception: - logger.exception('Failed to reload config_file=%s', self._config_file) - self._reload_config_scheduled = False + logger.exception('Failed to reload config_file=%s', self.config.config_file) def sighup_handler(self, *args): - self._reload_config_scheduled = True + self._received_sighup = True @property def noloadbalance(self): @@ -84,9 +92,21 @@ class Patroni(object): self.next_run = time.time() while True: - if self._reload_config_scheduled: + if self._received_sighup: + self._received_sighup = False + self.config.reload_local_configuration() self.reload_config() + logger.info(self.ha.run_cycle()) + + cluster = self.dcs.cluster + if cluster and cluster.config and cluster.config.data and \ + self.config.set_dynamic_configuration(cluster.config.data): + self.reload_config() + + if not self.postgresql.data_directory_empty(): + self.config.save_cache() + reap_children() self.schedule_next_run() diff --git a/patroni/config.py b/patroni/config.py new file mode 100644 index 00000000..cf5ba85d --- /dev/null +++ b/patroni/config.py @@ -0,0 +1,168 @@ +import json +import logging +import os +import tempfile +import yaml + +from copy import deepcopy +from patroni.postgresql import Postgresql + +logger = logging.getLogger(__name__) + + +class Config(object): + """ + This class is responsible for: + + 1) Building and giving access to `effective_configuration` from: + * `Config.__DEFAULT_CONFIG` -- some sane default values + * `dynamic_configuration` -- configuration stored in DCS + * `local_configuration` -- configuration from `config.yml` or environment + + 2) Saving and loading `dynamic_configuration` into 'patroni.dynamic.json' file + located in local_configuration['postgresql']['data_dir'] directory. + This is necessary to be able to restore `dynamic_configuration` + if DCS was accidentally wiped + + 3) Loading of configuration file in the old format and converting it into new format + + 4) Mimicking some of the `dict` interfaces to make it possible + to work with it as with the old `config` object. + """ + + __CACHE_FILENAME = 'patroni.dynamic.json' + __DEFAULT_CONFIG = { + 'ttl': 30, 'loop_wait': 10, 'retry_timeout': 5, + 'maximum_lag_on_failover': 1048576, + 'postgresql': { + 'parameters': Postgresql.CMDLINE_OPTIONS + } + } + + def __init__(self, config_file=None, config_env=None): + self._config_file = None if config_env else config_file + self._dynamic_configuration = {} + self._local_configuration = yaml.safe_load(config_env) if config_env else self._load_config_file() + self._build_effective_configuration() + self._data_dir = self.__effective_configuration['postgresql']['data_dir'] + self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME) + self._load_cache() + self._cache_needs_saving = False + + @property + def config_file(self): + return self._config_file + + @property + def dynamic_configuration(self): + return deepcopy(self._dynamic_configuration) + + def _load_config_file(self): + with open(self._config_file) as f: + return yaml.safe_load(f) + + def _load_cache(self): + if os.path.isfile(self._cache_file): + try: + with open(self._cache_file) as f: + self.set_dynamic_configuration(json.load(f)) + except Exception: + logger.exception('Exception when loading file: %s', self._cache_file) + + def save_cache(self): + if self._cache_needs_saving: + tmpfile = fd = None + try: + (fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir) + with os.fdopen(fd, 'w') as f: + fd = None + json.dump(self.dynamic_configuration, f) + tmpfile = os.rename(tmpfile, self._cache_file) + self._cache_needs_saving = False + except Exception: + logger.exception('Exception when saving file: %s', self._cache_file) + if fd: + try: + os.close(fd) + except Exception: + logger.error('Can not close temporary file %s', tmpfile) + if tmpfile and os.path.exists(tmpfile): + try: + os.remove(tmpfile) + except Exception: + logger.error('Can not remove temporary file %s', tmpfile) + + def set_dynamic_configuration(self, configuration): + if configuration and self._dynamic_configuration != configuration: + self._dynamic_configuration = configuration + self._build_effective_configuration() + self._cache_needs_saving = True + return True + + def reload_local_configuration(self): + self._local_configuration = self._load_config_file() + self._build_effective_configuration() + + def _process_postgresql_parameters(self, parameters, is_local=False): + ret = {} + for name, value in (parameters or {}).items(): + if (is_local and name not in self.__DEFAULT_CONFIG['postgresql']['parameters']) \ + or not ((name == 'wal_level' and value not in ('hot_standby', 'logical')) or + (name in ('max_replication_slots', 'max_wal_senders', 'wal_keep_segments') and + int(value) < self.__DEFAULT_CONFIG['postgresql']['parameters'][name]) or + name in ('hot_standby', 'wal_log_hints')): + ret[name] = value + return ret + + def _safe_copy_dynamic_configuration(self): + config = deepcopy(self.__DEFAULT_CONFIG) + + for name, value in self._dynamic_configuration.items(): + if name == 'postgresql': + for name, value in (value or {}).items(): + if name == 'parameters': + config['postgresql'][name].update(self._process_postgresql_parameters(value)) + elif name not in ('connect_address', 'listen', 'data_dir', 'pgpass', 'authentication'): + config['postgresql'][name] = deepcopy(value) + elif name in config: + config[name] = value + return config + + def _build_effective_configuration(self): + config = self._safe_copy_dynamic_configuration() + for name, value in self._local_configuration.items(): + if name == 'postgresql': + for name, value in (value or {}).items(): + if name == 'parameters': + config['postgresql'][name].update(self._process_postgresql_parameters(value, True)) + else: + config['postgresql'][name] = deepcopy(value) + elif name not in config: + config[name] = deepcopy(value) if value else {} + + pg_config = config['postgresql'] + + # special treatment for old config + if 'authentication' not in pg_config: + pg_config['use_pg_rewind'] = 'pg_rewind' in pg_config + pg_config['authentication'] = {u: pg_config[u] for u in ('replication', 'superuser') if u in pg_config} + + if 'superuser' not in pg_config['authentication'] and 'pg_rewind' in pg_config: + pg_config['authentication']['superuser'] = pg_config['pg_rewind'] + + if 'name' not in config and 'name' in pg_config: + config['name'] = pg_config['name'] + + pg_config.update({p: config[p] for p in ('name', 'scope', 'retry_timeout', + 'maximum_lag_on_failover') if p in config}) + + self.__effective_configuration = config + + def get(self, key, default=None): + return self.__effective_configuration.get(key, default) + + def __contains__(self, key): + return key in self.__effective_configuration + + def __getitem__(self, key): + return self.__effective_configuration[key] diff --git a/patroni/ctl.py b/patroni/ctl.py index 19b48feb..08622086 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -95,8 +95,9 @@ def ctl(ctx): def get_dcs(config, scope): config.setdefault('scope', scope) + config.setdefault('name', scope) try: - return _get_dcs(scope, config) + return _get_dcs(config) except PatroniException as e: raise PatroniCtlException(str(e)) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 2b543ff4..fcf258b7 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -30,7 +30,7 @@ def parse_connection_string(value): return conn_url, api_url -def get_dcs(node_name, config): +def get_dcs(config): available_implementations = [] for name in os.listdir(os.path.dirname(__file__)): if name.endswith('.py') and not name.startswith('__'): # find module @@ -44,8 +44,9 @@ def get_dcs(node_name, config): available_implementations.append(name) if name in config: # which has configuration section in the config file # propagate some parameters - config[name].update({p: config[p] for p in ('namespace', 'scope', 'ttl') if p in config}) - return value(node_name, config[name]) + config[name].update({p: config[p] for p in ('namespace', 'name', + 'scope', 'ttl', 'retry_timeout') if p in config}) + return value(config[name]) raise PatroniException("""Can not find suitable configuration of distributed configuration store Available implementations: """ + ', '.join(available_implementations)) @@ -163,11 +164,28 @@ class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at')) -class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')): +class ClusterConfig(namedtuple('ClusterConfig', 'index,data')): + + @staticmethod + def from_node(index, data): + """ + >>> ClusterConfig.from_node(1, '{').data + {} + """ + + try: + data = json.loads(data) + except (TypeError, ValueError): + data = {} + return ClusterConfig(index, data) + + +class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover')): """Immutable object (namedtuple) which represents PostgreSQL cluster. Consists of the following fields: :param initialize: boolean, shows whether this cluster has initialization key stored in DC or not. + :param config: global dynamic configuration, reference to `ClusterConfig` object :param leader: `Leader` object which represents current leader of the cluster :param last_leader_operation: int or long object containing position of last known leader operation. This value is stored in `/optime/leader` key @@ -192,19 +210,19 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem class AbstractDCS(object): _INITIALIZE = 'initialize' + _CONFIG = 'config' _LEADER = 'leader' _FAILOVER = 'failover' _MEMBERS = 'members/' _OPTIME = 'optime' _LEADER_OPTIME = _OPTIME + '/' + _LEADER - def __init__(self, name, config): + def __init__(self, config): """ - :param name: name of current instance (the same value as `~Postgresql.name`) :param config: dict, reference to config section of selected DCS. i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc... """ - self._name = name + self._name = config['name'] self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/')) self._base_path = '/'.join([self._namespace, config['scope']]) @@ -219,6 +237,10 @@ class AbstractDCS(object): def initialize_path(self): return self.client_path(self._INITIALIZE) + @property + def config_path(self): + return self.client_path(self._CONFIG) + @property def members_path(self): return self.client_path(self._MEMBERS) @@ -310,7 +332,11 @@ class AbstractDCS(object): if scheduled_at: failover_value['scheduled_at'] = scheduled_at.isoformat() - return self.set_failover_value(json.dumps(failover_value), index) + return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), index) + + @abc.abstractmethod + def set_config_value(self, value, index=None): + """Create or update `/config` key""" @abc.abstractmethod def touch_member(self, data, ttl=None): diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 81f430bc..e941d025 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -5,7 +5,7 @@ import time import six from consul import ConsulException, NotFound, base, std -from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member from patroni.exceptions import DCSError from patroni.utils import sleep from requests.exceptions import RequestException @@ -70,8 +70,8 @@ def catch_consul_errors(func): class Consul(AbstractDCS): - def __init__(self, name, config): - super(Consul, self).__init__(name, config) + def __init__(self, config): + super(Consul, self).__init__(config) self._ttl = None self._session = None self._my_member_data = None @@ -139,6 +139,10 @@ class Consul(AbstractDCS): initialize = nodes.get(self._INITIALIZE) initialize = initialize and initialize['Value'] + # get global dynamic configuration + config = nodes.get(self._CONFIG) + config = config and ClusterConfig.from_node(config['ModifyIndex'], config['Value']) + # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME) last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['Value']) @@ -163,9 +167,9 @@ class Consul(AbstractDCS): if failover: failover = Failover.from_node(failover['ModifyIndex'], failover['Value']) - self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover) except NotFound: - self._cluster = Cluster(False, None, None, [], None) + self._cluster = Cluster(False, None, None, None, [], None) except: logger.exception('get_cluster') raise ConsulError('Consul is not responding properly') @@ -205,6 +209,10 @@ class Consul(AbstractDCS): def set_failover_value(self, value, index=None): return self._client.kv.put(self.failover_path, value, cas=index) + @catch_consul_errors + def set_config_value(self, value, index=None): + return self._client.kv.put(self.config_path, value, cas=index) + @catch_consul_errors def write_leader_optime(self, last_operation): return self._client.kv.put(self.leader_optime_path, last_operation) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index f06cca3d..0c64cbfd 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -9,7 +9,7 @@ import time from dns.exception import DNSException from dns import resolver -from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member from patroni.exceptions import DCSError from patroni.utils import Retry, RetryFailedError, sleep from urllib3.exceptions import HTTPError, ReadTimeoutError @@ -191,8 +191,8 @@ def catch_etcd_errors(func): class Etcd(AbstractDCS): - def __init__(self, name, config): - super(Etcd, self).__init__(name, config) + def __init__(self, config): + super(Etcd, self).__init__(config) self.set_ttl(config.get('ttl', 30)) self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, retry_exceptions=(etcd.EtcdConnectionFailed, @@ -231,6 +231,10 @@ class Etcd(AbstractDCS): initialize = nodes.get(self._INITIALIZE) initialize = initialize and initialize.value + # get global dynamic configuration + config = nodes.get(self._CONFIG) + config = config and ClusterConfig.from_node(config.modifiedIndex, config.value) + # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME) last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value) @@ -250,9 +254,9 @@ class Etcd(AbstractDCS): if failover: failover = Failover.from_node(failover.modifiedIndex, failover.value) - self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover) + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover) except etcd.EtcdKeyNotFound: - self._cluster = Cluster(False, None, None, [], None) + self._cluster = Cluster(False, None, None, None, [], None) except: logger.exception('get_cluster') raise EtcdError('Etcd is not responding properly') @@ -278,6 +282,10 @@ class Etcd(AbstractDCS): def set_failover_value(self, value, index=None): return self._client.write(self.failover_path, value, prevIndex=index or 0) + @catch_etcd_errors + def set_config_value(self, value, index=None): + return self._client.write(self.config_path, value, prevIndex=index or 0) + @catch_etcd_errors def write_leader_optime(self, last_operation): return self._client.set(self.leader_optime_path, last_operation) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 22f2f5ad..b374ab51 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -5,7 +5,7 @@ import time from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError -from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member from patroni.exceptions import DCSError from patroni.utils import sleep from requests.exceptions import RequestException @@ -69,8 +69,8 @@ class ExhibitorEnsembleProvider(object): class ZooKeeper(AbstractDCS): - def __init__(self, name, config): - super(ZooKeeper, self).__init__(name, config) + def __init__(self, config): + super(ZooKeeper, self).__init__(config) hosts = config.get('hosts', []) if isinstance(hosts, list): @@ -83,7 +83,7 @@ class ZooKeeper(AbstractDCS): self.exhibitor = ExhibitorEnsembleProvider(exhibitor['hosts'], exhibitor['port'], poll_interval=interval) hosts = self.exhibitor.zookeeper_hosts - self._client = KazooClient(hosts=hosts, timeout=(config.get('session_timeout') or 30), + self._client = KazooClient(hosts=hosts, timeout=(config.get('session_timeout') or config.get('ttl') or 30), command_retry={'deadline': (config.get('reconnect_timeout') or 10), 'max_delay': 1, 'max_tries': -1}, connection_retry={'max_delay': 1, 'max_tries': -1}) @@ -146,6 +146,10 @@ class ZooKeeper(AbstractDCS): # get initialize flag initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None + # get global dynamic configuration + config = self.get_node(self.config_path, watch=self.cluster_watcher) if self._CONFIG in nodes else None + config = config and ClusterConfig.from_node(config[1].version, config[0]) + # get list of members members = self.load_members() if self._MEMBERS[:-1] in nodes else [] @@ -166,13 +170,12 @@ class ZooKeeper(AbstractDCS): # failover key failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None - if failover: - failover = Failover.from_node(failover[1].version, failover[0]) + failover = failover and Failover.from_node(failover[1].version, failover[0]) # get last leader operation optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self._fetch_cluster else None self._last_leader_operation = 0 if optime is None else int(optime[0]) - self._cluster = Cluster(initialize, leader, self._last_leader_operation, members, failover) + self._cluster = Cluster(initialize, config, leader, self._last_leader_operation, members, failover) def _load_cluster(self): if self.exhibitor and self.exhibitor.poll(): @@ -209,6 +212,16 @@ class ZooKeeper(AbstractDCS): logging.exception('set_failover_value') return False + def set_config_value(self, value, index=None): + try: + self._client.retry(self._client.set, self.config_path, value.encode('utf-8'), version=index or -1) + return True + except NoNodeError: + return value == '' or (not index and self._create(self.config_path, value)) + except Exception: + logging.exception('set_config_value') + return False + def initialize(self, create_new=True, sysid=""): return self._create(self.initialize_path, sysid, makepath=True) if create_new \ else self._client.retry(self._client.set, self.initialize_path, sysid.encode("utf-8")) diff --git a/patroni/ha.py b/patroni/ha.py index e1772eb5..ad791be5 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -86,10 +86,11 @@ class Ha(object): self._async_executor.schedule('bootstrap {0}'.format(msg)) self._async_executor.run_async(self.clone, args=(clone_member, msg)) return 'trying to bootstrap {0}'.format(msg) - elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key + # no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file + elif not (self.cluster.initialize or self.patroni.nofailover) and 'bootstrap' in self.patroni.config: if self.dcs.initialize(create_new=True): # race for initialization try: - self.state_handler.bootstrap() + self.state_handler.bootstrap(self.patroni.config['bootstrap']) self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid) except: # initdb or start failed # remove initialization key and give a chance to other members @@ -98,6 +99,7 @@ class Ha(object): self.state_handler.stop('immediate') self.state_handler.move_data_directory() raise + self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':'))) self.dcs.take_leader() self.load_cluster_from_dcs() return 'initialized a new cluster' @@ -440,9 +442,12 @@ class Ha(object): self.touch_member() # cluster has leader key but not initialize key - if not self.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock(): + if not (self.cluster.is_unlocked() or self.sysid_valid(self.cluster.initialize)) and self.has_lock(): self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) + if not (self.cluster.is_unlocked() or self.cluster.config and self.cluster.config.data) and self.has_lock(): + self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':'))) + if self._async_executor.busy: return self.handle_long_action_in_progress() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 1197aa08..35aecae7 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -41,27 +41,44 @@ def parseurl(url): 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. + CMDLINE_OPTIONS = { + 'listen_addresses': None, + 'port': None, + 'wal_level': 'hot_standby', + 'hot_standby': 'on', + 'max_wal_senders': 5, + 'wal_keep_segments': 8, + 'max_replication_slots': 5, + 'wal_log_hints': 'on' + } + def __init__(self, config): self.config = config self.name = config['name'] - self._restart_pending = False - self._server_parameters = self.get_server_parameters(config) - self._listen_addresses, self._port = (config['listen'] + ':5432').split(':')[:2] - self._connect_address = config.get('connect_address') - self.replication = config['replication'] - self.resolve_connection_addresses() - self.scope = config['scope'] self._data_dir = config['data_dir'] - self.superuser = config.get('superuser') or {} - self.admin = config.get('admin') or {} + self._restart_pending = False + self._server_parameters = self.get_server_parameters(config) - self.initdb_options = config.get('initdb') or [] - self.pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass') - self.pg_rewind = config.get('pg_rewind') or {} - self.callback = config.get('callbacks') or {} - self.use_slots = config.get('use_slots', True) + self._connect_address = config.get('connect_address') + self._superuser = config['authentication'].get('superuser', {}) + self._replication = config['authentication']['replication'] + self.resolve_connection_addresses() + + self._use_pg_rewind = config.get('use_pg_rewind', False) + self._use_slots = config.get('use_slots', True) + self._major_version = self.get_major_version() self._schedule_load_slots = self.use_slots + + self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass') + self.callback = config.get('callbacks') or {} self._postgresql_conf = os.path.join(self._data_dir, 'postgresql.conf') self._postgresql_base_conf_name = 'postgresql.base.conf' self._postgresql_base_conf = os.path.join(self._data_dir, self._postgresql_base_conf_name) @@ -90,39 +107,58 @@ class Postgresql(object): self.set_role('master' if self.is_leader() else 'replica') self._write_postgresql_conf() # we are "joining" already running postgres - @staticmethod - def get_server_parameters(config): - return {p: v for p, v in (config.get('parameters') or {}).items() if p not in ('listen_addresses', 'port')} + @property + def use_slots(self): + return self._use_slots and self._major_version >= 9.4 + + def get_major_version(self): + if not self.data_directory_empty(): + try: + with open(os.path.join(self._data_dir, 'PG_VERSION')) as f: + return float(f.read()) + except Exception: + logger.exception('Failed to read PG_VERSION from %s', self._data_dir) + return 0.0 + + def get_server_parameters(self, config): + parameters = config['parameters'].copy() + listen_addresses, port = (config['listen'] + ':5432').split(':')[:2] + parameters.update({'listen_addresses': listen_addresses, 'port': port}) + return parameters def resolve_connection_addresses(self): - self.local_address = self.get_local_address() + self._local_address = self.get_local_address() self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format( - connect_address=self._connect_address or self.local_address, **self.replication) + connect_address=self._connect_address or self._local_address, **self._replication) def reload_config(self, config): server_parameters = self.get_server_parameters(config) - self._connect_address = config.get('connect_address') - listen_addresses, port = (config['listen'] + ':5432').split(':')[:2] - if self._listen_addresses == listen_addresses and self._port == port: - self.resolve_connection_addresses() - self._listen_addresses = listen_addresses - self._port = port + listen_address_changed = reload_pending = False if self.is_healthy(): changes = server_parameters.copy() changes.update({p: None for p, v in self._server_parameters.items() if p not in server_parameters}) if changes: - for r in self.query("""SELECT name, setting + for r in self.query("""SELECT name, setting, context FROM pg_settings - WHERE context in ('internal', 'postmaster') - AND name IN (""" + ', '.join('%s' for _ in changes.keys()) + ')', + WHERE name IN (""" + ', '.join('%s' for _ in changes.keys()) + ')', *(list(changes.keys()))): if server_parameters[r[0]] is None or str(server_parameters[r[0]]) != str(r[1]): - self._restart_pending = True - break + reload_pending = True + if r[2] in ('internal', 'postmaster'): + self._restart_pending = True + if r[0] in ('listen_addresses', 'port'): + listen_address_changed = True + self.config = config self._server_parameters = server_parameters - self._write_postgresql_conf() - self.reload() + self._connect_address = config.get('connect_address') + + if not listen_address_changed: + self.resolve_connection_addresses() + + if reload_pending: + self._write_postgresql_conf() + self.reload() @property def restart_pending(self): @@ -134,8 +170,7 @@ class Postgresql(object): we have either wal_log_hints or checksums turned on """ # low-hanging fruit: check if pg_rewind configuration is there - if not self.pg_rewind or\ - not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')): + if not (self._use_pg_rewind and all(self._superuser.get(n) for n in ('username', 'password'))): return False cmd = ['pg_rewind', '--help'] @@ -157,14 +192,14 @@ class Postgresql(object): return self._sysid def get_local_address(self): - listen_addresses = self._listen_addresses.split(',') + listen_addresses = self._server_parameters['listen_addresses'].split(',') local_address = listen_addresses[0].strip() # take first address from listen_addresses for la in listen_addresses: - if la.strip() in ['*', '0.0.0.0']: # we are listening on * + if la.strip() in ('*', '0.0.0.0', '127.0.0.1', 'localhost'): # we are listening on '*' or localhost local_address = 'localhost' # connection via localhost is preferred break - return local_address + ':' + self._port + return local_address + ':' + self._server_parameters['port'] def get_postgres_role_from_data_directory(self): if self.data_directory_empty(): @@ -176,11 +211,11 @@ class Postgresql(object): @property def _connect_kwargs(self): - r = parseurl('postgres://{0}/postgres'.format(self.local_address)) - if 'username' in self.superuser: - r['user'] = self.superuser['username'] - if 'password' in self.superuser: - r['password'] = self.superuser['password'] + r = parseurl('postgres://{0}/postgres'.format(self._local_address)) + if 'username' in self._superuser: + r['user'] = self._superuser['username'] + if 'password' in self._superuser: + r['password'] = self._superuser['password'] return r def connection(self): @@ -229,9 +264,9 @@ class Postgresql(object): raise Exception('{0} option for initdb is not allowed'.format(name)) return True - def get_initdb_options(self): + def get_initdb_options(self, config): options = [] - for o in self.initdb_options: + for o in config: if isinstance(o, string_types) and self.initdb_allowed_option(o): options.append('--{0}'.format(o)) elif isinstance(o, dict): @@ -243,17 +278,17 @@ class Postgresql(object): raise Exception('Unknown type of initdb option: {0}'.format(o)) return options - def initialize(self): + def _initialize(self, config): self.set_state('initalizing new cluster') - options = self.get_initdb_options() + options = self.get_initdb_options(config.get('initdb') or []) pwfile = None - if self.superuser: - if 'username' in self.superuser: - options.append('--username={0}'.format(self.superuser['username'])) - if 'password' in self.superuser: + 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.write(fd, self._superuser['password'].encode('utf-8')) os.close(fd) options.append('--pwfile={0}'.format(pwfile)) @@ -261,7 +296,8 @@ class Postgresql(object): if pwfile: os.remove(pwfile) if ret: - self.write_pg_hba() + self.write_pg_hba(config.get('pg_hba', [])) + self._major_version = self.get_major_version() else: self.set_state('initdb failed') return ret @@ -271,12 +307,12 @@ class Postgresql(object): os.unlink(self._trigger_file) def write_pgpass(self, record): - with open(self.pgpass, 'w') as f: + with open(self._pgpass, 'w') as f: os.fchmod(f.fileno(), 0o600) f.write('{host}:{port}:*:{user}:{password}\n'.format(**record)) env = os.environ.copy() - env['PGPASSFILE'] = self.pgpass + env['PGPASSFILE'] = self._pgpass return env def replica_method_can_work_without_replication_connection(self, method): @@ -405,16 +441,20 @@ class Postgresql(object): env = {'PATH': os.environ.get('PATH')} # pg_ctl will write a FATAL if the username is incorrect. exporting PGUSER if necessary - if 'username' in self.superuser and self.superuser['username'] != os.environ.get('USER'): - env['PGUSER'] = self.superuser['username'] + if 'username' in self._superuser and self._superuser['username'] != os.environ.get('USER'): + env['PGUSER'] = self._superuser['username'] + self._write_postgresql_conf() - server_arguments = ['-o', "--listen_addresses='{0}' --port={1}".format(self._listen_addresses, self._port)] - ret = subprocess.call(self._pg_ctl + ['start'] + server_arguments, env=env, preexec_fn=os.setsid) == 0 + self.resolve_connection_addresses() + + options = ' '.join("--{0}='{1}'".format(p, self._server_parameters[p]) for p in self.CMDLINE_OPTIONS + if not (self._major_version < 9.4 and p in ('max_replication_slots', 'wal_log_hints'))) + + ret = subprocess.call(self._pg_ctl + ['start', '-o', options], env=env, preexec_fn=os.setsid) == 0 self._restart_pending = False self.set_state('running' if ret else 'start failed') if ret: - self.resolve_connection_addresses() self._schedule_load_slots = self.use_slots self.save_configuration_files() @@ -488,8 +528,9 @@ class Postgresql(object): with open(self._postgresql_conf, 'w') as f: f.write('# Do not edit this file manually!\n# It will be overwritten by Patroni!\n') f.write("include '{0}'\n\n".format(self._postgresql_base_conf_name)) - for setting, value in sorted(self._server_parameters.items()): - f.write("{0} = '{1}'\n".format(setting, value)) + for name, value in sorted(self._server_parameters.items()): + if name not in self.CMDLINE_OPTIONS: + f.write("{0} = '{1}'\n".format(name, value)) def is_healthy(self): if not self.is_running(): @@ -500,9 +541,9 @@ class Postgresql(object): def check_replication_lag(self, last_leader_operation): return (last_leader_operation or 0) - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0) - def write_pg_hba(self): + def write_pg_hba(self, config): with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f: - f.write('\n{}\n'.format('\n'.join(self.config.get('pg_hba', [])))) + f.write('\n{}\n'.format('\n'.join(config))) def primary_conninfo(self, leader_url): r = parseurl(leader_url) @@ -536,7 +577,7 @@ class Postgresql(object): def rewind(self, leader): # prepare pg_rewind connection r = parseurl(leader.conn_url) - r.update(self.pg_rewind) + r.update(self._superuser) r['user'] = r.pop('username') env = self.write_pgpass(r) pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) @@ -546,12 +587,9 @@ class Postgresql(object): logger.info("running pg_rewind from %s", pc) pg_rewind = ['pg_rewind', '-D', self._data_dir, '--source-server', pc] try: - ret = subprocess.call(pg_rewind, env=env) == 0 + return subprocess.call(pg_rewind, env=env) == 0 except OSError: - ret = False - if ret: - self.write_recovery_conf(leader) - return ret + return False def controldata(self): """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ @@ -566,6 +604,22 @@ class Postgresql(object): logger.exception("Error when calling pg_controldata") return result + 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() + opts = [opt.strip('"\n') for opt in data.split(' "')] + for opt in opts: + if '=' in opt and opt.startswith('--'): + name, val = opt.split('=', 1) + name = name.strip('-') + result[name] = val + 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 = ['postgres', '--single', '-D', self._data_dir] @@ -603,7 +657,6 @@ class Postgresql(object): need_rewind = change_role and self.can_rewind if need_rewind: logger.info("set the rewind flag after demote") - self.write_recovery_conf(leader) if leader and need_rewind: # we have a leader and need to rewind if self.is_running(): self.stop() @@ -613,20 +666,24 @@ class Postgresql(object): # XXX: if recovery.conf is linked, it will be written anew as a normal file. if os.path.islink(self._recovery_conf): os.unlink(self._recovery_conf) - else: + elif os.path.isfile(self._recovery_conf): os.remove(self._recovery_conf) # Archived segments might be useful to pg_rewind, # clean the flags that tell we should remove them. self.cleanup_archive_status() # Start in a single user mode and stop to produce a clean shutdown - self.single_user_mode(options={'archive_mode': 'on', 'archive_command': 'false'}) + opts = self.read_postmaster_opts() + opts.update({'archive_mode': 'on', 'archive_command': 'false'}) + self.single_user_mode(options=opts) if self.rewind(leader): + self.write_recovery_conf(leader) ret = self.start() else: logger.error("unable to rewind the former master") self.remove_data_directory() ret = True else: # do not rewind until the leader becomes available + self.write_recovery_conf(leader) ret = self.restart() if change_role and ret: self.call_nowait(ACTION_ON_ROLE_CHANGE) @@ -664,26 +721,19 @@ class Postgresql(object): self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret - def create_or_update_role(self, name, password, options): + def create_or_update_user(self, name, password, options): self.query("""DO $$ BEGIN SET local synchronous_commit = 'local'; PERFORM * FROM pg_authid WHERE rolname = %s; IF FOUND THEN - ALTER ROLE "{0}" WITH LOGIN {1} PASSWORD %s; + ALTER USER "{0}" WITH {1} PASSWORD %s; ELSE - CREATE ROLE "{0}" WITH LOGIN {1} PASSWORD %s; + CREATE USER "{0}" WITH {1} PASSWORD %s; END IF; END; $$""".format(name, options), name, password, password) - def create_replication_user(self): - self.create_or_update_role(self.replication['username'], self.replication['password'], 'REPLICATION') - - def create_connection_user(self): - if self.admin: - self.create_or_update_role(self.admin['username'], self.admin['password'], 'CREATEDB CREATEROLE') - def xlog_position(self): return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() THEN pg_last_xlog_replay_location() @@ -741,15 +791,18 @@ $$""".format(name, options), name, password, password) ret = self.create_replica(clone_member) == 0 if ret: + self._major_version = self.get_major_version() self.delete_trigger_file() self.restore_configuration_files() return ret - def bootstrap(self): + def bootstrap(self, config): """ Initialize a new node from scratch and start it. """ - if self.initialize() and self.start(): - self.create_replication_user() - self.create_connection_user() + if self._initialize(config) and self.start(): + for name, value in config['users'].items(): + if name not in (self._superuser.get('username'), self._replication['username']): + self.create_or_update_user(name, value['password'], ' '.join(value.get('options', [])).upper()) + self.create_or_update_user(self._replication['username'], self._replication['password'], 'REPLICATION') else: raise PostgresException("Could not bootstrap master PostgreSQL") diff --git a/postgres0.yml b/postgres0.yml index 22ab5a73..f6c6ca07 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -1,104 +1,70 @@ -ttl: &ttl 30 -loop_wait: &loop_wait 10 -scope: &scope batman +scope: batman +#namespace: /service/ +name: postgresql0 + restapi: listen: 127.0.0.1:8008 connect_address: 127.0.0.1:8008 -# auth: 'username:password' -# certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem -# keyfile: /etc/ssl/private/ssl-cert-snakeoil.key + etcd: - scope: *scope - ttl: *ttl host: 127.0.0.1:4001 - #discovery_srv: my-etcd.domain -#consul: -# scope: *scope -# ttl: *ttl -# host: 127.0.0.1:8500 -#zookeeper: -# scope: *scope -# session_timeout: *ttl -# reconnect_timeout: *loop_wait -# hosts: -# - 127.0.0.1:2181 -# - 127.0.0.2:2181 -# exhibitor: -# poll_interval: 300 -# port: 8181 -# hosts: -# - host1 -# - host2 -# - host3 + +bootstrap: + # this section will be written into Etcd:///config after initializing new cluster + # and all other cluster members will use it as a `global configuration` + dcs: + ttl: 30 + loop_wait: 10 + retry_timeout: 5 + maximum_lag_on_failover: 1048576 + postgresql: + use_pg_rewind: true +# use_slots: true + parameters: +# wal_level: hot_standby +# hot_standby: "on" +# wal_keep_segments: 8 +# max_wal_senders: 5 +# max_replication_slots: 5 +# wal_log_hints: "on" + archive_mode: "on" + archive_timeout: 1800s + archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f + recovery_conf: + restore_command: cp ../wal_archive/%f %p + + # some desired options for 'initdb' + initdb: # Note: It needs to be a list (some options need values, others are switches) + - encoding: UTF8 + - data-checksums + + pg_hba: # Add following lines to pg_hba.conf after running 'initdb' + - host replication replicator 127.0.0.1/32 md5 + - host all all 0.0.0.0/0 md5 +# - hostssl all all 0.0.0.0/0 md5 + + # Some additional users users which needs to be created after initializing new cluster + users: + admin: + password: admin + options: + - createrole + - createdb postgresql: - name: postgresql0 - scope: *scope listen: 127.0.0.1:5432 connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 - maximum_lag_on_failover: 1048576 # 1 megabyte in bytes - use_slots: True pgpass: /tmp/pgpass0 - initdb: ## We allow the following options to be passed on to initdb - # - auth: authmethod - # - auth-host: authmethod - # - auth-local: authmethod - - encoding: UTF8 - # - data-checksums # When pg_rewind is needed on 9.3, this needs to be enabled - # - locale: locale - # - lc-collate: locale - # - lc-ctype: locale - # - lc-messages: locale - # - lc-monetary: locale - # - lc-numeric: locale - # - lc-time: locale - # - text-search-config: CFG - # - xlogdir: directory - # - debug - # - noclean - pg_rewind: - username: postgres - password: zalando - pg_hba: - - host replication replicator 127.0.0.1/32 md5 - - host all all 0.0.0.0/0 md5 - # - hostssl all all 0.0.0.0/0 md5 - replication: - username: replicator - password: rep-pass - superuser: - username: postgres - password: zalando - admin: - username: admin - password: admin - create_replica_method: - - basebackup -# - wal_e -# commented-out example for wal-e provisioning - #wal_e: - #command: /patroni/scripts/wale_restore.py - #env_dir: /etc/wal-e.d/env - #threshold_megabytes: 10240 - #threshold_backup_size_percentage: 30 - #retries: 2 - #use_iam: 1 - #recovery_conf: - #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1 - recovery_conf: - restore_command: cp ../wal_archive/%f %p + authentication: + replication: + username: replicator + password: rep-pass + superuser: + username: postgres + password: zalando parameters: - archive_mode: "on" - wal_level: hot_standby - archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f - max_wal_senders: 10 - wal_keep_segments: 8 - archive_timeout: 1800s - max_replication_slots: 10 - hot_standby: "on" - wal_log_hints: "on" unix_socket_directories: '.' tags: - nofailover: False - noloadbalance: False - clonefrom: False + nofailover: false + noloadbalance: false + clonefrom: false diff --git a/tests/test_api.py b/tests/test_api.py index 34fbc353..23541cec 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,4 +1,5 @@ import psycopg2 +import socket import unittest from mock import Mock, patch @@ -7,7 +8,6 @@ from patroni.dcs import Member from six import BytesIO as IO from six.moves import BaseHTTPServer from six.moves.BaseHTTPServer import BaseHTTPRequestHandler -import socket from test_postgresql import psycopg2_connect, MockCursor @@ -71,6 +71,7 @@ class MockRestApiServer(RestApiServer): def __init__(self, Handler, request): self.socket = 0 + self.serve_forever = Mock() BaseHTTPServer.HTTPServer.__init__ = Mock() MockRestApiServer._BaseServer__is_shut_down = Mock() MockRestApiServer._BaseServer__shutdown_request = True diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..9c259796 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,30 @@ +import unittest + +from mock import MagicMock, Mock, patch +from patroni.config import Config +from six.moves import builtins + + +class TestConfig(unittest.TestCase): + + @patch('os.path.isfile', Mock(return_value=True)) + @patch('json.load', Mock(side_effect=Exception)) + @patch.object(builtins, 'open', MagicMock()) + def setUp(self): + self.config = Config(config_env='postgresql: {data_dir: foo}') + + def test_reload_local_configuration(self): + Config(config_file='postgres0.yml').reload_local_configuration() + + @patch('tempfile.mkstemp', Mock(return_value=[3000, 'blabla'])) + @patch('os.path.exists', Mock(return_value=True)) + @patch('os.remove', Mock(side_effect=IOError)) + @patch('os.close', Mock(side_effect=IOError)) + @patch('os.rename', Mock(return_value=None)) + @patch('json.dump', Mock()) + def test_save_cache(self): + self.config.set_dynamic_configuration({'ttl': 30, 'postgresql': {'foo': 'bar'}}) + with patch('os.fdopen', Mock(side_effect=IOError)): + self.config.save_cache() + with patch('os.fdopen', MagicMock()): + self.config.save_cache() diff --git a/tests/test_consul.py b/tests/test_consul.py index 22665174..b3c5ff16 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -51,7 +51,7 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'get', kv_get) @patch.object(consul.Consul.KV, 'delete', Mock()) def setUp(self): - self.c = Consul('postgresql1', {'ttl': 30, 'scope': 'test', 'host': 'localhost:1'}) + self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1'}) self.c._base_path = '/service/good' self.c._load_cluster() @@ -96,6 +96,10 @@ class TestConsul(unittest.TestCase): def test_set_failover_value(self): self.c.set_failover_value('') + @patch.object(consul.Consul.KV, 'put', Mock(return_value=True)) + def test_set_config_value(self): + self.c.set_config_value('') + @patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException)) def test_write_leader_optime(self): self.c.write_leader_optime('') diff --git a/tests/test_ctl.py b/tests/test_ctl.py index d7394db3..d8dbecb7 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -51,7 +51,7 @@ class TestCtl(unittest.TestCase): self.runner = CliRunner() with patch.object(etcd.Client, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) - self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}}, 'foo') + self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379'}}, 'foo') @patch('psycopg2.connect', psycopg2_connect) def test_get_cursor(self): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 508cd480..5631f817 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -78,6 +78,8 @@ def etcd_read(self, key, **kwargs): raise etcd.EtcdKeyNotFound response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [ + {"key": "/service/batman5/config", "value": '{"foo": "bar"}', + "modifiedIndex": 1582, "createdIndex": 1582}, {"key": "/service/batman5/failover", "value": "", "modifiedIndex": 1582, "createdIndex": 1582}, {"key": "/service/batman5/initialize", "value": "postgresql0", @@ -191,7 +193,8 @@ class TestEtcd(unittest.TestCase): def setUp(self): with patch.object(Client, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001']) - self.etcd = Etcd('foo', {'namespace': '/patroni/', 'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'}) + self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, + 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}) def test_base_path(self): self.assertEquals(self.etcd._base_path, '/patroni/test') diff --git a/tests/test_ha.py b/tests/test_ha.py index 5ceb3392..f8f1f90c 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -4,6 +4,7 @@ import datetime import pytz from mock import Mock, MagicMock, patch +from patroni.config import Config from patroni.dcs import Cluster, Failover, Leader, Member, get_dcs from patroni.exceptions import DCSError, PostgresException from patroni.ha import Ha @@ -20,7 +21,7 @@ def false(*args, **kwargs): def get_cluster(initialize, leader, members, failover): - return Cluster(initialize, leader, 10, members, failover) + return Cluster(initialize, None, leader, 10, members, failover) def get_cluster_not_initialized_without_leader(): @@ -48,6 +49,20 @@ def get_cluster_initialized_with_only_leader(failover=None): class MockPatroni(object): def __init__(self, p, d): + self.config = Config(config_env=""" +bootstrap: + users: + replicator: + password: rep-pass + options: + - replication +postgresql: + name: foo + data_dir: data/postgresql0 + pg_rewind: + username: postgres + password: postgres +""") self.postgresql = p self.dcs = d self.api = Mock() @@ -87,13 +102,16 @@ class TestHa(unittest.TestCase): with patch.object(etcd.Client, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432', - 'data_dir': 'data/postgresql0', 'superuser': {}, 'admin': {}, - 'replication': {'username': '', 'password': '', 'network': ''}}) + 'data_dir': 'data/postgresql0', + 'authentication': {'superuser': {'username': 'foo', 'password': 'bar'}, + 'replication': {'username': '', 'password': ''}}, + 'parameters': {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar', + 'hot_standby': 'on', 'max_wal_senders': 5, 'wal_keep_segments': 8}}) self.p.set_state('running') self.p.set_role('replica') self.p.check_replication_lag = true self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False) - self.e = get_dcs('foo', {'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}}) + self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', 'name': 'foo'}}) self.ha = Ha(MockPatroni(self.p, self.e)) self.ha._async_executor.run_async = run_async self.ha.old_cluster = self.e.get_cluster() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index c4492a9f..1cff2611 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -7,6 +7,7 @@ import unittest from mock import Mock, patch from patroni.api import RestApiServer from patroni.async_executor import AsyncExecutor +from patroni.exceptions import DCSError from patroni import Patroni, main as _main from six.moves import BaseHTTPServer from test_etcd import SleepException, etcd_read, etcd_write @@ -25,6 +26,7 @@ from test_postgresql import Postgresql, psycopg2_connect @patch.object(etcd.Client, 'read', etcd_read) class TestPatroni(unittest.TestCase): + @patch.object(etcd.Client, 'read', etcd_read) def setUp(self): RestApiServer._BaseServer__is_shut_down = Mock() RestApiServer._BaseServer__shutdown_request = True @@ -33,6 +35,12 @@ class TestPatroni(unittest.TestCase): mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) self.p = Patroni('postgres0.yml') + @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(side_effect=[None, DCSError('foo'), None])) + def test_load_dynamic_configuration(self): + self.p.config._dynamic_configuration = {} + self.p.load_dynamic_configuration() + self.p.load_dynamic_configuration() + @patch('time.sleep', Mock(side_effect=SleepException)) @patch.object(etcd.Client, 'delete', Mock()) @patch.object(etcd.Client, 'machines') @@ -55,11 +63,15 @@ class TestPatroni(unittest.TestCase): self.assertRaises(SleepException, _main) del os.environ[Patroni.PATRONI_CONFIG_VARIABLE] + @patch('patroni.config.Config.save_cache', Mock()) def test_run(self): self.p.sighup_handler() self.p.ha.dcs.watch = Mock(side_effect=SleepException) self.p.api.start = Mock() + self.p.config._dynamic_configuration = {} self.assertRaises(SleepException, self.p.run) + with patch('patroni.postgresql.Postgresql.data_directory_empty', Mock(return_value=False)): + self.assertRaises(SleepException, self.p.run) def test_schedule_next_run(self): self.p.ha.dcs.watch = Mock(return_value=True) @@ -84,5 +96,5 @@ class TestPatroni(unittest.TestCase): def test_reload_config(self): self.p.reload_config() - with patch('yaml.load', Mock(side_effect=Exception)): - self.p.reload_config() + self.p.get_tags = Mock(side_effect=Exception) + self.p.reload_config() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index c6efa841..84f33e2e 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -5,7 +5,7 @@ import shutil import subprocess import unittest -from mock import Mock, MagicMock, PropertyMock, patch +from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.dcs import Cluster, Leader, Member from patroni.exceptions import PostgresException, PostgresConnectionException from patroni.postgresql import Postgresql @@ -35,7 +35,7 @@ class MockCursor(object): elif sql.startswith('SELECT to_char(pg_postmaster_start_time'): self.results = [('', True, '', '', '', '', False)] elif sql.startswith('SELECT name, setting'): - self.results = [('archive_mode', 'off')] + self.results = [('port', '5433', 'postmaster')] else: self.results = [(None, None, None, None, None, None, None, None, None, None)] @@ -131,6 +131,13 @@ Data page checksum version: 0 """ +def postmaster_opts_string(*args, **kwargs): + return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" \ +"--port=5432" "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" \ +"--archive_command=mkdir -p ../wal_archive && cp %p ../wal_archive/%f" "--wal_log_hints=on" \ +"--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on" "--max_replication_slots=5"\n' + + def psycopg2_connect(*args, **kwargs): return MockConnect() @@ -142,25 +149,24 @@ def fake_listdir(path): @patch('subprocess.call', Mock(return_value=0)) @patch('psycopg2.connect', psycopg2_connect) class TestPostgresql(unittest.TestCase): + _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar', + 'hot_standby': 'on', 'max_wal_senders': 5, 'wal_keep_segments': 8, 'wal_log_hints': 'on'} @patch('subprocess.call', Mock(return_value=0)) @patch('psycopg2.connect', psycopg2_connect) @patch('os.rename', Mock()) + @patch.object(Postgresql, 'get_major_version', Mock(return_value=9.4)) def setUp(self): self.data_dir = 'data/test0' if not os.path.exists(self.data_dir): os.makedirs(self.data_dir) self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, 'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432', - 'pg_hba': ['host replication replicator 127.0.0.1/32 md5', - 'hostssl all all 0.0.0.0/0 md5', - 'host all all 0.0.0.0/0 md5'], - 'superuser': {'username': 'test', 'password': 'test'}, - 'admin': {'username': 'admin', 'password': 'admin'}, - 'pg_rewind': {'username': 'admin', 'password': 'admin'}, - 'replication': {'username': 'replicator', - 'password': 'rep-pass'}, - 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}, + 'authentication': {'superuser': {'username': 'test', 'password': 'test'}, + 'replication': {'username': 'replicator', 'password': 'rep-pass'}}, + 'use_pg_rewind': True, + 'parameters': self._PARAMETERS, + 'recovery_conf': {'foo': 'bar'}, 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_restart': 'true', 'on_role_change': 'true', 'on_reload': 'true' @@ -176,22 +182,11 @@ class TestPostgresql(unittest.TestCase): shutil.rmtree('data') def test_get_initdb_options(self): - self.p.initdb_options = [{'encoding': 'UTF8'}, 'data-checksums'] - self.assertEquals(self.p.get_initdb_options(), ['--encoding=UTF8', '--data-checksums']) - self.p.initdb_options = [{'pgdata': 'bar'}] - self.assertRaises(Exception, self.p.get_initdb_options) - self.p.initdb_options = [{'foo': 'bar', 1: 2}] - self.assertRaises(Exception, self.p.get_initdb_options) - self.p.initdb_options = [1] - self.assertRaises(Exception, self.p.get_initdb_options) - - def test_initialize(self): - self.assertTrue(self.p.initialize()) - - with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f: - lines = f.readlines() - assert 'host replication replicator 127.0.0.1/32 md5\n' in lines - assert 'host all all 0.0.0.0/0 md5\n' in lines + self.assertEquals(self.p.get_initdb_options([{'encoding': 'UTF8'}, 'data-checksums']), + ['--encoding=UTF8', '--data-checksums']) + self.assertRaises(Exception, self.p.get_initdb_options, [{'pgdata': 'bar'}]) + self.assertRaises(Exception, self.p.get_initdb_options, [{'foo': 'bar', 1: 2}]) + self.assertRaises(Exception, self.p.get_initdb_options, [1]) @patch('os.path.exists', Mock(return_value=True)) @patch('os.unlink', Mock()) @@ -258,10 +253,6 @@ class TestPostgresql(unittest.TestCase): @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)) def test_can_rewind(self): - tmp = self.p.pg_rewind - self.p.pg_rewind = None - self.assertFalse(self.p.can_rewind) - self.p.pg_rewind = tmp with mock.patch('subprocess.call', MagicMock(return_value=1)): self.assertFalse(self.p.can_rewind) with mock.patch('subprocess.call', side_effect=OSError): @@ -291,7 +282,7 @@ class TestPostgresql(unittest.TestCase): def test_sync_replication_slots(self): self.p.start() - cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem], None) + cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None) self.p.sync_replication_slots(cluster) self.p.query = Mock(side_effect=psycopg2.OperationalError) self.p.schedule_load_slots = True @@ -355,8 +346,16 @@ class TestPostgresql(unittest.TestCase): def test_bootstrap(self): with patch('subprocess.call', Mock(return_value=1)): - self.assertRaises(PostgresException, self.p.bootstrap) - self.p.bootstrap() + self.assertRaises(PostgresException, self.p.bootstrap, {}) + + self.p.bootstrap({'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}, + 'pg_hba': ['host replication replicator 127.0.0.1/32 md5', + 'hostssl all all 0.0.0.0/0 md5', + 'host all all 0.0.0.0/0 md5']}) + with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f: + lines = f.readlines() + assert 'host replication replicator 127.0.0.1/32 md5\n' in lines + assert 'host all all 0.0.0.0/0 md5\n' in lines @patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0)) def test_clone(self): @@ -387,6 +386,18 @@ class TestPostgresql(unittest.TestCase): with patch('subprocess.check_output', Mock(side_effect=subprocess.CalledProcessError(1, ''))): self.assertEquals(self.p.controldata(), {}) + def test_read_postmaster_opts(self): + m = mock_open(read_data=postmaster_opts_string()) + with patch.object(builtins, 'open', m): + data = self.p.read_postmaster_opts() + self.assertEquals(data['wal_level'], 'hot_standby') + self.assertEquals(int(data['max_replication_slots']), 5) + self.assertEqual(data.get('D'), None) + + m.side_effect = IOError + data = self.p.read_postmaster_opts() + self.assertEqual(data, dict()) + @patch('subprocess.Popen') @patch.object(builtins, 'open', MagicMock(return_value=42)) def test_single_user_mode(self, subprocess_popen_mock): @@ -462,3 +473,11 @@ class TestPostgresql(unittest.TestCase): self.assertTrue(self.p.replica_method_can_work_without_replication_connection('foo')) self.p.config['foo'] = {'command': 'bar'} self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo')) + + def test_reload_config(self): + self.p.reload_config({'listen': '*', 'parameters': self._PARAMETERS}) + self.p.reload_config({'listen': '*:5433', 'parameters': self._PARAMETERS}) + + @patch.object(builtins, 'open', mock_open(read_data='9.4')) + def test_get_major_version(self): + self.assertEquals(self.p.get_major_version(), 9.4) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3b5c9a6a..b2e6e57e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -70,7 +70,7 @@ class MockKazooClient(Mock): raise Exception if path == '/service/test/members/bar' and value == b'retry': return - if path == '/service/test/failover': + if path in ('/service/test/failover', '/service/test/config'): if value == b'Exception': raise Exception elif value == b'ok': @@ -103,7 +103,8 @@ class TestZooKeeper(unittest.TestCase): @patch('requests.get', requests_get) @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient) def setUp(self): - self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'}) + self.zk = ZooKeeper({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, + 'scope': 'test', 'name': 'foo'}) def test_session_listener(self): self.zk.session_listener(KazooState.SUSPENDED) @@ -136,6 +137,11 @@ class TestZooKeeper(unittest.TestCase): self.zk.set_failover_value('ok') self.zk.set_failover_value('Exception') + def test_set_config_value(self): + self.zk.set_config_value('') + self.zk.set_config_value('ok') + self.zk.set_config_value('Exception') + def test_initialize(self): self.assertFalse(self.zk.initialize())