diff --git a/.gitignore b/.gitignore index 699794c2..fbb294d4 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ build/ coverage.xml junit.xml pgpass +scm-source.json diff --git a/.travis.yml b/.travis.yml index 5fdf1d88..85b1395f 100644 --- a/.travis.yml +++ b/.travis.yml @@ -51,7 +51,7 @@ install: for pv in "2.7" "3.4" "3.5"; do source ~/virtualenv/python${pv}/bin/activate # explicitly install all needed python modules to cache them - for p in '-r requirements.txt' 'behave codacy-coverage coverage coveralls flake8 mock pytest-cov pytest'; do + for p in '-r requirements.txt' 'behave codacy-coverage coverage coveralls flake8 mock>=2.0.0 pytest-cov pytest'; do pip install $p done done diff --git a/Dockerfile b/Dockerfile index 12f6157c..421e775f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,39 +1,36 @@ ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine ## It has all the necessary components to play/debug with a single node appliance, running etcd -FROM ubuntu:14.04 +FROM ubuntu:16.04 MAINTAINER Feike Steenbergen -# We need curl -RUN apt-get update -y && apt-get install curl -y - -# Add PGDG repositories -RUN echo "deb http://apt.postgresql.org/pub/repos/apt/ $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list -RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add - -RUN apt-get update -y -RUN apt-get upgrade -y +RUN echo 'APT::Install-Recommends "0";' > /etc/apt/apt.conf.d/01norecommend +RUN echo 'APT::Install-Suggests "0";' >> /etc/apt/apt.conf.d/01norecommend ENV PGVERSION 9.5 -RUN apt-get install postgresql-${PGVERSION} postgresql-server-dev-${PGVERSION} -y -RUN apt-get install python python-dev python-pip -y -ADD requirements-py2.txt /requirements-py2.txt -RUN pip install -r /requirements-py2.txt +RUN apt-get update -y \ + && apt-get upgrade -y \ + && apt-get install -y curl postgresql-${PGVERSION} python-psycopg2 python-yaml python-requests python-six python-click \ + python-dateutil python-tzlocal python-urllib3 python-dnspython python-pip python-setuptools python-kazoo python \ + && pip install python-etcd==0.4.3 python-consul \ + && apt-get remove -y python-pip python-setuptools \ + && apt-get autoremove -y \ + # Clean up + && apt-get clean -y \ + && rm -rf /var/lib/apt/lists/* ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH -ADD patroni.py /patroni.py -ADD patronictl.py /patronictl.py -ADD patroni/ /patroni +ADD patronictl.py patroni.py docker/entrypoint.sh / +ADD patroni /patroni/ +RUN ln -s /patroni/patroni.py /usr/local/bin/patroni \ + && ln -s /patroni/patronictl.py /usr/local/bin/patronictl -RUN ln -s /patroni.py /usr/local/bin/patroni -RUN ln -s /patronictl.py /usr/local/bin/patronictl - -ENV ETCDVERSION 2.2.5 +ENV ETCDVERSION 2.3.6 RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C /bin --strip=1 --wildcards --no-anchored etcd etcdctl ### Setting up a simple script that will serve as an entrypoint -RUN mkdir /data/ && touch /var/log/etcd.log /var/log/etcd.err /pgpass /patroni/postgres.yml -RUN chown postgres:postgres -R /patroni/ /data/ /pgpass /var/log/etcd.* /patroni/postgres.yml -ADD docker/entrypoint.sh /entrypoint.sh +RUN mkdir /data/ && touch /var/log/etcd.log /var/log/etcd.err /pgpass /patroni/postgres.yml \ + && chown postgres:postgres -R /patroni/ /data/ /pgpass /var/log/etcd.* /patroni/postgres.yml EXPOSE 4001 5432 2380 diff --git a/README.rst b/README.rst index edf0357e..6449c65a 100644 --- a/README.rst +++ b/README.rst @@ -8,6 +8,8 @@ Patroni is a template for you to create your own customized, high-availability s We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely. +**Note to Kubernetes users**: We're currently developing Patroni to be as useful as possible for teams running Kubernetes on top of Google Compute Engine; Patroni can be the HA solution for Postgres in such an environment. Please contact us via our Issues Tracker if this describes your team's current setup, and we'll follow up. + .. contents:: :local: :depth: 1 diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 9bf176bb..c329d2bb 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -70,6 +70,7 @@ while getopts "$optspec" optchar; do esac done +## We start an etcd if [ -z ${ETCD_CLUSTER} ] then etcd --data-dir /tmp/etcd.data \ @@ -79,60 +80,35 @@ then ETCD_CLUSTER="127.0.0.1:4001" fi -mkdir -p ~postgres/.config/patroni -cat > ~postgres/.config/patroni/patronictl.yaml <<__EOF__ -{dcs_api: 'etcd://${ETCD_CLUSTER}', namespace: /service/} -__EOF__ +export PATRONI_SCOPE +export PATRONI_NAME="${HOSTNAME}" +export PATRONI_ETCD_HOST="$ETCD_CLUSTER" +export PATRONI_RESTAPI_CONNECT_ADDRESS="${DOCKER_IP}:8008" +export PATRONI_RESTAPI_LISTEN="0.0.0.0:8008" +export PATRONI_admin_PASSWORD="admin" +export PATRONI_admin_OPTIONS="createdb, createrole" +export PATRONI_POSTGRESQL_CONNECT_ADDRESS="${DOCKER_IP}:5432" +export PATRONI_POSTGRESQL_LISTEN="0.0.0.0:5432" +export PATRONI_POSTGRESQL_DATA_DIR="data/${PATRONI_SCOPE}" +export PATRONI_REPLICATION_USERNAME="replicator" +export PATRONI_REPLICATION_PASSWORD="abcd" +export PATRONI_SUPERUSER_USERNAME="postgres" +export PATRONI_SUPERUSER_PASSWORD="postgres" +export PATRONI_POSTGRESQL_PGPASS="$HOME/.pgpass" cat > /patroni/postgres.yaml <<__EOF__ +bootstrap: + dcs: + postgresql: + use_pg_rewind: true -ttl: &ttl 30 -loop_wait: &loop_wait 10 -scope: &scope '${PATRONI_SCOPE}' -namespace: 'patroni' -restapi: - listen: 0.0.0.0:8008 - connect_address: ${DOCKER_IP}:8008 -etcd: - scope: *scope - ttl: *ttl - host: ${ETCD_CLUSTER} -postgresql: - name: ${HOSTNAME} - scope: *scope - listen: 0.0.0.0:5432 - connect_address: ${DOCKER_IP}:5432 - data_dir: data/postgresql0 - maximum_lag_on_failover: 1048576 # 1 megabyte in bytes pg_hba: - host all all 0.0.0.0/0 md5 - - hostssl all all 0.0.0.0/0 md5 - host replication replicator ${DOCKER_IP}/16 md5 - replication: - username: replicator - password: rep-pass - network: 127.0.0.1/32 - superuser: - password: zalando - restore: patroni/scripts/restore.py - admin: - username: admin - password: admin - parameters: - archive_mode: "on" - wal_level: hot_standby - archive_command: 'true' - max_wal_senders: 20 - listen_addresses: 0.0.0.0 - max_wal_size: 1GB - min_wal_size: 128MB - wal_keep_segments: 64 - archive_timeout: 1800s - max_replication_slots: 20 - hot_standby: "on" __EOF__ -cat /patroni/postgres.yaml +mkdir -p "$HOME/.config/patroni" +ln -s /patroni/postgres.yaml "$HOME/.config/patroni/patronictl.yaml" if [ ! -z $CHEAT ] then diff --git a/patroni/__init__.py b/patroni/__init__.py index 763bd24d..c6d95259 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -1,5 +1,6 @@ import logging import signal +import sys import time from patroni.api import RestApiServer @@ -8,7 +9,7 @@ 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 +from patroni.utils import reap_children, sigchld_handler from patroni.version import __version__ logger = logging.getLogger(__name__) @@ -33,6 +34,7 @@ class Patroni(object): self._reload_config_scheduled = False self._received_sighup = False + self._received_sigterm = False def load_dynamic_configuration(self): while True: @@ -64,6 +66,11 @@ class Patroni(object): def sighup_handler(self, *args): self._received_sighup = True + def sigterm_handler(self, *args): + if not self._received_sigterm: + self._received_sigterm = True + sys.exit() + @property def noloadbalance(self): return self.tags.get('noloadbalance', False) @@ -87,10 +94,9 @@ class Patroni(object): def run(self): self.api.start() - signal.signal(signal.SIGHUP, self.sighup_handler) self.next_run = time.time() - while True: + while not self._received_sigterm: if self._received_sighup: self._received_sighup = False if self.config.reload_local_configuration(): @@ -108,17 +114,22 @@ class Patroni(object): reap_children() self.schedule_next_run() + def setup_signal_handlers(self): + signal.signal(signal.SIGHUP, self.sighup_handler) + signal.signal(signal.SIGTERM, self.sigterm_handler) + signal.signal(signal.SIGCHLD, sigchld_handler) + def main(): logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) logging.getLogger('requests').setLevel(logging.WARNING) - setup_signal_handlers() patroni = Patroni() + patroni.setup_signal_handlers() try: patroni.run() except KeyboardInterrupt: - set_ignore_sigterm() + pass finally: patroni.api.shutdown() patroni.postgresql.stop(checkpoint=False) diff --git a/patroni/api.py b/patroni/api.py index ba37d5a0..02d8b168 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -4,7 +4,7 @@ import json import logging import psycopg2 import time -import dateutil +import dateutil.parser import datetime import pytz import re diff --git a/patroni/config.py b/patroni/config.py index e749612e..ce81de17 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -66,8 +66,7 @@ class Config(object): format(self.PATRONI_CONFIG_VARIABLE)) exit(1) - self.__effective_configuration = self._build_effective_configuration(self._dynamic_configuration, - self._local_configuration) + self.__effective_configuration = self._build_effective_configuration({}, self._local_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() @@ -202,7 +201,7 @@ class Config(object): value = _popenv(name + '_' + param) if value: ret[param] = value - return len(ret) == 2 and ret or None + return ret restapi_auth = _get_auth('restapi') if restapi_auth: @@ -306,3 +305,6 @@ class Config(object): def __getitem__(self, key): return self.__effective_configuration[key] + + def copy(self): + return deepcopy(self.__effective_configuration) diff --git a/patroni/ctl.py b/patroni/ctl.py index 499f6563..a0f0c565 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -2,23 +2,26 @@ Patroni Control ''' +import base64 import click import datetime -import dateutil +import dateutil.parser import json import logging import os import psycopg2 import random import requests +import sys import time import tzlocal import yaml from click import ClickException +from patroni.config import Config from patroni.dcs import get_dcs as _get_dcs from patroni.exceptions import PatroniException -from patroni.postgresql import parseurl +from patroni.postgresql import get_conn_kwargs from prettytable import PrettyTable from six.moves.urllib_parse import urlparse @@ -56,14 +59,23 @@ def parse_dcs(dcs): def load_config(path, dcs): logging.debug('Loading configuration from file %s', path) - config = dict() + config = {} + old_argv = list(sys.argv) try: - with open(path, 'rb') as fd: - config = yaml.safe_load(fd) - except (IOError, yaml.YAMLError): - logging.exception('Could not load configuration file') + sys.argv[1] = path + if Config.PATRONI_CONFIG_VARIABLE not in os.environ: + for p in ('PATRONI_RESTAPI_LISTEN', 'PATRONI_POSTGRESQL_DATA_DIR'): + if p not in os.environ: + os.environ[p] = '.' + config = Config().copy() + finally: + sys.argv = old_argv - config.update(parse_dcs(dcs) or parse_dcs(config.get('dcs_api')) or {}) + dcs = parse_dcs(dcs) or parse_dcs(config.get('dcs_api')) or {} + if dcs: + for d in DCS_DEFAULTS: + config.pop(d, None) + config.update(dcs) return config @@ -102,11 +114,19 @@ def get_dcs(config, scope): raise PatroniCtlException(str(e)) +def auth_header(config): + if config.get('restapi', {}).get('auth', ''): + return {'Authorization': 'Basic ' + base64.b64encode(config['restapi']['auth'].encode('utf-8')).decode('utf-8')} + + def post_patroni(member, endpoint, content, headers=None): + headers = headers or {} url = urlparse(member.api_url) logging.debug(url) + if 'Content-Type' not in headers: + headers['Content-Type'] = 'application/json' return requests.post('{0}://{1}/{2}'.format(url.scheme, url.netloc, endpoint), - headers=headers or {'Content-Type': 'application/json'}, + headers=headers, data=json.dumps(content), timeout=60) @@ -122,10 +142,7 @@ def print_output(columns, rows=None, alignment=None, fmt='pretty', header=True, return if fmt == 'json': - elements = list() - for r in rows: - elements.append(dict(zip(columns, r))) - + elements = [dict(zip(columns, r)) for r in rows] click.echo(json.dumps(elements)) if fmt == 'tsv': @@ -165,14 +182,13 @@ def watching(w, watch, max_count=None, clear=True): yield 0 -def build_connect_parameters(conn_url, connect_parameters=None): - params = (connect_parameters or {}).copy() - parsed = parseurl(conn_url) - params['host'] = parsed['host'] - params['port'] = parsed['port'] - params['fallback_application_name'] = 'Patroni ctl' - params['connect_timeout'] = '5' - +def build_connect_parameters(conn_url, connect_parameters): + params = get_conn_kwargs(conn_url, connect_parameters) + params.update({'fallback_application_name': 'Patroni ctl', 'connect_timeout': '5'}) + if 'database' in connect_parameters: + params['database'] = connect_parameters['database'] + else: + params.pop('database') return params @@ -195,7 +211,7 @@ def get_any_member(cluster, role='master', member=None): return m -def get_cursor(cluster, role='master', member=None, connect_parameters=None): +def get_cursor(cluster, connect_parameters, role='master', member=None): member = get_any_member(cluster, role=role, member=member) if member is None: return None @@ -237,7 +253,7 @@ def dsn(cluster_name, config_file, dcs, role, member): if m is None: raise PatroniCtlException('Can not find a suitable member') - params = build_connect_parameters(m.conn_url) + params = get_conn_kwargs(m.conn_url) click.echo('host={host} port={port}'.format(**params)) @@ -287,7 +303,7 @@ def query( connect_parameters = dict() if username: - connect_parameters['user'] = username + connect_parameters['username'] = username if password: connect_parameters['password'] = click.prompt('Password', hide_input=True, type=str) if dbname: @@ -308,10 +324,10 @@ def query( cluster = dcs.get_cluster() -def query_member(cluster, cursor, member, role, command, connect_parameters=None): +def query_member(cluster, cursor, member, role, command, connect_parameters): try: if cursor is None: - cursor = get_cursor(cluster, role=role, member=member, connect_parameters=connect_parameters) + cursor = get_cursor(cluster, connect_parameters, role=role, member=member) if cursor is None: if role is None: @@ -382,17 +398,15 @@ def wait_for_leader(dcs, timeout=30): raise PatroniCtlException('Timeout occured') -def empty_post_to_members(cluster, member_names, force, endpoint): - candidates = dict() - for m in cluster.members: - candidates[m.name] = m +def empty_post_to_members(cluster, member_names, force, endpoint, headers=None): + candidates = {m.name: m for m in cluster.members} if not member_names: member_names = [click.prompt('Which member do you want to {0} [{1}]?'.format(endpoint, ', '.join(candidates.keys())), type=str, default='')] for mn in member_names: - if mn not in candidates.keys(): + if mn not in candidates: raise PatroniCtlException('{0} is not a member of cluster'.format(mn)) if not force: @@ -401,7 +415,7 @@ def empty_post_to_members(cluster, member_names, force, endpoint): raise PatroniCtlException('Aborted {0}'.format(endpoint)) for mn in member_names: - r = post_patroni(candidates[mn], endpoint, '') + r = post_patroni(candidates[mn], endpoint, '', headers) if r.status_code != 200: click.echo('{0} failed for member {1}, status code={2}, ({3})'.format(endpoint, mn, r.status_code, r.text)) else: @@ -426,7 +440,7 @@ def ctl_load_config(cluster_name, config_file, dcs): @option_force @option_dcs def restart(cluster_name, member_names, config_file, dcs, force, role, p_any): - _, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) role_names = [m.name for m in get_all_members(cluster, role)] @@ -440,7 +454,7 @@ def restart(cluster_name, member_names, config_file, dcs, force, role, p_any): member_names = member_names[:1] output_members(cluster, cluster_name) - empty_post_to_members(cluster, member_names, force, 'restart') + empty_post_to_members(cluster, member_names, force, 'restart', auth_header(config)) @ctl.command('reinit', help='Reinitialize cluster member') @@ -450,8 +464,8 @@ def restart(cluster_name, member_names, config_file, dcs, force, role, p_any): @option_force @option_dcs def reinit(cluster_name, member_names, config_file, dcs, force): - _, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - empty_post_to_members(cluster, member_names, force, 'reinitialize') + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + empty_post_to_members(cluster, member_names, force, 'reinitialize', auth_header(config)) @ctl.command('failover', help='Failover to a replica') @@ -471,7 +485,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled If so, we trigger a failover and keep the client up to date. """ - _, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) if cluster.leader is None: raise PatroniCtlException('This cluster has no master') @@ -533,7 +547,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled r = None try: - r = post_patroni(cluster.leader.member, 'failover', failover_value) + r = post_patroni(cluster.leader.member, 'failover', failover_value, auth_header(config)) if r.status_code in (200, 202): logging.debug(r) cluster = dcs.get_cluster() @@ -570,7 +584,7 @@ def output_members(cluster, name, fmt='pretty'): if m.name == leader_name: leader = '*' - host = build_connect_parameters(m.conn_url)['host'] + host = get_conn_kwargs(m.conn_url)['host'] xlog_location = m.data.get('xlog_location') or 0 lag = '' diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index c068ef8e..c1c31174 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -31,22 +31,22 @@ def parse_connection_string(value): 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 - module = importlib.import_module(__package__ + '.' + name[:-3]) - for name in dir(module): # iterate through module content - if not name.startswith('__'): # skip internal stuff - value = getattr(module, name) - name = name.lower() - # try to find implementation of AbstractDCS interface - if inspect.isclass(value) and issubclass(value, AbstractDCS): - 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', 'name', - 'scope', 'ttl', 'retry_timeout') if p in config}) - return value(config[name]) + available_implementations = set() + for module in os.listdir(os.path.dirname(__file__)): + if module.endswith('.py') and not module.startswith('__'): # find module + module_name = module[:-3].lower() + module = importlib.import_module(__package__ + '.' + module[:-3]) + for name in filter(lambda name: not name.startswith('__'), dir(module)): # iterate through module content + value = getattr(module, name) + name = name.lower() + # try to find implementation of AbstractDCS interface, class name must match with module_name + if inspect.isclass(value) and issubclass(value, AbstractDCS) and name == module_name: + available_implementations.add(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', '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)) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index aa3adf0c..4016c6d7 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -22,7 +22,7 @@ ACTION_ON_RELOAD = "on_reload" ACTION_ON_ROLE_CHANGE = "on_role_change" -def parseurl(url): +def get_conn_kwargs(url, auth=None): r = urlparse(url) ret = { 'host': r.hostname, @@ -32,10 +32,11 @@ def parseurl(url): 'connect_timeout': 3, 'options': '-c statement_timeout=2000', } - if r.username: - ret['user'] = r.username - if r.password: - ret['password'] = r.password + if auth and isinstance(auth, dict): + if 'username' in auth: + ret['user'] = auth['username'] + if 'password' in auth: + ret['password'] = auth['password'] return ret @@ -148,8 +149,8 @@ class Postgresql(object): def resolve_connection_addresses(self): self._local_address = self.get_local_address() - self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format( - connect_address=self._connect_address or self._local_address, database=self._database, **self._replication) + self.connection_string = 'postgres://{connect_address}/{database}'.format( + connect_address=self._connect_address or self._local_address, database=self._database) def reload_config(self, config): server_parameters = self.get_server_parameters(config) @@ -248,7 +249,7 @@ class Postgresql(object): 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', '127.0.0.1', 'localhost'): # we are listening on '*' or localhost + if la.strip().lower() 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._server_parameters['port'] @@ -263,12 +264,7 @@ class Postgresql(object): @property def _connect_kwargs(self): - r = parseurl('postgres://{0}/{1}'.format(self._local_address, self._database)) - if 'username' in self._superuser: - r['user'] = self._superuser['username'] - if 'password' in self._superuser: - r['password'] = self._superuser['password'] - return r + return get_conn_kwargs('postgres://{0}/{1}'.format(self._local_address, self._database), self._superuser) def connection(self): if not self._connection or self._connection.closed != 0: @@ -392,7 +388,7 @@ class Postgresql(object): replica_methods = self.config.get('create_replica_method') or ['basebackup'] if clone_member: - r = parseurl(clone_member.conn_url) + r = get_conn_kwargs(clone_member.conn_url, 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) @@ -606,17 +602,17 @@ class Postgresql(object): with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f: f.write('\n{}\n'.format('\n'.join(config))) - def primary_conninfo(self, leader_url): - r = parseurl(leader_url) + def primary_conninfo(self, node_to_follow_url): + r = get_conn_kwargs(node_to_follow_url, 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, leader): + def check_recovery_conf(self, node_to_follow): if not os.path.isfile(self._recovery_conf): return False - pattern = leader and leader.conn_url and self.primary_conninfo(leader.conn_url) + pattern = node_to_follow and node_to_follow.conn_url and self.primary_conninfo(node_to_follow.conn_url) with open(self._recovery_conf, 'r') as f: for line in f: @@ -624,11 +620,11 @@ class Postgresql(object): return pattern and (pattern in line) return not pattern - def write_recovery_conf(self, leader): + def write_recovery_conf(self, node_to_follow): with open(self._recovery_conf, 'w') as f: f.write("standby_mode = 'on'\nrecovery_target_timeline = 'latest'\n") - if leader and leader.conn_url: - f.write("primary_conninfo = '{0}'\n".format(self.primary_conninfo(leader.conn_url))) + if node_to_follow and node_to_follow.conn_url: + f.write("primary_conninfo = '{0}'\n".format(self.primary_conninfo(node_to_follow.conn_url))) if self.use_slots: f.write("primary_slot_name = '{0}'\n".format(self.name)) for name, value in self.config.get('recovery_conf', {}).items(): @@ -637,10 +633,7 @@ class Postgresql(object): def rewind(self, leader): # prepare pg_rewind connection - r = parseurl(leader.conn_url) - r.update(self._superuser) - r['user'] = r.pop('username') - r['database'] = self._database + r = get_conn_kwargs(leader.conn_url, self._superuser) env = self.write_pgpass(r) pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r) # first run a checkpoint on a promoted master in order @@ -656,7 +649,8 @@ class Postgresql(object): def controldata(self): """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ result = {} - if self.state != 'creating replica': # Don't try to call pg_controldata during backup restore + # Don't try to call pg_controldata during backup restore + if self._version_file_exists() and self.state != 'creating replica': try: data = subprocess.check_output(['pg_controldata', self._data_dir]) if data: diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index f3691907..4383e986 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -33,7 +33,7 @@ import sys import argparse -if sys.hexversion >= 0x03000000: +if sys.hexversion >= 0x0300000: long = int logger = logging.getLogger(__name__) diff --git a/patroni/utils.py b/patroni/utils.py index 1b604406..9c7a9e9d 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -1,39 +1,17 @@ -import datetime import os import random -import signal -import six import sys import time -import pytz -import dateutil.parser from patroni.exceptions import PatroniException -__ignore_sigterm = False +if sys.hexversion >= 0x0300000: + long = int + __interrupted_sleep = False __reap_children = False -def calculate_ttl(expiration): - """ - >>> calculate_ttl(None) - >>> calculate_ttl('2015-06-10 12:56:30.552539016Z') < 0 - True - >>> calculate_ttl('2015-06-10T12:56:30.552539016Z') < 0 - True - >>> calculate_ttl('fail-06-10T12:56:30.552539016Z') - """ - if not expiration: - return None - try: - expiration = dateutil.parser.parse(expiration) - except (ValueError, TypeError): - return None - now = datetime.datetime.now(pytz.utc) - return int((expiration - now).total_seconds()) - - def deep_compare(obj1, obj2): """ >>> deep_compare({'1': None}, {}) @@ -137,7 +115,7 @@ def strtol(value, strict=True): while i < l: try: # try to find maximally long number i += 1 # by giving to `int` longer and longer strings - ret = int(value[:i], base) if six.PY3 else long(value[:i], base) + ret = long(value[:i], base) except ValueError: # until we will not get an exception or end of the string i -= 1 break @@ -208,17 +186,6 @@ def compare_values(vartype, unit, old_value, new_value): return old_value is not None and new_value is not None and old_value == new_value -def set_ignore_sigterm(value=True): - global __ignore_sigterm - __ignore_sigterm = value - - -def sigterm_handler(signo, stack_frame): - if not __ignore_sigterm: - set_ignore_sigterm() - sys.exit() - - def sigchld_handler(signo, stack_frame): global __interrupted_sleep, __reap_children __reap_children = __interrupted_sleep = True @@ -237,11 +204,6 @@ def sleep(interval): __interrupted_sleep = False -def setup_signal_handlers(): - signal.signal(signal.SIGTERM, sigterm_handler) - signal.signal(signal.SIGCHLD, sigchld_handler) - - def reap_children(): global __reap_children if __reap_children: diff --git a/setup.py b/setup.py index 6af7b953..8a2b2bfb 100644 --- a/setup.py +++ b/setup.py @@ -33,7 +33,7 @@ LICENSE = 'The MIT License' URL = 'https://github.com/zalando/patroni' AUTHOR = 'Alexander Kukushkin, Oleksii Kliukin, Feike Steenbergen' AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, oleksii.kliukin@zalando.de, feike.steenbergen@zalando.de' -KEYWORDS = 'etcd governor patroni postgresql postgres ha zookeeper streaming replication' +KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd zookeeper exhibitor consul streaming replication' COVERAGE_XML = True COVERAGE_HTML = False @@ -147,7 +147,7 @@ def setup_package(): install_requires=install_reqs, setup_requires=['flake8'], cmdclass=cmdclass, - tests_require=['mock', 'pytest-cov', 'pytest'], + tests_require=['mock>=2.0.0', 'pytest-cov', 'pytest'], command_options=command_options, entry_points={'console_scripts': CONSOLE_SCRIPTS}, ) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index eb63e153..42929a43 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -1,7 +1,7 @@ import etcd import os -import pytest -import requests.exceptions +import requests +import sys import unittest from click.testing import CliRunner @@ -19,31 +19,16 @@ CONFIG_FILE_PATH = './test-ctl.yaml' def test_rw_config(): runner = CliRunner() - config = {'a': 'b'} with runner.isolated_filesystem(): - store_config(config, CONFIG_FILE_PATH + '/dummy') + store_config({'etcd': {'host': 'localhost:2379'}}, CONFIG_FILE_PATH + '/dummy') + sys.argv = ['patronictl.py', ''] + load_config(CONFIG_FILE_PATH + '/dummy', None) + load_config(CONFIG_FILE_PATH + '/dummy', '0.0.0.0') os.remove(CONFIG_FILE_PATH + '/dummy') os.rmdir(CONFIG_FILE_PATH) - with pytest.raises(Exception): - result = load_config(CONFIG_FILE_PATH, None) - assert 'Could not load configuration file' in result.output - os.mkdir(CONFIG_FILE_PATH) - with pytest.raises(Exception): - store_config(config, CONFIG_FILE_PATH) - - os.rmdir(CONFIG_FILE_PATH) - - store_config(config, CONFIG_FILE_PATH) - load_config(CONFIG_FILE_PATH, None) - load_config(CONFIG_FILE_PATH, '0.0.0.0') - - store_config({'dcs_api': None}, CONFIG_FILE_PATH) - load_config(CONFIG_FILE_PATH, None) - - -@patch('patroni.ctl.load_config', Mock(return_value={'etcd': {'host': 'localhost:4001'}})) +@patch('patroni.ctl.load_config', Mock(return_value={'restapi': {'auth': 'u:p'}, 'etcd': {'host': 'localhost:4001'}})) class TestCtl(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) @@ -55,14 +40,14 @@ class TestCtl(unittest.TestCase): @patch('psycopg2.connect', psycopg2_connect) def test_get_cursor(self): - self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), role='master')) + self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), {}, role='master')) - self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), role='master')) + self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {}, role='master')) # MockCursor returns pg_is_in_recovery as false - self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), role='replica')) + self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), {}, role='replica')) - self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), role='any')) + self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {'database': 'foo'}, role='any')) def test_parse_dcs(self): assert parse_dcs(None) is None @@ -183,24 +168,24 @@ class TestCtl(unittest.TestCase): def test_query_member(self): with patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor())): - rows = query_member(None, None, None, 'master', 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, 'master', 'SELECT pg_is_in_recovery()', {}) self.assertTrue('False' in str(rows)) - rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()', {}) self.assertEquals(rows, (None, None)) with patch('test_postgresql.MockCursor.execute', Mock(side_effect=OperationalError('bla'))): - rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()', {}) with patch('patroni.ctl.get_cursor', Mock(return_value=None)): - rows = query_member(None, None, None, None, 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, None, 'SELECT pg_is_in_recovery()', {}) self.assertTrue('No connection to' in str(rows)) - rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()', {}) self.assertTrue('No connection to' in str(rows)) with patch('patroni.ctl.get_cursor', Mock(side_effect=OperationalError('bla'))): - rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()', {}) @patch('patroni.ctl.get_dcs') def test_dsn(self, mock_get_dcs): diff --git a/tests/test_patroni.py b/tests/test_patroni.py index f9246cac..af8b36b0 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -68,6 +68,9 @@ class TestPatroni(unittest.TestCase): with patch('patroni.postgresql.Postgresql.data_directory_empty', Mock(return_value=False)): self.assertRaises(SleepException, self.p.run) + def test_sigterm_handler(self): + self.assertRaises(SystemExit, self.p.sigterm_handler) + def test_schedule_next_run(self): self.p.ha.dcs.watch = Mock(return_value=True) self.p.schedule_next_run() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 7fa7279a..42dcab0a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -336,7 +336,9 @@ class TestPostgresql(unittest.TestCase): @patch('os.path.isfile', Mock(return_value=True)) @patch('os.kill', Mock(side_effect=Exception)) - @patch.object(builtins, 'open', mock_open(read_data='-999999999999999')) + @patch('os.getpid', Mock(return_value=2)) + @patch('os.getppid', Mock(return_value=2)) + @patch.object(builtins, 'open', mock_open(read_data='-1')) @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True)) def test_is_running(self): self.assertFalse(self.p.is_running()) @@ -396,6 +398,7 @@ class TestPostgresql(unittest.TestCase): self.p.remove_data_directory() self.p.remove_data_directory() + @patch('patroni.postgresql.Postgresql._version_file_exists', Mock(return_value=True)) def test_controldata(self): with patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)): data = self.p.controldata() @@ -466,6 +469,7 @@ class TestPostgresql(unittest.TestCase): mock_unlink.assert_not_called() mock_remove.assert_not_called() + @patch('patroni.postgresql.Postgresql._version_file_exists', Mock(return_value=True)) @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string)) def test_sysid(self): self.assertEqual(self.p.sysid, "6200971513092291716") diff --git a/tests/test_utils.py b/tests/test_utils.py index af4cd861..b32d6b35 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,8 +2,7 @@ import unittest from mock import Mock, patch from patroni.exceptions import PatroniException -from patroni.utils import reap_children, Retry, RetryFailedError, set_ignore_sigterm,\ - sigchld_handler, sigterm_handler, sleep +from patroni.utils import reap_children, Retry, RetryFailedError, sigchld_handler, sleep def time_sleep(_): @@ -12,10 +11,6 @@ def time_sleep(_): class TestUtils(unittest.TestCase): - def test_sigterm_handler(self): - set_ignore_sigterm(False) - self.assertRaises(SystemExit, sigterm_handler, None, None) - @patch('time.sleep', Mock()) def test_reap_children(self): self.assertIsNone(reap_children())