mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Merge branch 'master' into feature/scheduled_restarts
This commit is contained in:
Oleksii Kliukin
@@ -9,3 +9,4 @@ build/
|
||||
coverage.xml
|
||||
junit.xml
|
||||
pgpass
|
||||
scm-source.json
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+20
-23
@@ -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 <[email protected]>
|
||||
|
||||
# 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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+22
-46
@@ -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
|
||||
|
||||
+16
-5
@@ -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)
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ import json
|
||||
import logging
|
||||
import psycopg2
|
||||
import time
|
||||
import dateutil
|
||||
import dateutil.parser
|
||||
import datetime
|
||||
import pytz
|
||||
import re
|
||||
|
||||
+5
-3
@@ -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)
|
||||
|
||||
+53
-39
@@ -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 = ''
|
||||
|
||||
+16
-16
@@ -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))
|
||||
|
||||
|
||||
+21
-27
@@ -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:
|
||||
|
||||
@@ -33,7 +33,7 @@ import sys
|
||||
import argparse
|
||||
|
||||
|
||||
if sys.hexversion >= 0x03000000:
|
||||
if sys.hexversion >= 0x0300000:
|
||||
long = int
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
+4
-42
@@ -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:
|
||||
|
||||
@@ -33,7 +33,7 @@ LICENSE = 'The MIT License'
|
||||
URL = 'https://github.com/zalando/patroni'
|
||||
AUTHOR = 'Alexander Kukushkin, Oleksii Kliukin, Feike Steenbergen'
|
||||
AUTHOR_EMAIL = '[email protected], [email protected], [email protected]'
|
||||
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},
|
||||
)
|
||||
|
||||
+17
-32
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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")
|
||||
|
||||
+1
-6
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user