gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Switch to a streaming watcher (#1189)

Browse Source 
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.
This commit is contained in:
Alexander Kukushkin
2019-10-07 15:16:35 +02:00
committed by GitHub
parent facee0186d
commit 86ee22efab
2 changed files with 91 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/kubernetes.py
+51 -18
View File
@@ -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:
tests/test_kubernetes.py
+40 -6
View File
@@ -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)