Merge pull request #288 from finos/feat/k8s-perf-and-parity

feat(k8s): status-poll perf, orphan-GC parallelism, contract-mock fix

Author: fgogolli

src/orb/providers/k8s/configuration/config.py

@@ -299,6 +299,22 @@ class K8sProviderConfig(BaseSettings, BaseProviderConfig):  # type: ignore[misc]
         ),
     )
 
+    # Controller-status cache
+    controller_status_cache_ttl_seconds: float = Field(
+        5.0,
+        description=(
+            "How long (in seconds) to serve a cached ``read_namespaced_*`` controller "
+            "response before re-issuing the GET to the API server.  Applied to "
+            "Deployment, StatefulSet and Job status polls.  At the default of 5 s, "
+            "1 000 concurrent requests polling every 5 s produce at most ~200 "
+            "controller GETs/s instead of the unthrottled ~200 GETs/s per workload.  "
+            "Set to 0 (or any value <= 0) to disable the cache entirely — every poll "
+            "will issue a direct GET.  Disabling prevents unbounded in-memory growth "
+            "in environments where the handler is polled at very high frequency; the "
+            "cache dict is not populated at all when TTL <= 0."
+        ),
+    )
+
     # Prometheus metrics
     metrics_enabled: bool = Field(
         True,

src/orb/providers/k8s/infrastructure/handlers/deployment_status.py

@@ -15,6 +15,8 @@
 
 from __future__ import annotations
 
+import threading
+import time
 from typing import TYPE_CHECKING, Any, Optional
 
 from orb.domain.base.provider_fulfilment import CheckHostsStatusResult, ProviderFulfilment
@@ -29,6 +31,11 @@ class DeploymentStatusResolver:
 
     def __init__(self, handler: K8sDeploymentHandler) -> None:
         self._handler = handler
+        # Per-workload TTL cache:
+        #   (namespace, deployment_name) -> (view, fetch_monotonic_ts, stored_requested_count)
+        # Guarded by _cache_lock so concurrent check_hosts_status calls are safe.
+        self._controller_cache: dict[tuple[str, str], tuple[dict[str, Any], float, int]] = {}
+        self._cache_lock = threading.Lock()
 
     def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         """Return per-pod details + the Deployment-driven fulfilment verdict.
@@ -56,7 +63,9 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
             # fulfilment verdict on the Deployment status because the
             # controller's view is authoritative for selective scale.
             cached_instances = handler.apply_pod_timeouts(list(cached.instances))
-            controller_view = self.read_deployment_status(namespace, deployment_name)
+            controller_view = self._get_cached_deployment_status(
+                namespace, deployment_name, requested_count=request.requested_count
+            )
             fulfilment = self.compute_fulfilment(
                 cached_instances,
                 request.requested_count,
@@ -67,10 +76,14 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         selector = handler.build_label_selector(request)
 
         try:
+            # resource_version='0' serves from the apiserver reflector cache
+            # (sub-ms, ~500 ms staleness) instead of reading from etcd.
+            # Acceptable for status polls — not used on create/update paths.
             response = handler.with_retry(
                 handler.client.core_v1.list_namespaced_pod,
                 namespace=namespace,
                 label_selector=selector,
+                resource_version="0",
                 operation_name="list_namespaced_pod",
             )
         except Exception as exc:
@@ -97,14 +110,74 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
             handler._instance_dict_for_pod(pod, namespace=namespace) for pod in pods
         ]
         instances = handler.apply_pod_timeouts(instances)
-        controller_view = self.read_deployment_status(namespace, deployment_name)
+        controller_view = self._get_cached_deployment_status(
+            namespace, deployment_name, requested_count=request.requested_count
+        )
         fulfilment = self.compute_fulfilment(
             instances,
             request.requested_count,
             controller_view=controller_view,
         )
         return CheckHostsStatusResult(instances=instances, fulfilment=fulfilment)
 
+    def _get_cached_deployment_status(
+        self, namespace: str, deployment_name: str, *, requested_count: int
+    ) -> dict[str, Any]:
+        """Return the controller view, serving from the TTL cache when fresh.
+
+        Cache semantics:
+
+        * **TTL disabled** — when ``controller_status_cache_ttl_seconds <= 0``
+          the cache is bypassed entirely (no store, no read) to avoid
+          unbounded dict growth during high-frequency polls.
+        * **Scale-down guard** — a cached entry is only served when the
+          *stored* ``requested_count`` equals the *current* ``requested_count``
+          AND the stored view shows the workload fully ready.  If the replica
+          target has changed since the entry was stored (e.g. scale 5 → 3),
+          the entry is treated as a miss and a fresh GET is issued.
+        * **TOCTOU guard** — the fetch timestamp is captured *before* the
+          blocking GET.  On store, an existing entry is only overwritten when
+          its timestamp is older than the new one, so a slow thread cannot
+          clobber a fresher entry with a stale view.
+
+        The TTL is controlled by
+        ``K8sProviderConfig.controller_status_cache_ttl_seconds`` (default 5 s).
+        """
+        ttl = self._handler.config.controller_status_cache_ttl_seconds
+
+        # TTL <= 0 means disabled — bypass cache entirely (Finding 5).
+        if ttl <= 0:
+            return self.read_deployment_status(namespace, deployment_name)
+
+        cache_key = (namespace, deployment_name)
+        now = time.monotonic()
+
+        with self._cache_lock:
+            entry = self._controller_cache.get(cache_key)
+            if entry is not None:
+                view, ts, stored_requested = entry  # type: ignore[misc]
+                ready = view.get("ready_replicas")
+                # Finding 1: only serve when stored requested_count matches current
+                # AND workload was fully ready when stored.
+                fully_ready = isinstance(ready, int) and ready >= stored_requested
+                if stored_requested == requested_count and fully_ready and now - ts < ttl:
+                    return view
+
+        # Cache miss, expired, workload not fully ready, or requested_count changed —
+        # fetch outside the lock to avoid blocking concurrent callers during the API
+        # GET.  Capture the timestamp BEFORE the GET (Finding 2 / TOCTOU guard).
+        fetch_ts = time.monotonic()
+        view = self.read_deployment_status(namespace, deployment_name)
+
+        with self._cache_lock:
+            existing = self._controller_cache.get(cache_key)
+            # Only overwrite if there is no existing entry or the existing entry
+            # is older than this fetch (Finding 2: don't clobber a newer entry).
+            if existing is None or existing[1] < fetch_ts:  # type: ignore[misc]
+                self._controller_cache[cache_key] = (view, fetch_ts, requested_count)  # type: ignore[assignment]
+
+        return view
+
     def read_deployment_status(self, namespace: str, deployment_name: str) -> dict[str, Any]:
         """Read ``availableReplicas``/``readyReplicas``/``conditions`` from the controller.
 

src/orb/providers/k8s/infrastructure/handlers/job_status.py

@@ -15,6 +15,8 @@
 
 from __future__ import annotations
 
+import threading
+import time
 from typing import TYPE_CHECKING, Any, Optional
 
 from orb.domain.base.provider_fulfilment import CheckHostsStatusResult, ProviderFulfilment
@@ -29,6 +31,11 @@ class JobStatusResolver:
 
     def __init__(self, handler: K8sJobHandler) -> None:
         self._handler = handler
+        # Per-workload TTL cache:
+        #   (namespace, job_name) -> (view, fetch_monotonic_ts)
+        # Guarded by _cache_lock so concurrent check_hosts_status calls are safe.
+        self._controller_cache: dict[tuple[str, str], tuple[dict[str, Any], float]] = {}
+        self._cache_lock = threading.Lock()
 
     def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         """Return per-pod details + the Job-driven fulfilment verdict.
@@ -56,7 +63,7 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
             # controller's view is authoritative for run-to-completion
             # semantics.
             cached_instances = handler.apply_pod_timeouts(list(cached.instances))
-            controller_view = self.read_job_status(namespace, job_name)
+            controller_view = self._get_cached_job_status(namespace, job_name)
             fulfilment = self.compute_fulfilment(
                 cached_instances,
                 request.requested_count,
@@ -67,10 +74,14 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         selector = handler.build_label_selector(request)
 
         try:
+            # resource_version='0' serves from the apiserver reflector cache
+            # (sub-ms, ~500 ms staleness) instead of reading from etcd.
+            # Acceptable for status polls — not used on create/update paths.
             response = handler.with_retry(
                 handler.client.core_v1.list_namespaced_pod,
                 namespace=namespace,
                 label_selector=selector,
+                resource_version="0",
                 operation_name="list_namespaced_pod",
             )
         except Exception as exc:
@@ -97,14 +108,84 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
             handler._instance_dict_for_pod(pod, namespace=namespace) for pod in pods
         ]
         instances = handler.apply_pod_timeouts(instances)
-        controller_view = self.read_job_status(namespace, job_name)
+        controller_view = self._get_cached_job_status(namespace, job_name)
         fulfilment = self.compute_fulfilment(
             instances,
             request.requested_count,
             controller_view=controller_view,
         )
         return CheckHostsStatusResult(instances=instances, fulfilment=fulfilment)
 
+    @staticmethod
+    def _is_terminal(view: dict[str, Any]) -> bool:
+        """Return True when the Job has reached a terminal condition.
+
+        A Job is terminal when it has a ``Complete`` or ``Failed`` condition
+        with ``status=True``.  Both states are equally steady — neither
+        transitions further — so both are safe to cache.
+
+        Caching ``Failed`` prevents re-fetching the Job on every poll after it
+        fails (Finding 3: re-fetch loop / log flood fix).
+        """
+        return any(
+            (c.get("type") in ("Complete", "Failed") and c.get("status") == "True")
+            for c in (view.get("conditions") or [])
+            if isinstance(c, dict)
+        )
+
+    def _get_cached_job_status(self, namespace: str, job_name: str) -> dict[str, Any]:
+        """Return the controller view, serving from the TTL cache when fresh.
+
+        Cache semantics:
+
+        * **TTL disabled** — when ``controller_status_cache_ttl_seconds <= 0``
+          the cache is bypassed entirely (no store, no read) to avoid
+          unbounded dict growth during high-frequency polls (Finding 5).
+        * **Terminal-state gate** — the cached view is only served when the
+          Job has reached a terminal condition (``Complete`` OR ``Failed``).
+          While the job is still running, every poll issues a fresh GET so
+          the active→complete/failed transition is observed promptly.
+          Caching ``Failed`` avoids the re-fetch loop that floods logs when
+          the Job never recovers (Finding 3).
+        * **TOCTOU guard** — the fetch timestamp is captured *before* the
+          blocking GET.  On store, an existing entry is only overwritten when
+          its timestamp is older than the new one, so a slow thread cannot
+          clobber a fresher entry with a stale view (Finding 2).
+
+        The TTL is controlled by
+        ``K8sProviderConfig.controller_status_cache_ttl_seconds`` (default 5 s).
+        """
+        ttl = self._handler.config.controller_status_cache_ttl_seconds
+
+        # TTL <= 0 means disabled — bypass cache entirely (Finding 5).
+        if ttl <= 0:
+            return self.read_job_status(namespace, job_name)
+
+        cache_key = (namespace, job_name)
+        now = time.monotonic()
+
+        with self._cache_lock:
+            entry = self._controller_cache.get(cache_key)
+            if entry is not None:
+                view, ts = entry
+                if self._is_terminal(view) and now - ts < ttl:
+                    return view
+
+        # Cache miss, expired, or job not yet terminal — fetch outside the lock
+        # to avoid blocking concurrent callers during the API GET.  Capture
+        # the timestamp BEFORE the GET (Finding 2 / TOCTOU guard).
+        fetch_ts = time.monotonic()
+        view = self.read_job_status(namespace, job_name)
+
+        with self._cache_lock:
+            existing = self._controller_cache.get(cache_key)
+            # Only overwrite if there is no existing entry or the existing entry
+            # is older than this fetch (Finding 2: don't clobber a newer entry).
+            if existing is None or existing[1] < fetch_ts:
+                self._controller_cache[cache_key] = (view, fetch_ts)
+
+        return view
+
     def read_job_status(self, namespace: str, job_name: str) -> dict[str, Any]:
         """Read controller view from ``batch/v1 Job.status``.
 
@@ -129,6 +210,9 @@ def read_job_status(self, namespace: str, job_name: str) -> dict[str, Any]:
             )
         except Exception as exc:
             if handler.is_not_found(exc):
+                # Job not found is normal after release or before creation.
+                # Log at debug to avoid flooding on the terminal-state poll
+                # path where the Job has already been deleted (Finding 3).
                 handler._logger.debug(
                     "Job %s in %s not found — assuming pre-create or post-release",
                     job_name,

src/orb/providers/k8s/infrastructure/handlers/pod_status.py

@@ -29,7 +29,12 @@ class PodStatusResolver:
     def __init__(self, handler: K8sPodHandler) -> None:
         self._handler = handler
 
-    def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
+    def check_hosts_status(
+        self,
+        request: Request,
+        *,
+        consistent_read: bool = False,
+    ) -> CheckHostsStatusResult:
         """Return per-pod details + the fulfilment verdict for ``request``.
 
         Cache-first read path: when a :class:`PodStateCache` has been
@@ -38,6 +43,18 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         A cache miss (no entry) or a stale cache (any entry older than
         :attr:`K8sProviderConfig.stale_cache_timeout_seconds`)
         falls back to a single ``list_namespaced_pod`` call.
+
+        Parameters
+        ----------
+        request:
+            The request whose pods are being queried.
+        consistent_read:
+            When ``True``, omit ``resource_version='0'`` from the fallback
+            list call, forcing a consistent read from etcd instead of the
+            apiserver reflector cache.  Use on any path that must confirm
+            release completion — the ~500 ms reflector lag can otherwise
+            cause just-deleted pods to appear alive (Finding 4).  Defaults
+            to ``False`` (reflector-cached read) for normal status polls.
         """
         handler = self._handler
         cached = handler._read_from_cache(request)
@@ -49,12 +66,22 @@ def check_hosts_status(self, request: Request) -> CheckHostsStatusResult:
         namespace = handler._resolve_request_namespace(request)
         selector = handler.build_label_selector(request)
 
+        # resource_version='0' serves from the apiserver reflector cache
+        # (sub-ms, ~500 ms staleness) instead of reading from etcd.
+        # Acceptable for status polls — omitted when consistent_read=True
+        # (e.g. release-confirmation paths that need a strong read).
+        list_kwargs: dict[str, Any] = {
+            "namespace": namespace,
+            "label_selector": selector,
+            "operation_name": "list_namespaced_pod",
+        }
+        if not consistent_read:
+            list_kwargs["resource_version"] = "0"
+
         try:
             response = handler.with_retry(
                 handler.client.core_v1.list_namespaced_pod,
-                namespace=namespace,
-                label_selector=selector,
-                operation_name="list_namespaced_pod",
+                **list_kwargs,
             )
         except Exception as exc:
             handler._logger.error(