diff --git a/docs/source/getting-started.md b/docs/source/getting-started.md index d0284e90a..0c0b726c4 100644 --- a/docs/source/getting-started.md +++ b/docs/source/getting-started.md @@ -155,7 +155,7 @@ with EchoEnv(base_url="http://localhost:8000").sync() as client: result = client.reset() ``` -The same client also runs environments on cloud sandboxes (Daytona, Modal, Novita, Azure Container Apps, Hugging Face) through runtime providers. `HFSandboxProvider` ships with core OpenEnv and needs a Hugging Face token for an account that can run Jobs (the sandboxes are billed as Jobs): it runs an environment's Space image in a Hugging Face sandbox ([example](guides/runtime-providers#hfsandboxprovider)). See the [Runtime Providers guide](guides/runtime-providers) to pick one. +The same client also runs environments on cloud sandboxes (Daytona, Modal, Novita, Azure Container Apps, Hugging Face) and on a Kubernetes cluster through runtime providers. `HFSandboxProvider` ships with core OpenEnv and needs a Hugging Face token for an account that can run Jobs (the sandboxes are billed as Jobs): it runs an environment's Space image in a Hugging Face sandbox ([example](guides/runtime-providers#hfsandboxprovider)). `KubernetesProvider` is an optional extra (`pip install openenv[kubernetes]`). See the [Runtime Providers guide](guides/runtime-providers) to pick one. ## Next Steps diff --git a/docs/source/guides/runtime-providers.md b/docs/source/guides/runtime-providers.md index 2fa61196c..876ed973b 100644 --- a/docs/source/guides/runtime-providers.md +++ b/docs/source/guides/runtime-providers.md @@ -61,8 +61,7 @@ is a one-line change. | `ModalProvider` | Modal sandboxes | `pip install openenv[modal]` | ✅ | | `NovitaSandboxProvider` | Novita AI sandboxes | `pip install openenv[novita]` | ✅ | | `HFSandboxProvider` | Hugging Face sandboxes (billed as HF Jobs) | core | ✅ | - -A `KubernetesProvider` is planned but not available yet. +| `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`, @@ -237,8 +236,22 @@ Full example: [`examples/hf_sandbox_coding_env.py`](https://github.com/huggingfa ### 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://{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 + +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 f07c72688..fc0b67269 100644 --- a/docs/source/reference/core.md +++ b/docs/source/reference/core.md @@ -291,3 +291,5 @@ See [Evaluating with Environments](../tutorials/evaluation-inspect) for a worked [[autodoc]] openenv.core.containers.runtime.novita_provider.NovitaSandboxProvider [[autodoc]] openenv.core.containers.runtime.hf_sandbox_provider.HFSandboxProvider + +[[autodoc]] openenv.core.containers.runtime.kubernetes_provider.KubernetesProvider diff --git a/pyproject.toml b/pyproject.toml index 8603b3b84..e7e6f6245 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,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 0caa1bed7..0315d7d8f 100644 --- a/src/openenv/core/README.md +++ b/src/openenv/core/README.md @@ -165,14 +165,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 9a3ce7622..16caec4c8 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 000000000..94652b1b5 --- /dev/null +++ b/src/openenv/core/containers/runtime/kubernetes_provider.py @@ -0,0 +1,985 @@ +# 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 +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 + +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 + + +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.""" + + +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. + # ``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: + """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) + 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] + ) -> 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 _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 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 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) + 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, timeout_s: float = 30.0 + ) -> dict[str, Optional[str]]: + 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 / 2, timeout_s / 2), + ) + 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) + 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}:{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. + + 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 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``. + + 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 + self._pod_unconfirmed = False + self._service_unconfirmed = 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}:{port}``. + """ + 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 " + "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.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 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) + except Exception: + pass + raise + + 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. 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 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: + if self._created(kind): + self._delete_tracked(kind) + return + 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. Callers record that as + unconfirmed. They do not treat it as a confirmed create. + """ + 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: + return None + if labels is None: + return False + return _owned_by_provider(labels, self._name) + + 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: + 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 _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. 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._has_tracked_resources(): + self._base_url = None + self._service_port = None + return + + errors: list[str] = [] + self._stop_kind("pod", errors) + self._stop_kind("service", errors) + 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 _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. + + 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 __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. + + 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": "", "https": "", "all": ""} + deadline = time.monotonic() + timeout_s + while (remaining := deadline - time.monotonic()) > 0: + 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() + if remaining <= 0: + break + if phase == "Running": + try: + response = requests.get( + health_url, timeout=min(2.0, remaining), proxies=proxies + ) + if response.status_code == 200: + return + except requests.RequestException: + pass + 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, 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, 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): + 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 e76b0b651..c386fc2b4 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 @@ -662,19 +664,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 000000000..be6e0fde1 --- /dev/null +++ b/tests/test_core/test_kubernetes_provider.py @@ -0,0 +1,991 @@ +# 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 ( + _CreateConflict, + _DefaultKubernetesAdapter, + _NamespaceMissing, + _TransientStatusError, + 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.phase = "Running" + self.reason = None + self.fail_pod: Exception | None = None + self.fail_service: Exception | None = None + self.fail_delete_pod: Exception | None = None + 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) + + 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_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, timeout_s=30.0): + self.status_timeout = timeout_s + 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: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: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 == [] + + 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.pods == [] + + def test_missing_namespace_raises_and_creates_nothing(self, provider, adapter): + 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"): + 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: + @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"): + 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_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_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 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 + 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") + 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": "", "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_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" + 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." 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.monotonic", + 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." 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 __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: + 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") + if calls.get("create_pod_status") == 404: + raise ApiException(status=404, body="secret-namespace-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, **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( + phase="Pending", + container_statuses=[ + _Obj(state=_Obj(waiting=waiting, terminated=None)) + ], + ) + ) + + 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 + 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 + ) + 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) + _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_create_404_reports_missing_namespace_without_api_body(self, monkeypatch): + calls = _install_fake_kubernetes(monkeypatch) + calls["create_pod_status"] = 404 + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + 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): + calls = _install_fake_kubernetes(monkeypatch) + adapter = _DefaultKubernetesAdapter( + namespace="openenv", kubeconfig=None, context=None + ) + status = adapter.pod_status("openenv-echo", timeout_s=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) + + 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( + _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__ + + 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_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 + # 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 + )