diff --git a/kubernetes/dynamic/client.py b/kubernetes/dynamic/client.py index 6e62a96c3a..05f6bdb81c 100644 --- a/kubernetes/dynamic/client.py +++ b/kubernetes/dynamic/client.py @@ -158,7 +158,7 @@ def server_side_apply(self, resource, body=None, name=None, namespace=None, forc return self.request('patch', path, body=body, force_conflicts=force_conflicts, **kwargs) - def watch(self, resource, namespace=None, name=None, label_selector=None, field_selector=None, resource_version=None, timeout=None, watcher=None, allow_watch_bookmarks=None): + def watch(self, resource, namespace=None, name=None, label_selector=None, field_selector=None, resource_version=None, timeout=None, watcher=None, allow_watch_bookmarks=None, send_initial_events=None, resource_version_match=None): """ Stream events for a resource from the Kubernetes API @@ -172,6 +172,10 @@ def watch(self, resource, namespace=None, name=None, label_selector=None, field_ :param timeout: The amount of time in seconds to wait before terminating the stream :param watcher: The Watcher object that will be used to stream the resource :param allow_watch_bookmarks: Ask the API server to send BOOKMARK events + :param send_initial_events: Ask the API server to begin the stream with synthetic events + for the current state, followed by a BOOKMARK event + :param resource_version_match: How resource_version is matched, e.g. "NotOlderThan". + Required by the API server when send_initial_events is set :return: Event object with these keys: 'type': The type of event such as "ADDED", "DELETED", etc. @@ -204,6 +208,8 @@ def watch(self, resource, namespace=None, name=None, label_selector=None, field_ serialize=False, timeout_seconds=timeout, allow_watch_bookmarks=allow_watch_bookmarks, + send_initial_events=send_initial_events, + resource_version_match=resource_version_match, ): event['object'] = ResourceInstance(resource, event['object']) yield event @@ -229,6 +235,8 @@ def request(self, method, path, body=None, **params): query_params.append(('limit', params['limit'])) if params.get('resource_version') is not None: query_params.append(('resourceVersion', params['resource_version'])) + if params.get('resource_version_match') is not None: + query_params.append(('resourceVersionMatch', params['resource_version_match'])) if params.get('timeout_seconds') is not None: query_params.append(('timeoutSeconds', params['timeout_seconds'])) if params.get('watch') is not None: @@ -247,6 +255,8 @@ def request(self, method, path, body=None, **params): query_params.append(('force', params['force_conflicts'])) if params.get('allow_watch_bookmarks') is not None: query_params.append(('allowWatchBookmarks', params['allow_watch_bookmarks'])) + if params.get('send_initial_events') is not None: + query_params.append(('sendInitialEvents', params['send_initial_events'])) header_params = params.get('header_params', {}) form_params = [] diff --git a/kubernetes/dynamic/client_test.py b/kubernetes/dynamic/client_test.py index a67a2e2e3f..b424b78e9f 100644 --- a/kubernetes/dynamic/client_test.py +++ b/kubernetes/dynamic/client_test.py @@ -208,6 +208,77 @@ def log_message(self, format, *args): target.server_close() proxy.server_close() + def test_watch_forwards_send_initial_events(self): + class FakeWatcher: + def __init__(self): + self.kwargs = None + + def stream(self, func, **kwargs): + self.kwargs = kwargs + return iter(()) + + class FakeResource: + def get(self, **kwargs): + pass + + dynamic = DynamicClient.__new__(DynamicClient) + watcher = FakeWatcher() + + list(dynamic.watch( + FakeResource(), + namespace='default', + watcher=watcher, + send_initial_events=True, + resource_version_match='NotOlderThan', + )) + + self.assertEqual(True, watcher.kwargs['send_initial_events']) + self.assertEqual('NotOlderThan', watcher.kwargs['resource_version_match']) + + def test_request_builds_send_initial_events_query_params(self): + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.server.request_path = self.path + body = json.dumps({'kind': 'APIResourceList'}).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + dynamic = DynamicClient.__new__(DynamicClient) + dynamic.client = ApiClient(Configuration( + host=f'http://127.0.0.1:{server.server_port}', + proxy='', + no_proxy='', + )) + + dynamic.request( + 'get', + '/apis', + resource_version='0', + resource_version_match='NotOlderThan', + send_initial_events=True, + serializer=lambda _, data: data, + ) + + self.assertEqual( + '/apis?resourceVersion=0&resourceVersionMatch=NotOlderThan' + '&sendInitialEvents=true', + server.request_path, + ) + finally: + server.shutdown() + thread.join() + server.server_close() + if __name__ == '__main__': unittest.main()