diff --git a/pyproject.toml b/pyproject.toml index 018c2b3..0e88023 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "slo-api" -version = "0.2.0" +version = "0.2.1" description = "HTTP facade for LLMSLORequirement CRs — exposes route-keyed REST API, translates to/from CRD specs" requires-python = ">=3.10" readme = "README.md" diff --git a/src/slo_api/cr_index.py b/src/slo_api/cr_index.py index 49d803e..2354621 100644 --- a/src/slo_api/cr_index.py +++ b/src/slo_api/cr_index.py @@ -118,9 +118,20 @@ def _run(self): def _watch_once(self): api = self._api_client() + # List first, then watch from the list's resourceVersion. A bare watch only marks the index + # synced on its first event, and with no CRs in the cluster there is none: the API server + # sends no ADDED event and its first BOOKMARK can take minutes, so readiness stayed 503 and + # the liveness probe killed the pod before it ever served. An empty list is a complete one. + listing = api.list_cluster_custom_object(group(), VERSION, PLURAL) + self._replace((listing or {}).get("items") or []) + self._synced.set() + rv = ((listing or {}).get("metadata") or {}).get("resourceVersion") + wf = self._watch_factory or watch.Watch w = wf() kwargs = {} if self._watch_factory else {"allow_watch_bookmarks": True} + if rv and not self._watch_factory: + kwargs["resource_version"] = rv try: for event in w.stream( api.list_cluster_custom_object, @@ -142,6 +153,25 @@ def _watch_once(self): if callable(stop): stop() + def _replace(self, items): + """Rebuild the index from a full list: drops CRs deleted while no watch was running. + + Built aside and swapped in under the lock, so a reader sees the old index or the new one, + never the empty or half-built one in between -- this runs on every watch reconnect.""" + by_key, by_service = {}, {} + for cr in items: + meta = cr.get("metadata") or {} + ns, name = meta.get("namespace"), meta.get("name") + if not ns or not name: + continue + sid = (cr.get("spec") or {}).get("serviceId") or "" + by_key[(ns, name)] = cr + lst = by_service.setdefault(sid, []) + if (ns, name) not in lst: + lst.append((ns, name)) + with self._lock: + self._by_key, self._by_service = by_key, by_service + def _apply(self, etype, cr): meta = cr.get("metadata") or {} spec = cr.get("spec") or {} diff --git a/tests/test_cr_index_sync.py b/tests/test_cr_index_sync.py new file mode 100644 index 0000000..d9240b2 --- /dev/null +++ b/tests/test_cr_index_sync.py @@ -0,0 +1,70 @@ +import threading + +from slo_api.cr_index import CRIndex + +from conftest import CRD_TEMPLATE, FakeCustomObjectsApi + + +class SilentWatch: + """A watch the way a real API server answers one on an empty cluster: no ADDED events and no + BOOKMARK for a long while. (FakeWatch in conftest always sends a BOOKMARK, which is why the + tests never saw an index that could not get synced.)""" + + def __init__(self, stop): + self._stop = stop + + def stream(self, fn, *args, **kwargs): + while not self._stop.is_set(): + self._stop.wait(0.01) + return + yield # pragma: no cover -- makes this a generator + + def stop(self): + pass + + +def _index(api): + idx = CRIndex(api=api, watch_factory=lambda: SilentWatch(idx._stop)) + return idx + + +def test_empty_cluster_is_synced(): + idx = _index(FakeCustomObjectsApi()) + idx.start() + try: + assert idx.wait_synced(timeout=2), "an empty list is complete: the index must report synced" + assert idx.list_routes() == [] + finally: + idx.stop() + + +def test_list_fills_the_index_before_any_watch_event(): + api = FakeCustomObjectsApi() + api.seed(CRD_TEMPLATE("ns1", "a", "route-a")) + api.seed(CRD_TEMPLATE("ns2", "b", "route-b")) + idx = _index(api) + idx.start() + try: + assert idx.wait_synced(timeout=2) + assert idx.list_routes() == ["route-a", "route-b"] + finally: + idx.stop() + + +def test_relist_never_shows_an_empty_index(): + """A watch reconnect re-lists and rebuilds the index. Readers in between must keep seeing the + old index, not a cleared or half-built one.""" + api = FakeCustomObjectsApi() + old = [CRD_TEMPLATE("ns1", "a", "route-a"), CRD_TEMPLATE("ns2", "b", "route-b")] + idx = _index(api) + idx._replace(old) + seen = [] + + def relisted(): + for cr in old + [CRD_TEMPLATE("ns3", "c", "route-c")]: + seen.append(idx.list_routes()) # what a reader sees while the rebuild is under way + yield cr + + idx._replace(relisted()) + assert all(s == ["route-a", "route-b"] for s in seen), seen + assert idx.list_routes() == ["route-a", "route-b", "route-c"]