List+watch cache for Kubernetes objects

[homm]

I've been working on the implementation of zero-cost always green list of the Kubernetes objects for my application. I believe this is similar to k8s.io/client-go/informers module from GO client.

watch.py

The Idea is to list arbitrary resources and then only watch for changes. Here is an example how this could be used:

import asyncio
from typing import cast

from watch import ResourceWatch
from kr8s.asyncio import api
from kr8s.asyncio.objects import Pod


def is_ready_for_connect(pod: Pod):
    if pod.metadata.get("deletionTimestamp"):
        print("-deletionTimestamp", pod.metadata.deletionTimestamp)
        return False

    for condition in (pod.status.get("conditions") or []):
        if condition.type == "Ready" and condition.status == "True":
            break
    else:
        print("-conditions", pod.status.get("conditions"))
        return False

    if not pod.status.get("podIP"):
        print("-pod_ip", pod.status.get("podIP"))
        return False

    return True


async def wait_changes(w: ResourceWatch):
    print("Start waiting")
    async for _ in w.wait_changes():
        snapshot = w.snapshot()
        for pod in snapshot.values():
            pod = cast(Pod, pod)
            print("   ", pod.metadata.name, "ready =", is_ready_for_connect(pod))
    print("Stop waiting")


async def main():
    selector = "app.kubernetes.io/name=robots,app.kubernetes.io/component=redis"
    selector = {}
    w = ResourceWatch(
        Pod, namespace="processing", label_selector=selector,
    )
    wait = asyncio.create_task(wait_changes(w))
    run = asyncio.create_task(w.run_forever())
    done, pending = await asyncio.wait([wait, run], return_when=asyncio.FIRST_COMPLETED)
    for task in pending:
        task.cancel()
    for task in done:
        task.result()

watch_task = asyncio.run(main())

The most important part is calling w.snapshot(), which instantly returns full list of current resources.

Instead of implementing this as a separate library, I'd like to propose to include this functionality right into kr8s. So I want to know your opinion, is it worth this?

[jacobtomlinson]

This sounds very similar to the existing kr8s.watch(label_selector=...) method.

In your version every time you get an event are you iterating all currently watched resources with .snapshot()?

[homm]

This sounds very similar to the existing kr8s.watch method

It does. With important differences.

1. kr8s.watch method doesn't provide any way to tell that the full initial list of resources is fetched. This is a problem then you need consistent list of resources. ResourceWatch provides .wait_changes() interface which guarantees that the list is consistent.
2. kr8s.watch provides only updates. You need a wrapper to get a list. ResourceWatch is this wrapper which maintains always green list of the Kubernetes objects via .snapshot() method.
3. ResourceWatch handles BOOKMARK events which is not possible with current kr8s.watch implementation.

In your version every time you get an event are you iterating all currently watched resources with .snapshot()?

Sorry, I don't get the question. Every time ResourceWatch gets event from the async_get_kind(watch=True), it just updates snapshot and notifies user who .wait_changes(). The user can get the actual resource list at any time with .snapshot() method, not matter on event or just when the user needs a list. I'm not sure what you are calling "iterating", of course the list could be iterated, but this is not an api call:

    def snapshot(self) -> dict[str, APIObject]:
        with self._snapshot_lock:
            return dict(self._snapshot)

[jacobtomlinson]

The current implementation of kr8s.watch listens for events and yields them as they arrive. This aligns with the kubectl watch behavior which prints events as they arrive.

Just to check my understanding, your proposal does the following:

• Gathers an initial list of all resources that match the selector
• Keeps them up to date as events come in
• Removes them if the event is a deletion
• Allows you to grab the latest version of a resource whenever you like
• Tracks resource versions with bookmark events to allow for better resuming

I agree that this feels separate to kr8s.watch, and goes beyond the functionality of kubectl watch. Therefore it's probably out of scope for kr8s, but I'm happy to be persuaded that we could include it. Maybe we should have our own informers submodule and follow that design more closely? Or maybe a separate package?

We could also improve kr8s.watch to cache bookmarks and if the watch is restarted without a resource version specified we resume from the most recent bookmark. That feels like a separate change to this.