NVIDIA-NeMo · tylersbray · Jul 1, 2026 · Jul 1, 2026 · Jul 1, 2026 · Jul 1, 2026
@@ -14,6 +14,7 @@ dependencies = [
 
 [project.optional-dependencies]
 docker = ["docker>=7.0"]
+k8s = ["kubernetes>=30.1.0"]
 
 [project.entry-points."nemo.services"]
 deployments = "nemo_deployments_plugin.service:DeploymentsService"
@@ -33,7 +34,7 @@ nemo-platform = { workspace = true }
 nemo-platform-plugin = { workspace = true }
 
 [dependency-groups]
-dev = ["pytest>=8.3.4", "pytest-asyncio>=0.25.3", "httpx>=0.27", "fastapi>=0.115", "docker>=7.0"]
+dev = ["pytest>=8.3.4", "pytest-asyncio>=0.25.3", "httpx>=0.27", "fastapi>=0.115", "docker>=7.0", "kubernetes>=30.1.0"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

@@ -31,7 +31,15 @@
     validate_config_for_docker,
 )
 from nemo_deployments_plugin.backends.docker.gpu import GPUAllocationError, get_shared_gpu_pool
-from nemo_deployments_plugin.backends.docker.labels import (
+from nemo_deployments_plugin.backends.docker.ports import find_available_port
+from nemo_deployments_plugin.backends.docker.probes import check_readiness_probe, host_url_for_port
+from nemo_deployments_plugin.backends.docker.status import (
+    LOG_MAX_CHARS,
+    map_docker_state_to_starting,
+    map_exited_status,
+    missing_container_status,
+)
+from nemo_deployments_plugin.backends.labels import (
     BACKOFF_LIMIT_LABEL,
     CONFIG_NAME_LABEL,
     DEPLOYMENT_NAME_LABEL,
@@ -43,14 +51,6 @@
     deployment_key,
     managed_by_filter,
 )
-from nemo_deployments_plugin.backends.docker.ports import find_available_port
-from nemo_deployments_plugin.backends.docker.probes import check_readiness_probe, host_url_for_port
-from nemo_deployments_plugin.backends.docker.status import (
-    LOG_MAX_CHARS,
-    map_docker_state_to_starting,
-    map_exited_status,
-    missing_container_status,
-)
 from nemo_deployments_plugin.constants import MANAGED_BY_LABEL
 from nemo_deployments_plugin.entities import Container, Deployment, DeploymentConfig
 from nemo_deployments_plugin.types import Endpoint, RestartPolicy

@@ -7,7 +7,7 @@
 
 from typing import Any
 
-from nemo_deployments_plugin.backends.docker.labels import docker_volume_name
+from nemo_deployments_plugin.backends.labels import docker_volume_name
 from nemo_deployments_plugin.entities import Container, DeploymentConfig, DockerDeploymentConfig, VolumeMount
 from nemo_deployments_plugin.types import RestartPolicy
 

@@ -131,7 +131,7 @@ def parse_gpu_device_ids(device_requests: list[Any] | None) -> list[int]:
 
 def discover_managed_gpu_allocations(client: docker.DockerClient) -> dict[str, list[int]]:
     """Return workload_id -> GPU IDs for running deployment-managed containers."""
-    from nemo_deployments_plugin.backends.docker.labels import (
+    from nemo_deployments_plugin.backends.labels import (
         DEPLOYMENT_NAME_LABEL,
         DEPLOYMENT_WORKSPACE_LABEL,
         MANAGED_BY_KEY,

@@ -11,7 +11,7 @@
 import socket
 from typing import TYPE_CHECKING
 
-from nemo_deployments_plugin.backends.docker.labels import managed_by_filter
+from nemo_deployments_plugin.backends.labels import managed_by_filter
 
 import docker
 

@@ -10,7 +10,7 @@
 from typing import Any
 
 from nemo_deployments_plugin.backends.base import VolumeStatusUpdate
-from nemo_deployments_plugin.backends.docker.labels import docker_volume_name, volume_identity_labels
+from nemo_deployments_plugin.backends.labels import docker_volume_name, volume_identity_labels
 
 import docker
 

@@ -0,0 +1,110 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Kubernetes substrate backend for the deployments plugin (scaffold)."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from nemo_deployments_plugin.backends.base import (
+    BackendStatusUpdate,
+    DeploymentBackend,
+    LogResult,
+    VolumeStatusUpdate,
+)
+from nemo_deployments_plugin.backends.k8s.client import KubernetesClients
+from nemo_deployments_plugin.backends.k8s.config import K8sExecutorConfig
+
+logger = logging.getLogger(__name__)
+
+_K8S_INSTALL_HINT = (
+    "kubernetes package is required for K8sDeploymentBackend. "
+    "Install with: uv sync --package nemo-deployments-plugin --extra k8s"
+)
+
+
+class K8sDeploymentBackend(DeploymentBackend):
+    """Manage deployments and volumes as native Kubernetes objects.
+
+    Lifecycle methods not yet implemented raise ``NotImplementedError`` (not ``...``) so
+    accidental calls fail loudly during phased rollout; ``...`` is for ``@abstractmethod``
+    stubs on the ABC itself.
+    """
+
+    _clients: KubernetesClients
+
+    def init(self) -> None:
+        try:
+            import kubernetes  # noqa: F401
+        except ImportError as exc:
+            raise RuntimeError(_K8S_INSTALL_HINT) from exc
+
+        self._executor_config = K8sExecutorConfig.model_validate(self._config)
+        self._clients = KubernetesClients(
+            kubeconfig_path=self._executor_config.kubeconfig_path,
+            request_timeout=self._executor_config.request_timeout,
+        )
+        logger.debug(
+            "K8sDeploymentBackend initialized (default_namespace=%s)",
+            self._executor_config.default_namespace,
+        )
+
+    def shutdown(self) -> None:
+        if hasattr(self, "_clients"):
+            self._clients.close()
+
+    @property
+    def executor_config(self) -> K8sExecutorConfig:
+        return self._executor_config
+
+    @property
+    def clients(self) -> KubernetesClients:
+        return self._clients
+
+    async def create_deployment(
+        self,
+        *,
+        workspace: str,
+        name: str,
+        config_name: str,
+        labels: dict[str, str],
+        backend_config: dict[str, Any],
+    ) -> BackendStatusUpdate:
+        raise NotImplementedError("K8s create_deployment is implemented in a later phase.")
+
+    async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate:
+        raise NotImplementedError("K8s read_status is implemented in a later phase.")
+
+    async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpdate:
+        raise NotImplementedError("K8s delete_deployment is implemented in a later phase.")
+
+    async def list_managed_deployment_names(self) -> list[str]:
+        raise NotImplementedError("K8s list_managed_deployment_names is implemented in a later phase.")
+
+    async def get_logs(
+        self,
+        *,
+        workspace: str,
+        name: str,
+        tail: int = 100,
+    ) -> LogResult:
+        raise NotImplementedError("K8s get_logs is implemented in a later phase.")
+
+    async def create_volume(
+        self,
+        *,
+        workspace: str,
+        name: str,
+        size: str,
+        access_modes: list[str],
+        backend_config: dict[str, Any],
+    ) -> VolumeStatusUpdate:
+        raise NotImplementedError("K8s create_volume is implemented in a later phase.")
+
+    async def read_volume_status(self, *, workspace: str, name: str) -> VolumeStatusUpdate:
+        raise NotImplementedError("K8s read_volume_status is implemented in a later phase.")
+
+    async def delete_volume(self, workspace: str, name: str) -> VolumeStatusUpdate:
+        raise NotImplementedError("K8s delete_volume is implemented in a later phase.")
@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Kubernetes client bootstrap for the deployments plugin.
+
+Copied from the jobs service pattern; tagged for future extraction to a shared substrate lib.
+
+Imports are centralized in ``_kubernetes_modules()`` rather than hoisted to module scope so
+``registry`` can load without requiring the optional ``kubernetes`` package until a k8s
+executor is actually constructed.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from kubernetes.client import ApiClient, AppsV1Api, BatchV1Api, CoreV1Api
+
+logger = logging.getLogger(__name__)
+
+_kubernetes_modules_cache: tuple[Any, Any] | None = None
+
+
+def _kubernetes_modules() -> tuple[Any, Any]:
+    """Return ``(kubernetes.client, kubernetes.config)``, importing on first use."""
+    global _kubernetes_modules_cache
+    if _kubernetes_modules_cache is None:
+        from kubernetes import client, config
+
+        _kubernetes_modules_cache = (client, config)
+    return _kubernetes_modules_cache
+
+
+def build_api_client(*, kubeconfig_path: str | None = None) -> ApiClient:
+    """Create an ``ApiClient`` for the given kubeconfig (in-cluster when path is unset)."""
+    client, config = _kubernetes_modules()
+    configuration = client.Configuration()
+    if kubeconfig_path:
+        config.load_kube_config(config_file=kubeconfig_path, client_configuration=configuration)
+    else:
+        try:
+            config.load_incluster_config(client_configuration=configuration)
+        except config.ConfigException:
+            config.load_kube_config(client_configuration=configuration)
+    return client.ApiClient(configuration)
+
+
+class KubernetesClients:
+    """Lazy Kubernetes API clients with per-instance kubeconfig and request timeout."""
+
+    def __init__(self, *, kubeconfig_path: str | None = None, request_timeout: int = 60) -> None:
+        self._kubeconfig_path = kubeconfig_path
+        self._request_timeout = request_timeout
+        self._api_client: ApiClient | None = None
+        self._core_v1: CoreV1Api | None = None
+        self._apps_v1: AppsV1Api | None = None
+        self._batch_v1: BatchV1Api | None = None
+
+    @property
+    def request_timeout(self) -> int:
+        """Per-request timeout (seconds) for Kubernetes API calls in later phases."""
+        return self._request_timeout
+
+    def _api(self) -> ApiClient:
+        if self._api_client is None:
+            self._api_client = build_api_client(kubeconfig_path=self._kubeconfig_path)
+            logger.debug(
+                "Kubernetes ApiClient created (kubeconfig_path=%s, request_timeout=%s)",
+                self._kubeconfig_path,
+                self._request_timeout,
+            )
+        return self._api_client
+
+    @property
+    def core_v1(self) -> CoreV1Api:
+        if self._core_v1 is None:
+            client, _ = _kubernetes_modules()
+            self._core_v1 = client.CoreV1Api(self._api())
+        return self._core_v1
+
+    @property
+    def apps_v1(self) -> AppsV1Api:
+        if self._apps_v1 is None:
+            client, _ = _kubernetes_modules()
+            self._apps_v1 = client.AppsV1Api(self._api())
+        return self._apps_v1
+
+    @property
+    def batch_v1(self) -> BatchV1Api:
+        if self._batch_v1 is None:
+            client, _ = _kubernetes_modules()
+            self._batch_v1 = client.BatchV1Api(self._api())
+        return self._batch_v1
+
+    def close(self) -> None:
+        """Release the underlying ``ApiClient`` connection pool, if created.
+
+        ``CoreV1Api`` / ``AppsV1Api`` / ``BatchV1Api`` share the same ``ApiClient`` instance;
+        closing it invalidates the cached API wrappers (reset below).
+        """
+        if self._api_client is not None:
+            self._api_client.close()
+            self._api_client = None
+            self._core_v1 = None
+            self._apps_v1 = None
+            self._batch_v1 = None
@@ -0,0 +1,39 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Executor-level Kubernetes backend configuration."""
+
+from __future__ import annotations
+
+import re
+
+from pydantic import BaseModel, Field, field_validator
+
+_DNS_LABEL_PATTERN = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$")
+
+
+class K8sExecutorConfig(BaseModel):
+    """Knobs for a named k8s executor instance (not entity backend_config)."""
+
+    kubeconfig_path: str | None = Field(
+        default=None,
+        description="Path to kubeconfig file. When unset, uses in-cluster config or default kubeconfig.",
+    )
+    default_namespace: str = Field(
+        default="default",
+        min_length=1,
+        max_length=63,
+        description="Namespace for resources when entity backend_config.k8s.namespace is unset.",
+    )
+    request_timeout: int = Field(
+        default=60,
+        ge=1,
+        description="Kubernetes API client timeout in seconds.",
+    )
+
+    @field_validator("default_namespace")
+    @classmethod
+    def _validate_default_namespace(cls, value: str) -> str:
+        if not _DNS_LABEL_PATTERN.fullmatch(value):
+            raise ValueError("default_namespace must be a lowercase DNS-1123 label (alphanumeric, interior hyphens)")
+        return value