diff --git a/kubespawner/reflector.py b/kubespawner/reflector.py index 2249fd79..a6b1334f 100644 --- a/kubespawner/reflector.py +++ b/kubespawner/reflector.py @@ -208,6 +208,12 @@ def __init__(self, *args, **kwargs): self.watch_task = None + # Generation counter for stale-entry detection. + # Incremented on each watch reconnect so callers can tell whether a + # cached entry was seen in the current watch cycle or a previous one. + self._generation = 0 + self._entry_generations = {} # ref_key → generation when last seen + async def _list_and_update(self, resource_version=None): """ Update current list of resources by doing a full fetch. @@ -307,6 +313,7 @@ async def _watch_and_update(self): try: resource_version = await self._list_and_update(resource_version) cur_delay = 0.1 + self._generation += 1 # new watch cycle; old entries are now stale watch_args = { "label_selector": self.label_selector, "field_selector": self.field_selector, @@ -346,9 +353,17 @@ async def _watch_and_update(self): if watch_event['type'] == 'DELETED': # This is an atomic delete operation on the dictionary! self.resources.pop(ref_key, None) + self._entry_generations.pop(ref_key, None) else: # This is an atomic operation on the dictionary! self.resources[ref_key] = resource + # _entry_generations is updated on the next line after + # resources; there is a brief window of inconsistency + # between the two dicts. is_stale() is advisory so + # this is safe: a caller that races here will either + # see the old generation (slightly conservative) or + # the new one (correct), never a crash. + self._entry_generations[ref_key] = self._generation resource_version = resource["metadata"]["resourceVersion"] if self._stopping: self.log.info("%s watcher stopped: inner", self.kind) @@ -437,6 +452,20 @@ async def stop(self): ) self.watch_task = None + def is_stale(self, ref_key: str) -> bool: + """Return True if the cached entry was not seen in the current watch cycle. + + After a watch reconnects, entries that existed before the reconnect + but have not yet been confirmed by a new watch event carry a previous + generation number. Callers (e.g. ``poll()``) can use this to decide + whether to re-query the API rather than trust a potentially stale value. + + Returns False if the key is not in the cache (absent is not stale). + """ + if ref_key not in self.resources: + return False + return self._entry_generations.get(ref_key, 0) < self._generation + class NamespacedResourceReflector(ResourceReflector): """ diff --git a/tests/test_reflector.py b/tests/test_reflector.py new file mode 100644 index 00000000..93ec7f49 --- /dev/null +++ b/tests/test_reflector.py @@ -0,0 +1,181 @@ +""" +Unit tests for ResourceReflector — generation tracking and is_stale() (PR #921). + +These tests do not require a running Kubernetes cluster; the k8s client is +mocked so the cache logic can be exercised in isolation. +""" + +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from kubespawner.reflector import ResourceReflector + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_reflector(**kwargs): + """Return a ResourceReflector with a mocked k8s client.""" + defaults = dict(kind="pods", namespace="test-ns") + defaults.update(kwargs) + with patch("kubespawner.reflector.shared_client", return_value=Mock()): + return ResourceReflector(**defaults) + + +@asynccontextmanager +async def stream_yielding(events): + """Async context manager that yields *events* then exits cleanly.""" + + async def _gen(): + for e in events: + yield e + + yield _gen() + + +# --------------------------------------------------------------------------- +# Generation tracking — construction +# --------------------------------------------------------------------------- + + +def test_generation_starts_at_zero(): + """_generation is initialised to 0 at construction.""" + r = make_reflector() + assert r._generation == 0 + + +def test_entry_generations_starts_empty(): + """_entry_generations is empty at construction.""" + r = make_reflector() + assert r._entry_generations == {} + + +# --------------------------------------------------------------------------- +# is_stale() — pure logic +# --------------------------------------------------------------------------- + + +def test_is_stale_missing_key_returns_false(): + """is_stale() returns False for a key not in the cache (absent ≠ stale).""" + r = make_reflector() + assert not r.is_stale("test-ns/nonexistent") + + +def test_is_stale_current_generation_returns_false(): + """Entry stamped with the current generation is not stale.""" + r = make_reflector() + r._generation = 5 + r.resources["test-ns/pod-a"] = {} + r._entry_generations["test-ns/pod-a"] = 5 + assert not r.is_stale("test-ns/pod-a") + + +def test_is_stale_old_generation_returns_true(): + """Entry from a previous watch cycle is detected as stale.""" + r = make_reflector() + r._generation = 5 + r.resources["test-ns/pod-a"] = {} + r._entry_generations["test-ns/pod-a"] = 3 # seen two cycles ago + assert r.is_stale("test-ns/pod-a") + + +def test_is_stale_no_stamp_after_generation_advance(): + """Entry with no generation stamp is stale once _generation advances.""" + r = make_reflector() + r._generation = 1 + r.resources["test-ns/pod-a"] = {} + # _entry_generations not set → .get() returns 0 < _generation=1 + assert r.is_stale("test-ns/pod-a") + + +# --------------------------------------------------------------------------- +# Generation tracking — watch cycle integration +# --------------------------------------------------------------------------- + + +async def test_generation_increments_on_watch_reconnect(): + """_generation increments each time the watch successfully reconnects.""" + r = make_reflector() + r._list_and_update = AsyncMock(return_value="1") + + class QuickWatch: + def stream(self, *a, **kw): + return stream_yielding([]) # exits immediately + + def stop(self): + pass + + async def close(self): + pass + + import asyncio + + with patch("kubespawner.reflector.watch.Watch", QuickWatch): + with patch("asyncio.sleep", new_callable=AsyncMock): + task = asyncio.create_task(r._watch_and_update()) + # Wait until at least two reconnects have occurred. + for _ in range(50): + await asyncio.sleep(0) + if r._generation >= 2: + break + r._stopping = True + try: + await asyncio.wait_for(task, timeout=2) + except (asyncio.TimeoutError, asyncio.CancelledError): + task.cancel() + + assert r._generation >= 2, "Expected _generation to increment across reconnects" + + +async def test_watch_event_stamps_entry_generation(): + """ADDED events stamp _entry_generations with the current _generation.""" + r = make_reflector() + r._generation = 3 + + ref_key = "test-ns/pod-x" + # Simulate what _watch_and_update does on an ADDED event. + r.resources[ref_key] = {} + r._entry_generations[ref_key] = r._generation + + assert r._entry_generations[ref_key] == 3 + assert not r.is_stale(ref_key) + + +async def test_deleted_event_clears_entry_generation(): + """DELETED events remove the entry from _entry_generations.""" + r = make_reflector() + r._generation = 2 + ref_key = "test-ns/pod-gone" + r.resources[ref_key] = {} + r._entry_generations[ref_key] = 2 + + # Simulate DELETED event processing. + r.resources.pop(ref_key, None) + r._entry_generations.pop(ref_key, None) + + assert ref_key not in r._entry_generations + assert not r.is_stale(ref_key) # absent is not stale + + +async def test_entry_becomes_stale_after_reconnect(): + """Entry not refreshed after a reconnect is detected as stale.""" + r = make_reflector() + ref_key = "test-ns/pod-old" + + # Populate during generation 1. + r._generation = 1 + r.resources[ref_key] = {} + r._entry_generations[ref_key] = 1 + + # Simulate a watch reconnect (generation advances to 2). + r._generation = 2 + + # The entry has not been updated yet — it should be stale. + assert r.is_stale(ref_key) + + # Once the watch delivers a new event for this pod, it is fresh again. + r._entry_generations[ref_key] = r._generation + assert not r.is_stale(ref_key)