mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Merge branch 'master' into feature/scheduled_restarts
This commit is contained in:
Oleksii Kliukin
+21
-15
@@ -3,36 +3,42 @@
|
||||
FROM ubuntu:16.04
|
||||
MAINTAINER Feike Steenbergen <[email protected]>
|
||||
|
||||
RUN echo 'APT::Install-Recommends "0";' > /etc/apt/apt.conf.d/01norecommend
|
||||
RUN echo 'APT::Install-Suggests "0";' >> /etc/apt/apt.conf.d/01norecommend
|
||||
RUN echo 'APT::Install-Recommends "0";' > /etc/apt/apt.conf.d/01norecommend \
|
||||
&& echo 'APT::Install-Suggests "0";' >> /etc/apt/apt.conf.d/01norecommend
|
||||
|
||||
ENV PGVERSION 9.5
|
||||
ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
|
||||
RUN apt-get update -y \
|
||||
&& apt-get upgrade -y \
|
||||
&& apt-get install -y curl postgresql-${PGVERSION} python-psycopg2 python-yaml python-requests python-six python-click \
|
||||
python-dateutil python-tzlocal python-urllib3 python-dnspython python-pip python-setuptools python-kazoo python \
|
||||
&& pip install python-etcd==0.4.3 python-consul \
|
||||
&& apt-get install -y curl jq haproxy zookeeper postgresql-${PGVERSION} python-psycopg2 python-yaml \
|
||||
python-requests python-six python-click python-dateutil python-tzlocal python-urllib3 \
|
||||
python-dnspython python-pip python-setuptools python-kazoo python-prettytable python \
|
||||
&& pip install python-etcd==0.4.3 python-consul==0.6.0 --upgrade \
|
||||
&& apt-get remove -y python-pip python-setuptools \
|
||||
&& apt-get autoremove -y \
|
||||
# Clean up
|
||||
&& apt-get clean -y \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
&& rm -rf /var/lib/apt/lists/* /root/.cache
|
||||
|
||||
ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH
|
||||
ENV ETCDVERSION 2.3.6
|
||||
RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \
|
||||
| tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl
|
||||
|
||||
ENV CONFDVERSION 0.11.0
|
||||
RUN curl -L https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-amd64 > /usr/local/bin/confd \
|
||||
&& chmod +x /usr/local/bin/confd
|
||||
|
||||
ADD patronictl.py patroni.py docker/entrypoint.sh /
|
||||
ADD patroni /patroni/
|
||||
RUN ln -s /patroni/patroni.py /usr/local/bin/patroni \
|
||||
&& ln -s /patroni/patronictl.py /usr/local/bin/patronictl
|
||||
|
||||
ENV ETCDVERSION 2.3.6
|
||||
RUN curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C /bin --strip=1 --wildcards --no-anchored etcd etcdctl
|
||||
ADD extras/confd /etc/confd
|
||||
RUN ln -s /patronictl.py /usr/local/bin/patronictl
|
||||
|
||||
### 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 \
|
||||
&& chown postgres:postgres -R /patroni/ /data/ /pgpass /var/log/etcd.* /patroni/postgres.yml
|
||||
RUN mkdir /data/ && touch /pgpass /patroni.yml \
|
||||
&& chown postgres:postgres -R /patroni/ /data/ /pgpass /patroni.yml /etc/haproxy /var/run/ /var/lib/ /var/log/ \
|
||||
&& echo 1 > /etc/zookeeper/conf/myid
|
||||
|
||||
EXPOSE 4001 5432 2380
|
||||
EXPOSE 2379 5432 8008
|
||||
|
||||
ENTRYPOINT ["/bin/bash", "/entrypoint.sh"]
|
||||
USER postgres
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
# docker compose file for running a 3-node PostgreSQL cluster
|
||||
# with etcd as the SIS
|
||||
|
||||
patroni_etcd:
|
||||
container_name: patroni_etcd
|
||||
image: patroni
|
||||
command: --etcd
|
||||
|
||||
dbnode1:
|
||||
image: patroni
|
||||
hostname: dbnode1
|
||||
links:
|
||||
- patroni_etcd:patroni_etcd
|
||||
volumes:
|
||||
- ./patroni:/patroni
|
||||
env_file: docker/patroni-secrets.env
|
||||
environment:
|
||||
PATRONI_ETCD_HOST: patroni_etcd:2379
|
||||
PATRONI_NAME: dbnode1
|
||||
PATRONI_SCOPE: testcluster
|
||||
|
||||
dbnode2:
|
||||
image: patroni
|
||||
hostname: dbnode2
|
||||
links:
|
||||
- patroni_etcd:patroni_etcd
|
||||
volumes:
|
||||
- ./patroni:/patroni
|
||||
env_file: docker/patroni-secrets.env
|
||||
environment:
|
||||
PATRONI_ETCD_HOST: patroni_etcd:2379
|
||||
PATRONI_NAME: dbnode2
|
||||
PATRONI_SCOPE: testcluster
|
||||
|
||||
dbnode3:
|
||||
image: patroni
|
||||
hostname: dbnode3
|
||||
links:
|
||||
- patroni_etcd:patroni_etcd
|
||||
volumes:
|
||||
- ./patroni:/patroni
|
||||
env_file: docker/patroni-secrets.env
|
||||
environment:
|
||||
PATRONI_ETCD_HOST: patroni_etcd:2379
|
||||
PATRONI_NAME: dbnode3
|
||||
PATRONI_SCOPE: testcluster
|
||||
|
||||
haproxy:
|
||||
image: patroni
|
||||
links:
|
||||
- patroni_etcd:patroni_etcd
|
||||
ports:
|
||||
- "5000"
|
||||
- "5001"
|
||||
environment:
|
||||
PATRONI_ETCD_HOST: patroni_etcd:2379
|
||||
PATRONI_SCOPE: testcluster
|
||||
command: --confd
|
||||
@@ -67,24 +67,38 @@ while getopts "$optspec" optchar; do
|
||||
esac
|
||||
done
|
||||
|
||||
function random_name()
|
||||
{
|
||||
cat /dev/urandom | LC_ALL=C tr -dc 'a-zA-Z0-9' | head -c 8
|
||||
}
|
||||
|
||||
if [ -z ${PATRONI_SCOPE} ]
|
||||
then
|
||||
PATRONI_SCOPE=$(random_name)
|
||||
if [ -z ${PATRONI_SCOPE} ]; then
|
||||
PATRONI_SCOPE=$(cat /dev/urandom | LC_ALL=C tr -dc 'a-z0-9' | head -c 8)
|
||||
fi
|
||||
|
||||
etcd_container=$(docker run -P -d --name="${PATRONI_SCOPE}_etcd" "${DOCKER_IMAGE}" --etcd-only)
|
||||
etcd_container_ip=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' ${etcd_container})
|
||||
echo "The etcd container is ${etcd_container}, ip=${etcd_container_ip}"
|
||||
function docker_run()
|
||||
{
|
||||
local name=$1
|
||||
shift
|
||||
container=$(docker run -d --name=$name $*)
|
||||
container_ip=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' ${container})
|
||||
echo "Started container ${name}, ip=${container_ip}"
|
||||
}
|
||||
|
||||
for i in $(seq 1 "${MEMBERS}")
|
||||
do
|
||||
container_name=$(random_name)
|
||||
patroni_container=$(docker run -P -d --name="${PATRONI_SCOPE}_${container_name}" "${DOCKER_IMAGE}" --etcd="${etcd_container_ip}:4001" --name="${PATRONI_SCOPE}")
|
||||
patroni_container_ip=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' ${patroni_container})
|
||||
echo "Started Patroni container ${patroni_container}, ip=${patroni_container_ip}"
|
||||
|
||||
ETCD_CONTAINER="${PATRONI_SCOPE}_etcd"
|
||||
docker_run ${ETCD_CONTAINER} ${DOCKER_IMAGE} --etcd
|
||||
|
||||
DOCKER_ARGS="--link=${ETCD_CONTAINER}:${ETCD_CONTAINER} -e PATRONI_SCOPE=${PATRONI_SCOPE} -e PATRONI_ETCD_HOST=${ETCD_CONTAINER}:2379"
|
||||
PATRONI_ENV=$(sed 's/#.*//g' docker/patroni-secrets.env | sed -n 's/^PATRONI_.*$/-e &/p' | tr '\n' ' ')
|
||||
PATRONI_VOLUME="-v $(dirname $(dirname $(realpath $0)))/patroni:/patroni"
|
||||
|
||||
for i in $(seq 1 "${MEMBERS}"); do
|
||||
container_name=postgres${i}
|
||||
docker_run "${PATRONI_SCOPE}_${container_name}" \
|
||||
$PATRONI_VOLUME \
|
||||
$DOCKER_ARGS \
|
||||
$PATRONI_ENV \
|
||||
-e PATRONI_NAME=${container_name} \
|
||||
${DOCKER_IMAGE}
|
||||
done
|
||||
|
||||
docker_run "${PATRONI_SCOPE}_haproxy" \
|
||||
-p=5000 -p=5001 \
|
||||
$DOCKER_ARGS \
|
||||
${DOCKER_IMAGE} --confd
|
||||
|
||||
+44
-50
@@ -7,49 +7,52 @@ Usage: $0
|
||||
|
||||
Options:
|
||||
|
||||
--etcd ETCD Provide an external etcd to connect to
|
||||
--name NAME Give the cluster a specific name
|
||||
--etcd-only Do not run Patroni, run a standalone etcd
|
||||
--etcd Do not run Patroni, run a standalone etcd
|
||||
--confd Do not run Patroni, run a standalone confd
|
||||
--zookeeper Do not run Patroni, run a standalone zookeeper
|
||||
|
||||
Examples:
|
||||
|
||||
$0 --etcd=127.17.0.84:4001
|
||||
$0 --etcd-only
|
||||
$0 --etcd
|
||||
$0 --confd
|
||||
$0 --zookeeper
|
||||
$0
|
||||
$0 --name=true_scotsman
|
||||
__EOF__
|
||||
}
|
||||
|
||||
DOCKER_IP=$(hostname --ip-address)
|
||||
PATRONI_SCOPE=${PATRONI_SCOPE:-batman}
|
||||
ETCD_ARGS="--data-dir /tmp/etcd.data -advertise-client-urls=http://${DOCKER_IP}:2379 -listen-client-urls=http://0.0.0.0:2379 -listen-peer-urls=http://0.0.0.0:2380"
|
||||
|
||||
optspec=":vh-:"
|
||||
while getopts "$optspec" optchar; do
|
||||
case "${optchar}" in
|
||||
-)
|
||||
case "${OPTARG}" in
|
||||
etcd-only)
|
||||
exec etcd --data-dir /tmp/etcd.data \
|
||||
-advertise-client-urls=http://${DOCKER_IP}:4001 \
|
||||
-listen-client-urls=http://0.0.0.0:4001 \
|
||||
-listen-peer-urls=http://0.0.0.0:2380
|
||||
exit 0
|
||||
confd)
|
||||
haproxy -f /etc/haproxy/haproxy.cfg -p /var/run/haproxy.pid -D
|
||||
CONFD="confd -prefix=${PATRONI_NAMESPACE:-/service}/$PATRONI_SCOPE -interval=10 -backend"
|
||||
if [ ! -z ${PATRONI_ZOOKEEPER_HOSTS} ]; then
|
||||
while ! /usr/share/zookeeper/bin/zkCli.sh -server ${PATRONI_ZOOKEEPER_HOSTS} ls /; do
|
||||
sleep 1
|
||||
done
|
||||
exec $CONFD zookeeper -node ${PATRONI_ZOOKEEPER_HOSTS}
|
||||
else
|
||||
while ! curl -s ${PATRONI_ETCD_HOST}/v2/members | jq -r '.members[0].clientURLs[0]' | grep -q http; do
|
||||
sleep 1
|
||||
done
|
||||
exec $CONFD etcd -node $PATRONI_ETCD_HOST
|
||||
fi
|
||||
;;
|
||||
etcd)
|
||||
exec etcd $ETCD_ARGS
|
||||
;;
|
||||
zookeeper)
|
||||
exec /usr/share/zookeeper/bin/zkServer.sh start-foreground
|
||||
;;
|
||||
cheat)
|
||||
CHEAT=1
|
||||
;;
|
||||
name)
|
||||
PATRONI_SCOPE="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
name=*)
|
||||
PATRONI_SCOPE=${OPTARG#*=}
|
||||
;;
|
||||
etcd)
|
||||
ETCD_CLUSTER="${!OPTIND}"; OPTIND=$(( $OPTIND + 1 ))
|
||||
;;
|
||||
etcd=*)
|
||||
ETCD_CLUSTER=${OPTARG#*=}
|
||||
;;
|
||||
help)
|
||||
usage
|
||||
exit 0
|
||||
@@ -71,32 +74,27 @@ while getopts "$optspec" optchar; do
|
||||
done
|
||||
|
||||
## We start an etcd
|
||||
if [ -z ${ETCD_CLUSTER} ]
|
||||
then
|
||||
etcd --data-dir /tmp/etcd.data \
|
||||
-advertise-client-urls=http://${DOCKER_IP}:4001 \
|
||||
-listen-client-urls=http://0.0.0.0:4001 \
|
||||
-listen-peer-urls=http://0.0.0.0:2380 > /var/log/etcd.log 2> /var/log/etcd.err &
|
||||
ETCD_CLUSTER="127.0.0.1:4001"
|
||||
if [[ -z ${PATRONI_ETCD_HOST} && -z ${PATRONI_ZOOKEEPER_HOSTS} ]]; then
|
||||
etcd $ETCD_ARGS > /var/log/etcd.log 2> /var/log/etcd.err &
|
||||
export PATRONI_ETCD_HOST="127.0.0.1:2379"
|
||||
fi
|
||||
|
||||
export PATRONI_SCOPE
|
||||
export PATRONI_NAME="${HOSTNAME}"
|
||||
export PATRONI_ETCD_HOST="$ETCD_CLUSTER"
|
||||
export PATRONI_NAME="${PATRONI_NAME:-${HOSTNAME}}"
|
||||
export PATRONI_RESTAPI_CONNECT_ADDRESS="${DOCKER_IP}:8008"
|
||||
export PATRONI_RESTAPI_LISTEN="0.0.0.0:8008"
|
||||
export PATRONI_admin_PASSWORD="admin"
|
||||
export PATRONI_admin_OPTIONS="createdb, createrole"
|
||||
export PATRONI_admin_PASSWORD="${PATRONI_admin_PASSWORD:=admin}"
|
||||
export PATRONI_admin_OPTIONS="${PATRONI_admin_OPTIONS:-createdb, createrole}"
|
||||
export PATRONI_POSTGRESQL_CONNECT_ADDRESS="${DOCKER_IP}:5432"
|
||||
export PATRONI_POSTGRESQL_LISTEN="0.0.0.0:5432"
|
||||
export PATRONI_POSTGRESQL_DATA_DIR="data/${PATRONI_SCOPE}"
|
||||
export PATRONI_REPLICATION_USERNAME="replicator"
|
||||
export PATRONI_REPLICATION_PASSWORD="abcd"
|
||||
export PATRONI_SUPERUSER_USERNAME="postgres"
|
||||
export PATRONI_SUPERUSER_PASSWORD="postgres"
|
||||
export PATRONI_REPLICATION_USERNAME="${PATRONI_REPLICATION_USERNAME:-replicator}"
|
||||
export PATRONI_REPLICATION_PASSWORD="${PATRONI_REPLICATION_PASSWORD:-abcd}"
|
||||
export PATRONI_SUPERUSER_USERNAME="${PATRONI_SUPERUSER_USERNAME:-postgres}"
|
||||
export PATRONI_SUPERUSER_PASSWORD="${PATRONI_SUPERUSER_PASSWORD:-postgres}"
|
||||
export PATRONI_POSTGRESQL_PGPASS="$HOME/.pgpass"
|
||||
|
||||
cat > /patroni/postgres.yaml <<__EOF__
|
||||
cat > /patroni.yml <<__EOF__
|
||||
bootstrap:
|
||||
dcs:
|
||||
postgresql:
|
||||
@@ -108,14 +106,10 @@ bootstrap:
|
||||
__EOF__
|
||||
|
||||
mkdir -p "$HOME/.config/patroni"
|
||||
ln -s /patroni/postgres.yaml "$HOME/.config/patroni/patronictl.yaml"
|
||||
[ -h "$HOME/.config/patroni/patronictl.yaml" ] || ln -s /patroni.yml "$HOME/.config/patroni/patronictl.yaml"
|
||||
|
||||
if [ ! -z $CHEAT ]
|
||||
then
|
||||
while :
|
||||
do
|
||||
sleep 60
|
||||
done
|
||||
else
|
||||
exec python /patroni.py /patroni/postgres.yaml
|
||||
fi
|
||||
[ -z $CHEAT ] && exec python /patroni.py /patroni.yml
|
||||
|
||||
while true; do
|
||||
sleep 60
|
||||
done
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
PATRONI_RESTAPI_USERNAME=admin
|
||||
PATRONI_RESTAPI_PASSWORD=admin
|
||||
PATRONI_SUPERUSER_USERNAME=postgres
|
||||
PATRONI_SUPERUSER_PASSWORD=postgres
|
||||
PATRONI_REPLICATION_USERNAME=replicator
|
||||
PATRONI_REPLICATION_PASSWORD=replicate
|
||||
PATRONI_admin_PASSWORD=admin
|
||||
PATRONI_admin_OPTIONS=createdb,createrole
|
||||
@@ -55,4 +55,4 @@ REST API
|
||||
|
||||
ZooKeeper
|
||||
---------
|
||||
- **PATRONI\_ZOOKEEPER\_HOSTS**: comma separated list of ZooKeeper cluster members: 'host1:port1,host2:port2,etc...'
|
||||
- **PATRONI\_ZOOKEEPER\_HOSTS**: comma separated list of ZooKeeper cluster members: "'host1:port1','host2:port2','etc...'". It is important to quote every single entity!
|
||||
|
||||
@@ -69,6 +69,9 @@ PostgreSQL
|
||||
- **pgpass**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni.
|
||||
- **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.
|
||||
- **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds.
|
||||
- **use\_pg\_rewind**: try to use pg\_rewind on the former leader when it joins cluster as a replica.
|
||||
- **remove\_data\_directory\_on\_rewind\_failure**: If this option is enabled, Patroni will remove postgres data directory and recreate replica. Otherwise it will try to follow the new leader. Default value is **false**.
|
||||
- **replica\_method** for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value".
|
||||
|
||||
REST API
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
### confd
|
||||
|
||||
`confd` directory contains haproxy template files for the [confd](https://github.com/kelseyhightower/confd) -- lightweight configuration management tool
|
||||
You need to copy content of `confd` directory into /etcd/confd and run confd service:
|
||||
```bash
|
||||
$ confd -prefix=/service/$PATRONI_SCOPE -backend etcd -node $PATRONI_ETCD_HOST -interval=10
|
||||
```
|
||||
It will periodically update haproxy.cfg with the actual list of Patroni nodes from `etcd` and "reload" haproxy when it is necessary.
|
||||
|
||||
|
||||
### startup-scripts
|
||||
|
||||
`startup-scripts` directory contains startup scripts for various OSes and management tools for Patroni.
|
||||
@@ -0,0 +1,13 @@
|
||||
[template]
|
||||
#prefix = "/service/batman"
|
||||
#owner = "haproxy"
|
||||
#mode = "0644"
|
||||
src = "haproxy.tmpl"
|
||||
dest = "/etc/haproxy/haproxy.cfg"
|
||||
|
||||
check_cmd = "/usr/sbin/haproxy -c -f {{ .src }}"
|
||||
reload_cmd = "haproxy -f /etc/haproxy/haproxy.cfg -p /var/run/haproxy.pid -D -sf $(cat /var/run/haproxy.pid)"
|
||||
|
||||
keys = [
|
||||
"/members/",
|
||||
]
|
||||
@@ -0,0 +1,28 @@
|
||||
global
|
||||
maxconn 100
|
||||
|
||||
defaults
|
||||
log global
|
||||
mode tcp
|
||||
retries 2
|
||||
timeout client 30m
|
||||
timeout connect 4s
|
||||
timeout server 30m
|
||||
timeout check 5s
|
||||
|
||||
frontend master_postgresql
|
||||
bind *:5000
|
||||
default_backend backend_master
|
||||
|
||||
frontend replicas_postgresql
|
||||
bind *:5001
|
||||
default_backend backend_replicas
|
||||
|
||||
backend backend_master
|
||||
option httpchk OPTIONS /master
|
||||
{{range gets "/members/*"}} server {{base .Key}} {{$data := json .Value}}{{base (replace (index (split $data.conn_url "/") 2) "@" "/" -1)}} maxconn 100 check port {{index (split (index (split $data.api_url "/") 2) ":") 1}}
|
||||
{{end}}
|
||||
backend backend_replicas
|
||||
option httpchk OPTIONS /replica
|
||||
{{range gets "/members/*"}} server {{base .Key}} {{$data := json .Value}}{{base (replace (index (split $data.conn_url "/") 2) "@" "/" -1)}} maxconn 100 check port {{index (split (index (split $data.api_url "/") 2) ":") 1}}
|
||||
{{end}}
|
||||
+16
-14
@@ -118,6 +118,7 @@ class PatroniController(AbstractController):
|
||||
|
||||
with open(patroni_config_name) as f:
|
||||
config = yaml.safe_load(f)
|
||||
config.pop('etcd')
|
||||
|
||||
host = config['postgresql']['listen'].split(':')[0]
|
||||
|
||||
@@ -139,16 +140,6 @@ class PatroniController(AbstractController):
|
||||
if tags:
|
||||
config['tags'] = tags
|
||||
|
||||
if dcs != 'etcd':
|
||||
dcs_config = config.pop('etcd')
|
||||
dcs_config.pop('host')
|
||||
|
||||
if dcs == 'exhibitor':
|
||||
dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181})
|
||||
elif dcs == 'zookeeper':
|
||||
dcs_config['hosts'] = ['127.0.0.1:2181']
|
||||
config[dcs] = dcs_config
|
||||
|
||||
with open(patroni_config_path, 'w') as f:
|
||||
yaml.safe_dump(config, f, default_flow_style=False)
|
||||
|
||||
@@ -220,6 +211,7 @@ class ConsulController(AbstractDcsController):
|
||||
|
||||
def __init__(self, output_dir):
|
||||
super(ConsulController, self).__init__('consul', tempfile.mkdtemp(), output_dir)
|
||||
os.environ['PATRONI_CONSUL_HOST'] = 'localhost:8500'
|
||||
self._client = consul.Consul()
|
||||
|
||||
def _start(self):
|
||||
@@ -252,6 +244,7 @@ class EtcdController(AbstractDcsController):
|
||||
|
||||
def __init__(self, output_dir):
|
||||
super(EtcdController, self).__init__('etcd', tempfile.mkdtemp(), output_dir)
|
||||
os.environ['PATRONI_ETCD_HOST'] = 'localhost:4001'
|
||||
self._client = etcd.Client()
|
||||
|
||||
def _start(self):
|
||||
@@ -287,8 +280,10 @@ class ZooKeeperController(AbstractDcsController):
|
||||
|
||||
""" handles all zookeeper related tasks, used for the tests setup and cleanup """
|
||||
|
||||
def __init__(self, output_dir):
|
||||
def __init__(self, output_dir, export_env=True):
|
||||
super(ZooKeeperController, self).__init__('zookeeper', None, output_dir)
|
||||
if export_env:
|
||||
os.environ['PATRONI_ZOOKEEPER_HOSTS'] = "'localhost:2181'"
|
||||
self._client = kazoo.client.KazooClient()
|
||||
|
||||
def _start(self):
|
||||
@@ -321,10 +316,17 @@ class ZooKeeperController(AbstractDcsController):
|
||||
return False
|
||||
|
||||
|
||||
class ExhibitorController(ZooKeeperController):
|
||||
|
||||
def __init__(self, output_dir):
|
||||
super(ExhibitorController, self).__init__(output_dir, False)
|
||||
os.environ.update({'PATRONI_EXHIBITOR_HOSTS': 'localhost', 'PATRONI_EXHIBITOR_PORT': '8181'})
|
||||
|
||||
|
||||
class PatroniPoolController(object):
|
||||
|
||||
KNOWN_DCS = {'consul': ConsulController, 'etcd': EtcdController,
|
||||
'zookeeper': ZooKeeperController, 'exhibitor': ZooKeeperController}
|
||||
'zookeeper': ZooKeeperController, 'exhibitor': ExhibitorController}
|
||||
|
||||
def __init__(self):
|
||||
self._dcs = None
|
||||
@@ -376,8 +378,8 @@ class PatroniPoolController(object):
|
||||
@property
|
||||
def dcs(self):
|
||||
if self._dcs is None:
|
||||
self._dcs = os.environ.get('DCS', 'etcd')
|
||||
assert self._dcs in self.KNOWN_DCS, 'Unsupported dcs: ' + self.dcs
|
||||
self._dcs = os.environ.pop('DCS', 'etcd')
|
||||
assert self._dcs in self.KNOWN_DCS, 'Unsupported dcs: ' + self._dcs
|
||||
return self._dcs
|
||||
|
||||
|
||||
|
||||
@@ -10,9 +10,12 @@ Scenario: check API requests on a stand-alone server
|
||||
And I receive a response role master
|
||||
When I issue a GET request to http://127.0.0.1:8008/replica
|
||||
Then I receive a response code 503
|
||||
When I issue an empty POST request to http://127.0.0.1:8008/reinitialize
|
||||
Then I receive a response code 503
|
||||
And I receive a response text "I am the leader, can not reinitialize"
|
||||
When I run patronictl.py reinit batman postgres0 --force
|
||||
Then I receive a response returncode 0
|
||||
And I receive a response output "reinitialize failed for member postgres0, status code=503, (I am the leader, can not reinitialize)"
|
||||
When I run patronictl.py failover batman --master postgres0 --force
|
||||
Then I receive a response returncode 1
|
||||
And I receive a response output "Error: No candidates found to failover to"
|
||||
When I issue a POST request to http://127.0.0.1:8008/failover with {"leader": "postgres0"}
|
||||
Then I receive a response code 500
|
||||
And I receive a response text failover is not possible: cluster does not have members except leader
|
||||
@@ -23,24 +26,24 @@ Scenario: check API requests on a stand-alone server
|
||||
And I receive a response text "No values given for required parameters leader and candidate"
|
||||
|
||||
Scenario: check local configuration reload
|
||||
Given I issue an empty POST request to http://127.0.0.1:8008/reload
|
||||
Then I receive a response code 200
|
||||
And I receive a response text nothing changed
|
||||
When I add tag new_tag new_value to postgres0 config
|
||||
And I issue an empty POST request to http://127.0.0.1:8008/reload
|
||||
Then I receive a response code 202
|
||||
Given I issue an empty POST request to http://127.0.0.1:8008/reload
|
||||
Then I receive a response code 200
|
||||
And I receive a response text nothing changed
|
||||
When I add tag new_tag new_value to postgres0 config
|
||||
And I issue an empty POST request to http://127.0.0.1:8008/reload
|
||||
Then I receive a response code 202
|
||||
|
||||
Scenario: check dynamic configuration change via DCS
|
||||
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 1, "postgresql": {"parameters": {"max_connections": 101}}}
|
||||
Then I receive a response code 200
|
||||
And I receive a response loop_wait 1
|
||||
And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
|
||||
When I issue a GET request to http://127.0.0.1:8008/config
|
||||
Then I receive a response code 200
|
||||
And I receive a response loop_wait 1
|
||||
When I issue a GET request to http://127.0.0.1:8008/patroni
|
||||
Then I receive a response code 200
|
||||
And I receive a response tags {'tag': 'new_value'}
|
||||
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 1, "postgresql": {"parameters": {"max_connections": 101}}}
|
||||
Then I receive a response code 200
|
||||
And I receive a response loop_wait 1
|
||||
And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
|
||||
When I issue a GET request to http://127.0.0.1:8008/config
|
||||
Then I receive a response code 200
|
||||
And I receive a response loop_wait 1
|
||||
When I issue a GET request to http://127.0.0.1:8008/patroni
|
||||
Then I receive a response code 200
|
||||
And I receive a response tags {'tag': 'new_value'}
|
||||
|
||||
Scenario: check API requests for the primary-replica pair
|
||||
Given I start postgres1
|
||||
@@ -49,28 +52,30 @@ Scenario: check API requests for the primary-replica pair
|
||||
Then I receive a response code 200
|
||||
And I receive a response state running
|
||||
And I receive a response role replica
|
||||
When I issue an empty POST request to http://127.0.0.1:8009/reinitialize
|
||||
Then I receive a response code 200
|
||||
When I issue an empty POST request to http://127.0.0.1:8008/restart
|
||||
Then I receive a response code 200
|
||||
And postgres0 role is the primary after 5 seconds
|
||||
When I sleep for 10 seconds
|
||||
Then postgres1 role is the secondary after 15 seconds
|
||||
When I run patronictl.py reinit batman postgres1 --force
|
||||
Then I receive a response returncode 0
|
||||
And I receive a response output "Succesful reinitialize on member postgres1"
|
||||
When I run patronictl.py restart batman postgres0 --force
|
||||
Then I receive a response returncode 0
|
||||
And I receive a response output "Succesful restart on member postgres0"
|
||||
And postgres0 role is the primary after 5 seconds
|
||||
When I sleep for 10 seconds
|
||||
Then postgres1 role is the secondary after 15 seconds
|
||||
|
||||
Scenario: check the failover via the API
|
||||
Given I issue a POST request to http://127.0.0.1:8008/failover with {"leader": "postgres0", "candidate": "postgres1"}
|
||||
Then I receive a response code 200
|
||||
Given I run patronictl.py failover batman --master postgres0 --candidate postgres1 --force
|
||||
Then I receive a response returncode 0
|
||||
And postgres1 is a leader after 5 seconds
|
||||
And postgres1 role is the primary after 5 seconds
|
||||
And postgres0 role is the secondary after 10 seconds
|
||||
And postgres1 role is the primary after 5 seconds
|
||||
And postgres0 role is the secondary after 10 seconds
|
||||
And replication works from postgres1 to postgres0 after 20 seconds
|
||||
|
||||
Scenario: check the scheduled failover
|
||||
Given I issue a scheduled failover at http://127.0.0.1:8009 from postgres1 to postgres0 in 1 seconds
|
||||
Then I receive a response code 202
|
||||
Given I issue a scheduled failover from postgres1 to postgres0 in 1 seconds
|
||||
Then I receive a response returncode 0
|
||||
And postgres0 is a leader after 20 seconds
|
||||
And postgres0 role is the primary after 5 seconds
|
||||
And postgres1 role is the secondary after 10 seconds
|
||||
And postgres0 role is the primary after 5 seconds
|
||||
And postgres1 role is the secondary after 10 seconds
|
||||
And replication works from postgres0 to postgres1 after 25 seconds
|
||||
|
||||
Scenario: check the scheduled restart
|
||||
|
||||
@@ -2,6 +2,8 @@ import json
|
||||
import parse
|
||||
import pytz
|
||||
import requests
|
||||
import shlex
|
||||
import subprocess
|
||||
import time
|
||||
import yaml
|
||||
|
||||
@@ -84,23 +86,39 @@ def do_request(context, request_method, url, data):
|
||||
_set_response(context, r)
|
||||
|
||||
|
||||
@step('I run {cmd}')
|
||||
def do_run(context, cmd):
|
||||
cmd = ['coverage', 'run', '--source=patroni', '-p'] + shlex.split(cmd)
|
||||
try:
|
||||
response = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
|
||||
context.status_code = 0
|
||||
except subprocess.CalledProcessError as e:
|
||||
response = e.output
|
||||
context.status_code = e.returncode
|
||||
context.response = response.decode('utf-8').strip()
|
||||
|
||||
|
||||
@then('I receive a response {component:w} {data}')
|
||||
def check_response(context, component, data):
|
||||
if component == 'code':
|
||||
assert context.status_code == int(data),\
|
||||
"status code {0} != {1}, response: {2}".format(context.status_code, int(data), context.response)
|
||||
"status code {0} != {1}, response: {2}".format(context.status_code, data, context.response)
|
||||
elif component == 'returncode':
|
||||
assert context.status_code == int(data), "return code {0} != {1}".format(context.status_code, data)
|
||||
elif component == 'text':
|
||||
assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data)
|
||||
elif component == 'output':
|
||||
assert data.strip('"') in context.response, "response {0} does not contain {1}".format(context.response, data)
|
||||
else:
|
||||
assert component in context.response, "{0} is not part of the response".format(component)
|
||||
assert str(context.response[component]) == str(data), "{0} does not contain {1}".format(component, data)
|
||||
|
||||
|
||||
@step('I issue a scheduled failover at {at_url:url} from {from_host:w} to {to_host:w} in {in_seconds:d} seconds')
|
||||
def scheduled_failover(context, at_url, from_host, to_host, in_seconds):
|
||||
@step('I issue a scheduled failover from {from_host:w} to {to_host:w} in {in_seconds:d} seconds')
|
||||
def scheduled_failover(context, from_host, to_host, in_seconds):
|
||||
context.execute_steps(u"""
|
||||
Given I issue a POST request to {0}/failover with {{"leader": "{1}", "candidate": "{2}", "scheduled_at": "{3}"}}
|
||||
""".format(at_url, from_host, to_host, datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds))))
|
||||
Given I run patronictl.py failover batman --master {0} --candidate {1} --scheduled "{2}" --force
|
||||
""".format(from_host, to_host, datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds))))
|
||||
|
||||
|
||||
@step('I issue a scheduled restart at {url:url} in {in_seconds:d} seconds with {data}')
|
||||
|
||||
+12
-13
@@ -18,6 +18,8 @@ logger = logging.getLogger(__name__)
|
||||
class Patroni(object):
|
||||
|
||||
def __init__(self):
|
||||
self.setup_signal_handlers()
|
||||
|
||||
self.version = __version__
|
||||
self.config = Config()
|
||||
self.dcs = get_dcs(self.config)
|
||||
@@ -32,10 +34,6 @@ class Patroni(object):
|
||||
self.next_run = time.time()
|
||||
self.scheduled_restart = {}
|
||||
|
||||
self._reload_config_scheduled = False
|
||||
self._received_sighup = False
|
||||
self._received_sigterm = False
|
||||
|
||||
def load_dynamic_configuration(self):
|
||||
while True:
|
||||
try:
|
||||
@@ -52,6 +50,10 @@ class Patroni(object):
|
||||
return {tag: value for tag, value in self.config.get('tags', {}).items()
|
||||
if tag not in ('clonefrom', 'nofailover', 'noloadbalance') or value}
|
||||
|
||||
@property
|
||||
def nofailover(self):
|
||||
return self.tags.get('nofailover', False)
|
||||
|
||||
def reload_config(self):
|
||||
try:
|
||||
self.tags = self.get_tags()
|
||||
@@ -63,6 +65,10 @@ class Patroni(object):
|
||||
except Exception:
|
||||
logger.exception('Failed to reload config_file=%s', self.config.config_file)
|
||||
|
||||
@property
|
||||
def replicatefrom(self):
|
||||
return self.tags.get('replicatefrom')
|
||||
|
||||
def sighup_handler(self, *args):
|
||||
self._received_sighup = True
|
||||
|
||||
@@ -75,14 +81,6 @@ class Patroni(object):
|
||||
def noloadbalance(self):
|
||||
return self.tags.get('noloadbalance', False)
|
||||
|
||||
@property
|
||||
def nofailover(self):
|
||||
return self.tags.get('nofailover', False)
|
||||
|
||||
@property
|
||||
def replicatefrom(self):
|
||||
return self.tags.get('replicatefrom')
|
||||
|
||||
def schedule_next_run(self):
|
||||
self.next_run += self.nap_time
|
||||
current_time = time.time()
|
||||
@@ -115,6 +113,8 @@ class Patroni(object):
|
||||
self.schedule_next_run()
|
||||
|
||||
def setup_signal_handlers(self):
|
||||
self._received_sighup = False
|
||||
self._received_sigterm = False
|
||||
signal.signal(signal.SIGHUP, self.sighup_handler)
|
||||
signal.signal(signal.SIGTERM, self.sigterm_handler)
|
||||
signal.signal(signal.SIGCHLD, sigchld_handler)
|
||||
@@ -125,7 +125,6 @@ def main():
|
||||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||||
|
||||
patroni = Patroni()
|
||||
patroni.setup_signal_handlers()
|
||||
try:
|
||||
patroni.run()
|
||||
except KeyboardInterrupt:
|
||||
|
||||
+1
-1
@@ -397,7 +397,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
if state == 'running':
|
||||
logger.exception('get_postgresql_status')
|
||||
state = 'unknown'
|
||||
return {'state': state}
|
||||
return {'state': state, 'role': self.server.patroni.postgresql.role}
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
logger.debug("API thread: %s - - [%s] %s", self.client_address[0], self.log_date_time_string(), fmt % args)
|
||||
|
||||
@@ -224,6 +224,7 @@ class Config(object):
|
||||
try:
|
||||
return yaml.safe_load(value)
|
||||
except Exception:
|
||||
logger.exception('Exception when parsing list %s', value)
|
||||
return None
|
||||
|
||||
for param in list(os.environ.keys()):
|
||||
|
||||
@@ -184,7 +184,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
|
||||
|
||||
"""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 initialize: shows whether this cluster has initialization key stored in DC or not.
|
||||
:param config: global dynamic configuration, reference to `ClusterConfig` object
|
||||
: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.
|
||||
|
||||
+24
-11
@@ -26,7 +26,7 @@ class EtcdError(DCSError):
|
||||
class Client(etcd.Client):
|
||||
|
||||
def __init__(self, config):
|
||||
super(Client, self).__init__(read_timeout=5)
|
||||
super(Client, self).__init__(read_timeout=config['retry_timeout'])
|
||||
self._config = config
|
||||
self._load_machines_cache()
|
||||
self._allow_reconnect = True
|
||||
@@ -50,6 +50,9 @@ class Client(etcd.Client):
|
||||
self._update_machines_cache = True
|
||||
return [self._base_uri]
|
||||
|
||||
def set_read_timeout(self, timeout):
|
||||
self._read_timeout = timeout
|
||||
|
||||
def _do_http_request(self, request_executor, method, url, fields=None, **kwargs):
|
||||
try:
|
||||
response = request_executor(method, url, fields=fields, **kwargs)
|
||||
@@ -70,13 +73,7 @@ class Client(etcd.Client):
|
||||
if not path.startswith('/'):
|
||||
raise ValueError('Path does not start with /')
|
||||
|
||||
if timeout is None:
|
||||
timeout = self.read_timeout
|
||||
|
||||
if timeout == 0:
|
||||
timeout = None
|
||||
|
||||
kwargs = {'timeout': timeout, 'fields': params, 'redirect': self.allow_redirect,
|
||||
kwargs = {'fields': params, 'redirect': self.allow_redirect,
|
||||
'headers': self._get_headers(), 'preload_content': False}
|
||||
|
||||
if method in [self._MGET, self._MDELETE]:
|
||||
@@ -91,6 +88,22 @@ class Client(etcd.Client):
|
||||
if self._update_machines_cache:
|
||||
self._load_machines_cache()
|
||||
|
||||
if timeout is None:
|
||||
# calculate the number of retries and timeout *per node*
|
||||
# actual number of retries depends on the number of nodes
|
||||
etcd_nodes = len(self._machines_cache) + 1
|
||||
kwargs['retries'] = 0 if etcd_nodes > 3 else (1 if etcd_nodes > 1 else 2)
|
||||
|
||||
# if etcd_nodes > 3:
|
||||
# kwargs.update({'retries': 0, 'timeout': float(self.read_timeout)/etcd_nodes})
|
||||
# elif etcd_nodes > 1:
|
||||
# kwargs.update({'retries': 1, 'timeout': self.read_timeout/2.0/etcd_nodes})
|
||||
# else:
|
||||
# kwargs.update({'retries': 2, 'timeout': self.read_timeout/3.0})
|
||||
kwargs['timeout'] = self.read_timeout/float(kwargs['retries'] + 1)/etcd_nodes
|
||||
else:
|
||||
kwargs.update({'retries': 0, 'timeout': timeout})
|
||||
|
||||
response = False
|
||||
|
||||
try:
|
||||
@@ -122,7 +135,7 @@ class Client(etcd.Client):
|
||||
for host, port in self.get_srv_record(discovery_srv):
|
||||
url = '{0}://{1}:{2}/members'.format(self._protocol, host, port)
|
||||
try:
|
||||
response = requests.get(url, timeout=5)
|
||||
response = requests.get(url, timeout=self.read_timeout)
|
||||
if response.ok:
|
||||
for member in response.json():
|
||||
ret.extend(member['clientURLs'])
|
||||
@@ -195,8 +208,7 @@ class Etcd(AbstractDCS):
|
||||
super(Etcd, self).__init__(config)
|
||||
self._ttl = int(config.get('ttl') or 30)
|
||||
self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
|
||||
retry_exceptions=(etcd.EtcdConnectionFailed,
|
||||
etcd.EtcdLeaderElectionInProgress,
|
||||
retry_exceptions=(etcd.EtcdLeaderElectionInProgress,
|
||||
etcd.EtcdWatcherCleared,
|
||||
etcd.EtcdEventIndexCleared))
|
||||
self._client = self.get_etcd_client(config)
|
||||
@@ -223,6 +235,7 @@ class Etcd(AbstractDCS):
|
||||
|
||||
def set_retry_timeout(self, retry_timeout):
|
||||
self._retry.deadline = retry_timeout
|
||||
self._client.set_read_timeout(retry_timeout)
|
||||
|
||||
@staticmethod
|
||||
def member(node):
|
||||
|
||||
@@ -2,6 +2,7 @@ import logging
|
||||
|
||||
from kazoo.client import KazooClient, KazooState
|
||||
from kazoo.exceptions import NoNodeError, NodeExistsError
|
||||
from kazoo.handlers.threading import SequentialThreadingHandler
|
||||
from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member
|
||||
from patroni.exceptions import DCSError
|
||||
|
||||
@@ -12,6 +13,34 @@ class ZooKeeperError(DCSError):
|
||||
pass
|
||||
|
||||
|
||||
class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
|
||||
|
||||
def __init__(self, connect_timeout):
|
||||
super(PatroniSequentialThreadingHandler, self).__init__()
|
||||
self.set_connect_timeout(connect_timeout)
|
||||
|
||||
def set_connect_timeout(self, connect_timeout):
|
||||
self._connect_timeout = max(1.0, connect_timeout/4.0)
|
||||
|
||||
def create_connection(self, *args, **kwargs):
|
||||
"""This method is trying to establish connection with one of the zookeeper nodes.
|
||||
Somehow strategy "fail earlier and retry more often" works way better comparing to
|
||||
the original strategy "try to connect with specified timeout".
|
||||
Since we want to try connect to zookeeper more often (with the smaller connect_timeout),
|
||||
he have to override `create_connection` method in the `SequentialThreadingHandler`
|
||||
class (which is used by `kazoo.Client`).
|
||||
|
||||
:param args: always contains `tuple(host, port)` as the first element and could contain
|
||||
`connect_timeout` (negotiated session timeout) as the second element."""
|
||||
|
||||
args = list(args)
|
||||
if len(args) == 1:
|
||||
args.append(self._connect_timeout)
|
||||
else:
|
||||
args[1] = max(self._connect_timeout, args[1]/10.0)
|
||||
return super(PatroniSequentialThreadingHandler, self).create_connection(*args, **kwargs)
|
||||
|
||||
|
||||
class ZooKeeper(AbstractDCS):
|
||||
|
||||
def __init__(self, config):
|
||||
@@ -21,7 +50,8 @@ class ZooKeeper(AbstractDCS):
|
||||
if isinstance(hosts, list):
|
||||
hosts = ','.join(hosts)
|
||||
|
||||
self._client = KazooClient(hosts, timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1},
|
||||
self._client = KazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
|
||||
timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1},
|
||||
command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1})
|
||||
self._client.add_listener(self.session_listener)
|
||||
|
||||
@@ -47,6 +77,7 @@ class ZooKeeper(AbstractDCS):
|
||||
self._client.restart()
|
||||
|
||||
def set_retry_timeout(self, retry_timeout):
|
||||
self._client.handler.set_connect_timeout(retry_timeout)
|
||||
self._client._retry.deadline = retry_timeout
|
||||
|
||||
def get_node(self, key, watch=None):
|
||||
|
||||
+6
-12
@@ -92,7 +92,7 @@ class Ha(object):
|
||||
self._async_executor.run_async(self.clone, args=(clone_member, msg))
|
||||
return 'trying to bootstrap {0}'.format(msg)
|
||||
# no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file
|
||||
elif not (self.cluster.initialize or self.patroni.nofailover) and 'bootstrap' in self.patroni.config:
|
||||
elif self.cluster.initialize is None and not self.patroni.nofailover and 'bootstrap' in self.patroni.config:
|
||||
if self.dcs.initialize(create_new=True): # race for initialization
|
||||
try:
|
||||
self.state_handler.bootstrap(self.patroni.config['bootstrap'])
|
||||
@@ -170,9 +170,8 @@ class Ha(object):
|
||||
logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content)
|
||||
json = response.json()
|
||||
is_master = json['role'] == 'master'
|
||||
xlog_location = json['xlog']['location' if is_master else 'replayed_location']
|
||||
tags = json.get('tags', dict())
|
||||
return (member, True, not is_master, xlog_location, tags)
|
||||
xlog_location = None if is_master else json['xlog']['replayed_location']
|
||||
return (member, True, not is_master, xlog_location, json.get('tags', {}))
|
||||
except:
|
||||
logger.exception('request failed: GET %s', member.api_url)
|
||||
return (member, False, None, 0, {})
|
||||
@@ -187,12 +186,6 @@ class Ha(object):
|
||||
def _is_healthiest_node(self, members, check_replication_lag=True):
|
||||
"""This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
|
||||
|
||||
if self.state_handler.is_leader():
|
||||
return True
|
||||
|
||||
if self.patroni.nofailover is True:
|
||||
return False
|
||||
|
||||
if check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation):
|
||||
return False # Too far behind last reported xlog location on master
|
||||
|
||||
@@ -264,7 +257,6 @@ class Ha(object):
|
||||
return self._is_healthiest_node(members, check_replication_lag=False)
|
||||
|
||||
def is_healthiest_node(self):
|
||||
|
||||
if self.state_handler.is_leader(): # leader is always the healthiest
|
||||
return True
|
||||
|
||||
@@ -281,6 +273,7 @@ class Ha(object):
|
||||
def demote(self, delete_leader=True):
|
||||
if delete_leader:
|
||||
self.state_handler.stop()
|
||||
self.state_handler.set_role('unknown')
|
||||
self.dcs.delete_leader()
|
||||
self.touch_member()
|
||||
self.dcs.reset_cluster()
|
||||
@@ -516,7 +509,8 @@ class Ha(object):
|
||||
def sysid_valid(sysid):
|
||||
# sysid does tv_sec << 32, where tv_sec is the number of seconds sine 1970,
|
||||
# so even 1 << 32 would have 10 digits.
|
||||
return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit()
|
||||
sysid = str(sysid)
|
||||
return len(sysid) >= 10 and sysid.isdigit()
|
||||
|
||||
def post_recover(self):
|
||||
if not self.state_handler.is_running():
|
||||
|
||||
+83
-31
@@ -87,7 +87,7 @@ class Postgresql(object):
|
||||
self._replication = config['authentication']['replication']
|
||||
self.resolve_connection_addresses()
|
||||
|
||||
self._use_pg_rewind = config.get('use_pg_rewind', False)
|
||||
self._need_rewind = False
|
||||
self._use_slots = config.get('use_slots', True)
|
||||
self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
|
||||
self._major_version = self.get_major_version()
|
||||
@@ -106,8 +106,6 @@ class Postgresql(object):
|
||||
self._trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote'
|
||||
self._trigger_file = os.path.abspath(os.path.join(self._data_dir, self._trigger_file))
|
||||
|
||||
self._pg_ctl = ['pg_ctl', '-w', '-D', self._data_dir]
|
||||
|
||||
self._connection = None
|
||||
self._cursor_holder = None
|
||||
self._sysid = None
|
||||
@@ -152,6 +150,22 @@ class Postgresql(object):
|
||||
self.connection_string = 'postgres://{connect_address}/{database}'.format(
|
||||
connect_address=self._connect_address or self._local_address, database=self._database)
|
||||
|
||||
def pg_ctl(self, cmd, *args, **kwargs):
|
||||
"""Builds and executes pg_ctl command
|
||||
|
||||
:returns: `!True` when return_code == 0, otherwise `!False`"""
|
||||
|
||||
pg_ctl = ['pg_ctl', cmd]
|
||||
if cmd in ('start', 'stop', 'restart'):
|
||||
pg_ctl += ['-w']
|
||||
timeout = self.config.get('pg_ctl_timeout')
|
||||
if timeout:
|
||||
try:
|
||||
pg_ctl += ['-t', str(int(timeout))]
|
||||
except Exception:
|
||||
logger.error('Bad value of pg_ctl_timeout: %s', timeout)
|
||||
return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0
|
||||
|
||||
def reload_config(self, config):
|
||||
server_parameters = self.get_server_parameters(config)
|
||||
|
||||
@@ -223,7 +237,7 @@ class Postgresql(object):
|
||||
we have either wal_log_hints or checksums turned on
|
||||
"""
|
||||
# low-hanging fruit: check if pg_rewind configuration is there
|
||||
if not (self._use_pg_rewind and all(self._superuser.get(n) for n in ('username', 'password'))):
|
||||
if not (self.config.get('use_pg_rewind') and all(self._superuser.get(n) for n in ('username', 'password'))):
|
||||
return False
|
||||
|
||||
cmd = ['pg_rewind', '--help']
|
||||
@@ -339,8 +353,9 @@ class Postgresql(object):
|
||||
os.write(fd, self._superuser['password'].encode('utf-8'))
|
||||
os.close(fd)
|
||||
options.append('--pwfile={0}'.format(pwfile))
|
||||
options = ['-o', ' '.join(options)] if options else []
|
||||
|
||||
ret = subprocess.call(self._pg_ctl + ['initdb'] + (['-o', ' '.join(options)] if options else [])) == 0
|
||||
ret = self.pg_ctl('initdb', *options)
|
||||
if pwfile:
|
||||
os.remove(pwfile)
|
||||
if ret:
|
||||
@@ -508,7 +523,7 @@ class Postgresql(object):
|
||||
options = ' '.join("--{0}='{1}'".format(p, self._server_parameters[p]) for p, v in self.CMDLINE_OPTIONS.items()
|
||||
if self._major_version >= v[2])
|
||||
|
||||
ret = subprocess.call(self._pg_ctl + ['start', '-o', options], env=env, preexec_fn=os.setsid) == 0
|
||||
ret = self.pg_ctl('start', '-o', options, env=env, preexec_fn=os.setsid)
|
||||
self._pending_restart = False
|
||||
|
||||
self.set_state('running' if ret else 'start failed')
|
||||
@@ -522,6 +537,7 @@ class Postgresql(object):
|
||||
return ret
|
||||
|
||||
def checkpoint(self, connect_kwargs=None):
|
||||
check_not_is_in_recovery = connect_kwargs is not None
|
||||
connect_kwargs = connect_kwargs or self._connect_kwargs
|
||||
for p in ['connect_timeout', 'options']:
|
||||
connect_kwargs.pop(p, None)
|
||||
@@ -530,9 +546,14 @@ class Postgresql(object):
|
||||
conn.autocommit = True
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SET statement_timeout = 0")
|
||||
cur.execute('CHECKPOINT')
|
||||
if check_not_is_in_recovery:
|
||||
cur.execute('SELECT pg_is_in_recovery()')
|
||||
if cur.fetchone()[0]:
|
||||
return 'is_in_recovery=true'
|
||||
return cur.execute('CHECKPOINT')
|
||||
except psycopg2.Error:
|
||||
logging.exception('Exception during CHECKPOINT')
|
||||
return 'not accessible or not healty'
|
||||
|
||||
def stop(self, mode='fast', block_callbacks=False, checkpoint=True):
|
||||
# make sure we close all connections established against
|
||||
@@ -552,10 +573,11 @@ class Postgresql(object):
|
||||
if not block_callbacks:
|
||||
self.set_state('stopping')
|
||||
|
||||
ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0
|
||||
ret = self.pg_ctl('stop', '-m', mode)
|
||||
# block_callbacks is used during restart to avoid
|
||||
# running start/stop callbacks in addition to restart ones
|
||||
if not ret:
|
||||
logger.warning('pg_ctl stop failed')
|
||||
self.set_state('stop failed')
|
||||
elif not block_callbacks:
|
||||
self.set_state('stopped')
|
||||
@@ -563,7 +585,7 @@ class Postgresql(object):
|
||||
return ret
|
||||
|
||||
def reload(self):
|
||||
ret = subprocess.call(self._pg_ctl + ['reload']) == 0
|
||||
ret = self.pg_ctl('reload')
|
||||
if ret:
|
||||
self.call_nowait(ACTION_ON_RELOAD)
|
||||
return ret
|
||||
@@ -631,18 +653,13 @@ class Postgresql(object):
|
||||
if name not in ('standby_mode', 'recovery_target_timeline', 'primary_conninfo', 'primary_slot_name'):
|
||||
f.write("{0} = '{1}'\n".format(name, value))
|
||||
|
||||
def rewind(self, leader):
|
||||
def rewind(self, r):
|
||||
# prepare pg_rewind connection
|
||||
r = get_conn_kwargs(leader.conn_url, self._superuser)
|
||||
env = self.write_pgpass(r)
|
||||
pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r)
|
||||
# first run a checkpoint on a promoted master in order
|
||||
# to make it store the new timeline ([email protected])
|
||||
self.checkpoint(r)
|
||||
logger.info("running pg_rewind from %s", pc)
|
||||
pg_rewind = ['pg_rewind', '-D', self._data_dir, '--source-server', pc]
|
||||
dsn = 'user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1'.format(**r)
|
||||
logger.info('running pg_rewind from %s', dsn)
|
||||
try:
|
||||
return subprocess.call(pg_rewind, env=env) == 0
|
||||
return subprocess.call(['pg_rewind', '-D', self._data_dir, '--source-server', dsn], env=env) == 0
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@@ -709,39 +726,73 @@ class Postgresql(object):
|
||||
def follow(self, member, leader, recovery=False):
|
||||
if self.check_recovery_conf(member) and not recovery:
|
||||
return True
|
||||
|
||||
change_role = self.role == 'master'
|
||||
need_rewind = change_role and self.can_rewind
|
||||
if need_rewind:
|
||||
|
||||
if change_role:
|
||||
if leader:
|
||||
if leader.name == self.name:
|
||||
self._need_rewind = False
|
||||
member = None
|
||||
if self.is_running():
|
||||
return
|
||||
else:
|
||||
self._need_rewind = bool(leader.conn_url) and self.can_rewind
|
||||
else:
|
||||
self._need_rewind = False
|
||||
member = None
|
||||
|
||||
if self._need_rewind:
|
||||
logger.info("set the rewind flag after demote")
|
||||
if leader and leader.name != self.name and need_rewind: # we have a leader and need to rewind
|
||||
if self.is_running():
|
||||
self.stop()
|
||||
|
||||
self.set_role('unknown')
|
||||
if self.is_running() and not self.stop():
|
||||
return logger.warning('Can not run pg_rewind because postgres is still running')
|
||||
|
||||
if not (leader and leader.conn_url):
|
||||
return logger.info('Leader unknown, can not rewind')
|
||||
|
||||
# prepare pg_rewind connection
|
||||
r = get_conn_kwargs(leader.conn_url, self._superuser)
|
||||
|
||||
# first make sure that we are really trying to rewind
|
||||
# from the master and run a checkpoint on a t in order to
|
||||
# make it store the new timeline ([email protected])
|
||||
leader_status = self.checkpoint(r)
|
||||
if leader_status:
|
||||
return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
|
||||
|
||||
# at present, pg_rewind only runs when the cluster is shut down cleanly
|
||||
# and not shutdown in recovery. We have to remove the recovery.conf if present
|
||||
# and start/shutdown in a single user mode to emulate this.
|
||||
# XXX: if recovery.conf is linked, it will be written anew as a normal file.
|
||||
if os.path.islink(self._recovery_conf):
|
||||
if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf):
|
||||
os.unlink(self._recovery_conf)
|
||||
elif os.path.isfile(self._recovery_conf):
|
||||
os.remove(self._recovery_conf)
|
||||
|
||||
# Archived segments might be useful to pg_rewind,
|
||||
# clean the flags that tell we should remove them.
|
||||
self.cleanup_archive_status()
|
||||
|
||||
# Start in a single user mode and stop to produce a clean shutdown
|
||||
opts = self.read_postmaster_opts()
|
||||
opts.update({'archive_mode': 'on', 'archive_command': 'false'})
|
||||
self.single_user_mode(options=opts)
|
||||
if self.rewind(leader):
|
||||
|
||||
if self.rewind(r) or not self.config.get('remove_data_directory_on_rewind_failure', False):
|
||||
self.write_recovery_conf(member)
|
||||
ret = self.start()
|
||||
else:
|
||||
logger.error("unable to rewind the former master")
|
||||
logger.error('unable to rewind the former master')
|
||||
self.remove_data_directory()
|
||||
self.set_role('uninitialized')
|
||||
ret = True
|
||||
else: # do not rewind until the leader becomes available
|
||||
self._need_rewind = False
|
||||
else:
|
||||
self.write_recovery_conf(member)
|
||||
ret = self.restart()
|
||||
if change_role and ret:
|
||||
self.set_role('replica')
|
||||
|
||||
if change_role:
|
||||
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
||||
return ret
|
||||
|
||||
@@ -770,10 +821,11 @@ class Postgresql(object):
|
||||
def promote(self):
|
||||
if self.role == 'master':
|
||||
return True
|
||||
ret = subprocess.call(self._pg_ctl + ['promote']) == 0
|
||||
ret = self.pg_ctl('promote')
|
||||
if ret:
|
||||
self.set_role('master')
|
||||
logger.info("cleared rewind flag after becoming the leader")
|
||||
self._need_rewind = False
|
||||
self.call_nowait(ACTION_ON_ROLE_CHANGE)
|
||||
return ret
|
||||
|
||||
|
||||
+9
-1
@@ -81,14 +81,20 @@ def parse_bool(value):
|
||||
def strtol(value, strict=True):
|
||||
"""As most as possible close equivalent of strtol(3) function (with base=0),
|
||||
used by postgres to parse parameter values.
|
||||
>>> strtol(0) == (0, '')
|
||||
True
|
||||
>>> strtol(1) == (1, '')
|
||||
True
|
||||
>>> strtol(9) == (9, '')
|
||||
True
|
||||
>>> strtol(' +0x400MB') == (1024, 'MB')
|
||||
True
|
||||
>>> strtol(' -070d') == (-56, 'd')
|
||||
True
|
||||
>>> strtol(' d ') == (None, 'd')
|
||||
True
|
||||
>>> strtol('9s', False) == (9, 's')
|
||||
True
|
||||
>>> strtol(' s ', False) == (1, 's')
|
||||
True
|
||||
"""
|
||||
@@ -112,7 +118,7 @@ def strtol(value, strict=True):
|
||||
base = 10
|
||||
|
||||
ret = None
|
||||
while i < l:
|
||||
while i <= l:
|
||||
try: # try to find maximally long number
|
||||
i += 1 # by giving to `int` longer and longer strings
|
||||
ret = long(value[:i], base)
|
||||
@@ -137,6 +143,8 @@ def parse_int(value, base_unit=None):
|
||||
True
|
||||
>>> parse_int('1GB', 'MB') is None
|
||||
True
|
||||
>>> parse_int(0) == 0
|
||||
True
|
||||
"""
|
||||
|
||||
convert = {
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
__version__ = '0.90'
|
||||
__version__ = '1.0'
|
||||
|
||||
@@ -41,7 +41,7 @@ class TestConfig(unittest.TestCase):
|
||||
'PATRONI_POSTGRESQL_PGPASS': '/tmp/pgpass0',
|
||||
'PATRONI_ETCD_HOST': '127.0.0.1:2379',
|
||||
'PATRONI_CONSUL_HOST': '127.0.0.1:8500',
|
||||
'PATRONI_ZOOKEEPER_HOSTS': 'host1,host2',
|
||||
'PATRONI_ZOOKEEPER_HOSTS': "'host1:2181','host2:2181'",
|
||||
'PATRONI_EXHIBITOR_HOSTS': 'host1,host2',
|
||||
'PATRONI_EXHIBITOR_PORT': '8181',
|
||||
'PATRONI_foo_HOSTS': '[host1,host2', # Exception in parse_list
|
||||
|
||||
+3
-2
@@ -147,7 +147,7 @@ class TestClient(unittest.TestCase):
|
||||
def setUp(self):
|
||||
with patch.object(etcd.Client, 'machines') as mock_machines:
|
||||
mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
|
||||
self.client = Client({'discovery_srv': 'test'})
|
||||
self.client = Client({'discovery_srv': 'test', 'retry_timeout': 3})
|
||||
self.client.http.request = http_request
|
||||
self.client.http.request_encode_body = http_request
|
||||
|
||||
@@ -204,7 +204,8 @@ class TestEtcd(unittest.TestCase):
|
||||
with patch.object(etcd.Client, 'machines') as mock_machines:
|
||||
mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
|
||||
with patch('time.sleep', Mock(side_effect=SleepException())):
|
||||
self.assertRaises(SleepException, self.etcd.get_etcd_client, {'discovery_srv': 'test'})
|
||||
self.assertRaises(SleepException, self.etcd.get_etcd_client,
|
||||
{'discovery_srv': 'test', 'retry_timeout': 10})
|
||||
|
||||
def test_get_cluster(self):
|
||||
self.assertIsInstance(self.etcd.get_cluster(), Cluster)
|
||||
|
||||
+48
-29
@@ -171,7 +171,8 @@ class TestPostgresql(unittest.TestCase):
|
||||
'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
|
||||
'authentication': {'superuser': {'username': 'test', 'password': 'test'},
|
||||
'replication': {'username': 'replicator', 'password': 'rep-pass'}},
|
||||
'use_pg_rewind': True,
|
||||
'remove_data_directory_on_rewind_failure': True,
|
||||
'use_pg_rewind': True, 'pg_ctl_timeout': 'bla',
|
||||
'parameters': self._PARAMETERS,
|
||||
'recovery_conf': {'foo': 'bar'},
|
||||
'callbacks': {'on_start': 'true', 'on_stop': 'true',
|
||||
@@ -230,47 +231,65 @@ class TestPostgresql(unittest.TestCase):
|
||||
def test_write_pgpass(self):
|
||||
self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'})
|
||||
|
||||
def test_checkpoint(self):
|
||||
with patch.object(MockCursor, 'fetchone', Mock(return_value=(True, ))):
|
||||
self.assertEquals(self.p.checkpoint({'user': 'postgres'}), 'is_in_recovery=true')
|
||||
with patch.object(MockCursor, 'execute', Mock(return_value=None)):
|
||||
self.assertIsNone(self.p.checkpoint())
|
||||
self.assertEquals(self.p.checkpoint(), 'not accessible or not healty')
|
||||
|
||||
@patch('subprocess.call', side_effect=OSError)
|
||||
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
|
||||
def test_pg_rewind(self, mock_call):
|
||||
self.assertTrue(self.p.rewind(self.leader))
|
||||
r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''}
|
||||
self.assertTrue(self.p.rewind(r))
|
||||
subprocess.call = mock_call
|
||||
self.assertFalse(self.p.rewind(self.leader))
|
||||
self.assertFalse(self.p.rewind(r))
|
||||
|
||||
@patch('patroni.postgresql.Postgresql.rewind', return_value=False)
|
||||
@patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
|
||||
@patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
|
||||
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
|
||||
@patch('os.unlink', Mock(return_value=True))
|
||||
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
|
||||
@patch.object(Postgresql, 'remove_data_directory', Mock(return_value=True))
|
||||
@patch.object(Postgresql, 'single_user_mode', Mock(return_value=1))
|
||||
@patch.object(Postgresql, 'write_pgpass', Mock(return_value={}))
|
||||
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
|
||||
@patch.object(Postgresql, 'can_rewind', PropertyMock(return_value=True))
|
||||
@patch.object(Postgresql, 'rewind', return_value=False)
|
||||
def test_follow(self, mock_pg_rewind):
|
||||
self.p.follow(None, None)
|
||||
self.p.follow(self.leader, self.leader)
|
||||
self.p.follow(Leader(-1, 28, self.other), self.leader)
|
||||
self.p.rewind = mock_pg_rewind
|
||||
self.p.follow(self.leader, self.leader)
|
||||
with mock.patch('os.path.islink', MagicMock(return_value=True)):
|
||||
with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
|
||||
with mock.patch('os.unlink', MagicMock(return_value=True)):
|
||||
self.p.follow(self.leader, self.leader, recovery=True)
|
||||
with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
|
||||
self.p.rewind.return_value = True
|
||||
self.p.follow(self.leader, self.leader, recovery=True)
|
||||
self.p.rewind.return_value = False
|
||||
self.p.follow(self.leader, self.leader, recovery=True)
|
||||
with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)):
|
||||
self.assertTrue(self.p.follow(None, None))
|
||||
with patch.object(Postgresql, 'check_recovery_conf', Mock(return_value=True)):
|
||||
self.assertTrue(self.p.follow(None, None)) # nothing to do, recovery.conf has good primary_conninfo
|
||||
|
||||
self.p.follow(self.me, self.me) # follow is called when the node is holding leader lock
|
||||
|
||||
with patch.object(Postgresql, 'restart', Mock(return_value=False)):
|
||||
self.p.set_role('replica')
|
||||
self.p.follow(None, None) # restart without rewind
|
||||
self.p.set_role('master')
|
||||
|
||||
with patch.object(Postgresql, 'stop', Mock(return_value=False)):
|
||||
self.p.follow(self.leader, self.leader) # failed to stop postgres
|
||||
|
||||
self.p.follow(self.leader, None) # Leader unknown, can not rewind
|
||||
|
||||
self.p.follow(self.leader, self.leader) # "leader" is not accessible or is_in_recovery
|
||||
|
||||
with patch.object(Postgresql, 'checkpoint', Mock(return_value=None)):
|
||||
self.p.follow(self.leader, self.leader)
|
||||
self.p.set_role('master')
|
||||
mock_pg_rewind.return_value = True
|
||||
self.p.follow(self.leader, self.leader)
|
||||
|
||||
self.assertTrue(self.p.follow(None, None)) # check_recovery_conf...
|
||||
|
||||
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
|
||||
def test_can_rewind(self):
|
||||
with mock.patch('subprocess.call', MagicMock(return_value=1)):
|
||||
with patch('subprocess.call', MagicMock(return_value=1)):
|
||||
self.assertFalse(self.p.can_rewind)
|
||||
with mock.patch('subprocess.call', side_effect=OSError):
|
||||
with patch('subprocess.call', side_effect=OSError):
|
||||
self.assertFalse(self.p.can_rewind)
|
||||
tmp = self.p.controldata
|
||||
self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
|
||||
self.assertTrue(self.p.can_rewind)
|
||||
self.p.controldata = tmp
|
||||
with patch.object(Postgresql, 'controldata', Mock(return_value={'wal_log_hints setting': 'on'})):
|
||||
self.assertTrue(self.p.can_rewind)
|
||||
self.p.config['use_pg_rewind'] = False
|
||||
self.assertFalse(self.p.can_rewind)
|
||||
|
||||
@patch('time.sleep', Mock())
|
||||
def test_create_replica(self):
|
||||
|
||||
+13
-1
@@ -3,9 +3,10 @@ import unittest
|
||||
|
||||
from kazoo.client import KazooState
|
||||
from kazoo.exceptions import NoNodeError, NodeExistsError
|
||||
from kazoo.handlers.threading import SequentialThreadingHandler
|
||||
from kazoo.protocol.states import ZnodeStat
|
||||
from mock import Mock, patch
|
||||
from patroni.dcs.zookeeper import Leader, ZooKeeper, ZooKeeperError
|
||||
from patroni.dcs.zookeeper import Leader, PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
|
||||
|
||||
|
||||
class MockKazooClient(Mock):
|
||||
@@ -92,6 +93,17 @@ class MockKazooClient(Mock):
|
||||
raise NoNodeError
|
||||
|
||||
|
||||
class TestPatroniSequentialThreadingHandler(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.handler = PatroniSequentialThreadingHandler(10)
|
||||
|
||||
@patch.object(SequentialThreadingHandler, 'create_connection', Mock())
|
||||
def test_create_connection(self):
|
||||
self.assertIsNotNone(self.handler.create_connection(()))
|
||||
self.assertIsNotNone(self.handler.create_connection((), 40))
|
||||
|
||||
|
||||
class TestZooKeeper(unittest.TestCase):
|
||||
|
||||
@patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
|
||||
|
||||
Reference in New Issue
Block a user