Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion plugins/nemo-deployments/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ dependencies = [

[project.optional-dependencies]
docker = ["docker>=7.0"]
k8s = ["kubernetes>=30.1.0"]
Comment thread
tylersbray marked this conversation as resolved.

[project.entry-points."nemo.services"]
deployments = "nemo_deployments_plugin.service:DeploymentsService"
Expand All @@ -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"]
Expand Down
Original file line number Diff line number Diff line change
@@ -1,80 +1,38 @@
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Docker resource naming and identity labels for orphan cleanup.

Resource names are derived from a human-readable prefix plus a deterministic
8-character hash. The hash is computed from ``{workspace}/{name}`` (see
``deployment_key``), not from the hyphen-joined display string, so ambiguous
pairs like ``("foo", "bar-baz")`` and ``("foo-bar", "baz")`` produce distinct
names even when their joined prefixes collide.

Naming logic is shared via ``nemo_platform_plugin.k8s_naming`` (plugins cannot
import ``nmp_common``). Orphan cleanup and idempotency rely on identity labels,
not container names alone.
"""

from nemo_deployments_plugin.constants import MANAGED_BY_LABEL
from nemo_platform_plugin.k8s_naming import k8s_safe_name, workspace_name_identity

MANAGED_BY_KEY = "managed-by"
DEPLOYMENT_WORKSPACE_LABEL = "nemo.nvidia.com/deployment-workspace"
DEPLOYMENT_NAME_LABEL = "nemo.nvidia.com/deployment-name"
RESTART_POLICY_LABEL = "nemo.nvidia.com/restart-policy"
CONFIG_NAME_LABEL = "nemo.nvidia.com/deployment-config"
VOLUME_WORKSPACE_LABEL = "nemo.nvidia.com/volume-workspace"
VOLUME_NAME_LABEL = "nemo.nvidia.com/volume-name"


def deployment_key(workspace: str, name: str) -> str:
"""Return the canonical identity string used for hashing and label keys."""
return workspace_name_identity(workspace, name)


def container_name(workspace: str, deployment_name: str) -> str:
"""Docker container name for a deployment (``dep-`` prefix, hashed identity)."""
return k8s_safe_name(
f"dep-{workspace}-{deployment_name}",
hash_input=deployment_key(workspace, deployment_name),
)


def docker_volume_name(workspace: str, volume_name: str) -> str:
"""Docker volume name for a deployment volume (``dep-vol-`` prefix, hashed identity)."""
return k8s_safe_name(
f"dep-vol-{workspace}-{volume_name}",
hash_input=deployment_key(workspace, volume_name),
)


BACKOFF_LIMIT_LABEL = "nemo.nvidia.com/backoff-limit"


def deployment_identity_labels(
workspace: str,
name: str,
restart_policy: str,
*,
config_name: str,
backoff_limit: int = 6,
) -> dict[str, str]:
return {
MANAGED_BY_KEY: MANAGED_BY_LABEL,
DEPLOYMENT_WORKSPACE_LABEL: workspace,
DEPLOYMENT_NAME_LABEL: name,
RESTART_POLICY_LABEL: restart_policy,
CONFIG_NAME_LABEL: config_name,
BACKOFF_LIMIT_LABEL: str(backoff_limit),
}


def volume_identity_labels(workspace: str, name: str) -> dict[str, str]:
return {
MANAGED_BY_KEY: MANAGED_BY_LABEL,
VOLUME_WORKSPACE_LABEL: workspace,
VOLUME_NAME_LABEL: name,
}


def managed_by_filter() -> dict[str, str | bool]:
return {"label": f"{MANAGED_BY_KEY}={MANAGED_BY_LABEL}"}
"""Docker resource naming and identity labels (re-exported from shared ``backends.labels``)."""

from nemo_deployments_plugin.backends.labels import (
Comment thread
tylersbray marked this conversation as resolved.
Outdated
BACKOFF_LIMIT_LABEL,
CONFIG_NAME_LABEL,
DEPLOYMENT_NAME_LABEL,
DEPLOYMENT_WORKSPACE_LABEL,
MANAGED_BY_KEY,
RESTART_POLICY_LABEL,
VOLUME_NAME_LABEL,
VOLUME_WORKSPACE_LABEL,
container_name,
deployment_identity_labels,
deployment_key,
docker_volume_name,
managed_by_filter,
volume_identity_labels,
)

__all__ = [
"BACKOFF_LIMIT_LABEL",
"CONFIG_NAME_LABEL",
"DEPLOYMENT_NAME_LABEL",
"DEPLOYMENT_WORKSPACE_LABEL",
"MANAGED_BY_KEY",
"RESTART_POLICY_LABEL",
"VOLUME_NAME_LABEL",
"VOLUME_WORKSPACE_LABEL",
"container_name",
"deployment_identity_labels",
"deployment_key",
"docker_volume_name",
"managed_by_filter",
"volume_identity_labels",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
# 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."""

_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:
pass

@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.")
Comment thread
tylersbray marked this conversation as resolved.

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.")
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
# 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.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from kubernetes import client

logger = logging.getLogger(__name__)


def build_api_client(*, kubeconfig_path: str | None = None) -> client.ApiClient:
"""Create an ``ApiClient`` for the given kubeconfig (in-cluster when path is unset)."""
from kubernetes import client, config

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: client.ApiClient | None = None
self._core_v1: client.CoreV1Api | None = None
self._apps_v1: client.AppsV1Api | None = None
self._batch_v1: client.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) -> client.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) -> client.CoreV1Api:
if self._core_v1 is None:
from kubernetes import client

self._core_v1 = client.CoreV1Api(self._api())
return self._core_v1

@property
def apps_v1(self) -> client.AppsV1Api:
if self._apps_v1 is None:
from kubernetes import client

self._apps_v1 = client.AppsV1Api(self._api())
return self._apps_v1

@property
def batch_v1(self) -> client.BatchV1Api:
if self._batch_v1 is None:
from kubernetes import client

self._batch_v1 = client.BatchV1Api(self._api())
return self._batch_v1
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# 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

from pydantic import BaseModel, Field


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,
Comment thread
tylersbray marked this conversation as resolved.
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.",
)
Loading
Loading