mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Merge branch 'master' of github.com:zalando/patroni into feature/faster_shutdown
This commit is contained in:
Alexander Kukushkin
+4
-7
@@ -8,7 +8,7 @@ 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 | sudo apt-key add -
|
||||
RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
|
||||
RUN apt-get update -y
|
||||
RUN apt-get upgrade -y
|
||||
|
||||
@@ -18,16 +18,13 @@ RUN pip install python-etcd
|
||||
|
||||
ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
|
||||
|
||||
RUN mkdir -p /patroni/helpers
|
||||
RUN mkdir -p /patroni/scripts
|
||||
ADD patroni.py /patroni/patroni.py
|
||||
ADD helpers /patroni/helpers
|
||||
ADD scripts /patroni/scripts
|
||||
ADD patroni.py /patroni.py
|
||||
ADD patroni/ /patroni
|
||||
|
||||
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
|
||||
### 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
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
include requirements*
|
||||
include *.rst
|
||||
recursive-include patroni *.py
|
||||
@@ -1,129 +0,0 @@
|
||||
[](https://travis-ci.org/zalando/patroni)
|
||||
[](https://coveralls.io/r/zalando/patroni?branch=master)
|
||||
# Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd
|
||||
|
||||
Patroni was previously known as Governor.
|
||||
|
||||
*There are many ways to run high availability with PostgreSQL; here we present a template for you to create your own custom fit high availability solution using python and distributed configuration store (like ZooKeeper or etcd) for maximum accessibility.*
|
||||
|
||||
## Getting Started
|
||||
To get started, do the following from different terminals:
|
||||
|
||||
```
|
||||
> etcd --data-dir=data/etcd
|
||||
> ./patroni.py postgres0.yml
|
||||
> ./patroni.py postgres1.yml
|
||||
```
|
||||
|
||||
From there, you will see a high-availability cluster start up. Test
|
||||
different settings in the YAML files to see how behavior changes. Kill
|
||||
some of the different components to see how the system behaves.
|
||||
|
||||
Add more `postgres*.yml` files to create an even larger cluster.
|
||||
|
||||
We provide a haproxy configuration, which will give your application a single endpoint for connecting to the cluster's leader. To configure, run:
|
||||
|
||||
```
|
||||
> haproxy -f haproxy.cfg
|
||||
```
|
||||
|
||||
```
|
||||
> psql --host 127.0.0.1 --port 5000 postgres
|
||||
```
|
||||
|
||||
## How Patroni works
|
||||
|
||||
For a diagram of the high availability decision loop, see the included a PDF: [postgres-ha.pdf](https://github.com/zalando/patroni/blob/master/postgres-ha.pdf)
|
||||
|
||||
## YAML Configuration
|
||||
|
||||
For an example file, see `postgres0.yml`. Below is an explanation of settings:
|
||||
|
||||
* *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
|
||||
* *loop_wait*: the number of seconds the loop will sleep
|
||||
|
||||
* *restapi*
|
||||
* *listen*: ip address + port that Patroni will listen to provide health-check information for haproxy.
|
||||
* *connect_address*: ip address + port through which restapi is accessible.
|
||||
|
||||
* *etcd*
|
||||
* *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd
|
||||
* *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
|
||||
* *host*: the host:port for the etcd endpoint
|
||||
|
||||
* *zookeeper*
|
||||
* *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...']
|
||||
* *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
|
||||
* *listen*: ip address + port that Postgres listening. Must be accessible from other nodes in the cluster if using streaming replication.
|
||||
* *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
|
||||
|
||||
## Replication choices
|
||||
|
||||
Patroni uses Postgres' streaming replication. By default, this replication is asynchronous. For more information, see the [Postgres documentation on streaming replication](http://www.postgresql.org/docs/current/static/warm-standby.html#STREAMING-REPLICATION).
|
||||
|
||||
Patroni's asynchronous replication configuration allows for `maximum_lag_on_failover` settings. This setting ensures failover will not occur if a follower is more than a certain number of bytes behind the follower. This setting should be increased or decreased based on business requirements.
|
||||
|
||||
When asynchronous replication is not best for your use-case, investigate how Postgres's [synchronous replication](http://www.postgresql.org/docs/current/static/warm-standby.html#SYNCHRONOUS-REPLICATION) works. Synchronous replication ensures consistency across a cluster by confirming that writes are written to a secondary before returning to the connecting client with a success. The cost of synchronous replication will be reduced throughput on writes. This throughput will be entirely based on network performance. In hosted datacenter environments (like AWS, Rackspace, or any network you do not control), synchrous replication increases the variability of write performance significantly. If followers become inaccessible from the leader, the leader will becomes effectively readonly.
|
||||
|
||||
To enable a simple synchronous replication test, add the follow lines to the `parameters` section of your YAML configuration files.
|
||||
|
||||
```YAML
|
||||
synchronous_commit: "on"
|
||||
synchronous_standby_names: "*"
|
||||
```
|
||||
|
||||
When using synchronous replication, use at least a 3-Postgres data nodes to ensure write availability if one host fails.
|
||||
|
||||
Choosing your replication schema is dependent on the many business decisions. Investigate both async and sync replication, as well as other HA solutions, to determine which solution is best for you.
|
||||
|
||||
## Applications should not use superusers
|
||||
|
||||
When connecting from an application, always use a non-superuser. Patroni requires access to the database to function properly. By using a superuser from application, you can potentially use the entire connection pool, including the connections reserved for superusers with the `superuser_reserved_connections` setting. If Patroni cannot access the Primary, because the connection pool is full, behavior will be undesireable.
|
||||
|
||||
## Requirements on a Mac
|
||||
|
||||
Run the following on a Mac to install requirements:
|
||||
|
||||
```
|
||||
brew install postgresql etcd haproxy libyaml python
|
||||
pip install psycopg2 pyyaml
|
||||
```
|
||||
|
||||
## Notice
|
||||
|
||||
There are many different ways to do HA with PostgreSQL, see [the
|
||||
PostgreSQL documentation](https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling) for a complete list.
|
||||
|
||||
We call this project a "template" because it is far from a one-size fits
|
||||
all, or a plug-and-play replication system. It will have it's own
|
||||
caveats. Use wisely.
|
||||
+222
@@ -0,0 +1,222 @@
|
||||
|Build Status| |Coverage Status|
|
||||
|
||||
Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd
|
||||
------------------------------------------------------------
|
||||
|
||||
Patroni was previously known as Governor.
|
||||
|
||||
*There are many ways to run high availability with PostgreSQL; here we
|
||||
present a template for you to create your own custom fit high
|
||||
availability solution using python and distributed configuration store
|
||||
(like ZooKeeper or etcd) for maximum accessibility.*
|
||||
|
||||
Getting Started
|
||||
---------------
|
||||
|
||||
To get started, do the following from different terminals:
|
||||
|
||||
::
|
||||
|
||||
> etcd --data-dir=data/etcd
|
||||
> ./patroni.py postgres0.yml
|
||||
> ./patroni.py postgres1.yml
|
||||
|
||||
From there, you will see a high-availability cluster start up. Test
|
||||
different settings in the YAML files to see how behavior changes. Kill
|
||||
some of the different components to see how the system behaves.
|
||||
|
||||
Add more ``postgres*.yml`` files to create an even larger cluster.
|
||||
|
||||
We provide a haproxy configuration, which will give your application a
|
||||
single endpoint for connecting to the cluster's leader. To configure,
|
||||
run:
|
||||
|
||||
::
|
||||
|
||||
> haproxy -f haproxy.cfg
|
||||
|
||||
::
|
||||
|
||||
> psql --host 127.0.0.1 --port 5000 postgres
|
||||
|
||||
How Patroni works
|
||||
-----------------
|
||||
|
||||
For a diagram of the high availability decision loop, see the included a
|
||||
PDF:
|
||||
`postgres-ha.pdf <https://github.com/zalando/patroni/blob/master/postgres-ha.pdf>`__
|
||||
|
||||
YAML Configuration
|
||||
------------------
|
||||
|
||||
For an example file, see ``postgres0.yml``. Below is an explanation of
|
||||
settings:
|
||||
|
||||
- *ttl*: the TTL to acquire the leader lock. Think of it as the length
|
||||
of time before automatic failover process is initiated.
|
||||
- *loop\_wait*: the number of seconds the loop will sleep
|
||||
|
||||
- *restapi*
|
||||
- *listen*: ip address + port that Patroni will listen to provide
|
||||
health-check information for haproxy.
|
||||
- *connect\_address*: ip address + port through which restapi is
|
||||
accessible.
|
||||
|
||||
- *etcd*
|
||||
- *scope*: the relative path used on etcd's http api for this
|
||||
deployment, thus you can run multiple HA deployments from a single
|
||||
etcd
|
||||
- *ttl*: the TTL to acquire the leader lock. Think of it as the length
|
||||
of time before automatic failover process is initiated.
|
||||
- *host*: the host:port for the etcd endpoint
|
||||
|
||||
- *zookeeper*
|
||||
- *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.
|
||||
- *reconnect\_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...']
|
||||
- *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
|
||||
- *listen*: ip address + port that Postgres listening. Must be
|
||||
accessible from other nodes in the cluster if using streaming
|
||||
replication.
|
||||
- *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
|
||||
- *use\_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3, and you should comment out max_replication_slots.
|
||||
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*: additional configuration settings written to recovery.conf when configuring follower
|
||||
- *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work.
|
||||
|
||||
Replication choices
|
||||
-------------------
|
||||
|
||||
Patroni uses Postgres' streaming replication. By default, this
|
||||
replication is asynchronous. For more information, see the `Postgres
|
||||
documentation on streaming
|
||||
replication <http://www.postgresql.org/docs/current/static/warm-standby.html#STREAMING-REPLICATION>`__.
|
||||
|
||||
Patroni's asynchronous replication configuration allows for
|
||||
``maximum_lag_on_failover`` settings. This setting ensures failover will
|
||||
not occur if a follower is more than a certain number of bytes behind
|
||||
the follower. This setting should be increased or decreased based on
|
||||
business requirements.
|
||||
|
||||
When asynchronous replication is not best for your use-case, investigate
|
||||
how Postgres's `synchronous
|
||||
replication <http://www.postgresql.org/docs/current/static/warm-standby.html#SYNCHRONOUS-REPLICATION>`__
|
||||
works. Synchronous replication ensures consistency across a cluster by
|
||||
confirming that writes are written to a secondary before returning to
|
||||
the connecting client with a success. The cost of synchronous
|
||||
replication will be reduced throughput on writes. This throughput will
|
||||
be entirely based on network performance. In hosted datacenter
|
||||
environments (like AWS, Rackspace, or any network you do not control),
|
||||
synchrous replication increases the variability of write performance
|
||||
significantly. If followers become inaccessible from the leader, the
|
||||
leader will becomes effectively readonly.
|
||||
|
||||
To enable a simple synchronous replication test, add the follow lines to
|
||||
the ``parameters`` section of your YAML configuration files.
|
||||
|
||||
.. code:: YAML
|
||||
|
||||
synchronous_commit: "on"
|
||||
synchronous_standby_names: "*"
|
||||
|
||||
When using synchronous replication, use at least a 3-Postgres data nodes
|
||||
to ensure write availability if one host fails.
|
||||
|
||||
Choosing your replication schema is dependent on the many business
|
||||
decisions. Investigate both async and sync replication, as well as other
|
||||
HA solutions, to determine which solution is best for you.
|
||||
|
||||
Applications should not use superusers
|
||||
--------------------------------------
|
||||
|
||||
When connecting from an application, always use a non-superuser. Patroni
|
||||
requires access to the database to function properly. By using a
|
||||
superuser from application, you can potentially use the entire
|
||||
connection pool, including the connections reserved for superusers with
|
||||
the ``superuser_reserved_connections`` setting. If Patroni cannot access
|
||||
the Primary, because the connection pool is full, behavior will be
|
||||
undesireable.
|
||||
|
||||
Requirements on a Mac
|
||||
---------------------
|
||||
|
||||
Run the following on a Mac to install requirements:
|
||||
|
||||
::
|
||||
|
||||
brew install postgresql etcd haproxy libyaml python
|
||||
pip install psycopg2 pyyaml
|
||||
|
||||
Notice
|
||||
------
|
||||
|
||||
There are many different ways to do HA with PostgreSQL, see `the
|
||||
PostgreSQL
|
||||
documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__
|
||||
for a complete list.
|
||||
|
||||
We call this project a "template" because it is far from a one-size fits
|
||||
all, or a plug-and-play replication system. It will have it's own
|
||||
caveats. Use wisely.
|
||||
|
||||
.. |Build Status| image:: https://travis-ci.org/zalando/patroni.svg?branch=master
|
||||
:target: https://travis-ci.org/zalando/patroni
|
||||
.. |Coverage Status| image:: https://coveralls.io/repos/zalando/patroni/badge.svg?branch=master
|
||||
:target: https://coveralls.io/r/zalando/patroni?branch=master
|
||||
+3
-2
@@ -1,6 +1,7 @@
|
||||
# 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-SNAPSHOT
|
||||
You can run Patroni in a docker container using this Dockerfile, or by using one of the Docker image at
|
||||
|
||||
https://os-registry.stups.zalan.do/v1/repositories/acid/patroni/tags
|
||||
|
||||
This Dockerfile is meant in aiding development of Patroni and quick testing of features. It is not a production-worthy
|
||||
Dockerfile
|
||||
|
||||
@@ -133,5 +133,5 @@ then
|
||||
sleep 60
|
||||
done
|
||||
else
|
||||
exec /patroni/patroni.py /patroni/postgres.yml
|
||||
exec python /patroni.py /patroni/postgres.yml
|
||||
fi
|
||||
|
||||
@@ -1,109 +0,0 @@
|
||||
import datetime
|
||||
import os
|
||||
import re
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
|
||||
interrupted_sleep = False
|
||||
reap_children = False
|
||||
|
||||
_DATE_TIME_RE = re.compile(r'''^
|
||||
(?P<year>\d{4})\-(?P<month>\d{2})\-(?P<day>\d{2}) # date
|
||||
T
|
||||
(?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})\.(?P<microsecond>\d{6}) # time
|
||||
\d*Z$''', re.X)
|
||||
|
||||
|
||||
def parse_datetime(time_str):
|
||||
"""
|
||||
>>> parse_datetime('2015-06-10T12:56:30.552539016Z')
|
||||
datetime.datetime(2015, 6, 10, 12, 56, 30, 552539)
|
||||
>>> parse_datetime('2015-06-10 12:56:30.552539016Z')
|
||||
"""
|
||||
m = _DATE_TIME_RE.match(time_str)
|
||||
if not m:
|
||||
return None
|
||||
p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' '))
|
||||
return datetime.datetime(**p)
|
||||
|
||||
|
||||
def calculate_ttl(expiration):
|
||||
"""
|
||||
>>> calculate_ttl(None)
|
||||
>>> calculate_ttl('2015-06-10 12:56:30.552539016Z')
|
||||
"""
|
||||
if not expiration:
|
||||
return None
|
||||
expiration = parse_datetime(expiration)
|
||||
if not expiration:
|
||||
return None
|
||||
now = datetime.datetime.utcnow()
|
||||
return int((expiration - now).total_seconds())
|
||||
|
||||
|
||||
def lsn_to_bytes(value):
|
||||
"""
|
||||
>>> lsn_to_bytes('1/66000060')
|
||||
6006243424
|
||||
>>> lsn_to_bytes('j/66000060')
|
||||
0
|
||||
"""
|
||||
try:
|
||||
e = value.split('/')
|
||||
if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0:
|
||||
return (int(e[0], 16) << 32) | int(e[1], 16)
|
||||
except ValueError:
|
||||
pass
|
||||
return 0
|
||||
|
||||
|
||||
def bytes_to_lsn(value):
|
||||
"""
|
||||
>>> bytes_to_lsn(6006243424)
|
||||
'1/66000060'
|
||||
"""
|
||||
id = value >> 32
|
||||
off = value & 0xffffffff
|
||||
return '%x/%x' % (id, off)
|
||||
|
||||
|
||||
def sigterm_handler(signo, stack_frame):
|
||||
sys.exit()
|
||||
|
||||
|
||||
def sigchld_handler(signo, stack_frame):
|
||||
global interrupted_sleep, reap_children
|
||||
reap_children = interrupted_sleep = True
|
||||
|
||||
|
||||
def sleep(interval):
|
||||
global interrupted_sleep
|
||||
current_time = time.time()
|
||||
end_time = current_time + interval
|
||||
while current_time < end_time:
|
||||
interrupted_sleep = False
|
||||
time.sleep(end_time - current_time)
|
||||
if not interrupted_sleep: # we will ignore only sigchld
|
||||
break
|
||||
current_time = time.time()
|
||||
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:
|
||||
try:
|
||||
while True:
|
||||
ret = os.waitpid(-1, os.WNOHANG)
|
||||
if ret == (0, 0):
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
reap_children = False
|
||||
+1
-119
@@ -1,123 +1,5 @@
|
||||
#!/usr/bin/env python
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import yaml
|
||||
|
||||
from helpers.api import RestApiServer
|
||||
from helpers.etcd import Etcd
|
||||
from helpers.ha import Ha
|
||||
from helpers.postgresql import Postgresql
|
||||
from helpers.utils import setup_signal_handlers, sleep, reap_children
|
||||
from helpers.zookeeper import ZooKeeper
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Patroni:
|
||||
|
||||
def __init__(self, config):
|
||||
self.nap_time = config['loop_wait']
|
||||
self.postgresql = Postgresql(config['postgresql'])
|
||||
self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
|
||||
host, port = config['restapi']['listen'].split(':')
|
||||
self.api = RestApiServer(self, config['restapi'])
|
||||
self.next_run = time.time()
|
||||
self.shutdown_member_ttl = 300
|
||||
|
||||
@staticmethod
|
||||
def get_dcs(name, config):
|
||||
if 'etcd' in config:
|
||||
return Etcd(name, config['etcd'])
|
||||
if 'zookeeper' in config:
|
||||
return ZooKeeper(name, config['zookeeper'])
|
||||
raise Exception('Can not find sutable configuration of distributed configuration store')
|
||||
|
||||
def touch_member(self, ttl=None):
|
||||
connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
|
||||
if self.ha.cluster:
|
||||
for m in self.ha.cluster.members:
|
||||
# Do not update member TTL when it is far from being expired
|
||||
if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
|
||||
return True
|
||||
return self.ha.dcs.touch_member(connection_string, ttl)
|
||||
|
||||
def initialize(self):
|
||||
# wait for etcd to be available
|
||||
while not self.touch_member():
|
||||
logger.info('waiting on DCS')
|
||||
sleep(5)
|
||||
|
||||
# is data directory empty?
|
||||
if self.postgresql.data_directory_empty():
|
||||
# racing to initialize
|
||||
if self.ha.dcs.race('/initialize'):
|
||||
self.postgresql.initialize()
|
||||
self.ha.dcs.take_leader()
|
||||
self.postgresql.start()
|
||||
self.postgresql.create_replication_user()
|
||||
self.postgresql.create_connection_users()
|
||||
else:
|
||||
while True:
|
||||
leader = self.ha.dcs.current_leader()
|
||||
if leader and self.postgresql.sync_from_leader(leader):
|
||||
self.postgresql.write_recovery_conf(leader)
|
||||
self.postgresql.start()
|
||||
break
|
||||
sleep(5)
|
||||
elif self.postgresql.is_running():
|
||||
self.postgresql.load_replication_slots()
|
||||
|
||||
def schedule_next_run(self):
|
||||
self.next_run += self.nap_time
|
||||
current_time = time.time()
|
||||
nap_time = self.next_run - current_time
|
||||
if nap_time <= 0:
|
||||
self.next_run = current_time
|
||||
else:
|
||||
self.ha.dcs.sleep(nap_time)
|
||||
|
||||
def run(self):
|
||||
self.api.start()
|
||||
self.next_run = time.time()
|
||||
|
||||
while True:
|
||||
self.touch_member()
|
||||
logger.info(self.ha.run_cycle())
|
||||
try:
|
||||
if self.ha.state_handler.is_leader():
|
||||
self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
|
||||
else:
|
||||
self.ha.state_handler.drop_replication_slots()
|
||||
except:
|
||||
logger.exception('Exception when changing replication slots')
|
||||
self.schedule_next_run()
|
||||
reap_children()
|
||||
|
||||
|
||||
def main():
|
||||
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
|
||||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||||
setup_signal_handlers()
|
||||
|
||||
if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
|
||||
print('Usage: {} config.yml'.format(sys.argv[0]))
|
||||
return
|
||||
|
||||
with open(sys.argv[1], 'r') as f:
|
||||
config = yaml.load(f)
|
||||
|
||||
patroni = Patroni(config)
|
||||
try:
|
||||
patroni.initialize()
|
||||
patroni.run()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal
|
||||
patroni.postgresql.stop()
|
||||
patroni.ha.dcs.delete_leader()
|
||||
from patroni import main
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import yaml
|
||||
|
||||
from patroni.api import RestApiServer
|
||||
from patroni.etcd import Etcd
|
||||
from patroni.exceptions import DCSError
|
||||
from patroni.ha import Ha
|
||||
from patroni.postgresql import Postgresql
|
||||
from patroni.utils import setup_signal_handlers, sleep, reap_children
|
||||
from patroni.zookeeper import ZooKeeper
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Patroni:
|
||||
|
||||
def __init__(self, config):
|
||||
self.nap_time = config['loop_wait']
|
||||
self.postgresql = Postgresql(config['postgresql'])
|
||||
self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
|
||||
host, port = config['restapi']['listen'].split(':')
|
||||
self.api = RestApiServer(self, config['restapi'])
|
||||
self.next_run = time.time()
|
||||
self.shutdown_member_ttl = 300
|
||||
|
||||
@staticmethod
|
||||
def get_dcs(name, config):
|
||||
if 'etcd' in config:
|
||||
return Etcd(name, config['etcd'])
|
||||
if 'zookeeper' in config:
|
||||
return ZooKeeper(name, config['zookeeper'])
|
||||
raise Exception('Can not find sutable configuration of distributed configuration store')
|
||||
|
||||
def touch_member(self, ttl=None):
|
||||
connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
|
||||
if self.ha.cluster:
|
||||
for m in self.ha.cluster.members:
|
||||
# Do not update member TTL when it is far from being expired
|
||||
if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
|
||||
return True
|
||||
return self.ha.dcs.touch_member(connection_string, ttl)
|
||||
|
||||
def cleanup_on_failed_initialization(self):
|
||||
""" cleanup the DCS if initialization was not successfull """
|
||||
logger.info("removing initialize key after failed attempt to initialize the cluster")
|
||||
self.ha.dcs.cancel_initialization()
|
||||
self.touch_member(self.shutdown_member_ttl)
|
||||
self.postgresql.stop()
|
||||
self.postgresql.move_data_directory()
|
||||
|
||||
def initialize(self):
|
||||
# wait for etcd to be available
|
||||
while not self.touch_member():
|
||||
logger.info('waiting on DCS')
|
||||
sleep(5)
|
||||
|
||||
# is data directory empty?
|
||||
if self.postgresql.data_directory_empty():
|
||||
while True:
|
||||
try:
|
||||
cluster = self.ha.dcs.get_cluster()
|
||||
if not cluster.is_unlocked(): # the leader already exists
|
||||
if not cluster.initialize:
|
||||
self.ha.dcs.initialize()
|
||||
self.postgresql.bootstrap(cluster.leader)
|
||||
break
|
||||
# racing to initialize
|
||||
elif not cluster.initialize and self.ha.dcs.initialize():
|
||||
try:
|
||||
self.postgresql.bootstrap()
|
||||
except:
|
||||
# bail out and clean the initialize flag.
|
||||
self.cleanup_on_failed_initialization()
|
||||
raise
|
||||
self.ha.dcs.take_leader()
|
||||
break
|
||||
except DCSError:
|
||||
logger.info('waiting on DCS')
|
||||
sleep(5)
|
||||
elif self.postgresql.is_running():
|
||||
self.postgresql.load_replication_slots()
|
||||
|
||||
def schedule_next_run(self):
|
||||
if self.postgresql.is_promoted:
|
||||
self.next_run = time.time()
|
||||
self.next_run += self.nap_time
|
||||
current_time = time.time()
|
||||
nap_time = self.next_run - current_time
|
||||
if nap_time <= 0:
|
||||
self.next_run = current_time
|
||||
else:
|
||||
self.ha.dcs.watch(nap_time)
|
||||
|
||||
def run(self):
|
||||
self.api.start()
|
||||
self.next_run = time.time()
|
||||
|
||||
while True:
|
||||
self.touch_member()
|
||||
logger.info(self.ha.run_cycle())
|
||||
try:
|
||||
if self.ha.state_handler.is_leader():
|
||||
self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
|
||||
else:
|
||||
self.ha.state_handler.drop_replication_slots()
|
||||
except:
|
||||
logger.exception('Exception when changing replication slots')
|
||||
reap_children()
|
||||
self.schedule_next_run()
|
||||
|
||||
|
||||
def main():
|
||||
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
|
||||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||||
setup_signal_handlers()
|
||||
|
||||
if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
|
||||
print('Usage: {} config.yml'.format(sys.argv[0]))
|
||||
return
|
||||
|
||||
with open(sys.argv[1], 'r') as f:
|
||||
config = yaml.load(f)
|
||||
|
||||
patroni = Patroni(config)
|
||||
patroni.initialize()
|
||||
try:
|
||||
patroni.run()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal
|
||||
patroni.postgresql.stop()
|
||||
patroni.ha.dcs.delete_leader()
|
||||
@@ -0,0 +1,5 @@
|
||||
from patroni import main
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -1,7 +1,8 @@
|
||||
import abc
|
||||
|
||||
from collections import namedtuple
|
||||
from helpers.utils import calculate_ttl, sleep
|
||||
from patroni.exceptions import DCSError
|
||||
from patroni.utils import calculate_ttl, sleep
|
||||
from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
|
||||
|
||||
|
||||
@@ -22,24 +23,11 @@ def parse_connection_string(value):
|
||||
return conn_url, api_url
|
||||
|
||||
|
||||
class DCSError(Exception):
|
||||
"""Parent class for all kind of exceptions related to selected distributed configuration store"""
|
||||
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
|
||||
def __str__(self):
|
||||
"""
|
||||
>>> str(DCSError('foo'))
|
||||
"'foo'"
|
||||
"""
|
||||
return repr(self.value)
|
||||
|
||||
|
||||
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 index: modification index of a given member key in a Configuration Store
|
||||
: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
|
||||
@@ -50,11 +38,30 @@ class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl'))
|
||||
return calculate_ttl(self.expiration) or -1
|
||||
|
||||
|
||||
class Leader(namedtuple('Leader', 'index,expiration,ttl,member')):
|
||||
|
||||
"""Immutable object (namedtuple) which represents leader key.
|
||||
Consists of the following fields:
|
||||
:param index: modification index of a leader key in a Configuration Store
|
||||
:param expiration: expiration time of the leader key
|
||||
:param ttl: ttl of the leader key
|
||||
:param member: reference to a `Member` object which represents current leader (see `Cluster.members`)"""
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.member.name
|
||||
|
||||
@property
|
||||
def conn_url(self):
|
||||
return self.member.conn_url
|
||||
|
||||
|
||||
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 leader: `Leader` object which represents current leader of the cluster
|
||||
:param last_leader_operation: int or long object containing position of last known leader operation.
|
||||
This value is stored in `/optime/leader` key
|
||||
:param members: list of Member object, all PostgreSQL cluster members including leader"""
|
||||
@@ -67,6 +74,12 @@ class AbstractDCS:
|
||||
|
||||
__metaclass__ = abc.ABCMeta
|
||||
|
||||
_INITIALIZE = 'initialize'
|
||||
_LEADER = 'leader'
|
||||
_MEMBERS = 'members/'
|
||||
_OPTIME = 'optime'
|
||||
_LEADER_OPTIME = _OPTIME + '/' + _LEADER
|
||||
|
||||
def __init__(self, name, config):
|
||||
"""
|
||||
:param name: name of current instance (the same value as `~Postgresql.name`)
|
||||
@@ -74,10 +87,31 @@ class AbstractDCS:
|
||||
i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
|
||||
"""
|
||||
self._name = name
|
||||
self._base_path = '/service/' + config['scope']
|
||||
self._scope = config['scope']
|
||||
self._base_path = '/service/' + self._scope
|
||||
|
||||
def client_path(self, path):
|
||||
return self._base_path + path
|
||||
return '/'.join([self._base_path, path.lstrip('/')])
|
||||
|
||||
@property
|
||||
def initialize_path(self):
|
||||
return self.client_path(self._INITIALIZE)
|
||||
|
||||
@property
|
||||
def members_path(self):
|
||||
return self.client_path(self._MEMBERS)
|
||||
|
||||
@property
|
||||
def member_path(self):
|
||||
return self.client_path(self._MEMBERS + self._name)
|
||||
|
||||
@property
|
||||
def leader_path(self):
|
||||
return self.client_path(self._LEADER)
|
||||
|
||||
@property
|
||||
def leader_optime_path(self):
|
||||
return self.client_path(self._LEADER_OPTIME)
|
||||
|
||||
@abc.abstractmethod
|
||||
def get_cluster(self):
|
||||
@@ -131,12 +165,11 @@ class AbstractDCS:
|
||||
overwriting the key if necessary."""
|
||||
|
||||
@abc.abstractmethod
|
||||
def race(self, path):
|
||||
def initialize(self):
|
||||
"""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`
|
||||
this method should create atomically initialize key and return `!True`
|
||||
otherwise it should return `!False`"""
|
||||
|
||||
@abc.abstractmethod
|
||||
@@ -144,5 +177,9 @@ class AbstractDCS:
|
||||
"""Voluntarily remove leader key from DCS
|
||||
This method should remove leader key if current instance is the leader"""
|
||||
|
||||
def sleep(self, timeout):
|
||||
@abc.abstractmethod
|
||||
def cancel_initialization(self):
|
||||
""" Removes the initialize key for a cluster """
|
||||
|
||||
def watch(self, timeout):
|
||||
sleep(timeout)
|
||||
@@ -5,11 +5,13 @@ import os
|
||||
import random
|
||||
import requests
|
||||
import socket
|
||||
import time
|
||||
import urllib3
|
||||
|
||||
from dns.exception import DNSException
|
||||
from dns import resolver
|
||||
from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
|
||||
from helpers.utils import sleep
|
||||
from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
|
||||
from patroni.utils import Retry, RetryFailedError, sleep
|
||||
from requests.exceptions import RequestException
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -59,6 +61,16 @@ class Client(etcd.Client):
|
||||
logger.exception('Can not resolve SRV for %s', host)
|
||||
return []
|
||||
|
||||
# try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81
|
||||
def _result_from_response(self, response):
|
||||
try:
|
||||
response.data.decode('utf-8')
|
||||
except urllib3.exceptions.TimeoutError:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise etcd.EtcdException('Unable to decode server response: %s' % e)
|
||||
return super(Client, self)._result_from_response(response)
|
||||
|
||||
def _get_machines_cache_from_srv(self, discovery_srv):
|
||||
"""Fetch list of etcd-cluster member by resolving _etcd-server._tcp. SRV record.
|
||||
This record should contain list of host and peer ports which could be used to run
|
||||
@@ -124,7 +136,7 @@ def catch_etcd_errors(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return not func(*args, **kwargs) is None
|
||||
except etcd.EtcdException:
|
||||
except (RetryFailedError, etcd.EtcdException):
|
||||
return False
|
||||
return wrapper
|
||||
|
||||
@@ -135,7 +147,16 @@ class Etcd(AbstractDCS):
|
||||
super(Etcd, self).__init__(name, config)
|
||||
self.ttl = config['ttl']
|
||||
self.member_ttl = config.get('member_ttl', 3600)
|
||||
self._retry = Retry(deadline=10, max_delay=1, max_tries=-1,
|
||||
retry_exceptions=(etcd.EtcdConnectionFailed,
|
||||
etcd.EtcdLeaderElectionInProgress,
|
||||
etcd.EtcdWatcherCleared,
|
||||
etcd.EtcdEventIndexCleared))
|
||||
self.client = self.get_etcd_client(config)
|
||||
self.cluster = None
|
||||
|
||||
def retry(self, *args, **kwargs):
|
||||
return self._retry.copy()(*args, **kwargs)
|
||||
|
||||
def get_etcd_client(self, config):
|
||||
client = None
|
||||
@@ -154,61 +175,91 @@ class Etcd(AbstractDCS):
|
||||
|
||||
def get_cluster(self):
|
||||
try:
|
||||
result = self.client.read(self.client_path(''), recursive=True)
|
||||
result = self.retry(self.client.read, self.client_path(''), recursive=True)
|
||||
nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
|
||||
|
||||
# get initialize flag
|
||||
initialize = bool(nodes.get('initialize', False))
|
||||
initialize = bool(nodes.get(self._INITIALIZE, False))
|
||||
|
||||
# get last leader operation
|
||||
last_leader_operation = nodes.get('optime/leader', None)
|
||||
last_leader_operation = nodes.get(self._LEADER_OPTIME, None)
|
||||
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
|
||||
|
||||
# get list of members
|
||||
members = [self.member(n) for k, n in nodes.items() if k.startswith('members/') and len(k.split('/')) == 2]
|
||||
members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
|
||||
|
||||
# get leader
|
||||
leader = nodes.get('leader', None)
|
||||
leader = nodes.get(self._LEADER, None)
|
||||
if leader:
|
||||
leader = Member(-1, leader.value, None, None, None, None)
|
||||
leader = ([m for m in members if m.name == leader.name] or [leader])[0]
|
||||
member = Member(-1, leader.value, None, None, None, None)
|
||||
member = ([m for m in members if m.name == leader.value] or [member])[0]
|
||||
leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member)
|
||||
|
||||
return Cluster(initialize, leader, last_leader_operation, members)
|
||||
self.cluster = Cluster(initialize, leader, last_leader_operation, members)
|
||||
except etcd.EtcdKeyNotFound:
|
||||
return Cluster(False, None, None, [])
|
||||
self.cluster = Cluster(False, None, None, [])
|
||||
except:
|
||||
self.cluster = None
|
||||
logger.exception('get_cluster')
|
||||
|
||||
raise EtcdError('Etcd is not responding properly')
|
||||
raise EtcdError('Etcd is not responding properly')
|
||||
return self.cluster
|
||||
|
||||
@catch_etcd_errors
|
||||
def touch_member(self, connection_string, ttl=None):
|
||||
return self.client.set(self.client_path('/members/' + self._name), connection_string, ttl or self.member_ttl)
|
||||
return self.retry(self.client.set, self.member_path, connection_string, ttl or self.member_ttl)
|
||||
|
||||
@catch_etcd_errors
|
||||
def take_leader(self):
|
||||
return self.client.set(self.client_path('/leader'), self._name, self.ttl)
|
||||
return self.retry(self.client.set, self.leader_path, self._name, self.ttl)
|
||||
|
||||
@catch_etcd_errors
|
||||
def attempt_to_acquire_leader(self):
|
||||
ret = self.client.write(self.client_path('/leader'), self._name, ttl=self.ttl, prevExist=False)
|
||||
ret or logger.info('Could not take out TTL lock')
|
||||
return ret
|
||||
try:
|
||||
return bool(self.retry(self.client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
|
||||
except etcd.EtcdAlreadyExist:
|
||||
logger.info('Could not take out TTL lock')
|
||||
except (RetryFailedError, etcd.EtcdException):
|
||||
pass
|
||||
return False
|
||||
|
||||
@catch_etcd_errors
|
||||
def write_leader_optime(self, state_handler):
|
||||
return self.client.set(self.client_path('/optime/leader'), state_handler.last_operation())
|
||||
return self.client.set(self.leader_optime_path, state_handler.last_operation())
|
||||
|
||||
@catch_etcd_errors
|
||||
def update_leader(self, state_handler):
|
||||
ret = self.client.test_and_set(self.client_path('/leader'), self._name, self._name, self.ttl)
|
||||
ret = self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
|
||||
ret and self.write_leader_optime(state_handler)
|
||||
return ret
|
||||
|
||||
@catch_etcd_errors
|
||||
def race(self, path):
|
||||
return self.client.write(self.client_path(path), self._name, prevExist=False)
|
||||
def initialize(self):
|
||||
return self.client.write(self.initialize_path, self._name, prevExist=False)
|
||||
|
||||
@catch_etcd_errors
|
||||
def delete_leader(self):
|
||||
return self.client.delete(self.client_path('/leader'), prevValue=self._name)
|
||||
return self.client.delete(self.leader_path, prevValue=self._name)
|
||||
|
||||
@catch_etcd_errors
|
||||
def cancel_initialization(self):
|
||||
return self.client.delete(self.initialize_path, prevValue=self._name)
|
||||
|
||||
def watch(self, timeout):
|
||||
# watch on leader key changes if it is defined and current node is not lock owner
|
||||
if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name:
|
||||
end_time = time.time() + timeout
|
||||
index = self.cluster.leader.index
|
||||
while index and timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect
|
||||
try:
|
||||
res = self.client.watch(self.leader_path, index=index + 1, timeout=timeout)
|
||||
if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.name:
|
||||
return
|
||||
index = res.modifiedIndex
|
||||
except urllib3.exceptions.TimeoutError:
|
||||
self.client.http.clear()
|
||||
return
|
||||
except etcd.EtcdException:
|
||||
index = None
|
||||
|
||||
timeout = end_time - time.time()
|
||||
|
||||
timeout > 0 and super(Etcd, self).watch(timeout)
|
||||
@@ -0,0 +1,21 @@
|
||||
class PatroniException(Exception):
|
||||
|
||||
"""Parent class for all kind of exceptions related to selected distributed configuration store"""
|
||||
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
|
||||
def __str__(self):
|
||||
"""
|
||||
>>> str(DCSError('foo'))
|
||||
"'foo'"
|
||||
"""
|
||||
return repr(self.value)
|
||||
|
||||
|
||||
class PostgresException(PatroniException):
|
||||
pass
|
||||
|
||||
|
||||
class DCSError(PatroniException):
|
||||
pass
|
||||
@@ -1,6 +1,6 @@
|
||||
import logging
|
||||
|
||||
from helpers.dcs import DCSError
|
||||
from patroni.dcs import DCSError
|
||||
from psycopg2 import InterfaceError, OperationalError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -4,14 +4,12 @@ import psycopg2
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import six
|
||||
import time
|
||||
|
||||
from helpers.utils import sleep
|
||||
from patroni.exceptions import PostgresException
|
||||
from patroni.utils import sleep
|
||||
from six.moves.urllib_parse import urlparse
|
||||
|
||||
if six.PY3:
|
||||
long = int
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ACTION_ON_START = "on_start"
|
||||
@@ -50,6 +48,7 @@ class Postgresql:
|
||||
self.superuser = config['superuser']
|
||||
self.admin = config['admin']
|
||||
self.callback = config.get('callbacks', {})
|
||||
self.use_slots = config.get('use_slots', True)
|
||||
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'))
|
||||
@@ -162,7 +161,7 @@ class Postgresql:
|
||||
return ret
|
||||
|
||||
def is_running(self):
|
||||
return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0
|
||||
return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null 2>&1', 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 """
|
||||
@@ -268,8 +267,8 @@ class Postgresql:
|
||||
member_conn.autocommit = True
|
||||
member_cursor = member_conn.cursor()
|
||||
member_cursor.execute(
|
||||
"SELECT pg_is_in_recovery(), %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)",
|
||||
(self.xlog_position(), ))
|
||||
"SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')",
|
||||
(self.xlog_position(),))
|
||||
row = member_cursor.fetchone()
|
||||
member_cursor.close()
|
||||
member_conn.close()
|
||||
@@ -317,10 +316,9 @@ class Postgresql:
|
||||
recovery_target_timeline = 'latest'
|
||||
""")
|
||||
if leader and leader.conn_url:
|
||||
f.write("""
|
||||
primary_slot_name = '{}'
|
||||
primary_conninfo = '{}'
|
||||
""".format(self.name, self.primary_conninfo(leader.conn_url)))
|
||||
f.write("""primary_conninfo = '{}'\n""".format(self.primary_conninfo(leader.conn_url)))
|
||||
if self.use_slots:
|
||||
f.write("""primary_slot_name = '{}'\n""".format(self.name))
|
||||
for name, value in self.config.get('recovery_conf', {}).items():
|
||||
f.write("{} = '{}'\n".format(name, value))
|
||||
|
||||
@@ -366,32 +364,37 @@ primary_conninfo = '{}'
|
||||
self.query('CREATE ROLE "{0}" WITH LOGIN SUPERUSER PASSWORD %s'.format(
|
||||
self.superuser['username']), self.superuser['password'])
|
||||
else:
|
||||
self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(os.environ['USER']), self.superuser['password'])
|
||||
rolsuper = self.query("""SELECT rolname FROM pg_authid WHERE rolsuper = 't'""").fetchone()[0]
|
||||
self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(rolsuper), self.superuser['password'])
|
||||
if self.admin:
|
||||
self.query('CREATE ROLE "{0}" WITH LOGIN CREATEDB CREATEROLE PASSWORD %s'.format(
|
||||
self.admin['username']), self.admin['password'])
|
||||
|
||||
def xlog_position(self):
|
||||
return self.query("""SELECT CASE WHEN pg_is_in_recovery()
|
||||
THEN pg_last_xlog_replay_location() - '0/0000000'::pg_lsn
|
||||
ELSE pg_current_xlog_location() - '0/00000'::pg_lsn END""").fetchone()[0]
|
||||
return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
|
||||
THEN pg_last_xlog_replay_location()
|
||||
ELSE pg_current_xlog_location()
|
||||
END, '0/0')""").fetchone()[0]
|
||||
|
||||
def load_replication_slots(self):
|
||||
cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
|
||||
self.members = [r[0] for r in cursor]
|
||||
if self.use_slots:
|
||||
cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
|
||||
self.members = [r[0] for r in cursor]
|
||||
|
||||
def sync_replication_slots(self, members):
|
||||
# drop unused slots
|
||||
for slot in set(self.members) - set(members):
|
||||
self.query("""SELECT pg_drop_replication_slot(%s)
|
||||
WHERE EXISTS(SELECT 1 FROM pg_replication_slots
|
||||
WHERE slot_name = %s)""", slot, slot)
|
||||
if self.use_slots:
|
||||
# drop unused slots
|
||||
for slot in set(self.members) - set(members):
|
||||
self.query("""SELECT pg_drop_replication_slot(%s)
|
||||
WHERE EXISTS(SELECT 1 FROM pg_replication_slots
|
||||
WHERE slot_name = %s)""", slot, slot)
|
||||
|
||||
# create new slots
|
||||
for slot in set(members) - set(self.members):
|
||||
self.query("""SELECT pg_create_physical_replication_slot(%s)
|
||||
WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
|
||||
WHERE slot_name = %s)""", slot, slot)
|
||||
|
||||
# create new slots
|
||||
for slot in set(members) - set(self.members):
|
||||
self.query("""SELECT pg_create_physical_replication_slot(%s)
|
||||
WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
|
||||
WHERE slot_name = %s)""", slot, slot)
|
||||
self.members = members
|
||||
|
||||
def create_replication_slots(self, cluster):
|
||||
@@ -402,3 +405,34 @@ primary_conninfo = '{}'
|
||||
|
||||
def last_operation(self):
|
||||
return str(self.xlog_position())
|
||||
|
||||
def bootstrap(self, current_leader=None):
|
||||
"""
|
||||
Initially bootstrap PostgreSQL, either by creating a data
|
||||
directory with initdb, or by initalizing a replica from an
|
||||
exiting leader. Failure in the first case always leads to
|
||||
exception, since there is no point in continuing if initdb failed.
|
||||
In the second case, however, a False is returned on failure, since
|
||||
it is normal for the replica to retry a failed attempt to initialize
|
||||
from the master.
|
||||
"""
|
||||
ret = False
|
||||
if not current_leader:
|
||||
ret = self.initialize() and self.start()
|
||||
if ret:
|
||||
self.create_replication_user()
|
||||
self.create_connection_users()
|
||||
else:
|
||||
raise PostgresException("Could not bootstrap master PostgreSQL")
|
||||
else:
|
||||
if self.sync_from_leader(current_leader):
|
||||
self.write_recovery_conf(current_leader)
|
||||
ret = self.start()
|
||||
return ret
|
||||
|
||||
def move_data_directory(self):
|
||||
if os.path.isdir(self.data_dir) and not self.is_running():
|
||||
try:
|
||||
os.rename(self.data_dir, '{0}_{1}'.format(self.data_dir, time.strftime('%Y-%m-%d-%H-%M-%S')))
|
||||
except:
|
||||
logger.exception("Could not rename data directory {0}".format(self.data_dir))
|
||||
@@ -0,0 +1,162 @@
|
||||
import datetime
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
|
||||
from patroni.exceptions import DCSError
|
||||
|
||||
interrupted_sleep = False
|
||||
reap_children = False
|
||||
|
||||
_DATE_TIME_RE = re.compile(r'''^
|
||||
(?P<year>\d{4})\-(?P<month>\d{2})\-(?P<day>\d{2}) # date
|
||||
T
|
||||
(?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})\.(?P<microsecond>\d{6}) # time
|
||||
\d*Z$''', re.X)
|
||||
|
||||
|
||||
def parse_datetime(time_str):
|
||||
"""
|
||||
>>> parse_datetime('2015-06-10T12:56:30.552539016Z')
|
||||
datetime.datetime(2015, 6, 10, 12, 56, 30, 552539)
|
||||
>>> parse_datetime('2015-06-10 12:56:30.552539016Z')
|
||||
"""
|
||||
m = _DATE_TIME_RE.match(time_str)
|
||||
if not m:
|
||||
return None
|
||||
p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' '))
|
||||
return datetime.datetime(**p)
|
||||
|
||||
|
||||
def calculate_ttl(expiration):
|
||||
"""
|
||||
>>> calculate_ttl(None)
|
||||
>>> calculate_ttl('2015-06-10 12:56:30.552539016Z')
|
||||
"""
|
||||
if not expiration:
|
||||
return None
|
||||
expiration = parse_datetime(expiration)
|
||||
if not expiration:
|
||||
return None
|
||||
now = datetime.datetime.utcnow()
|
||||
return int((expiration - now).total_seconds())
|
||||
|
||||
|
||||
def sigterm_handler(signo, stack_frame):
|
||||
sys.exit()
|
||||
|
||||
|
||||
def sigchld_handler(signo, stack_frame):
|
||||
global interrupted_sleep, reap_children
|
||||
reap_children = interrupted_sleep = True
|
||||
|
||||
|
||||
def sleep(interval):
|
||||
global interrupted_sleep
|
||||
current_time = time.time()
|
||||
end_time = current_time + interval
|
||||
while current_time < end_time:
|
||||
interrupted_sleep = False
|
||||
time.sleep(end_time - current_time)
|
||||
if not interrupted_sleep: # we will ignore only sigchld
|
||||
break
|
||||
current_time = time.time()
|
||||
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:
|
||||
try:
|
||||
while True:
|
||||
ret = os.waitpid(-1, os.WNOHANG)
|
||||
if ret == (0, 0):
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
reap_children = False
|
||||
|
||||
|
||||
class RetryFailedError(DCSError):
|
||||
|
||||
"""Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts."""
|
||||
|
||||
|
||||
class Retry:
|
||||
|
||||
"""Helper for retrying a method in the face of retry-able exceptions"""
|
||||
|
||||
def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8, max_delay=3600,
|
||||
sleep_func=time.sleep, deadline=None, retry_exceptions=DCSError):
|
||||
"""Create a :class:`Retry` instance for retrying function calls
|
||||
|
||||
:param max_tries: How many times to retry the command. -1 means infinite tries.
|
||||
:param delay: Initial delay between retry attempts.
|
||||
:param backoff: Backoff multiplier between retry attempts. Defaults to 2 for exponential backoff.
|
||||
:param max_jitter: Additional max jitter period to wait between retry attempts to avoid slamming the server.
|
||||
:param max_delay: Maximum delay in seconds, regardless of other backoff settings. Defaults to one hour.
|
||||
:param retry_exceptions: single exception or tuple"""
|
||||
|
||||
self.max_tries = max_tries
|
||||
self.delay = delay
|
||||
self.backoff = backoff
|
||||
self.max_jitter = int(max_jitter * 100)
|
||||
self.max_delay = float(max_delay)
|
||||
self._attempts = 0
|
||||
self._cur_delay = delay
|
||||
self.deadline = deadline
|
||||
self._cur_stoptime = None
|
||||
self.sleep_func = sleep_func
|
||||
self.retry_exceptions = retry_exceptions
|
||||
|
||||
def reset(self):
|
||||
"""Reset the attempt counter"""
|
||||
self._attempts = 0
|
||||
self._cur_delay = self.delay
|
||||
self._cur_stoptime = None
|
||||
|
||||
def copy(self):
|
||||
"""Return a clone of this retry manager"""
|
||||
return Retry(max_tries=self.max_tries, delay=self.delay, backoff=self.backoff,
|
||||
max_jitter=self.max_jitter / 100.0, max_delay=self.max_delay, sleep_func=self.sleep_func,
|
||||
deadline=self.deadline, retry_exceptions=self.retry_exceptions)
|
||||
|
||||
def __call__(self, func, *args, **kwargs):
|
||||
"""Call a function with arguments until it completes without throwing a `retry_exceptions`
|
||||
|
||||
:param func: Function to call
|
||||
:param args: Positional arguments to call the function with
|
||||
:params kwargs: Keyword arguments to call the function with
|
||||
|
||||
The function will be called until it doesn't throw one of the retryable exceptions"""
|
||||
self.reset()
|
||||
|
||||
while True:
|
||||
try:
|
||||
if self.deadline is not None and self._cur_stoptime is None:
|
||||
self._cur_stoptime = time.time() + self.deadline
|
||||
return func(*args, **kwargs)
|
||||
except self.retry_exceptions:
|
||||
# Note: max_tries == -1 means infinite tries.
|
||||
if self._attempts == self.max_tries:
|
||||
raise RetryFailedError("Too many retry attempts")
|
||||
self._attempts += 1
|
||||
sleeptime = self._cur_delay + (
|
||||
random.randint(0, self.max_jitter) / 100.0)
|
||||
|
||||
if self._cur_stoptime is not None and \
|
||||
time.time() + sleeptime >= self._cur_stoptime:
|
||||
raise RetryFailedError("Exceeded retry deadline")
|
||||
else:
|
||||
self.sleep_func(sleeptime)
|
||||
self._cur_delay = min(self._cur_delay * self.backoff,
|
||||
self.max_delay)
|
||||
@@ -0,0 +1 @@
|
||||
__version__ = '0.2'
|
||||
@@ -3,10 +3,10 @@ import random
|
||||
import requests
|
||||
import time
|
||||
|
||||
from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string
|
||||
from helpers.utils import sleep
|
||||
from kazoo.client import KazooClient, KazooState
|
||||
from kazoo.exceptions import NoNodeError, NodeExistsError
|
||||
from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
|
||||
from patroni.utils import sleep
|
||||
from requests.exceptions import RequestException
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -92,9 +92,8 @@ class ZooKeeper(AbstractDCS):
|
||||
self.client.add_listener(self.session_listener)
|
||||
self.cluster_event = self.client.handler.event_object()
|
||||
|
||||
self.cluster = None
|
||||
self.fetch_cluster = True
|
||||
self.members = []
|
||||
self.leader = None
|
||||
self.last_leader_operation = 0
|
||||
|
||||
self.client.start(None)
|
||||
@@ -107,53 +106,61 @@ class ZooKeeper(AbstractDCS):
|
||||
self.fetch_cluster = True
|
||||
self.cluster_event.set()
|
||||
|
||||
def get_node(self, name, watch=None):
|
||||
def get_node(self, key, watch=None):
|
||||
try:
|
||||
return self.client.get(self.client_path(name), watch)
|
||||
ret = self.client.get(key, watch)
|
||||
return (ret[0].decode('utf-8'), ret[1])
|
||||
except NoNodeError:
|
||||
pass
|
||||
except:
|
||||
logger.exception('get_node')
|
||||
return None
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def member(name, value, znode):
|
||||
conn_url, api_url = parse_connection_string(value)
|
||||
return Member(znode.mzxid, name, conn_url, api_url, None, None)
|
||||
return Member(znode.version, name, conn_url, api_url, None, None)
|
||||
|
||||
def get_children(self, key, watch=None):
|
||||
try:
|
||||
return self.client.get_children(key, watch)
|
||||
except NoNodeError:
|
||||
return []
|
||||
|
||||
def load_members(self):
|
||||
members = []
|
||||
for member in self.client.get_children(self.client_path('/members'), self.cluster_watcher):
|
||||
data = self.get_node('/members/' + member)
|
||||
for member in self.get_children(self.members_path, self.cluster_watcher):
|
||||
data = self.get_node(self.members_path + member)
|
||||
if data is not None:
|
||||
members.append(self.member(member, *data))
|
||||
return members
|
||||
|
||||
def _inner_load_cluster(self):
|
||||
self.cluster_event.clear()
|
||||
leader = self.get_node('/leader', self.cluster_watcher)
|
||||
self.members = self.load_members()
|
||||
nodes = set(self.get_children(self.client_path('')))
|
||||
|
||||
# get initialize flag
|
||||
initialize = self._INITIALIZE in nodes
|
||||
|
||||
# get list of members
|
||||
members = self.load_members() if self._MEMBERS[:-1] in nodes else []
|
||||
|
||||
# get leader
|
||||
leader = self.get_node(self.leader_path, self.cluster_watcher) if self._LEADER in nodes else None
|
||||
if leader:
|
||||
if leader[0] == self._name:
|
||||
client_id = self.client.client_id
|
||||
if client_id is not None and client_id[0] != leader[1].ephemeralOwner:
|
||||
logger.info('I am leader but not owner of the session. Removing leader node')
|
||||
self.client.delete(self.client_path('/leader'))
|
||||
leader = None
|
||||
client_id = self.client.client_id
|
||||
if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
|
||||
logger.info('I am leader but not owner of the session. Removing leader node')
|
||||
self.client.delete(self.leader_path)
|
||||
leader = None
|
||||
|
||||
if leader:
|
||||
for member in self.members:
|
||||
if member.name == leader[0]:
|
||||
leader = member
|
||||
self.fetch_cluster = False
|
||||
break
|
||||
if not isinstance(leader, Member):
|
||||
leader = Member(-1, leader, None, None, None, None)
|
||||
self.leader = leader
|
||||
if self.fetch_cluster:
|
||||
last_leader_operation = self.get_node('/optime/leader')
|
||||
if last_leader_operation:
|
||||
self.last_leader_operation = int(last_leader_operation[0])
|
||||
member = Member(-1, leader[0], None, None, None, None)
|
||||
member = ([m for m in members if m.name == leader[0]] or [member])[0]
|
||||
leader = Leader(leader[1].version, None, None, member)
|
||||
self.fetch_cluster = member.index == -1
|
||||
|
||||
# get last leader operation
|
||||
self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None
|
||||
self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0])
|
||||
self.cluster = Cluster(initialize, leader, self.last_leader_operation, members)
|
||||
|
||||
def get_cluster(self):
|
||||
if self.exhibitor and self.exhibitor.poll():
|
||||
@@ -163,31 +170,32 @@ class ZooKeeper(AbstractDCS):
|
||||
try:
|
||||
self.client.retry(self._inner_load_cluster)
|
||||
except:
|
||||
self.cluster = None
|
||||
logger.exception('get_cluster')
|
||||
self.session_listener(KazooState.LOST)
|
||||
raise ZooKeeperError('ZooKeeper in not responding properly')
|
||||
return Cluster(True, self.leader, self.last_leader_operation, self.members)
|
||||
return self.cluster
|
||||
|
||||
def _create(self, path, value, **kwargs):
|
||||
try:
|
||||
self.client.retry(self.client.create, self.client_path(path), value, **kwargs)
|
||||
self.client.retry(self.client.create, path, value.encode('utf-8'), **kwargs)
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
def attempt_to_acquire_leader(self):
|
||||
ret = self._create('/leader', self._name, makepath=True, ephemeral=True)
|
||||
ret = self._create(self.leader_path, self._name, makepath=True, ephemeral=True)
|
||||
ret or logger.info('Could not take out TTL lock')
|
||||
return ret
|
||||
|
||||
def race(self, path):
|
||||
return self._create(path, self._name, makepath=True)
|
||||
def initialize(self):
|
||||
return self._create(self.initialize_path, self._name, makepath=True)
|
||||
|
||||
def touch_member(self, connection_string, ttl=None):
|
||||
for m in self.members:
|
||||
if m.name == self._name:
|
||||
return True
|
||||
path = self.client_path('/members/' + self._name)
|
||||
if self.cluster and any(m.name == self._name for m in self.cluster.members):
|
||||
return True
|
||||
path = self.member_path
|
||||
connection_string = connection_string.encode('utf-8')
|
||||
try:
|
||||
self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
|
||||
return True
|
||||
@@ -204,10 +212,10 @@ class ZooKeeper(AbstractDCS):
|
||||
return self.attempt_to_acquire_leader()
|
||||
|
||||
def update_leader(self, state_handler):
|
||||
last_operation = state_handler.last_operation()
|
||||
last_operation = state_handler.last_operation().encode('utf-8')
|
||||
if last_operation != self.last_leader_operation:
|
||||
self.last_leader_operation = last_operation
|
||||
path = self.client_path('/optime/leader')
|
||||
path = self.leader_optime_path
|
||||
try:
|
||||
self.client.retry(self.client.set, path, last_operation)
|
||||
except NoNodeError:
|
||||
@@ -220,10 +228,21 @@ class ZooKeeper(AbstractDCS):
|
||||
return True
|
||||
|
||||
def delete_leader(self):
|
||||
if isinstance(self.leader, Member) and self.leader.name == self._name:
|
||||
self.client.delete(self.client_path('/leader'))
|
||||
if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name:
|
||||
self.client.delete(self.leader_path, version=self.cluster.leader.index)
|
||||
|
||||
def sleep(self, timeout):
|
||||
def _cancel_initialization(self):
|
||||
node = self.get_node(self.initialize_path)
|
||||
if node and node[0] == self._name:
|
||||
self.client.delete(self.initialize_path, version=node[1].version)
|
||||
|
||||
def cancel_initialization(self):
|
||||
try:
|
||||
self.client.retry(self._cancel_initialization)
|
||||
except:
|
||||
logger.exception("Unable to delete initialize key")
|
||||
|
||||
def watch(self, timeout):
|
||||
self.cluster_event.wait(timeout)
|
||||
if self.cluster_event.isSet():
|
||||
self.fetch_cluster = True
|
||||
+2
-1
@@ -30,6 +30,7 @@ postgresql:
|
||||
connect_address: 127.0.0.1:5432
|
||||
data_dir: data/postgresql0
|
||||
maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
|
||||
use_slots: True
|
||||
pg_hba:
|
||||
- host all all 0.0.0.0/0 md5
|
||||
- hostssl all all 0.0.0.0/0 md5
|
||||
@@ -46,7 +47,7 @@ postgresql:
|
||||
env_dir: /home/postgres/etc/wal-e.d/env
|
||||
threshold_megabytes: 10240
|
||||
threshold_backup_size_percentage: 30
|
||||
restore: scripts/restore.py
|
||||
restore: patroni/scripts/restore.py
|
||||
#recovery_conf:
|
||||
#restore_command: cp ../wal_archive/%f %p
|
||||
parameters:
|
||||
|
||||
+2
-1
@@ -30,6 +30,7 @@ postgresql:
|
||||
connect_address: 127.0.0.1:5433
|
||||
data_dir: data/postgresql1
|
||||
maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
|
||||
use_slots: True
|
||||
pg_hba:
|
||||
- host all all 0.0.0.0/0 md5
|
||||
- hostssl all all 0.0.0.0/0 md5
|
||||
@@ -48,7 +49,7 @@ postgresql:
|
||||
env_dir: /home/postgres/etc/wal-e.d/env
|
||||
threshold_megabytes: 10240
|
||||
threshold_backup_size_percentage: 30
|
||||
restore: scripts/restore.py
|
||||
restore: patroni/scripts/restore.py
|
||||
parameters:
|
||||
archive_mode: "on"
|
||||
wal_level: hot_standby
|
||||
|
||||
Executable
+31
@@ -0,0 +1,31 @@
|
||||
#!/bin/sh
|
||||
|
||||
if [ $# -ne 1 ]; then
|
||||
>&2 echo "usage: $0 <version>"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
readonly VERSIONFILE="patroni/version.py"
|
||||
|
||||
## Bail out on any non-zero exitcode from the called processes
|
||||
set -xe
|
||||
|
||||
python3 --version
|
||||
git --version
|
||||
|
||||
version=$1
|
||||
|
||||
sed -i "s/__version__ = .*/__version__ = '${version}'/" "${VERSIONFILE}"
|
||||
python3 setup.py clean
|
||||
python3 setup.py test
|
||||
python3 setup.py flake8
|
||||
|
||||
git add "${VERSIONFILE}"
|
||||
|
||||
git commit -m "Bumped version to $version"
|
||||
git push
|
||||
|
||||
python3 setup.py sdist bdist_wheel upload
|
||||
|
||||
git tag v${version}
|
||||
git push --tags
|
||||
@@ -19,13 +19,22 @@ if sys.version_info < (2, 7, 0):
|
||||
__location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
|
||||
|
||||
|
||||
def read_version(package):
|
||||
data = {}
|
||||
with open(os.path.join(package, 'version.py'), 'r') as fd:
|
||||
exec(fd.read(), data)
|
||||
return data['__version__']
|
||||
|
||||
NAME = 'patroni'
|
||||
MAIN_PACKAGE = 'patroni.py'
|
||||
HELPERS = 'helpers'
|
||||
MAIN_PACKAGE = NAME
|
||||
SCRIPTS = 'scripts'
|
||||
VERSION = '0.1'
|
||||
DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
|
||||
VERSION = read_version(MAIN_PACKAGE)
|
||||
DESCRIPTION = 'PostgreSQL High-Available orchestrator and CLI'
|
||||
LICENSE = 'The MIT License'
|
||||
URL = 'https://github.com/zalando/patroni'
|
||||
AUTHOR = 'Alexander Kukushkin, Alexey Klyukin, Feike Steenbergen'
|
||||
AUTHOR_EMAIL = '[email protected], [email protected], [email protected]'
|
||||
KEYWORDS = 'etcd governor patroni postgresql postgres ha zookeeper streaming replication'
|
||||
|
||||
COVERAGE_XML = True
|
||||
COVERAGE_HTML = False
|
||||
@@ -38,7 +47,7 @@ CLASSIFIERS = [
|
||||
'Environment :: Console',
|
||||
'Intended Audience :: Developers',
|
||||
'Intended Audience :: System Administrators',
|
||||
'License :: OSI Approved :: The MIT License',
|
||||
'License :: OSI Approved :: MIT License',
|
||||
'Operating System :: POSIX :: Linux',
|
||||
'Programming Language :: Python',
|
||||
'Programming Language :: Python :: 2.7',
|
||||
@@ -47,6 +56,8 @@ CLASSIFIERS = [
|
||||
'Programming Language :: Python :: Implementation :: CPython',
|
||||
]
|
||||
|
||||
CONSOLE_SCRIPTS = ['patroni = patroni:main']
|
||||
|
||||
|
||||
class PyTest(TestCommand):
|
||||
|
||||
@@ -62,8 +73,7 @@ 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', SCRIPTS, '--cov-report',
|
||||
'term-missing']
|
||||
self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
|
||||
if self.cov_xml:
|
||||
self.cov.extend(['--cov-report', 'xml'])
|
||||
if self.cov_html:
|
||||
@@ -82,7 +92,7 @@ class PyTest(TestCommand):
|
||||
params['plugins'] = ['cov']
|
||||
if self.junitxml:
|
||||
params['args'] += self.junitxml
|
||||
params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s']
|
||||
params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-s', '-vv']
|
||||
errno = pytest.main(**params)
|
||||
sys.exit(errno)
|
||||
|
||||
@@ -118,10 +128,13 @@ def setup_package():
|
||||
setup(
|
||||
name=NAME,
|
||||
version=version,
|
||||
url=URL,
|
||||
author=AUTHOR,
|
||||
author_email=AUTHOR_EMAIL,
|
||||
description=DESCRIPTION,
|
||||
license=LICENSE,
|
||||
keywords='etcd governor patroni postgresql postgres ha zookeeper',
|
||||
long_description=read('README.md'),
|
||||
keywords=KEYWORDS,
|
||||
long_description=read('README.rst'),
|
||||
classifiers=CLASSIFIERS,
|
||||
test_suite='tests',
|
||||
packages=setuptools.find_packages(exclude=['tests', 'tests.*']),
|
||||
@@ -131,6 +144,7 @@ def setup_package():
|
||||
cmdclass=cmdclass,
|
||||
tests_require=['pytest-cov', 'pytest'],
|
||||
command_options=command_options,
|
||||
entry_points={'console_scripts': CONSOLE_SCRIPTS},
|
||||
)
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
import psycopg2
|
||||
import unittest
|
||||
|
||||
from helpers.api import RestApiHandler, RestApiServer
|
||||
from patroni.api import RestApiHandler, RestApiServer
|
||||
from six import BytesIO as IO
|
||||
from test_postgresql import psycopg2_connect
|
||||
|
||||
|
||||
+1
-1
@@ -2,7 +2,7 @@ import unittest
|
||||
import requests
|
||||
import boto.ec2
|
||||
from collections import namedtuple
|
||||
from scripts.aws import AWSConnection
|
||||
from patroni.scripts.aws import AWSConnection
|
||||
from requests.exceptions import RequestException
|
||||
|
||||
|
||||
|
||||
+62
-10
@@ -3,14 +3,15 @@ import dns.resolver
|
||||
import etcd
|
||||
import json
|
||||
import requests
|
||||
import urllib3
|
||||
import socket
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from dns.exception import DNSException
|
||||
from helpers.dcs import Cluster, DCSError, Member
|
||||
from helpers.etcd import Client, Etcd
|
||||
from mock import Mock, patch
|
||||
from patroni.dcs import Cluster, DCSError, Leader, Member
|
||||
from patroni.etcd import Client, Etcd
|
||||
|
||||
|
||||
class MockResponse:
|
||||
@@ -25,6 +26,10 @@ class MockResponse:
|
||||
|
||||
@property
|
||||
def data(self):
|
||||
if self.content == 'TimeoutError':
|
||||
raise urllib3.exceptions.TimeoutError
|
||||
if self.content == 'Exception':
|
||||
raise Exception
|
||||
return self.content
|
||||
|
||||
@property
|
||||
@@ -39,7 +44,7 @@ class MockPostgresql:
|
||||
name = ''
|
||||
|
||||
def last_operation(self):
|
||||
return 0
|
||||
return '0'
|
||||
|
||||
|
||||
def requests_get(url, **kwargs):
|
||||
@@ -61,7 +66,22 @@ def requests_get(url, **kwargs):
|
||||
return response
|
||||
|
||||
|
||||
def etcd_watch(key, index=None, timeout=None, recursive=None):
|
||||
if timeout == 1:
|
||||
raise urllib3.exceptions.TimeoutError
|
||||
elif timeout == 5:
|
||||
return etcd.EtcdResult('delete', {})
|
||||
elif timeout == 10:
|
||||
raise etcd.EtcdException
|
||||
elif index == 20729:
|
||||
return etcd.EtcdResult('set', {'value': 'postgresql1', 'modifiedIndex': index + 1})
|
||||
elif index == 20731:
|
||||
return etcd.EtcdResult('set', {'value': 'postgresql2', 'modifiedIndex': index + 1})
|
||||
|
||||
|
||||
def etcd_write(key, value, **kwargs):
|
||||
if key == '/service/exists/leader':
|
||||
raise etcd.EtcdAlreadyExist
|
||||
if key == '/service/test/leader':
|
||||
if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True):
|
||||
return True
|
||||
@@ -73,9 +93,9 @@ def etcd_delete(key, **kwargs):
|
||||
|
||||
|
||||
def etcd_read(key, **kwargs):
|
||||
if key == '/service/noleader':
|
||||
if key == '/service/noleader/':
|
||||
raise DCSError('noleader')
|
||||
elif key == '/service/nocluster':
|
||||
elif key == '/service/nocluster/':
|
||||
raise etcd.EtcdKeyNotFound
|
||||
|
||||
response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
|
||||
@@ -107,8 +127,12 @@ def time_sleep(_):
|
||||
pass
|
||||
|
||||
|
||||
class SleepException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def time_sleep_exception(_):
|
||||
raise Exception()
|
||||
raise SleepException()
|
||||
|
||||
|
||||
class MockSRV:
|
||||
@@ -172,6 +196,15 @@ class TestClient(unittest.TestCase):
|
||||
self.assertEquals(self.client.get_srv_record('blabla'), [])
|
||||
self.assertEquals(self.client.get_srv_record('exception'), [])
|
||||
|
||||
def test__result_from_response(self):
|
||||
response = MockResponse()
|
||||
response.content = 'TimeoutError'
|
||||
self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response)
|
||||
response.content = 'Exception'
|
||||
self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
|
||||
response.content = b'{}'
|
||||
self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
|
||||
|
||||
def test__get_machines_cache_from_srv(self):
|
||||
self.client.get_srv_record = lambda e: [('localhost', 2380)]
|
||||
self.client._get_machines_cache_from_srv('blabla')
|
||||
@@ -204,7 +237,7 @@ class TestEtcd(unittest.TestCase):
|
||||
time.sleep = time_sleep_exception
|
||||
with patch.object(etcd.Client, 'machines') as mock_machines:
|
||||
mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
|
||||
self.assertRaises(Exception, self.etcd.get_etcd_client, {'discovery_srv': 'test'})
|
||||
self.assertRaises(SleepException, self.etcd.get_etcd_client, {'discovery_srv': 'test'})
|
||||
|
||||
def test_get_cluster(self):
|
||||
self.assertIsInstance(self.etcd.get_cluster(), Cluster)
|
||||
@@ -214,7 +247,7 @@ class TestEtcd(unittest.TestCase):
|
||||
self.assertIsNone(cluster.leader)
|
||||
|
||||
def test_current_leader(self):
|
||||
self.assertIsInstance(self.etcd.current_leader(), Member)
|
||||
self.assertIsInstance(self.etcd.current_leader(), Leader)
|
||||
self.etcd._base_path = '/service/noleader'
|
||||
self.assertIsNone(self.etcd.current_leader())
|
||||
|
||||
@@ -224,12 +257,31 @@ class TestEtcd(unittest.TestCase):
|
||||
def test_take_leader(self):
|
||||
self.assertFalse(self.etcd.take_leader())
|
||||
|
||||
def testattempt_to_acquire_leader(self):
|
||||
self.etcd._base_path = '/service/exists'
|
||||
self.assertFalse(self.etcd.attempt_to_acquire_leader())
|
||||
self.etcd._base_path = '/service/failed'
|
||||
self.assertFalse(self.etcd.attempt_to_acquire_leader())
|
||||
|
||||
def test_update_leader(self):
|
||||
self.assertTrue(self.etcd.update_leader(MockPostgresql()))
|
||||
|
||||
def test_race(self):
|
||||
self.assertFalse(self.etcd.race(''))
|
||||
def test_initialize(self):
|
||||
self.assertFalse(self.etcd.initialize())
|
||||
|
||||
def test_cancel_initializion(self):
|
||||
self.etcd.client.delete = etcd_delete
|
||||
self.assertFalse(self.etcd.cancel_initialization())
|
||||
|
||||
def test_delete_leader(self):
|
||||
self.etcd.client.delete = etcd_delete
|
||||
self.assertFalse(self.etcd.delete_leader())
|
||||
|
||||
def test_watch(self):
|
||||
self.etcd.client.watch = etcd_watch
|
||||
self.etcd.watch(100)
|
||||
self.etcd.get_cluster()
|
||||
self.etcd.watch(1)
|
||||
self.etcd.watch(5)
|
||||
self.etcd.watch(10)
|
||||
self.etcd.watch(100)
|
||||
|
||||
+3
-3
@@ -1,9 +1,9 @@
|
||||
import unittest
|
||||
|
||||
from helpers.dcs import Cluster, DCSError
|
||||
from helpers.etcd import Client, Etcd
|
||||
from helpers.ha import Ha
|
||||
from mock import Mock, patch
|
||||
from patroni.dcs import Cluster, DCSError
|
||||
from patroni.etcd import Client, Etcd
|
||||
from patroni.ha import Ha
|
||||
from test_etcd import etcd_read, etcd_write
|
||||
|
||||
|
||||
|
||||
+71
-15
@@ -1,5 +1,5 @@
|
||||
import datetime
|
||||
import helpers.zookeeper
|
||||
import patroni.zookeeper
|
||||
import psycopg2
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -7,12 +7,13 @@ import time
|
||||
import unittest
|
||||
import yaml
|
||||
|
||||
from helpers.api import RestApiServer
|
||||
from helpers.dcs import Cluster, Member
|
||||
from helpers.etcd import Etcd
|
||||
from helpers.zookeeper import ZooKeeper
|
||||
from mock import Mock, patch
|
||||
from patroni.api import RestApiServer
|
||||
from patroni.dcs import Cluster, Member, Leader
|
||||
from patroni.etcd import Etcd
|
||||
from patroni.exceptions import PostgresException
|
||||
from patroni import Patroni, main
|
||||
from patroni.zookeeper import ZooKeeper
|
||||
from six.moves import BaseHTTPServer
|
||||
from test_etcd import Client, etcd_read, etcd_write
|
||||
from test_ha import true, false
|
||||
@@ -24,8 +25,12 @@ def nop(*args, **kwargs):
|
||||
pass
|
||||
|
||||
|
||||
class SleepException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def time_sleep(*args):
|
||||
raise Exception()
|
||||
raise SleepException()
|
||||
|
||||
|
||||
class Mock_BaseServer__is_shut_down:
|
||||
@@ -37,6 +42,30 @@ class Mock_BaseServer__is_shut_down:
|
||||
pass
|
||||
|
||||
|
||||
def get_cluster(initialize, leader):
|
||||
return Cluster(initialize, leader, None, None)
|
||||
|
||||
|
||||
def get_cluster_not_initialized_without_leader():
|
||||
return get_cluster(None, None)
|
||||
|
||||
|
||||
def get_cluster_initialized_without_leader():
|
||||
return get_cluster(True, None)
|
||||
|
||||
|
||||
def get_cluster_not_initialized_with_leader():
|
||||
return get_cluster(False, Leader(0, 0, 0,
|
||||
Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
|
||||
None, None, 28)))
|
||||
|
||||
|
||||
def get_cluster_initialized_with_leader():
|
||||
return get_cluster(True, Leader(0, 0, 0,
|
||||
Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
|
||||
None, None, 28)))
|
||||
|
||||
|
||||
class TestPatroni(unittest.TestCase):
|
||||
|
||||
def __init__(self, method_name='runTest'):
|
||||
@@ -46,6 +75,7 @@ class TestPatroni(unittest.TestCase):
|
||||
|
||||
def set_up(self):
|
||||
self.touched = False
|
||||
self.init_cancelled = False
|
||||
subprocess.call = subprocess_call
|
||||
psycopg2.connect = psycopg2_connect
|
||||
self.time_sleep = time.sleep
|
||||
@@ -70,7 +100,7 @@ class TestPatroni(unittest.TestCase):
|
||||
Postgresql.write_recovery_conf = self.write_recovery_conf
|
||||
|
||||
def test_get_dcs(self):
|
||||
helpers.zookeeper.KazooClient = MockKazooClient
|
||||
patroni.zookeeper.KazooClient = MockKazooClient
|
||||
self.assertIsInstance(self.p.get_dcs('', {'zookeeper': {'scope': '', 'hosts': ''}}), ZooKeeper)
|
||||
self.assertRaises(Exception, self.p.get_dcs, '', {})
|
||||
|
||||
@@ -90,7 +120,7 @@ class TestPatroni(unittest.TestCase):
|
||||
|
||||
Etcd.delete_leader = nop
|
||||
|
||||
self.assertRaises(Exception, main)
|
||||
self.assertRaises(SleepException, main)
|
||||
|
||||
Patroni.run = run
|
||||
Patroni.touch_member = touch_member
|
||||
@@ -100,10 +130,11 @@ class TestPatroni(unittest.TestCase):
|
||||
self.p.touch_member = self.touch_member
|
||||
self.p.ha.state_handler.sync_replication_slots = time_sleep
|
||||
self.p.ha.dcs.client.read = etcd_read
|
||||
self.assertRaises(Exception, self.p.run)
|
||||
self.p.ha.dcs.watch = time_sleep
|
||||
self.assertRaises(SleepException, self.p.run)
|
||||
self.p.ha.state_handler.is_leader = lambda: False
|
||||
self.p.api.start = nop
|
||||
self.assertRaises(Exception, self.p.run)
|
||||
self.assertRaises(SleepException, self.p.run)
|
||||
|
||||
def touch_member(self, ttl=None):
|
||||
if not self.touched:
|
||||
@@ -121,24 +152,49 @@ class TestPatroni(unittest.TestCase):
|
||||
self.p.touch_member()
|
||||
|
||||
def test_patroni_initialize(self):
|
||||
self.p.postgresql.should_use_s3_to_create_replica = false
|
||||
self.p.ha.dcs.client.write = etcd_write
|
||||
self.p.ha.dcs.client.read = etcd_read
|
||||
self.p.touch_member = self.touch_member
|
||||
self.p.postgresql.data_directory_empty = true
|
||||
self.p.ha.dcs.race = true
|
||||
self.p.ha.dcs.initialize = true
|
||||
self.p.postgresql.initialize = true
|
||||
self.p.postgresql.start = true
|
||||
self.p.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader
|
||||
self.p.initialize()
|
||||
|
||||
self.p.ha.dcs.race = false
|
||||
self.p.ha.dcs.initialize = false
|
||||
self.p.ha.dcs.get_cluster = get_cluster_initialized_with_leader
|
||||
time.sleep = time_sleep
|
||||
self.p.ha.dcs.client.read = etcd_read
|
||||
self.p.initialize()
|
||||
|
||||
self.p.ha.dcs.current_leader = nop
|
||||
self.assertRaises(Exception, self.p.initialize)
|
||||
self.p.ha.dcs.get_cluster = get_cluster_initialized_without_leader
|
||||
self.assertRaises(SleepException, self.p.initialize)
|
||||
|
||||
self.p.postgresql.data_directory_empty = false
|
||||
self.p.initialize()
|
||||
|
||||
self.p.ha.dcs.get_cluster = get_cluster_not_initialized_with_leader
|
||||
self.p.postgresql.data_directory_empty = true
|
||||
self.p.initialize()
|
||||
|
||||
def test_schedule_next_run(self):
|
||||
self.p.next_run = time.time() - self.p.nap_time - 1
|
||||
self.p.schedule_next_run()
|
||||
|
||||
def cancel_initialization(self):
|
||||
self.init_cancelled = True
|
||||
|
||||
def test_cleanup_on_initialization(self):
|
||||
self.p.ha.dcs.client.write = etcd_write
|
||||
self.p.ha.dcs.client.read = etcd_read
|
||||
self.p.ha.dcs.get_cluster = get_cluster_not_initialized_without_leader
|
||||
self.p.touch_member = self.touch_member
|
||||
self.p.postgresql.data_directory_empty = true
|
||||
self.p.ha.dcs.initialize = true
|
||||
self.p.postgresql.initialize = true
|
||||
self.p.postgresql.start = false
|
||||
|
||||
self.p.ha.dcs.cancel_initialization = self.cancel_initialization
|
||||
self.assertRaises(PostgresException, self.p.initialize)
|
||||
self.assertTrue(self.init_cancelled)
|
||||
|
||||
+17
-10
@@ -4,8 +4,9 @@ import shutil
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
from helpers.dcs import Cluster, Member
|
||||
from helpers.postgresql import Postgresql
|
||||
from patroni.dcs import Cluster, Leader, Member
|
||||
from patroni.postgresql import Postgresql
|
||||
from test_ha import true, false
|
||||
|
||||
|
||||
def nop(*args, **kwargs):
|
||||
@@ -24,7 +25,6 @@ class MockCursor:
|
||||
|
||||
def __init__(self):
|
||||
self.closed = False
|
||||
self.current = 0
|
||||
self.results = []
|
||||
|
||||
def execute(self, sql, *params):
|
||||
@@ -43,7 +43,7 @@ class MockCursor:
|
||||
self.results = [(True, -1)]
|
||||
else:
|
||||
self.results = [(False, 0)]
|
||||
elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
|
||||
elif sql.startswith('SELECT pg_xlog_location_diff'):
|
||||
self.results = [(0,)]
|
||||
elif sql.startswith('SELECT pg_is_in_recovery()'):
|
||||
self.results = [(False, )]
|
||||
@@ -119,11 +119,12 @@ class TestPostgresql(unittest.TestCase):
|
||||
'on_restart': 'true', 'on_role_change': 'true',
|
||||
'on_reload': 'true'
|
||||
},
|
||||
'restore': '/usr/bin/true'})
|
||||
'restore': 'true'})
|
||||
psycopg2.connect = psycopg2_connect
|
||||
if not os.path.exists(self.p.data_dir):
|
||||
os.makedirs(self.p.data_dir)
|
||||
self.leader = Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres', None, None, 28)
|
||||
self.leadermem = Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres', None, None, 28)
|
||||
self.leader = Leader(-1, None, 28, self.leadermem)
|
||||
self.other = Member(0, 'test1', 'postgres://replicator:[email protected]:5433/postgres', None, None, 28)
|
||||
self.me = Member(0, 'test0', 'postgres://replicator:[email protected]:5434/postgres', None, None, 28)
|
||||
|
||||
@@ -156,7 +157,7 @@ class TestPostgresql(unittest.TestCase):
|
||||
self.p.follow_the_leader(None)
|
||||
self.p.demote(self.leader)
|
||||
self.p.follow_the_leader(self.leader)
|
||||
self.p.follow_the_leader(self.other)
|
||||
self.p.follow_the_leader(Leader(-1, None, 28, self.other))
|
||||
|
||||
def test_create_connection_users(self):
|
||||
cfg = self.p.config
|
||||
@@ -166,7 +167,7 @@ class TestPostgresql(unittest.TestCase):
|
||||
|
||||
def test_create_replication_slots(self):
|
||||
self.p.start()
|
||||
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader])
|
||||
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem])
|
||||
self.p.create_replication_slots(cluster)
|
||||
|
||||
def test_query(self):
|
||||
@@ -180,7 +181,7 @@ class TestPostgresql(unittest.TestCase):
|
||||
self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
|
||||
|
||||
def test_is_healthiest_node(self):
|
||||
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader])
|
||||
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem])
|
||||
self.assertTrue(self.p.is_healthiest_node(cluster))
|
||||
self.p.is_leader = false
|
||||
self.assertFalse(self.p.is_healthiest_node(cluster))
|
||||
@@ -188,7 +189,7 @@ class TestPostgresql(unittest.TestCase):
|
||||
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.p.config['maximum_lag_on_failover'] = -3
|
||||
self.assertFalse(self.p.is_healthiest_node(cluster))
|
||||
|
||||
def test_is_leader(self):
|
||||
@@ -217,3 +218,9 @@ class TestPostgresql(unittest.TestCase):
|
||||
self.p.start()
|
||||
self.p.query = self.mock_query
|
||||
self.assertTrue(self.p.stop())
|
||||
|
||||
def test_move_data_directory(self):
|
||||
self.p.is_running = is_running
|
||||
os.rename = nop
|
||||
os.path.isdir = true
|
||||
self.p.move_data_directory()
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import unittest
|
||||
from mock import MagicMock, patch
|
||||
import os
|
||||
from scripts.restore import Restore, WALERestore
|
||||
from patroni.scripts.restore import Restore, WALERestore
|
||||
|
||||
|
||||
def fake_cursor_fetchone(*args, **kwargs):
|
||||
|
||||
+60
-3
@@ -2,7 +2,8 @@ import os
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from helpers.utils import sigchld_handler, sigterm_handler, sleep
|
||||
from patroni.exceptions import DCSError
|
||||
from patroni.utils import Retry, RetryFailedError, reap_children, sigchld_handler, sigterm_handler, sleep
|
||||
|
||||
|
||||
def nop(*args, **kwargs):
|
||||
@@ -34,11 +35,67 @@ class TestUtils(unittest.TestCase):
|
||||
def test_sigterm_handler(self):
|
||||
self.assertRaises(SystemExit, sigterm_handler, None, None)
|
||||
|
||||
def test_sigchld_handler(self):
|
||||
sigchld_handler(None, None)
|
||||
def test_reap_children(self):
|
||||
reap_children()
|
||||
os.waitpid = os_waitpid
|
||||
sigchld_handler(None, None)
|
||||
reap_children()
|
||||
|
||||
def test_sleep(self):
|
||||
time.sleep = time_sleep
|
||||
sleep(0.01)
|
||||
|
||||
|
||||
class TestRetrySleeper(unittest.TestCase):
|
||||
|
||||
def _pass(self):
|
||||
pass
|
||||
|
||||
def _fail(self, times=1):
|
||||
scope = dict(times=0)
|
||||
|
||||
def inner():
|
||||
if scope['times'] >= times:
|
||||
pass
|
||||
else:
|
||||
scope['times'] += 1
|
||||
raise DCSError('Failed!')
|
||||
return inner
|
||||
|
||||
def _makeOne(self, *args, **kwargs):
|
||||
return Retry(*args, **kwargs)
|
||||
|
||||
def test_reset(self):
|
||||
retry = self._makeOne(delay=0, max_tries=2)
|
||||
retry(self._fail())
|
||||
self.assertEquals(retry._attempts, 1)
|
||||
retry.reset()
|
||||
self.assertEquals(retry._attempts, 0)
|
||||
|
||||
def test_too_many_tries(self):
|
||||
retry = self._makeOne(delay=0)
|
||||
self.assertRaises(RetryFailedError, retry, self._fail(times=999))
|
||||
self.assertEquals(retry._attempts, 1)
|
||||
|
||||
def test_maximum_delay(self):
|
||||
def sleep_func(_time):
|
||||
pass
|
||||
|
||||
retry = self._makeOne(delay=10, max_tries=100, sleep_func=sleep_func)
|
||||
retry(self._fail(times=10))
|
||||
self.assertTrue(retry._cur_delay < 4000, retry._cur_delay)
|
||||
# gevent's sleep function is picky about the type
|
||||
self.assertEquals(type(retry._cur_delay), float)
|
||||
|
||||
def test_deadline(self):
|
||||
def sleep_func(_time):
|
||||
pass
|
||||
|
||||
retry = self._makeOne(deadline=0.0001, sleep_func=sleep_func)
|
||||
self.assertRaises(RetryFailedError, retry, self._fail(times=100))
|
||||
|
||||
def test_copy(self):
|
||||
_sleep = lambda t: None
|
||||
retry = self._makeOne(sleep_func=_sleep)
|
||||
rcopy = retry.copy()
|
||||
self.assertTrue(rcopy.sleep_func is _sleep)
|
||||
|
||||
+53
-21
@@ -1,8 +1,10 @@
|
||||
import helpers.zookeeper
|
||||
import patroni.zookeeper
|
||||
import requests
|
||||
import six
|
||||
import unittest
|
||||
|
||||
from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
|
||||
from patroni.dcs import Leader
|
||||
from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
|
||||
from kazoo.client import KazooState
|
||||
from kazoo.exceptions import NoNodeError, NodeExistsError
|
||||
from kazoo.protocol.states import ZnodeStat
|
||||
@@ -30,6 +32,10 @@ class MockEventHandler:
|
||||
return MockEvent()
|
||||
|
||||
|
||||
class SleepException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class MockKazooClient:
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
@@ -51,50 +57,70 @@ class MockKazooClient:
|
||||
func(*args, **kwargs)
|
||||
|
||||
def get(self, path, watch=None):
|
||||
if path == '/service/test/no_node':
|
||||
if not isinstance(path, six.string_types):
|
||||
raise TypeError("Invalid type for 'path' (string expected)")
|
||||
if path == '/no_node':
|
||||
raise NoNodeError
|
||||
elif path == '/service/test/other_exception':
|
||||
raise Exception()
|
||||
elif '/members/' in path:
|
||||
return (
|
||||
'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni',
|
||||
b'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni',
|
||||
ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
|
||||
)
|
||||
elif path.endswith('/optime/leader'):
|
||||
return '1'
|
||||
return (b'1', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
|
||||
elif path.endswith('/leader'):
|
||||
if self.leader:
|
||||
return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
|
||||
return ('foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
|
||||
return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
|
||||
return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
|
||||
elif path.endswith('/initialize'):
|
||||
return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
|
||||
|
||||
def get_children(self, path, watch=None, include_data=False):
|
||||
if not isinstance(path, six.string_types):
|
||||
raise TypeError("Invalid type for 'path' (string expected)")
|
||||
if path == '/no_node':
|
||||
raise NoNodeError
|
||||
elif path in ['/service/bla/', '/service/test/']:
|
||||
return ['initialize', 'leader', 'members', 'optime']
|
||||
return ['foo', 'bar', 'buzz']
|
||||
|
||||
def create(self, path, value="", acl=None, ephemeral=False, sequence=False, makepath=False):
|
||||
def create(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False):
|
||||
if not isinstance(path, six.string_types):
|
||||
raise TypeError("Invalid type for 'path' (string expected)")
|
||||
if not isinstance(value, (six.binary_type,)):
|
||||
raise TypeError("Invalid type for 'value' (must be a byte string)")
|
||||
if path.endswith('/initialize') or path == '/service/test/optime/leader':
|
||||
raise Exception
|
||||
elif value == 'retry' or (value == 'exists' and self.exists):
|
||||
elif value == b'retry' or (value == b'exists' and self.exists):
|
||||
raise NodeExistsError
|
||||
|
||||
def set(self, path, value, version=-1):
|
||||
if not isinstance(path, six.string_types):
|
||||
raise TypeError("Invalid type for 'path' (string expected)")
|
||||
if not isinstance(value, (six.binary_type,)):
|
||||
raise TypeError("Invalid type for 'value' (must be a byte string)")
|
||||
if path == '/service/bla/optime/leader':
|
||||
raise Exception
|
||||
raise NoNodeError
|
||||
|
||||
def delete(self, path, version=-1, recursive=False):
|
||||
if not isinstance(path, six.string_types):
|
||||
raise TypeError("Invalid type for 'path' (string expected)")
|
||||
self.exists = False
|
||||
if path == '/service/test/leader':
|
||||
if self.leader:
|
||||
return
|
||||
self.leader = True
|
||||
raise Exception
|
||||
elif path.endswith('/initialize'):
|
||||
raise NoNodeError
|
||||
|
||||
def set_hosts(self, hosts, randomize_hosts=None):
|
||||
pass
|
||||
|
||||
|
||||
def exhibitor_sleep(_):
|
||||
raise Exception
|
||||
raise SleepException
|
||||
|
||||
|
||||
class TestExhibitorEnsembleProvider(unittest.TestCase):
|
||||
@@ -105,10 +131,10 @@ class TestExhibitorEnsembleProvider(unittest.TestCase):
|
||||
|
||||
def set_up(self):
|
||||
requests.get = requests_get
|
||||
helpers.zookeeper.sleep = exhibitor_sleep
|
||||
patroni.zookeeper.sleep = exhibitor_sleep
|
||||
|
||||
def test_init(self):
|
||||
self.assertRaises(Exception, ExhibitorEnsembleProvider, ['localhost'], 8181)
|
||||
self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181)
|
||||
|
||||
|
||||
class TestZooKeeper(unittest.TestCase):
|
||||
@@ -119,7 +145,7 @@ class TestZooKeeper(unittest.TestCase):
|
||||
|
||||
def set_up(self):
|
||||
requests.get = requests_get
|
||||
helpers.zookeeper.KazooClient = MockKazooClient
|
||||
patroni.zookeeper.KazooClient = MockKazooClient
|
||||
self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'})
|
||||
|
||||
def test_session_listener(self):
|
||||
@@ -127,7 +153,9 @@ class TestZooKeeper(unittest.TestCase):
|
||||
|
||||
def test_get_node(self):
|
||||
self.assertIsNone(self.zk.get_node('/no_node'))
|
||||
self.assertIsNone(self.zk.get_node('/other_exception'))
|
||||
|
||||
def test_get_children(self):
|
||||
self.assertListEqual(self.zk.get_children('/no_node'), [])
|
||||
|
||||
def test__inner_load_cluster(self):
|
||||
self.zk._base_path = self.zk._base_path.replace('test', 'bla')
|
||||
@@ -136,12 +164,16 @@ class TestZooKeeper(unittest.TestCase):
|
||||
def test_get_cluster(self):
|
||||
self.assertRaises(ZooKeeperError, self.zk.get_cluster)
|
||||
self.zk.exhibitor.poll = lambda: True
|
||||
self.zk.get_cluster()
|
||||
cluster = self.zk.get_cluster()
|
||||
self.assertIsInstance(cluster.leader, Leader)
|
||||
self.zk.touch_member('foo')
|
||||
self.zk.delete_leader()
|
||||
|
||||
def test_race(self):
|
||||
self.assertFalse(self.zk.race('/initialize'))
|
||||
def test_initialize(self):
|
||||
self.assertFalse(self.zk.initialize())
|
||||
|
||||
def test_cancel_initialization(self):
|
||||
self.zk.cancel_initialization()
|
||||
|
||||
def test_touch_member(self):
|
||||
self.zk.touch_member('new')
|
||||
@@ -158,5 +190,5 @@ class TestZooKeeper(unittest.TestCase):
|
||||
self.zk.last_leader_operation = -1
|
||||
self.assertTrue(self.zk.update_leader(MockPostgresql()))
|
||||
|
||||
def test_sleep(self):
|
||||
self.zk.sleep(0)
|
||||
def test_watch(self):
|
||||
self.zk.watch(0)
|
||||
|
||||
Reference in New Issue
Block a user