gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/external_backup

Browse Source 
Conflicts:
	requirements-py2.txt
This commit is contained in:
Feike Steenbergen
2015-08-11 11:27:53 +02:00
parent 164b5760a4 e51ea4c1ca
commit b0707c8ddd
13 changed files with 394 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+12 -11
View File
@@ -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
docker/README.md
+46
View File
@@ -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
docker/dev_patroni_cluster.sh
Executable
+90
View File
@@ -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
docker/entrypoint.sh
Executable
+136
View File
@@ -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
helpers/api.py
+2 -8
View File
@@ -1,17 +1,11 @@
import json
import logging
import psycopg2
import sys
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from six.moves.socketserver import ThreadingMixIn
from threading import Thread
if sys.hexversion >= 0x03000000:
from http.server import BaseHTTPRequestHandler, HTTPServer
from socketserver import ThreadingMixIn
else:
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from SocketServer import ThreadingMixIn
logger = logging.getLogger(__name__)
helpers/dcs.py
+69 -13
View File
@@ -1,16 +1,21 @@
import abc
import sys
from collections import namedtuple
from helpers.utils import calculate_ttl, sleep
if sys.hexversion >= 0x03000000:
from urllib.parse import urlparse, urlunparse, parse_qsl
else:
from urlparse import urlparse, urlunparse, parse_qsl
from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
def parse_connection_string(value):
"""Original Governor stores connection strings for each cluster members if a following format:
postgres://{username}:{password}@{connect_address}/postgres
Since each of our patroni instances provides own REST API endpoint it's good to store this information
in DCS among with postgresql connection string. In order to not introduce new keys and be compatible with
original Governor we decided to extend original connection string in a following way:
postgres://{username}:{password}@{connect_address}/postgres?application_name={api_url}
This way original Governor could use such connection string as it is, because of feature of `libpq` library.
This method is able to split connection string stored in DCS into two parts, `conn_url` and `api_url`"""
scheme, netloc, path, params, query, fragment = urlparse(value)
conn_url = urlunparse((scheme, netloc, path, params, '', fragment))
api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0]
@@ -18,6 +23,7 @@ def parse_connection_string(value):
class DCSError(Exception):
"""Parent class for all kind of exceptions related to selected distributed configuration store"""
def __init__(self, value):
self.value = value
@@ -31,12 +37,27 @@ class DCSError(Exception):
class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')):
"""Immutable object (namedtuple) which represents single member of PostgreSQL cluster.
Consists of the following fields:
:param index: modification index of a given member key in DCS
:param name: name of PostgreSQL cluster member
:param conn_url: connection string containing host, user and password which could be used to access this member.
:param api_url: REST API url of patroni instance
:param expiration: expiration time of given member key
:param ttl: ttl of given member key in seconds"""
def real_ttl(self):
return calculate_ttl(self.expiration) or -1
class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')):
"""Immutable object (namedtuple) which represents PostgreSQL cluster.
Consists of the following fields:
:param initialize: boolean, shows whether this cluster has initialization key stored in DC or not.
:param leader: `Member` object which represents current leader of the cluster
:param last_leader_operation: int or long object containing position of last known leader operation.
This value is stored in `/optime/leader` key
:param members: list of Member object, all PostgreSQL cluster members including leader"""
def is_unlocked(self):
return not (self.leader and self.leader.name)
@@ -47,6 +68,11 @@ class AbstractDCS:
__metaclass__ = abc.ABCMeta
def __init__(self, name, config):
"""
:param name: name of current instance (the same value as `~Postgresql.name`)
:param config: dict, reference to config section of selected DCS.
i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
"""
self._name = name
self._base_path = '/service/' + config['scope']
@@ -55,15 +81,30 @@ class AbstractDCS:
@abc.abstractmethod
def get_cluster(self):
"""get_cluster"""
""":returns: `Cluster` object which represent current state and topology of the cluster
raise `~DCSError` in case of communication or other problems with DCS. If current instance was
running as a master and exception raised instance would be demoted."""
@abc.abstractmethod
def update_leader(self, state_handler):
"""update_leader"""
"""Update leader key (or session) ttl and `/optime/leader` key in DCS.
:param state_handler: reference to `Postgresql` object
:returns: `!True` if leader lock (or session) has been updated successfully.
If not, `!False` must be returned and current instance would be demoted.
If you failed to update `/optime/leader` this error is not critical and you can return `!True`
You have to use CAS operation on order to update leader key, for example for etcd `prevValue`
parameter have to be used."""
@abc.abstractmethod
def attempt_to_acquire_leader(self):
"""attempt_to_acquire_leader"""
"""Attempt to acquire leader lock
This method should create `/leader` key with value=`~self._name`
:returns: `!True` if key has been created successfully.
Key has to be created atomically. In case if key already exists it should not be
overwritten and `!False` must be returned"""
def current_leader(self):
try:
@@ -74,19 +115,34 @@ class AbstractDCS:
@abc.abstractmethod
def touch_member(self, connection_string, ttl=None):
"""touch_member"""
"""Update member key in DCS.
This method should create or update key with the name = '/members/' + `~self._name`
and value = connection_string in a given DCS.
:param connection_string: how this instance can be accessed by other instances
:param ttl: ttl for member key, optional parameter. If it is None `~self.member_ttl will be used`
:returns: `!True` on success otherwise `!False`
"""
@abc.abstractmethod
def take_leader(self):
"""take_leader"""
"""This method should create leader key with value = `~self._name` and ttl=`~self.ttl`
Since it could be called only on initial cluster bootstrap it could create this key regardless,
overwriting the key if necessary."""
@abc.abstractmethod
def race(self, path):
"""race"""
"""Race for cluster initialization.
:param path: usually this is just '/initialize'
:returns: `!True` if key has been created successfully.
this method should create atomically `path` key and return `!True`
otherwise it should return `!False`"""
@abc.abstractmethod
def delete_leader(self):
"""delete_leader"""
"""Voluntarily remove leader key from DCS
This method should remove leader key if current instance is the leader"""
def sleep(self, timeout):
sleep(timeout)
helpers/ha.py
+10 -2
View File
@@ -12,9 +12,17 @@ class Ha:
self.state_handler = state_handler
self.dcs = etcd
self.cluster = None
self.old_cluster = None
def load_cluster_from_dcs(self):
self.cluster = self.dcs.get_cluster()
cluster = self.dcs.get_cluster()
# We want to keep the state of cluster when it was healhy
if cluster.is_unlocked() and self.cluster and not self.cluster.is_unlocked():
self.old_cluster = self.cluster
if not self.old_cluster:
self.old_cluster = cluster
self.cluster = cluster
def acquire_lock(self):
return self.dcs.attempt_to_acquire_leader()
@@ -46,7 +54,7 @@ class Ha:
self.load_cluster_from_dcs()
if self.cluster.is_unlocked():
if self.state_handler.is_healthiest_node(self.cluster):
if self.state_handler.is_healthiest_node(self.old_cluster):
if self.acquire_lock():
if self.state_handler.is_leader() or self.state_handler.is_promoted:
return 'acquired session lock as a leader'
helpers/postgresql.py
+7 -6
View File
@@ -4,15 +4,13 @@ import psycopg2
import shlex
import shutil
import subprocess
import sys
import six
from helpers.utils import sleep
from six.moves.urllib_parse import urlparse
if sys.hexversion >= 0x03000000:
from urllib.parse import urlparse
if six.PY3:
long = int
else:
from urlparse import urlparse
logger = logging.getLogger(__name__)
@@ -266,7 +264,10 @@ class Postgresql:
member_cursor.close()
member_conn.close()
logger.error([self.name, member.name, row])
if not row[0] or row[1] < 0:
if not row[0]:
logger.warning('Master (%s) is still alive', member.name)
return False
if row[1] < 0:
return False
except psycopg2.Error:
continue
requirements-py3.txt
+1
View File
@@ -4,4 +4,5 @@ dnspython3
psycopg2
PyYAML
requests
six
kazoo>=2.2.1
tests/test_api.py
+1 -5
View File
@@ -4,11 +4,7 @@ import unittest
from helpers.api import RestApiHandler, RestApiServer
from test_postgresql import psycopg2_connect
if sys.hexversion >= 0x03000000:
from io import BytesIO as IO
else:
from StringIO import StringIO as IO
from six import BytesIO as IO
def throws(*args, **kwargs):
tests/test_ha.py
+11 -1
View File
@@ -60,6 +60,10 @@ def dead_etcd():
raise DCSError('Etcd is not responding properly')
def get_unlocked_cluster():
return Cluster(False, None, None, [])
class TestHa(unittest.TestCase):
def __init__(self, method_name='runTest'):
@@ -74,9 +78,15 @@ class TestHa(unittest.TestCase):
self.e = Etcd('foo', {'ttl': 30, 'host': 'remotehost:2379', 'scope': 'test'})
self.ha = Ha(self.p, self.e)
self.ha.load_cluster_from_dcs()
self.ha.cluster = Cluster(False, None, None, [])
self.ha.cluster = get_unlocked_cluster()
self.ha.load_cluster_from_dcs = nop
def test_load_cluster_from_dcs(self):
ha = Ha(self.p, self.e)
ha.load_cluster_from_dcs()
self.e.get_cluster = get_unlocked_cluster
ha.load_cluster_from_dcs()
def test_start_as_slave(self):
self.p.is_healthy = false
self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
tests/test_patroni.py
+1 -5
View File
@@ -11,16 +11,12 @@ import yaml
from patroni import Patroni, main
from helpers.dcs import Cluster, Member
from helpers.zookeeper import ZooKeeper
from six.moves import BaseHTTPServer
from test_etcd import requests_get, requests_put, requests_delete
from test_ha import true, false
from test_postgresql import Postgresql, subprocess_call, psycopg2_connect
from test_zookeeper import MockKazooClient
if sys.hexversion >= 0x03000000:
import http.server as BaseHTTPServer
else:
import BaseHTTPServer
def nop(*args, **kwargs):
pass
tests/test_postgresql.py
+8 -7
View File
@@ -23,10 +23,6 @@ def false(*args, **kwargs):
return False
def xlog_position():
return 1
class MockCursor:
def __init__(self):
@@ -44,9 +40,12 @@ class MockCursor:
elif sql.startswith('SELECT pg_current_xlog_location()'):
self.results = [(0,)]
elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
if params[0][0] != 0:
if params[0][0] == 1:
raise psycopg2.OperationalError()
self.results = [(False, 0)]
elif params[0][0] == 2:
self.results = [(True, -1)]
else:
self.results = [(False, 0)]
elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
self.results = [(0,)]
elif sql.startswith('SELECT pg_is_in_recovery()'):
@@ -188,8 +187,10 @@ class TestPostgresql(unittest.TestCase):
self.assertTrue(self.p.is_healthiest_node(cluster))
self.p.is_leader = false
self.assertFalse(self.p.is_healthiest_node(cluster))
self.p.xlog_position = xlog_position
self.p.xlog_position = lambda: 1
self.assertTrue(self.p.is_healthiest_node(cluster))
self.p.xlog_position = lambda: 2
self.assertFalse(self.p.is_healthiest_node(cluster))
self.p.config['maximum_lag_on_failover'] = -2
self.assertFalse(self.p.is_healthiest_node(cluster))