mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-26 07:30:14 +00:00
Merge branch 'master' of github.com:zalando/patroni into feature/lost-dcs-data
This commit is contained in:
Alexander Kukushkin
+12
-11
@@ -1,4 +1,4 @@
|
||||
## This Dockerfile is meant to aid in the building and debugging governor whilst developing on your local machine
|
||||
## 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
|
||||
MAINTAINER Feike Steenbergen <[email protected]>
|
||||
@@ -13,22 +13,23 @@ RUN apt-get update -y
|
||||
RUN apt-get upgrade -y
|
||||
|
||||
ENV PGVERSION 9.4
|
||||
RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython -y
|
||||
RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo -y
|
||||
|
||||
ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
|
||||
|
||||
RUN mkdir -p /governor/helpers
|
||||
ADD governor.py /governor/governor.py
|
||||
ADD helpers /governor/helpers
|
||||
ADD postgres0.yml /governor/
|
||||
RUN mkdir -p /patroni/helpers
|
||||
ADD patroni.py /patroni/patroni.py
|
||||
ADD helpers /patroni/helpers
|
||||
|
||||
ENV ETCDVERSION 2.0.12
|
||||
ENV ETCDVERSION 2.0.13
|
||||
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 && chown postgres:postgres /var/log/etcd.*
|
||||
RUN chown postgres:postgres -R /governor/ /data/
|
||||
RUN /bin/echo -e "etcd --data-dir /tmp/etcd.data > /var/log/etcd.log 2> /var/log/etcd.err &\n/governor/governor.py /governor/postgres0.yml \"\$@\"" >> /entrypoint.sh && chmod +x /entrypoint.sh
|
||||
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
|
||||
|
||||
ENTRYPOINT /entrypoint.sh
|
||||
EXPOSE 4001 5432 2380
|
||||
|
||||
ENTRYPOINT ["/bin/bash", "/entrypoint.sh"]
|
||||
USER postgres
|
||||
|
||||
@@ -55,8 +55,11 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings:
|
||||
* *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd
|
||||
* *session_timeout*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
|
||||
* *reconnects_timeout*: how long we should try to reconnect to ZooKeeper after connection loss. After this timeout we assume that we don't have lock anymore and will restart in read-only mode.
|
||||
* *hosts*: List of ZooKeeper cluster members in format: 'host1:port1,host2:port2,..etc...'
|
||||
|
||||
* *hosts*: list of ZooKeeper cluster members in format: [ 'host1:port1', 'host2:port2', 'etc...']
|
||||
* *exhibitor*: if you are running ZooKeeper cluster under Exhibitor supervisory the following section could be interesting for you
|
||||
* *poll_interval*: how often list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
|
||||
* *port*: Exhibitor port
|
||||
* *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: [ 'host1', 'host2', 'etc...' ]. This list would be updated automatically when Exhibitor (ZooKeeper) cluster topology changes.
|
||||
|
||||
* *postgresql*
|
||||
* *name*: the name of the Postgres host, must be unique for the cluster
|
||||
@@ -64,10 +67,23 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings:
|
||||
* *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications.
|
||||
* *data_dir*: file path to initialize and store Postgres data files
|
||||
* *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader
|
||||
* *pg_hba*: list of lines which should be added to pg_hba.conf
|
||||
* *- host all all 0.0.0.0/0 md5*
|
||||
* *replication*
|
||||
* *username*: replication username, user will be created during initialization
|
||||
* *password*: replication password, user will be created during initialization
|
||||
* *network*: network setting for replication in pg_hba.conf
|
||||
* *callbacks* callback scripts to run on certain actions. Patroni will pass current action, role and cluster name. See scripts/aws.py as an example on how to write them.
|
||||
* *on_start*: a script to run when the cluster starts
|
||||
* *on_stop*: a script to run when the cluster stops
|
||||
* *on_restart*: a script to run when the cluster restarts
|
||||
* *on_reload*: a script to run when configuration reload is triggered
|
||||
* *on_role_change*: a script to run when the cluster is being promoted or demoted
|
||||
* *superuser*
|
||||
* *password*: password for postgres user. It would be set during initialization
|
||||
* *admin*:
|
||||
* *username*: admin username, user will be created during initialization. It would have CREATEDB and CREATEROLE privileges
|
||||
* *password*: admin password, user will be created during initialization.
|
||||
* *recovery_conf*: configuration settings written to recovery.conf when configuring follower
|
||||
* *parameters*: list of configuration settings for Postgres
|
||||
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
# Patroni Dockerfile
|
||||
You can run Patroni in a docker container using this Dockerfile, or by using the Docker image at
|
||||
https://os-registry.stups.zalan.do/acid/patroni-1.0
|
||||
|
||||
This Dockerfile is meant in aiding development of Patroni and quick testing of features. It is not a production-worthy
|
||||
Dockerfile
|
||||
|
||||
# Examples
|
||||
|
||||
## Standalone Patroni
|
||||
|
||||
docker run -d os-registry.stups.zalan.do/acid/patroni:1.0
|
||||
|
||||
## Multiple Patroni's communicating with a standalone etcd inside Docker
|
||||
|
||||
Basically what you would do would be:
|
||||
|
||||
* Run 1 container which provides etcd
|
||||
|
||||
docker run -d <IMAGE> --etcd-only
|
||||
|
||||
* Run n containers running Patroni, passing the `--etcd` option to the `docker run` command
|
||||
|
||||
docker run -d <IMAGE> --etcd=<IP FROM etcd CONTAINER:PORT>
|
||||
|
||||
To automate this you can run the following script:
|
||||
|
||||
dev_patroni_cluster.sh [OPTIONS]
|
||||
|
||||
Options:
|
||||
|
||||
--image IMAGE The Docker image to use for the cluster
|
||||
--members INT The number of members for the cluster
|
||||
--name NAME The name of the new cluster
|
||||
|
||||
Example session:
|
||||
|
||||
$ ./dev_patroni_cluster.sh --image os-registry.stups.zalan.do/acid/patroni:1.0 --members=2 --name=bravo
|
||||
The etcd container is 6be871a11cb373406ca5ea1c6b39e140fdde9fb1d6177212d6ad0c0d1bd9b563, ip=172.17.1.24
|
||||
Started Patroni container 67e611f2eca7c40f9e6e0e24a4a8f2cba7e3e56d22a420e15ab9240a37a9d7a4, ip=172.17.1.25
|
||||
Started Patroni container 47dd12ae635ab83b039f5889e250048b606ed5e48e3650b69e365e7e1d4acbcf, ip=172.17.1.26
|
||||
$ docker ps
|
||||
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
|
||||
47dd12ae635a os-registry.stups.zalan.do/acid/patroni:1.0 "/bin/bash /entrypoi 10 seconds ago Up 8 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_OR64g8bx
|
||||
67e611f2eca7 os-registry.stups.zalan.do/acid/patroni:1.0 "/bin/bash /entrypoi 11 seconds ago Up 10 seconds 2380/tcp, 4001/tcp, 5432/tcp bravo_si9no8iz
|
||||
6be871a11cb3 os-registry.stups.zalan.do/acid/patroni:1.0 "/bin/bash /entrypoi 12 seconds ago Up 10 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_etcd
|
||||
Executable
+90
@@ -0,0 +1,90 @@
|
||||
#!/bin/bash
|
||||
|
||||
DOCKER_IMAGE="os-registry.stups.zalan.do/acid/patroni:1.0"
|
||||
MEMBERS=3
|
||||
|
||||
|
||||
function usage()
|
||||
{
|
||||
cat <<__EOF__
|
||||
Usage: $0
|
||||
|
||||
Options:
|
||||
|
||||
--image IMAGE The Docker image to use for the cluster
|
||||
--members INT The number of members for the cluster
|
||||
--name NAME The name of the new cluster
|
||||
|
||||
Examples:
|
||||
|
||||
$0 --image ${DOCKER_IMAGE}
|
||||
$0
|
||||
$0 --image ${DOCKER_IMAGE} --members=2
|
||||
__EOF__
|
||||
}
|
||||
|
||||
|
||||
optspec=":-:"
|
||||
while getopts "$optspec" optchar; do
|
||||
case "${optchar}" in
|
||||
-)
|
||||
case "${OPTARG}" in
|
||||
help)
|
||||
usage
|
||||
exit 0
|
||||
;;
|
||||
name)
|
||||
PATRONI_SCOPE="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
name=*)
|
||||
PATRONI_SCOPE="${OPTARG#*=}"
|
||||
;;
|
||||
image)
|
||||
DOCKER_IMAGE="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
image=*)
|
||||
DOCKER_IMAGE="${OPTARG#*=}"
|
||||
;;
|
||||
members)
|
||||
MEMBERS="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
members=*)
|
||||
MEMBERS="${OPTARG#*=}"
|
||||
;;
|
||||
*)
|
||||
if [ "$OPTERR" = 1 ] && [ "${optspec:0:1}" != ":" ]; then
|
||||
echo "Unknown option --${OPTARG}" >&2
|
||||
fi
|
||||
;;
|
||||
esac;;
|
||||
*)
|
||||
if [ "$OPTERR" != 1 ] || [ "${optspec:0:1}" = ":" ]; then
|
||||
echo "Non-option argument: '-${OPTARG}'" >&2
|
||||
usage
|
||||
exit 1
|
||||
fi
|
||||
;;
|
||||
esac
|
||||
done
|
||||
|
||||
function random_name()
|
||||
{
|
||||
cat /dev/urandom | env LC_CTYPE=C tr -dc 'a-zA-Z0-9' | head -c 8
|
||||
}
|
||||
|
||||
if [ -z ${PATRONI_SCOPE} ]
|
||||
then
|
||||
PATRONI_SCOPE=$(random_name)
|
||||
fi
|
||||
|
||||
etcd_container=$(docker run -d --name="${PATRONI_SCOPE}_etcd" "${DOCKER_IMAGE}" --etcd-only)
|
||||
etcd_container_ip=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' ${etcd_container})
|
||||
echo "The etcd container is ${etcd_container}, ip=${etcd_container_ip}"
|
||||
|
||||
for i in $(seq 1 "${MEMBERS}")
|
||||
do
|
||||
container_name=$(random_name)
|
||||
patroni_container=$(docker run -d --name="${PATRONI_SCOPE}_${container_name}" "${DOCKER_IMAGE}" --etcd="${etcd_container_ip}:4001" --name="${PATRONI_SCOPE}")
|
||||
patroni_container_ip=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' ${patroni_container})
|
||||
echo "Started Patroni container ${patroni_container}, ip=${patroni_container_ip}"
|
||||
done
|
||||
Executable
+136
@@ -0,0 +1,136 @@
|
||||
#!/bin/bash
|
||||
|
||||
function usage()
|
||||
{
|
||||
cat <<__EOF__
|
||||
Usage: $0
|
||||
|
||||
Options:
|
||||
|
||||
--etcd ETCD Provide an external etcd to connect to
|
||||
--name NAME Give the cluster a specific name
|
||||
--etcd-only Do not run Patroni, run a standalone etcd
|
||||
|
||||
Examples:
|
||||
|
||||
$0 --etcd=127.17.0.84:4001
|
||||
$0 --etcd-only
|
||||
$0
|
||||
$0 --name=true_scotsman
|
||||
__EOF__
|
||||
}
|
||||
|
||||
DOCKER_IP=$(hostname --ip-address)
|
||||
PATRONI_SCOPE=batman
|
||||
|
||||
optspec=":vh-:"
|
||||
while getopts "$optspec" optchar; do
|
||||
case "${optchar}" in
|
||||
-)
|
||||
case "${OPTARG}" in
|
||||
etcd-only)
|
||||
exec etcd --data-dir /tmp/etcd.data \
|
||||
-advertise-client-urls=http://${DOCKER_IP}:4001 \
|
||||
-listen-client-urls=http://0.0.0.0:4001 \
|
||||
-listen-peer-urls=http://0.0.0.0:2380
|
||||
exit 0
|
||||
;;
|
||||
cheat)
|
||||
CHEAT=1
|
||||
;;
|
||||
name)
|
||||
PATRONI_SCOPE="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
name=*)
|
||||
PATRONI_SCOPE=${OPTARG#*=}
|
||||
;;
|
||||
etcd)
|
||||
ETCD_CLUSTER="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
etcd=*)
|
||||
ETCD_CLUSTER=${OPTARG#*=}
|
||||
;;
|
||||
help)
|
||||
usage
|
||||
exit 0
|
||||
;;
|
||||
*)
|
||||
if [ "$OPTERR" = 1 ] && [ "${optspec:0:1}" != ":" ]; then
|
||||
echo "Unknown option --${OPTARG}" >&2
|
||||
fi
|
||||
;;
|
||||
esac;;
|
||||
*)
|
||||
if [ "$OPTERR" != 1 ] || [ "${optspec:0:1}" = ":" ]; then
|
||||
echo "Non-option argument: '-${OPTARG}'" >&2
|
||||
usage
|
||||
exit 1
|
||||
fi
|
||||
;;
|
||||
esac
|
||||
done
|
||||
|
||||
if [ -z ${ETCD_CLUSTER} ]
|
||||
then
|
||||
etcd --data-dir /tmp/etcd.data \
|
||||
-advertise-client-urls=http://${DOCKER_IP}:4001 \
|
||||
-listen-client-urls=http://0.0.0.0:4001 \
|
||||
-listen-peer-urls=http://0.0.0.0:2380 > /var/log/etcd.log 2> /var/log/etcd.err &
|
||||
ETCD_CLUSTER="127.0.0.1:4001"
|
||||
fi
|
||||
|
||||
cat > /patroni/postgres.yml <<__EOF__
|
||||
|
||||
ttl: &ttl 30
|
||||
loop_wait: &loop_wait 10
|
||||
scope: &scope ${PATRONI_SCOPE}
|
||||
restapi:
|
||||
listen: 127.0.0.1:8008
|
||||
connect_address: 127.0.0.1:8008
|
||||
etcd:
|
||||
scope: *scope
|
||||
ttl: *ttl
|
||||
host: ${ETCD_CLUSTER}
|
||||
postgresql:
|
||||
name: postgresql_${DOCKER_IP//./_} ## Replication slots do not allow dots in their name
|
||||
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
|
||||
admin:
|
||||
username: admin
|
||||
password: admin
|
||||
parameters:
|
||||
archive_mode: "on"
|
||||
wal_level: hot_standby
|
||||
archive_command: mkdir -p ../wal_archive && cp %p ../wal_archive/%f
|
||||
max_wal_senders: 20
|
||||
listen_addresses: 0.0.0.0
|
||||
wal_keep_segments: 8
|
||||
archive_timeout: 1800s
|
||||
max_replication_slots: 20
|
||||
hot_standby: "on"
|
||||
__EOF__
|
||||
|
||||
cat /patroni/postgres.yml
|
||||
|
||||
if [ ! -z $CHEAT ]
|
||||
then
|
||||
while :
|
||||
do
|
||||
sleep 60
|
||||
done
|
||||
else
|
||||
exec /patroni/patroni.py /patroni/postgres.yml
|
||||
fi
|
||||
+60
-13
@@ -1,6 +1,7 @@
|
||||
import logging
|
||||
import os
|
||||
import psycopg2
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -15,6 +16,12 @@ else:
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ACTION_ON_START = "on_start"
|
||||
ACTION_ON_STOP = "on_stop"
|
||||
ACTION_ON_RESTART = "on_restart"
|
||||
ACTION_ON_RELOAD = "on_reload"
|
||||
ACTION_ON_ROLE_CHANGE = "on_role_change"
|
||||
|
||||
|
||||
def parseurl(url):
|
||||
r = urlparse(url)
|
||||
@@ -35,14 +42,16 @@ def parseurl(url):
|
||||
|
||||
class Postgresql:
|
||||
|
||||
def __init__(self, config, on_change_callback=None):
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.name = config['name']
|
||||
self.scope = config['scope']
|
||||
self.listen_addresses, self.port = config['listen'].split(':')
|
||||
self.data_dir = config['data_dir']
|
||||
self.replication = config['replication']
|
||||
self.superuser = config['superuser']
|
||||
self.admin = config['admin']
|
||||
self.callback = config.get('callbacks', {})
|
||||
self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
|
||||
self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'),
|
||||
os.path.join(self.data_dir, 'postgresql.conf'))
|
||||
@@ -65,7 +74,6 @@ class Postgresql:
|
||||
self._connection = None
|
||||
self._cursor_holder = None
|
||||
self.members = [] # list of already existing replication slots
|
||||
self.on_change_callback = on_change_callback
|
||||
|
||||
def get_local_address(self):
|
||||
listen_addresses = self.listen_addresses.split(',')
|
||||
@@ -230,9 +238,9 @@ class Postgresql:
|
||||
return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\
|
||||
(diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100)
|
||||
|
||||
def is_leader(self):
|
||||
def is_leader(self, check_only=False):
|
||||
ret = not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
|
||||
if ret and self.is_promoted:
|
||||
if ret and self.is_promoted and not check_only:
|
||||
self.delete_trigger_file()
|
||||
self.is_promoted = False
|
||||
return ret
|
||||
@@ -240,6 +248,26 @@ class Postgresql:
|
||||
def is_running(self):
|
||||
return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0
|
||||
|
||||
def call_nowait(self, cb_name, is_leader=None):
|
||||
""" pick a callback command and call it without waiting for it to finish """
|
||||
if not self.callback or cb_name not in self.callback:
|
||||
return False
|
||||
cmd = self.callback[cb_name]
|
||||
if is_leader is None:
|
||||
try:
|
||||
is_leader = self.is_leader(check_only=True)
|
||||
except psycopg2.OperationalError as e:
|
||||
logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e))
|
||||
return False
|
||||
scope = self.scope
|
||||
try:
|
||||
role = "master" if is_leader else "replica"
|
||||
subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, scope])
|
||||
except Exception as e:
|
||||
logger.warning("callback {0} {1} {2} {3} failed: {4}".format(os.path.abspath(cmd), cb_name, role, scope, e))
|
||||
return False
|
||||
return True
|
||||
|
||||
def start(self):
|
||||
if self.is_running():
|
||||
self.load_replication_slots()
|
||||
@@ -253,18 +281,37 @@ class Postgresql:
|
||||
ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0
|
||||
ret and self.load_replication_slots()
|
||||
self.save_configuration_files()
|
||||
if self.on_change_callback:
|
||||
self.on_change_callback('replica' if os.path.exists(self.recovery_conf) else 'master')
|
||||
if ret and ACTION_ON_START in self.callback:
|
||||
self.call_nowait(ACTION_ON_START)
|
||||
return ret
|
||||
|
||||
def stop(self):
|
||||
return subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) != 0
|
||||
try:
|
||||
is_leader = self.is_leader(check_only=True)
|
||||
except:
|
||||
is_leader = None
|
||||
pass
|
||||
ret = subprocess.call(self._pg_ctl + ['stop', '-m', 'fast'])
|
||||
if ret == 0 and ACTION_ON_STOP in self.callback:
|
||||
self.call_nowait(ACTION_ON_STOP, is_leader=is_leader)
|
||||
return ret == 0
|
||||
|
||||
def reload(self):
|
||||
return subprocess.call(self._pg_ctl + ['reload']) == 0
|
||||
ret = subprocess.call(self._pg_ctl + ['reload'])
|
||||
if ret == 0 and ACTION_ON_RELOAD in self.callback:
|
||||
self.call_nowait(ACTION_ON_RELOAD)
|
||||
return ret == 0
|
||||
|
||||
def restart(self):
|
||||
return subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) == 0
|
||||
try:
|
||||
is_leader = self.is_leader(check_only=True)
|
||||
except:
|
||||
is_leader = None
|
||||
pass
|
||||
ret = subprocess.call(self._pg_ctl + ['restart', '-m', 'fast'])
|
||||
if ret == 0 and ACTION_ON_RESTART in self.callback:
|
||||
self.call_nowait(ACTION_ON_RESTART, is_leader=is_leader)
|
||||
return ret == 0
|
||||
|
||||
def server_options(self):
|
||||
options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port)
|
||||
@@ -354,8 +401,8 @@ primary_conninfo = '{}'
|
||||
if not self.check_recovery_conf(leader):
|
||||
self.write_recovery_conf(leader)
|
||||
self.restart()
|
||||
if self.on_change_callback:
|
||||
self.on_change_callback('replica')
|
||||
if ACTION_ON_ROLE_CHANGE in self.callback:
|
||||
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
||||
|
||||
def save_configuration_files(self):
|
||||
"""
|
||||
@@ -375,8 +422,8 @@ primary_conninfo = '{}'
|
||||
|
||||
def promote(self):
|
||||
self.is_promoted = subprocess.call(self._pg_ctl + ['promote']) == 0
|
||||
if self.on_change_callback:
|
||||
self.on_change_callback('master')
|
||||
if self.is_promoted and ACTION_ON_ROLE_CHANGE in self.callback:
|
||||
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
||||
return self.is_promoted
|
||||
|
||||
def demote(self, leader):
|
||||
|
||||
@@ -6,7 +6,6 @@ import time
|
||||
import yaml
|
||||
|
||||
from helpers.api import RestApiServer
|
||||
from helpers.aws import AWSConnection
|
||||
from helpers.etcd import Etcd
|
||||
from helpers.ha import Ha
|
||||
from helpers.postgresql import Postgresql
|
||||
@@ -18,7 +17,6 @@ class Patroni:
|
||||
|
||||
def __init__(self, config):
|
||||
self.nap_time = config['loop_wait']
|
||||
self.aws = AWSConnection(config)
|
||||
self.postgresql = Postgresql(config['postgresql'])
|
||||
self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
|
||||
host, port = config['restapi']['listen'].split(':')
|
||||
|
||||
+4
-2
@@ -1,15 +1,16 @@
|
||||
ttl: &ttl 30
|
||||
loop_wait: &loop_wait 10
|
||||
scope: &scope batman
|
||||
restapi:
|
||||
listen: 127.0.0.1:8008
|
||||
connect_address: 127.0.0.1:8008
|
||||
etcd:
|
||||
scope: batman
|
||||
scope: *scope
|
||||
ttl: *ttl
|
||||
host: 127.0.0.1:4001
|
||||
#discovery_srv: my-etcd.domain
|
||||
#zookeeper:
|
||||
# scope: batman
|
||||
# scope: *scope
|
||||
# session_timeout: *ttl
|
||||
# reconnect_timeout: *loop_wait
|
||||
# hosts:
|
||||
@@ -24,6 +25,7 @@ etcd:
|
||||
# - host3
|
||||
postgresql:
|
||||
name: postgresql0
|
||||
scope: *scope
|
||||
listen: 127.0.0.1:5432
|
||||
connect_address: 127.0.0.1:5432
|
||||
data_dir: data/postgresql0
|
||||
|
||||
+4
-2
@@ -1,15 +1,16 @@
|
||||
ttl: &ttl 30
|
||||
loop_wait: &loop_wait 10
|
||||
scope: &scope batman
|
||||
restapi:
|
||||
listen: 127.0.0.1:8009
|
||||
connect_address: 127.0.0.1:8009
|
||||
etcd:
|
||||
scope: batman
|
||||
scope: *scope
|
||||
ttl: *ttl
|
||||
host: 127.0.0.1:4001
|
||||
#discovery_srv: my-etcd.domain
|
||||
#zookeeper:
|
||||
# scope: batman
|
||||
# scope: *scope
|
||||
# session_timeout: *ttl
|
||||
# reconnect_timeout: *loop_wait
|
||||
# hosts:
|
||||
@@ -24,6 +25,7 @@ etcd:
|
||||
# - host3
|
||||
postgresql:
|
||||
name: postgresql1
|
||||
scope: *scope
|
||||
listen: 127.0.0.1:5433
|
||||
connect_address: 127.0.0.1:5433
|
||||
data_dir: data/postgresql1
|
||||
|
||||
Regular → Executable
+13
-12
@@ -1,24 +1,18 @@
|
||||
#!/usr/bin/python
|
||||
|
||||
import logging
|
||||
import requests
|
||||
from requests.exceptions import RequestException
|
||||
import yaml
|
||||
import sys
|
||||
import boto.ec2
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AWSConnection:
|
||||
|
||||
def __init__(self, config):
|
||||
def __init__(self, cluster_name):
|
||||
self.available = False
|
||||
self.config = config
|
||||
|
||||
if 'cluster_name' in config:
|
||||
self.cluster_name = config.get('cluster_name')
|
||||
elif 'etcd' in config and isinstance(config['etcd'], dict):
|
||||
self.cluster_name = config['etcd'].get('scope', 'unknown')
|
||||
else:
|
||||
self.cluster_name = 'unknown'
|
||||
self.cluster_name = cluster_name if cluster_name is not None else 'unknown'
|
||||
try:
|
||||
# get the instance id
|
||||
r = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=0.1)
|
||||
@@ -27,7 +21,7 @@ class AWSConnection:
|
||||
return
|
||||
if r.ok:
|
||||
try:
|
||||
content = yaml.load(r.content)
|
||||
content = r.json()
|
||||
self.instance_id = content['instanceId']
|
||||
self.region = content['region']
|
||||
except Exception as e:
|
||||
@@ -69,3 +63,10 @@ class AWSConnection:
|
||||
def on_role_change(self, new_role):
|
||||
ret = self._tag_ec2(new_role)
|
||||
return self._tag_ebs(new_role) and ret
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv) == 4 and sys.argv[1] in ('on_start', 'on_stop', 'on_role_change'):
|
||||
AWSConnection(cluster_name=sys.argv[3]).on_role_change(sys.argv[2])
|
||||
else:
|
||||
sys.exit("Usage: {0} action role name".format(sys.argv[0]))
|
||||
@@ -22,6 +22,7 @@ __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect
|
||||
NAME = 'patroni'
|
||||
MAIN_PACKAGE = 'patroni.py'
|
||||
HELPERS = 'helpers'
|
||||
SCRIPTS = 'scripts'
|
||||
VERSION = '0.1'
|
||||
DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
|
||||
LICENSE = 'The MIT License'
|
||||
@@ -61,7 +62,8 @@ class PyTest(TestCommand):
|
||||
def finalize_options(self):
|
||||
TestCommand.finalize_options(self)
|
||||
if self.cov_xml or self.cov_html:
|
||||
self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov-report', 'term-missing']
|
||||
self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov', SCRIPTS, '--cov-report',
|
||||
'term-missing']
|
||||
if self.cov_xml:
|
||||
self.cov.extend(['--cov-report', 'xml'])
|
||||
if self.cov_html:
|
||||
@@ -80,7 +82,7 @@ class PyTest(TestCommand):
|
||||
params['plugins'] = ['cov']
|
||||
if self.junitxml:
|
||||
params['args'] += self.junitxml
|
||||
params['args'] += ['--doctest-modules', HELPERS, '-s']
|
||||
params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s']
|
||||
errno = pytest.main(**params)
|
||||
sys.exit(errno)
|
||||
|
||||
|
||||
+25
-43
@@ -2,9 +2,8 @@ import unittest
|
||||
import requests
|
||||
import boto.ec2
|
||||
from collections import namedtuple
|
||||
from helpers.aws import AWSConnection
|
||||
from scripts.aws import AWSConnection
|
||||
from requests.exceptions import RequestException
|
||||
import yaml
|
||||
|
||||
|
||||
class MockEc2Connection:
|
||||
@@ -24,6 +23,16 @@ class MockEc2Connection:
|
||||
return True
|
||||
|
||||
|
||||
class MockResponse:
|
||||
|
||||
def __init__(self, content):
|
||||
self.content = content
|
||||
self.ok = True
|
||||
|
||||
def json(self):
|
||||
return self.content
|
||||
|
||||
|
||||
class TestAWSConnection(unittest.TestCase):
|
||||
|
||||
def __init__(self, method_name='runTest'):
|
||||
@@ -32,8 +41,8 @@ class TestAWSConnection(unittest.TestCase):
|
||||
def set_error(self):
|
||||
self.error = True
|
||||
|
||||
def set_ok(self):
|
||||
self.error = False
|
||||
def set_json_error(self):
|
||||
self.json_error = True
|
||||
|
||||
def boto_ec2_connect_to_region(self, region):
|
||||
return MockEc2Connection(self.error)
|
||||
@@ -43,50 +52,18 @@ class TestAWSConnection(unittest.TestCase):
|
||||
raise RequestException("foo")
|
||||
result = namedtuple('Request', 'ok content')
|
||||
result.ok = True
|
||||
if url.split('/')[-1] == 'document':
|
||||
result.content = '{\n "instanceId" : "012345",\n "region" : "eu-west-1"\n}'
|
||||
if url.split('/')[-1] == 'document' and not self.json_error:
|
||||
result = {"instanceId": "012345", "region": "eu-west-1"}
|
||||
else:
|
||||
result.content = 'foo'
|
||||
return result
|
||||
result = 'foo'
|
||||
return MockResponse(result)
|
||||
|
||||
def setUp(self):
|
||||
self.error = False
|
||||
self.json_error = False
|
||||
requests.get = self.requests_get
|
||||
boto.ec2.connect_to_region = self.boto_ec2_connect_to_region
|
||||
self.config_string = """
|
||||
loop_wait: 10
|
||||
restapi:
|
||||
listen: 0.0.0.0:8008
|
||||
connect_address: 127.0.0.1:5432
|
||||
etcd:
|
||||
scope: test
|
||||
ttl: 30
|
||||
host: 127.0.0.1:8080
|
||||
postgresql:
|
||||
name: postgresql_foo
|
||||
listen: 0.0.0.0:5432
|
||||
connect_address: 127.0.0.1:5432
|
||||
data_dir: /home/postgres/pgdata/data
|
||||
replication:
|
||||
username: standby
|
||||
password: standby
|
||||
network: 0.0.0.0/0
|
||||
superuser:
|
||||
password: zalando
|
||||
admin:
|
||||
username: admin
|
||||
password: admin
|
||||
parameters:
|
||||
archive_mode: "on"
|
||||
wal_level: hot_standby
|
||||
max_wal_senders: 5
|
||||
wal_keep_segments: 8
|
||||
archive_timeout: 1800s
|
||||
max_replication_slots: 5
|
||||
hot_standby: "on"
|
||||
ssl: "on"
|
||||
"""
|
||||
self.conn = AWSConnection(yaml.load(self.config_string))
|
||||
self.conn = AWSConnection('test')
|
||||
|
||||
def test_aws_available(self):
|
||||
self.assertTrue(self.conn.aws_available())
|
||||
@@ -98,11 +75,16 @@ postgresql:
|
||||
|
||||
def test_non_aws(self):
|
||||
self.set_error()
|
||||
conn = AWSConnection(yaml.load(self.config_string))
|
||||
conn = AWSConnection('test')
|
||||
self.assertFalse(conn.aws_available())
|
||||
self.assertFalse(conn._tag_ebs('master'))
|
||||
self.assertFalse(conn._tag_ec2('master'))
|
||||
|
||||
def test_aws_bizare_response(self):
|
||||
self.set_json_error()
|
||||
conn = AWSConnection('test')
|
||||
self.assertFalse(conn.aws_available())
|
||||
|
||||
def test_aws_tag_ebs_error(self):
|
||||
self.set_error()
|
||||
self.assertFalse(self.conn._tag_ebs("master"))
|
||||
|
||||
@@ -109,15 +109,19 @@ class TestPostgresql(unittest.TestCase):
|
||||
def set_up(self):
|
||||
subprocess.call = subprocess_call
|
||||
shutil.copy = nop
|
||||
self.p = Postgresql({'name': 'test0', 'data_dir': 'data/test0', 'listen': '127.0.0.1, *:5432',
|
||||
'connect_address': '127.0.0.2:5432',
|
||||
self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': 'data/test0',
|
||||
'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
|
||||
'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'],
|
||||
'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'},
|
||||
'superuser': {'password': ''},
|
||||
'admin': {'username': 'admin', 'password': 'admin'},
|
||||
'replication': {'username': 'replicator',
|
||||
'password': 'rep-pass',
|
||||
'network': '127.0.0.1/32'},
|
||||
'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}},
|
||||
on_change_callback=lambda state: True)
|
||||
'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'},
|
||||
'callbacks': {'on_start': '/usr/bin/true', 'on_stop': '/usr/bin/true',
|
||||
'on_restart': '/usr/bin/true', 'on_role_change': '/bin/true',
|
||||
'on_reload': '/usr/bin/true'
|
||||
}})
|
||||
psycopg2.connect = psycopg2_connect
|
||||
if not os.path.exists(self.p.data_dir):
|
||||
os.makedirs(self.p.data_dir)
|
||||
@@ -128,6 +132,9 @@ class TestPostgresql(unittest.TestCase):
|
||||
def tear_down(self):
|
||||
shutil.rmtree('data')
|
||||
|
||||
def mock_query(self, p):
|
||||
raise psycopg2.OperationalError("not supported")
|
||||
|
||||
def test_data_directory_empty(self):
|
||||
self.assertTrue(self.p.data_directory_empty())
|
||||
|
||||
@@ -135,12 +142,13 @@ class TestPostgresql(unittest.TestCase):
|
||||
self.assertTrue(self.p.initialize())
|
||||
self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf')))
|
||||
|
||||
def test_start(self):
|
||||
def test_start_stop(self):
|
||||
self.assertFalse(self.p.start())
|
||||
self.p.is_running = is_running
|
||||
with open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w'):
|
||||
pass
|
||||
self.assertTrue(self.p.start())
|
||||
self.assertTrue(self.p.stop())
|
||||
|
||||
def test_sync_from_leader(self):
|
||||
self.assertTrue(self.p.sync_from_leader(self.leader))
|
||||
@@ -203,3 +211,11 @@ class TestPostgresql(unittest.TestCase):
|
||||
|
||||
def test_last_operation(self):
|
||||
self.assertEquals(self.p.last_operation(), '0')
|
||||
|
||||
def test_non_existing_callback(self):
|
||||
self.assertFalse(self.p.call_nowait('foobar'))
|
||||
|
||||
def test_is_leader_exception(self):
|
||||
self.p.start()
|
||||
self.p.query = self.mock_query
|
||||
self.assertTrue(self.p.stop())
|
||||
|
||||
Reference in New Issue
Block a user