From 86ee22efab2cb9c22f5b21bf6a30a6242c2e25c2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 7 Oct 2019 15:16:35 +0200 Subject: [PATCH] Switch to a streaming watcher (#1189) Watch requests to K8s API either streaming the data or close connection by timeout. In any case it requires a second connection open, but opening a new connection every 10 seconds is more expensive for both, Patroni and K8s API. Switching to the streaming model also brings other benefits: we can watch not only on leader object, but also on config and wake up Patroni main thread if the config was changed. --- patroni/dcs/kubernetes.py | 69 +++++++++++++++++++++++++++++---------- tests/test_kubernetes.py | 46 ++++++++++++++++++++++---- 2 files changed, 91 insertions(+), 24 deletions(-) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index fb491058..8b261485 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -11,8 +11,10 @@ from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_ from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError from patroni.utils import deep_compare, tzutc, Retry, RetryFailedError +from urllib3 import Timeout from urllib3.exceptions import HTTPError from six.moves.http_client import HTTPException +from threading import Thread logger = logging.getLogger(__name__) @@ -70,6 +72,48 @@ def catch_kubernetes_errors(func): return wrapper +class KubernetesWatcher(Thread): + + def __init__(self, dcs): + super(KubernetesWatcher, self).__init__() + self.daemon = True + self._dcs = dcs + self._leader_value = None + self._config_value = None + self.start() + + def _process_event(self, event): + ev_type = event.get('type') + metadata = event['raw_object'].get('metadata', {}) + name = metadata.get('name') + annotations_map = {self._dcs.leader_path: self._dcs._LEADER, self._dcs.config_path: self._dcs._CONFIG} + value = None if ev_type == 'DELETED' else metadata.get('annotations', {}).get(annotations_map.get(name)) + + if name == self._dcs.leader_path: + if ev_type != 'ADDED' and self._leader_value and value != self._leader_value: + logger.debug('Leader changed from %s to %s', self._leader_value, value) + self._dcs.event.set() + self._leader_value = value + elif name == self._dcs.config_path: + if value != self._config_value and (ev_type != 'ADDED' or self._config_value): + logger.debug('Config changed to %s', value) + self._dcs.event.set() + self._config_value = value + + def _do_watch(self): + stream = self._dcs.start_watch_stream() + for event in stream: + self._process_event(event) + + def run(self): + while True: + try: + self._do_watch() + except Exception as e: + logger.debug('Watcher.run %r', e) + time.sleep(self._dcs._retry.deadline) + + class Kubernetes(AbstractDCS): def __init__(self, config): @@ -109,6 +153,8 @@ class Kubernetes(AbstractDCS): self._leader_observed_subsets = [] self._config_resource_version = None self.__do_not_watch = False + if not config.get('patronictl'): + self._watcher = KubernetesWatcher(self) def retry(self, *args, **kwargs): return self._retry.copy()(*args, **kwargs) @@ -412,29 +458,16 @@ class Kubernetes(AbstractDCS): def delete_sync_state(self, index=None): return self.write_sync_state(None, None, index) + def start_watch_stream(self): + watch = k8s_watch.Watch() + return watch.stream(self._api.list_namespaced_kind, self._namespace, label_selector=self._label_selector, + _request_timeout=(self._retry.deadline, Timeout.DEFAULT_TIMEOUT)) + 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: - logger.exception('watch') - - timeout = end_time - time.time() - try: return super(Kubernetes, self).watch(None, timeout) finally: diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 93b74807..2faa41a4 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -3,6 +3,8 @@ import unittest from mock import Mock, patch from patroni.dcs.kubernetes import Kubernetes, KubernetesError, k8s_client, k8s_watch, RetryFailedError +from threading import Thread +from . import SleepException def mock_list_namespaced_config_map(self, *args, **kwargs): @@ -27,11 +29,13 @@ def mock_list_namespaced_pod(self, *args, **kwargs): @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', Mock()) @patch.object(k8s_client.CoreV1Api, 'create_namespaced_config_map', Mock()) +@patch.object(Thread, 'start', 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) + @patch.object(Thread, 'start', Mock()) def setUp(self): self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}}) self.k.get_cluster() @@ -101,16 +105,14 @@ class TestKubernetes(unittest.TestCase): 'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'}) self.assertFalse(k.delete_sync_state()) + @patch.object(k8s_watch.Watch, 'stream', Mock(return_value=[{'raw_object': {'metadata': {}}}])) + def test_start_watch_stream(self): + self.assertIsNotNone(self.k.start_watch_stream()) + 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)) def test_set_history_value(self): self.k.set_history_value('{}') @@ -126,3 +128,35 @@ class TestKubernetes(unittest.TestCase): self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'})) self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'})) k.touch_member({'state': 'running', 'role': 'replica'}) + + +@patch('time.sleep', Mock(side_effect=SleepException)) +@patch.object(Kubernetes, 'start_watch_stream') +class TestKubernetesWatcher(unittest.TestCase): + + @patch('kubernetes.config.load_kube_config', Mock()) + @patch.object(Thread, 'start', Mock()) + def setUp(self): + self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}}) + + def test_leader_update(self, mock_stream): + mock_stream.return_value = [ + {'raw_object': {'type': 'MODIFIED', + 'metadata': {'name': self.k.leader_path, 'annotations': {self.k._LEADER: 'foo'}}}}, + {'raw_object': {'type': 'MODIFIED', + 'metadata': {'name': self.k.leader_path, 'annotations': {self.k._LEADER: 'bar'}}}} + ] + self.assertRaises(SleepException, self.k._watcher.run) + + def test_config_update(self, mock_stream): + mock_stream.return_value = [ + {'raw_object': {'type': 'MODIFIED', + 'metadata': {'name': self.k.config_path, 'annotations': {self.k._CONFIG: 'foo'}}}}, + {'raw_object': {'type': 'MODIFIED', + 'metadata': {'name': self.k.config_path, 'annotations': {self.k._CONFIG: 'bar'}}}} + ] + self.assertRaises(SleepException, self.k._watcher.run) + + def test_run(self, mock_stream): + mock_stream.side_effect = Exception + self.assertRaises(SleepException, self.k._watcher.run)