Your advanced code editor is loading.

Please wait a moment.

#editor-loading { display: none; }

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 json import pydoc import sys

from kubernetes import client

PYDOC_RETURN_LABEL = “:return:” PYDOC_FOLLOW_PARAM = “:param bool 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”

PY2 = sys.version_info[0] == 2 if PY2: import httplib HTTP_STATUS_GONE = httplib.GONE else: import http HTTP_STATUS_GONE = http.HTTPStatus.GONE

class SimpleNamespace:

def __init__(self, **kwargs):
          self.__dict__.update(kwargs)

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 “”

def iter_resp_lines(resp): buffer = bytearray() for segment in resp.stream(amt=None, decode_content=False):

    # Append the segment (chunk) to the buffer
          #
          # Performance note: depending on contents of buffer and the type+value of segment,
          # encoding segment into the buffer could be a wasteful step. The approach used here
          # simplifies the logic farther down, but in the future it may be reasonable to
          # sacrifice readability for performance.
          if isinstance(segment, bytes):
          buffer.extend(segment)
          elif isinstance(segment, str):
          buffer.extend(segment.encode("utf-8"))
          else:
          raise TypeError(
          f"Received invalid segment type, {type(segment)}, from stream. Accepts only 'str' or 'bytes'.")
          
          Split by newline (safe for utf-8 because multi-byte sequences cannot contain the newline byte)
          next_newline = buffer.find(b'\n')
          while next_newline != -1:
          Convert bytes to a valid utf-8 string, replacing any invalid utf-8 with the '�' character
          line = buffer[:next_newline].decode(
          "utf-8", errors="replace")
          buffer = buffer[next_newline+1:]
          if line:
          yield line
          next_newline = buffer.find(b'\n')

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
          
          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, return_type):
          js = json.loads(data)
          js['raw_object'] = js['object']
          BOOKMARK event is treated the same as ERROR for a quick fix of
          decoding exception
          TODO: make use of the resource_version in BOOKMARK event for more
          efficient WATCH
          if return_type and js['type'] != 'ERROR' and js['type'] != 'BOOKMARK':
          obj = SimpleNamespace(data=json.dumps(js['raw_object']))
          js['object'] = self._api_client.deserialize(obj, return_type)
          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']
          return js
          def stream(self, func, *args, **kwargs):
          """Watch an API resource and stream the result back via a generator.
          Note that watching an API resource can expire. The method tries to
          resume automatically once from the last result, but if that last result
          is too old as well, an ApiException exception will be thrown with
          code 410. In that case you have to recover yourself, probably
          by listing the API resource to obtain the latest state and then
          watching from that state on by setting resource_version to
          one returned from listing.
          :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.client.CoreV1Api()
          watch = kubernetes.watch.Watch()
          for e in watch.stream(v1.list_namespace, resource_version=1127):
          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
          return_type = self.get_return_type(func)
          watch_arg = self.get_watch_argument_name(func)
          kwargs[watch_arg] = True
          kwargs[‘_preload_content’] = False
          if ‘resource_version’ in kwargs:
          self.resource_version = kwargs[‘resource_version’]
          Do not attempt retries if user specifies a timeout.
          We want to ensure we are returning within that timeout.
          disable_retries = (‘timeout_seconds’ in kwargs)
          retry_after_410 = False
          while True:
          resp = func(*args, **kwargs)
          try:
          for line in iter_resp_lines(resp):
          unmarshal when we are receiving events from watch,
          return raw string when we are streaming log
          if watch_arg == “watch”:
          event = self.unmarshal_event(line, return_type)
                              if isinstance(event, dict) 
and event[‘type’] == ‘ERROR’:
          obj = event[‘raw_object’]
          Current request expired, let’s retry, (if enabled)
          but only if we have not already retried.
                              if not disable_retries and not retry_after_410 and \
          
          obj[‘code’] == HTTP_STATUS_GONE:
          retry_after_410 = True
          break
          else:
          reason = “%s: %s” % (
          obj[‘reason’], obj[‘message’])
          raise client.rest.ApiException(
          status=obj[‘code’], reason=reason)
          else:
          retry_after_410 = False
          yield event
          else:
          yield line
          if self._stop:
          break
          finally:
          resp.close()
          resp.release_conn()
          if self.resource_version is not None:
          kwargs[‘resource_version’] = self.resource_version
          else:
          self._stop = True
          if self._stop or disable_retries:
          break

Explanation

This code defines a Watch class for streaming Kubernetes API events in Python. It provides a mechanism to observe changes to resources like pods, deployments, and namespaces in real-time. Here’s a breakdown of the code:

Core Components

Watch class: The main class responsible for managing watch operations.
__init__(self, return_type=None): Initializes the watch object.
return_type: Optional type of the object returned by the watched API function. If not provided, it’s inferred from the function’s docstring.
stop(self): Sets a flag to stop the watch loop.
get_return_type(self, func): Determines the return type of the watched API function, either from the return_type argument or by parsing the function’s docstring.
get_watch_argument_name(self, func): Identifies the name of the watch argument in the API function (e.g., ‘follow’ or ‘watch’).
unmarshal_event(self, data, return_type): Parses the received event data from the stream, deserializes the object, and updates the resource version.
stream(self, func, *args, **kwargs): The core watch method that:

Sets up the watch arguments.
Enters a loop that retrieves data from the API using the provided function.
Iterates over the stream of events received from the API.
Unmarshals events into 'type', 'raw_object', and 'object' dictionary keys.
Handles ‘ERROR’ events, retrying if the API call expired.
Yields the unmarshalled event data.
Handles the stop() flag to terminate the watch.

Key Concepts

Kubernetes API Watch: The Kubernetes API allows you to watch resources for changes. It provides a stream of events (ADD, DELETE, UPDATE) representing modifications.
Resource Version: A unique version identifier used to ensure consistency when working with the Kubernetes API. When watching resources, a resource_version is used to determine the starting point for the watch and ensure you don’t miss any events.
Event Handling: The code handles various event types (ADD, DELETE, UPDATE, ERROR), and provides a mechanism to retry expired watches.

Example Usage

from kubernetes import client
from kubernetes.watch import Watch

v1 = client.CoreV1Api()
watch = Watch()

for event in watch.stream(v1.list_namespace, resource_version="1127"):
type = event['type']
object = event['object']  # object is one of type return_type
raw_object = event['raw_object']  # raw_object is a dict

if should_stop:
watch.stop()

Additional Notes

The code assumes that the API functions have docstrings specifying the return type and watch argument name.
This code simplifies the watch process by automatically inferring the object type from the API function’s docstring, but users can also explicitly specify the return_type if needed.
The code provides retry logic in case the watch expires, ensuring a certain level of robustness in handling transient errors.

Overall

This Python code provides a robust and convenient mechanism to implement watch operations for Kubernetes resources, enabling real-time monitoring and event-driven application development.

Your advanced code editor is loading.

Please wait a moment.

Explanation

Graph

Symbols

We couldn't identify any entrypoints. If you believe this to be incorrect then please contact support.