From 4328c150103e6b3119fd57f0898be8c69bcb8928 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 8 Dec 2017 16:55:00 +0100 Subject: [PATCH] Make Patroni Kubernetes native (#500) * Use ConfigMaps or Endpoins for leader elections and to keep cluster state * Label pods with a postgres role * change behavior of pip install. From now on it will not install all dependencies, you have to specify explicitly DCS you want to use Patroni with: `pip install patroni[etcd,zookeeper,kubernetes]` --- .travis.yml | 28 +- features/basic_replication.feature | 11 +- features/environment.py | 93 ++++++- features/steps/patroni_api.py | 7 +- kubernetes/Dockerfile | 32 +++ kubernetes/callback.py | 62 +++++ kubernetes/entrypoint.sh | 36 +++ kubernetes/patroni_k8s.yaml | 122 +++++++++ patroni/__init__.py | 6 +- patroni/config.py | 17 +- patroni/ctl.py | 2 +- patroni/dcs/__init__.py | 68 +++-- patroni/dcs/consul.py | 15 +- patroni/dcs/etcd.py | 6 +- patroni/dcs/kubernetes.py | 396 +++++++++++++++++++++++++++++ patroni/dcs/zookeeper.py | 15 +- patroni/ha.py | 15 +- requirements.txt | 7 +- setup.py | 22 +- tests/test_config.py | 3 + tests/test_consul.py | 10 +- tests/test_etcd.py | 2 +- tests/test_kubernetes.py | 103 ++++++++ tests/test_zookeeper.py | 22 +- 24 files changed, 1000 insertions(+), 100 deletions(-) create mode 100644 kubernetes/Dockerfile create mode 100755 kubernetes/callback.py create mode 100755 kubernetes/entrypoint.sh create mode 100644 kubernetes/patroni_k8s.yaml create mode 100644 patroni/dcs/kubernetes.py create mode 100644 tests/test_kubernetes.py diff --git a/.travis.yml b/.travis.yml index ca96dbca..42c97f31 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,4 +1,4 @@ -sudo: false +sudo: true dist: trusty language: python python: @@ -7,11 +7,13 @@ env: global: - ETCDVERSION=3.0.17 ZKVERSION=3.4.9 CONSULVERSION=0.7.4 - PYVERSIONS="2.7 3.4 3.5" + - BOTO_CONFIG=/doesnotexist matrix: - TEST_SUITE="python setup.py" - DCS="etcd" TEST_SUITE="behave" - DCS="exhibitor" TEST_SUITE="behave" - DCS="consul" TEST_SUITE="behave" + - DCS="kubernetes" TEST_SUITE="behave" cache: directories: - $HOME/mycache @@ -51,6 +53,26 @@ install: ln -s $EC etcd } + function get_kubernetes() { + wget -O localkube "https://storage.googleapis.com/minikube/k8sReleases/v1.7.0/localkube-linux-amd64" + chmod +x localkube + sudo nohup ./localkube --logtostderr=true --enable-dns=false > localkube.log 2>&1 & + + echo "Waiting for localkube to start..." + if ! timeout 120 sh -c "while ! curl -ks http://127.0.0.1:8080/ >/dev/null; do sleep 1; done"; then + sudo cat localkube.log + echo "localkube did not start" + exit 1 + fi + echo "Check certificate permissions" + sudo chmod 644 /var/lib/localkube/certs/* + sudo ls -altr /var/lib/localkube/certs/ + + echo "Set up .kube/config" + mkdir ~/.kube + echo -e "apiVersion: v1\nclusters:\n- cluster:\n certificate-authority: /var/lib/localkube/certs/ca.crt\n server: https://127.0.0.1:8443\n name: local\ncontexts:\n- context:\n cluster: local\n user: myself\n name: local\ncurrent-context: local\nkind: Config\npreferences: {}\nusers:\n- name: myself\n user:\n client-certificate: /var/lib/localkube/certs/apiserver.crt\n client-key: /var/lib/localkube/certs/apiserver.key\n" > ~/.kube/config + } + function get_exhibitor() { ZC=~/mycache/zookeeper-${ZKVERSION} if [[ ! -d $ZC ]]; then @@ -65,7 +87,6 @@ install: echo -e 'HTTP/1.0 200 OK\nContent-Type: application/json\n\n{"servers":["127.0.0.1"],"port":2181}' \ | nc -l 8181 &> /dev/null done& - ZK_PID=$! } attempt_num=1 @@ -115,4 +136,5 @@ after_success: - fpv=$(basename $(readlink $HOME/virtualenv/python3.5)) && mv $HOME/mycache/${fpv} $HOME/virtualenv/${fpv} - coveralls - if [[ $TEST_SUITE != "behave" ]]; then python-codacy-coverage -r coverage.xml; fi - - if [[ $DCS == "exhibitor" ]]; then ~/mycache/zookeeper-${ZKVERSION}/bin/zkServer.sh stop; kill -9 $ZK_PID; fi + - if [[ $DCS == "exhibitor" ]]; then ~/mycache/zookeeper-${ZKVERSION}/bin/zkServer.sh stop; fi + - sudo kill $(jobs -p) diff --git a/features/basic_replication.feature b/features/basic_replication.feature index 4d30f272..67b44004 100644 --- a/features/basic_replication.feature +++ b/features/basic_replication.feature @@ -21,9 +21,14 @@ Feature: basic replication And "sync" key in DCS has sync_standby=postgres2 after 10 seconds Scenario: check the basic failover in synchronous mode - When I kill postgres0 - Then postgres2 role is the primary after 22 seconds - When I issue a PATCH request to http://127.0.0.1:8009/config with {"synchronous_mode": null, "master_start_timeout": 0} + Given I run patronictl.py pause batman + Then I receive a response returncode 0 + When I sleep for 2 seconds + And I shut down postgres0 + And I run patronictl.py resume batman + Then I receive a response returncode 0 + And postgres2 role is the primary after 24 seconds + When I issue a PATCH request to http://127.0.0.1:8010/config with {"synchronous_mode": null, "master_start_timeout": 0} Then I receive a response code 200 When I add the table bar to postgres2 Then table bar is present on postgres1 after 20 seconds diff --git a/features/environment.py b/features/environment.py index d82a95ff..8a729995 100644 --- a/features/environment.py +++ b/features/environment.py @@ -7,6 +7,7 @@ import kazoo.exceptions import os import psutil import psycopg2 +import json import shutil import signal import six @@ -98,6 +99,7 @@ class PatroniController(AbstractController): else: self.watchdog = None + self._scope = (custom_config or {}).get('scope', 'batman') self._config = self._make_patroni_test_config(name, custom_config) self._closables = [] @@ -125,6 +127,9 @@ class PatroniController(AbstractController): def _start(self): if self.watchdog: self.watchdog.start() + if isinstance(self._context.dcs_ctl, KubernetesController): + self._context.dcs_ctl.create_pod(self._name[8:], self._scope) + os.environ['PATRONI_KUBERNETES_POD_IP'] = '10.0.0.' + self._name[-1] return subprocess.Popen(['coverage', 'run', '--source=patroni', '-p', 'patroni.py', self._config], stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory) @@ -132,6 +137,8 @@ class PatroniController(AbstractController): if postgres: return subprocess.call(['pg_ctl', '-D', self._data_dir, 'stop', '-mi', '-w']) super(PatroniController, self).stop(kill, timeout) + if isinstance(self._context.dcs_ctl, KubernetesController): + self._context.dcs_ctl.delete_pod(self._name[8:]) if self.watchdog: self.watchdog.stop() @@ -147,7 +154,7 @@ class PatroniController(AbstractController): with open(patroni_config_name) as f: config = yaml.safe_load(f) - config.pop('etcd') + config.pop('etcd', None) host = config['postgresql']['listen'].split(':')[0] @@ -335,10 +342,6 @@ class AbstractDcsController(AbstractController): def query(self, key, scope='batman'): """ query for a value of a given key """ - @abc.abstractmethod - def set(self, key, value): - """ set a value to a given key """ - @abc.abstractmethod def cleanup_service_tree(self): """ clean all contents stored in the tree used for the tests """ @@ -388,9 +391,6 @@ class ConsulController(AbstractDcsController): _, value = self._client.kv.get(self.path(key, scope)) return value and value['Value'].decode('utf-8') - def set(self, key, value): - self._client.kv.put(self.path(key), value) - def cleanup_service_tree(self): self._client.kv.delete(self.path(scope=''), recurse=True) @@ -417,9 +417,6 @@ class EtcdController(AbstractDcsController): except etcd.EtcdKeyNotFound: return None - def set(self, key, value): - self._client.set(self.path(key), value) - def cleanup_service_tree(self): try: self._client.delete(self.path(scope=''), recursive=True) @@ -436,6 +433,76 @@ class EtcdController(AbstractDcsController): return False +class KubernetesController(AbstractDcsController): + + def __init__(self, context): + super(KubernetesController, self).__init__(context) + self._namespace = 'default' + self._labels = {"application": "patroni"} + self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) + os.environ['PATRONI_KUBERNETES_LABELS'] = json.dumps(self._labels) + os.environ['PATRONI_KUBERNETES_USE_ENDPOINTS'] = 'true' + + from kubernetes import client as k8s_client, config as k8s_config + k8s_config.load_kube_config(context='local') + self._client = k8s_client + self._api = self._client.CoreV1Api() + + def _start(self): + pass + + def create_pod(self, name, scope): + labels = self._labels.copy() + labels['cluster-name'] = scope + metadata = self._client.V1ObjectMeta(namespace=self._namespace, name=name, labels=labels) + spec = self._client.V1PodSpec(containers=[self._client.V1Container(name=name, image='empty')]) + body = self._client.V1Pod(metadata=metadata, spec=spec) + self._api.create_namespaced_pod(self._namespace, body) + + def delete_pod(self, name): + try: + self._api.delete_namespaced_pod(name, self._namespace, self._client.V1DeleteOptions()) + except: + pass + while True: + try: + self._api.read_namespaced_pod(name, self._namespace) + except: + break + + def query(self, key, scope='batman'): + if key.startswith('members/'): + pod = self._api.read_namespaced_pod(key[8:], self._namespace) + return (pod.metadata.annotations or {}).get('status', '') + else: + try: + e = self._api.read_namespaced_endpoints(scope + ('' if key == 'leader' else '-' + key), self._namespace) + if key == 'leader': + return e.metadata.annotations[key] + else: + return json.dumps(e.metadata.annotations) + except: + return None + + def cleanup_service_tree(self): + try: + self._api.delete_collection_namespaced_pod(self._namespace, label_selector=self._label_selector) + except: + pass + try: + self._api.delete_collection_namespaced_endpoints(self._namespace, label_selector=self._label_selector) + except: + pass + + while True: + result = self._api.list_namespaced_pod(self._namespace, label_selector=self._label_selector) + if len(result.items) < 1: + break + + def _is_running(self): + return True + + class ZooKeeperController(AbstractDcsController): """ handles all zookeeper related tasks, used for the tests setup and cleanup """ @@ -455,9 +522,6 @@ class ZooKeeperController(AbstractDcsController): except kazoo.exceptions.NoNodeError: return None - def set(self, key, value): - self._client.set(self.path(key), value.encode('utf-8')) - def cleanup_service_tree(self): try: self._client.delete(self.path(scope=''), recursive=True) @@ -713,6 +777,7 @@ class WatchdogMonitor(object): # actions to execute on start/stop of the tests and before running invidual features def before_all(context): + os.environ.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'}) context.ci = 'TRAVIS_BUILD_NUMBER' in os.environ or 'BUILD_NUMBER' in os.environ context.timeout_multiplier = 2 if context.ci else 1 context.pctl = PatroniPoolController(context) diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index e6734f36..ba0ee086 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -1,3 +1,4 @@ +import base64 import json import os import parse @@ -78,11 +79,13 @@ def do_post_empty(context, url): @step('I issue a {request_method:w} request to {url:url} with {data}') def do_request(context, request_method, url, data): data = data and json.loads(data) or {} + headers = {'Authorization': 'Basic ' + base64.b64encode('username:password'.encode('utf-8')).decode('utf-8'), + 'Content-Type': 'application/json'} try: if request_method == 'PATCH': - r = requests.patch(url, json=data) + r = requests.patch(url, headers=headers, json=data) else: - r = requests.post(url, json=data) + r = requests.post(url, headers=headers, json=data) except requests.exceptions.RequestException: context.status_code = None context.response = None diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile new file mode 100644 index 00000000..ce9b2b21 --- /dev/null +++ b/kubernetes/Dockerfile @@ -0,0 +1,32 @@ +FROM postgres:9.6 +MAINTAINER Alexander Kukushkin + +RUN export DEBIAN_FRONTEND=noninteractive \ + && echo 'APT::Install-Recommends "0";\nAPT::Install-Suggests "0";' > /etc/apt/apt.conf.d/01norecommend \ + && apt-get update -y \ + && apt-get upgrade -y \ + && apt-get install -y git curl jq python-psycopg2 python-yaml python-requests python-six python-pysocks \ + python-dateutil python-pip python-prettytable python-wheel python-psutil python locales \ + + ## 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 \ + + && pip install setuptools pip --upgrade \ + && pip install 'git+https://github.com/zalando/patroni.git@feature/k8s#egg=patroni[kubernetes]' \ + + && mkdir -p /home/postgres \ + && chown postgres:postgres /home/postgres \ + + # Clean up + && apt-get remove -y git python-pip python-setuptools \ + && apt-get autoremove -y \ + && apt-get clean -y \ + && rm -rf /var/lib/apt/lists/* /root/.cache + +ADD entrypoint.sh callback.py / + +EXPOSE 5432 8008 +ENV LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8 +USER postgres +WORKDIR /home/postgres +CMD ["/bin/bash", "/entrypoint.sh"] diff --git a/kubernetes/callback.py b/kubernetes/callback.py new file mode 100755 index 00000000..c3293149 --- /dev/null +++ b/kubernetes/callback.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python + +import logging +import os +import socket +import sys +import time + +from kubernetes import client as k8s_client, config as k8s_config +from urllib3.exceptions import HTTPError +from six.moves.http_client import HTTPException + +logger = logging.getLogger(__name__) + + +class CoreV1Api(k8s_client.CoreV1Api): + + def retry(func): + def wrapped(*args, **kwargs): + count = 0 + while True: + try: + return func(*args, **kwargs) + except (HTTPException, HTTPError, socket.error, socket.timeout): + if count >= 10: + raise + logger.info('Throttling API requests...') + time.sleep(2 ** count * 0.5) + count += 1 + return wrapped + + @retry + def patch_namespaced_endpoints(self, *args, **kwargs): + return super(CoreV1Api, self).patch_namespaced_endpoints(*args, **kwargs) + + +def patch_master_endpoint(api, namespace, cluster): + addresses = [k8s_client.V1EndpointAddress(ip=os.environ['POD_IP'])] + ports = [k8s_client.V1EndpointPort(port=5432)] + subsets = [k8s_client.V1EndpointSubset(addresses=addresses, ports=ports)] + body = k8s_client.V1Endpoints(subsets=subsets) + return api.patch_namespaced_endpoints(cluster, namespace, body) + + +def main(): + logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) + if len(sys.argv) != 4 or sys.argv[1] not in ('on_start', 'on_stop', 'on_role_change'): + sys.exit('Usage: %s ', sys.argv[0]) + + action, role, cluster = sys.argv[1:4] + + k8s_config.load_incluster_config() + k8s_api = CoreV1Api() + + namespace = os.environ['KUBERNETES_NAMESPACE'] + + if role == 'master' and action in ('on_start', 'on_role_change'): + patch_master_endpoint(k8s_api, namespace, cluster) + + +if __name__ == '__main__': + main() diff --git a/kubernetes/entrypoint.sh b/kubernetes/entrypoint.sh new file mode 100755 index 00000000..c665726c --- /dev/null +++ b/kubernetes/entrypoint.sh @@ -0,0 +1,36 @@ +#!/bin/bash + +cat > /home/postgres/patroni.yml <<__EOF__ +bootstrap: + dcs: + postgresql: + use_pg_rewind: true + initdb: + - auth-host: md5 + - auth-local: trust + - encoding: UTF8 + - locale: en_US.UTF-8 + - data-checksums + pg_hba: + - host all all 0.0.0.0/0 md5 + - host replication ${PATRONI_REPLICATION_USERNAME} ${POD_IP}/16 md5 +restapi: + connect_address: '${POD_IP}:8008' +postgresql: + connect_address: '${POD_IP}:5432' + authentication: + superuser: + password: '${PATRONI_SUPERUSER_PASSWORD}' + replication: + password: '${PATRONI_REPLICATION_PASSWORD}' + callbacks: + on_start: /callback.py + on_stop: /callback.py + on_role_change: /callback.py +__EOF__ + +unset PATRONI_SUPERUSER_PASSWORD PATRONI_REPLICATION_PASSWORD +export KUBERNETES_NAMESPACE=$PATRONI_KUBERNETES_NAMESPACE +export POD_NAME=$PATRONI_NAME + +exec /usr/bin/python /usr/local/bin/patroni /home/postgres/patroni.yml diff --git a/kubernetes/patroni_k8s.yaml b/kubernetes/patroni_k8s.yaml new file mode 100644 index 00000000..ed465352 --- /dev/null +++ b/kubernetes/patroni_k8s.yaml @@ -0,0 +1,122 @@ +apiVersion: apps/v1beta1 +kind: StatefulSet +metadata: + name: &cluster_name patronidemo + labels: + application: patroni + cluster-name: *cluster_name +spec: + replicas: 3 + serviceName: *cluster_name + template: + metadata: + labels: + application: patroni + cluster-name: *cluster_name + spec: + containers: + - name: *cluster_name + image: patroni # docker build -t patroni . + imagePullPolicy: IfNotPresent + ports: + - containerPort: 8008 + protocol: TCP + - containerPort: 5432 + protocol: TCP + volumeMounts: + - mountPath: /home/postgres/pgdata + name: pgdata + env: + - name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP + - name: PATRONI_KUBERNETES_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: PATRONI_KUBERNETES_LABELS + value: '{application: patroni, cluster-name: patronidemo}' + - name: PATRONI_SUPERUSER_USERNAME + value: postgres + - name: PATRONI_SUPERUSER_PASSWORD + valueFrom: + secretKeyRef: + name: *cluster_name + key: superuser-password + - name: PATRONI_REPLICATION_USERNAME + value: standby + - name: PATRONI_REPLICATION_PASSWORD + valueFrom: + secretKeyRef: + name: *cluster_name + key: replication-password + - name: PATRONI_SCOPE + value: *cluster_name + - name: PATRONI_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: PATRONI_POSTGRESQL_DATA_DIR + value: /home/postgres/pgdata/pgroot/data + - name: PATRONI_POSTGRESQL_PGPASS + value: /tmp/pgpass + - name: PATRONI_POSTGRESQL_LISTEN + value: '0.0.0.0:5432' + - name: PATRONI_RESTAPI_LISTEN + value: '0.0.0.0:8008' + terminationGracePeriodSeconds: 0 + volumes: + - name: pgdata + emptyDir: {} +# volumeClaimTemplates: +# - metadata: +# labels: +# application: spilo +# spilo-cluster: *cluster_name +# annotations: +# volume.alpha.kubernetes.io/storage-class: anything +# name: pgdata +# spec: +# accessModes: +# - ReadWriteOnce +# resources: +# requests: +# storage: 5Gi + +--- +apiVersion: v1 +kind: Endpoints +metadata: + name: &cluster_name patronidemo + labels: + application: patroni + cluster-name: *cluster_name +subsets: [] + +--- +apiVersion: v1 +kind: Service +metadata: + name: &cluster_name patronidemo + labels: + application: patroni + cluster-name: *cluster_name +spec: + type: ClusterIP + ports: + - port: 5432 + targetPort: 5432 + +--- +apiVersion: v1 +kind: Secret +metadata: + name: &cluster_name patronidemo + labels: + application: patroni + cluster-name: *cluster_name +type: Opaque +data: + superuser-password: emFsYW5kbw== + replication-password: cmVwLXBhc3M= diff --git a/patroni/__init__.py b/patroni/__init__.py index 0f207f75..6bc56396 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -39,7 +39,7 @@ class Patroni(object): while True: try: cluster = self.dcs.get_cluster() - if cluster and cluster.config: + if cluster and cluster.config and cluster.config.data: if self.config.set_dynamic_configuration(cluster.config): self.dcs.reload_config(self.config) self.watchdog.reload_config(self.config) @@ -113,8 +113,8 @@ class Patroni(object): logger.info(self.ha.run_cycle()) - cluster = self.dcs.cluster - if cluster and cluster.config and self.config.set_dynamic_configuration(cluster.config): + if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \ + and self.config.set_dynamic_configuration(self.dcs.cluster.config): self.reload_config() if self.postgresql.role != 'uninitialized': diff --git a/patroni/config.py b/patroni/config.py index 8a8ef3cc..5dabe8ec 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -239,16 +239,25 @@ class Config(object): for param in list(os.environ.keys()): if param.startswith(Config.PATRONI_ENV_PREFIX): - name, suffix = (param[8:].rsplit('_', 1) + [''])[:2] + name, suffix = (param[8:].split('_', 1) + [''])[:2] if name and suffix: # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..) - if suffix in ('HOST', 'HOSTS', 'PORT', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT', 'KEY', - 'VERIFY', 'TOKEN', 'CHECKS', 'DC') and '_' not in name: + if suffix in ('HOST', 'HOSTS', 'PORT', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT', + 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'NAMESPACE', 'CONTEXT', + 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS'): value = os.environ.pop(param) if suffix == 'PORT': value = value and parse_int(value) - elif suffix in ('HOSTS', 'CHECKS'): + elif suffix in ('HOSTS', 'PORTS', 'CHECKS'): value = value and _parse_list(value) + elif suffix == 'LABELS': + if not value.strip().startswith('{'): + value = '{{{0}}}'.format(value) + try: + value = yaml.safe_load(value) + except Exception: + logger.exception('Exception when parsing dict %s', value) + value = None if value: ret[name.lower()][suffix.lower()] = value # PATRONI__PASSWORD=, PATRONI__OPTIONS= diff --git a/patroni/ctl.py b/patroni/ctl.py index 2afec1b4..b5643bdb 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -406,7 +406,7 @@ def remove(obj, cluster_name, fmt): if message != confirm: raise PatroniCtlException('You did not exactly type "{0}"'.format(message)) - if cluster.leader: + if cluster.leader and cluster.leader.name: confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue') if confirm != cluster.leader.name: raise PatroniCtlException('You did not specify the current master of the cluster') diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index fb97eedc..717a0801 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -183,34 +183,38 @@ class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): """ >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}')) True + >>> 'Failover' in str(Failover.from_node(1, {"leader": "cluster_leader"})) + True >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}')) True >>> Failover.from_node(1, 'null') is None - True + False >>> n = '{"leader": "cluster_leader", "member": "cluster_candidate", "scheduled_at": "2016-01-14T10:09:57.1394Z"}' >>> 'tzinfo=' in str(Failover.from_node(1, n)) True >>> Failover.from_node(1, None) is None - True + False >>> Failover.from_node(1, '{}') is None - True + False >>> 'abc' in Failover.from_node(1, 'abc:def') True """ @staticmethod def from_node(index, value): - if not value: - return None - - try: - data = json.loads(value) - if not data: - return None - except ValueError: - 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) if leader or candidate else None + if isinstance(value, dict): + data = value + elif value: + try: + data = json.loads(value) + if not isinstance(data, dict): + data = {} + except ValueError: + 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) if leader or candidate else None + else: + data = {} if data.get('scheduled_at'): data['scheduled_at'] = dateutil.parser.parse(data['scheduled_at']) @@ -258,8 +262,12 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')): True >>> SyncState.from_node(1, '{"leader": "leader"}').leader == "leader" True + >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader" + True """ - if value: + if isinstance(value, dict): + data = value + elif value: try: data = json.loads(value) if not isinstance(data, dict): @@ -345,8 +353,7 @@ class AbstractDCS(object): i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc... """ self._name = config['name'] - self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/')) - self._base_path = '/'.join([self._namespace, config['scope']]) + self._base_path = os.path.join('/', config.get('namespace', '/service/').strip('/'), config['scope']) self._set_loop_wait(config.get('loop_wait', 10)) self._ctl = bool(config.get('patronictl', False)) @@ -449,7 +456,7 @@ class AbstractDCS(object): self._last_leader_operation = last_operation @abc.abstractmethod - def update_leader(self): + def _update_leader(self): """Update leader key (or session) ttl :returns: `!True` if leader key (or session) has been updated successfully. @@ -458,6 +465,18 @@ class AbstractDCS(object): You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd `prevValue` parameter must be used.""" + def update_leader(self, last_operation): + """Update leader key (or session) ttl and optime/leader + + :param last_operation: absolute xlog location in bytes + :returns: `!True` if leader key (or session) has been updated successfully. + If not, `!False` must be returned and current instance would be demoted.""" + + ret = self._update_leader() + if ret and last_operation: + self.write_leader_optime(last_operation) + return ret + @abc.abstractmethod def attempt_to_acquire_leader(self, permanent=False): """Attempt to acquire leader lock @@ -483,7 +502,6 @@ class AbstractDCS(object): if scheduled_at: failover_value['scheduled_at'] = scheduled_at.isoformat() - return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), index) @abc.abstractmethod @@ -496,7 +514,7 @@ class AbstractDCS(object): This method should create or update key with the name = '/members/' + `~self._name` and value = data in a given DCS. - :param data: json serialized information about instance (including connection strings) + :param data: information about instance (including connection strings) :param ttl: ttl for member key, optional parameter. If it is None `~self.member_ttl will be used` :param permanent: if set to `!True`, the member key will never expire. Used in patronictl for the external master. @@ -533,8 +551,14 @@ class AbstractDCS(object): def delete_cluster(self): """Delete cluster from DCS""" + @staticmethod + def sync_state(leader, sync_standby): + """Build sync_state dict""" + return {'leader': leader, 'sync_standby': sync_standby} + def write_sync_state(self, leader, sync_standby, index=None): - return self.set_sync_state_value(json.dumps({'leader': leader, 'sync_standby': sync_standby}), index=index) + sync_value = self.sync_state(leader, sync_standby) + return self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index) @abc.abstractmethod def set_sync_state_value(self, value, index=None): diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 51aa4dd8..d941a2c0 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -1,4 +1,5 @@ from __future__ import absolute_import +import json import logging import os import socket @@ -9,7 +10,7 @@ import urllib3 from consul import ConsulException, NotFound, base from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState from patroni.exceptions import DCSError -from patroni.utils import parse_bool, Retry, RetryFailedError +from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError from urllib3.exceptions import HTTPError from six.moves.urllib.parse import urlencode, urlparse from six.moves.http_client import HTTPException @@ -141,7 +142,7 @@ class Consul(AbstractDCS): retry_exceptions=(ConsulInternalError, HTTPException, HTTPError, socket.error, socket.timeout)) - self._my_member_data = None + self._my_member_data = {} kwargs = {} if 'url' in config: r = urlparse(config['url']) @@ -311,12 +312,12 @@ class Consul(AbstractDCS): except Exception: return False - if not create_member and member and data == self._my_member_data: + if not create_member and member and deep_compare(data, self._my_member_data): return True try: args = {} if permanent else {'acquire': self._session} - self._client.kv.put(self.member_path, data, **args) + self._client.kv.put(self.member_path, json.dumps(data, separators=(',', ':')), **args) self._my_member_data = data return True except Exception: @@ -352,7 +353,7 @@ class Consul(AbstractDCS): return self._client.kv.put(self.leader_optime_path, last_operation) @catch_consul_errors - def update_leader(self): + def _update_leader(self): if self._session: self.retry(self._client.session.renew, self._session) self._last_session_refresh = time.time() @@ -379,11 +380,11 @@ class Consul(AbstractDCS): @catch_consul_errors def set_sync_state_value(self, value, index=None): - return self._client.kv.put(self.sync_path, value, cas=index) + return self.retry(self._client.kv.put, self.sync_path, value, cas=index) @catch_consul_errors def delete_sync_state(self, index=None): - return self._client.kv.delete(self.sync_path, cas=index) + return self.retry(self._client.kv.delete, self.sync_path, cas=index) def watch(self, leader_index, timeout): if self.__do_not_watch: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 8a6ec569..bd06f977 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -1,5 +1,6 @@ from __future__ import absolute_import import etcd +import json import logging import os import urllib3.util.connection @@ -467,6 +468,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def touch_member(self, data, ttl=None, permanent=False): + data = json.dumps(data, separators=(',', ':')) return self.retry(self._client.set, self.member_path, data, None if permanent else ttl or self._ttl) @catch_etcd_errors @@ -499,7 +501,7 @@ class Etcd(AbstractDCS): return self._client.set(self.leader_optime_path, last_operation) @catch_etcd_errors - def update_leader(self): + def _update_leader(self): return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self._ttl) @catch_etcd_errors @@ -520,7 +522,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def set_sync_state_value(self, value, index=None): - return self._client.write(self.sync_path, value, prevIndex=index or 0) + return self.retry(self._client.write, self.sync_path, value, prevIndex=index or 0) @catch_etcd_errors def delete_sync_state(self, index=None): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py new file mode 100644 index 00000000..7d424848 --- /dev/null +++ b/patroni/dcs/kubernetes.py @@ -0,0 +1,396 @@ +from __future__ import absolute_import +import datetime +import functools +import json +import logging +import socket +import sys +import time + +from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch +from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState +from patroni.exceptions import DCSError +from patroni.utils import deep_compare, tzutc, Retry, RetryFailedError +from urllib3.exceptions import HTTPError +from six.moves.http_client import HTTPException + +logger = logging.getLogger(__name__) + + +class KubernetesError(DCSError): + pass + + +class KubernetesRetriableException(k8s_client.rest.ApiException): + + def __init__(self, orig): + super(KubernetesRetriableException, self).__init__(orig.status, orig.reason) + self.body = orig.body + self.headers = orig.headers + + +class CoreV1Api(object): + + def __init__(self, use_endpoints=False): + self._api = k8s_client.CoreV1Api() + self._request_timeout = None + self._use_endpoints = use_endpoints + + def set_timeout(self, timeout): + self._request_timeout = (1, timeout / 3.0) + + def __getattr__(self, func): + if func.endswith('_kind'): + func = func[:-4] + ('endpoints' if self._use_endpoints else 'config_map') + + def wrapper(*args, **kwargs): + if '_request_timeout' not in kwargs: + kwargs['_request_timeout'] = self._request_timeout + try: + return getattr(self._api, func)(*args, **kwargs) + except k8s_client.rest.ApiException as e: + if e.status in (502, 503, 504): # XXX + raise KubernetesRetriableException(e) + raise + return wrapper + + +class Kubernetes(AbstractDCS): + + def __init__(self, config): + self._labels = config['labels'] + self._labels[config.get('scope_label', 'cluster-name')] = config['scope'] + self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) + self._namespace = config.get('namespace') or 'default' + self._role_label = config.get('role_label', 'role') + config['namespace'] = '' + super(Kubernetes, self).__init__(config) + self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, + retry_exceptions=(KubernetesRetriableException, HTTPException, + HTTPError, socket.error, socket.timeout)) + self._ttl = None + try: + k8s_config.load_incluster_config() + except k8s_config.ConfigException: + k8s_config.load_kube_config(context=config.get('context', 'local')) + + self.__subsets = None + use_endpoints = config.get('use_endpoints') and (config.get('patronictl') or 'pod_ip' in config) + if use_endpoints: + addresses = [k8s_client.V1EndpointAddress(ip=config['pod_ip'])] + ports = [] + for p in config.get('ports', [{}]): + port = {'port': int(p.get('port', '5432'))} + port.update({n: p[n] for n in ('name', 'protocol') if p.get(n)}) + ports.append(k8s_client.V1EndpointPort(**port)) + self.__subsets = [k8s_client.V1EndpointSubset(addresses=addresses, ports=ports)] + self._api = CoreV1Api(use_endpoints) + self.set_retry_timeout(config['retry_timeout']) + self.set_ttl(config.get('ttl') or 30) + self._leader_observed_record = {} + self._leader_observed_time = None + self._leader_resource_version = None + self._leader_observed_subsets = [] + self.__do_not_watch = False + + def retry(self, *args, **kwargs): + return self._retry.copy()(*args, **kwargs) + + def catch_kubernetes_errors(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except (RetryFailedError, k8s_client.rest.ApiException, + HTTPException, HTTPError, socket.error, socket.timeout): + return False + return wrapper + + def client_path(self, path): + return super(Kubernetes, self).client_path(path)[1:].replace('/', '-') + + @property + def leader_path(self): + return self._base_path[1:] if self.__subsets else super(Kubernetes, self).leader_path + + def set_ttl(self, ttl): + ttl = int(ttl) + self.__do_not_watch = self._ttl != ttl + self._ttl = ttl + + def set_retry_timeout(self, retry_timeout): + self._retry.deadline = retry_timeout + self._api.set_timeout(retry_timeout) + + @staticmethod + def member(pod): + annotations = pod.metadata.annotations or {} + member = Member.from_node(pod.metadata.resource_version, pod.metadata.name, None, annotations.get('status', '')) + member.data['pod_labels'] = pod.metadata.labels + return member + + def _load_cluster(self): + try: + # get list of members + response = self.retry(self._api.list_namespaced_pod, self._namespace, label_selector=self._label_selector) + members = [self.member(pod) for pod in response.items] + + response = self.retry(self._api.list_namespaced_kind, self._namespace, label_selector=self._label_selector) + nodes = {item.metadata.name: item for item in response.items} + + config = nodes.get(self.config_path) + metadata = config and config.metadata + annotations = metadata and metadata.annotations or {} + + # get initialize flag + initialize = annotations.get(self._INITIALIZE) + + # get global dynamic configuration + config = ClusterConfig.from_node(metadata and metadata.resource_version, + annotations.get(self._CONFIG) or '{}') + + leader = nodes.get(self.leader_path) + metadata = leader and leader.metadata + self._leader_resource_version = metadata.resource_version if metadata else None + self._leader_observed_subsets = leader.subsets if self.__subsets and leader else [] + annotations = metadata and metadata.annotations or {} + + # get last leader operation + last_leader_operation = annotations.get(self._OPTIME) + last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation) + + # get leader + leader_record = {n: annotations.get(n) for n in (self._LEADER, 'acquireTime', + 'ttl', 'renewTime', 'transitions') if n in annotations} + if (leader_record or self._leader_observed_record) and leader_record != self._leader_observed_record: + self._leader_observed_record = leader_record + self._leader_observed_time = time.time() + + leader = leader_record.get(self._LEADER) + try: + ttl = int(leader_record.get('ttl')) or self._ttl + except (TypeError, ValueError): + ttl = self._ttl + + if not metadata or not self._leader_observed_time or self._leader_observed_time + ttl < time.time(): + leader = None + + if metadata: + member = Member(-1, leader, None, {}) + member = ([m for m in members if m.name == leader] or [member])[0] + leader = Leader(response.metadata.resource_version, None, member) + + # failover key + failover = nodes.get(self.failover_path) + metadata = failover and failover.metadata + failover = Failover.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) + + # get synchronization state + sync = nodes.get(self.sync_path) + metadata = sync and sync.metadata + sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) + + self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) + except Exception: + logger.exception('get_cluster') + raise KubernetesError('Kubernetes API is not responding properly') + + @staticmethod + def compare_ports(p1, p2): + return p1.name == p2.name and p1.port == p2.port and (p1.protocol or 'TCP') == (p2.protocol or 'TCP') + + @staticmethod + def subsets_changed(last_observed_subsets, subsets): + """ + >>> Kubernetes.subsets_changed([], []) + False + >>> Kubernetes.subsets_changed([], [k8s_client.V1EndpointSubset()]) + True + >>> s1 = [k8s_client.V1EndpointSubset(addresses=[k8s_client.V1EndpointAddress(ip='1.2.3.4')])] + >>> s2 = [k8s_client.V1EndpointSubset(addresses=[k8s_client.V1EndpointAddress(ip='1.2.3.5')])] + >>> Kubernetes.subsets_changed(s1, s2) + True + >>> a = [k8s_client.V1EndpointAddress(ip='1.2.3.4')] + >>> s1 = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(protocol='TCP', port=1)])] + >>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(port=5432)])] + >>> Kubernetes.subsets_changed(s1, s2) + True + >>> p1 = k8s_client.V1EndpointPort(name='port1', port=1) + >>> p2 = k8s_client.V1EndpointPort(name='port2', port=2) + >>> p3 = k8s_client.V1EndpointPort(name='port3', port=3) + >>> s1 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p1, p2])] + >>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p3])] + >>> Kubernetes.subsets_changed(s1, s2) + True + >>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p1])] + >>> Kubernetes.subsets_changed(s1, s2) + False + """ + if len(last_observed_subsets) != len(subsets): + return True + if subsets == []: + return False + if len(last_observed_subsets[0].addresses or []) != 1 or \ + last_observed_subsets[0].addresses[0].ip != subsets[0].addresses[0].ip or \ + len(last_observed_subsets[0].ports) != len(subsets[0].ports): + return True + if len(subsets[0].ports) == 1: + return not Kubernetes.compare_ports(last_observed_subsets[0].ports[0], subsets[0].ports[0]) + observed_ports = {p.name: p for p in last_observed_subsets[0].ports} + for p in subsets[0].ports: + if p.name not in observed_ports or not Kubernetes.compare_ports(p, observed_ports.pop(p.name)): + return True + return False + + @catch_kubernetes_errors + def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, subsets=None): + metadata = {'namespace': self._namespace, 'name': name, 'labels': self._labels, 'annotations': annotations} + if patch or resource_version: + if resource_version is not None: + metadata['resource_version'] = resource_version + func = functools.partial(self._api.patch_namespaced_kind, name) + else: + func = functools.partial(self._api.create_namespaced_kind) + # skip annotations with null values + metadata['annotations'] = {k: v for k, v in metadata['annotations'].items() if v is not None} + + metadata = k8s_client.V1ObjectMeta(**metadata) + if subsets is not None and self.__subsets: + endpoints = {'metadata': metadata} + if self.subsets_changed(self._leader_observed_subsets, subsets): + endpoints['subsets'] = subsets + body = k8s_client.V1Endpoints(**endpoints) + else: + body = k8s_client.V1ConfigMap(metadata=metadata) + return self.retry(func, self._namespace, body) if retry else func(self._namespace, body) + + def _write_leader_optime(self, last_operation): + """Unused""" + + def _update_leader(self): + """Unused""" + + def update_leader(self, last_operation): + now = datetime.datetime.now(tzutc).isoformat() + annotations = {self._LEADER: self._name, 'ttl': str(self._ttl), 'renewTime': now, + 'acquireTime': self._leader_observed_record.get('acquireTime') or now, + 'transitions': self._leader_observed_record.get('transitions') or '0'} + if last_operation: + annotations[self._OPTIME] = last_operation + + ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=self.__subsets) + if ret: + self._leader_resource_version = ret.metadata.resource_version + return ret + + def attempt_to_acquire_leader(self, permanent=False): + now = datetime.datetime.now(tzutc).isoformat() + annotations = {self._LEADER: self._name, 'ttl': str(sys.maxsize if permanent else self._ttl), + 'renewTime': now, 'acquireTime': now, 'transitions': '0'} + if self._leader_observed_record: + try: + transitions = int(self._leader_observed_record.get('transitions')) + except (TypeError, ValueError): + transitions = 0 + + if self._leader_observed_record.get(self._LEADER) != self._name: + transitions += 1 + else: + annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now + annotations['transitions'] = str(transitions) + ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, subsets=self.__subsets) + if ret: + self._leader_resource_version = ret.metadata.resource_version + else: + logger.info('Could not take out TTL lock') + return ret + + def take_leader(self): + return self.attempt_to_acquire_leader() + + def set_failover_value(self, value, index=None): + """Unused""" + + def manual_failover(self, leader, candidate, scheduled_at=None, index=None): + annotations = {'leader': leader or None, 'member': candidate or None, 'scheduled_at': scheduled_at} + patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.index) + return self.patch_or_create(self.failover_path, annotations, index, bool(index or patch), False) + + def set_config_value(self, value, index=None): + patch = bool(index or self.cluster and self.cluster.config and self.cluster.config.index) + return self.patch_or_create(self.config_path, {self._CONFIG: value}, index, patch, False) + + @catch_kubernetes_errors + def touch_member(self, data, ttl=None, permanent=False): + cluster = self.cluster + if cluster and cluster.leader and cluster.leader.name == self._name: + role = 'master' + elif data['state'] == 'running' and data['role'] != 'master': + role = data['role'] + else: + role = None + + member = cluster and cluster.get_member(self._name, fallback_to_leader=False) + pod_labels = member and member.data.pop('pod_labels', None) + ret = pod_labels is not None and pod_labels.get(self._role_label) == role and deep_compare(data, member.data) + + if not ret: + metadata = {'namespace': self._namespace, 'name': self._name, 'labels': {self._role_label: role}, + 'annotations': {'status': json.dumps(data, separators=(',', ':'))}} + body = k8s_client.V1Pod(metadata=k8s_client.V1ObjectMeta(**metadata)) + ret = self._api.patch_namespaced_pod(self._name, self._namespace, body) + return ret + + def initialize(self, create_new=True, sysid=""): + cluster = self.cluster + resource_version = cluster.config.index if cluster and cluster.config and cluster.config.index else None + return self.patch_or_create(self.config_path, {self._INITIALIZE: sysid}, resource_version) + + def delete_leader(self): + if self.cluster and isinstance(self.cluster.leader, Leader) and self.cluster.leader.name == self._name: + self.patch_or_create(self.leader_path, {self._LEADER: None}, self._leader_resource_version, True, False, []) + self.reset_cluster() + + def cancel_initialization(self): + self.patch_or_create(self.config_path, {self._INITIALIZE: None}, self.cluster.config.index, True) + + @catch_kubernetes_errors + def delete_cluster(self): + self.retry(self._api.delete_collection_namespaced_kind, self._namespace, label_selector=self._label_selector) + + def set_sync_state_value(self, value, index=None): + """Unused""" + + def write_sync_state(self, leader, sync_standby, index=None): + return self.patch_or_create(self.sync_path, self.sync_state(leader, sync_standby), index, False) + + def delete_sync_state(self, index=None): + return self.write_sync_state(None, None, index) + + def watch(self, leader_index, timeout): + if self.__do_not_watch: + self.__do_not_watch = False + return True + + if leader_index: + end_time = time.time() + timeout + w = k8s_watch.Watch() + while timeout >= 1: + try: + for event in w.stream(self._api.list_namespaced_kind, self._namespace, + resource_version=leader_index, timeout_seconds=int(timeout + 0.5), + field_selector='metadata.name=' + self.leader_path, + _request_timeout=(1, timeout + 1)): + return event['raw_object'].get('metadata', {}).get('resourceVersion') != leader_index + return False + except KeyboardInterrupt: + raise + except Exception: + logging.exception('watch') + + timeout = end_time - time.time() + + try: + return super(Kubernetes, self).watch(None, timeout) + finally: + self.event.clear() diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 43bcb42b..40ae3cb9 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -1,3 +1,4 @@ +import json import logging import time @@ -6,6 +7,7 @@ from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.handlers.threading import SequentialThreadingHandler from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState from patroni.exceptions import DCSError +from patroni.utils import deep_compare logger = logging.getLogger(__name__) @@ -57,7 +59,7 @@ class ZooKeeper(AbstractDCS): max_delay=1, max_tries=-1, sleep_func=time.sleep)) self._client.add_listener(self.session_listener) - self._my_member_data = None + self._my_member_data = {} self._fetch_cluster = True self._orig_kazoo_connect = self._client._connection._connect @@ -242,7 +244,7 @@ class ZooKeeper(AbstractDCS): def touch_member(self, data, ttl=None, permanent=False): cluster = self.cluster member = cluster and cluster.get_member(self._name, fallback_to_leader=False) - data = data.encode('utf-8') + encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8') if member and self._client.client_id is not None and member.session != self._client.client_id[0]: try: self._client.delete_async(self.member_path).get(timeout=1) @@ -253,11 +255,12 @@ class ZooKeeper(AbstractDCS): member = None if member: - if data == self._my_member_data: + if deep_compare(data, self._my_member_data): return True else: try: - self._client.create_async(self.member_path, data, makepath=True, ephemeral=not permanent).get(timeout=1) + self._client.create_async(self.member_path, encoded_data, makepath=True, + ephemeral=not permanent).get(timeout=1) self._my_member_data = data return True except Exception as e: @@ -265,7 +268,7 @@ class ZooKeeper(AbstractDCS): logger.exception('touch_member') return False try: - self._client.set_async(self.member_path, data).get(timeout=1) + self._client.set_async(self.member_path, encoded_data).get(timeout=1) self._my_member_data = data return True except Exception: @@ -291,7 +294,7 @@ class ZooKeeper(AbstractDCS): logger.exception('Failed to update %s', self.leader_optime_path) return False - def update_leader(self): + def _update_leader(self): return True def delete_leader(self): diff --git a/patroni/ha.py b/patroni/ha.py index 67c25c31..9971c170 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -86,14 +86,15 @@ class Ha(object): return self.dcs.attempt_to_acquire_leader() def update_lock(self, write_leader_optime=False): - ret = self.dcs.update_leader() + last_operation = None + if write_leader_optime: + try: + last_operation = self.state_handler.last_operation() + except Exception: + logger.exception('Exception when called state_handler.last_operation()') + ret = self.dcs.update_leader(last_operation) if ret: self.watchdog.keepalive() - if write_leader_optime: - try: - self.dcs.write_leader_optime(self.state_handler.last_operation()) - except Exception: - pass return ret def has_lock(self): @@ -132,7 +133,7 @@ class Ha(object): scheduled_restart_data['schedule'] = scheduled_restart_data['schedule'].isoformat() data['scheduled_restart'] = scheduled_restart_data - return self.dcs.touch_member(json.dumps(data, separators=(',', ':'))) + return self.dcs.touch_member(data) def clone(self, clone_member=None, msg='(without leader)'): if self.state_handler.clone(clone_member): diff --git a/requirements.txt b/requirements.txt index a66aaef5..0a4bb6da 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,10 @@ -urllib3>=1.9 +urllib3>=1.19.1,!=1.21 boto -psycopg2>=2.6.1 +psycopg2>=2.5.4 PyYAML requests six >= 1.7 -kazoo==2.2.1 +kazoo>=1.3.1 python-etcd>=0.4.3,<0.5 python-consul>=0.7.0 click>=4.1 @@ -13,3 +13,4 @@ tzlocal python-dateutil psutil cdiff +kubernetes==3.0.0 diff --git a/setup.py b/setup.py index 424d4d90..2d28ebab 100644 --- a/setup.py +++ b/setup.py @@ -115,13 +115,23 @@ def read(fname): def setup_package(): # Assemble additional setup commands - cmdclass = {} - cmdclass['test'] = PyTest + cmdclass = {'test': PyTest} # Some helper variables version = os.getenv('GO_PIPELINE_LABEL', VERSION) - install_reqs = get_install_requirements('requirements.txt') + install_requires = [] + extras_require = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'], + 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']} + + for r in get_install_requirements('requirements.txt'): + extra = False + for e, v in extras_require.items(): + if r.startswith(v[0]): + extras_require[e] = [r] + extra = True + if not extra: + install_requires.append(r) command_options = {'test': {'test_suite': ('setup.py', 'tests')}} if JUNIT_XML: @@ -145,10 +155,10 @@ def setup_package(): test_suite='tests', packages=find_packages(exclude=['tests', 'tests.*']), package_data={MAIN_PACKAGE: ["*.json"]}, - install_requires=install_reqs, - setup_requires=['flake8'], + install_requires=install_requires, + extras_require=extras_require, cmdclass=cmdclass, - tests_require=['mock>=2.0.0', 'pytest-cov', 'pytest'], + tests_require=['flake8', 'mock>=2.0.0', 'pytest-cov', 'pytest'], command_options=command_options, entry_points={'console_scripts': CONSOLE_SCRIPTS}, ) diff --git a/tests/test_config.py b/tests/test_config.py index 9e7ddafa..26d64395 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -49,6 +49,9 @@ class TestConfig(unittest.TestCase): 'PATRONI_ETCD_CERT': '/cert', 'PATRONI_ETCD_KEY': '/key', 'PATRONI_CONSUL_HOST': '127.0.0.1:8500', + 'PATRONI_KUBERNETES_LABELS': 'a:b:c', + 'PATRONI_KUBERNETES_SCOPE_LABEL': 'a', + 'PATRONI_KUBERNETES_PORTS': '[{"name": "postgresql"}]', 'PATRONI_ZOOKEEPER_HOSTS': "'host1:2181','host2:2181'", 'PATRONI_EXHIBITOR_HOSTS': 'host1,host2', 'PATRONI_EXHIBITOR_PORT': '8181', diff --git a/tests/test_consul.py b/tests/test_consul.py index 5703eb28..47df9550 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -112,11 +112,11 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException])) def test_touch_member(self): self.c.refresh_session = Mock(return_value=True) - self.c.touch_member('balbla') - self.c.touch_member('balbla') - self.c.touch_member('balbla') + self.c.touch_member({'balbla': 'blabla'}) + self.c.touch_member({'balbla': 'blabla'}) + self.c.touch_member({'balbla': 'blabla'}) self.c.refresh_session = Mock(return_value=False) - self.c.touch_member('balbla') + self.c.touch_member({'balbla': 'blabla'}) @patch.object(consul.Consul.KV, 'put', Mock(return_value=False)) def test_take_leader(self): @@ -138,7 +138,7 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.Session, 'renew', Mock()) def test_update_leader(self): - self.c.update_leader() + self.c.update_leader(None) @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)) def test_delete_leader(self): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b9ba6475..90f523d1 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -288,7 +288,7 @@ class TestEtcd(unittest.TestCase): self.etcd.write_leader_optime('0') def test_update_leader(self): - self.assertTrue(self.etcd.update_leader()) + self.assertTrue(self.etcd.update_leader(None)) def test_initialize(self): self.assertFalse(self.etcd.initialize()) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py new file mode 100644 index 00000000..c9c012e9 --- /dev/null +++ b/tests/test_kubernetes.py @@ -0,0 +1,103 @@ +import unittest + +from mock import Mock, patch +from patroni.dcs.kubernetes import Kubernetes, KubernetesError, k8s_client, k8s_watch + + +def mock_list_namespaced_config_map(self, *args, **kwargs): + metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config', + 'annotations': {'initialize': '123', 'config': '{}'}} + items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))] + metadata.update({'name': 'test-leader', 'annotations': {'optime': '1234', 'leader': 'p-0', 'ttl': '30s'}}) + items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) + metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}}) + items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) + metadata.update({'name': 'test-sync', 'annotations': {'leader': 'p-0'}}) + items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) + metadata = k8s_client.V1ObjectMeta(resource_version='1') + return k8s_client.V1ConfigMapList(metadata=metadata, items=items) + + +def mock_list_namespaced_pod(self, *args, **kwargs): + metadata = k8s_client.V1ObjectMeta(resource_version='1', name='p-0', annotations={'status': '{}'}) + items = [k8s_client.V1Pod(metadata=metadata)] + return k8s_client.V1PodList(items=items) + + +@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', Mock()) +@patch.object(k8s_client.CoreV1Api, 'create_namespaced_config_map', Mock()) +class TestKubernetes(unittest.TestCase): + + @patch('kubernetes.config.load_kube_config', Mock()) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod) + def setUp(self): + self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}}) + with patch('time.time', Mock(return_value=1)): + self.k.get_cluster() + + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod) + def test_get_cluster(self): + self.k.get_cluster() + with patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', Mock(side_effect=Exception)): + self.assertRaises(KubernetesError, self.k.get_cluster) + + @patch('kubernetes.config.load_kube_config', Mock()) + @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock()) + def test_update_leader(self): + k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, + 'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'}) + self.assertIsNotNone(k.update_leader('123')) + + def test_take_leader(self): + self.k.take_leader() + self.k._leader_observed_record['leader'] = 'test' + self.k.patch_or_create = Mock(return_value=False) + self.k.take_leader() + + def test_manual_failover(self): + self.k.manual_failover('foo', 'bar') + + def test_set_config_value(self): + self.k.set_config_value('{}') + + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', Mock(return_value=True)) + def test_touch_member(self): + self.k.touch_member({}) + self.k._name = 'p-1' + self.k.touch_member({'state': 'running', 'role': 'replica'}) + self.k.touch_member({'state': 'stopped', 'role': 'master'}) + + def test_initialize(self): + self.k.initialize() + + def test_delete_leader(self): + self.k.delete_leader() + + def test_cancel_initialization(self): + self.k.cancel_initialization() + + @patch.object(k8s_client.CoreV1Api, 'delete_collection_namespaced_config_map', + Mock(side_effect=k8s_client.rest.ApiException(500, ''))) + def test_delete_cluster(self): + self.k.delete_cluster() + + @patch('kubernetes.config.load_kube_config', Mock()) + @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', + Mock(side_effect=[k8s_client.rest.ApiException(502, ''), k8s_client.rest.ApiException(500, '')])) + def test_delete_sync_state(self): + k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, + 'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'}) + self.assertFalse(k.delete_sync_state()) + + def test_watch(self): + self.k.set_ttl(10) + self.k.watch(None, 0) + self.k.watch(None, 0) + with patch.object(k8s_watch.Watch, 'stream', + Mock(side_effect=[Exception, [], KeyboardInterrupt, + [{'raw_object': {'metadata': {'resourceVersion': '2'}}}]])): + self.assertFalse(self.k.watch('1', 2)) + self.assertRaises(KeyboardInterrupt, self.k.watch, '1', 2) + self.assertTrue(self.k.watch('1', 2)) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 9166138a..c70949cd 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -58,11 +58,11 @@ class MockKazooClient(Mock): raise TypeError("Invalid type for 'path' (string expected)") if not isinstance(value, (six.binary_type,)): raise TypeError("Invalid type for 'value' (must be a byte string)") - if value == b'Exception': + if b'Exception' in value: raise Exception if path.endswith('/initialize') or path == '/service/test/optime/leader': raise Exception - elif value == b'retry' or (value == b'exists' and self.exists): + elif b'retry' in value or (b'exists' in value and self.exists): raise NodeExistsError def create_async(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False): @@ -76,10 +76,10 @@ class MockKazooClient(Mock): raise TypeError("Invalid type for 'value' (must be a byte string)") if path == '/service/bla/optime/leader': raise Exception - if path == '/service/test/members/bar' and value == b'retry': + if path == '/service/test/members/bar' and b'retry' in value: return if path in ('/service/test/failover', '/service/test/config', '/service/test/sync'): - if value == b'Exception': + if b'Exception' in value: raise Exception elif value == b'ok': return @@ -145,7 +145,7 @@ class TestZooKeeper(unittest.TestCase): self.assertRaises(ZooKeeperError, self.zk.get_cluster) cluster = self.zk.get_cluster() self.assertIsInstance(cluster.leader, Leader) - self.zk.touch_member('foo') + self.zk.touch_member({'foo': 'foo'}) def test_delete_leader(self): self.assertTrue(self.zk.delete_leader()) @@ -169,17 +169,17 @@ class TestZooKeeper(unittest.TestCase): def test_touch_member(self): self.zk._name = 'buzz' self.zk.get_cluster() - self.zk.touch_member('new') + self.zk.touch_member({'new': 'new'}) self.zk._name = 'bar' - self.zk.touch_member('new') + self.zk.touch_member({'new': 'new'}) self.zk._name = 'na' self.zk._client.exists = 1 - self.zk.touch_member('Exception') + self.zk.touch_member({'Exception': 'Exception'}) self.zk._name = 'bar' - self.zk.touch_member('retry') + self.zk.touch_member({'retry': 'retry'}) self.zk._fetch_cluster = True self.zk.get_cluster() - self.zk.touch_member('retry') + self.zk.touch_member({'retry': 'retry'}) def test_take_leader(self): self.zk.take_leader() @@ -187,7 +187,7 @@ class TestZooKeeper(unittest.TestCase): self.zk.take_leader() def test_update_leader(self): - self.assertTrue(self.zk.update_leader()) + self.assertTrue(self.zk.update_leader(None)) def test_write_leader_optime(self): self.zk.last_leader_operation = '0'