gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-09-01 09:09:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

Browse Source
This commit is contained in:
Alexander Kukushkin
2023-05-23 09:59:16 +02:00
parent dbfe844a6f 2f5bcbd877
commit f298921315
36 changed files with 1268 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/ISSUE_TEMPLATE/bug_report.yml
+1 -1
View File
@@ -6,7 +6,7 @@ body:
- type: markdown
attributes:
value: |
If you have a question please post it on channel [#patroni](https://postgresteam.slack.com/archives/C9XPYG92A) in the [PostgreSQL Slack](https://join.slack.com/t/postgresteam/shared_invite/zt-1qj14i9sj-E9WqIFlvcOiHsEk2yFEMjA).
If you have a question please post it on channel [#patroni](https://postgresteam.slack.com/archives/C9XPYG92A) in the [PostgreSQL Slack](https://pgtreats.info/slack-invite).
Before reporting a bug please make sure to **reproduce it with the latest Patroni version**!
Please fill the form below and provide as much information as possible.
Not doing so may result in your bug not being addressed in a timely manner.
.github/ISSUE_TEMPLATE/config.yml
+1 -1
View File
@@ -1,5 +1,5 @@
blank_issues_enabled: false
contact_links:
- name: Question
url: https://join.slack.com/t/postgresteam/shared_invite/zt-1qj14i9sj-E9WqIFlvcOiHsEk2yFEMjA
url: https://pgtreats.info/slack-invite
about: "Please ask questions on channel #patroni in the PostgreSQL Slack"
.github/workflows/tests.yaml
+19 -2
View File
@@ -116,8 +116,8 @@ jobs:
sudo apt-get install -y wget ca-certificates gnupg debian-archive-keyring apt-transport-https
sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
sudo sh -c 'wget -qO - https://www.postgresql.org/media/keys/ACCC4CF8.asc | gpg --dearmor > /etc/apt/trusted.gpg.d/apt.postgresql.org.gpg'
sudo sh -c 'echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://repos.citusdata.com/community/ubuntu/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list'
sudo sh -c 'wget -qO - https://repos.citusdata.com/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg'
sudo sh -c 'echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/ubuntu/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list'
sudo sh -c 'wget -qO - https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg'
if: matrix.os == 'ubuntu'
- name: Install dependencies
run: python .github/workflows/install_deps.py
@@ -157,3 +157,20 @@ jobs:
steps:
- run: bash <(curl -Ls https://coverage.codacy.com/get.sh) final
if: ${{ env.SECRETS_AVAILABLE == 'true' }}
pyright:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python 3.11
uses: actions/setup-python@v4
with:
python-version: 3.11
- name: Install dependencies
run: python -m pip install -r requirements.txt psycopg2-binary psycopg
- uses: jakebailey/pyright-action@v1
with:
version: 1.1.309
Dockerfile
+15 -3
View File
@@ -53,14 +53,26 @@ RUN set -ex \
&& curl -sL "https://github.com/coreos/etcd/releases/download/v$ETCDVERSION/etcd-v$ETCDVERSION-linux-$(dpkg --print-architecture).tar.gz" \
| tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl \
\
# Download confd
&& curl -sL "https://github.com/kelseyhightower/confd/releases/download/v$CONFDVERSION/confd-$CONFDVERSION-linux-$(dpkg --print-architecture)" \
> /usr/local/bin/confd && chmod +x /usr/local/bin/confd \
&& if [ $(dpkg --print-architecture) = 'arm64' ]; then \
# Build confd
apt-get install -y git make \
&& curl -sL https://go.dev/dl/go1.20.4.linux-arm64.tar.gz | tar xz -C /usr/local go \
&& export GOROOT=/usr/local/go && export PATH=$PATH:$GOROOT/bin \
&& git clone --recurse-submodules https://github.com/kelseyhightower/confd.git \
&& make -C confd \
&& cp confd/bin/confd /usr/local/bin/confd \
&& rm -rf /confd /usr/local/go; \
else \
# Download confd
curl -sL "https://github.com/kelseyhightower/confd/releases/download/v$CONFDVERSION/confd-$CONFDVERSION-linux-$(dpkg --print-architecture)" \
> /usr/local/bin/confd && chmod +x /usr/local/bin/confd; \
fi \
\
# Clean up all useless packages and some files
&& apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
git make \
&& apt-get autoremove -y \
&& apt-get clean -y \
&& rm -rf /var/lib/apt/lists/* \
Dockerfile.citus
+33 -6
View File
@@ -20,14 +20,28 @@ RUN set -ex \
&& export DEBIAN_FRONTEND=noninteractive \
&& echo 'APT::Install-Recommends "0";\nAPT::Install-Suggests "0";' > /etc/apt/apt.conf.d/01norecommend \
&& apt-get update -y \
# postgres:10 is based on debian, which has the patroni package. We will install all required dependencies
# postgres:PG_MAJOR is based on debian, which has the patroni package. We will install all required dependencies
&& apt-cache depends patroni | sed -n -e 's/.*Depends: \(python3-.\+\)$/\1/p' \
| grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
| xargs apt-get install -y vim curl less jq locales haproxy sudo \
python3-etcd python3-kazoo python3-pip busybox \
net-tools iputils-ping --fix-missing \
&& curl https://install.citusdata.com/community/deb.sh | bash \
&& apt-get -y install postgresql-$PG_MAJOR-citus-11.2 \
&& if [ $(dpkg --print-architecture) = 'arm64' ]; then \
apt-get install -y postgresql-server-dev-$PG_MAJOR \
git gcc make autoconf \
libc6-dev flex libcurl4-gnutls-dev \
libicu-dev libkrb5-dev liblz4-dev \
libpam0g-dev libreadline-dev libselinux1-dev\
libssl-dev libxslt1-dev libzstd-dev uuid-dev \
&& git clone -b "main" https://github.com/citusdata/citus.git \
&& MAKEFLAGS="-j $(grep -c ^processor /proc/cpuinfo)" \
&& cd citus && ./configure && make install && cd ../ && rm -rf /citus; \
else \
echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
&& curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
&& apt-get update -y \
&& apt-get -y install postgresql-$PG_MAJOR-citus-11.3 \
fi \
&& pip3 install dumb-init \
\
# Cleanup all locales but en_US.UTF-8
@@ -55,9 +69,19 @@ RUN set -ex \
&& curl -sL https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-$(dpkg --print-architecture).tar.gz \
| tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl \
\
# Download confd
&& curl -sL https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-$(dpkg --print-architecture) \
> /usr/local/bin/confd && chmod +x /usr/local/bin/confd \
&& if [ $(dpkg --print-architecture) = 'arm64' ]; then \
# Build confd
curl -sL https://go.dev/dl/go1.20.4.linux-arm64.tar.gz | tar xz -C /usr/local go \
&& export GOROOT=/usr/local/go && export PATH=$PATH:$GOROOT/bin \
&& git clone --recurse-submodules https://github.com/kelseyhightower/confd.git \
&& make -C confd \
&& cp confd/bin/confd /usr/local/bin/confd \
&& rm -rf /confd /usr/local/go; \
else \
# Download confd
curl -sL "https://github.com/kelseyhightower/confd/releases/download/v$CONFDVERSION/confd-$CONFDVERSION-linux-$(dpkg --print-architecture)" \
> /usr/local/bin/confd && chmod +x /usr/local/bin/confd; \
fi \
# Prepare client cert for HAProxy
&& cat /etc/ssl/private/ssl-cert-snakeoil.key /etc/ssl/certs/ssl-cert-snakeoil.pem > /etc/ssl/private/ssl-cert-snakeoil.crt \
\
@@ -65,6 +89,9 @@ RUN set -ex \
&& apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
postgresql-server-dev-$PG_MAJOR git gcc make autoconf \
libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \
libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \
&& apt-get autoremove -y \
&& apt-get clean -y \
&& rm -rf /var/lib/apt/lists/* \
README.rst
+1 -1
View File
@@ -49,7 +49,7 @@ We report new releases information `here <https://github.com/zalando/patroni/rel
Community
=========
There are two places to connect with the Patroni community: `on github <https://github.com/zalando/patroni>`__, via Issues and PRs, and on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://join.slack.com/t/postgresteam/shared_invite/zt-1qj14i9sj-E9WqIFlvcOiHsEk2yFEMjA>`__. If you're using Patroni, or just interested, please join us.
There are two places to connect with the Patroni community: `on github <https://github.com/zalando/patroni>`__, via Issues and PRs, and on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://pgtreats.info/slack-invite>`__. If you're using Patroni, or just interested, please join us.
===================================
Technical Requirements/Installation
docker-compose-citus.yml
+10 -9
View File
@@ -16,7 +16,7 @@ networks:
services:
etcd1: &etcd
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
environment:
ETCDCTL_API: 3
@@ -25,6 +25,7 @@ services:
ETCD_INITIAL_CLUSTER: etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
ETCD_INITIAL_CLUSTER_STATE: new
ETCD_INITIAL_CLUSTER_TOKEN: tutorial
ETCD_UNSUPPORTED_ARCH: arm64
container_name: demo-etcd1
hostname: etcd1
command: etcd -name etcd1 -initial-advertise-peer-urls http://etcd1:2380
@@ -42,7 +43,7 @@ services:
command: etcd -name etcd3 -initial-advertise-peer-urls http://etcd3:2380
haproxy:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: haproxy
@@ -64,7 +65,7 @@ services:
PGSSLROOTCERT: /etc/ssl/certs/ssl-cert-snakeoil.pem
coord1:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: coord1
@@ -75,7 +76,7 @@ services:
PATRONI_CITUS_GROUP: 0
coord2:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: coord2
@@ -85,7 +86,7 @@ services:
PATRONI_NAME: coord2
coord3:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: coord3
@@ -96,7 +97,7 @@ services:
work1-1:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: work1-1
@@ -107,7 +108,7 @@ services:
PATRONI_CITUS_GROUP: 1
work1-2:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: work1-2
@@ -118,7 +119,7 @@ services:
work2-1:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: work2-1
@@ -129,7 +130,7 @@ services:
PATRONI_CITUS_GROUP: 2
work2-2:
image: patroni-citus
image: ${PATRONI_TEST_IMAGE:-patroni-citus}
networks: [ demo ]
env_file: docker/patroni.env
hostname: work2-2
docker-compose.yml
+6 -5
View File
@@ -14,7 +14,7 @@ networks:
services:
etcd1: &etcd
image: patroni
image: ${PATRONI_TEST_IMAGE:-patroni}
networks: [ demo ]
environment:
ETCD_LISTEN_PEER_URLS: http://0.0.0.0:2380
@@ -22,6 +22,7 @@ services:
ETCD_INITIAL_CLUSTER: etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
ETCD_INITIAL_CLUSTER_STATE: new
ETCD_INITIAL_CLUSTER_TOKEN: tutorial
ETCD_UNSUPPORTED_ARCH: arm64
container_name: demo-etcd1
hostname: etcd1
command: etcd -name etcd1 -initial-advertise-peer-urls http://etcd1:2380
@@ -39,7 +40,7 @@ services:
command: etcd -name etcd3 -initial-advertise-peer-urls http://etcd3:2380
haproxy:
image: patroni
image: ${PATRONI_TEST_IMAGE:-patroni}
networks: [ demo ]
env_file: docker/patroni.env
hostname: haproxy
@@ -54,7 +55,7 @@ services:
PATRONI_SCOPE: demo
patroni1:
image: patroni
image: ${PATRONI_TEST_IMAGE:-patroni}
networks: [ demo ]
env_file: docker/patroni.env
hostname: patroni1
@@ -64,7 +65,7 @@ services:
PATRONI_NAME: patroni1
patroni2:
image: patroni
image: ${PATRONI_TEST_IMAGE:-patroni}
networks: [ demo ]
env_file: docker/patroni.env
hostname: patroni2
@@ -74,7 +75,7 @@ services:
PATRONI_NAME: patroni2
patroni3:
image: patroni
image: ${PATRONI_TEST_IMAGE:-patroni}
networks: [ demo ]
env_file: docker/patroni.env
hostname: patroni3
docs/CONTRIBUTING.rst
+1 -1
View File
@@ -8,7 +8,7 @@ Wanna contribute to Patroni? Yay - here is how!
Chatting
--------
Just want to chat with other Patroni users? Looking for interactive troubleshooting help? Join us on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://join.slack.com/t/postgresteam/shared_invite/zt-1qj14i9sj-E9WqIFlvcOiHsEk2yFEMjA>`__.
Just want to chat with other Patroni users? Looking for interactive troubleshooting help? Join us on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://pgtreats.info/slack-invite>`__.
Running tests
-------------
docs/yaml_configuration.rst
+1 -1
View File
@@ -262,7 +262,7 @@ PostgreSQL
own config item. See :ref:`custom replica creation methods documentation <custom_replica_creation>` for further explanation.
- **data\_dir**: The location of the Postgres data directory, either :ref:`existing <existing_data>` or to be initialized by Patroni.
- **config\_dir**: The location of the Postgres configuration directory, defaults to the data directory. Must be writable by Patroni.
- **bin\_dir**: Path to PostgreSQL binaries (pg_ctl, pg_rewind, pg_basebackup, postgres). The default value is an empty string meaning that PATH environment variable will be used to find the executables.
- **bin\_dir**: (optional) Path to PostgreSQL binaries (pg_ctl, pg_rewind, pg_basebackup, postgres). If not provided or is an empty string, PATH environment variable will be used to find the executables.
- **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
- **use\_unix\_socket**: specifies that Patroni should prefer to use unix sockets to connect to the cluster. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
- **use\_unix\_socket\_repl**: specifies that Patroni should prefer to use unix sockets for replication user cluster connection. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
features/environment.py
+3 -1
View File
@@ -7,6 +7,7 @@ import psutil
import re
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
@@ -1060,10 +1061,11 @@ def before_all(context):
try:
with open(os.devnull, 'w') as null:
ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni',
'--addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile,
'-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile,
'-out', context.certfile], stdout=null, stderr=null)
if ret != 0:
raise Exception
os.chmod(context.keyfile, stat.S_IWRITE | stat.S_IREAD)
except Exception:
context.keyfile = context.certfile = None
kubernetes/Dockerfile.citus
+4 -2
View File
@@ -10,8 +10,10 @@ RUN export DEBIAN_FRONTEND=noninteractive \
| xargs apt-get install -y busybox vim-tiny curl jq less locales git python3-pip python3-wheel \
## Make sure we have a en_US.UTF-8 locale available
&& localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
&& curl https://install.citusdata.com/community/deb.sh | bash \
&& apt-get -y install postgresql-15-citus-11.2 \
&& echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
&& curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
&& apt-get update -y \
&& apt-get -y install postgresql-$PG_MAJOR-citus-11.3 \
&& pip3 install setuptools \
&& pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
&& PGHOME=/home/postgres \
patroni/api.py
+784 -139
View File
File diff suppressed because it is too large Load Diff
patroni/config.py
+8 -7
View File
@@ -7,7 +7,7 @@ import yaml
from collections import defaultdict
from copy import deepcopy
from typing import Any, Callable, Collection, Dict, List, Optional, Union
from typing import Any, Callable, Collection, Dict, List, Optional, Union, TYPE_CHECKING
from . import PATRONI_ENV_PREFIX
from .collections import CaseInsensitiveDict
@@ -156,7 +156,7 @@ def get_global_config(cluster: Union[Cluster, None], default: Optional[Dict[str,
:returns: :class:`GlobalConfig` object
"""
# Try to protect from the case when DCS was wiped out
if cluster and cluster.config and cluster.config.modify_index:
if cluster and cluster.config and cluster.config.modify_version:
config = cluster.config.data
else:
config = default or {}
@@ -207,7 +207,7 @@ class Config(object):
def __init__(self, configfile: str,
validator: Optional[Callable[[Dict[str, Any]], List[str]]] = default_validator) -> None:
self._modify_index = -1
self._modify_version = -1
self._dynamic_configuration = {}
self.__environment_configuration = self._build_environment_configuration()
@@ -262,7 +262,8 @@ class Config(object):
def _load_config_file(self) -> Dict[str, Any]:
"""Loads config.yaml from filesystem and applies some values which were set via ENV"""
assert self._config_file is not None
if TYPE_CHECKING: # pragma: no cover
assert self._config_file is not None
config = self._load_config_path(self._config_file)
patch_config(config, self.__environment_configuration)
return config
@@ -301,9 +302,9 @@ class Config(object):
# configuration could be either ClusterConfig or dict
def set_dynamic_configuration(self, configuration: Union[ClusterConfig, Dict[str, Any]]) -> bool:
if isinstance(configuration, ClusterConfig):
if self._modify_index == configuration.modify_index:
return False # If the index didn't changed there is nothing to do
self._modify_index = configuration.modify_index
if self._modify_version == configuration.modify_version:
return False # If the version didn't changed there is nothing to do
self._modify_version = configuration.modify_version
configuration = configuration.data
if not deep_compare(self._dynamic_configuration, configuration):
patroni/ctl.py
+11 -9
View File
@@ -32,9 +32,9 @@ if TYPE_CHECKING: # pragma: no cover
from psycopg2 import cursor
try:
from ydiff import markup_to_pager, PatchStream
from ydiff import markup_to_pager, PatchStream # pyright: ignore [reportMissingModuleSource]
except ImportError: # pragma: no cover
from cdiff import markup_to_pager, PatchStream
from cdiff import markup_to_pager, PatchStream # pyright: ignore [reportMissingModuleSource]
from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member
from .exceptions import PatroniException
@@ -812,7 +812,8 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
r = None
try:
member = cluster.leader.member if cluster.leader else candidate and cluster.get_member(candidate, False)
assert isinstance(member, Member)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(member, Member)
r = request_patroni(member, 'post', action, failover_value)
# probably old patroni, which doesn't support switchover yet
@@ -1052,7 +1053,7 @@ def flush(obj: Dict[str, Any], cluster_name: str, group: Optional[int],
logging.warning('Failing over to DCS')
click.echo('{0} Could not find any accessible member of cluster {1}'.format(timestamp(), cluster_name))
dcs.manual_failover('', '', index=failover.index)
dcs.manual_failover('', '', version=failover.version)
def wait_until_pause_is_applied(dcs: AbstractDCS, paused: bool, old_cluster: Cluster) -> None:
@@ -1060,7 +1061,7 @@ def wait_until_pause_is_applied(dcs: AbstractDCS, paused: bool, old_cluster: Clu
config = get_global_config(old_cluster)
click.echo("'{0}' request sent, waiting until it is recognized by all nodes".format(paused and 'pause' or 'resume'))
old = {m.name: m.index for m in old_cluster.members if m.api_url}
old = {m.name: m.version for m in old_cluster.members if m.api_url}
loop_wait = config.get('loop_wait') or dcs.loop_wait
cluster = None
@@ -1072,7 +1073,7 @@ def wait_until_pause_is_applied(dcs: AbstractDCS, paused: bool, old_cluster: Clu
if TYPE_CHECKING: # pragma: no cover
assert cluster is not None
remaining = [m.name for m in cluster.members if m.data.get('pause', False) != paused
and m.name in old and old[m.name] != m.index]
and m.name in old and old[m.name] != m.version]
if remaining:
return click.echo("{0} members didn't recognized pause state after {1} seconds"
.format(', '.join(remaining), loop_wait))
@@ -1169,7 +1170,7 @@ def show_diff(before_editing: str, after_editing: str) -> None:
(
os.path.basename(p)
for p in (os.environ.get('PAGER'), "less", "more")
if p is not None and shutil.which(p)
if p is not None and bool(shutil.which(p))
),
None,
)
@@ -1347,7 +1348,7 @@ def edit_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int],
return
if force or click.confirm('Apply these changes?'):
if not dcs.set_config_value(json.dumps(changed_data), cluster.config.index):
if not dcs.set_config_value(json.dumps(changed_data), cluster.config.version):
raise PatroniCtlException("Config modification aborted due to concurrent changes")
click.echo("Configuration changed")
@@ -1396,7 +1397,8 @@ def version(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member
@click.pass_obj
def history(obj: Dict[str, Any], cluster_name: str, group: Optional[int], fmt: str) -> None:
cluster = get_dcs(obj, cluster_name, group).get_cluster()
history: List[List[Any]] = list(map(list, cluster.history and cluster.history.lines or []))
cluster_history = cluster.history.lines if cluster.history else []
history: List[List[Any]] = list(map(list, cluster_history))
table_header_row = ['TL', 'LSN', 'Reason', 'Timestamp', 'New Leader']
for line in history:
if len(line) < len(table_header_row):
patroni/dcs/__init__.py
+47 -45
View File
@@ -126,7 +126,7 @@ _Session = Union[int, float, str, None]
class Member(NamedTuple):
"""Immutable object (namedtuple) which represents single member of PostgreSQL cluster.
Consists of the following fields:
:param index: modification index of a given member key in a Configuration Store
:param version: modification version of a given member key in a Configuration Store
:param name: name of PostgreSQL cluster member
:param session: either session id or just ttl in seconds
:param data: arbitrary data i.e. conn_url, api_url, xlog location, state, role, tags, etc...
@@ -135,18 +135,18 @@ class Member(NamedTuple):
conn_url: connection string containing host, user and password which could be used to access this member.
api_url: REST API url of patroni instance
"""
index: _Version
version: _Version
name: str
session: _Session
data: Dict[str, Any]
@staticmethod
def from_node(index: _Version, name: str, session: _Session, value: str) -> 'Member':
def from_node(version: _Version, name: str, session: _Session, value: str) -> 'Member':
"""
>>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None
True
>>> Member.from_node(-1, '', '', '{')
Member(index=-1, name='', session='', data={})
Member(version=-1, name='', session='', data={})
"""
if value.startswith('postgres'):
conn_url, api_url = parse_connection_string(value)
@@ -157,7 +157,7 @@ class Member(NamedTuple):
assert isinstance(data, dict)
except (AssertionError, TypeError, ValueError):
data: Dict[str, Any] = {}
return Member(index, name, session, data)
return Member(version, name, session, data)
@property
def conn_url(self) -> Optional[str]:
@@ -229,7 +229,7 @@ class Member(NamedTuple):
return self.state == 'running'
@property
def version(self) -> Optional[Tuple[int, ...]]:
def patroni_version(self) -> Optional[Tuple[int, ...]]:
version = self.data.get('version')
if version:
try:
@@ -240,7 +240,9 @@ class Member(NamedTuple):
class RemoteMember(Member):
"""Represents a remote member (typically a primary) for a standby cluster"""
def __new__(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember':
@classmethod
def from_name_and_data(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember':
return super(RemoteMember, cls).__new__(cls, -1, name, None, data)
@staticmethod
@@ -261,11 +263,11 @@ class Leader(NamedTuple):
"""Immutable object (namedtuple) which represents leader key.
Consists of the following fields:
:param index: modification index of a leader key in a Configuration Store
:param version: modification version of a leader key in a Configuration Store
:param session: either session id or just ttl in seconds
:param member: reference to a `Member` object which represents current leader (see `Cluster.members`)
"""
index: _Version
version: _Version
session: _Session
member: Member
@@ -294,7 +296,7 @@ class Leader(NamedTuple):
>>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote
"""
version = self.member.version
version = self.member.patroni_version
# 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false
if version and version > (1, 5, 6):
return self.data.get('role') in ('master', 'primary') and 'checkpoint_after_promote' not in self.data
@@ -321,13 +323,13 @@ class Failover(NamedTuple):
>>> 'abc' in Failover.from_node(1, 'abc:def')
True
"""
index: _Version
version: _Version
leader: Optional[str]
candidate: Optional[str]
scheduled_at: Optional[datetime.datetime]
@staticmethod
def from_node(index: _Version, value: Union[str, Dict[str, str]]) -> 'Failover':
def from_node(version: _Version, value: Union[str, Dict[str, str]]) -> 'Failover':
if isinstance(value, dict):
data: Dict[str, Any] = value
elif value:
@@ -340,26 +342,26 @@ class Failover(NamedTuple):
t = [a.strip() for a in value.split(':')]
leader = t[0]
candidate = t[1] if len(t) > 1 else None
return Failover(index, leader, candidate, None)
return Failover(version, leader, candidate, None)
else:
data = {}
if data.get('scheduled_at'):
data['scheduled_at'] = dateutil.parser.parse(data['scheduled_at'])
return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at'))
return Failover(version, data.get('leader'), data.get('member'), data.get('scheduled_at'))
def __len__(self) -> int:
return int(bool(self.leader)) + int(bool(self.candidate))
class ClusterConfig(NamedTuple):
index: _Version
version: _Version
data: Dict[str, Any]
modify_index: _Version
modify_version: _Version
@staticmethod
def from_node(index: _Version, value: str, modify_index: Optional[_Version] = None) -> 'ClusterConfig':
def from_node(version: _Version, value: str, modify_version: Optional[_Version] = None) -> 'ClusterConfig':
"""
>>> ClusterConfig.from_node(1, '{') is None
False
@@ -370,8 +372,8 @@ class ClusterConfig(NamedTuple):
assert isinstance(data, dict)
except (AssertionError, TypeError, ValueError):
data: Dict[str, Any] = {}
modify_index = 0
return ClusterConfig(index, data, index if modify_index is None else modify_index)
modify_version = 0
return ClusterConfig(version, data, version if modify_version is None else modify_version)
@property
def permanent_slots(self) -> Dict[str, Any]:
@@ -390,19 +392,19 @@ class ClusterConfig(NamedTuple):
class SyncState(NamedTuple):
"""Immutable object (namedtuple) which represents last observed synhcronous replication state
:param index: modification index of a synchronization key in a Configuration Store
:param version: modification version of a synchronization key in a Configuration Store
:param leader: reference to member that was leader
:param sync_standby: synchronous standby list (comma delimited) which are last synchronized to leader
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
"""
index: Optional[_Version]
version: Optional[_Version]
leader: Optional[str]
sync_standby: Optional[str]
quorum: int
@staticmethod
def from_node(index: Optional[_Version], value: Union[str, Dict[str, Any], None]) -> 'SyncState':
def from_node(version: Optional[_Version], value: Union[str, Dict[str, Any], None]) -> 'SyncState':
"""
>>> SyncState.from_node(1, None).leader is None
True
@@ -423,13 +425,13 @@ class SyncState(NamedTuple):
assert isinstance(value, dict)
leader = value.get('leader')
quorum = value.get('quorum')
return SyncState(index, leader, value.get('sync_standby'), int(quorum) if leader and quorum else 0)
return SyncState(version, leader, value.get('sync_standby'), int(quorum) if leader and quorum else 0)
except (AssertionError, TypeError, ValueError):
return SyncState.empty(index)
return SyncState.empty(version)
@staticmethod
def empty(index: Optional[_Version] = None) -> 'SyncState':
return SyncState(index, None, None, 0)
def empty(version: Optional[_Version] = None) -> 'SyncState':
return SyncState(version, None, None, 0)
@property
def is_empty(self) -> bool:
@@ -494,12 +496,12 @@ _HistoryTuple = Union[Tuple[int, int, str], Tuple[int, int, str, str], Tuple[int
class TimelineHistory(NamedTuple):
"""Object representing timeline history file"""
index: _Version
version: _Version
value: Any
lines: List[_HistoryTuple]
@staticmethod
def from_node(index: _Version, value: str) -> 'TimelineHistory':
def from_node(version: _Version, value: str) -> 'TimelineHistory':
"""
>>> h = TimelineHistory.from_node(1, 2)
>>> h.lines
@@ -510,7 +512,7 @@ class TimelineHistory(NamedTuple):
assert isinstance(lines, list)
except (AssertionError, TypeError, ValueError):
lines: List[_HistoryTuple] = []
return TimelineHistory(index, value, lines)
return TimelineHistory(version, value, lines)
class Cluster(NamedTuple):
@@ -547,7 +549,7 @@ class Cluster(NamedTuple):
def is_empty(self):
return self.initialize is None and self.config is None and self.leader is None and self.last_lsn == 0\
and self.members == [] and self.failover is None and self.sync.index is None\
and self.members == [] and self.failover is None and self.sync.version is None\
and self.history is None and self.slots is None and self.failsafe is None and self.workers == {}
def __len__(self) -> int:
@@ -711,7 +713,7 @@ class Cluster(NamedTuple):
@property
def min_version(self) -> Optional[Tuple[int, ...]]:
return next(iter(sorted(m.version for m in self.members if m.version)), None)
return next(iter(sorted(m.patroni_version for m in self.members if m.patroni_version)), None)
class ReturnFalseException(Exception):
@@ -880,7 +882,8 @@ class AbstractDCS(abc.ABC):
if path is None:
path = self.client_path('')
cluster = self._load_cluster(path, self._cluster_loader)
assert isinstance(cluster, Cluster)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(cluster, Cluster)
return cluster
def is_citus_coordinator(self) -> bool:
@@ -897,7 +900,6 @@ class AbstractDCS(abc.ABC):
if isinstance(groups, Cluster): # Zookeeper could return a cached version
cluster = groups
else:
assert isinstance(groups, dict)
cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
cluster.workers.update(groups)
return cluster
@@ -1016,11 +1018,11 @@ class AbstractDCS(abc.ABC):
process requests (hopefuly temporary), the ~DCSError exception should be raised"""
@abc.abstractmethod
def set_failover_value(self, value: str, index: Optional[Any] = None) -> bool:
def set_failover_value(self, value: str, version: Optional[Any] = None) -> bool:
"""Create or update `/failover` key"""
def manual_failover(self, leader: Optional[str], candidate: Optional[str],
scheduled_at: Optional[datetime.datetime] = None, index: Optional[Any] = None) -> bool:
scheduled_at: Optional[datetime.datetime] = None, version: Optional[Any] = None) -> bool:
failover_value = {}
if leader:
failover_value['leader'] = leader
@@ -1030,10 +1032,10 @@ class AbstractDCS(abc.ABC):
if scheduled_at:
failover_value['scheduled_at'] = scheduled_at.isoformat()
return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), index)
return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), version)
@abc.abstractmethod
def set_config_value(self, value: str, index: Optional[Any] = None) -> bool:
def set_config_value(self, value: str, version: Optional[Any] = None) -> bool:
"""Create or update `/config` key"""
@abc.abstractmethod
@@ -1101,18 +1103,18 @@ class AbstractDCS(abc.ABC):
'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None}
def write_sync_state(self, leader: Optional[str], sync_standby: Optional[Collection[str]],
quorum: Optional[int], index: Optional[Any] = None) -> Optional[SyncState]:
quorum: Optional[int], version: Optional[Any] = None) -> Optional[SyncState]:
"""Write the new synchronous state to DCS.
Calls :func:`sync_state` method to build a dict and than calls DCS specific :func:`set_sync_state_value` method.
:param leader: name of the leader node that manages /sync key
:param sync_standby: collection of currently known synchronous standby node names
:param index: for conditional update of the key/object
:param version: for conditional update of the key/object
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
:returns: the new :class:`SyncState` object or None
"""
sync_value = self.sync_state(leader, sync_standby, quorum)
ret = self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index)
ret = self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), version)
if not isinstance(ret, bool):
return SyncState.from_node(ret, sync_value)
@@ -1121,23 +1123,23 @@ class AbstractDCS(abc.ABC):
""""""
@abc.abstractmethod
def set_sync_state_value(self, value: str, index: Optional[Any] = None) -> Union[Any, bool]:
def set_sync_state_value(self, value: str, version: Optional[Any] = None) -> Union[Any, bool]:
"""Set synchronous state in DCS, should be implemented in the child class.
:param value: the new value of /sync key
:param index: for conditional update of the key/object
:param version: for conditional update of the key/object
:returns: version of the new object or `False` in case of error
"""
@abc.abstractmethod
def delete_sync_state(self, index: Optional[Any] = None) -> bool:
def delete_sync_state(self, version: Optional[Any] = None) -> bool:
""""""
def watch(self, leader_index: Optional[Any], timeout: float) -> bool:
def watch(self, leader_version: Optional[Any], timeout: float) -> bool:
"""If the current node is a leader it should just sleep.
Any other node should watch for changes of leader key with a given timeout
:param leader_index: index of a leader key
:param leader_version: version of a leader key
:param timeout: timeout in seconds
:returns: `!True` if you would like to reschedule the next run of ha cycle"""
patroni/dcs/consul.py
+16 -15
View File
@@ -578,12 +578,12 @@ class Consul(AbstractDCS):
return self.attempt_to_acquire_leader()
@catch_consul_errors
def set_failover_value(self, value: str, index: Optional[int] = None) -> bool:
return self._client.kv.put(self.failover_path, value, cas=index)
def set_failover_value(self, value: str, version: Optional[int] = None) -> bool:
return self._client.kv.put(self.failover_path, value, cas=version)
@catch_consul_errors
def set_config_value(self, value: str, index: Optional[int] = None) -> bool:
return self._client.kv.put(self.config_path, value, cas=index)
def set_config_value(self, value: str, version: Optional[int] = None) -> bool:
return self._client.kv.put(self.config_path, value, cas=version)
@catch_consul_errors
def _write_leader_optime(self, last_lsn: str) -> bool:
@@ -622,7 +622,8 @@ class Consul(AbstractDCS):
raise ConsulError('update_leader timeout')
logger.warning('Recreating the leader key due to session mismatch')
if cluster and cluster.leader:
self._run_and_handle_exceptions(self._client.kv.delete, self.leader_path, cas=cluster.leader.index)
self._run_and_handle_exceptions(self._client.kv.delete, self.leader_path,
cas=cluster.leader.version)
retry.deadline = retry.stoptime - time.time()
if retry.deadline < 0.5:
@@ -653,14 +654,14 @@ class Consul(AbstractDCS):
def _delete_leader(self) -> bool:
cluster = self.cluster
if cluster and isinstance(cluster.leader, Leader) and\
cluster.leader.name == self._name and isinstance(cluster.leader.index, int):
return self._client.kv.delete(self.leader_path, cas=cluster.leader.index)
cluster.leader.name == self._name and isinstance(cluster.leader.version, int):
return self._client.kv.delete(self.leader_path, cas=cluster.leader.version)
return True
@catch_consul_errors
def set_sync_state_value(self, value: str, index: Optional[int] = None) -> Union[int, bool]:
def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]:
retry = self._retry.copy()
ret = retry(self._client.kv.put, self.sync_path, value, cas=index)
ret = retry(self._client.kv.put, self.sync_path, value, cas=version)
if ret: # We have no other choise, only read after write :(
retry.deadline = retry.stoptime - time.time()
if retry.deadline < 0.5:
@@ -671,21 +672,21 @@ class Consul(AbstractDCS):
return False
@catch_consul_errors
def delete_sync_state(self, index: Optional[int] = None) -> bool:
return self.retry(self._client.kv.delete, self.sync_path, cas=index)
def delete_sync_state(self, version: Optional[int] = None) -> bool:
return self.retry(self._client.kv.delete, self.sync_path, cas=version)
def watch(self, leader_index: Optional[int], timeout: float) -> bool:
def watch(self, leader_version: Optional[int], timeout: float) -> bool:
self._last_session_refresh = 0
if self.__do_not_watch:
self.__do_not_watch = False
return True
if leader_index:
if leader_version:
end_time = time.time() + timeout
while timeout >= 1:
try:
idx, _ = self._client.kv.get(self.leader_path, index=leader_index, wait=str(timeout) + 's')
return str(idx) != str(leader_index)
idx, _ = self._client.kv.get(self.leader_path, index=leader_version, wait=str(timeout) + 's')
return str(idx) != str(leader_version)
except (ConsulException, HTTPException, HTTPError, socket.error, socket.timeout):
logger.exception('watch')
patroni/dcs/etcd.py
+31 -26
View File
@@ -290,7 +290,8 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
etcd_nodes = len(machines_cache)
except Exception as e:
logger.debug('Failed to update list of etcd nodes: %r', e)
assert isinstance(retry, Retry) # etcd.EtcdConnectionFailed is raised only if retry is not None!
if TYPE_CHECKING: # pragma: no cover
assert isinstance(retry, Retry) # etcd.EtcdConnectionFailed is raised only if retry is not None!
sleeptime = retry.sleeptime
remaining_time = retry.stoptime - sleeptime - time.time()
nodes, timeout, retries = self._calculate_timeouts(etcd_nodes, remaining_time)
@@ -502,6 +503,17 @@ class AbstractEtcd(AbstractDCS):
if isinstance(raise_ex, Exception):
raise raise_ex
def handle_etcd_exceptions(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
try:
retval = func(self, *args, **kwargs)
self._has_failed = False
return retval
except (RetryFailedError, etcd.EtcdException) as e:
self._handle_exception(e)
return False
except Exception as e:
self._handle_exception(e, raise_ex=self._client.ERROR_CLS('unexpected error'))
def _run_and_handle_exceptions(self, method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
retry = kwargs.pop('retry', self.retry)
try:
@@ -624,16 +636,7 @@ class AbstractEtcd(AbstractDCS):
def catch_etcd_errors(func: Callable[..., Any]) -> Any:
def wrapper(self: AbstractEtcd, *args: Any, **kwargs: Any) -> Any:
try:
retval = func(self, *args, **kwargs)
self._has_failed = False
return retval
except (RetryFailedError, etcd.EtcdException) as e:
self._handle_exception(e)
return False
except Exception as e:
self._handle_exception(e, raise_ex=self._client.ERROR_CLS('unexpected error'))
return self.handle_etcd_exceptions(func, *args, **kwargs)
return wrapper
@@ -645,7 +648,8 @@ class Etcd(AbstractEtcd):
@property
def _client(self) -> EtcdClient:
assert isinstance(self._abstract_client, EtcdClient)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(self._abstract_client, EtcdClient)
return self._abstract_client
def set_ttl(self, ttl: int) -> Optional[bool]:
@@ -696,8 +700,8 @@ class Etcd(AbstractEtcd):
if leader:
member = Member(-1, leader.value, None, {})
member = ([m for m in members if m.name == leader.value] or [member])[0]
index = etcd_index if etcd_index > leader.modifiedIndex else leader.modifiedIndex + 1
leader = Leader(index, leader.ttl, member)
version = etcd_index if etcd_index > leader.modifiedIndex else leader.modifiedIndex + 1
leader = Leader(version, leader.ttl, member)
# failover key
failover = nodes.get(self._FAILOVER)
@@ -742,7 +746,8 @@ class Etcd(AbstractEtcd):
except Exception as e:
self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
self._has_failed = False
assert cluster is not None
if TYPE_CHECKING: # pragma: no cover
assert cluster is not None
return cluster
@catch_etcd_errors
@@ -766,12 +771,12 @@ class Etcd(AbstractEtcd):
return self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, retry=None)
@catch_etcd_errors
def set_failover_value(self, value: str, index: Optional[int] = None) -> bool:
return bool(self._client.write(self.failover_path, value, prevIndex=index or 0))
def set_failover_value(self, value: str, version: Optional[int] = None) -> bool:
return bool(self._client.write(self.failover_path, value, prevIndex=version or 0))
@catch_etcd_errors
def set_config_value(self, value: str, index: Optional[int] = None) -> bool:
return bool(self._client.write(self.config_path, value, prevIndex=index or 0))
def set_config_value(self, value: str, version: Optional[int] = None) -> bool:
return bool(self._client.write(self.config_path, value, prevIndex=version or 0))
@catch_etcd_errors
def _write_leader_optime(self, last_lsn: str) -> bool:
@@ -817,24 +822,24 @@ class Etcd(AbstractEtcd):
return bool(self._client.write(self.history_path, value))
@catch_etcd_errors
def set_sync_state_value(self, value: str, index: Optional[int] = None) -> Union[int, bool]:
return self.retry(self._client.write, self.sync_path, value, prevIndex=index or 0).modifiedIndex
def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]:
return self.retry(self._client.write, self.sync_path, value, prevIndex=version or 0).modifiedIndex
@catch_etcd_errors
def delete_sync_state(self, index: Optional[int] = None) -> bool:
return bool(self.retry(self._client.delete, self.sync_path, prevIndex=index or 0))
def delete_sync_state(self, version: Optional[int] = None) -> bool:
return bool(self.retry(self._client.delete, self.sync_path, prevIndex=version or 0))
def watch(self, leader_index: Optional[int], timeout: float) -> bool:
def watch(self, leader_version: Optional[int], timeout: float) -> bool:
if self.__do_not_watch:
self.__do_not_watch = False
return True
if leader_index:
if leader_version:
end_time = time.time() + timeout
while timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect
try:
result = self._client.watch(self.leader_path, index=leader_index, timeout=timeout + 0.5)
result = self._client.watch(self.leader_path, index=leader_version, timeout=timeout + 0.5)
self._has_failed = False
if result.action == 'compareAndSwap':
time.sleep(0.01)
patroni/dcs/etcd3.py
+48 -42
View File
@@ -13,7 +13,7 @@ from collections import defaultdict
from enum import IntEnum
from urllib3.exceptions import ReadTimeoutError, ProtocolError
from threading import Condition, Lock, Thread
from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, Union
from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
@@ -145,7 +145,8 @@ errCodeToClientError = {getattr(s, 'code'): s for s in Etcd3ClientError.__subcla
def _raise_for_data(data: Union[bytes, str, Dict[str, Union[Any, Dict[str, Any]]]],
status_code: Optional[int] = None) -> Etcd3ClientError:
try:
assert isinstance(data, dict)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(data, dict)
data_error: Optional[Dict[str, Any]] = data.get('error') or data.get('Error')
if isinstance(data_error, dict): # streaming response
status_code = data_error.get('http_code')
@@ -153,7 +154,8 @@ def _raise_for_data(data: Union[bytes, str, Dict[str, Union[Any, Dict[str, Any]]
error: str = data_error['message']
else:
data_code = data.get('code') or data.get('Code')
assert not isinstance(data_code, dict)
if TYPE_CHECKING: # pragma: no cover
assert not isinstance(data_code, dict)
code = data_code
error = str(data_error)
except Exception:
@@ -193,28 +195,7 @@ def build_range_request(key: str, range_end: Union[bytes, str, None] = None) ->
def _handle_auth_errors(func: Callable[..., Any]) -> Any:
def wrapper(self: 'Etcd3Client', *args: Any, **kwargs: Any) -> Any:
def retry(ex: Exception) -> Any:
if self.username and self.password:
self.authenticate()
return func(self, *args, **kwargs)
else:
logger.fatal('Username or password not set, authentication is not possible')
raise ex
try:
return func(self, *args, **kwargs)
except (UserEmpty, PermissionDenied) as e: # no token provided
# PermissionDenied is raised on 3.0 and 3.1
if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied)
or self._cluster_version < (3, 2)):
raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not '
'supported on version lower than 3.3.0. Cluster version: '
'{0}'.format('.'.join(map(str, self._cluster_version))))
return retry(e)
except InvalidAuthToken as e:
logger.error('Invalid auth token: %s', self._token)
return retry(e)
return self.handle_auth_errors(func, *args, **kwargs)
return wrapper
@@ -322,6 +303,29 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
self._token = response.get('token')
return old_token != self._token
def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
def retry(ex: Exception) -> Any:
if self.username and self.password:
self.authenticate()
return func(self, *args, **kwargs)
else:
logger.fatal('Username or password not set, authentication is not possible')
raise ex
try:
return func(self, *args, **kwargs)
except (UserEmpty, PermissionDenied) as e: # no token provided
# PermissionDenied is raised on 3.0 and 3.1
if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied)
or self._cluster_version < (3, 2)):
raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not '
'supported on version lower than 3.3.0. Cluster version: '
'{0}'.format('.'.join(map(str, self._cluster_version))))
return retry(e)
except InvalidAuthToken as e:
logger.error('Invalid auth token: %s', self._token)
return retry(e)
@_handle_auth_errors
def range(self, key: str, range_end: Union[bytes, str, None] = None,
retry: Optional[Retry] = None) -> Dict[str, Any]:
@@ -401,7 +405,7 @@ class KVCache(Thread):
self._leader_key = base64_encode(dcs.leader_path)
self._optime_key = base64_encode(dcs.leader_optime_path)
self._status_key = base64_encode(dcs.status_path)
self._name = base64_encode(dcs._name)
self._name = base64_encode(getattr(dcs, '_name')) # pyright
self._is_ready = False
self._response = None
self._response_lock = Lock()
@@ -582,9 +586,9 @@ class PatroniEtcd3Client(Etcd3Client):
self._kv_cache.condition.wait(timeout)
def get_cluster(self, path: str) -> List[Dict[str, Any]]:
if self._kv_cache and self._etcd3._retry.deadline is not None and path.startswith(self._etcd3.cluster_prefix):
if self._kv_cache and path.startswith(self._etcd3.cluster_prefix):
with self._kv_cache.condition:
self._wait_cache(self._etcd3._retry.deadline)
self._wait_cache(self.read_timeout)
ret = self._kv_cache.copy()
else:
ret = self._etcd3.retry(self.prefix, path).get('kvs', [])
@@ -621,7 +625,6 @@ class Etcd3(AbstractEtcd):
def __init__(self, config: Dict[str, Any]) -> None:
super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition))
assert isinstance(self._client, PatroniEtcd3Client)
self.__do_not_watch = False
self._lease = None
self._last_lease_refresh = 0
@@ -633,12 +636,14 @@ class Etcd3(AbstractEtcd):
@property
def _client(self) -> PatroniEtcd3Client:
assert isinstance(self._abstract_client, PatroniEtcd3Client)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(self._abstract_client, PatroniEtcd3Client)
return self._abstract_client
def set_socket_options(self, sock: socket.socket,
socket_options: Optional[Collection[Tuple[int, int, int]]]) -> None:
assert self._retry.deadline is not None
if TYPE_CHECKING: # pragma: no cover
assert self._retry.deadline is not None
enable_keepalive(sock, self.ttl, int(self.loop_wait + self._retry.deadline))
def set_ttl(self, ttl: int) -> Optional[bool]:
@@ -773,7 +778,8 @@ class Etcd3(AbstractEtcd):
except Exception as e:
self._handle_exception(e, 'get_cluster', raise_ex=Etcd3Error('Etcd is not responding properly'))
self._has_failed = False
assert cluster is not None
if TYPE_CHECKING: # pragma: no cover
assert cluster is not None
return cluster
@catch_etcd_errors
@@ -841,12 +847,12 @@ class Etcd3(AbstractEtcd):
return ret
@catch_etcd_errors
def set_failover_value(self, value: str, index: Optional[str] = None) -> bool:
return bool(self._client.put(self.failover_path, value, mod_revision=index))
def set_failover_value(self, value: str, version: Optional[str] = None) -> bool:
return bool(self._client.put(self.failover_path, value, mod_revision=version))
@catch_etcd_errors
def set_config_value(self, value: str, index: Optional[str] = None) -> bool:
return bool(self._client.put(self.config_path, value, mod_revision=index))
def set_config_value(self, value: str, version: Optional[str] = None) -> bool:
return bool(self._client.put(self.config_path, value, mod_revision=version))
@catch_etcd_errors
def _write_leader_optime(self, last_lsn: str) -> bool:
@@ -893,7 +899,7 @@ class Etcd3(AbstractEtcd):
def _delete_leader(self) -> bool:
cluster = self.cluster
if cluster and isinstance(cluster.leader, Leader) and cluster.leader.name == self._name:
return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.index)
return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.version)
return True
@catch_etcd_errors
@@ -909,15 +915,15 @@ class Etcd3(AbstractEtcd):
return bool(self._client.put(self.history_path, value))
@catch_etcd_errors
def set_sync_state_value(self, value: str, index: Optional[str] = None) -> Union[str, bool]:
return self.retry(self._client.put, self.sync_path, value, mod_revision=index)\
def set_sync_state_value(self, value: str, version: Optional[str] = None) -> Union[str, bool]:
return self.retry(self._client.put, self.sync_path, value, mod_revision=version)\
.get('header', {}).get('revision', False)
@catch_etcd_errors
def delete_sync_state(self, index: Optional[str] = None) -> bool:
return self.retry(self._client.deleterange, self.sync_path, mod_revision=index)
def delete_sync_state(self, version: Optional[str] = None) -> bool:
return self.retry(self._client.deleterange, self.sync_path, mod_revision=version)
def watch(self, leader_index: Optional[str], timeout: float) -> bool:
def watch(self, leader_version: Optional[str], timeout: float) -> bool:
if self.__do_not_watch:
self.__do_not_watch = False
return True
patroni/dcs/kubernetes.py
+50 -40
View File
@@ -135,11 +135,14 @@ class K8sConfig(object):
context = context or config['current-context']
context_value = self._get_by_name(config, 'context', context)
assert isinstance(context_value, dict)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(context_value, dict)
cluster = self._get_by_name(config, 'cluster', context_value['cluster'])
assert isinstance(cluster, dict)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(cluster, dict)
user = self._get_by_name(config, 'user', context_value['user'])
assert isinstance(user, dict)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(user, dict)
self._server = cluster['server'].rstrip('/')
if self._server.startswith('https'):
@@ -281,7 +284,8 @@ class K8sClient(object):
try:
response = self.pool_manager.request('GET', base_uri + path, **kwargs)
endpoint = self._handle_server_response(response, True)
assert isinstance(endpoint, K8sObject)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(endpoint, K8sObject)
for subset in endpoint.subsets:
for port in subset.ports:
if port.name == 'https' and port.protocol == 'TCP':
@@ -412,7 +416,8 @@ class K8sClient(object):
except Exception as e:
logger.debug('Failed to update list of K8s master nodes: %r', e)
assert isinstance(retry, Retry) # K8sConnectionFailed is raised only if retry is not None!
if TYPE_CHECKING: # pragma: no cover
assert isinstance(retry, Retry) # K8sConnectionFailed is raised only if retry is not None!
sleeptime = retry.sleeptime
remaining_time = (retry.stoptime or time.time()) - sleeptime - time.time()
nodes, timeout, retries = self._calculate_timeouts(api_servers, remaining_time)
@@ -559,10 +564,23 @@ class CoreV1ApiProxy(object):
return self._use_endpoints
def _run_and_handle_exceptions(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
try:
return method(*args, **kwargs)
except k8s_client.rest.ApiException as e:
if e.status == 403:
logger.exception('Permission denied')
elif e.status != 409: # Object exists or conflict in resource_version
logger.exception('Unexpected error from Kubernetes API')
return False
except (RetryFailedError, K8sException) as e:
raise KubernetesError(e)
def catch_kubernetes_errors(func: Callable[..., Any]) -> Callable[..., Any]:
def wrapper(self: 'Kubernetes', *args: Any, **kwargs: Any) -> Any:
try:
return self._run_and_handle_exceptions(func, self, *args, **kwargs)
return _run_and_handle_exceptions(func, self, *args, **kwargs)
except KubernetesError:
return False
return wrapper
@@ -584,7 +602,8 @@ class ObjectCache(Thread):
self._response_lock = Lock() # protect the `self._response` from concurrent access
self._object_cache: Dict[str, K8sObject] = {}
self._object_cache_lock = Lock()
self._annotations_map = {self._dcs.leader_path: self._dcs._LEADER, self._dcs.config_path: self._dcs._CONFIG}
self._annotations_map = {self._dcs.leader_path: getattr(self._dcs, '_LEADER'),
self._dcs.config_path: getattr(self._dcs, '_CONFIG')} # pyright
self.start()
def _list(self) -> K8sObject:
@@ -779,19 +798,6 @@ class Kubernetes(AbstractDCS):
kwargs['_retry'] = retry
return retry(method, *args, **kwargs)
@staticmethod
def _run_and_handle_exceptions(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
try:
return method(*args, **kwargs)
except k8s_client.rest.ApiException as e:
if e.status == 403:
logger.exception('Permission denied')
elif e.status != 409: # Object exists or conflict in resource_version
logger.exception('Unexpected error from Kubernetes API')
return False
except (RetryFailedError, K8sException) as e:
raise KubernetesError(e)
def client_path(self, path: str) -> str:
return super(Kubernetes, self).client_path(path)[1:].replace('/', '-')
@@ -818,7 +824,8 @@ class Kubernetes(AbstractDCS):
Either cause by changes in the local configuration file + SIGHUP or by changes of dynamic configuration"""
super(Kubernetes, self).reload_config(config)
assert self._retry.deadline is not None
if TYPE_CHECKING: # pragma: no cover
assert self._retry.deadline is not None
self._api.configure_timeouts(self.loop_wait, self._retry.deadline, self.ttl)
# retriable_http_codes supposed to be either int, list of integers or comma-separated string with integers.
@@ -954,7 +961,8 @@ class Kubernetes(AbstractDCS):
def __load_cluster(
self, group: Optional[str], loader: Callable[[Dict[str, Any]], Union[Cluster, Dict[int, Cluster]]]
) -> Union[Cluster, Dict[int, Cluster]]:
assert self._retry.deadline is not None
if TYPE_CHECKING: # pragma: no cover
assert self._retry.deadline is not None
stop_time = time.time() + self._retry.deadline
self._api.refresh_api_servers_cache()
try:
@@ -978,7 +986,8 @@ class Kubernetes(AbstractDCS):
def get_citus_coordinator(self) -> Optional[Cluster]:
try:
ret = self.__load_cluster(str(CITUS_COORDINATOR_GROUP_ID), self._cluster_loader)
assert isinstance(ret, Cluster)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(ret, Cluster)
return ret
except Exception as e:
logger.error('Failed to load Citus coordinator cluster from Kubernetes: %r', e)
@@ -1170,8 +1179,8 @@ class Kubernetes(AbstractDCS):
if kind and (kind_annotations.get(self._LEADER) != self._name or kind_resource_version == resource_version):
return False
return bool(self._run_and_handle_exceptions(self._patch_or_create, self.leader_path, annotations,
kind_resource_version, ips=ips, retry=_retry))
return bool(_run_and_handle_exceptions(self._patch_or_create, self.leader_path, annotations,
kind_resource_version, ips=ips, retry=_retry))
def update_leader(self, last_lsn: Optional[int], slots: Optional[Dict[str, int]] = None,
failsafe: Optional[Dict[str, str]] = None) -> bool:
@@ -1232,24 +1241,24 @@ class Kubernetes(AbstractDCS):
def take_leader(self) -> bool:
return self.attempt_to_acquire_leader()
def set_failover_value(self, value: str, index: Optional[str] = None) -> bool:
def set_failover_value(self, value: str, version: Optional[str] = None) -> bool:
"""Unused"""
raise NotImplementedError # pragma: no cover
def manual_failover(self, leader: Optional[str], candidate: Optional[str],
scheduled_at: Optional[datetime.datetime] = None, index: Optional[str] = None) -> bool:
scheduled_at: Optional[datetime.datetime] = None, version: Optional[str] = None) -> bool:
annotations = {'leader': leader or None, 'member': candidate or None,
'scheduled_at': scheduled_at and scheduled_at.isoformat()}
patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.index)
return bool(self.patch_or_create(self.failover_path, annotations, index, bool(index or patch), False))
patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.version)
return bool(self.patch_or_create(self.failover_path, annotations, version, bool(version or patch), False))
@property
def _config_resource_version(self) -> Optional[str]:
config = self._kinds.get(self.config_path)
return config and config.metadata.resource_version
def set_config_value(self, value: str, index: Optional[str] = None) -> bool:
return self.patch_or_create_config({self._CONFIG: value}, index, bool(self._config_resource_version), False)
def set_config_value(self, value: str, version: Optional[str] = None) -> bool:
return self.patch_or_create_config({self._CONFIG: value}, version, bool(self._config_resource_version), False)
@catch_kubernetes_errors
def touch_member(self, data: Dict[str, Any]) -> bool:
@@ -1279,7 +1288,8 @@ class Kubernetes(AbstractDCS):
def initialize(self, create_new: bool = True, sysid: str = "") -> bool:
cluster = self.cluster
resource_version = str(cluster.config.index) if cluster and cluster.config and cluster.config.index else None
resource_version = str(cluster.config.version)\
if cluster and cluster.config and cluster.config.version else None
return self.patch_or_create_config({self._INITIALIZE: sysid}, resource_version)
def _delete_leader(self) -> bool:
@@ -1308,37 +1318,37 @@ class Kubernetes(AbstractDCS):
def set_history_value(self, value: str) -> bool:
return self.patch_or_create_config({self._HISTORY: value}, None, bool(self._config_resource_version), False)
def set_sync_state_value(self, value: str, index: Optional[str] = None) -> bool:
def set_sync_state_value(self, value: str, version: Optional[str] = None) -> bool:
"""Unused"""
raise NotImplementedError # pragma: no cover
def write_sync_state(self, leader: Optional[str], sync_standby: Optional[Collection[str]],
quorum: Optional[int], index: Optional[str] = None) -> Optional[SyncState]:
quorum: Optional[int], version: Optional[str] = None) -> Optional[SyncState]:
"""Prepare and write annotations to $SCOPE-sync Endpoint or ConfigMap.
:param leader: name of the leader node that manages /sync key
:param sync_standby: collection of currently known synchronous standby node names
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
:param index: last known `resource_version` for conditional update of the object
:param version: last known `resource_version` for conditional update of the object
:returns: the new :class:`SyncState` object or None
"""
sync_state = self.sync_state(leader, sync_standby, quorum)
sync_state['quorum'] = str(sync_state['quorum']) if sync_state['quorum'] is not None else None
ret = self.patch_or_create(self.sync_path, sync_state, index, False)
ret = self.patch_or_create(self.sync_path, sync_state, version, False)
if not isinstance(ret, bool):
return SyncState.from_node(ret.metadata.resource_version, sync_state)
def delete_sync_state(self, index: Optional[str] = None) -> bool:
def delete_sync_state(self, version: Optional[str] = None) -> bool:
"""Patch annotations of $SCOPE-sync Endpoint or ConfigMap with empty values.
Effectively it removes "leader" and "sync_standby" annotations from the object.
:param index: last known `resource_version` for conditional update of the object
:param version: last known `resource_version` for conditional update of the object
:returns: `True` if "delete" was successful
"""
return self.write_sync_state(None, None, None, index=index) is not None
return self.write_sync_state(None, None, None, version=version) is not None
def watch(self, leader_index: Optional[str], timeout: float) -> bool:
def watch(self, leader_version: Optional[str], timeout: float) -> bool:
if self.__do_not_watch:
self.__do_not_watch = False
return True
patroni/dcs/raft.py
+12 -12
View File
@@ -10,7 +10,7 @@ from pysyncobj.dns_resolver import globalDnsResolver
from pysyncobj.node import TCPNode
from pysyncobj.transport import TCPTransport, CONNECTION_STATE
from pysyncobj.utility import TcpUtility
from typing import Any, Callable, Collection, Dict, List, Optional, Union, TYPE_CHECKING
from typing import Any, Callable, Collection, Dict, List, Optional, Set, Union, TYPE_CHECKING
from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory, citus_group_re
from ..exceptions import DCSError
@@ -114,7 +114,7 @@ class KVStoreTTL(DynMemberSyncObj):
self.set_retry_timeout(int(config.get('retry_timeout') or 10))
self_addr = config.get('self_addr')
partner_addrs = set(config.get('partner_addrs', []))
partner_addrs: Set[str] = set(config.get('partner_addrs', []))
if config.get('patronictl'):
if self_addr:
partner_addrs.add(self_addr)
@@ -430,11 +430,11 @@ class Raft(AbstractDCS):
return self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl,
handle_raft_error=False, prevExist=False) is not False
def set_failover_value(self, value: str, index: Optional[int] = None) -> bool:
return self._sync_obj.set(self.failover_path, value, prevIndex=index) is not False
def set_failover_value(self, value: str, version: Optional[int] = None) -> bool:
return self._sync_obj.set(self.failover_path, value, prevIndex=version) is not False
def set_config_value(self, value: str, index: Optional[int] = None) -> bool:
return self._sync_obj.set(self.config_path, value, prevIndex=index) is not False
def set_config_value(self, value: str, version: Optional[int] = None) -> bool:
return self._sync_obj.set(self.config_path, value, prevIndex=version) is not False
def touch_member(self, data: Dict[str, Any]) -> bool:
value = json.dumps(data, separators=(',', ':'))
@@ -458,17 +458,17 @@ class Raft(AbstractDCS):
def set_history_value(self, value: str) -> bool:
return self._sync_obj.set(self.history_path, value) is not False
def set_sync_state_value(self, value: str, index: Optional[int] = None) -> Union[int, bool]:
ret = self._sync_obj.set(self.sync_path, value, prevIndex=index)
def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]:
ret = self._sync_obj.set(self.sync_path, value, prevIndex=version)
if isinstance(ret, dict):
return ret['index']
return ret
def delete_sync_state(self, index: Optional[int] = None) -> bool:
return self._sync_obj.delete(self.sync_path, prevIndex=index)
def delete_sync_state(self, version: Optional[int] = None) -> bool:
return self._sync_obj.delete(self.sync_path, prevIndex=version)
def watch(self, leader_index: Optional[int], timeout: float) -> bool:
def watch(self, leader_version: Optional[int], timeout: float) -> bool:
try:
return super(Raft, self).watch(leader_index, timeout)
return super(Raft, self).watch(leader_version, timeout)
finally:
self.event.clear()
patroni/dcs/zookeeper.py
+15 -15
View File
@@ -273,7 +273,7 @@ class ZooKeeper(AbstractDCS):
member = Member(-1, leader[0], None, {})
member = ([m for m in members if m.name == leader[0]] or [member])[0]
leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
self._fetch_cluster = member.index == -1
self._fetch_cluster = member.version == -1
# get last known leader lsn and slots
last_lsn, slots = self.get_status(path, leader)
@@ -357,19 +357,19 @@ class ZooKeeper(AbstractDCS):
logger.info('Could not take out TTL lock')
return False
def _set_or_create(self, key: str, value: str, index: Optional[int] = None,
def _set_or_create(self, key: str, value: str, version: Optional[int] = None,
retry: bool = False, do_not_create_empty: bool = False) -> Union[int, bool]:
value_bytes = value.encode('utf-8')
try:
if retry:
ret = self._client.retry(self._client.set, key, value_bytes, version=index or -1)
ret = self._client.retry(self._client.set, key, value_bytes, version=version or -1)
else:
ret = self._client.set_async(key, value_bytes, version=index or -1).get(timeout=1)
ret = self._client.set_async(key, value_bytes, version=version or -1).get(timeout=1)
return ret.version
except NoNodeError:
if do_not_create_empty and not value_bytes:
return True
elif index is None:
elif version is None:
if self._create(key, value_bytes, retry):
return 0
else:
@@ -378,11 +378,11 @@ class ZooKeeper(AbstractDCS):
logger.exception('Failed to update %s', key)
return False
def set_failover_value(self, value: str, index: Optional[int] = None) -> bool:
return self._set_or_create(self.failover_path, value, index) is not False
def set_failover_value(self, value: str, version: Optional[int] = None) -> bool:
return self._set_or_create(self.failover_path, value, version) is not False
def set_config_value(self, value: str, index: Optional[int] = None) -> bool:
return self._set_or_create(self.config_path, value, index, retry=True) is not False
def set_config_value(self, value: str, version: Optional[int] = None) -> bool:
return self._set_or_create(self.config_path, value, version, retry=True) is not False
def initialize(self, create_new: bool = True, sysid: str = "") -> bool:
sysid_bytes = sysid.encode('utf-8')
@@ -494,14 +494,14 @@ class ZooKeeper(AbstractDCS):
def set_history_value(self, value: str) -> bool:
return self._set_or_create(self.history_path, value) is not False
def set_sync_state_value(self, value: str, index: Optional[int] = None) -> Union[int, bool]:
return self._set_or_create(self.sync_path, value, index, retry=True, do_not_create_empty=True)
def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]:
return self._set_or_create(self.sync_path, value, version, retry=True, do_not_create_empty=True)
def delete_sync_state(self, index: Optional[int] = None) -> bool:
return self.set_sync_state_value("{}", index) is not False
def delete_sync_state(self, version: Optional[int] = None) -> bool:
return self.set_sync_state_value("{}", version) is not False
def watch(self, leader_index: Optional[int], timeout: float) -> bool:
ret = super(ZooKeeper, self).watch(leader_index, timeout + 0.5)
def watch(self, leader_version: Optional[int], timeout: float) -> bool:
ret = super(ZooKeeper, self).watch(leader_version, timeout + 0.5)
if ret and not self._fetch_status:
self._fetch_cluster = True
return ret or self._fetch_cluster
patroni/ha.py
+23 -23
View File
@@ -101,10 +101,10 @@ class Failsafe(object):
@property
def leader(self) -> Optional[Leader]:
with self._lock:
if self._last_update + self._dcs.ttl > time.time():
return Leader('', '', RemoteMember(self._name, {'api_url': self._api_url,
'conn_url': self._conn_url,
'slots': self._slots}))
if self._last_update + self._dcs.ttl > time.time() and self._name:
return Leader('', '', RemoteMember.from_name_and_data(self._name, {'api_url': self._api_url,
'conn_url': self._conn_url,
'slots': self._slots}))
def update_cluster(self, cluster: Cluster) -> Cluster:
# Enreach cluster with the real leader if there was a ping from it
@@ -591,7 +591,7 @@ class Ha(object):
def disable_synchronous_replication(self) -> None:
"""Cleans up /sync key in DCS if synchronous replication is disabled."""
if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(index=self.cluster.sync.index):
if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(version=self.cluster.sync.version):
logger.info("Disabled synchronous replication")
self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet())
@@ -611,7 +611,7 @@ class Ha(object):
sync = self.cluster.sync
leader = sync.leader or self.state_handler.name
if sync.is_empty:
sync = self.dcs.write_sync_state(leader, None, 0, index=sync.index)
sync = self.dcs.write_sync_state(leader, None, 0, version=sync.version)
if not sync:
return logger.warning("Updating sync state failed")
@@ -630,7 +630,7 @@ class Ha(object):
if transition == 'quorum':
logger.info("Setting leader to %s, quorum to %d of %d (%s)",
leader, num, len(nodes), ", ".join(sorted(nodes)))
sync = self.dcs.write_sync_state(leader, nodes, num, index=sync.index)
sync = self.dcs.write_sync_state(leader, nodes, num, version=sync.version)
if not sync:
return logger.info('Synchronous replication key updated by someone else.')
elif transition == 'sync':
@@ -675,7 +675,7 @@ class Ha(object):
if sync_common != voters:
logger.info("Updating synchronous privilege temporarily from %s to %s",
list(voters), list(sync_common))
sync = self.dcs.write_sync_state(self.state_handler.name, sync_common, 0, index=sync.index)
sync = self.dcs.write_sync_state(self.state_handler.name, sync_common, 0, version=sync.version)
if not sync:
return logger.info('Synchronous replication key updated by someone else.')
@@ -694,7 +694,7 @@ class Ha(object):
allow_promote = self.state_handler.sync_handler.current_state(self.cluster).sync
if allow_promote and allow_promote != sync_common:
if self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, index=sync.index):
if self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, version=sync.version):
logger.info("Synchronous standby status assigned to %s", list(allow_promote))
else:
logger.info("Synchronous replication key updated by someone else")
@@ -738,7 +738,7 @@ class Ha(object):
# Just set ourselves as the authoritative source of truth for now. We don't want to wait for standbys
# to connect. We will try finding a synchronous standby in the next cycle.
if not self.dcs.write_sync_state(self.state_handler.name, None, 0, index=self.cluster.sync.index):
if not self.dcs.write_sync_state(self.state_handler.name, None, 0, version=self.cluster.sync.version):
return False
self.state_handler.sync_handler.set_synchronous_standby_names(sync, numsync)
@@ -921,9 +921,8 @@ class Ha(object):
data['slots'] = self.state_handler.slots()
except Exception:
logger.exception('Exception when called state_handler.slots()')
members = [RemoteMember(name, {'api_url': url})
for name, url in failsafe.items()
if name != self.state_handler.name]
members = [RemoteMember.from_name_and_data(name, {'api_url': url})
for name, url in failsafe.items() if name != self.state_handler.name]
if not members: # A sinlge node cluster
return True
pool = ThreadPool(len(members))
@@ -1065,7 +1064,7 @@ class Ha(object):
if not self.cluster.get_member(failover.candidate, fallback_to_leader=False)\
and self.state_handler.is_leader():
logger.warning("manual failover: removing failover key because failover candidate is not running")
self.dcs.manual_failover('', '', index=failover.index)
self.dcs.manual_failover('', '', version=failover.version)
return None
return False
@@ -1157,7 +1156,8 @@ class Ha(object):
if failsafe_members and self.state_handler.name not in failsafe_members:
return False
# Race among not only existing cluster members, but also all known members from the failsafe config
all_known_members += [RemoteMember(name, {'api_url': url}) for name, url in failsafe_members.items()]
all_known_members += [RemoteMember.from_name_and_data(name, {'api_url': url})
for name, url in failsafe_members.items()]
all_known_members += self.cluster.members
# Special handling if synchronous mode was requested and activated (the leader in /sync is not empty)
@@ -1309,7 +1309,7 @@ class Ha(object):
if (failover.scheduled_at and not
self.should_run_scheduled_action("failover", failover.scheduled_at, lambda:
self.dcs.manual_failover('', '', index=failover.index))):
self.dcs.manual_failover('', '', version=failover.version))):
return
if not failover.leader or failover.leader == self.state_handler.name:
@@ -1339,7 +1339,7 @@ class Ha(object):
failover.leader, self.state_handler.name)
logger.info('Cleaning up failover key')
self.dcs.manual_failover('', '', index=failover.index)
self.dcs.manual_failover('', '', version=failover.version)
def process_unhealthy_cluster(self) -> str:
"""Cluster has no leader key"""
@@ -1350,7 +1350,7 @@ class Ha(object):
if failover:
if self.is_paused() and failover.leader and failover.candidate:
logger.info('Updating failover key after acquiring leader lock...')
self.dcs.manual_failover('', failover.candidate, failover.scheduled_at, failover.index)
self.dcs.manual_failover('', failover.candidate, failover.scheduled_at, failover.version)
else:
logger.info('Cleaning up failover key after acquiring leader lock...')
self.dcs.manual_failover('', '')
@@ -1732,7 +1732,7 @@ class Ha(object):
self.global_config = self.patroni.config.get_global_config(self.cluster)
self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover, self.global_config)
except Exception:
self.state_handler.reset_cluster_info_state(None)
self.state_handler.reset_cluster_info_state(None, self.patroni.nofailover, self.global_config)
raise
if self.is_paused():
@@ -2000,11 +2000,11 @@ class Ha(object):
def watch(self, timeout: float) -> bool:
# watch on leader key changes if the postgres is running and leader is known and current node is not lock owner
if self._async_executor.busy or not self.cluster or self.cluster.is_unlocked() or self.has_lock(False):
leader_index = None
leader_version = None
else:
leader_index = self.cluster.leader.index if self.cluster.leader else None
leader_version = self.cluster.leader.version if self.cluster.leader else None
return self.dcs.watch(leader_index, timeout)
return self.dcs.watch(leader_version, timeout)
def wakeup(self) -> None:
"""Call of this method will trigger the next run of HA loop if there is
@@ -2029,4 +2029,4 @@ class Ha(object):
data['conn_kwargs'] = conn_kwargs
name = member.name if member else 'remote_member:{}'.format(uuid.uuid1())
return RemoteMember(name, data)
return RemoteMember.from_name_and_data(name, data)
patroni/log.py
+6 -4
View File
@@ -13,7 +13,7 @@ from patroni.utils import deep_compare
from queue import Queue, Full
from threading import Lock, Thread
from typing import Any, Dict, List, Optional, Union
from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
_LOGGER = logging.getLogger(__name__)
@@ -249,8 +249,9 @@ class PatroniLogger(Thread):
if not isinstance(self.log_handler, RotatingFileHandler):
new_handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
handler = new_handler or self.log_handler
assert isinstance(handler, RotatingFileHandler)
handler.maxBytes = int(config.get('file_size', 25000000))
if TYPE_CHECKING: # pragma: no cover
assert isinstance(handler, RotatingFileHandler)
handler.maxBytes = int(config.get('file_size', 25000000)) # pyright: ignore [reportGeneralTypeIssues]
handler.backupCount = int(config.get('file_num', 4))
else:
if self.log_handler is None or isinstance(self.log_handler, RotatingFileHandler):
@@ -306,7 +307,8 @@ class PatroniLogger(Thread):
while True:
self._close_old_handlers()
assert self.log_handler is not None
if TYPE_CHECKING: # pragma: no cover
assert self.log_handler is not None
record = self._queue_handler.queue.get(True)
# special message that indicates Patroni is shutting down
patroni/postgresql/__init__.py
+6 -2
View File
@@ -197,7 +197,7 @@ class Postgresql(object):
"FROM pg_catalog.pg_stat_get_wal_senders() w,"
" pg_catalog.pg_stat_get_activity(w.pid)"
" WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name)
if (not self._global_config or self._global_config.is_synchronous_mode)
if (not self.global_config or self.global_config.is_synchronous_mode)
and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL")
if self._major_version >= 90600:
@@ -380,6 +380,10 @@ class Postgresql(object):
self.config.write_postgresql_conf()
self.reload()
@property
def global_config(self) -> Optional['GlobalConfig']:
return self._global_config
def reset_cluster_info_state(self, cluster: Union[Cluster, None], nofailover: bool = False,
global_config: Optional['GlobalConfig'] = None) -> None:
"""Reset monitoring query cache.
@@ -393,7 +397,7 @@ class Postgresql(object):
:param global_config: last known :class:`GlobalConfig` object
"""
self._cluster_info_state = {}
if cluster and cluster.config and cluster.config.modify_index:
if cluster and cluster.config and cluster.config.modify_version:
self._has_permanent_logical_slots =\
cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
patroni/postgresql/config.py
+2 -2
View File
@@ -860,9 +860,9 @@ class ConfigHandler(object):
parameters = config['parameters'].copy()
listen_addresses, port = split_host_port(config['listen'], 5432)
parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port))
if not self._postgresql._global_config or self._postgresql._global_config.is_synchronous_mode:
if not self._postgresql.global_config or self._postgresql.global_config.is_synchronous_mode:
if self._synchronous_standby_names is None:
if self._postgresql._global_config and self._postgresql._global_config.is_synchronous_mode_strict\
if self._postgresql.global_config and self._postgresql.global_config.is_synchronous_mode_strict\
and self._postgresql.role in ('master', 'primary', 'promoted'):
parameters['synchronous_standby_names'] = '*'
else:
patroni/postgresql/sync.py
+9 -9
View File
@@ -249,10 +249,10 @@ class SyncHandler(object):
if len(replica_list) > 1 else self._postgresql.last_operation()
if TYPE_CHECKING: # pragma: no cover
assert self._postgresql._global_config is not None
sync_node_count = self._postgresql._global_config.synchronous_node_count\
assert self._postgresql.global_config is not None
sync_node_count = self._postgresql.global_config.synchronous_node_count\
if self._postgresql.supports_multiple_sync else 1
sync_node_maxlag = self._postgresql._global_config.maximum_lag_on_syncnode
sync_node_maxlag = self._postgresql.global_config.maximum_lag_on_syncnode
active = CaseInsensitiveSet()
sync_nodes = CaseInsensitiveSet()
@@ -260,7 +260,7 @@ class SyncHandler(object):
# Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
for pid, app_name, sync_state, replica_lsn, nofailover in sorted(replica_list, key=lambda x: x[4]):
if app_name not in self._ready_replicas and app_name in self._ssn_data.members:
if self._postgresql._global_config.is_quorum_commit_mode:
if self._postgresql.global_config.is_quorum_commit_mode:
# When quorum commit is enabled we can't check against cluster.sync because nodes
# are written there when at least one of them caught up with _primary_flush_lsn.
if replica_lsn >= self._primary_flush_lsn\
@@ -273,7 +273,7 @@ class SyncHandler(object):
self._ready_replicas[app_name] = pid
if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
if self._postgresql._global_config.is_quorum_commit_mode:
if self._postgresql.global_config.is_quorum_commit_mode:
# add nodes with nofailover tag only to get enough "active" nodes
if not nofailover or len(active) < sync_node_count:
if app_name in self._ready_replicas:
@@ -287,7 +287,7 @@ class SyncHandler(object):
if len(active) >= sync_node_count:
break
if self._postgresql._global_config.is_quorum_commit_mode:
if self._postgresql.global_config.is_quorum_commit_mode:
sync_nodes = CaseInsensitiveSet() if self._ssn_data.has_star else self._ssn_data.members
return _SyncState(
@@ -319,11 +319,11 @@ class SyncHandler(object):
sync_param = next(iter(sync), None)
if TYPE_CHECKING: # pragma: no cover
assert self._postgresql._global_config is not None
assert self._postgresql.global_config is not None
if self._postgresql._global_config.is_quorum_commit_mode and sync or\
if self._postgresql.global_config.is_quorum_commit_mode and sync or\
self._postgresql.supports_multiple_sync and len(sync) > 1:
prefix = 'ANY ' if self._postgresql._global_config.is_quorum_commit_mode\
prefix = 'ANY ' if self._postgresql.global_config.is_quorum_commit_mode\
and self._postgresql.supports_quorum_commit else ''
sync_param = '{0}{1} ({2})'.format(prefix, num, sync_param)
patroni/scripts/wale_restore.py
+3 -2
View File
@@ -32,7 +32,7 @@ import sys
import time
from enum import IntEnum
from typing import Any, List, NamedTuple, Optional, Tuple
from typing import Any, List, NamedTuple, Optional, Tuple, TYPE_CHECKING
from .. import psycopg
@@ -365,7 +365,8 @@ def main() -> int:
break
time.sleep(RETRY_SLEEP_INTERVAL)
assert exit_code is not None
if TYPE_CHECKING: # pragma: no cover
assert exit_code is not None
return exit_code
patroni/validator.py
+29 -14
View File
@@ -11,7 +11,7 @@ import shutil
import socket
import subprocess
from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType
from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, TYPE_CHECKING
from .utils import parse_int, split_host_port, data_directory_is_empty
from .dcs import dcs_modules
@@ -196,7 +196,8 @@ def get_major_version(bin_dir: OptionalType[str] = None) -> str:
binary = os.path.join(bin_dir, 'postgres')
version = subprocess.check_output([binary, '--version']).decode()
version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version)
assert version is not None
if TYPE_CHECKING: # pragma: no cover
assert version is not None
return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)
@@ -340,14 +341,17 @@ class Optional(object):
"""Mark a configuration option as optional.
:ivar name: name of the configuration option.
:ivar default: value to set if the configuration option is not explicitly provided
"""
def __init__(self, name: str) -> None:
def __init__(self, name: str, default: OptionalType[Any] = None) -> None:
"""Create an :class:`Optional` object.
:param name: name of the configuration option.
:param default: value to set if the configuration option is not explicitly provided
"""
self.name = name
self.default = default
class Directory(object):
@@ -369,14 +373,27 @@ class Directory(object):
self.contains = contains
self.contains_executable = contains_executable
def _check_executables(self, path: OptionalType[str] = None) -> Iterator[Result]:
"""Check that all executables from contains_executable list exist within the given directory or within PATH.
:param path: optional path to the base directory against which executables will be validated.
If not provided, check within PATH.
:rtype: Iterator[:class:`Result`] objects with the error message containing the name of the executable,
if any check fails.
"""
for program in self.contains_executable or []:
if not shutil.which(program, path=path):
yield Result(False, f"does not contain '{program}' in '{(path or '$PATH')}'")
def validate(self, name: str) -> Iterator[Result]:
"""Check if the expected paths and executables can be found under *name* directory.
:param name: path to the base directory against which paths and executables will be validated.
Check against PATH if name is not provided.
:rtype: Iterator[:class:`Result`] objects with the error message related to the failure, if any check fails.
"""
if not name:
yield Result(False, "is an empty string")
yield from self._check_executables()
elif not os.path.exists(name):
yield Result(False, "Directory '{}' does not exist.".format(name))
elif not os.path.isdir(name):
@@ -386,10 +403,7 @@ class Directory(object):
for path in self.contains:
if not os.path.exists(os.path.join(name, path)):
yield Result(False, "'{}' does not contain '{}'".format(name, path))
if self.contains_executable:
for program in self.contains_executable:
if not shutil.which(program, path=name):
yield Result(False, "'{}' does not contain '{}'".format(name, program))
yield from self._check_executables(path=name)
class Schema(object):
@@ -471,8 +485,7 @@ class Schema(object):
* It must contain a ``bind.host`` entry which value should be valid as per function ``validate_host``;
* It must contain a ``bind.port`` entry which value should be an :class:`int` instance;
* It must contain a ``aliases`` entry which value should be a :class:`list` of :class:`str` instances;
* It may optionally contain a ``data_directory`` entry. If not given it will assume the value
``/var/lib/myapp``;
* It may optionally contain a ``data_directory`` entry, with a value which should be a string;
* It must contain at least one of ``log_to_file`` or ``log_to_db``, with a value which should be a
:class:`bool` instance;
* It must contain a ``version`` entry which value should be either an :class:`int` or a :class:`float`
@@ -582,9 +595,11 @@ class Schema(object):
for d in self._data_key(key):
if d not in self.data and not isinstance(key, Optional):
yield Result(False, "is not defined.", path=d)
elif d not in self.data and isinstance(key, Optional):
elif d not in self.data and isinstance(key, Optional) and key.default is None:
continue
else:
if d not in self.data and isinstance(key, Optional):
self.data[d] = key.default
validator = self.validator[key]
if isinstance(key, Or) and isinstance(self.validator[key], Case):
validator = self.validator[key]._schema[d]
@@ -806,11 +821,11 @@ schema = Schema({
"authentication": {
"replication": userattributes,
"superuser": userattributes,
"rewind": userattributes
Optional("rewind"): userattributes
},
"data_dir": validate_data_dir,
Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
"postgres", "pg_isready"]),
Optional("bin_dir", ""): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
"postgres", "pg_isready"]),
Optional("parameters"): {
Optional("unix_socket_directories"): str
},
patroni/watchdog/base.py
+2 -2
View File
@@ -34,7 +34,7 @@ def parse_mode(mode: Union[bool, str]) -> str:
def synchronized(func: Callable[..., Any]) -> Callable[..., Any]:
def wrapped(self: 'Watchdog', *args: Any, **kwargs: Any) -> Any:
with self._lock:
with self.lock:
return func(self, *args, **kwargs)
return wrapped
@@ -90,7 +90,7 @@ class Watchdog(object):
def __init__(self, config: Config) -> None:
self.config = WatchdogConfig(config)
self.active_config: WatchdogConfig = self.config
self._lock = RLock()
self.lock = RLock()
self.active = False
if self.config.mode == MODE_OFF:
patroni/watchdog/linux.py
+1
View File
@@ -1,3 +1,4 @@
# pyright: reportConstantRedefinition=false
import ctypes
import os
import platform
pyrightconfig.json
+1 -1
View File
@@ -19,7 +19,7 @@
"reportMissingImports": true,
"reportMissingTypeStubs": false,
"pythonVersion": "3.6",
"pythonVersion": "3.11",
"pythonPlatform": "All",
"typeCheckingMode": "strict"
tests/test_api.py
+23 -6
View File
@@ -180,6 +180,7 @@ class MockRestApiServer(RestApiServer):
@patch('ssl.SSLContext.load_cert_chain', Mock())
@patch('ssl.SSLContext.wrap_socket', Mock(return_value=0))
@patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
@patch.object(HTTPServer, '__init__', Mock())
class TestRestApiHandler(unittest.TestCase):
@@ -589,6 +590,7 @@ class TestRestApiServer(unittest.TestCase):
@patch('ssl.SSLContext.load_cert_chain', Mock())
@patch('ssl.SSLContext.set_ciphers', Mock())
@patch('ssl.SSLContext.wrap_socket', Mock(return_value=0))
@patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
@patch.object(HTTPServer, '__init__', Mock())
def setUp(self):
self.srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required',
@@ -622,24 +624,39 @@ class TestRestApiServer(unittest.TestCase):
try:
raise Exception()
except Exception:
self.assertIsNone(MockRestApiServer.handle_error(None, ('127.0.0.1', 55555)))
self.assertIsNone(self.srv.handle_error(None, ('127.0.0.1', 55555)))
@patch.object(HTTPServer, '__init__', Mock(side_effect=socket.error))
def test_socket_error(self):
self.assertRaises(socket.error, MockRestApiServer, Mock(), '', {'listen': '*:8008'})
def __create_socket(self):
sock = socket.socket()
try:
import ssl
ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
ctx.check_hostname = False
sock = ctx.wrap_socket(sock=sock)
sock.do_handshake = Mock()
sock.unwrap = Mock(side_effect=Exception)
except Exception:
pass
return sock
@patch.object(ThreadingMixIn, 'process_request_thread', Mock())
def test_process_request_thread(self):
self.srv.process_request_thread(Mock(), '2')
self.srv.process_request_thread(self.__create_socket(), ('2', 54321))
@patch.object(MockRestApiServer, 'process_request', Mock(side_effect=RuntimeError))
@patch.object(MockRestApiServer, 'get_request')
def test_process_request_error(self, mock_get_request):
mock_request = Mock()
mock_request.unwrap.side_effect = Exception
mock_get_request.return_value = (mock_request, ('127.0.0.1', 55555))
mock_get_request.return_value = (self.__create_socket(), ('127.0.0.1', 55555))
self.srv._handle_request_noblock()
@patch('ssl._ssl._test_decode_cert', Mock())
@patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()]))
def test_reload_local_certificate(self):
self.assertTrue(self.srv.reload_local_certificate())
@patch('ssl.SSLContext.load_verify_locations', Mock(side_effect=Exception))
def test_get_certificate_serial_number(self):
self.assertIsNone(self.srv.get_certificate_serial_number())
tests/test_ha.py
+11 -10
View File
@@ -201,6 +201,7 @@ class TestHa(PostgresInit):
@patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.etcd']))
@patch.object(etcd.Client, 'read', etcd_read)
@patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
@patch.object(Config, '_load_cache', Mock())
def setUp(self):
super(TestHa, self).setUp()
self.p.set_state('running')
@@ -1206,8 +1207,8 @@ class TestHa(PostgresInit):
# When we just became primary nobody is sync
self.assertEqual(self.ha.enforce_primary_role('msg', 'promote msg'), 'promote msg')
mock_set_sync.assert_called_once_with(frozenset(), 0)
mock_write_sync.assert_called_once_with('leader', None, 0, index=0)
mock_set_sync.assert_called_once_with(CaseInsensitiveSet(), 0)
mock_write_sync.assert_called_once_with('leader', None, 0, version=0)
mock_set_sync.reset_mock()
@@ -1245,7 +1246,7 @@ class TestHa(PostgresInit):
mock_acquire.assert_called_once()
mock_follow.assert_not_called()
mock_promote.assert_called_once()
mock_write_sync.assert_called_once_with('other', None, 0, index=0)
mock_write_sync.assert_called_once_with('other', None, 0, version=0)
def test_disable_sync_when_restarting(self):
self.ha.is_synchronous_mode = true
@@ -1447,7 +1448,7 @@ class TestHa(PostgresInit):
'Postponing promotion because synchronous replication state was updated by somebody else')
self.assertEqual(self.ha.dcs.write_sync_state.call_count, 1)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, None, 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': 0})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': 0})
mock_set_sync = self.p.config.set_synchronous_standby_names = Mock()
mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=True)
@@ -1455,7 +1456,7 @@ class TestHa(PostgresInit):
self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
self.assertEqual(self.ha.dcs.write_sync_state.call_count, 1)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, None, 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': 0})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': 0})
self.assertEqual(mock_set_sync.call_count, 1)
self.assertEqual(mock_set_sync.call_args_list[0][0], (None,))
@@ -1496,7 +1497,7 @@ class TestHa(PostgresInit):
self.ha.run_cycle()
self.assertEqual(mock_write_sync.call_count, 1)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, None, 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': None})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': None})
self.assertEqual(mock_set_sync.call_count, 0)
mock_write_sync = self.ha.dcs.write_sync_state = Mock(side_effect=[SyncState.empty(), None])
@@ -1505,9 +1506,9 @@ class TestHa(PostgresInit):
self.ha.run_cycle()
self.assertEqual(mock_write_sync.call_count, 2)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, None, 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': None})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': None})
self.assertEqual(mock_write_sync.call_args_list[1][0], (self.p.name, CaseInsensitiveSet(['other']), 0))
self.assertEqual(mock_write_sync.call_args_list[1][1], {'index': None})
self.assertEqual(mock_write_sync.call_args_list[1][1], {'version': None})
self.assertEqual(mock_set_sync.call_count, 0)
self.p.sync_handler.current_state = Mock(side_effect=[_SyncState('quorum', 1, 0, CaseInsensitiveSet(['foo']),
@@ -1524,7 +1525,7 @@ class TestHa(PostgresInit):
self.ha.run_cycle()
self.assertEqual(mock_write_sync.call_count, 1)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, CaseInsensitiveSet(), 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': 0})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': 0})
self.assertEqual(mock_set_sync.call_count, 1)
self.assertEqual(mock_set_sync.call_args_list[0][0], ('ANY 1 (other)',))
@@ -1539,6 +1540,6 @@ class TestHa(PostgresInit):
self.ha.run_cycle()
self.assertEqual(mock_write_sync.call_count, 1)
self.assertEqual(mock_write_sync.call_args_list[0][0], (self.p.name, CaseInsensitiveSet(), 0))
self.assertEqual(mock_write_sync.call_args_list[0][1], {'index': 0})
self.assertEqual(mock_write_sync.call_args_list[0][1], {'version': 0})
self.assertEqual(mock_set_sync.call_count, 1)
self.assertEqual(mock_set_sync.call_args_list[0][0], ('ANY 1 (*)',))
tests/test_postgresql.py
+2 -1
View File
@@ -339,7 +339,8 @@ class TestPostgresql(BaseTestPostgresql):
@patch.object(Postgresql, 'start', Mock())
def test_follow(self):
self.p.call_nowait(CallbackAction.ON_START)
m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
m = RemoteMember.from_name_and_data('1', {'restore_command': '2', 'primary_slot_name': 'foo',
'conn_kwargs': {'host': 'bar'}})
self.p.follow(m)
with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)):
self.assertIsNone(self.p.follow(m))
tests/test_validator.py
+33 -11
View File
@@ -7,7 +7,7 @@ import unittest
from io import StringIO
from mock import Mock, patch, mock_open
from patroni.dcs import dcs_modules
from patroni.validator import schema
from patroni.validator import schema, Directory, Schema
available_dcs = [m.split(".")[-1] for m in dcs_modules()]
config = {
@@ -92,6 +92,16 @@ config = {
}
}
config_2 = {
"some_dir": "very_interesting_dir"
}
schema2 = Schema({
"some_dir": Directory(contains=["very_interesting_subdir", "another_interesting_subdir"])
})
required_binaries = ["pg_ctl", "initdb", "pg_controldata", "pg_basebackup", "postgres", "pg_isready"]
directories = []
files = []
binaries = []
@@ -190,6 +200,14 @@ class TestValidator(unittest.TestCase):
self.assertEqual(['consul.host', 'etcd.host', 'postgresql.bin_dir', 'postgresql.data_dir', 'postgresql.listen',
'raft.bind_addr', 'raft.self_addr', 'restapi.connect_address'], parse_output(output))
def test_bin_dir_is_empty_string_excutables_in_path(self, mock_out, mock_err):
binaries.extend(required_binaries)
c = copy.deepcopy(config)
c["postgresql"]["bin_dir"] = ""
errors = schema(c)
output = "\n".join(errors)
self.assertEqual(['raft.bind_addr', 'raft.self_addr'], parse_output(output))
@patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 12.1"))
def test_data_dir_contains_pg_version(self, mock_out, mock_err):
directories.append(config["postgresql"]["data_dir"])
@@ -197,14 +215,11 @@ class TestValidator(unittest.TestCase):
directories.append(os.path.join(config["postgresql"]["data_dir"], "pg_wal"))
files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "pg_ctl"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "initdb"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "pg_controldata"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "pg_basebackup"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "postgres"))
binaries.append(os.path.join(config["postgresql"]["bin_dir"], "pg_isready"))
binaries.extend(required_binaries)
c = copy.deepcopy(config)
c["postgresql"]["bin_dir"] = "" # to cover postgres --version call from PATH
with patch('patroni.validator.open', mock_open(read_data='12')):
errors = schema(config)
errors = schema(c)
output = "\n".join(errors)
self.assertEqual(['raft.bind_addr', 'raft.self_addr'], parse_output(output))
@@ -215,10 +230,10 @@ class TestValidator(unittest.TestCase):
directories.append(os.path.join(config["postgresql"]["data_dir"], "pg_wal"))
files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
binaries.extend([os.path.join(config["postgresql"]["bin_dir"], i) for i in required_binaries])
c = copy.deepcopy(config)
c["etcd"]["hosts"] = []
c["postgresql"]["listen"] = '127.0.0.2,*:543'
del c["postgresql"]["bin_dir"]
with patch('patroni.validator.open', mock_open(read_data='11')):
errors = schema(c)
output = "\n".join(errors)
@@ -227,18 +242,19 @@ class TestValidator(unittest.TestCase):
@patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 12.1"))
def test_pg_wal_doesnt_exist(self, mock_out, mock_err):
binaries.extend([os.path.join(config["postgresql"]["bin_dir"], i) for i in required_binaries])
directories.append(config["postgresql"]["data_dir"])
directories.append(config["postgresql"]["bin_dir"])
files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
c = copy.deepcopy(config)
del c["postgresql"]["bin_dir"]
with patch('patroni.validator.open', mock_open(read_data='11')):
errors = schema(c)
output = "\n".join(errors)
self.assertEqual(['postgresql.data_dir', 'raft.bind_addr', 'raft.self_addr'], parse_output(output))
def test_data_dir_is_empty_string(self, mock_out, mock_err):
binaries.extend(required_binaries)
directories.append(config["postgresql"]["data_dir"])
directories.append(config["postgresql"]["bin_dir"])
c = copy.deepcopy(config)
@@ -248,5 +264,11 @@ class TestValidator(unittest.TestCase):
c["postgresql"]["bin_dir"] = ""
errors = schema(c)
output = "\n".join(errors)
self.assertEqual(['kubernetes', 'postgresql.bin_dir', 'postgresql.data_dir',
self.assertEqual(['kubernetes', 'postgresql.data_dir',
'postgresql.pg_hba', 'raft.bind_addr', 'raft.self_addr'], parse_output(output))
def test_directory_contains(self, mock_out, mock_err):
directories.extend([config_2["some_dir"], os.path.join(config_2["some_dir"], "very_interesting_subdir")])
errors = schema2(config_2)
output = "\n".join(errors)
self.assertEqual(['some_dir'], parse_output(output))