diff --git a/.travis.yml b/.travis.yml index e7111e8b..3cda4951 100644 --- a/.travis.yml +++ b/.travis.yml @@ -7,16 +7,17 @@ addons: - expect-dev # for unbuffer env: global: - - ETCDVERSION=3.0.17 ZKVERSION=3.4.14 CONSULVERSION=0.7.4 - - PYVERSIONS="2.7 3.5 3.6" - - EXCLUDE_BEHAVE="3.5" + - ETCDVERSION=3.3.13 ZKVERSION=3.4.14 CONSULVERSION=0.7.4 + - PYVERSIONS="2.7 3.6" - BOTO_CONFIG=/doesnotexist matrix: include: - python: "3.5" - env: TEST_SUITE="python setup.py" + env: PYVERSIONS="2.7 3.5 3.6" TEST_SUITE="python setup.py" - python: "3.6" env: DCS="etcd" TEST_SUITE="behave" + - python: "3.6" + env: DCS="etcd3" TEST_SUITE="behave" - python: "3.6" env: DCS="exhibitor" TEST_SUITE="behave" - python: "3.6" @@ -36,10 +37,8 @@ before_cache: - | rm -fr $HOME/mycache/python* for pv in $PYVERSIONS; do - if [[ $TEST_SUITE != "behave" || $pv != $EXCLUDE_BEHAVE ]]; then - fpv=$(basename $(readlink $HOME/virtualenv/python${pv})) - mv $HOME/virtualenv/${fpv} $HOME/mycache/${fpv} - fi + fpv=$(basename $(readlink $HOME/virtualenv/python${pv})) + mv $HOME/virtualenv/${fpv} $HOME/mycache/${fpv} done install: - | @@ -60,6 +59,7 @@ install: function get_etcd() { EC=~/mycache/etcd_${ETCDVERSION} if [[ ! -x $EC ]]; then + rm -rf ~/mycache/etcd_* curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \ | tar xz -C . --strip=1 --wildcards --no-anchored etcd [[ ${PIPESTATUS[0]} == 0 ]] || return 1 @@ -68,6 +68,10 @@ install: ln -s $EC etcd } + function get_etcd3() { + get_etcd + } + function get_kubernetes() { wget -O localkube "https://storage.googleapis.com/minikube/k8sReleases/v1.7.0/localkube-linux-amd64" chmod +x localkube @@ -117,41 +121,38 @@ install: fi for pv in $PYVERSIONS; do - if [[ $TEST_SUITE != "behave" || $pv != $EXCLUDE_BEHAVE ]]; then - fpv=$(basename $(readlink $HOME/virtualenv/python$pv)) - if [[ -d ~/mycache/${fpv} ]]; then - mv ~/virtualenv/${fpv} ~/virtualenv/${fpv}.bckp - mv ~/mycache/${fpv} ~/virtualenv/${fpv} - fi - source ~/virtualenv/python${pv}/bin/activate - # explicitly install all needed python modules to cache them - for p in '-r requirements.txt' 'psycopg2-binary behave codacy-coverage coverage coveralls flake8 mock pytest-cov pytest setuptools'; do - pip install $p --upgrade - done + fpv=$(basename $(readlink $HOME/virtualenv/python$pv)) + if [[ -d ~/mycache/${fpv} ]]; then + mv ~/virtualenv/${fpv} ~/virtualenv/${fpv}.bckp + mv ~/mycache/${fpv} ~/virtualenv/${fpv} fi + source ~/virtualenv/python${pv}/bin/activate + # explicitly install all needed python modules to cache them + for p in '-r requirements.txt' 'psycopg2-binary behave codacy-coverage coverage coveralls flake8 mock pytest-cov pytest setuptools'; do + pip install $p --upgrade + done done script: - | for pv in $PYVERSIONS; do - if [[ $TEST_SUITE == "behave" && $pv == $EXCLUDE_BEHAVE ]]; then - continue - fi - source ~/virtualenv/python${pv}/bin/activate - if [[ $TEST_SUITE != "behave" ]]; then - echo Running unit tests using python${pv} - unbuffer $TEST_SUITE test - $TEST_SUITE flake8 - elif [[ $pv != $EXCLUDE_BEHAVE ]]; then - echo Running acceptance tests using python${pv} + if [[ $TEST_SUITE = "behave" ]]; then + echo Running integration tests using python${pv} if ! PATH=.:/usr/lib/postgresql/9.6/bin:$PATH unbuffer $TEST_SUITE; then # output all log files when tests are failing grep . features/output/*_failed/*postgres?.* exit 1 fi + else + echo Running unit tests using python${pv} + unbuffer $TEST_SUITE test + $TEST_SUITE flake8 fi + mv .coverage /tmp/.coverage.$pv done + mv /tmp/.coverage.* . + python -m coverage combine set +e after_success: diff --git a/Dockerfile b/Dockerfile index a5e23e33..824865e0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,7 +14,7 @@ ARG PGDATA ARG LC_ALL ARG LANG -ENV ETCDVERSION=2.3.8 CONFDVERSION=0.16.0 +ENV ETCDVERSION=3.3.13 CONFDVERSION=0.16.0 RUN set -ex \ && export DEBIAN_FRONTEND=noninteractive \ diff --git a/README.rst b/README.rst index d305f8ba..55b44b10 100644 --- a/README.rst +++ b/README.rst @@ -96,7 +96,7 @@ Patroni can be installed with pip: where dependencies can be either empty, or consist of one or more of the following: -etcd +etcd or etcd3 `python-etcd` module in order to use Etcd as DCS consul `python-consul` module in order to use Consul as DCS @@ -106,6 +106,8 @@ exhibitor `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper) kubernetes `kubernetes` module in order to use Kubernetes as DCS in Patroni +raft + `pysyncobj` module in order to use python Raft implementation as DCS aws `boto` in order to use AWS callbacks diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 898bb4b7..6892f938 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -25,7 +25,7 @@ case "$1" in while ! etcdctl cluster-health 2> /dev/null; do sleep 1 done - exec dumb-init $CONFD etcd -node $(echo $ETCDCTL_ENDPOINTS | sed 's/,/ -node /g') + exec dumb-init $CONFD etcdv3 -node $(echo $ETCDCTL_ENDPOINTS | sed 's/,/ -node /g') fi ;; etcd) @@ -37,7 +37,7 @@ case "$1" in esac ## We start an etcd -if [ -z "$PATRONI_ETCD_HOSTS" ] && [ -z "$PATRONI_ZOOKEEPER_HOSTS" ]; then +if [ -z "$PATRONI_ETCD3_HOSTS" ] && [ -z "$PATRONI_ZOOKEEPER_HOSTS" ]; then export PATRONI_ETCD_URL="http://127.0.0.1:2379" etcd --data-dir /tmp/etcd.data -advertise-client-urls=$PATRONI_ETCD_URL -listen-client-urls=http://0.0.0.0:2379 > /var/log/etcd.log 2> /var/log/etcd.err & fi diff --git a/docker/patroni.env b/docker/patroni.env index e2993a4d..10df2eb4 100644 --- a/docker/patroni.env +++ b/docker/patroni.env @@ -1,5 +1,5 @@ PATRONI_SCOPE=demo -PATRONI_ETCD_HOSTS='etcd1:2379','etcd2:2379','etcd3:2379' +PATRONI_ETCD3_HOSTS='etcd1:2379','etcd2:2379','etcd3:2379' PATRONI_RESTAPI_USERNAME=admin PATRONI_RESTAPI_PASSWORD=admin diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index c86a35f8..3f125909 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -66,6 +66,14 @@ Etcd - **PATRONI\_ETCD\_CERT**: File with the client certificate. - **PATRONI\_ETCD\_KEY**: File with the client key. Can be empty if the key is part of certificate. +Etcdv3 +------ +Environment names for Etcdv3 are similar as for Etcd, you just need to use ``ETCD3`` instead of ``ETCD`` in the variable name. Example: ``PATRONI_ETCD3_HOST``, ``PATRONI_ETCD3_CACERT``, and so on. + +.. warning:: + Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from Etcd to Etcdv3 just by updating Patroni configuration. + + ZooKeeper --------- - **PATRONI\_ZOOKEEPER\_HOSTS**: comma separated list of ZooKeeper cluster members: "'host1:port1','host2:port2','etc...'". It is important to quote every single entity! diff --git a/docs/README.rst b/docs/README.rst index 3fbc4366..405d54d1 100644 --- a/docs/README.rst +++ b/docs/README.rst @@ -72,7 +72,7 @@ Patroni can be installed with pip: where dependencies can be either empty, or consist of one or more of the following: -etcd +etcd or etcd3 `python-etcd` module in order to use Etcd as DCS consul `python-consul` module in order to use Consul as DCS @@ -82,6 +82,8 @@ exhibitor `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper) kubernetes `kubernetes` module in order to use Kubernetes as DCS in Patroni +raft + `pysyncobj` module in order to use python Raft implementation as DCS aws `boto` in order to use AWS callbacks diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index abc308e8..f8004bac 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -137,6 +137,14 @@ Most of the parameters are optional, but you have to specify one of the **host** - **cert**: (optional) file with the client certificate. - **key**: (optional) file with the client key. Can be empty if the key is part of **cert**. +Etcdv3 +------ +If you want that Patroni works with Etcd cluster via protocol version 3, you need to use the ``etcd3`` section in the Patroni configuration file. All configuration parameters are the same as for ``etcd``. + +.. warning:: + Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from ``etcd`` to ``etcd3`` just by updating Patroni config file. + + ZooKeeper ---------- - **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...']. diff --git a/features/environment.py b/features/environment.py index 563ea0c6..13f37aa2 100644 --- a/features/environment.py +++ b/features/environment.py @@ -378,21 +378,36 @@ class ConsulController(AbstractDcsController): super(ConsulController, self).start(max_wait_limit) -class EtcdController(AbstractDcsController): +class AbstractEtcdController(AbstractDcsController): """ handles all etcd related tasks, used for the tests setup and cleanup """ - def __init__(self, context): - super(EtcdController, self).__init__(context) - os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379' - - import etcd - self._client = etcd.Client(port=2379) + def __init__(self, context, client_cls): + super(AbstractEtcdController, self).__init__(context) + self._client_cls = client_cls def _start(self): return subprocess.Popen(["etcd", "--debug", "--data-dir", self._work_directory], stdout=self._log, stderr=subprocess.STDOUT) + def _is_running(self): + from patroni.dcs.etcd import DnsCachingResolver + # if etcd is running, but we didn't start it + try: + self._client = self._client_cls({'host': 'localhost', 'port': 2379, 'retry_timeout': 30, + 'patronictl': 1}, DnsCachingResolver()) + return True + except Exception: + return False + + +class EtcdController(AbstractEtcdController): + + def __init__(self, context): + from patroni.dcs.etcd import EtcdClient + super(EtcdController, self).__init__(context, EtcdClient) + os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379' + def query(self, key, scope='batman'): import etcd try: @@ -409,12 +424,25 @@ class EtcdController(AbstractDcsController): except Exception as e: assert False, "exception when cleaning up etcd contents: {0}".format(e) - def _is_running(self): - # if etcd is running, but we didn't start it + +class Etcd3Controller(AbstractEtcdController): + + def __init__(self, context): + from patroni.dcs.etcd3 import Etcd3Client + super(Etcd3Controller, self).__init__(context, Etcd3Client) + os.environ['PATRONI_ETCD3_HOST'] = 'localhost:2379' + + def query(self, key, scope='batman'): + import base64 + response = self._client.range(self.path(key, scope)) + for k in response.get('kvs', []): + return base64.b64decode(k['value']).decode('utf-8') if 'value' in k else None + + def cleanup_service_tree(self): try: - return bool(self._client.machines) - except Exception: - return False + self._client.deleteprefix(self.path(scope='')) + except Exception as e: + assert False, "exception when cleaning up etcd contents: {0}".format(e) class KubernetesController(AbstractDcsController): diff --git a/patroni/config.py b/patroni/config.py index 7644a8e7..8fe65307 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -316,8 +316,9 @@ class Config(object): value = parse_bool(value) if value: ret[name.lower()][suffix.lower()] = value - if 'etcd' in ret: - ret['etcd'].update(_get_auth('etcd')) + for dcs in ('etcd', 'etcd3'): + if dcs in ret: + ret[dcs].update(_get_auth(dcs)) users = {} for param in list(os.environ.keys()): diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 51110a90..a7077be7 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -1,4 +1,5 @@ from __future__ import absolute_import +import abc import etcd import json import logging @@ -82,7 +83,8 @@ class DnsCachingResolver(Thread): return [] -class Client(etcd.Client): +@six.add_metaclass(abc.ABCMeta) +class AbstractEtcdClientWithFailover(etcd.Client): def __init__(self, config, dns_resolver, cache_ttl=300): self._dns_resolver = dns_resolver @@ -90,7 +92,7 @@ class Client(etcd.Client): self._machines_cache_updated = 0 args = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'username', 'password', 'cert', 'ca_cert') if config.get(p)} - super(Client, self).__init__(read_timeout=config['retry_timeout'], **args) + super(AbstractEtcdClientWithFailover, self).__init__(read_timeout=config['retry_timeout'], **args) # For some reason python3-etcd on debian and ubuntu are not based on the latest version # Workaround for the case when https://github.com/jplana/python-etcd/pull/196 is not applied self.http.connection_pool_kw.pop('ssl_version', None) @@ -130,12 +132,16 @@ class Client(etcd.Client): return etcd_nodes, per_node_timeout, per_node_retries - 1 + def reload_config(self, config): + self.username = config.get('username') + self.password = config.get('password') + def _get_headers(self): basic_auth = ':'.join((self.username, self.password)) if self.username and self.password else None return urllib3.make_headers(basic_auth=basic_auth, user_agent=USER_AGENT) - def _build_request_parameters(self, etcd_nodes, timeout=None): - kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect} + def _prepare_common_parameters(self, etcd_nodes, timeout=None): + kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect, 'preload_content': False} if timeout is not None: kwargs.update(retries=0, timeout=timeout) @@ -148,6 +154,14 @@ class Client(etcd.Client): def set_machines_cache_ttl(self, cache_ttl): self._machines_cache_ttl = cache_ttl + @abc.abstractmethod + def _prepare_get_members(self, etcd_nodes): + """returns: request parameters""" + + @abc.abstractmethod + def _get_members(self, base_uri, **kwargs): + """returns: list of clientURLs""" + @property def machines_cache(self): base_uri, cache = self._base_uri, self._machines_cache @@ -166,13 +180,11 @@ class Client(etcd.Client): the original method was retrying 2 times with the `read_timeout` on each node.""" machines_cache = self.machines_cache - kwargs = self._build_request_parameters(len(machines_cache)) + kwargs = self._prepare_get_members(len(machines_cache)) for base_uri in machines_cache: try: - response = self.http.request(self._MGET, base_uri + self.version_prefix + '/machines', **kwargs) - data = self._handle_server_response(response).data.decode('utf-8') - machines = [m.strip() for m in data.split(',') if m.strip()] + machines = list(self._get_members(base_uri, **kwargs)) logger.debug("Retrieved list of machines: %s", machines) if machines: random.shuffle(machines) @@ -188,14 +200,15 @@ class Client(etcd.Client): self._read_timeout = timeout def _do_http_request(self, retry, machines_cache, request_executor, method, path, fields=None, **kwargs): + if fields is not None: + kwargs['fields'] = fields some_request_failed = False for i, base_uri in enumerate(machines_cache): if i > 0: logger.info("Retrying on %s", base_uri) try: - response = request_executor(method, base_uri + path, fields=fields, **kwargs) + response = request_executor(method, base_uri + path, **kwargs) response.data.decode('utf-8') - self._check_cluster_id(response) if some_request_failed: self.set_base_uri(base_uri) self._refresh_machines_cache() @@ -218,20 +231,12 @@ class Client(etcd.Client): raise etcd.EtcdConnectionFailed('No more machines in the cluster') + @abc.abstractmethod + def _prepare_request(self, kwargs, params=None, method=None): + """returns: request_executor""" + def api_execute(self, path, method, params=None, timeout=None): - if not path.startswith('/'): - raise ValueError('Path does not start with /') - retry = params.pop('retry', None) if isinstance(params, dict) else None - kwargs = {'fields': params, 'preload_content': False} - - if method in [self._MGET, self._MDELETE]: - request_executor = self.http.request - elif method in [self._MPUT, self._MPOST]: - request_executor = self.http.request_encode_body - kwargs['encode_multipart'] = False - else: - raise etcd.EtcdException('HTTP method {0} not supported'.format(method)) # Update machines_cache if previous attempt of update has failed if self._update_machines_cache: @@ -241,7 +246,9 @@ class Client(etcd.Client): machines_cache = self.machines_cache etcd_nodes = len(machines_cache) - kwargs.update(self._build_request_parameters(etcd_nodes, timeout)) + + kwargs = self._prepare_common_parameters(etcd_nodes, timeout) + request_executor = self._prepare_request(kwargs, params, method) while True: try: @@ -389,17 +396,47 @@ class Client(etcd.Client): self._base_uri = value -class Etcd(AbstractDCS): +class EtcdClient(AbstractEtcdClientWithFailover): - def __init__(self, config): - super(Etcd, self).__init__(config) - self._ttl = int(config.get('ttl') or 30) + ERROR_CLS = EtcdError + + def __del__(self): + if self.http is not None: + try: + self.http.clear() + except (ReferenceError, TypeError): + pass + + def _prepare_get_members(self, etcd_nodes): + return self._prepare_common_parameters(etcd_nodes) + + def _get_members(self, base_uri, **kwargs): + response = self.http.request(self._MGET, base_uri + self.version_prefix + '/machines', **kwargs) + data = self._handle_server_response(response).data.decode('utf-8') + return [m.strip() for m in data.split(',') if m.strip()] + + def _prepare_request(self, kwargs, params=None, method=None): + kwargs['fields'] = params + if method in (self._MPOST, self._MPUT): + kwargs['encode_multipart'] = False + return self.http.request + + +class AbstractEtcd(AbstractDCS): + + def __init__(self, config, client_cls, retry_errors_cls): + super(AbstractEtcd, self).__init__(config) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, - retry_exceptions=(etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) - self._client = self.get_etcd_client(config) + retry_exceptions=retry_errors_cls) + self._ttl = int(config.get('ttl') or 30) + self._client = self.get_etcd_client(config, client_cls) self.__do_not_watch = False self._has_failed = False + def reload_config(self, config): + super(AbstractEtcd, self).reload_config(config) + self._client.reload_config(config.get(self.__class__.__name__.lower(), {})) + def retry(self, *args, **kwargs): retry = self._retry.copy() kwargs['retry'] = retry @@ -416,22 +453,13 @@ class Etcd(AbstractDCS): if isinstance(raise_ex, Exception): raise raise_ex - def catch_etcd_errors(func): - def wrapper(self, *args, **kwargs): - try: - retval = func(self, *args, **kwargs) is not None - 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=EtcdError('unexpected error')) - - return wrapper - @staticmethod - def get_etcd_client(config): + def set_socket_options(sock, socket_options): + if socket_options: + for opt in socket_options: + sock.setsockopt(*opt) + + def get_etcd_client(self, config, client_cls): if 'proxy' in config: config['use_proxies'] = True config['url'] = config['proxy'] @@ -480,9 +508,7 @@ class Etcd(AbstractDCS): sock = None try: sock = socket.socket(af, socktype, proto) - if socket_options: - for opt in socket_options: - sock.setsockopt(*opt) + self.set_socket_options(sock, socket_options) if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT: sock.settimeout(timeout) if source_address: @@ -506,7 +532,7 @@ class Etcd(AbstractDCS): client = None while not client: try: - client = Client(config, dns_resolver) + client = client_cls(config, dns_resolver) if 'use_proxies' in config and not client.machines: raise etcd.EtcdException except etcd.EtcdException: @@ -516,9 +542,10 @@ class Etcd(AbstractDCS): def set_ttl(self, ttl): ttl = int(ttl) - self.__do_not_watch = self._ttl != ttl + ret = self._ttl != ttl self._ttl = ttl self._client.set_machines_cache_ttl(ttl*10) + return ret @property def ttl(self): @@ -528,6 +555,31 @@ class Etcd(AbstractDCS): self._retry.deadline = retry_timeout self._client.set_read_timeout(retry_timeout) + +def catch_etcd_errors(func): + def wrapper(self, *args, **kwargs): + try: + retval = func(self, *args, **kwargs) is not None + 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 wrapper + + +class Etcd(AbstractEtcd): + + def __init__(self, config): + super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) + self.__do_not_watch = False + + def set_ttl(self, ttl): + self.__do_not_watch = super(Etcd, self).set_ttl(ttl) + @staticmethod def member(node): return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py new file mode 100644 index 00000000..1e84d5ce --- /dev/null +++ b/patroni/dcs/etcd3.py @@ -0,0 +1,795 @@ +from __future__ import absolute_import +import base64 +import etcd +import json +import logging +import os +import six +import socket +import sys +import time +import urllib3 + +from threading import Condition, Lock, Thread + +from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory +from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors +from ..exceptions import DCSError, PatroniException +from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT + +logger = logging.getLogger(__name__) + + +class Etcd3Error(DCSError): + pass + + +class UnsupportedEtcdVersion(PatroniException): + pass + + +# google.golang.org/grpc/codes +GRPCCode = type('Enum', (), {'OK': 0, 'Canceled': 1, 'Unknown': 2, 'InvalidArgument': 3, 'DeadlineExceeded': 4, + 'NotFound': 5, 'AlreadyExists': 6, 'PermissionDenied': 7, 'ResourceExhausted': 8, + 'FailedPrecondition': 9, 'Aborted': 10, 'OutOfRange': 11, 'Unimplemented': 12, + 'Internal': 13, 'Unavailable': 14, 'DataLoss': 15, 'Unauthenticated': 16}) +GRPCcodeToText = {v: k for k, v in GRPCCode.__dict__.items() if not k.startswith('__') and isinstance(v, int)} + + +class Etcd3Exception(etcd.EtcdException): + pass + + +class Etcd3ClientError(Etcd3Exception): + + def __init__(self, code=None, error=None, status=None): + if not hasattr(self, 'error'): + self.error = error and error.strip() + self.codeText = GRPCcodeToText.get(code) + self.status = status + + def __repr__(self): + return "<{0} error: '{1}', code: {2}>".format(self.__class__.__name__, self.error, self.code) + + __str__ = __repr__ + + def as_dict(self): + return {'error': self.error, 'code': self.code, 'codeText': self.codeText, 'status': self.status} + + @classmethod + def get_subclasses(cls): + for subclass in cls.__subclasses__(): + for subsubclass in subclass.get_subclasses(): + yield subsubclass + yield subclass + + +class Unknown(Etcd3ClientError): + code = GRPCCode.Unknown + + +class InvalidArgument(Etcd3ClientError): + code = GRPCCode.InvalidArgument + + +class DeadlineExceeded(Etcd3ClientError): + code = GRPCCode.DeadlineExceeded + error = "context deadline exceeded" + + +class NotFound(Etcd3ClientError): + code = GRPCCode.NotFound + + +class FailedPrecondition(Etcd3ClientError): + code = GRPCCode.FailedPrecondition + + +class Unavailable(Etcd3ClientError): + code = GRPCCode.Unavailable + + +# https://github.com/etcd-io/etcd/blob/master/etcdserver/api/v3rpc/rpctypes/error.go +class LeaseNotFound(NotFound): + error = "etcdserver: requested lease not found" + + +class UserEmpty(InvalidArgument): + error = "etcdserver: user name is empty" + + +class PermissionDenied(Etcd3ClientError): + code = GRPCCode.PermissionDenied + error = "etcdserver: permission denied" + + +class AuthNotEnabled(FailedPrecondition): + error = "etcdserver: authentication is not enabled" + + +class InvalidAuthToken(Etcd3ClientError): + code = GRPCCode.Unauthenticated + error = "etcdserver: invalid auth token" + + +errStringToClientError = {s.error: s for s in Etcd3ClientError.get_subclasses() if hasattr(s, 'error')} +errCodeToClientError = {s.code: s for s in Etcd3ClientError.__subclasses__()} + + +def _raise_for_data(data, status_code=None): + try: + error = data.get('error') or data.get('Error') + if isinstance(error, dict): # streaming response + status_code = error.get('http_code') + code = error['grpc_code'] + error = error['message'] + else: + code = data.get('code') or data.get('Code') + except Exception: + error = str(data) + code = GRPCCode.Unknown + err = errStringToClientError.get(error) or errCodeToClientError.get(code) or Unknown + raise err(code, error, status_code) + + +def to_bytes(v): + return v if isinstance(v, bytes) else v.encode('utf-8') + + +def prefix_range_end(v): + v = bytearray(to_bytes(v)) + for i in range(len(v) - 1, -1, -1): + if v[i] < 0xff: + v[i] += 1 + break + return bytes(v) + + +def base64_encode(v): + return base64.b64encode(to_bytes(v)).decode('utf-8') + + +def base64_decode(v): + return base64.b64decode(v).decode('utf-8') + + +def build_range_request(key, range_end=None): + fields = {'key': base64_encode(key)} + if range_end: + fields['range_end'] = base64_encode(range_end) + return fields + + +class Etcd3Client(AbstractEtcdClientWithFailover): + + ERROR_CLS = Etcd3Error + + def __init__(self, config, dns_resolver, cache_ttl=300): + self._token = None + self._cluster_version = None + self.version_prefix = '/v3beta' + super(Etcd3Client, self).__init__(config, dns_resolver, cache_ttl) + + if six.PY2: # pragma: no cover + # Old grpc-gateway sometimes sends double 'transfer-encoding: chunked' headers, + # what breaks the old (python2.7) httplib.HTTPConnection (it closes the socket). + def dedup_addheader(httpm, key, value): + prev = httpm.dict.get(key) + if prev is None: + httpm.dict[key] = value + elif key != 'transfer-encoding' or prev != value: + combined = ", ".join((prev, value)) + httpm.dict[key] = combined + + import httplib + httplib.HTTPMessage.addheader = dedup_addheader + + try: + self.authenticate() + except Exception as e: + logger.fatal('Etcd3 authentication failed: %r', e) + sys.exit(1) + + def _get_headers(self): + headers = urllib3.make_headers(user_agent=USER_AGENT) + if self._token and self._cluster_version >= (3, 3, 0): + headers['authorization'] = self._token + return headers + + def _prepare_request(self, kwargs, params=None, method=None): + if params is not None: + kwargs['body'] = json.dumps(params) + kwargs['headers']['Content-Type'] = 'application/json' + return self.http.urlopen + + @staticmethod + def _handle_server_response(response): + data = response.data + try: + data = data.decode('utf-8') + data = json.loads(data) + except (TypeError, ValueError, UnicodeError) as e: + if response.status < 400: + raise etcd.EtcdException('Server response was not valid JSON: %r' % e) + if response.status < 400: + return data + _raise_for_data(data, response.status) + + def _ensure_version_prefix(self, base_uri, **kwargs): + if self.version_prefix != '/v3': + response = self.http.urlopen(self._MGET, base_uri + '/version', **kwargs) + response = self._handle_server_response(response) + + server_version_str = response['etcdserver'] + server_version = tuple(int(x) for x in server_version_str.split('.')) + cluster_version_str = response['etcdcluster'] + self._cluster_version = tuple(int(x) for x in cluster_version_str.split('.')) + + if self._cluster_version < (3, 0) or server_version < (3, 0, 4): + raise UnsupportedEtcdVersion('Detected Etcd version {0} is lower than 3.0.4'.format(server_version_str)) + + if self._cluster_version < (3, 3): + if self.version_prefix != '/v3alpha': + if self._cluster_version < (3, 1): + logger.warning('Detected Etcd version %s is lower than 3.1.0, watches are not supported', + cluster_version_str) + if self.username and self.password: + logger.warning('Detected Etcd version %s is lower than 3.3.0, authentication is not supported', + cluster_version_str) + self.version_prefix = '/v3alpha' + elif self._cluster_version < (3, 4): + self.version_prefix = '/v3beta' + else: + self.version_prefix = '/v3' + + def _prepare_get_members(self, etcd_nodes): + kwargs = self._prepare_common_parameters(etcd_nodes) + self._prepare_request(kwargs, {}) + return kwargs + + def _get_members(self, base_uri, **kwargs): + self._ensure_version_prefix(base_uri, **kwargs) + resp = self.http.urlopen(self._MPOST, base_uri + self.version_prefix + '/cluster/member/list', **kwargs) + members = self._handle_server_response(resp)['members'] + return set(url for member in members for url in member.get('clientURLs', [])) + + def call_rpc(self, method, fields, retry=None): + fields['retry'] = retry + return self.api_execute(self.version_prefix + method, self._MPOST, fields) + + def authenticate(self): + if self._cluster_version >= (3, 3) and self.username and self.password: + logger.info('Trying to authenticate on Etcd...') + old_token, self._token = self._token, None + try: + response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password}) + except AuthNotEnabled: + logger.info('Etcd authentication is not enabled') + self._token = None + except Exception: + self._token = old_token + raise + else: + self._token = response.get('token') + return old_token != self._token + + def _handle_auth_errors(func): + def wrapper(self, *args, **kwargs): + def retry(ex): + 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 wrapper + + @_handle_auth_errors + def range(self, key, range_end=None, retry=None): + params = build_range_request(key, range_end) + params['serializable'] = True # For better performance. We can tolerate stale reads. + return self.call_rpc('/kv/range', params, retry) + + def prefix(self, key, retry=None): + return self.range(key, prefix_range_end(key), retry) + + def lease_grant(self, ttl, retry=None): + return self.call_rpc('/lease/grant', {'TTL': ttl}, retry)['ID'] + + def lease_keepalive(self, ID, retry=None): + return self.call_rpc('/lease/keepalive', {'ID': ID}, retry).get('result', {}).get('TTL') + + def txn(self, compare, success, retry=None): + return self.call_rpc('/kv/txn', {'compare': [compare], 'success': [success]}, retry).get('succeeded') + + @_handle_auth_errors + def put(self, key, value, lease=None, create_revision=None, mod_revision=None, retry=None): + fields = {'key': base64_encode(key), 'value': base64_encode(value)} + if lease: + fields['lease'] = lease + if create_revision is not None: + compare = {'target': 'CREATE', 'create_revision': create_revision} + elif mod_revision is not None: + compare = {'target': 'MOD', 'mod_revision': mod_revision} + else: + return self.call_rpc('/kv/put', fields, retry) + compare['key'] = fields['key'] + return self.txn(compare, {'request_put': fields}, retry) + + @_handle_auth_errors + def deleterange(self, key, range_end=None, mod_revision=None, retry=None): + fields = build_range_request(key, range_end) + if mod_revision is None: + return self.call_rpc('/kv/deleterange', fields, retry) + compare = {'target': 'MOD', 'mod_revision': mod_revision, 'key': fields['key']} + return self.txn(compare, {'request_delete_range': fields}, retry) + + def deleteprefix(self, key, retry=None): + return self.deleterange(key, prefix_range_end(key), retry=retry) + + def watchrange(self, key, range_end=None, start_revision=None, filters=None): + """returns: response object""" + params = build_range_request(key, range_end) + if start_revision is not None: + params['start_revision'] = start_revision + params['filters'] = filters or [] + kwargs = self._prepare_common_parameters(1, self.read_timeout) + request_executor = self._prepare_request(kwargs, {'create_request': params}) + kwargs.update(timeout=urllib3.Timeout(connect=kwargs['timeout']), retries=0) + return request_executor(self._MPOST, self._base_uri + self.version_prefix + '/watch', **kwargs) + + def watchprefix(self, key, start_revision=None, filters=None): + return self.watchrange(key, prefix_range_end(key), start_revision, filters) + + +class KVCache(Thread): + + def __init__(self, dcs, client): + Thread.__init__(self) + self.daemon = True + self._dcs = dcs + self._client = client + self.condition = Condition() + self._config_key = base64_encode(dcs.config_path) + self._leader_key = base64_encode(dcs.leader_path) + self._optime_key = base64_encode(dcs.leader_optime_path) + self._name = base64_encode(dcs._name) + self._is_ready = False + self._response = None + self._response_lock = Lock() + self._object_cache = {} + self._object_cache_lock = Lock() + self.start() + + def set(self, value, overwrite=False): + with self._object_cache_lock: + name = value['key'] + old_value = self._object_cache.get(name) + ret = not old_value or int(old_value['mod_revision']) < int(value['mod_revision']) + if ret or overwrite and old_value['mod_revision'] == value['mod_revision']: + self._object_cache[name] = value + return ret, old_value + + def delete(self, name, mod_revision): + with self._object_cache_lock: + old_value = self._object_cache.get(name) + ret = old_value and int(old_value['mod_revision']) < int(mod_revision) + if ret: + del self._object_cache[name] + return not old_value or ret, old_value + + def copy(self): + with self._object_cache_lock: + return [v.copy() for v in self._object_cache.values()] + + def get(self, name): + with self._object_cache_lock: + return self._object_cache.get(name) + + def _process_event(self, event): + kv = event['kv'] + key = kv['key'] + if event.get('type') == 'DELETE': + success, old_value = self.delete(key, kv['mod_revision']) + else: + success, old_value = self.set(kv, True) + + if success: + old_value = old_value and old_value.get('value') + new_value = kv.get('value') + + value_changed = old_value != new_value and \ + (key == self._leader_key or key == self._optime_key and new_value is not None or + key == self._config_key and old_value is not None and new_value is not None) + + if value_changed: + logger.debug('%s changed from %s to %s', key, old_value, new_value) + + # We also want to wake up HA loop on replicas if leader optime was updated + if value_changed and (key != self._optime_key or self.get(self._leader_key) != self._name): + self._dcs.event.set() + + def _process_message(self, message): + logger.debug('Received message: %s', message) + if 'error' in message: + _raise_for_data(message) + for event in message.get('result', {}).get('events', []): + self._process_event(event) + + @staticmethod + def _finish_response(response): + try: + response.close() + finally: + response.release_conn() + + def _do_watch(self, revision): + with self._response_lock: + self._response = None + response = self._client.watchprefix(self._dcs.cluster_prefix, revision) + with self._response_lock: + if self._response is None: + self._response = response + + if not self._response: + return self._finish_response(response) + + for message in iter_response_objects(response): + self._process_message(message) + + def _build_cache(self): + result = self._dcs.retry(self._client.prefix, self._dcs.cluster_prefix) + with self._object_cache_lock: + self._object_cache = {node['key']: node for node in result.get('kvs', [])} + with self.condition: + self._is_ready = True + self.condition.notify() + + try: + self._do_watch(result['header']['revision']) + except Exception as e: + logger.error('watchprefix failed: %r', e) + finally: + with self.condition: + self._is_ready = False + with self._response_lock: + response, self._response = self._response, None + if response: + self._finish_response(response) + + def run(self): + while True: + try: + self._build_cache() + except Exception as e: + logger.error('KVCache.run %r', e) + time.sleep(1) + + def kill_stream(self): + sock = None + with self._response_lock: + if self._response: + try: + sock = self._response.connection.sock + except Exception: + sock = None + else: + self._response = False + if sock: + try: + sock.shutdown(socket.SHUT_RDWR) + sock.close() + except Exception as e: + logger.debug('Error on socket.shutdown: %r', e) + + def is_ready(self): + """Must be called only when holding the lock on `condition`""" + return self._is_ready + + +class PatroniEtcd3Client(Etcd3Client): + + def __init__(self, *args, **kwargs): + self._kv_cache = None + super(PatroniEtcd3Client, self).__init__(*args, **kwargs) + + def configure(self, etcd3): + self._etcd3 = etcd3 + + def start_watcher(self): + if self._cluster_version >= (3, 1): + self._kv_cache = KVCache(self._etcd3, self) + + def _restart_watcher(self): + if self._kv_cache: + self._kv_cache.kill_stream() + + def set_base_uri(self, value): + super(PatroniEtcd3Client, self).set_base_uri(value) + self._restart_watcher() + + def authenticate(self): + ret = super(PatroniEtcd3Client, self).authenticate() + if ret: + self._restart_watcher() + return ret + + def _wait_cache(self, timeout): + stop_time = time.time() + timeout + while not self._kv_cache.is_ready(): + timeout = stop_time - time.time() + if timeout <= 0: + raise RetryFailedError('Exceeded retry deadline') + self._kv_cache.condition.wait(timeout) + + def get_cluster(self): + if self._kv_cache: + with self._kv_cache.condition: + self._wait_cache(self._etcd3._retry.deadline) + return self._kv_cache.copy() + else: + return self._etcd3.retry(self.prefix, self._etcd3.cluster_prefix).get('kvs', []) + + def call_rpc(self, method, fields, retry=None): + ret = super(PatroniEtcd3Client, self).call_rpc(method, fields, retry) + + if self._kv_cache: + value = delete = None + if method == '/kv/txn' and ret.get('succeeded'): + on_success = fields['success'][0] + value = on_success.get('request_put') + delete = on_success.get('request_delete_range') + elif method == '/kv/put' and ret: + value = fields + elif method == '/kv/deleterange' and ret: + delete = fields + + if value: + value['mod_revision'] = ret['header']['revision'] + self._kv_cache.set(value) + elif delete and 'range_end' not in delete: + self._kv_cache.delete(delete['key'], ret['header']['revision']) + + return ret + + +class Etcd3(AbstractEtcd): + + def __init__(self, config): + super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition)) + self.__do_not_watch = False + self._lease = None + self._last_lease_refresh = 0 + + self._client.configure(self) + if not self._ctl: + self._client.start_watcher() + self.create_lease() + + def set_socket_options(self, sock, socket_options): + enable_keepalive(sock, self.ttl, int(self.loop_wait + self._retry.deadline)) + + def set_ttl(self, ttl): + self.__do_not_watch = super(Etcd3, self).set_ttl(ttl) + if self.__do_not_watch: + self._lease = None + + def _do_refresh_lease(self, retry=None): + if self._lease and self._last_lease_refresh + self._loop_wait > time.time(): + return False + + if self._lease and not self._client.lease_keepalive(self._lease, retry): + self._lease = None + + ret = not self._lease + if ret: + self._lease = self._client.lease_grant(self._ttl, retry) + + self._last_lease_refresh = time.time() + return ret + + def refresh_lease(self): + try: + return self.retry(self._do_refresh_lease) + except (Etcd3ClientError, RetryFailedError): + logger.exception('refresh_lease') + raise Etcd3Error('Failed ro keepalive/grant lease') + + def create_lease(self): + while not self._lease: + try: + self.refresh_lease() + except Etcd3Error: + logger.info('waiting on etcd') + time.sleep(5) + + @property + def cluster_prefix(self): + return self.client_path('') + + @staticmethod + def member(node): + return Member.from_node(node['mod_revision'], os.path.basename(node['key']), node['lease'], node['value']) + + def _load_cluster(self): + cluster = None + try: + path_len = len(self.cluster_prefix) + + nodes = {} + for node in self._client.get_cluster(): + node['key'] = base64_decode(node['key']) + node['value'] = base64_decode(node.get('value', '')) + node['lease'] = node.get('lease') + nodes[node['key'][path_len:].lstrip('/')] = node + + # get initialize flag + initialize = nodes.get(self._INITIALIZE) + initialize = initialize and initialize['value'] + + # get global dynamic configuration + config = nodes.get(self._CONFIG) + config = config and ClusterConfig.from_node(config['mod_revision'], config['value']) + + # get timeline history + history = nodes.get(self._HISTORY) + history = history and TimelineHistory.from_node(history['mod_revision'], history['value']) + + # get last leader operation + last_leader_operation = nodes.get(self._LEADER_OPTIME) + last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value']) + + # get list of members + members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1] + + # get leader + leader = nodes.get(self._LEADER) + if not self._ctl and leader and leader['value'] == self._name and self._lease != leader.get('lease'): + logger.warning('I am the leader but not owner of the lease') + + if leader: + member = Member(-1, leader['value'], None, {}) + member = ([m for m in members if m.name == leader['value']] or [member])[0] + leader = Leader(leader['mod_revision'], leader['lease'], member) + + # failover key + failover = nodes.get(self._FAILOVER) + if failover: + failover = Failover.from_node(failover['mod_revision'], failover['value']) + + # get synchronization state + sync = nodes.get(self._SYNC) + sync = SyncState.from_node(sync and sync['mod_revision'], sync and sync['value']) + + cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + except UnsupportedEtcdVersion: + raise + except Exception as e: + self._handle_exception(e, 'get_cluster', raise_ex=Etcd3Error('Etcd is not responding properly')) + self._has_failed = False + return cluster + + @catch_etcd_errors + def touch_member(self, data, permanent=False): + if not permanent: + try: + self.refresh_lease() + except Etcd3Error: + return False + + cluster = self.cluster + member = cluster and cluster.get_member(self._name, fallback_to_leader=False) + + if member and member.session == self._lease and deep_compare(data, member.data): + return True + + data = json.dumps(data, separators=(',', ':')) + try: + return self._client.put(self.member_path, data, None if permanent else self._lease) + except LeaseNotFound: + self._lease = None + logger.error('Our lease disappeared from Etcd, can not "touch_member"') + + @catch_etcd_errors + def take_leader(self): + return self.retry(self._client.put, self.leader_path, self._name, self._lease) + + @catch_etcd_errors + def _do_attempt_to_acquire_leader(self, permanent): + try: + return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0) + except LeaseNotFound: + self._lease = None + logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt') + self.refresh_lease() + return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0) + + def attempt_to_acquire_leader(self, permanent=False): + if not self._lease and not permanent: + self.refresh_lease() + + ret = self._do_attempt_to_acquire_leader(permanent) + if not ret: + logger.info('Could not take out TTL lock') + return ret + + @catch_etcd_errors + def set_failover_value(self, value, index=None): + return self._client.put(self.failover_path, value, mod_revision=index) + + @catch_etcd_errors + def set_config_value(self, value, index=None): + return self._client.put(self.config_path, value, mod_revision=index) + + @catch_etcd_errors + def _write_leader_optime(self, last_operation): + return self._client.put(self.leader_optime_path, last_operation) + + @catch_etcd_errors + def _update_leader(self): + if not self._lease: + self.refresh_lease() + elif self.retry(self._client.lease_keepalive, self._lease): + self._last_lease_refresh = time.time() + + if self._lease: + cluster = self.cluster + leader_lease = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session + if leader_lease != self._lease: + self.take_leader() + return bool(self._lease) + + @catch_etcd_errors + def initialize(self, create_new=True, sysid=""): + return self.retry(self._client.put, self.initialize_path, sysid, None, 0 if create_new else None) + + @catch_etcd_errors + def _delete_leader(self): + 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) + + @catch_etcd_errors + def cancel_initialization(self): + return self.retry(self._client.deleterange, self.initialize_path) + + @catch_etcd_errors + def delete_cluster(self): + return self.retry(self._client.deleteprefix, self.cluster_prefix) + + @catch_etcd_errors + def set_history_value(self, value): + return self._client.put(self.history_path, value) + + @catch_etcd_errors + def set_sync_state_value(self, value, index=None): + return self.retry(self._client.put, self.sync_path, value, mod_revision=index) + + @catch_etcd_errors + def delete_sync_state(self, index=None): + return self.retry(self._client.deleterange, self.sync_path, mod_revision=index) + + def watch(self, leader_index, timeout): + if self.__do_not_watch: + self.__do_not_watch = False + return True + + try: + return super(Etcd3, self).watch(None, timeout) + finally: + self.event.clear() diff --git a/setup.py b/setup.py index 9a0858a0..3f7609cd 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, dmitrii.dolgov@zalando.de, alexk KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\ ' zookeeper exhibitor consul streaming replication kubernetes k8s' -EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'], +EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': [], 'raft': ['pysyncobj']} COVERAGE_XML = True COVERAGE_HTML = False diff --git a/tests/__init__.py b/tests/__init__.py index 4958f152..9a2d73e2 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -169,6 +169,7 @@ class PostgresInit(unittest.TestCase): 'stats_temp_directory': '/tmp'} @patch('psycopg2.connect', psycopg2_connect) + @patch('patroni.postgresql.CallbackExecutor', Mock()) @patch.object(ConfigHandler, 'write_postgresql_conf', Mock()) @patch.object(ConfigHandler, 'replace_pg_hba', Mock()) @patch.object(ConfigHandler, 'replace_pg_ident', Mock()) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 02883af3..cf45e283 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -8,7 +8,7 @@ from mock import patch, Mock from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, find_executable, CONFIG_FILE_PATH -from patroni.dcs.etcd import Client, Failover +from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Failover from patroni.utils import tzutc from psycopg2 import OperationalError from urllib3 import PoolManager @@ -37,7 +37,7 @@ class TestCtl(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) def setUp(self): - with patch.object(Client, 'machines') as mock_machines: + with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) self.runner = CliRunner() self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo') diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 258ab9a7..66d604ce 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -5,7 +5,7 @@ import unittest from dns.exception import DNSException from mock import Mock, patch -from patroni.dcs.etcd import AbstractDCS, Client, Cluster, Etcd, EtcdError, DnsCachingResolver +from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver from patroni.exceptions import DCSError from patroni.utils import Retry from urllib3.exceptions import ReadTimeoutError @@ -123,9 +123,9 @@ class TestClient(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) @patch('patroni.dcs.etcd.requests_get', requests_get) def setUp(self): - with patch.object(Client, 'machines') as mock_machines: + with patch.object(EtcdClient, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001']) - self.client = Client({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver()) + self.client = EtcdClient({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver()) self.client.http.request = http_request self.client.http.request_encode_body = http_request @@ -143,10 +143,9 @@ class TestClient(unittest.TestCase): except Exception: self.assertIsNone(machines) - @patch.object(Client, 'machines') + @patch.object(EtcdClient, 'machines') def test_api_execute(self, mock_machines): mock_machines.__get__ = Mock(return_value=['http://localhost:4001', 'http://localhost:2379']) - self.assertRaises(ValueError, self.client.api_execute, '', '') self.client._base_uri = 'http://localhost:4001' self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'POST', timeout=0) self.client._base_uri = 'http://localhost:4001' @@ -157,18 +156,17 @@ class TestClient(unittest.TestCase): self.client._machines_cache = [self.client._base_uri] self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'}) self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'}) - self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', '') - with patch.object(Client, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ - patch.object(Client, '_load_machines_cache', Mock(side_effect=Exception)): + with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ + patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)): self.client.http.request = Mock(side_effect=socket.error) self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry}) - with patch.object(Client, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ - patch.object(Client, '_load_machines_cache', Mock(return_value=True)): + with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ + patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)): self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry}) - with patch.object(Client, '_do_http_request', Mock(side_effect=etcd.EtcdException)): + with patch.object(EtcdClient, '_do_http_request', Mock(side_effect=etcd.EtcdException)): self.client._read_timeout = 0.01 self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET') @@ -184,7 +182,7 @@ class TestClient(unittest.TestCase): def test__get_machines_cache_from_dns(self): self.client._get_machines_cache_from_dns('error', 2379) - @patch.object(Client, 'machines') + @patch.object(EtcdClient, 'machines') def test__refresh_machines_cache(self, mock_machines): mock_machines.__get__ = Mock(side_effect=etcd.EtcdConnectionFailed) self.assertIsNone(self.client._refresh_machines_cache()) @@ -205,6 +203,10 @@ class TestClient(unittest.TestCase): timeout=1, source_address=('localhost', 53333), socket_options=[(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)]) + def test___del__(self): + self.client.http.clear = Mock(side_effect=TypeError) + del self.client + @patch('patroni.dcs.etcd.requests_get', requests_get) @patch('socket.getaddrinfo', socket_getaddrinfo) @@ -215,7 +217,7 @@ class TestEtcd(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) def setUp(self): - with patch.object(Client, 'machines') as mock_machines: + with patch.object(EtcdClient, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001']) self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}) @@ -226,17 +228,18 @@ class TestEtcd(unittest.TestCase): @patch('dns.resolver.query', dns_query) def test_get_etcd_client(self): with patch('time.sleep', Mock(side_effect=SleepException)),\ - patch.object(Client, 'machines') as mock_machines: + patch.object(EtcdClient, 'machines') as mock_machines: mock_machines.__get__ = Mock(side_effect=etcd.EtcdException) self.assertRaises(SleepException, self.etcd.get_etcd_client, - {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1}) + {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1}, + EtcdClient) self.assertRaises(SleepException, self.etcd.get_etcd_client, - {'url': 'https://test:2379', 'retry_timeout': 10}) + {'url': 'https://test:2379', 'retry_timeout': 10}, EtcdClient) self.assertRaises(SleepException, self.etcd.get_etcd_client, - {'hosts': 'foo:4001,bar', 'retry_timeout': 10}) + {'hosts': 'foo:4001,bar', 'retry_timeout': 10}, EtcdClient) mock_machines.__get__ = Mock(return_value=[]) self.assertRaises(SleepException, self.etcd.get_etcd_client, - {'proxy': 'https://user:password@test:2379', 'retry_timeout': 10}) + {'proxy': 'https://user:password@test:2379', 'retry_timeout': 10}, EtcdClient) def test_get_cluster(self): cluster = self.etcd.get_cluster() diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py new file mode 100644 index 00000000..6afef8f1 --- /dev/null +++ b/tests/test_etcd3.py @@ -0,0 +1,246 @@ +import etcd +import json +import unittest +import urllib3 + +from mock import Mock, patch +from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Error, Etcd3ClientError, RetryFailedError,\ + InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, base64_encode +from threading import Thread + +from . import SleepException, MockResponse + + +def mock_urlopen(self, method, url, **kwargs): + ret = MockResponse() + if method == 'GET' and url.endswith('/version'): + ret.content = '{"etcdserver": "3.3.13", "etcdcluster": "3.3.0"}' + elif method != 'POST': + raise Exception('Unexpected request method: {0} {1} {2}'.format(method, url, kwargs)) + elif url.endswith('/cluster/member/list'): + ret.content = '{"members":[{"clientURLs":["http://localhost:2379", "http://localhost:4001"]}]}' + elif url.endswith('/auth/authenticate'): + ret.content = '{"token":"authtoken"}' + elif url.endswith('/lease/grant'): + ret.content = '{"ID": "123"}' + elif url.endswith('/lease/keepalive'): + ret.content = '{"result":{"TTL":30}}' + elif url.endswith('/kv/range'): + ret.content = json.dumps({ + "header": {"revision": "1"}, + "kvs": [ + {"key": base64_encode('/patroni/test/leader'), + "value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'}, + {"key": base64_encode('/patroni/test/members/foo'), + "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'}, + {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'} + ] + }) + elif url.endswith('/watch'): + key = base64_encode('/patroni/test/config') + ret.read_chunked = Mock(return_value=[json.dumps({ + 'result': {'events': [ + {'kv': {'key': key, 'value': base64_encode('bar'), 'mod_revision': '2'}}, + {'kv': {'key': key, 'value': base64_encode('buzz'), 'mod_revision': '3'}}, + {'type': 'DELETE', 'kv': {'key': key, 'mod_revision': '4'}}, + {'kv': {'key': base64_encode('/patroni/test/optime/leader'), + 'value': base64_encode('1234567'), 'mod_revision': '5'}}, + ]} + })[:-1].encode('utf-8'), b'}{"error":{"grpc_code":14,"message":"","http_code":503}}']) + elif url.endswith('/kv/put') or url.endswith('/kv/txn'): + ret.status_code = 400 + ret.content = '{"code":5,"error":"etcdserver: requested lease not found"}' + elif not url.endswith('/kv/deleterange'): + raise Exception('Unexpected url: {0} {1} {2}'.format(method, url, kwargs)) + return ret + + +class BaseTestEtcd3(unittest.TestCase): + + @patch.object(Thread, 'start', Mock()) + @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) + def setUp(self): + self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, + 'host': 'localhost:2378', 'scope': 'test', 'name': 'foo', + 'username': 'etcduser', 'password': 'etcdpassword'}) + self.client = self.etcd3._client + self.kv_cache = self.client._kv_cache + + +class TestKVCache(BaseTestEtcd3): + + def test__do_watch(self): + self.client.watchprefix = Mock(return_value=False) + self.assertRaises(AttributeError, self.kv_cache._do_watch, '1') + + @patch('time.sleep', Mock(side_effect=SleepException)) + @patch('patroni.dcs.etcd3.KVCache._build_cache', Mock(side_effect=Exception)) + def test_run(self): + self.assertRaises(SleepException, self.kv_cache.run) + + @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) + def test_kill_stream(self): + self.assertRaises(Unavailable, self.kv_cache._do_watch, '1') + self.kv_cache.kill_stream() + with patch.object(MockResponse, 'connection', create=True) as mock_conn: + self.kv_cache.kill_stream() + mock_conn.sock.close.side_effect = Exception + self.kv_cache.kill_stream() + + +class TestPatroniEtcd3Client(BaseTestEtcd3): + + @patch('patroni.dcs.etcd3.Etcd3Client.authenticate', Mock(side_effect=Exception)) + def test__init__(self): + self.assertRaises(SystemExit, self.setUp) + + @patch.object(urllib3.PoolManager, 'urlopen') + def test_call_rpc(self, mock_urlopen): + request = {'key': base64_encode('/patroni/test/leader')} + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}' + self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]}) + self.client.call_rpc('/kv/put', request) + self.client.call_rpc('/kv/deleterange', request) + + @patch('time.time', Mock(side_effect=[1, 10.9, 100])) + def test__wait_cache(self): + with self.kv_cache.condition: + self.assertRaises(RetryFailedError, self.client._wait_cache, 10) + + @patch.object(urllib3.PoolManager, 'urlopen') + def test__restart_watcher(self, mock_urlopen): + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.status_code = 400 + mock_urlopen.return_value.content = '{"code":9,"error":"etcdserver: authentication is not enabled"}' + self.client.authenticate() + + @patch.object(urllib3.PoolManager, 'urlopen') + def test__handle_auth_errors(self, mock_urlopen): + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: user name is empty"}' + mock_urlopen.return_value.status_code = 403 + self.client._cluster_version = (3, 1, 5) + self.assertRaises(UnsupportedEtcdVersion, self.client.deleteprefix, 'foo') + self.client._cluster_version = (3, 3, 13) + self.assertRaises(UserEmpty, self.client.deleteprefix, 'foo') + mock_urlopen.return_value.content = '{"code":16,"error":"etcdserver: invalid auth token"}' + self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') + with patch.object(PatroniEtcd3Client, 'authenticate', Mock(return_value=True)): + self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') + self.client.username = None + self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') + + def test__handle_server_response(self): + response = MockResponse() + response.content = '{"code":0,"error":"' + self.assertRaises(etcd.EtcdException, self.client._handle_server_response, response) + response.status_code = 400 + self.assertRaises(Unknown, self.client._handle_server_response, response) + response.content = '{"error":{"grpc_code":0,"message":"","http_code":400}}' + try: + self.client._handle_server_response(response) + except Unknown as e: + self.assertEqual(e.as_dict(), {'code': 2, 'codeText': 'OK', 'error': u'', 'status': 400}) + + @patch.object(urllib3.PoolManager, 'urlopen') + def test__ensure_version_prefix(self, mock_urlopen): + self.client.version_prefix = None + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = '{"etcdserver": "3.0.3", "etcdcluster": "3.0.0"}' + self.assertRaises(UnsupportedEtcdVersion, self.client._ensure_version_prefix, '') + mock_urlopen.return_value.content = '{"etcdserver": "3.0.4", "etcdcluster": "3.0.0"}' + self.client._ensure_version_prefix('') + self.assertEqual(self.client.version_prefix, '/v3alpha') + mock_urlopen.return_value.content = '{"etcdserver": "3.4.4", "etcdcluster": "3.4.0"}' + self.client._ensure_version_prefix('') + self.assertEqual(self.client.version_prefix, '/v3') + + +@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) +class TestEtcd3(BaseTestEtcd3): + + @patch.object(Thread, 'start', Mock()) + @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) + def setUp(self): + super(TestEtcd3, self).setUp() + self.assertRaises(AttributeError, self.kv_cache._build_cache) + self.kv_cache._is_ready = True + self.etcd3.get_cluster() + + def test_get_cluster(self): + self.assertIsInstance(self.etcd3.get_cluster(), Cluster) + self.client._kv_cache = None + with patch.object(urllib3.PoolManager, 'urlopen') as mock_urlopen: + mock_urlopen.side_effect = UnsupportedEtcdVersion('') + self.assertRaises(UnsupportedEtcdVersion, self.etcd3.get_cluster) + mock_urlopen.side_effect = SleepException() + self.assertRaises(Etcd3Error, self.etcd3.get_cluster) + + def test_touch_member(self): + self.etcd3.touch_member({}) + self.etcd3._lease = 'bla' + self.etcd3.touch_member({}) + with patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError)): + self.etcd3.touch_member({}) + + def test__update_leader(self): + self.etcd3._lease = None + self.etcd3.update_leader('123') + self.etcd3.update_leader('124') + + def test_attempt_to_acquire_leader(self): + self.etcd3._lease = None + self.assertFalse(self.etcd3.attempt_to_acquire_leader()) + + def test_set_ttl(self): + self.etcd3.set_ttl(20) + + @patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False)) + def test_refresh_lease(self): + self.etcd3._last_lease_refresh = 0 + self.etcd3.refresh_lease() + + @patch('time.sleep', Mock(side_effect=SleepException)) + @patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False)) + @patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError)) + def test_create_lease(self): + self.etcd3._lease = None + self.etcd3._last_lease_refresh = 0 + self.assertRaises(SleepException, self.etcd3.create_lease) + + def test_set_failover_value(self): + self.etcd3.set_failover_value('', 1) + + def test_set_config_value(self): + self.etcd3.set_config_value('') + + def test_initialize(self): + self.etcd3.initialize() + + def test_cancel_initialization(self): + self.etcd3.cancel_initialization() + + def test_delete_leader(self): + self.etcd3.delete_leader() + + def test_delete_cluster(self): + self.etcd3.delete_cluster() + + def test_set_history_value(self): + self.etcd3.set_history_value('') + + def test_set_sync_state_value(self): + self.etcd3.set_sync_state_value('') + + def test_delete_sync_state(self): + self.etcd3.delete_sync_state() + + def test_watch(self): + self.etcd3.set_ttl(10) + self.etcd3.watch(None, 0) + self.etcd3.watch(None, 0) + + def test_set_socket_options(self): + with patch('socket.SIO_KEEPALIVE_VALS', 1, create=True): + self.etcd3.set_socket_options(Mock(), None) diff --git a/tests/test_ha.py b/tests/test_ha.py index dbd33f7f..45d017b9 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -6,7 +6,7 @@ import sys from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory -from patroni.dcs.etcd import Client +from patroni.dcs.etcd import AbstractEtcdClientWithFailover from patroni.exceptions import DCSError, PostgresConnectionException, PatroniFatalException from patroni.ha import Ha, _MemberStatus from patroni.postgresql import Postgresql @@ -185,7 +185,7 @@ class TestHa(PostgresInit): @patch.object(etcd.Client, 'read', etcd_read) def setUp(self): super(TestHa, self).setUp() - with patch.object(Client, 'machines') as mock_machines: + with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines: mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) self.p.set_state('running') self.p.set_role('replica') diff --git a/tests/test_patroni.py b/tests/test_patroni.py index db1f69fa..225cf32a 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -9,7 +9,7 @@ import patroni.config as config from mock import Mock, PropertyMock, patch from patroni.api import RestApiServer from patroni.async_executor import AsyncExecutor -from patroni.dcs.etcd import Client +from patroni.dcs.etcd import AbstractEtcdClientWithFailover from patroni.exceptions import DCSError from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler @@ -53,7 +53,7 @@ class TestPatroni(unittest.TestCase): @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock()) @patch.object(etcd.Client, 'read', etcd_read) @patch.object(Thread, 'start', Mock()) - @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) + @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) def setUp(self): self._handlers = logging.getLogger().handlers[:] RestApiServer._BaseServer__is_shut_down = Mock() @@ -75,7 +75,7 @@ class TestPatroni(unittest.TestCase): @patch('sys.argv', ['patroni.py', 'postgres0.yml']) @patch('time.sleep', Mock(side_effect=SleepException)) @patch.object(etcd.Client, 'delete', Mock()) - @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) + @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379'])) @patch.object(Thread, 'join', Mock()) def test_patroni_patroni_main(self): with patch('subprocess.call', Mock(return_value=1)): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index adb92a32..028399bf 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -98,7 +98,6 @@ class TestPostgresql(BaseTestPostgresql): def setUp(self): super(TestPostgresql, self).setUp() self.p.config.write_postgresql_conf() - self.p._callback_executor = Mock() @patch('subprocess.Popen') @patch.object(Postgresql, 'wait_for_startup')