ported asyncio watch from github.com/tomplus/kubernetes_asyncio

CHANGELOG.md

@@ -1,3 +1,23 @@
+# v36.0.3
+
+Kubernetes API Version: v1.36.2
+
+### Bug or Regression
+- Fix Watch.stream selecting watch instead of follow when streaming pod logs.
+- Start the leader election worker thread as a daemon so it does not block process shutdown.
+- Fix readline_channel, readline_stdout and readline_stderr crashing with OverflowError when using the default timeout, and returning None when a timeout expires.
+- Fix format_quantity returning imprecise, non-canonical values for the milli, micro and nano suffixes, and ignoring quantize=Decimal(0).
+
+# v36.0.2
+
+Kubernetes API Version: v1.36.1
+
+### Uncategorized
+- Restored backward compatibility for `Configuration.auth_settings()`:
+  the legacy `api_key['authorization']` lookup is honored as a fallback
+  when `api_key['BearerToken']` is not set, fixing 401 Unauthorized
+  regressions seen after upgrading to v36.0.0 (#2595). (#2604, @GK-07)
+
 # v36.0.1
 
 Kubernetes API Version: v1.36.1

README.md

@@ -134,7 +134,7 @@ supported versions of Kubernetes clusters.
 - [client 33.y.z](https://pypi.org/project/kubernetes/33.1.0/): Kubernetes 1.32 or below (+-), Kubernetes 1.33 (✓), Kubernetes 1.34 or above (+-)
 - [client 34.y.z](https://pypi.org/project/kubernetes/34.1.0/): Kubernetes 1.33 or below (+-), Kubernetes 1.34 (✓), Kubernetes 1.35 or above (+-)
 - [client 35.y.z](https://pypi.org/project/kubernetes/35.0.0/): Kubernetes 1.34 or below (+-), Kubernetes 1.35 (✓), Kubernetes 1.36 or above (+-)
-- [client 36.y.z](https://pypi.org/project/kubernetes/36.0.1/): Kubernetes 1.35 or below (+-), Kubernetes 1.36 (✓), Kubernetes 1.37 or above (+-)
+- [client 36.y.z](https://pypi.org/project/kubernetes/36.0.3/): Kubernetes 1.35 or below (+-), Kubernetes 1.36 (✓), Kubernetes 1.37 or above (+-)
 
 
 > See [here](#homogenizing-the-kubernetes-python-client-versions) for an explanation of why there is no v13-v16 release.

examples_asyncio/watch_namespaces.py

@@ -0,0 +1,31 @@
+import asyncio
+
+from kubernetes.aio import client, config, watch
+
+
+async def main():
+    # Configs can be set in Configuration class directly or using helper
+    # utility. If no argument provided, the config will be loaded from
+    # default location.
+    await config.load_kube_config()
+
+    v1 = client.CoreV1Api()
+    count = 10
+    w = watch.Watch()
+
+    async for event in w.stream(v1.list_namespace, timeout_seconds=10):
+        print("Event: {} {}".format(event["type"], event["object"].metadata.name))
+        count -= 1
+        if not count:
+            w.stop()
+
+    print("Ended.")
+    # An explicit close is necessary to stop the stream
+    # or use async context manager like in example4.py
+    await w.close()
+
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main())
+    loop.close()

examples_asyncio/watch_ns_pods.py

@@ -0,0 +1,49 @@
+"""Watch multiple K8s event streams without threads."""
+
+import asyncio
+
+from kubernetes.aio import client, config, watch
+
+
+async def watch_namespaces():
+    async with client.ApiClient() as api:
+        v1 = client.CoreV1Api(api)
+        async with watch.Watch().stream(v1.list_namespace) as stream:
+            async for event in stream:
+                etype, obj = event["type"], event["object"]
+                print("{} namespace {}".format(etype, obj.metadata.name))
+
+
+async def watch_pods():
+    async with client.ApiClient() as api:
+        v1 = client.CoreV1Api(api)
+        async with watch.Watch().stream(v1.list_pod_for_all_namespaces) as stream:
+            async for event in stream:
+                evt, obj = event["type"], event["object"]
+                print(
+                    "{} pod {} in NS {}".format(
+                        evt, obj.metadata.name, obj.metadata.namespace
+                    )
+                )
+
+
+def main():
+    loop = asyncio.get_event_loop()
+
+    # Load the kubeconfig file specified in the KUBECONFIG environment
+    # variable, or fall back to `~/.kube/config`.
+    loop.run_until_complete(config.load_kube_config())
+
+    # Define the tasks to watch namespaces and pods.
+    tasks = [
+        asyncio.ensure_future(watch_namespaces()),
+        asyncio.ensure_future(watch_pods()),
+    ]
+
+    # Push tasks into event loop.
+    loop.run_until_complete(asyncio.wait(tasks))
+    loop.close()
+
+
+if __name__ == "__main__":
+    main()

kubernetes/aio/watch/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2016 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .watch import Watch

kubernetes/aio/watch/watch.py

@@ -0,0 +1,257 @@
+# Copyright 2016 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+import json
+import pydoc
+from functools import partial
+from types import SimpleNamespace
+
+from kubernetes.aio import client
+
+PYDOC_RETURN_LABEL = ":rtype:"
+PYDOC_FOLLOW_PARAM = ":param follow:"
+
+# Removing this suffix from return type name should give us event's object
+# type. e.g., if list_namespaces() returns "NamespaceList" type,
+# then list_namespaces(watch=true) returns a stream of events with objects
+# of type "Namespace". In case this assumption is not true, user should
+# provide return_type to Watch class's __init__.
+TYPE_LIST_SUFFIX = "List"
+
+
+def _find_return_type(func):
+    for line in pydoc.getdoc(func).splitlines():
+        if line.startswith(PYDOC_RETURN_LABEL):
+            return line[len(PYDOC_RETURN_LABEL):].strip()
+    return ""
+
+
+class Stream(object):
+
+    def __init__(self, func, *args, **kwargs):
+        pass
+
+
+class Watch(object):
+
+    def __init__(self, return_type=None):
+        self._raw_return_type = return_type
+        self._stop = False
+        self._api_client = client.ApiClient()
+        self.resource_version = None
+        self.resp = None
+
+    def stop(self):
+        self._stop = True
+
+    def get_return_type(self, func):
+        if self._raw_return_type:
+            return self._raw_return_type
+        return_type = _find_return_type(func)
+
+        if return_type.endswith(TYPE_LIST_SUFFIX):
+            return return_type[:-len(TYPE_LIST_SUFFIX)]
+        return return_type
+
+    def get_watch_argument_name(self, func):
+        if PYDOC_FOLLOW_PARAM in pydoc.getdoc(func):
+            return 'follow'
+        else:
+            return 'watch'
+
+    def unmarshal_event(self, data: str, response_type):
+        """Return the K8s response `data` in JSON format.
+
+        """
+        try:
+            js = json.loads(data)
+        except ValueError:
+            return data
+
+        if 'object' not in js or 'type' not in js:
+            # raise error with code if set
+            if 'code' in js:
+                reason = "{}: {}".format(js.get('reason'), js.get('message'))
+                raise client.exceptions.ApiException(status=js['code'], reason=reason)
+
+            raise Exception(("Malformed JSON response, the 'object' and/or "
+                             "'type' field is missing. JSON: {}").format(js))
+        # Make a copy of the original object and save it under the
+        # `raw_object` key because we will replace the data under `object` with
+        # a Python native type shortly.
+        js['raw_object'] = js['object']
+
+        # Something went wrong. A typical example would be that the user
+        # supplied a resource version that was too old. In that case K8s would
+        # not send a conventional ADDED/DELETED/... event but an error. Turn
+        # this error into a Python exception to save the user the hassle.
+        if js['type'].lower() == 'error':
+            obj = js['raw_object']
+            reason = "{}: {}".format(obj['reason'], obj['message'])
+            raise client.exceptions.ApiException(status=obj['code'], reason=reason)
+
+        if js['type'].lower() != 'bookmark':
+            # If possible, compile the JSON response into a Python native response
+            # type, eg `V1Namespace` or `V1Pod`,`ExtensionsV1beta1Deployment`, ...
+            if response_type:
+                js['object'] = self._api_client.deserialize(
+                    response=SimpleNamespace(data=json.dumps(js['raw_object'])),
+                    response_type=response_type
+                )
+
+            # decode and save resource_version to continue watching
+            if hasattr(js['object'], 'metadata'):
+                self.resource_version = js['object'].metadata.resource_version
+
+            # For custom objects that we don't have model defined, json
+            # deserialization results in dictionary
+            elif (isinstance(js['object'], dict)
+                    and 'metadata' in js['object']
+                    and 'resourceVersion' in js['object']['metadata']):
+                self.resource_version = js['object']['metadata']['resourceVersion']
+
+        elif js['type'].lower() == 'bookmark':
+            if (isinstance(js['raw_object'], dict)
+                    and 'metadata' in js['raw_object']
+                    and 'resourceVersion' in js['raw_object']['metadata']):
+                self.resource_version = js['raw_object']['metadata']['resourceVersion']
+            else:
+                raise Exception(("Malformed JSON response for bookmark event, "
+                                 "'metadata' or 'resourceVersion' field is missing. "
+                                 "JSON: {}").format(js))
+
+        return js
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            return await self.next()
+        except:  # noqa: E722
+            await self.close()
+            raise
+
+    def _reconnect(self):
+        self.resp.close()
+        self.resp = None
+        if self.resource_version:
+            self.func.keywords['resource_version'] = self.resource_version
+
+    async def next(self):
+
+        watch_forever = 'timeout_seconds' not in self.func.keywords
+        retry_410 = watch_forever
+
+        while 1:
+
+            # Set the response object to the user supplied function (eg
+            # `list_namespaced_pods`) if this is the first iteration.
+            if self.resp is None:
+                self.resp = await self.func()
+
+            # Abort at the current iteration if the user has called `stop` on this
+            # stream instance.
+            if self._stop:
+                raise StopAsyncIteration
+
+            # Fetch the next K8s response.
+            try:
+                line = await self.resp.content.readline()
+            except asyncio.TimeoutError:
+                # This exception can be raised by aiohttp (client timeout)
+                # but we don't retry if server side timeout is applied.
+                if watch_forever:
+                    self._reconnect()
+                    continue
+                else:
+                    raise
+
+            line = line.decode('utf8')
+
+            # Special case for faster log streaming
+            if self.return_type == 'str':
+                if line == '':
+                    # end of log
+                    raise StopAsyncIteration
+                return line
+
+            # Stop the iterator if K8s sends an empty response. This happens when
+            # eg the supplied timeout has expired.
+            if line == '':
+                if watch_forever:
+                    self._reconnect()
+                    continue
+                raise StopAsyncIteration
+
+            # retry 410 error only once
+            try:
+                event = self.unmarshal_event(line, self.return_type)
+            except client.exceptions.ApiException as ex:
+                if ex.status == 410 and retry_410:
+                    retry_410 = False  # retry only once
+                    self._reconnect()
+                    continue
+                raise
+            retry_410 = watch_forever
+            return event
+
+    def stream(self, func, *args, **kwargs):
+        """Watch an API resource and stream the result back via a generator.
+
+        :param func: The API function pointer. Any parameter to the function
+                     can be passed after this parameter.
+
+        :return: Event object with these keys:
+                   'type': The type of event such as "ADDED", "DELETED", etc.
+                   'raw_object': a dict representing the watched object.
+                   'object': A model representation of raw_object. The name of
+                             model will be determined based on
+                             the func's doc string. If it cannot be determined,
+                             'object' value will be the same as 'raw_object'.
+
+        Example:
+            v1 = kubernetes.aio.client.CoreV1Api()
+            watch = kubernetes.aio.watch.Watch()
+            async for e in watch.stream(v1.list_namespace, timeout_seconds=10):
+                type = e['type']
+                object = e['object']  # object is one of type return_type
+                raw_object = e['raw_object']  # raw_object is a dict
+                ...
+                if should_stop:
+                    watch.stop()
+        """
+        self._stop = False
+        self.return_type = self.get_return_type(func)
+        kwargs[self.get_watch_argument_name(func)] = True
+        kwargs['_preload_content'] = False
+        if 'resource_version' in kwargs:
+            self.resource_version = kwargs['resource_version']
+
+        self.func = partial(func, *args, **kwargs)
+
+        return self
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        await self.close()
+
+    async def close(self):
+        await self._api_client.close()
+        if self.resp is not None:
+            self.resp.release()
+            self.resp = None