fix(api): time out every apiserver call; retry read-modify-write on conflicts

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,27 +176,39 @@ 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
         data = resp.data
         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-<name>` (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()

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,23 +262,26 @@ 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:
         authorized_app(store, user, namespace, name)
         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:

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)