Support sendInitialEvents in the dynamic client watch

kubernetes/base/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 = []

kubernetes/base/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()