From 9053789fc33644eb913a1ed5268c01fc1e1ca88e Mon Sep 17 00:00:00 2001 From: Johnny Bouder Date: Mon, 19 Oct 2026 09:13:16 -0400 Subject: [PATCH] fix(api): time out every apiserver call; retry read-modify-write on conflicts - Every Kubernetes client call passes a (5s connect, 30s read) _request_timeout. List calls page with limit=500 and follow continue tokens rather than truncating. - Events are fetched by field selector (the App, app-, each current pod and its ReplicaSet) instead of listing the namespace. - PATCH, stop and start re-read and re-apply once when the write conflicts on resourceVersion; a second conflict returns 409, not 500. Closes #47 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016pvMUQJoLkBDk6B94TTm45 --- api/src/nebari_apps_api/k8s.py | 109 +++++++++++++++++++----- api/src/nebari_apps_api/main.py | 47 +++++++--- api/tests/conftest.py | 5 ++ api/tests/test_k8s_resilience.py | 142 +++++++++++++++++++++++++++++++ api/tests/test_k8s_scoping.py | 10 ++- 5 files changed, 277 insertions(+), 36 deletions(-) create mode 100644 api/tests/test_k8s_resilience.py diff --git a/api/src/nebari_apps_api/k8s.py b/api/src/nebari_apps_api/k8s.py index c6fcfd4..2066b56 100644 --- a/api/src/nebari_apps_api/k8s.py +++ b/api/src/nebari_apps_api/k8s.py @@ -6,6 +6,7 @@ from __future__ import annotations +from collections.abc import Callable from datetime import UTC from typing import TYPE_CHECKING, Any, Protocol, cast @@ -19,6 +20,11 @@ APP_LABEL = "apps.nebari.dev/app" # Seconds a readiness check may wait on a dependency. READY_TIMEOUT = 3 +# (connect, read) seconds for every other apiserver call, so a slow apiserver +# cannot pin the API's worker threads. +REQUEST_TIMEOUT = (5, 30) +# Page size for list calls; results are followed through `continue` tokens. +PAGE_SIZE = 500 class NotFoundError(Exception): @@ -94,37 +100,51 @@ def _wrap(self, exc: Exception) -> Exception: def list_apps(self, namespace: str | None) -> list[dict[str, Any]]: try: if namespace: - return _items(self._custom.list_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL)) + return _list_custom(self._custom.list_namespaced_custom_object, GROUP, VERSION, namespace, PLURAL) if self._namespaces is not None: items: list[dict[str, Any]] = [] for ns in self._namespaces: - items += _items(self._custom.list_namespaced_custom_object(GROUP, VERSION, ns, PLURAL)) + items += _list_custom(self._custom.list_namespaced_custom_object, GROUP, VERSION, ns, PLURAL) return items - return _items(self._custom.list_cluster_custom_object(GROUP, VERSION, PLURAL)) + return _list_custom(self._custom.list_cluster_custom_object, GROUP, VERSION, PLURAL) except Exception as exc: raise self._wrap(exc) from exc def get_app(self, namespace: str, name: str) -> dict[str, Any]: try: - return _obj(self._custom.get_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, name)) + return _obj( + self._custom.get_namespaced_custom_object( + GROUP, VERSION, namespace, PLURAL, name, _request_timeout=REQUEST_TIMEOUT + ) + ) except Exception as exc: raise self._wrap(exc) from exc def create_app(self, namespace: str, body: dict[str, Any]) -> dict[str, Any]: try: - return _obj(self._custom.create_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, body)) + return _obj( + self._custom.create_namespaced_custom_object( + GROUP, VERSION, namespace, PLURAL, body, _request_timeout=REQUEST_TIMEOUT + ) + ) except Exception as exc: raise self._wrap(exc) from exc def replace_app(self, namespace: str, name: str, body: dict[str, Any]) -> dict[str, Any]: try: - return _obj(self._custom.replace_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, name, body)) + return _obj( + self._custom.replace_namespaced_custom_object( + GROUP, VERSION, namespace, PLURAL, name, body, _request_timeout=REQUEST_TIMEOUT + ) + ) except Exception as exc: raise self._wrap(exc) from exc def delete_app(self, namespace: str, name: str) -> None: try: - self._custom.delete_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, name) + self._custom.delete_namespaced_custom_object( + GROUP, VERSION, namespace, PLURAL, name, _request_timeout=REQUEST_TIMEOUT + ) except Exception as exc: raise self._wrap(exc) from exc @@ -133,11 +153,13 @@ def check_ready(self) -> None: self._core.list_namespace(limit=1, _request_timeout=READY_TIMEOUT) def list_managed_namespaces(self) -> list[str]: - namespaces = self._core.list_namespace(label_selector=f"{MANAGED_LABEL}=true") - return sorted(ns.metadata.name for ns in namespaces.items if ns.metadata and ns.metadata.name) + namespaces = _list_typed(self._core.list_namespace, label_selector=f"{MANAGED_LABEL}=true") + return sorted(ns.metadata.name for ns in namespaces if ns.metadata and ns.metadata.name) def pod_logs(self, namespace: str, app_name: str, lines: int, container: str | None) -> str: - pods = self._core.list_namespaced_pod(namespace, label_selector=f"apps.nebari.dev/app={app_name}") + pods = self._core.list_namespaced_pod( + namespace, label_selector=f"{APP_LABEL}={app_name}", limit=1, _request_timeout=REQUEST_TIMEOUT + ) if not pods.items: raise NotFoundError(f"no pods found for app {app_name}") pod_name = pods.items[0].metadata.name if pods.items[0].metadata else None @@ -154,6 +176,7 @@ def pod_logs(self, namespace: str, app_name: str, lines: int, container: str | N container=container, tail_lines=lines, _preload_content=False, + _request_timeout=REQUEST_TIMEOUT, ) except Exception as exc: raise self._wrap(exc) from exc @@ -161,20 +184,31 @@ def pod_logs(self, namespace: str, app_name: str, lines: int, container: str | N return data.decode("utf-8", errors="replace") if isinstance(data, bytes) else str(data) def app_events(self, namespace: str, app_name: str) -> list[dict[str, Any]]: - events = self._core.list_namespaced_event(namespace) + """Events for the App and its children, fetched by field selector. + + A field selector cannot match a name prefix, so the children are named + explicitly: `app-` (Deployment, Service, ConfigMap, NebariApp), + and each current pod and the ReplicaSet that owns it. + """ + selectors = [f"involvedObject.kind=App,involvedObject.name={app_name}"] + names = {f"app-{app_name}"} + pods = _list_typed(self._core.list_namespaced_pod, namespace, label_selector=f"{APP_LABEL}={app_name}") + for pod in pods: + if pod.metadata and pod.metadata.name: + names.add(pod.metadata.name) + names.update(ref.name for ref in pod.metadata.owner_references or [] if ref.kind == "ReplicaSet") + selectors += [f"involvedObject.name={n}" for n in sorted(names)] related: list[dict[str, Any]] = [] - prefix = f"app-{app_name}" - for ev in events.items: - involved = ev.involved_object - name = involved.name or "" - if (involved.kind == "App" and name == app_name) or name.startswith(prefix): + for selector in selectors: + for ev in _list_typed(self._core.list_namespaced_event, namespace, field_selector=selector): + involved = ev.involved_object related.append( { "type": ev.type or "", "reason": ev.reason or "", "message": ev.message or "", "kind": involved.kind or "", - "object": name, + "object": involved.name or "", "count": ev.count or 1, "lastTimestamp": str(ev.last_timestamp or ev.event_time or ""), } @@ -189,7 +223,9 @@ def restart_app(self, namespace: str, app_name: str) -> None: stamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") patch = {"spec": {"template": {"metadata": {"annotations": {"kubectl.kubernetes.io/restartedAt": stamp}}}}} try: - self._apps.patch_namespaced_deployment(f"app-{app_name}", namespace, patch) + self._apps.patch_namespaced_deployment( + f"app-{app_name}", namespace, patch, _request_timeout=REQUEST_TIMEOUT + ) except Exception as exc: raise self._wrap(exc) from exc @@ -200,7 +236,8 @@ def pod_metrics(self, namespace: str, app_name: str) -> list[dict[str, Any]]: "v1beta1", namespace, "pods", - label_selector=f"apps.nebari.dev/app={app_name}", + label_selector=f"{APP_LABEL}={app_name}", + _request_timeout=REQUEST_TIMEOUT, ) except Exception as exc: raise self._wrap(exc) from exc @@ -228,11 +265,13 @@ def cluster_pod_metrics(self) -> list[dict[str, Any]]: report usage as unavailable rather than failing. """ try: - res = self._custom.list_cluster_custom_object("metrics.k8s.io", "v1beta1", "pods", label_selector=APP_LABEL) + items = _list_custom( + self._custom.list_cluster_custom_object, "metrics.k8s.io", "v1beta1", "pods", label_selector=APP_LABEL + ) except Exception as exc: raise self._wrap(exc) from exc out: list[dict[str, Any]] = [] - for item in _items(res): + for item in items: meta = item.get("metadata", {}) app = meta.get("labels", {}).get(APP_LABEL) if not app: @@ -248,9 +287,9 @@ def app_restarts(self) -> list[dict[str, Any]]: if self._namespaces is not None: items: list[V1Pod] = [] for ns in self._namespaces: - items += self._core.list_namespaced_pod(ns, label_selector=APP_LABEL).items + items += _list_typed(self._core.list_namespaced_pod, ns, label_selector=APP_LABEL) else: - items = self._core.list_pod_for_all_namespaces(label_selector=APP_LABEL).items + items = _list_typed(self._core.list_pod_for_all_namespaces, label_selector=APP_LABEL) except Exception as exc: raise self._wrap(exc) from exc out: list[dict[str, Any]] = [] @@ -274,6 +313,30 @@ def _items(res: object) -> list[dict[str, Any]]: return cast("list[dict[str, Any]]", _obj(res).get("items", [])) +def _list_custom(call: Callable[..., object], *args: Any, **kwargs: Any) -> list[dict[str, Any]]: + """Every item of a custom-object list, a page at a time.""" + items: list[dict[str, Any]] = [] + token = "" + while True: + res = _obj(call(*args, limit=PAGE_SIZE, _continue=token or None, _request_timeout=REQUEST_TIMEOUT, **kwargs)) + items += _items(res) + token = (res.get("metadata") or {}).get("continue") or "" + if not token: + return items + + +def _list_typed(call: Callable[..., Any], *args: Any, **kwargs: Any) -> list[Any]: + """Every item of a typed (core/apps) list, a page at a time.""" + items: list[Any] = [] + token = "" + while True: + res = call(*args, limit=PAGE_SIZE, _continue=token or None, _request_timeout=REQUEST_TIMEOUT, **kwargs) + items += res.items + token = (res.metadata and res.metadata._continue) or "" + if not token: + return items + + def _cpu_to_milli(value: str) -> int: """Parse a Kubernetes CPU quantity to integer millicores.""" value = value.strip() diff --git a/api/src/nebari_apps_api/main.py b/api/src/nebari_apps_api/main.py index 57248ce..f0a4bd2 100644 --- a/api/src/nebari_apps_api/main.py +++ b/api/src/nebari_apps_api/main.py @@ -7,6 +7,7 @@ import json import time from collections import Counter +from collections.abc import Callable from typing import Annotated, Any import httpx @@ -114,6 +115,29 @@ def authorized_app(store: AppStore, user: User, namespace: str, name: str) -> di raise HTTPException(403, f"you do not have access to app {namespace}/{name}") return cr + def update_app( + store: AppStore, + user: User, + namespace: str, + name: str, + change: Callable[[dict[str, Any]], dict[str, Any]], + ) -> dict[str, Any]: + """Read-modify-write an App. The write carries the read's resourceVersion, + so a concurrent edit (UI, MCP, kubectl) makes it conflict: re-read, which + also re-checks access, and apply the change once more. A second conflict + is the caller's to retry (409). + """ + for attempt in range(2): + existing = authorized_app(store, user, namespace, name) + try: + return store.replace_app(namespace, name, change(existing)) + except ConflictError: + if attempt: + raise HTTPException( + 409, f"app {namespace}/{name} was changed by someone else; reload and try again" + ) from None + raise AssertionError("unreachable") + def visible_apps(store: AppStore, user: User, namespace: str | None) -> list[dict[str, Any]]: """Apps the caller may see: launchable namespaces only, filtered by owner unless the caller is an admin. @@ -238,12 +262,14 @@ def get_app(namespace: str, name: str, store: Store, user: Me) -> AppOut: def patch_app(namespace: str, name: str, patch: AppPatch, store: Store, user: Me) -> AppOut: _reject_users(patch.access) _check_source(patch.source) - existing = authorized_app(store, user, namespace, name) - current = ((existing.get("spec") or {}).get("access") or {}).get("subdomain") - if patch.access is not None and current and patch.access.subdomain != current: - raise HTTPException(422, "access.subdomain cannot be changed after launch") - updated = store.replace_app(namespace, name, crmod.apply_patch(existing, patch)) - return crmod.from_cr(updated) + + def change(existing: dict[str, Any]) -> dict[str, Any]: + current = ((existing.get("spec") or {}).get("access") or {}).get("subdomain") + if patch.access is not None and current and patch.access.subdomain != current: + raise HTTPException(422, "access.subdomain cannot be changed after launch") + return crmod.apply_patch(existing, patch) + + return crmod.from_cr(update_app(store, user, namespace, name, change)) @app.delete(PREFIX + "/apps/{namespace}/{name}", status_code=204) def delete_app(namespace: str, name: str, store: Store, user: Me) -> None: @@ -251,10 +277,11 @@ def delete_app(namespace: str, name: str, store: Store, user: Me) -> None: store.delete_app(namespace, name) def _scale(store: AppStore, user: User, namespace: str, name: str, replicas: int) -> AppOut: - existing = authorized_app(store, user, namespace, name) - runtime = existing.setdefault("spec", {}).setdefault("runtime", {}) - runtime["replicas"] = replicas - return crmod.from_cr(store.replace_app(namespace, name, existing)) + def change(existing: dict[str, Any]) -> dict[str, Any]: + existing.setdefault("spec", {}).setdefault("runtime", {})["replicas"] = replicas + return existing + + return crmod.from_cr(update_app(store, user, namespace, name, change)) @app.post(PREFIX + "/apps/{namespace}/{name}/stop", response_model=AppOut) def stop_app(namespace: str, name: str, store: Store, user: Me) -> AppOut: diff --git a/api/tests/conftest.py b/api/tests/conftest.py index e184575..f59f91f 100644 --- a/api/tests/conftest.py +++ b/api/tests/conftest.py @@ -18,6 +18,8 @@ def __init__(self) -> None: self.apps: dict[tuple[str, str], dict[str, Any]] = {} self.namespaces = ["apps", "team-a"] self.ready = True + # replace_app raises ConflictError this many times (a concurrent writer). + self.conflicts = 0 self.logs = "hello from pod\n" self.restarted: list[tuple[str, str]] = [] self.metrics_available = True @@ -40,6 +42,9 @@ def create_app(self, namespace: str, body: dict[str, Any]) -> dict[str, Any]: return copy.deepcopy(body) def replace_app(self, namespace: str, name: str, body: dict[str, Any]) -> dict[str, Any]: + if self.conflicts: + self.conflicts -= 1 + raise ConflictError(f"{namespace}/{name} was modified") if (namespace, name) not in self.apps: raise NotFoundError(f"{namespace}/{name}") self.apps[(namespace, name)] = copy.deepcopy(body) diff --git a/api/tests/test_k8s_resilience.py b/api/tests/test_k8s_resilience.py new file mode 100644 index 0000000..6bec73b --- /dev/null +++ b/api/tests/test_k8s_resilience.py @@ -0,0 +1,142 @@ +"""Timeouts and paging on every apiserver call; conflict retry on read-modify-write.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from nebari_apps_api.k8s import PAGE_SIZE, REQUEST_TIMEOUT, KubernetesAppStore + +APP = { + "name": "docs-site", + "namespace": "apps", + "displayName": "Docs Site", + "source": {"type": "inline", "inline": {"files": {"index.html": "

hi

"}}}, + "access": {"public": True, "subdomain": "docs-site"}, +} + + +def page(items, token=None): + return SimpleNamespace(items=items, metadata=SimpleNamespace(_continue=token)) + + +def pod(name, rs): + return SimpleNamespace( + metadata=SimpleNamespace( + name=name, + namespace="apps", + labels={"apps.nebari.dev/app": "demo"}, + owner_references=[SimpleNamespace(kind="ReplicaSet", name=rs)], + ), + status=SimpleNamespace(container_statuses=[SimpleNamespace(restart_count=0)]), + ) + + +def event(kind, name, reason): + return SimpleNamespace( + involved_object=SimpleNamespace(kind=kind, name=name), + type="Normal", + reason=reason, + message="", + count=1, + last_timestamp=reason, + event_time=None, + ) + + +@pytest.fixture +def kube(): + store = KubernetesAppStore.__new__(KubernetesAppStore) + store._namespaces = None + store._custom = MagicMock() + store._core = MagicMock() + store._apps = MagicMock() + store._custom.list_namespaced_custom_object.return_value = {"items": [], "metadata": {}} + store._custom.list_cluster_custom_object.return_value = {"items": [], "metadata": {}} + store._custom.get_namespaced_custom_object.return_value = {} + store._custom.create_namespaced_custom_object.return_value = {} + store._custom.replace_namespaced_custom_object.return_value = {} + store._core.list_namespace.return_value = page([]) + store._core.list_namespaced_pod.return_value = page([pod("app-demo-abc-1", "app-demo-abc")]) + store._core.list_pod_for_all_namespaces.return_value = page([]) + store._core.list_namespaced_event.return_value = page([]) + store._core.read_namespaced_pod_log.return_value = SimpleNamespace(data=b"log") + return store + + +def test_every_call_has_a_timeout(kube): + kube.list_apps(None) + kube.list_apps("apps") + kube.get_app("apps", "demo") + kube.create_app("apps", {}) + kube.replace_app("apps", "demo", {}) + kube.delete_app("apps", "demo") + kube.list_managed_namespaces() + kube.pod_logs("apps", "demo", 10, None) + kube.app_events("apps", "demo") + kube.restart_app("apps", "demo") + kube.pod_metrics("apps", "demo") + kube.cluster_pod_metrics() + kube.app_restarts() + calls = kube._custom.mock_calls + kube._core.mock_calls + kube._apps.mock_calls + assert len(calls) >= 13 + for call in calls: + assert call.kwargs.get("_request_timeout") == REQUEST_TIMEOUT, call + + +def test_lists_page_through_continue_tokens(kube): + kube._custom.list_cluster_custom_object.side_effect = [ + {"items": [{"metadata": {"name": "a"}}], "metadata": {"continue": "t1"}}, + {"items": [{"metadata": {"name": "b"}}], "metadata": {}}, + ] + assert [a["metadata"]["name"] for a in kube.list_apps(None)] == ["a", "b"] + first, second = kube._custom.list_cluster_custom_object.call_args_list + assert first.kwargs["limit"] == PAGE_SIZE and first.kwargs["_continue"] is None + assert second.kwargs["_continue"] == "t1" + + +def test_events_are_field_selected(kube): + by_selector = { + "involvedObject.kind=App,involvedObject.name=demo": [event("App", "demo", "1-validated")], + "involvedObject.name=app-demo": [event("Deployment", "app-demo", "2-scaled")], + "involvedObject.name=app-demo-abc": [event("ReplicaSet", "app-demo-abc", "3-created")], + "involvedObject.name=app-demo-abc-1": [event("Pod", "app-demo-abc-1", "4-pulled")], + } + kube._core.list_namespaced_event.side_effect = lambda ns, field_selector, **kw: page(by_selector[field_selector]) + events = kube.app_events("apps", "demo") + assert [e["reason"] for e in events] == ["4-pulled", "3-created", "2-scaled", "1-validated"] + for call in kube._core.list_namespaced_event.call_args_list: + assert call.kwargs["field_selector"] in by_selector + + +@pytest.mark.parametrize( + ("method", "path", "body"), + [ + ("PATCH", "/api/v1/apps/apps/docs-site", {"displayName": "Renamed"}), + ("POST", "/api/v1/apps/apps/docs-site/stop", None), + ("POST", "/api/v1/apps/apps/docs-site/start", None), + ], +) +def test_one_conflict_is_retried(client, store, method, path, body): + assert client.post("/api/v1/apps", json=APP).status_code == 201 + store.conflicts = 1 + resp = client.request(method, path, json=body) + assert resp.status_code == 200, resp.text + assert store.conflicts == 0 + + +@pytest.mark.parametrize( + ("method", "path", "body"), + [ + ("PATCH", "/api/v1/apps/apps/docs-site", {"displayName": "Renamed"}), + ("POST", "/api/v1/apps/apps/docs-site/stop", None), + ], +) +def test_second_conflict_is_409(client, store, method, path, body): + assert client.post("/api/v1/apps", json=APP).status_code == 201 + store.conflicts = 2 + resp = client.request(method, path, json=body) + assert resp.status_code == 409 + assert "changed by someone else" in resp.json()["detail"] diff --git a/api/tests/test_k8s_scoping.py b/api/tests/test_k8s_scoping.py index 40965e8..3ccd97e 100644 --- a/api/tests/test_k8s_scoping.py +++ b/api/tests/test_k8s_scoping.py @@ -10,19 +10,23 @@ from nebari_apps_api.k8s import KubernetesAppStore +def page(items): + return SimpleNamespace(items=items, metadata=SimpleNamespace(_continue=None)) + + def scoped_store(namespaces): store = KubernetesAppStore.__new__(KubernetesAppStore) store._namespaces = namespaces store._custom = MagicMock() store._core = MagicMock() - store._custom.list_namespaced_custom_object.side_effect = lambda g, v, ns, p: { + store._custom.list_namespaced_custom_object.side_effect = lambda g, v, ns, p, **kw: { "items": [{"metadata": {"name": f"a-{ns}", "namespace": ns}}] } pod = lambda ns: SimpleNamespace( # noqa: E731 metadata=SimpleNamespace(namespace=ns, labels={"apps.nebari.dev/app": "x"}), status=SimpleNamespace(container_statuses=[SimpleNamespace(restart_count=2)]), ) - store._core.list_namespaced_pod.side_effect = lambda ns, label_selector: SimpleNamespace(items=[pod(ns)]) + store._core.list_namespaced_pod.side_effect = lambda ns, **kw: page([pod(ns)]) return store @@ -42,7 +46,7 @@ def test_scoped_restarts_iterate_namespaces(): def test_unscoped_uses_cluster_wide_calls(): store = scoped_store(None) store._custom.list_cluster_custom_object.return_value = {"items": []} - store._core.list_pod_for_all_namespaces.return_value = SimpleNamespace(items=[]) + store._core.list_pod_for_all_namespaces.return_value = page([]) store.list_apps(None) store.app_restarts() store._custom.list_cluster_custom_object.assert_called_once()