Add resilient streaming

[urimandujano]

This PR introduces a new ResilientWatcher utility to swap in for kubernetes.watch.Watch objects. This new class will "resiliently" watch streams produced by Kubernetes client functions. If an exception we want to retry is raised during the stream, the new utility will sleep for 1 second before restarting the stream from the begining. All other exceptions are raised and halt the stream.

The behavior of restarting the stream from the beginning means its possible to produce duplicate objects while streaming. This PR helps to address that by adding a cache (specifically for streaming API objects) that can track which data has been produced by a stream and filter redundant data out. This is the functionality of the api_objects_stream method - it tracks Kubenetes object IDs in an internal cache. There is another method log_stream which is intended to be used for client functions that return a stream of logs. Since there is no way to reliably deduplicate logs, logs will be repeated on restarted streams.

Closes #96 and #106

Example

# Using `ResilientStreamWatcher` on its own

from kubernetes import client, config

from prefect_kubernetes.utilities import ResilientStreamWatcher

config.load_kube_config(context="docker-desktop")


def list_namespace():
    watcher = ResilientStreamWatcher()
    for ns in watcher.api_object_stream(
        client.CoreV1Api().list_namespace, timeout_seconds=5
    ):
        print(
            f"Got item {ns['object'].metadata.name} with ID {ns['object'].metadata.uid} "
        )


def list_logs():
    watcher = ResilientStreamWatcher()
    for log in watcher.log_stream(
        client.CoreV1Api().read_namespaced_pod_log,
        name="mypod",
        namespace="default",
        follow=True,
        _preload_content=False,
    ):
        print(log)


if __name__ == "__main__":
    list_namespace()
    list_logs()

Screenshots

Checklist

• References any related issue by including "Closes #" or "Closes ".
  • If no issue exists and your change is not a small fix, please create an issue first.
• Includes tests or only affects documentation.
• Passes pre-commit checks.
  • Run pre-commit install && pre-commit run --all locally for formatting and linting.
• Includes screenshots of documentation updates.
  • Run mkdocs serve view documentation locally.
• Summarizes PR's changes in CHANGELOG.md

[desertaxle]

This is really cool! I have a couple of stylistic comments/questions. Also, +1 to @abrookins comment on using dequeue.

[desertaxle]

This function looks like it's a very thin wrapper around _stream and doesn't inherently have any knowledge of logs. Would we be able to remove this method and make _stream 'public' instead?

[urimandujano]

I think that it'll be a good experience to have stream publicly available for a more general use case. I removed log_stream and updated the stream docstring to note it's behaviors on log stream client methods.

[desertaxle]

Very cool implementation!