diff --git a/Dockerfile b/Dockerfile index 421e775f..f01c12cb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,36 +3,42 @@ FROM ubuntu:16.04 MAINTAINER Feike Steenbergen -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 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 00000000..2dbb7cfa --- /dev/null +++ b/docker-compose.yml @@ -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 diff --git a/docker/dev_patroni_cluster.sh b/docker/dev_patroni_cluster.sh index dcc18f87..9f21b128 100755 --- a/docker/dev_patroni_cluster.sh +++ b/docker/dev_patroni_cluster.sh @@ -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 diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index c329d2bb..5e5dac29 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -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 diff --git a/docker/patroni-secrets.env b/docker/patroni-secrets.env new file mode 100644 index 00000000..7c0f840e --- /dev/null +++ b/docker/patroni-secrets.env @@ -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 diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index c0bfa03c..3a13194e 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -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! diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 88be5031..4728dce8 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -69,6 +69,9 @@ PostgreSQL - **pgpass**: path to the `.pgpass `__ 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 diff --git a/extras/README.md b/extras/README.md new file mode 100644 index 00000000..a4c3d59a --- /dev/null +++ b/extras/README.md @@ -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. diff --git a/extras/confd/conf.d/haproxy.toml b/extras/confd/conf.d/haproxy.toml new file mode 100644 index 00000000..f9f04fbc --- /dev/null +++ b/extras/confd/conf.d/haproxy.toml @@ -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/", +] diff --git a/extras/confd/templates/haproxy.tmpl b/extras/confd/templates/haproxy.tmpl new file mode 100644 index 00000000..4309ed0c --- /dev/null +++ b/extras/confd/templates/haproxy.tmpl @@ -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}} diff --git a/features/environment.py b/features/environment.py index 3fc52a2d..be9c2f9b 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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 diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 934e8b76..8f5c415e 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -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 diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index dd260aab..fc4466e8 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -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}') diff --git a/patroni/__init__.py b/patroni/__init__.py index c6d95259..1403dacf 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -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: diff --git a/patroni/api.py b/patroni/api.py index 207dd746..fb9fae99 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -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) diff --git a/patroni/config.py b/patroni/config.py index ce81de17..f09cfca4 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -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()): diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index c1c31174..39049792 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -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. diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 7b2b7477..e6169271 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -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): diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index dae08ffb..be6fe45a 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -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): diff --git a/patroni/ha.py b/patroni/ha.py index 1da78027..f45a4585 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -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(): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a0424e3a..602cc5e5 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -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 (5540277D.8020309@iki.fi) - 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 (5540277D.8020309@iki.fi) + 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 diff --git a/patroni/utils.py b/patroni/utils.py index 9c7a9e9d..8b2389f7 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -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 = { diff --git a/patroni/version.py b/patroni/version.py index 77ac3b7d..7e49527e 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -1 +1 @@ -__version__ = '0.90' +__version__ = '1.0' diff --git a/tests/test_config.py b/tests/test_config.py index a8a5a6c4..94e5f5fc 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -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 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index ebdf6aa2..0fb6ca9c 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -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) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 0ef36441..bb3b17c6 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -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): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 01555d8e..a406ec82 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -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)