From 456edeb588a4ee936483d6d1b6c6286b2a09584b Mon Sep 17 00:00:00 2001 From: Rohit Sharma Date: Sat, 26 Sep 2026 17:47:05 +0100 Subject: [PATCH 1/7] Add a Kubernetes container provider. KubernetesProvider was a placeholder that could not be instantiated, while the runtime docs still described it as planned. A trainer running in the same cluster needs the same ContainerProvider contract as local Docker: one environment, one Pod and one Service, and an in-cluster HTTP URL. The kubernetes client is an optional extra, imported only when the provider is constructed. Readiness polls /health, as LocalDockerProvider does. The SDK sits behind a private adapter so tests can mock it, as ModalProvider does. A failed start deletes only the objects this instance created, and stop and close are safe to call twice. The Service URL and raw API bodies are left out of errors. --- README.md | 2 +- docs/source/getting-started.md | 4 +- docs/source/guides/runtime-providers.md | 19 +- docs/source/reference/core.md | 2 + pyproject.toml | 3 + src/openenv/core/README.md | 6 +- .../core/containers/runtime/__init__.py | 2 - .../containers/runtime/kubernetes_provider.py | 776 ++++++++++++++++++ .../core/containers/runtime/providers.py | 17 +- tests/test_core/test_kubernetes_provider.py | 649 +++++++++++++++ 10 files changed, 1455 insertions(+), 25 deletions(-) create mode 100644 src/openenv/core/containers/runtime/kubernetes_provider.py create mode 100644 tests/test_core/test_kubernetes_provider.py diff --git a/README.md b/README.md index 5b12ea0203..06b3526c9f 100644 --- a/README.md +++ b/README.md @@ -176,7 +176,7 @@ Manage container deployment: - `LocalDockerProvider`: Run containers on local Docker daemon - `DockerSwarmProvider`: Deploy to Docker Swarm clusters - `UVProvider`, `DaytonaProvider`, `ACASandboxProvider`: Additional runtime providers -- `KubernetesProvider`: Deploy to Kubernetes clusters (planned) +- `KubernetesProvider`: Deploy to Kubernetes clusters (`pip install openenv[kubernetes]`) #### 5. Models Type-safe data structures: diff --git a/docs/source/getting-started.md b/docs/source/getting-started.md index f40fe8a444..711d29cbcc 100644 --- a/docs/source/getting-started.md +++ b/docs/source/getting-started.md @@ -163,8 +163,8 @@ runtime can implement it without changing the client/server protocol. Providers shipped today: `LocalDockerProvider`, `DockerSwarmProvider`, `UVProvider`, `DaytonaProvider`, `ACASandboxProvider` (Azure Container Apps -Sandboxes), `ModalProvider`, and `NovitaSandboxProvider`. A `KubernetesProvider` -is planned. +Sandboxes), `ModalProvider`, `NovitaSandboxProvider`, and `KubernetesProvider` +(`pip install openenv[kubernetes]`). See the [Runtime Providers guide](guides/runtime-providers.md) for the full list, install extras, and how to select a provider, and the diff --git a/docs/source/guides/runtime-providers.md b/docs/source/guides/runtime-providers.md index ff061c3ed7..cbd28612ee 100644 --- a/docs/source/guides/runtime-providers.md +++ b/docs/source/guides/runtime-providers.md @@ -16,7 +16,7 @@ is a one-line change. | `ACASandboxProvider` | Azure Container Apps Sandboxes | `pip install openenv[aca]` | ✅ | | `ModalProvider` | Modal sandboxes | `pip install openenv[modal]` | ✅ | | `NovitaSandboxProvider` | Novita AI sandboxes | `pip install openenv[novita]` | ✅ | -| `KubernetesProvider` | Kubernetes cluster | core | 🚧 planned | +| `KubernetesProvider` | Kubernetes cluster | `pip install openenv[kubernetes]` | ✅ | Cloud-provider SDKs are optional extras, imported lazily, so installing core OpenEnv pulls in no cloud SDK. The core providers (`LocalDockerProvider`, @@ -162,8 +162,21 @@ provider = DockerSwarmProvider() ### KubernetesProvider -🚧 Not yet implemented. The class exists as a placeholder for the planned -Kubernetes backend. +Runs one environment as one Pod and one Service in an existing namespace. +Install with `pip install openenv[kubernetes]`. The returned URL is the +in-cluster Service DNS name over `http://`, which is the connectivity choice +for this provider. `ModalProvider` and `ACASandboxProvider` still require +`https://`. Use this provider from a trainer in the same cluster; it does not +create a namespace, Deployment, Ingress, or port-forward. + +```python +from openenv.core.containers.runtime.kubernetes_provider import KubernetesProvider + +provider = KubernetesProvider(namespace="openenv") +base_url = provider.start_container("echo-env:latest") +provider.wait_for_ready(base_url) +provider.stop_container() +``` ### LocalDockerProvider diff --git a/docs/source/reference/core.md b/docs/source/reference/core.md index dd92a51a57..471ae6e6b2 100644 --- a/docs/source/reference/core.md +++ b/docs/source/reference/core.md @@ -255,3 +255,5 @@ For dataset-backed environments that publish enumerable tasks and splits, see th [[autodoc]] openenv.core.containers.runtime.modal_provider.ModalProvider [[autodoc]] openenv.core.containers.runtime.novita_provider.NovitaSandboxProvider + +[[autodoc]] openenv.core.containers.runtime.kubernetes_provider.KubernetesProvider diff --git a/pyproject.toml b/pyproject.toml index d69d73be42..46673c4ac1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,9 @@ modal = [ "modal>=1.4.0", "pyyaml>=6.0", ] +kubernetes = [ + "kubernetes>=29.0.0", +] novita = [ "novita-sandbox>=2.1.1", ] diff --git a/src/openenv/core/README.md b/src/openenv/core/README.md index 80173590ed..3116feecd5 100644 --- a/src/openenv/core/README.md +++ b/src/openenv/core/README.md @@ -174,14 +174,14 @@ provider.wait_for_ready(base_url) provider.stop_container() ``` -### Kubernetes Provider (Coming Soon) +### Kubernetes Provider ```python -from openenv.core.containers.runtime import KubernetesProvider +from openenv.core.containers.runtime.kubernetes_provider import KubernetesProvider provider = KubernetesProvider(namespace="envs") base_url = provider.start_container("my-env:latest") -# Use environment... +provider.wait_for_ready(base_url) provider.stop_container() ``` diff --git a/src/openenv/core/containers/runtime/__init__.py b/src/openenv/core/containers/runtime/__init__.py index 9a3ce76225..16caec4c81 100644 --- a/src/openenv/core/containers/runtime/__init__.py +++ b/src/openenv/core/containers/runtime/__init__.py @@ -5,7 +5,6 @@ from .providers import ( ContainerProvider, DockerSwarmProvider, - KubernetesProvider, LocalDockerProvider, RuntimeProvider, ) @@ -20,7 +19,6 @@ "ContainerProvider", "DockerSwarmProvider", "LocalDockerProvider", - "KubernetesProvider", "RuntimeProvider", "UVProvider", ] diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py new file mode 100644 index 0000000000..e2390a09f4 --- /dev/null +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -0,0 +1,776 @@ +# SPDX-License-Identifier: BSD-3-Clause + +""" +Kubernetes container provider for running OpenEnv environments in-cluster. + +Requires the ``kubernetes`` client: ``pip install openenv[kubernetes]`` + +One provider instance owns one Pod and one Service, matching +``LocalDockerProvider`` owning one container. The returned ``base_url`` is the +in-cluster Service DNS name over ``http://``. That plaintext URL is the +in-cluster exception confirmed for this provider: ``ModalProvider`` and +``ACASandboxProvider`` still refuse non-HTTPS tunnel URLs. This provider never +returns a URL it did not construct from a validated name, namespace, and port, +and it does not copy that URL or raw API bodies into errors. +""" + +from __future__ import annotations + +import re +import secrets +import time +from typing import Any, Mapping, Optional + +from .providers import ContainerProvider + +_CONTAINER_PORT = 8000 +_CONTAINER_NAME = "env" +_DEFAULT_NAMESPACE = "default" +_NAME_PREFIX = "openenv-" + +# ``openenv-`` + slug + ``-`` + millisecond timestamp + ``-`` + 4 hex chars. +_SLUG_MAX = 32 + +_DNS_LABEL = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$") +_LABEL_NAME = re.compile(r"^[A-Za-z0-9]([A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?$") +_LABEL_VALUE = re.compile(r"^([A-Za-z0-9]([A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?)?$") +_SAFE_REASON = re.compile(r"^[A-Za-z][A-Za-z0-9]{0,63}$") + +# Applied last so a caller-supplied label cannot drop ownership or retarget +# the Service selector. +_SELECTOR_KEY = "app.kubernetes.io/instance" + +_START_KWARGS = { + "resources", + "runtime_class_name", + "image_pull_secrets", + "labels", + "service_account_name", + "automount_service_account_token", +} + + +def _raise_install_error(exc: ImportError) -> None: + raise RuntimeError( + "KubernetesProvider requires the kubernetes client. " + "Install it with `pip install openenv[kubernetes]`." + ) from exc + + +def _dns_label(value: str, *, max_length: int, what: str) -> str: + """Return *value* as a DNS-1123 label, or raise ``ValueError``. + + Namespace and runtime class names are validated, not rewritten: rewriting + them would target a different object than the one the caller named. + """ + if not isinstance(value, str) or not _DNS_LABEL.fullmatch(value): + raise ValueError( + f"{what} must be a DNS-1123 label: lowercase alphanumeric " + "characters or '-', starting and ending with an alphanumeric." + ) + if not 1 <= len(value) <= max_length: + raise ValueError(f"{what} must be 1 to {max_length} characters.") + return value + + +def _slug(image: str) -> str: + """DNS-1123 slug from an image reference. Raises if nothing usable remains.""" + raw = image.split("/")[-1].split("@")[0].split(":")[0] + cleaned = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-") + cleaned = re.sub(r"-{2,}", "-", cleaned)[:_SLUG_MAX].strip("-") + if not cleaned or not _DNS_LABEL.fullmatch(cleaned): + raise ValueError( + "image does not yield a valid Kubernetes name. " + "Use an image reference with an alphanumeric name." + ) + return cleaned + + +def _resource_name(image: str) -> str: + suffix = f"{int(time.time() * 1000)}-{secrets.token_hex(2)}" + name = f"{_NAME_PREFIX}{_slug(image)}-{suffix}" + if len(name) > 63 or not _DNS_LABEL.fullmatch(name): + raise ValueError("generated Kubernetes name is not a valid DNS-1123 label.") + return name + + +def _label_key(key: str) -> str: + if not isinstance(key, str) or len(key) > 253: + raise ValueError( + "Kubernetes label keys must be strings of at most 253 characters." + ) + prefix, sep, name = key.rpartition("/") + if not _LABEL_NAME.fullmatch(name): + raise ValueError(f"Invalid Kubernetes label key {key!r}.") + if sep and (len(prefix) > 253 or not _dns_subdomain(prefix)): + raise ValueError(f"Invalid Kubernetes label key prefix {prefix!r}.") + return key + + +def _dns_subdomain(value: str) -> bool: + if not value or len(value) > 253: + return False + return all(_DNS_LABEL.fullmatch(part) for part in value.split(".")) + + +def _label_value(value: str) -> str: + if not isinstance(value, str) or not _LABEL_VALUE.fullmatch(value): + raise ValueError( + "Kubernetes label values must be 63 characters or fewer and contain " + "only alphanumeric characters, '-', '_' or '.'." + ) + return value + + +def _ownership_labels(name: str) -> dict[str, str]: + return { + "app.kubernetes.io/name": "openenv", + "app.kubernetes.io/instance": name, + "app.kubernetes.io/managed-by": "openenv", + "app.kubernetes.io/part-of": "openenv", + "app.kubernetes.io/component": "environment", + } + + +def _merged_labels(name: str, extra: Optional[Mapping[str, str]]) -> dict[str, str]: + labels: dict[str, str] = {} + if extra: + if not isinstance(extra, Mapping): + raise ValueError("labels must be a mapping of strings.") + for key, value in extra.items(): + labels[_label_key(key)] = _label_value(value) + labels.update(_ownership_labels(name)) + return labels + + +def _safe_reason(reason: Optional[str]) -> Optional[str]: + """Keep kubelet reason codes. Drop anything else so API text cannot leak.""" + if isinstance(reason, str) and _SAFE_REASON.fullmatch(reason): + return reason + return None + + +def _incluster_base_url(name: str, namespace: str, port: int) -> str: + # *name* and *namespace* are already DNS-1123 labels and *port* is an int, + # so this cannot be steered by a status field from the API server. + return f"http://{name}.{namespace}.svc.cluster.local:{port}" + + +class _DefaultKubernetesAdapter: + """Thin adapter over the official ``kubernetes`` client. + + The provider talks to this private adapter instead of spreading SDK details + through its own logic; tests inject a duck-typed fake in its place. + """ + + def __init__( + self, + *, + namespace: str, + kubeconfig: Optional[str], + context: Optional[str], + ): + try: + from kubernetes import client, config + from kubernetes.client.exceptions import ApiException + except ImportError as exc: + _raise_install_error(exc) + + self._client = client + self._ApiException = ApiException + self._namespace = namespace + self._load_config(config, kubeconfig=kubeconfig, context=context) + self._api = client.CoreV1Api() + + def _load_config( + self, config: Any, *, kubeconfig: Optional[str], context: Optional[str] + ) -> None: + if kubeconfig is not None or context is not None: + config.load_kube_config(config_file=kubeconfig, context=context) + return + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + + def _api_error(self, exc: Exception, action: str, *, missing_ok: bool) -> None: + status = getattr(exc, "status", None) + if missing_ok and status == 404: + return + if status == 409: + raise RuntimeError(f"Kubernetes {action} name is already in use") from None + # The API body can echo object contents. Do not chain it into the error. + raise RuntimeError(f"Failed to {action} Kubernetes resource") from None + + def ensure_namespace(self) -> None: + try: + self._api.read_namespace(self._namespace) + except self._ApiException as exc: + if getattr(exc, "status", None) == 404: + raise RuntimeError( + "Kubernetes namespace does not exist. " + "KubernetesProvider does not create namespaces." + ) from None + raise RuntimeError("Failed to read Kubernetes namespace") from None + + def create_pod(self, manifest: Mapping[str, Any]) -> None: + try: + self._api.create_namespaced_pod(self._namespace, self._pod(manifest)) + except self._ApiException as exc: + self._api_error(exc, "create Pod", missing_ok=False) + + def create_service(self, manifest: Mapping[str, Any]) -> None: + try: + self._api.create_namespaced_service( + self._namespace, self._service(manifest) + ) + except self._ApiException as exc: + self._api_error(exc, "create Service", missing_ok=False) + + def delete_pod(self, name: str) -> None: + try: + self._api.delete_namespaced_pod(name, self._namespace) + except self._ApiException as exc: + self._api_error(exc, "delete Pod", missing_ok=True) + + def delete_service(self, name: str) -> None: + try: + self._api.delete_namespaced_service(name, self._namespace) + except self._ApiException as exc: + self._api_error(exc, "delete Service", missing_ok=True) + + def pod_status(self, name: str) -> dict[str, Optional[str]]: + try: + pod = self._api.read_namespaced_pod(name, self._namespace) + except self._ApiException as exc: + self._api_error(exc, "read Pod", missing_ok=False) + raise RuntimeError("Failed to read Kubernetes Pod") from None + status = getattr(pod, "status", None) + phase = getattr(status, "phase", None) if status is not None else None + reason = None + if status is not None: + for container in status.container_statuses or []: + state = getattr(container, "state", None) + waiting = getattr(state, "waiting", None) if state is not None else None + terminated = ( + getattr(state, "terminated", None) if state is not None else None + ) + if waiting is not None and getattr(waiting, "reason", None): + reason = waiting.reason + elif terminated is not None and getattr(terminated, "reason", None): + reason = terminated.reason + return {"phase": phase or "Unknown", "reason": reason} + + def _metadata(self, meta: Mapping[str, Any]) -> Any: + return self._client.V1ObjectMeta( + name=meta["name"], + namespace=meta["namespace"], + labels=dict(meta["labels"]), + ) + + def _pod(self, manifest: Mapping[str, Any]) -> Any: + spec = manifest["spec"] + container = spec["containers"][0] + container_kwargs: dict[str, Any] = { + "name": container["name"], + "image": container["image"], + "ports": [ + self._client.V1ContainerPort( + name=port["name"], container_port=port["container_port"] + ) + for port in container["ports"] + ], + } + if container.get("env"): + container_kwargs["env"] = [ + self._client.V1EnvVar(name=item["name"], value=item["value"]) + for item in container["env"] + ] + resources = container.get("resources") + if resources: + container_kwargs["resources"] = self._client.V1ResourceRequirements( + requests=resources.get("requests"), + limits=resources.get("limits"), + ) + pod_kwargs: dict[str, Any] = { + "restart_policy": spec["restart_policy"], + "automount_service_account_token": spec["automount_service_account_token"], + "containers": [self._client.V1Container(**container_kwargs)], + } + if spec.get("runtime_class_name"): + pod_kwargs["runtime_class_name"] = spec["runtime_class_name"] + if spec.get("service_account_name"): + pod_kwargs["service_account_name"] = spec["service_account_name"] + if spec.get("image_pull_secrets"): + pod_kwargs["image_pull_secrets"] = [ + self._client.V1LocalObjectReference(name=secret) + for secret in spec["image_pull_secrets"] + ] + return self._client.V1Pod( + metadata=self._metadata(manifest["metadata"]), + spec=self._client.V1PodSpec(**pod_kwargs), + ) + + def _service(self, manifest: Mapping[str, Any]) -> Any: + spec = manifest["spec"] + port = spec["ports"][0] + return self._client.V1Service( + metadata=self._metadata(manifest["metadata"]), + spec=self._client.V1ServiceSpec( + selector=dict(spec["selector"]), + ports=[ + self._client.V1ServicePort( + name=port["name"], + port=port["port"], + target_port=port["target_port"], + protocol=port["protocol"], + ) + ], + ), + ) + + +class KubernetesProvider(ContainerProvider): + """ + Container provider that runs one environment as one Pod and one Service. + + ``start_container`` returns ``http://{name}.{namespace}.svc.cluster.local:{port}``. + That URL is reachable only from inside the cluster. It is plaintext on + purpose: in-cluster HTTP is the connectivity decision for this provider. + Cloud sandbox providers keep their own HTTPS requirement. + + The namespace must already exist. This provider does not create namespaces, + Deployments, Ingress objects, or port-forwards. + + Only one Pod is active per provider. Calling ``start_container`` again + before ``stop_container()`` / ``close()`` raises ``RuntimeError`` rather + than orphaning the running Pod. A failed start deletes only the objects + this instance created. ``stop_container`` is safe to call more than once. + + The Pod does not mount a service-account token unless + ``automount_service_account_token=True`` is passed to ``start_container``. + + Example: + ```python + from openenv.core.containers.runtime.kubernetes_provider import ( + KubernetesProvider, + ) + + provider = KubernetesProvider(namespace="openenv") + base_url = provider.start_container("echo-env:latest") + provider.wait_for_ready(base_url) + provider.stop_container() + ``` + """ + + def __init__( + self, + *, + image: str | None = None, + env_vars: dict[str, str] | None = None, + namespace: str = _DEFAULT_NAMESPACE, + kubeconfig: str | None = None, + context: str | None = None, + _adapter: Any = None, + ): + """ + Args: + image (`str`, *optional*): + Registry image used when ``start_container()`` is called + without an image. + env_vars (`dict`, *optional*): + Environment variables used when ``start_container()`` is called + without ``env_vars``. + namespace (`str`, *optional*, defaults to ``"default"``): + Existing namespace for the Pod and Service. Not created. + kubeconfig (`str`, *optional*): + Kubeconfig path. When omitted, in-cluster config is tried and + then the default kubeconfig. + context (`str`, *optional*): + Kubeconfig context. Implies kubeconfig loading rather than + in-cluster config. + """ + self._image = image + self._env_vars = env_vars + self._namespace = _dns_label(namespace, max_length=63, what="namespace") + self._name: str | None = None + self._base_url: str | None = None + self._service_port: int | None = None + self._pod_created = False + self._service_created = False + + if _adapter is None: + self._adapter: Any = _DefaultKubernetesAdapter( + namespace=self._namespace, + kubeconfig=kubeconfig, + context=context, + ) + else: + self._adapter = _adapter + + @property + def base_url(self) -> str: + """URL returned by the last successful ``start_container``.""" + if self._base_url is None: + raise RuntimeError( + "KubernetesProvider has no active base_url. Start the provider " + "before reading base_url." + ) + return self._base_url + + def start_container( + self, + image: str | None = None, + port: int | None = None, + env_vars: dict[str, str] | None = None, + **kwargs: Any, + ) -> str: + """ + Create one Pod and one Service and return the in-cluster base URL. + + Args: + image (`str`, *optional*): + Registry image. May be omitted when supplied to the constructor. + port (`int`, *optional*): + Service port. Defaults to 8000. The container always listens + on 8000, matching ``LocalDockerProvider``. + env_vars (`dict`, *optional*): + Environment variables for the container. Replaces constructor + ``env_vars`` when passed, including an empty dict. + **kwargs: + ``resources``, ``runtime_class_name``, ``image_pull_secrets``, + ``labels``, ``service_account_name``, and + ``automount_service_account_token``. Unknown keys raise + ``ValueError``. + + Returns: + `str`: ``http://{name}.{namespace}.svc.cluster.local:{port}``. + """ + if self._pod_created or self._service_created or self._base_url is not None: + raise RuntimeError( + "KubernetesProvider already has an active Pod. Call " + "stop_container() (or close()) before starting another — a " + "second start would orphan the running Pod." + ) + + unknown = set(kwargs) - _START_KWARGS + if unknown: + raise ValueError( + f"Unsupported kwargs for KubernetesProvider: {sorted(unknown)}" + ) + + effective_image = image if image is not None else self._image + if not isinstance(effective_image, str) or not effective_image: + raise ValueError( + "KubernetesProvider requires an image. Pass it to the " + "constructor or start_container()." + ) + service_port = _CONTAINER_PORT if port is None else port + if isinstance(service_port, bool) or not isinstance(service_port, int): + raise ValueError("port must be an integer.") + if not 1 <= service_port <= 65535: + raise ValueError("port must be between 1 and 65535.") + + effective_env = self._env_vars if env_vars is None else env_vars + env_list = _env_list(effective_env) + resources = _resources(kwargs.get("resources")) + runtime_class_name = _optional_dns_subdomain( + kwargs.get("runtime_class_name"), what="runtime_class_name" + ) + service_account_name = _optional_dns_subdomain( + kwargs.get("service_account_name"), what="service_account_name" + ) + image_pull_secrets = _secret_names(kwargs.get("image_pull_secrets")) + automount = kwargs.get("automount_service_account_token", False) + if not isinstance(automount, bool): + raise ValueError("automount_service_account_token must be a bool.") + + name = _resource_name(effective_image) + labels = _merged_labels(name, kwargs.get("labels")) + self._name = name + self._service_port = service_port + try: + self._adapter.ensure_namespace() + self._adapter.create_pod( + _pod_manifest( + name=name, + namespace=self._namespace, + labels=labels, + image=effective_image, + env_list=env_list, + resources=resources, + runtime_class_name=runtime_class_name, + service_account_name=service_account_name, + image_pull_secrets=image_pull_secrets, + automount_service_account_token=automount, + ) + ) + self._pod_created = True + self._adapter.create_service( + _service_manifest( + name=name, + namespace=self._namespace, + labels=labels, + service_port=service_port, + ) + ) + self._service_created = True + except Exception: + # A cleanup failure must not mask the original error. + try: + self.stop_container() + except Exception: + pass + raise + + self._base_url = _incluster_base_url(name, self._namespace, service_port) + return self._base_url + + def stop_container(self) -> None: + """Delete the Pod and Service this provider created. + + Safe when nothing was started and safe to call twice. A 404 from the + API is treated as already deleted. Objects this instance did not + create are left in place, including a name that was already taken. + """ + if self._name is None and not self._pod_created and not self._service_created: + self._base_url = None + self._service_port = None + return + + name = self._name + errors: list[str] = [] + if self._pod_created and name is not None: + try: + self._adapter.delete_pod(name) + self._pod_created = False + except Exception: + errors.append("pod") + if self._service_created and name is not None: + try: + self._adapter.delete_service(name) + self._service_created = False + except Exception: + errors.append("service") + if errors: + raise RuntimeError( + "Failed to delete Kubernetes resources created by this provider. " + "Call stop_container() again to retry." + ) + self._name = None + self._base_url = None + self._service_port = None + + def close(self) -> None: + """Stop the active Pod and Service. + + Overrides the base no-op so a caller holding a bare ``ContainerProvider`` + reference can release the Pod polymorphically (also invoked on + context-manager exit). Equivalent to ``stop_container()``. + """ + self.stop_container() + + def wait_for_ready(self, base_url: str, timeout_s: float = 120.0) -> None: + """ + Wait until the Pod is Running and ``GET /health`` returns 200. + + A Failed or Succeeded Pod raises immediately instead of consuming the + timeout. The URL is not included in errors: this method only polls the + Service URL returned by ``start_container``. + + Args: + base_url (`str`): + URL returned by ``start_container()``. + timeout_s (`float`, *optional*, defaults to `120.0`): + Maximum seconds to wait. Image pulls are slower than local Docker. + + Raises: + TimeoutError: If the Pod does not become ready in time. + RuntimeError: If the Pod has already ended. + ValueError: If ``base_url`` is not the URL this provider returned. + """ + import requests + + expected = self._base_url + if expected is None or not isinstance(base_url, str): + raise ValueError( + "base_url does not match the in-cluster Service URL for this provider." + ) + if base_url.rstrip("/") != expected: + raise ValueError( + "base_url does not match the in-cluster Service URL for this provider." + ) + if isinstance(timeout_s, bool) or not isinstance(timeout_s, (int, float)): + raise ValueError("timeout_s must be a number.") + if timeout_s < 0: + raise ValueError("timeout_s must be non-negative.") + + health_url = f"{expected}/health" + proxies = {"http": None, "https": None} + deadline = time.time() + timeout_s + while time.time() < deadline: + phase, reason = self._phase() + if phase in ("Failed", "Succeeded"): + raise RuntimeError(self._ended_message(phase, reason)) + if phase == "Running": + try: + response = requests.get(health_url, timeout=2.0, proxies=proxies) + if response.status_code == 200: + return + except requests.RequestException: + pass + time.sleep(0.5) + + raise TimeoutError(f"Kubernetes Pod did not become ready within {timeout_s}s.") + + def _phase(self) -> tuple[str, Optional[str]]: + if self._name is None: + raise RuntimeError( + "KubernetesProvider has no active Pod. Start the provider " + "before waiting for readiness." + ) + status = self._adapter.pod_status(self._name) + phase = status.get("phase") if isinstance(status, Mapping) else None + reason = status.get("reason") if isinstance(status, Mapping) else None + if not isinstance(phase, str) or not _SAFE_REASON.fullmatch(phase): + phase = "Unknown" + return phase, _safe_reason(reason if isinstance(reason, str) else None) + + def _ended_message(self, phase: str, reason: Optional[str]) -> str: + if reason is None: + return ( + f"Kubernetes Pod entered phase {phase} before it became ready. " + "The Service URL and Pod status message are omitted." + ) + return ( + f"Kubernetes Pod entered phase {phase} ({reason}) before it became " + "ready. The Service URL and Pod status message are omitted." + ) + + +def _env_list(env_vars: Optional[Mapping[str, str]]) -> list[dict[str, str]]: + if env_vars is None: + return [] + if not isinstance(env_vars, Mapping): + raise ValueError("env_vars must be a mapping of strings.") + items: list[dict[str, str]] = [] + for key, value in env_vars.items(): + if not isinstance(key, str) or not isinstance(value, str): + raise ValueError("env_vars keys and values must be strings.") + items.append({"name": key, "value": value}) + return items + + +def _resources(resources: Any) -> Optional[dict[str, dict[str, str]]]: + if resources is None: + return None + if not isinstance(resources, Mapping): + raise ValueError("resources must be a mapping of requests and limits.") + unknown = set(resources) - {"requests", "limits"} + if unknown: + raise ValueError( + f"Unsupported resources keys for KubernetesProvider: {sorted(unknown)}" + ) + parsed: dict[str, dict[str, str]] = {} + for section in ("requests", "limits"): + values = resources.get(section) + if values is None: + continue + if not isinstance(values, Mapping) or not values: + raise ValueError(f"resources.{section} must be a non-empty mapping.") + parsed[section] = {} + for key, value in values.items(): + if not isinstance(key, str) or not isinstance(value, str) or not value: + raise ValueError( + f"resources.{section} keys and values must be non-empty strings." + ) + parsed[section][key] = value + return parsed or None + + +def _optional_dns_subdomain(value: Any, *, what: str) -> Optional[str]: + if value is None: + return None + if not isinstance(value, str) or not _dns_subdomain(value): + raise ValueError( + f"{what} must be a DNS-1123 subdomain: lowercase labels separated by '.'." + ) + return value + + +def _secret_names(value: Any) -> list[str]: + if value is None: + return [] + if isinstance(value, str) or not isinstance(value, (list, tuple)): + raise ValueError("image_pull_secrets must be a list of secret names.") + names: list[str] = [] + for item in value: + if not isinstance(item, str) or not _dns_subdomain(item): + raise ValueError("image_pull_secrets entries must be DNS-1123 subdomains.") + names.append(item) + return names + + +def _pod_manifest( + *, + name: str, + namespace: str, + labels: dict[str, str], + image: str, + env_list: list[dict[str, str]], + resources: Optional[dict[str, dict[str, str]]], + runtime_class_name: Optional[str], + service_account_name: Optional[str], + image_pull_secrets: list[str], + automount_service_account_token: bool, +) -> dict[str, Any]: + container: dict[str, Any] = { + "name": _CONTAINER_NAME, + "image": image, + "ports": [{"name": "http", "container_port": _CONTAINER_PORT}], + } + if env_list: + container["env"] = env_list + if resources: + container["resources"] = resources + spec: dict[str, Any] = { + "restart_policy": "Never", + "automount_service_account_token": automount_service_account_token, + "containers": [container], + } + if runtime_class_name: + spec["runtime_class_name"] = runtime_class_name + if service_account_name: + spec["service_account_name"] = service_account_name + if image_pull_secrets: + spec["image_pull_secrets"] = image_pull_secrets + return { + "metadata": {"name": name, "namespace": namespace, "labels": labels}, + "spec": spec, + } + + +def _service_manifest( + *, + name: str, + namespace: str, + labels: dict[str, str], + service_port: int, +) -> dict[str, Any]: + return { + "metadata": {"name": name, "namespace": namespace, "labels": labels}, + "spec": { + "selector": {_SELECTOR_KEY: name}, + "ports": [ + { + "name": "http", + "port": service_port, + "target_port": _CONTAINER_PORT, + "protocol": "TCP", + } + ], + }, + } + + +__all__ = ["KubernetesProvider"] diff --git a/src/openenv/core/containers/runtime/providers.py b/src/openenv/core/containers/runtime/providers.py index 3b7575f701..aa35fbbffd 100644 --- a/src/openenv/core/containers/runtime/providers.py +++ b/src/openenv/core/containers/runtime/providers.py @@ -21,7 +21,9 @@ class ContainerProvider(ABC): Providers implement this interface to support different container platforms: - LocalDockerProvider: Runs containers on local Docker daemon - - KubernetesProvider: Runs containers in Kubernetes cluster + - DockerSwarmProvider: Runs containers as a Docker Swarm service + - KubernetesProvider: optional ``openenv[kubernetes]`` extra + (``kubernetes_provider.py``) - FargateProvider: Runs containers on AWS Fargate - CloudRunProvider: Runs containers on Google Cloud Run @@ -644,19 +646,6 @@ def _generate_service_name(self, image: str) -> str: return f"{clean_image}-swarm-{timestamp}" -class KubernetesProvider(ContainerProvider): - """ - Planned container provider for Kubernetes clusters. - - Not yet implemented: this is a placeholder for the planned Kubernetes - backend and does not implement the abstract `ContainerProvider` methods, so - it cannot be instantiated. Use `LocalDockerProvider`, `DockerSwarmProvider`, - `DaytonaProvider`, or `ACASandboxProvider` instead. - """ - - pass - - class RuntimeProvider(ABC): """ Abstract base class for runtime providers that are not container providers. diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py new file mode 100644 index 0000000000..b2619be1c0 --- /dev/null +++ b/tests/test_core/test_kubernetes_provider.py @@ -0,0 +1,649 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Unit tests for KubernetesProvider. + +Provider-behavior tests inject a duck-typed fake adapter (``_adapter=``), so +they run without ``pip install kubernetes``. A separate test installs a minimal +fake ``kubernetes`` module to pin the real ``_DefaultKubernetesAdapter`` SDK +shape, including in-cluster config fallback and 404/409 handling that must not +leak API bodies. +""" + +from __future__ import annotations + +import sys +import types +from unittest.mock import MagicMock, patch + +import pytest +from openenv.core.containers.runtime import kubernetes_provider +from openenv.core.containers.runtime.kubernetes_provider import ( + _DefaultKubernetesAdapter, + KubernetesProvider, +) +from openenv.core.containers.runtime.providers import ContainerProvider + + +class _FakeAdapter: + def __init__(self): + self.pods: list[dict] = [] + self.services: list[dict] = [] + self.deleted_pods = 0 + self.deleted_services = 0 + self.ensured = 0 + self.phase = "Running" + self.reason = None + self.namespace_exists = True + self.fail_pod: Exception | None = None + self.fail_service: Exception | None = None + self.fail_delete_pod: Exception | None = None + + def ensure_namespace(self): + self.ensured += 1 + if not self.namespace_exists: + raise RuntimeError( + "Kubernetes namespace does not exist. " + "KubernetesProvider does not create namespaces." + ) + + def create_pod(self, manifest): + if self.fail_pod is not None: + raise self.fail_pod + self.pods.append(manifest) + + def create_service(self, manifest): + if self.fail_service is not None: + raise self.fail_service + self.services.append(manifest) + + def delete_pod(self, name): + self.deleted_pods += 1 + if self.fail_delete_pod is not None: + raise self.fail_delete_pod + + def delete_service(self, name): + self.deleted_services += 1 + + def pod_status(self, name): + return {"phase": self.phase, "reason": self.reason} + + +@pytest.fixture() +def adapter(): + return _FakeAdapter() + + +@pytest.fixture() +def provider(adapter): + return KubernetesProvider(namespace="openenv", _adapter=adapter) + + +@pytest.fixture(autouse=True) +def _fast_provider_sleep(): + with patch("openenv.core.containers.runtime.kubernetes_provider.time.sleep"): + yield + + +def _labels(manifest): + return manifest["metadata"]["labels"] + + +class TestStartContainer: + def test_returns_incluster_http_url(self, provider): + url = provider.start_container("echo-env:latest") + assert url.startswith("http://openenv-echo-env-") + assert url.endswith(".openenv.svc.cluster.local:8000") + assert not url.startswith("https://") + assert provider.base_url == url + + def test_explicit_port_is_service_port_only(self, provider, adapter): + url = provider.start_container("echo-env:latest", port=9000) + assert url.endswith(".openenv.svc.cluster.local:9000") + assert adapter.pods[0]["spec"]["containers"][0]["ports"] == [ + {"name": "http", "container_port": 8000} + ] + assert adapter.services[0]["spec"]["ports"][0]["target_port"] == 8000 + assert adapter.services[0]["spec"]["ports"][0]["port"] == 9000 + + def test_image_name_is_dns_safe(self, provider, adapter): + provider.start_container("registry.example.com/Org/Echo_Env:latest") + name = adapter.pods[0]["metadata"]["name"] + assert name.startswith("openenv-echo-env-") + assert len(name) <= 63 + assert "_" not in name + assert name == adapter.services[0]["metadata"]["name"] + + def test_long_image_name_stays_within_63_chars(self, provider, adapter): + provider.start_container(("a" * 80) + ":latest") + assert len(adapter.pods[0]["metadata"]["name"]) <= 63 + + def test_unusable_image_name_raises_before_create(self, provider, adapter): + with pytest.raises(ValueError, match="valid Kubernetes name"): + provider.start_container("---:latest") + assert adapter.pods == [] + assert adapter.ensured == 0 + + def test_constructor_image_used_when_start_omits_image(self, adapter): + provider = KubernetesProvider( + _adapter=adapter, image="constructor-env:latest", namespace="openenv" + ) + provider.start_container() + assert adapter.pods[0]["spec"]["containers"][0]["image"] == ( + "constructor-env:latest" + ) + + def test_start_image_overrides_constructor_image(self, adapter): + provider = KubernetesProvider( + _adapter=adapter, image="constructor-env:latest", namespace="openenv" + ) + provider.start_container("explicit-env:latest") + assert adapter.pods[0]["spec"]["containers"][0]["image"] == ( + "explicit-env:latest" + ) + + def test_requires_image_from_constructor_or_start(self, adapter): + provider = KubernetesProvider(_adapter=adapter) + with pytest.raises(ValueError, match="requires an image"): + provider.start_container() + + def test_env_vars_forwarded(self, provider, adapter): + provider.start_container("echo-env:latest", env_vars={"TOKEN": "supersecret"}) + assert adapter.pods[0]["spec"]["containers"][0]["env"] == [ + {"name": "TOKEN", "value": "supersecret"} + ] + + def test_empty_env_vars_override_constructor(self, adapter): + provider = KubernetesProvider( + _adapter=adapter, + namespace="openenv", + image="echo-env:latest", + env_vars={"TOKEN": "supersecret"}, + ) + provider.start_container(env_vars={}) + assert "env" not in adapter.pods[0]["spec"]["containers"][0] + + def test_ownership_labels_override_caller(self, provider, adapter): + provider.start_container( + "echo-env:latest", + labels={ + "app.kubernetes.io/managed-by": "attacker", + "app.kubernetes.io/instance": "hijack", + "team": "rl", + }, + ) + labels = _labels(adapter.pods[0]) + name = adapter.pods[0]["metadata"]["name"] + assert labels["app.kubernetes.io/managed-by"] == "openenv" + assert labels["app.kubernetes.io/instance"] == name + assert labels["app.kubernetes.io/part-of"] == "openenv" + assert labels["team"] == "rl" + assert adapter.services[0]["spec"]["selector"] == { + "app.kubernetes.io/instance": name + } + assert _labels(adapter.services[0])["app.kubernetes.io/managed-by"] == ( + "openenv" + ) + + def test_invalid_label_value_rejected_before_create(self, provider, adapter): + with pytest.raises(ValueError, match="label values"): + provider.start_container( + "echo-env:latest", labels={"team": "line\nsecret-value"} + ) + assert adapter.pods == [] + assert "secret-value" not in repr(adapter.pods) + + def test_functional_kwargs(self, provider, adapter): + provider.start_container( + "echo-env:latest", + resources={"requests": {"cpu": "100m"}, "limits": {"memory": "128Mi"}}, + runtime_class_name="gvisor", + image_pull_secrets=["reg-cred"], + service_account_name="openenv-runner", + ) + spec = adapter.pods[0]["spec"] + container = spec["containers"][0] + assert container["resources"]["requests"] == {"cpu": "100m"} + assert container["resources"]["limits"] == {"memory": "128Mi"} + assert spec["runtime_class_name"] == "gvisor" + assert spec["image_pull_secrets"] == ["reg-cred"] + assert spec["service_account_name"] == "openenv-runner" + assert spec["automount_service_account_token"] is False + assert spec["restart_policy"] == "Never" + + def test_unknown_kwarg_rejected_before_create(self, provider, adapter): + with pytest.raises(ValueError, match="ingress"): + provider.start_container("echo-env:latest", ingress="yes") + assert adapter.pods == [] + + def test_invalid_namespace_rejected_before_api(self, adapter): + with pytest.raises(ValueError, match="namespace"): + KubernetesProvider(namespace="Not_A_Namespace", _adapter=adapter) + assert adapter.ensured == 0 + + def test_missing_namespace_raises_and_creates_nothing(self, provider, adapter): + adapter.namespace_exists = False + with pytest.raises(RuntimeError, match="does not exist"): + provider.start_container("echo-env:latest") + assert adapter.pods == [] + assert adapter.deleted_pods == 0 + + def test_rejects_bad_port(self, provider): + with pytest.raises(ValueError, match="port"): + provider.start_container("echo-env:latest", port=0) + with pytest.raises(ValueError, match="port"): + provider.start_container("echo-env:latest", port=True) + + +class TestLifecycle: + def test_double_start_raises_and_preserves_existing_pod(self, provider, adapter): + provider.start_container("echo-env:latest") + with pytest.raises(RuntimeError, match="already has an active Pod"): + provider.start_container("other:latest") + assert len(adapter.pods) == 1 + assert adapter.deleted_pods == 0 + + def test_start_after_stop_is_allowed(self, provider, adapter): + provider.start_container("echo-env:latest") + provider.stop_container() + provider.start_container("other:latest") + assert len(adapter.pods) == 2 + + def test_stop_deletes_pod_and_service(self, provider, adapter): + provider.start_container("echo-env:latest") + provider.stop_container() + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 1 + assert provider._name is None + assert provider._base_url is None + + def test_stop_without_start_is_noop(self, provider, adapter): + provider.stop_container() + assert adapter.deleted_pods == 0 + assert adapter.deleted_services == 0 + + def test_stop_is_idempotent(self, provider, adapter): + provider.start_container("echo-env:latest") + provider.stop_container() + provider.stop_container() + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 1 + + def test_close_deletes(self, provider, adapter): + provider.start_container("echo-env:latest") + provider.close() + assert adapter.deleted_pods == 1 + assert provider._pod_created is False + + def test_context_manager_deletes_on_exit(self, adapter): + with KubernetesProvider(namespace="openenv", _adapter=adapter) as started: + started.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 1 + + def test_is_container_provider(self, provider): + assert isinstance(provider, ContainerProvider) + + def test_not_reexported_from_runtime_package(self): + from openenv.core.containers import runtime + + assert "KubernetesProvider" not in runtime.__all__ + assert not hasattr(runtime, "KubernetesProvider") + + +class TestCleanup: + def test_service_failure_deletes_only_the_pod(self, provider, adapter): + adapter.fail_service = RuntimeError("service failed") + with pytest.raises(RuntimeError, match="service failed"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 0 + assert provider._pod_created is False + assert provider._name is None + + def test_pod_create_failure_deletes_nothing(self, provider, adapter): + adapter.fail_pod = RuntimeError("Kubernetes create Pod name is already in use") + with pytest.raises(RuntimeError, match="already in use"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 0 + assert adapter.deleted_services == 0 + + def test_cleanup_failure_does_not_mask_original_error(self, provider, adapter): + adapter.fail_service = RuntimeError("service failed") + adapter.fail_delete_pod = RuntimeError("delete failed secret-body") + with pytest.raises(RuntimeError, match="service failed") as exc_info: + provider.start_container("echo-env:latest") + assert "secret-body" not in str(exc_info.value) + assert "delete failed" not in str(exc_info.value) + assert adapter.deleted_pods == 1 + + def test_failed_delete_can_be_retried(self, provider, adapter): + provider.start_container("echo-env:latest") + adapter.fail_delete_pod = RuntimeError("apiserver unavailable") + with pytest.raises(RuntimeError, match="retry"): + provider.stop_container() + assert provider._pod_created is True + adapter.fail_delete_pod = None + provider.stop_container() + assert provider._pod_created is False + assert adapter.deleted_services == 1 + + +class TestWaitForReady: + def test_ready_when_running_and_health_200(self, provider): + url = provider.start_container("echo-env:latest") + with patch("requests.get", return_value=MagicMock(status_code=200)) as get: + provider.wait_for_ready(url, timeout_s=5) + assert get.call_args.args[0] == f"{url}/health" + assert get.call_args.kwargs["proxies"] == {"http": None, "https": None} + + def test_failed_pod_raises_immediately_without_url(self, provider, adapter): + url = provider.start_container("echo-env:latest") + adapter.phase = "Failed" + adapter.reason = "ImagePullBackOff" + with ( + patch( + "openenv.core.containers.runtime.kubernetes_provider.time.sleep" + ) as sleep, + pytest.raises(RuntimeError, match="ImagePullBackOff") as exc_info, + ): + provider.wait_for_ready(url, timeout_s=30) + sleep.assert_not_called() + assert url not in str(exc_info.value) + assert "svc.cluster.local" not in str(exc_info.value) + + def test_unsafe_reason_is_omitted(self, provider, adapter): + url = provider.start_container("echo-env:latest") + adapter.phase = "Failed" + adapter.reason = "secret-token\nImagePullBackOff" + with pytest.raises(RuntimeError, match="phase Failed") as exc_info: + provider.wait_for_ready(url, timeout_s=5) + message = str(exc_info.value) + assert "secret-token" not in message + assert "ImagePullBackOff" not in message + assert url not in message + + def test_timeout_does_not_leak_url(self, provider, adapter): + import requests + + url = provider.start_container("echo-env:latest") + adapter.phase = "Running" + with ( + patch("requests.get", side_effect=requests.ConnectionError("refused")), + patch( + "openenv.core.containers.runtime.kubernetes_provider.time.time", + side_effect=[0, 1, 100], + ), + pytest.raises(TimeoutError, match="did not become ready") as exc_info, + ): + provider.wait_for_ready(url, timeout_s=5) + assert url not in str(exc_info.value) + assert "svc.cluster.local" not in str(exc_info.value) + + def test_mismatched_base_url_is_rejected_without_echoing_it(self, provider): + provider.start_container("echo-env:latest") + evil = "http://evil.example/secret-token" + with pytest.raises(ValueError, match="does not match") as exc_info: + provider.wait_for_ready(evil, timeout_s=5) + assert "secret-token" not in str(exc_info.value) + assert "evil.example" not in str(exc_info.value) + + def test_trailing_slash_matches(self, provider): + url = provider.start_container("echo-env:latest") + with patch("requests.get", return_value=MagicMock(status_code=200)): + provider.wait_for_ready(url + "/", timeout_s=5) + + +def _install_fake_kubernetes(monkeypatch, *, incluster_fails: bool = False): + """Install a minimal fake ``kubernetes`` module and return its call log.""" + calls: dict = { + "incluster": 0, + "kubeconfig": [], + "create_pod": [], + "create_service": [], + "delete_pod": [], + "delete_service": [], + "read_namespace": [], + "read_pod": [], + } + + class ConfigException(Exception): + pass + + class ApiException(Exception): + def __init__(self, status=None, body=None): + super().__init__("api failure") + self.status = status + self.body = body + + config_mod = types.ModuleType("kubernetes.config") + config_mod.ConfigException = ConfigException + + def load_incluster_config(): + calls["incluster"] += 1 + if incluster_fails: + raise ConfigException("not in a cluster") + + def load_kube_config(config_file=None, context=None): + calls["kubeconfig"].append((config_file, context)) + + config_mod.load_incluster_config = load_incluster_config + config_mod.load_kube_config = load_kube_config + + class _Obj: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + client_mod = types.ModuleType("kubernetes.client") + for name in ( + "V1ObjectMeta", + "V1ContainerPort", + "V1EnvVar", + "V1ResourceRequirements", + "V1Container", + "V1LocalObjectReference", + "V1PodSpec", + "V1Pod", + "V1ServicePort", + "V1ServiceSpec", + "V1Service", + ): + setattr(client_mod, name, type(name, (_Obj,), {})) + + class CoreV1Api: + def read_namespace(self, name): + calls["read_namespace"].append(name) + if calls.get("namespace_status") == 404: + raise ApiException(status=404, body="secret-namespace-body") + return _Obj() + + def create_namespaced_pod(self, namespace, body): + calls["create_pod"].append((namespace, body)) + if calls.get("create_pod_status") == 409: + raise ApiException(status=409, body="secret-pod-body") + return body + + def create_namespaced_service(self, namespace, body): + calls["create_service"].append((namespace, body)) + return body + + def delete_namespaced_pod(self, name, namespace): + calls["delete_pod"].append((name, namespace)) + if calls.get("delete_pod_status") == 404: + raise ApiException(status=404, body="secret-delete-body") + if calls.get("delete_pod_status") == 500: + raise ApiException(status=500, body="secret-delete-body") + + def delete_namespaced_service(self, name, namespace): + calls["delete_service"].append((name, namespace)) + + def read_namespaced_pod(self, name, namespace): + calls["read_pod"].append((name, namespace)) + waiting = _Obj(reason="ImagePullBackOff", message="secret-status-text") + return _Obj( + status=_Obj( + phase="Pending", + container_statuses=[ + _Obj(state=_Obj(waiting=waiting, terminated=None)) + ], + ) + ) + + client_mod.CoreV1Api = CoreV1Api + exceptions_mod = types.ModuleType("kubernetes.client.exceptions") + exceptions_mod.ApiException = ApiException + client_mod.exceptions = exceptions_mod + + kubernetes_mod = types.ModuleType("kubernetes") + kubernetes_mod.client = client_mod + kubernetes_mod.config = config_mod + + monkeypatch.setitem(sys.modules, "kubernetes", kubernetes_mod) + monkeypatch.setitem(sys.modules, "kubernetes.client", client_mod) + monkeypatch.setitem(sys.modules, "kubernetes.config", config_mod) + monkeypatch.setitem(sys.modules, "kubernetes.client.exceptions", exceptions_mod) + return calls + + +class TestDefaultAdapter: + def test_incluster_config_is_preferred(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + adapter.ensure_namespace() + assert calls["incluster"] == 1 + assert calls["kubeconfig"] == [] + assert calls["read_namespace"] == ["openenv"] + + def test_falls_back_to_kubeconfig_outside_cluster(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch, incluster_fails=True) + _DefaultKubernetesAdapter(namespace="openenv", kubeconfig=None, context=None) + assert calls["incluster"] == 1 + assert calls["kubeconfig"] == [(None, None)] + + def test_explicit_kubeconfig_skips_incluster(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig="/tmp/kubeconfig", context="dev" + ) + assert calls["incluster"] == 0 + assert calls["kubeconfig"] == [("/tmp/kubeconfig", "dev")] + + def test_create_translates_manifest_and_hides_conflict_body(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["create_pod_status"] = 409 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + manifest = { + "metadata": { + "name": "openenv-echo", + "namespace": "openenv", + "labels": {"app.kubernetes.io/instance": "openenv-echo"}, + }, + "spec": { + "restart_policy": "Never", + "automount_service_account_token": False, + "runtime_class_name": "gvisor", + "image_pull_secrets": ["reg-cred"], + "containers": [ + { + "name": "env", + "image": "echo-env:latest", + "ports": [{"name": "http", "container_port": 8000}], + "env": [{"name": "TOKEN", "value": "supersecret"}], + "resources": {"limits": {"memory": "128Mi"}}, + } + ], + }, + } + with pytest.raises(RuntimeError, match="already in use") as exc_info: + adapter.create_pod(manifest) + assert "secret-pod-body" not in str(exc_info.value) + assert "supersecret" not in str(exc_info.value) + body = calls["create_pod"][0][1] + assert body.spec.runtime_class_name == "gvisor" + assert body.spec.automount_service_account_token is False + assert body.spec.containers[0].image == "echo-env:latest" + assert body.spec.containers[0].env[0].value == "supersecret" + assert calls["create_pod"][0][0] == "openenv" + + def test_delete_404_is_idempotent_and_hides_body(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["delete_pod_status"] = 404 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + adapter.delete_pod("openenv-echo") + assert calls["delete_pod"] == [("openenv-echo", "openenv")] + + def test_delete_failure_hides_api_body(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["delete_pod_status"] = 500 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + with pytest.raises(RuntimeError, match="Failed to delete Pod") as exc_info: + adapter.delete_pod("openenv-echo") + assert "secret-delete-body" not in str(exc_info.value) + + def test_missing_namespace_hides_api_body(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["namespace_status"] = 404 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + with pytest.raises(RuntimeError, match="does not exist") as exc_info: + adapter.ensure_namespace() + assert "secret-namespace-body" not in str(exc_info.value) + + def test_pod_status_keeps_reason_code_not_status_message(self, monkeypatch): + _install_fake_kubernetes(monkeypatch) + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + status = adapter.pod_status("openenv-echo") + assert status["phase"] == "Pending" + assert status["reason"] == "ImagePullBackOff" + assert "secret-status-text" not in str(status) + + def test_missing_client_names_the_extra(self, monkeypatch): + real_import = __import__ + + def guarded(name, globals=None, locals=None, fromlist=(), level=0): + if name == "kubernetes" or name.startswith("kubernetes."): + raise ImportError("blocked") + return real_import(name, globals, locals, fromlist, level) + + for key in list(sys.modules): + if key == "kubernetes" or key.startswith("kubernetes."): + monkeypatch.delitem(sys.modules, key) + monkeypatch.setattr("builtins.__import__", guarded) + with pytest.raises(RuntimeError, match=r"openenv\[kubernetes\]"): + _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + + +def test_module_import_does_not_import_kubernetes(): + source = kubernetes_provider.__file__ + assert source is not None + # The optional client is imported inside the adapter constructor, not at + # module import. Importing the provider module must not be what loads it + # for callers who never construct the default adapter. + import ast + from pathlib import Path + + tree = ast.parse(Path(source).read_text(encoding="utf-8")) + module_imports = [ + node for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom)) + ] + imported = [] + for node in module_imports: + if isinstance(node, ast.ImportFrom) and node.module: + imported.append(node.module) + if isinstance(node, ast.Import): + imported.extend(alias.name for alias in node.names) + assert not any( + name == "kubernetes" or name.startswith("kubernetes.") for name in imported + ) From 74d80ba2ad851ce42ff494f77a6e887d4195ff72 Mon Sep 17 00:00:00 2001 From: Rohit Sharma Date: Sun, 27 Sep 2026 09:10:25 +0100 Subject: [PATCH 2/7] Fix Kubernetes provider cleanup and in-cluster DNS. The namespace preflight called a cluster-scoped API, so a trainer with only a namespace Role could not start. A 404 from Pod or Service create now reports that the namespace does not exist. A lost create response used to clear the name while leaving the object in the cluster. Cleanup now reads that name and deletes it only when the OpenEnv ownership labels match. A 409 is left untouched. If the read or delete cannot be confirmed, the name stays so stop_container() can retry. The Service URL is the short name http://{name}.{namespace}:{port}. Pod DNS search completes the cluster domain, so .svc.cluster.local is not hardcoded. --- docs/source/guides/runtime-providers.md | 9 +- .../containers/runtime/kubernetes_provider.py | 163 +++++++++++++++--- tests/test_core/test_kubernetes_provider.py | 160 ++++++++++++++--- 3 files changed, 276 insertions(+), 56 deletions(-) diff --git a/docs/source/guides/runtime-providers.md b/docs/source/guides/runtime-providers.md index cbd28612ee..34b8be8f97 100644 --- a/docs/source/guides/runtime-providers.md +++ b/docs/source/guides/runtime-providers.md @@ -164,10 +164,11 @@ provider = DockerSwarmProvider() Runs one environment as one Pod and one Service in an existing namespace. Install with `pip install openenv[kubernetes]`. The returned URL is the -in-cluster Service DNS name over `http://`, which is the connectivity choice -for this provider. `ModalProvider` and `ACASandboxProvider` still require -`https://`. Use this provider from a trainer in the same cluster; it does not -create a namespace, Deployment, Ingress, or port-forward. +in-cluster Service DNS name over `http://{name}.{namespace}:{port}`, which Pod +DNS search expands under `svc.`. `ModalProvider` and +`ACASandboxProvider` still require `https://`. Use this provider from a trainer +in the same cluster; it does not create a namespace, Deployment, Ingress, or +port-forward. ```python from openenv.core.containers.runtime.kubernetes_provider import KubernetesProvider diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index e2390a09f4..7ce4881ed4 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -7,11 +7,13 @@ One provider instance owns one Pod and one Service, matching ``LocalDockerProvider`` owning one container. The returned ``base_url`` is the -in-cluster Service DNS name over ``http://``. That plaintext URL is the -in-cluster exception confirmed for this provider: ``ModalProvider`` and -``ACASandboxProvider`` still refuse non-HTTPS tunnel URLs. This provider never -returns a URL it did not construct from a validated name, namespace, and port, -and it does not copy that URL or raw API bodies into errors. +short in-cluster Service address ``http://{name}.{namespace}:{port}``. Pod DNS +search completes that to the cluster domain, so the provider does not hardcode +``.svc.cluster.local``. That plaintext URL is the in-cluster exception +confirmed for this provider: ``ModalProvider`` and ``ACASandboxProvider`` still +refuse non-HTTPS tunnel URLs. This provider never returns a URL it did not +construct from a validated name, namespace, and port, and it does not copy +that URL or raw API bodies into errors. """ from __future__ import annotations @@ -150,10 +152,30 @@ def _safe_reason(reason: Optional[str]) -> Optional[str]: return None +class _NamespaceMissing(RuntimeError): + """Creating a namespaced object returned 404: the namespace is absent.""" + + +class _CreateConflict(RuntimeError): + """Creating a namespaced object returned 409: the name is already taken.""" + + def _incluster_base_url(name: str, namespace: str, port: int) -> str: # *name* and *namespace* are already DNS-1123 labels and *port* is an int, # so this cannot be steered by a status field from the API server. - return f"http://{name}.{namespace}.svc.cluster.local:{port}" + # ``name.namespace`` is completed by the Pod DNS search list + # (``svc.``), which works when the cluster domain is not + # ``cluster.local``. + return f"http://{name}.{namespace}:{port}" + + +def _owned_by_provider(labels: Any, name: str) -> bool: + if not isinstance(labels, Mapping): + return False + return ( + labels.get("app.kubernetes.io/managed-by") == "openenv" + and labels.get("app.kubernetes.io/instance") == name + ) class _DefaultKubernetesAdapter: @@ -198,21 +220,17 @@ def _api_error(self, exc: Exception, action: str, *, missing_ok: bool) -> None: if missing_ok and status == 404: return if status == 409: - raise RuntimeError(f"Kubernetes {action} name is already in use") from None + raise _CreateConflict( + f"Kubernetes {action} name is already in use" + ) from None + if status == 404 and action.startswith("create "): + raise _NamespaceMissing( + "Kubernetes namespace does not exist. " + "KubernetesProvider does not create namespaces." + ) from None # The API body can echo object contents. Do not chain it into the error. raise RuntimeError(f"Failed to {action} Kubernetes resource") from None - def ensure_namespace(self) -> None: - try: - self._api.read_namespace(self._namespace) - except self._ApiException as exc: - if getattr(exc, "status", None) == 404: - raise RuntimeError( - "Kubernetes namespace does not exist. " - "KubernetesProvider does not create namespaces." - ) from None - raise RuntimeError("Failed to read Kubernetes namespace") from None - def create_pod(self, manifest: Mapping[str, Any]) -> None: try: self._api.create_namespaced_pod(self._namespace, self._pod(manifest)) @@ -227,6 +245,32 @@ def create_service(self, manifest: Mapping[str, Any]) -> None: except self._ApiException as exc: self._api_error(exc, "create Service", missing_ok=False) + def pod_labels(self, name: str) -> Optional[dict[str, str]]: + """Return Pod labels, or ``None`` when the Pod does not exist.""" + return self._resource_labels(name, self._api.read_namespaced_pod, "read Pod") + + def service_labels(self, name: str) -> Optional[dict[str, str]]: + """Return Service labels, or ``None`` when the Service does not exist.""" + return self._resource_labels( + name, self._api.read_namespaced_service, "read Service" + ) + + def _resource_labels( + self, name: str, read: Any, action: str + ) -> Optional[dict[str, str]]: + try: + obj = read(name, self._namespace) + except self._ApiException as exc: + if getattr(exc, "status", None) == 404: + return None + self._api_error(exc, action, missing_ok=False) + raise RuntimeError(f"Failed to {action} Kubernetes resource") from None + metadata = getattr(obj, "metadata", None) + labels = getattr(metadata, "labels", None) if metadata is not None else None + if not isinstance(labels, Mapping): + return {} + return {str(key): str(value) for key, value in labels.items()} + def delete_pod(self, name: str) -> None: try: self._api.delete_namespaced_pod(name, self._namespace) @@ -334,10 +378,11 @@ class KubernetesProvider(ContainerProvider): """ Container provider that runs one environment as one Pod and one Service. - ``start_container`` returns ``http://{name}.{namespace}.svc.cluster.local:{port}``. - That URL is reachable only from inside the cluster. It is plaintext on - purpose: in-cluster HTTP is the connectivity decision for this provider. - Cloud sandbox providers keep their own HTTPS requirement. + ``start_container`` returns ``http://{name}.{namespace}:{port}``. That + address is reachable only from inside the cluster: Pod DNS search expands + it under ``svc.``. It is plaintext on purpose: in-cluster + HTTP is the connectivity decision for this provider. Cloud sandbox + providers keep their own HTTPS requirement. The namespace must already exist. This provider does not create namespaces, Deployments, Ingress objects, or port-forwards. @@ -444,7 +489,7 @@ def start_container( ``ValueError``. Returns: - `str`: ``http://{name}.{namespace}.svc.cluster.local:{port}``. + `str`: ``http://{name}.{namespace}:{port}``. """ if self._pod_created or self._service_created or self._base_url is not None: raise RuntimeError( @@ -490,7 +535,6 @@ def start_container( self._name = name self._service_port = service_port try: - self._adapter.ensure_namespace() self._adapter.create_pod( _pod_manifest( name=name, @@ -515,10 +559,10 @@ def start_container( ) ) self._service_created = True - except Exception: + except Exception as exc: # A cleanup failure must not mask the original error. try: - self.stop_container() + self._cleanup_after_failed_start(exc) except Exception: pass raise @@ -526,6 +570,73 @@ def start_container( self._base_url = _incluster_base_url(name, self._namespace, service_port) return self._base_url + def _cleanup_after_failed_start(self, exc: BaseException) -> None: + """Delete objects this start may have created. + + A 409 is someone else's name, and a 404 means the namespace is + missing, so those do not probe. Any other create error may mean the + API server applied the object and the client lost the response. Those + are deleted only when the OpenEnv ownership labels match. If the read + or delete cannot be confirmed, the name and flag are kept so + ``stop_container()`` can retry. + """ + uncertain = not isinstance(exc, (_CreateConflict, _NamespaceMissing)) + self._release("pod", uncertain=uncertain) + self._release("service", uncertain=uncertain) + if not self._pod_created and not self._service_created: + self._name = None + self._base_url = None + self._service_port = None + + def _release(self, kind: str, *, uncertain: bool) -> None: + flagged = self._pod_created if kind == "pod" else self._service_created + if not flagged and not uncertain: + return + if not flagged: + owned = self._lookup_owned(kind) + if owned is None or not owned: + return + self._delete_kind(kind) + + def _lookup_owned(self, kind: str) -> Optional[bool]: + """Return whether *kind* at ``_name`` is ours. + + ``None`` means the read itself failed. The flag is then set so a later + ``stop_container()`` retries the delete. + """ + if self._name is None: + return False + try: + if kind == "pod": + labels = self._adapter.pod_labels(self._name) + else: + labels = self._adapter.service_labels(self._name) + except Exception: + self._set_created(kind, True) + return None + if labels is None: + return False + return _owned_by_provider(labels, self._name) + + def _delete_kind(self, kind: str) -> None: + if self._name is None: + return + try: + if kind == "pod": + self._adapter.delete_pod(self._name) + else: + self._adapter.delete_service(self._name) + except Exception: + self._set_created(kind, True) + return + self._set_created(kind, False) + + def _set_created(self, kind: str, created: bool) -> None: + if kind == "pod": + self._pod_created = created + else: + self._service_created = created + def stop_container(self) -> None: """Delete the Pod and Service this provider created. diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index b2619be1c0..081d67bf68 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -18,7 +18,9 @@ import pytest from openenv.core.containers.runtime import kubernetes_provider from openenv.core.containers.runtime.kubernetes_provider import ( + _CreateConflict, _DefaultKubernetesAdapter, + _NamespaceMissing, KubernetesProvider, ) from openenv.core.containers.runtime.providers import ContainerProvider @@ -30,29 +32,36 @@ def __init__(self): self.services: list[dict] = [] self.deleted_pods = 0 self.deleted_services = 0 - self.ensured = 0 self.phase = "Running" self.reason = None - self.namespace_exists = True self.fail_pod: Exception | None = None self.fail_service: Exception | None = None self.fail_delete_pod: Exception | None = None - - def ensure_namespace(self): - self.ensured += 1 - if not self.namespace_exists: - raise RuntimeError( - "Kubernetes namespace does not exist. " - "KubernetesProvider does not create namespaces." - ) + self.fail_read_pod: Exception | None = None + self.fail_read_service: Exception | None = None + self.persist_pod_on_failure = False + self.persist_service_on_failure = False + self.foreign_pod_on_failure = False def create_pod(self, manifest): if self.fail_pod is not None: + if self.persist_pod_on_failure: + self.pods.append(manifest) + elif self.foreign_pod_on_failure: + foreign = { + "metadata": { + "name": manifest["metadata"]["name"], + "labels": {"app.kubernetes.io/managed-by": "someone-else"}, + } + } + self.pods.append(foreign) raise self.fail_pod self.pods.append(manifest) def create_service(self, manifest): if self.fail_service is not None: + if self.persist_service_on_failure: + self.services.append(manifest) raise self.fail_service self.services.append(manifest) @@ -64,6 +73,23 @@ def delete_pod(self, name): def delete_service(self, name): self.deleted_services += 1 + def pod_labels(self, name): + if self.fail_read_pod is not None: + raise self.fail_read_pod + return self._labels(self.pods, name) + + def service_labels(self, name): + if self.fail_read_service is not None: + raise self.fail_read_service + return self._labels(self.services, name) + + @staticmethod + def _labels(items, name): + for item in items: + if item["metadata"]["name"] == name: + return dict(item["metadata"]["labels"]) + return None + def pod_status(self, name): return {"phase": self.phase, "reason": self.reason} @@ -92,13 +118,14 @@ class TestStartContainer: def test_returns_incluster_http_url(self, provider): url = provider.start_container("echo-env:latest") assert url.startswith("http://openenv-echo-env-") - assert url.endswith(".openenv.svc.cluster.local:8000") + assert url.endswith(".openenv:8000") + assert ".svc." not in url assert not url.startswith("https://") assert provider.base_url == url def test_explicit_port_is_service_port_only(self, provider, adapter): url = provider.start_container("echo-env:latest", port=9000) - assert url.endswith(".openenv.svc.cluster.local:9000") + assert url.endswith(".openenv:9000") assert adapter.pods[0]["spec"]["containers"][0]["ports"] == [ {"name": "http", "container_port": 8000} ] @@ -121,7 +148,6 @@ def test_unusable_image_name_raises_before_create(self, provider, adapter): with pytest.raises(ValueError, match="valid Kubernetes name"): provider.start_container("---:latest") assert adapter.pods == [] - assert adapter.ensured == 0 def test_constructor_image_used_when_start_omits_image(self, adapter): provider = KubernetesProvider( @@ -218,14 +244,18 @@ def test_unknown_kwarg_rejected_before_create(self, provider, adapter): def test_invalid_namespace_rejected_before_api(self, adapter): with pytest.raises(ValueError, match="namespace"): KubernetesProvider(namespace="Not_A_Namespace", _adapter=adapter) - assert adapter.ensured == 0 + assert adapter.pods == [] def test_missing_namespace_raises_and_creates_nothing(self, provider, adapter): - adapter.namespace_exists = False + adapter.fail_pod = _NamespaceMissing( + "Kubernetes namespace does not exist. " + "KubernetesProvider does not create namespaces." + ) with pytest.raises(RuntimeError, match="does not exist"): provider.start_container("echo-env:latest") assert adapter.pods == [] assert adapter.deleted_pods == 0 + assert provider._name is None def test_rejects_bad_port(self, provider): with pytest.raises(ValueError, match="port"): @@ -300,12 +330,70 @@ def test_service_failure_deletes_only_the_pod(self, provider, adapter): assert provider._pod_created is False assert provider._name is None - def test_pod_create_failure_deletes_nothing(self, provider, adapter): - adapter.fail_pod = RuntimeError("Kubernetes create Pod name is already in use") - with pytest.raises(RuntimeError, match="already in use"): + def test_pod_create_conflict_deletes_nothing(self, provider, adapter): + adapter.fail_pod = _CreateConflict( + "Kubernetes create Pod name is already in use" + ) + adapter.persist_pod_on_failure = True + with pytest.raises(_CreateConflict, match="already in use"): provider.start_container("echo-env:latest") assert adapter.deleted_pods == 0 assert adapter.deleted_services == 0 + assert provider._name is None + + def test_service_conflict_deletes_the_pod_only(self, provider, adapter): + adapter.fail_service = _CreateConflict( + "Kubernetes create Service name is already in use" + ) + adapter.persist_service_on_failure = True + with pytest.raises(_CreateConflict, match="already in use"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 0 + assert provider._name is None + + def test_lost_pod_create_response_deletes_owned_pod(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.persist_pod_on_failure = True + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert provider._pod_created is False + assert provider._name is None + + def test_lost_service_create_response_deletes_owned_service( + self, provider, adapter + ): + adapter.fail_service = RuntimeError("apiserver timeout") + adapter.persist_service_on_failure = True + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 1 + assert provider._name is None + + def test_lost_create_response_leaves_foreign_object(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.foreign_pod_on_failure = True + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 0 + assert provider._name is None + + def test_unconfirmed_cleanup_keeps_name_for_retry(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.fail_read_pod = RuntimeError("read failed secret-body") + with pytest.raises(RuntimeError, match="apiserver timeout") as exc_info: + provider.start_container("echo-env:latest") + assert "secret-body" not in str(exc_info.value) + assert provider._pod_created is True + assert provider._name is not None + adapter.fail_read_pod = None + adapter.fail_delete_pod = None + provider.stop_container() + assert provider._pod_created is False + assert provider._name is None + assert adapter.deleted_pods == 1 def test_cleanup_failure_does_not_mask_original_error(self, provider, adapter): adapter.fail_service = RuntimeError("service failed") @@ -349,7 +437,7 @@ def test_failed_pod_raises_immediately_without_url(self, provider, adapter): provider.wait_for_ready(url, timeout_s=30) sleep.assert_not_called() assert url not in str(exc_info.value) - assert "svc.cluster.local" not in str(exc_info.value) + assert ".svc." not in str(exc_info.value) def test_unsafe_reason_is_omitted(self, provider, adapter): url = provider.start_container("echo-env:latest") @@ -377,7 +465,7 @@ def test_timeout_does_not_leak_url(self, provider, adapter): ): provider.wait_for_ready(url, timeout_s=5) assert url not in str(exc_info.value) - assert "svc.cluster.local" not in str(exc_info.value) + assert ".svc." not in str(exc_info.value) def test_mismatched_base_url_is_rejected_without_echoing_it(self, provider): provider.start_container("echo-env:latest") @@ -460,6 +548,8 @@ def create_namespaced_pod(self, namespace, body): calls["create_pod"].append((namespace, body)) if calls.get("create_pod_status") == 409: raise ApiException(status=409, body="secret-pod-body") + if calls.get("create_pod_status") == 404: + raise ApiException(status=404, body="secret-namespace-body") return body def create_namespaced_service(self, namespace, body): @@ -510,10 +600,9 @@ def test_incluster_config_is_preferred(self, monkeypatch): adapter = _DefaultKubernetesAdapter( namespace="openenv", kubeconfig=None, context=None ) - adapter.ensure_namespace() + assert adapter._namespace == "openenv" assert calls["incluster"] == 1 assert calls["kubeconfig"] == [] - assert calls["read_namespace"] == ["openenv"] def test_falls_back_to_kubeconfig_outside_cluster(self, monkeypatch): calls = _install_fake_kubernetes(monkeypatch, incluster_fails=True) @@ -587,14 +676,33 @@ def test_delete_failure_hides_api_body(self, monkeypatch): adapter.delete_pod("openenv-echo") assert "secret-delete-body" not in str(exc_info.value) - def test_missing_namespace_hides_api_body(self, monkeypatch): + def test_create_404_reports_missing_namespace_without_api_body(self, monkeypatch): calls = _install_fake_kubernetes(monkeypatch) - calls["namespace_status"] = 404 + calls["create_pod_status"] = 404 adapter = _DefaultKubernetesAdapter( namespace="openenv", kubeconfig=None, context=None ) - with pytest.raises(RuntimeError, match="does not exist") as exc_info: - adapter.ensure_namespace() + with pytest.raises(_NamespaceMissing, match="does not exist") as exc_info: + adapter.create_pod( + { + "metadata": { + "name": "openenv-echo", + "namespace": "openenv", + "labels": {}, + }, + "spec": { + "restart_policy": "Never", + "automount_service_account_token": False, + "containers": [ + { + "name": "env", + "image": "echo-env:latest", + "ports": [{"name": "http", "container_port": 8000}], + } + ], + }, + } + ) assert "secret-namespace-body" not in str(exc_info.value) def test_pod_status_keeps_reason_code_not_status_message(self, monkeypatch): From 7eed4395b50609b5c2a801350ab002fcaa27fbd4 Mon Sep 17 00:00:00 2001 From: Rohit Sharma Date: Fri, 2 Oct 2026 07:24:45 +0100 Subject: [PATCH 3/7] Keep unconfirmed Kubernetes ownership separate from a confirmed create. A failed ownership read was stored as a created flag, so stop_container() deleted that name without reading the labels again. A Pod create failure also marked the Service, which had never been created. Unknown ownership is now its own state, and only the resource whose create was attempted can enter it. stop_container() reads the labels again and deletes only on a match. A failed read keeps that state for another retry. --- .../containers/runtime/kubernetes_provider.py | 128 +++++++++++++----- tests/test_core/test_kubernetes_provider.py | 76 ++++++++++- 2 files changed, 165 insertions(+), 39 deletions(-) diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index 7ce4881ed4..2e5ebb2e58 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -389,8 +389,11 @@ class KubernetesProvider(ContainerProvider): Only one Pod is active per provider. Calling ``start_container`` again before ``stop_container()`` / ``close()`` raises ``RuntimeError`` rather - than orphaning the running Pod. A failed start deletes only the objects - this instance created. ``stop_container`` is safe to call more than once. + than orphaning the running Pod. A failed start deletes a resource only + after its OpenEnv ownership labels match. When a create response is lost + and that read fails, the resource stays unconfirmed: ``stop_container`` + reads the labels again and deletes only on a match. ``stop_container`` is + safe to call more than once. The Pod does not mount a service-account token unless ``automount_service_account_token=True`` is passed to ``start_container``. @@ -443,6 +446,8 @@ def __init__( self._service_port: int | None = None self._pod_created = False self._service_created = False + self._pod_unconfirmed = False + self._service_unconfirmed = False if _adapter is None: self._adapter: Any = _DefaultKubernetesAdapter( @@ -491,7 +496,7 @@ def start_container( Returns: `str`: ``http://{name}.{namespace}:{port}``. """ - if self._pod_created or self._service_created or self._base_url is not None: + if self._has_tracked_resources() or self._base_url is not None: raise RuntimeError( "KubernetesProvider already has an active Pod. Call " "stop_container() (or close()) before starting another — a " @@ -575,34 +580,42 @@ def _cleanup_after_failed_start(self, exc: BaseException) -> None: A 409 is someone else's name, and a 404 means the namespace is missing, so those do not probe. Any other create error may mean the - API server applied the object and the client lost the response. Those - are deleted only when the OpenEnv ownership labels match. If the read - or delete cannot be confirmed, the name and flag are kept so - ``stop_container()`` can retry. + API server applied the object and the client lost the response. Only + the resource whose create was attempted is checked. It is deleted only + when the OpenEnv ownership labels match. A failed read is recorded as + unconfirmed, which is not the same as a successful create: a later + ``stop_container()`` reads the labels again before deleting. """ uncertain = not isinstance(exc, (_CreateConflict, _NamespaceMissing)) + # Service create runs only after Pod create has succeeded. + service_was_attempted = self._pod_created self._release("pod", uncertain=uncertain) - self._release("service", uncertain=uncertain) - if not self._pod_created and not self._service_created: + self._release("service", uncertain=uncertain and service_was_attempted) + if not self._has_tracked_resources(): self._name = None self._base_url = None self._service_port = None def _release(self, kind: str, *, uncertain: bool) -> None: - flagged = self._pod_created if kind == "pod" else self._service_created - if not flagged and not uncertain: + if self._created(kind): + self._delete_tracked(kind) return - if not flagged: - owned = self._lookup_owned(kind) - if owned is None or not owned: - return - self._delete_kind(kind) + if not uncertain: + return + owned = self._lookup_owned(kind) + if owned is None: + self._set_unconfirmed(kind, True) + return + if not owned: + return + self._set_created(kind, True) + self._delete_tracked(kind) def _lookup_owned(self, kind: str) -> Optional[bool]: """Return whether *kind* at ``_name`` is ours. - ``None`` means the read itself failed. The flag is then set so a later - ``stop_container()`` retries the delete. + ``None`` means the read itself failed. Callers record that as + unconfirmed. They do not treat it as a confirmed create. """ if self._name is None: return False @@ -612,13 +625,16 @@ def _lookup_owned(self, kind: str) -> Optional[bool]: else: labels = self._adapter.service_labels(self._name) except Exception: - self._set_created(kind, True) return None if labels is None: return False return _owned_by_provider(labels, self._name) - def _delete_kind(self, kind: str) -> None: + def _delete_tracked(self, kind: str) -> None: + """Delete a resource already known to be ours. + + A failed delete stays confirmed so ``stop_container()`` can retry it. + """ if self._name is None: return try: @@ -631,38 +647,54 @@ def _delete_kind(self, kind: str) -> None: return self._set_created(kind, False) + def _created(self, kind: str) -> bool: + if kind == "pod": + return self._pod_created + return self._service_created + def _set_created(self, kind: str, created: bool) -> None: if kind == "pod": self._pod_created = created else: self._service_created = created + def _unconfirmed(self, kind: str) -> bool: + if kind == "pod": + return self._pod_unconfirmed + return self._service_unconfirmed + + def _set_unconfirmed(self, kind: str, unconfirmed: bool) -> None: + if kind == "pod": + self._pod_unconfirmed = unconfirmed + else: + self._service_unconfirmed = unconfirmed + + def _has_tracked_resources(self) -> bool: + return ( + self._pod_created + or self._service_created + or self._pod_unconfirmed + or self._service_unconfirmed + ) + def stop_container(self) -> None: """Delete the Pod and Service this provider created. Safe when nothing was started and safe to call twice. A 404 from the - API is treated as already deleted. Objects this instance did not - create are left in place, including a name that was already taken. + API is treated as already deleted. A confirmed resource is deleted by + name. An unconfirmed one is deleted only after a fresh ownership-label + match. If that read fails, the unconfirmed state is kept for another + retry. Objects this instance does not own are left in place, including + a name that was already taken. """ - if self._name is None and not self._pod_created and not self._service_created: + if self._name is None and not self._has_tracked_resources(): self._base_url = None self._service_port = None return - name = self._name errors: list[str] = [] - if self._pod_created and name is not None: - try: - self._adapter.delete_pod(name) - self._pod_created = False - except Exception: - errors.append("pod") - if self._service_created and name is not None: - try: - self._adapter.delete_service(name) - self._service_created = False - except Exception: - errors.append("service") + self._stop_kind("pod", errors) + self._stop_kind("service", errors) if errors: raise RuntimeError( "Failed to delete Kubernetes resources created by this provider. " @@ -672,6 +704,30 @@ def stop_container(self) -> None: self._base_url = None self._service_port = None + def _stop_kind(self, kind: str, errors: list[str]) -> None: + if self._name is None: + return + if self._unconfirmed(kind): + owned = self._lookup_owned(kind) + if owned is None: + errors.append(kind) + return + self._set_unconfirmed(kind, False) + if not owned: + return + self._set_created(kind, True) + if not self._created(kind): + return + try: + if kind == "pod": + self._adapter.delete_pod(self._name) + else: + self._adapter.delete_service(self._name) + except Exception: + errors.append(kind) + return + self._set_created(kind, False) + def close(self) -> None: """Stop the active Pod and Service. diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index 081d67bf68..440d1bce68 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -380,20 +380,90 @@ def test_lost_create_response_leaves_foreign_object(self, provider, adapter): assert adapter.deleted_pods == 0 assert provider._name is None - def test_unconfirmed_cleanup_keeps_name_for_retry(self, provider, adapter): + def test_unconfirmed_pod_is_not_treated_as_created(self, provider, adapter): adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.persist_pod_on_failure = True adapter.fail_read_pod = RuntimeError("read failed secret-body") + adapter.fail_read_service = RuntimeError("service read failed") with pytest.raises(RuntimeError, match="apiserver timeout") as exc_info: provider.start_container("echo-env:latest") assert "secret-body" not in str(exc_info.value) - assert provider._pod_created is True + assert provider._pod_created is False + assert provider._pod_unconfirmed is True + assert provider._service_created is False + assert provider._service_unconfirmed is False assert provider._name is not None + assert adapter.deleted_pods == 0 + assert adapter.deleted_services == 0 + + def test_unconfirmed_stop_rechecks_labels_before_delete(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.persist_pod_on_failure = True + adapter.fail_read_pod = RuntimeError("read failed") + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") adapter.fail_read_pod = None - adapter.fail_delete_pod = None provider.stop_container() + assert adapter.deleted_pods == 1 + assert provider._pod_unconfirmed is False assert provider._pod_created is False assert provider._name is None + + def test_unconfirmed_stop_leaves_foreign_object(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.foreign_pod_on_failure = True + adapter.fail_read_pod = RuntimeError("read failed") + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + adapter.fail_read_pod = None + provider.stop_container() + assert adapter.deleted_pods == 0 + assert provider._pod_unconfirmed is False + assert provider._name is None + + def test_unconfirmed_read_failure_on_stop_keeps_state(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.fail_read_pod = RuntimeError("read failed secret-body") + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + with pytest.raises(RuntimeError, match="retry") as exc_info: + provider.stop_container() + assert "secret-body" not in str(exc_info.value) + assert adapter.deleted_pods == 0 + assert provider._pod_unconfirmed is True + assert provider._pod_created is False + assert provider._name is not None + + def test_second_start_is_blocked_while_unconfirmed(self, provider, adapter): + adapter.fail_pod = RuntimeError("apiserver timeout") + adapter.fail_read_pod = RuntimeError("read failed") + with pytest.raises(RuntimeError, match="apiserver timeout"): + provider.start_container("echo-env:latest") + with pytest.raises(RuntimeError, match="already has an active Pod"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 0 + + def test_unconfirmed_service_retries_without_deleting_early( + self, provider, adapter + ): + adapter.fail_service = RuntimeError("apiserver timeout") + adapter.persist_service_on_failure = True + adapter.fail_read_service = RuntimeError("read failed secret-body") + with pytest.raises(RuntimeError, match="apiserver timeout") as exc_info: + provider.start_container("echo-env:latest") + assert "secret-body" not in str(exc_info.value) assert adapter.deleted_pods == 1 + assert adapter.deleted_services == 0 + assert provider._pod_created is False + assert provider._pod_unconfirmed is False + assert provider._service_created is False + assert provider._service_unconfirmed is True + assert provider._name is not None + adapter.fail_read_service = None + provider.stop_container() + assert adapter.deleted_services == 1 + assert provider._service_unconfirmed is False + assert provider._name is None def test_cleanup_failure_does_not_mask_original_error(self, provider, adapter): adapter.fail_service = RuntimeError("service failed") From aafa0f1d2fafe5580347aee2a5c44cde737106a4 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Mon, 19 Oct 2026 14:23:35 -0700 Subject: [PATCH 4/7] fix: preserve kubernetes cleanup on failure --- .../containers/runtime/kubernetes_provider.py | 10 +++++- tests/test_core/test_kubernetes_provider.py | 31 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index 2e5ebb2e58..324ab1298f 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -564,7 +564,8 @@ def start_container( ) ) self._service_created = True - except Exception as exc: + except BaseException as exc: + # Interrupts can also arrive after the API created a resource. # A cleanup failure must not mask the original error. try: self._cleanup_after_failed_start(exc) @@ -737,6 +738,13 @@ def close(self) -> None: """ self.stop_container() + def __exit__(self, exc_type, exc, tb) -> None: + try: + self.close() + except Exception: + if exc_type is None: + raise + def wait_for_ready(self, base_url: str, timeout_s: float = 120.0) -> None: """ Wait until the Pod is Running and ``GET /health`` returns 200. diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index 440d1bce68..3983416411 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -321,6 +321,37 @@ def test_not_reexported_from_runtime_package(self): class TestCleanup: + @pytest.mark.parametrize("resource", ["pod", "service"]) + @pytest.mark.parametrize("error", [KeyboardInterrupt, SystemExit]) + def test_interrupt_cleans_up_attempted_resources( + self, provider, adapter, resource, error + ): + setattr(adapter, f"fail_{resource}", error("interrupted create")) + setattr(adapter, f"persist_{resource}_on_failure", True) + with pytest.raises(error, match="interrupted create"): + provider.start_container("echo-env:latest") + assert adapter.deleted_pods == 1 + assert adapter.deleted_services == (1 if resource == "service" else 0) + assert provider._name is None + + def test_context_exit_preserves_start_failure(self, adapter): + adapter.fail_service = RuntimeError("original service failure") + adapter.fail_delete_pod = RuntimeError("cleanup failed") + provider = KubernetesProvider(namespace="openenv", _adapter=adapter) + with pytest.raises(RuntimeError, match="original service failure"): + with provider: + provider.start_container("echo-env:latest") + assert provider._pod_created + adapter.fail_delete_pod = None + provider.close() + assert provider._name is None + + def test_context_exit_surfaces_cleanup_failure_without_original(self, adapter): + adapter.fail_delete_pod = RuntimeError("cleanup failed") + with pytest.raises(RuntimeError, match="Failed to delete Kubernetes"): + with KubernetesProvider(namespace="openenv", _adapter=adapter) as provider: + provider.start_container("echo-env:latest") + def test_service_failure_deletes_only_the_pod(self, provider, adapter): adapter.fail_service = RuntimeError("service failed") with pytest.raises(RuntimeError, match="service failed"): From 14fbf7e3e24431c4bc2e910f89eef69db120b577 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Mon, 19 Oct 2026 14:30:16 -0700 Subject: [PATCH 5/7] fix: bound kubernetes readiness requests --- .../containers/runtime/kubernetes_provider.py | 36 ++++++---- tests/test_core/test_kubernetes_provider.py | 71 +++++++++++++++++-- 2 files changed, 89 insertions(+), 18 deletions(-) diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index 324ab1298f..200af17c79 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -202,7 +202,10 @@ def __init__( self._ApiException = ApiException self._namespace = namespace self._load_config(config, kubeconfig=kubeconfig, context=context) - self._api = client.CoreV1Api() + configuration = client.Configuration.get_default_copy() + # Readiness owns the retry deadline; SDK retries would multiply it. + configuration.retries = 0 + self._api = client.CoreV1Api(client.ApiClient(configuration=configuration)) def _load_config( self, config: Any, *, kubeconfig: Optional[str], context: Optional[str] @@ -283,10 +286,14 @@ def delete_service(self, name: str) -> None: except self._ApiException as exc: self._api_error(exc, "delete Service", missing_ok=True) - def pod_status(self, name: str) -> dict[str, Optional[str]]: + def pod_status( + self, name: str, timeout_s: float = 30.0 + ) -> dict[str, Optional[str]]: try: - pod = self._api.read_namespaced_pod(name, self._namespace) - except self._ApiException as exc: + pod = self._api.read_namespaced_pod( + name, self._namespace, _request_timeout=timeout_s + ) + except Exception as exc: self._api_error(exc, "read Pod", missing_ok=False) raise RuntimeError("Failed to read Kubernetes Pod") from None status = getattr(pod, "status", None) @@ -781,30 +788,35 @@ def wait_for_ready(self, base_url: str, timeout_s: float = 120.0) -> None: raise ValueError("timeout_s must be non-negative.") health_url = f"{expected}/health" - proxies = {"http": None, "https": None} - deadline = time.time() + timeout_s - while time.time() < deadline: - phase, reason = self._phase() + proxies = {"http": "", "https": "", "all": ""} + deadline = time.monotonic() + timeout_s + while (remaining := deadline - time.monotonic()) > 0: + phase, reason = self._phase(remaining) if phase in ("Failed", "Succeeded"): raise RuntimeError(self._ended_message(phase, reason)) + remaining = deadline - time.monotonic() + if remaining <= 0: + break if phase == "Running": try: - response = requests.get(health_url, timeout=2.0, proxies=proxies) + response = requests.get( + health_url, timeout=min(2.0, remaining), proxies=proxies + ) if response.status_code == 200: return except requests.RequestException: pass - time.sleep(0.5) + time.sleep(min(0.5, max(0.0, deadline - time.monotonic()))) raise TimeoutError(f"Kubernetes Pod did not become ready within {timeout_s}s.") - def _phase(self) -> tuple[str, Optional[str]]: + def _phase(self, timeout_s: float) -> tuple[str, Optional[str]]: if self._name is None: raise RuntimeError( "KubernetesProvider has no active Pod. Start the provider " "before waiting for readiness." ) - status = self._adapter.pod_status(self._name) + status = self._adapter.pod_status(self._name, timeout_s=timeout_s) phase = status.get("phase") if isinstance(status, Mapping) else None reason = status.get("reason") if isinstance(status, Mapping) else None if not isinstance(phase, str) or not _SAFE_REASON.fullmatch(phase): diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index 3983416411..6e6bc09724 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -90,7 +90,8 @@ def _labels(items, name): return dict(item["metadata"]["labels"]) return None - def pod_status(self, name): + def pod_status(self, name, timeout_s=30.0): + self.status_timeout = timeout_s return {"phase": self.phase, "reason": self.reason} @@ -523,7 +524,38 @@ def test_ready_when_running_and_health_200(self, provider): with patch("requests.get", return_value=MagicMock(status_code=200)) as get: provider.wait_for_ready(url, timeout_s=5) assert get.call_args.args[0] == f"{url}/health" - assert get.call_args.kwargs["proxies"] == {"http": None, "https": None} + assert get.call_args.kwargs["proxies"] == {"http": "", "https": "", "all": ""} + + def test_health_transport_ignores_ambient_proxies(self, provider, monkeypatch): + import requests + + url = provider.start_container("echo-env:latest") + monkeypatch.setenv("ALL_PROXY", "http://proxy.example:8123") + monkeypatch.setenv("HTTP_PROXY", "http://proxy.example:8123") + seen = [] + + def send(session, request, **kwargs): + seen.append(requests.utils.select_proxy(request.url, kwargs["proxies"])) + response = requests.Response() + response.status_code = 200 + return response + + with patch.object(requests.Session, "send", send): + provider.wait_for_ready(url, timeout_s=5) + assert seen == [""] + + def test_status_read_uses_remaining_deadline(self, provider, adapter): + url = provider.start_container("echo-env:latest") + with ( + patch( + "openenv.core.containers.runtime.kubernetes_provider.time.monotonic", + side_effect=[0.0, 1.0, 1.5], + ), + patch("requests.get", return_value=MagicMock(status_code=200)) as get, + ): + provider.wait_for_ready(url, timeout_s=2.0) + assert adapter.status_timeout == 1.0 + assert get.call_args.kwargs["timeout"] == 0.5 def test_failed_pod_raises_immediately_without_url(self, provider, adapter): url = provider.start_container("echo-env:latest") @@ -559,7 +591,7 @@ def test_timeout_does_not_leak_url(self, provider, adapter): with ( patch("requests.get", side_effect=requests.ConnectionError("refused")), patch( - "openenv.core.containers.runtime.kubernetes_provider.time.time", + "openenv.core.containers.runtime.kubernetes_provider.time.monotonic", side_effect=[0, 1, 100], ), pytest.raises(TimeoutError, match="did not become ready") as exc_info, @@ -639,6 +671,9 @@ def __init__(self, **kwargs): setattr(client_mod, name, type(name, (_Obj,), {})) class CoreV1Api: + def __init__(self, api_client): + calls["configuration"] = api_client.configuration + def read_namespace(self, name): calls["read_namespace"].append(name) if calls.get("namespace_status") == 404: @@ -667,8 +702,9 @@ def delete_namespaced_pod(self, name, namespace): def delete_namespaced_service(self, name, namespace): calls["delete_service"].append((name, namespace)) - def read_namespaced_pod(self, name, namespace): + def read_namespaced_pod(self, name, namespace, **kwargs): calls["read_pod"].append((name, namespace)) + calls["read_pod_kwargs"] = kwargs waiting = _Obj(reason="ImagePullBackOff", message="secret-status-text") return _Obj( status=_Obj( @@ -679,6 +715,10 @@ def read_namespaced_pod(self, name, namespace): ) ) + client_mod.Configuration = types.SimpleNamespace( + get_default_copy=lambda: types.SimpleNamespace(retries=None) + ) + client_mod.ApiClient = _Obj client_mod.CoreV1Api = CoreV1Api exceptions_mod = types.ModuleType("kubernetes.client.exceptions") exceptions_mod.ApiException = ApiException @@ -704,6 +744,7 @@ def test_incluster_config_is_preferred(self, monkeypatch): assert adapter._namespace == "openenv" assert calls["incluster"] == 1 assert calls["kubeconfig"] == [] + assert calls["configuration"].retries == 0 def test_falls_back_to_kubeconfig_outside_cluster(self, monkeypatch): calls = _install_fake_kubernetes(monkeypatch, incluster_fails=True) @@ -807,15 +848,33 @@ def test_create_404_reports_missing_namespace_without_api_body(self, monkeypatch assert "secret-namespace-body" not in str(exc_info.value) def test_pod_status_keeps_reason_code_not_status_message(self, monkeypatch): - _install_fake_kubernetes(monkeypatch) + calls = _install_fake_kubernetes(monkeypatch) adapter = _DefaultKubernetesAdapter( namespace="openenv", kubeconfig=None, context=None ) - status = adapter.pod_status("openenv-echo") + status = adapter.pod_status("openenv-echo", timeout_s=0.25) + assert calls["read_pod_kwargs"] == {"_request_timeout": 0.25} assert status["phase"] == "Pending" assert status["reason"] == "ImagePullBackOff" assert "secret-status-text" not in str(status) + def test_status_transport_error_does_not_expose_endpoint(self, monkeypatch): + _install_fake_kubernetes(monkeypatch) + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + with ( + patch.object( + adapter._api, + "read_namespaced_pod", + side_effect=TimeoutError("https://private-api.example/secret"), + ), + pytest.raises(RuntimeError, match="Failed to read Pod") as exc_info, + ): + adapter.pod_status("openenv-echo", timeout_s=0.25) + assert "secret" not in str(exc_info.value) + assert exc_info.value.__suppress_context__ + def test_missing_client_names_the_extra(self, monkeypatch): real_import = __import__ From ed3bf9da8c6d2773d8987d95e888f4cf88220632 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Mon, 19 Oct 2026 14:34:49 -0700 Subject: [PATCH 6/7] fix: retry transient kubernetes status reads --- .../containers/runtime/kubernetes_provider.py | 20 ++++++- tests/test_core/test_kubernetes_provider.py | 57 ++++++++++++++++++- 2 files changed, 75 insertions(+), 2 deletions(-) diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index 200af17c79..20067bae5a 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -160,6 +160,10 @@ class _CreateConflict(RuntimeError): """Creating a namespaced object returned 409: the name is already taken.""" +class _TransientStatusError(RuntimeError): + """A Pod status read can be retried within the readiness deadline.""" + + def _incluster_base_url(name: str, namespace: str, port: int) -> str: # *name* and *namespace* are already DNS-1123 labels and *port* is an int, # so this cannot be steered by a status field from the API server. @@ -289,11 +293,22 @@ def delete_service(self, name: str) -> None: def pod_status( self, name: str, timeout_s: float = 30.0 ) -> dict[str, Optional[str]]: + from urllib3.exceptions import HTTPError + try: pod = self._api.read_namespaced_pod( name, self._namespace, _request_timeout=timeout_s ) except Exception as exc: + status = getattr(exc, "status", None) + if ( + isinstance(exc, (TimeoutError, HTTPError)) + or status == 429 + or (isinstance(status, int) and 500 <= status < 600) + ): + raise _TransientStatusError( + "Kubernetes Pod status read failed" + ) from None self._api_error(exc, "read Pod", missing_ok=False) raise RuntimeError("Failed to read Kubernetes Pod") from None status = getattr(pod, "status", None) @@ -791,7 +806,10 @@ def wait_for_ready(self, base_url: str, timeout_s: float = 120.0) -> None: proxies = {"http": "", "https": "", "all": ""} deadline = time.monotonic() + timeout_s while (remaining := deadline - time.monotonic()) > 0: - phase, reason = self._phase(remaining) + try: + phase, reason = self._phase(remaining) + except _TransientStatusError: + phase, reason = "Unknown", None if phase in ("Failed", "Succeeded"): raise RuntimeError(self._ended_message(phase, reason)) remaining = deadline - time.monotonic() diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index 6e6bc09724..a474cec801 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -21,6 +21,7 @@ _CreateConflict, _DefaultKubernetesAdapter, _NamespaceMissing, + _TransientStatusError, KubernetesProvider, ) from openenv.core.containers.runtime.providers import ContainerProvider @@ -557,6 +558,35 @@ def test_status_read_uses_remaining_deadline(self, provider, adapter): assert adapter.status_timeout == 1.0 assert get.call_args.kwargs["timeout"] == 0.5 + def test_transient_status_error_retries_without_teardown(self, provider, adapter): + url = provider.start_container("echo-env:latest") + with ( + patch.object( + adapter, + "pod_status", + side_effect=[_TransientStatusError("retry"), {"phase": "Running"}], + ) as status, + patch("requests.get", return_value=MagicMock(status_code=200)) as get, + ): + provider.wait_for_ready(url, timeout_s=5) + assert status.call_count == 2 + get.assert_called_once() + assert adapter.deleted_pods == adapter.deleted_services == 0 + + def test_transient_status_errors_still_obey_deadline(self, provider, adapter): + url = provider.start_container("echo-env:latest") + with ( + patch.object( + adapter, "pod_status", side_effect=_TransientStatusError("retry") + ), + patch( + "openenv.core.containers.runtime.kubernetes_provider.time.monotonic", + side_effect=[0, 1, 5], + ), + pytest.raises(TimeoutError, match="did not become ready"), + ): + provider.wait_for_ready(url, timeout_s=5) + def test_failed_pod_raises_immediately_without_url(self, provider, adapter): url = provider.start_container("echo-env:latest") adapter.phase = "Failed" @@ -705,6 +735,8 @@ def delete_namespaced_service(self, name, namespace): def read_namespaced_pod(self, name, namespace, **kwargs): calls["read_pod"].append((name, namespace)) calls["read_pod_kwargs"] = kwargs + if calls.get("read_pod_status"): + raise ApiException(status=calls["read_pod_status"], body="secret-body") waiting = _Obj(reason="ImagePullBackOff", message="secret-status-text") return _Obj( status=_Obj( @@ -869,12 +901,35 @@ def test_status_transport_error_does_not_expose_endpoint(self, monkeypatch): "read_namespaced_pod", side_effect=TimeoutError("https://private-api.example/secret"), ), - pytest.raises(RuntimeError, match="Failed to read Pod") as exc_info, + pytest.raises( + _TransientStatusError, match="status read failed" + ) as exc_info, ): adapter.pod_status("openenv-echo", timeout_s=0.25) assert "secret" not in str(exc_info.value) assert exc_info.value.__suppress_context__ + @pytest.mark.parametrize("status", [429, 500, 503]) + def test_status_api_errors_are_retryable(self, monkeypatch, status): + calls = _install_fake_kubernetes(monkeypatch) + calls["read_pod_status"] = status + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + with pytest.raises(_TransientStatusError) as exc_info: + adapter.pod_status("openenv-echo") + assert "secret" not in str(exc_info.value) + + def test_status_permission_error_is_not_retryable(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["read_pod_status"] = 403 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + with pytest.raises(RuntimeError, match="Failed to read Pod") as exc_info: + adapter.pod_status("openenv-echo") + assert not isinstance(exc_info.value, _TransientStatusError) + def test_missing_client_names_the_extra(self, monkeypatch): real_import = __import__ From dacf113019983541a25c7de54a81ba7f03922c89 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Mon, 19 Oct 2026 14:38:53 -0700 Subject: [PATCH 7/7] fix: support kubernetes sdk timeout bounds --- .../containers/runtime/kubernetes_provider.py | 6 +++++- tests/test_core/test_kubernetes_provider.py | 21 ++++++++++++++++++- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/src/openenv/core/containers/runtime/kubernetes_provider.py b/src/openenv/core/containers/runtime/kubernetes_provider.py index 20067bae5a..94652b1b53 100644 --- a/src/openenv/core/containers/runtime/kubernetes_provider.py +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -296,8 +296,12 @@ def pod_status( from urllib3.exceptions import HTTPError try: + # Older supported SDKs ignore a float timeout. Split the budget + # between connect and read using their supported tuple form. pod = self._api.read_namespaced_pod( - name, self._namespace, _request_timeout=timeout_s + name, + self._namespace, + _request_timeout=(timeout_s / 2, timeout_s / 2), ) except Exception as exc: status = getattr(exc, "status", None) diff --git a/tests/test_core/test_kubernetes_provider.py b/tests/test_core/test_kubernetes_provider.py index a474cec801..be6e0fde1d 100644 --- a/tests/test_core/test_kubernetes_provider.py +++ b/tests/test_core/test_kubernetes_provider.py @@ -885,7 +885,7 @@ def test_pod_status_keeps_reason_code_not_status_message(self, monkeypatch): namespace="openenv", kubeconfig=None, context=None ) status = adapter.pod_status("openenv-echo", timeout_s=0.25) - assert calls["read_pod_kwargs"] == {"_request_timeout": 0.25} + assert calls["read_pod_kwargs"] == {"_request_timeout": (0.125, 0.125)} assert status["phase"] == "Pending" assert status["reason"] == "ImagePullBackOff" assert "secret-status-text" not in str(status) @@ -948,6 +948,25 @@ def guarded(name, globals=None, locals=None, fromlist=(), level=0): ) +def test_pod_status_bounds_real_sdk_transport(monkeypatch): + client = pytest.importorskip("kubernetes.client") + configuration = client.Configuration() + configuration.retries = 0 + with client.ApiClient(configuration=configuration) as api_client: + adapter = _DefaultKubernetesAdapter.__new__(_DefaultKubernetesAdapter) + adapter._namespace = "openenv" + adapter._api = client.CoreV1Api(api_client) + request = MagicMock(side_effect=TimeoutError("transport timeout")) + monkeypatch.setattr(api_client.rest_client.pool_manager, "request", request) + + with pytest.raises(_TransientStatusError): + adapter.pod_status("openenv-echo", timeout_s=0.25) + + timeout = request.call_args.kwargs["timeout"] + assert timeout is not None + assert timeout.connect_timeout + timeout.read_timeout <= 0.25 + + def test_module_import_does_not_import_kubernetes(): source = kubernetes_provider.__file__ assert source is not None