diff --git a/docs/set-up/config-reference.mdx b/docs/set-up/config-reference.mdx index e0e3244df6..2b82b44f0e 100644 --- a/docs/set-up/config-reference.mdx +++ b/docs/set-up/config-reference.mdx @@ -593,6 +593,54 @@ models: max_restart_count: 5 # Whether this backend is enabled | default: False enabled: false + deployments_plugin: + default_executor: + docker_executor: + k8s_executor: + # Maximum seconds a deployment may stay PENDING before ERROR. | default: 7200 + pending_timeout_seconds: 7200 + # default: 5 + max_restart_count: 5 + default_storage_class: + # default: '200Gi' + default_pvc_size: 200Gi + # default: 'nvcr.io/nim/meta/llama-3.1-8b-instruct' + default_nimservice_image: nvcr.io/nim/meta/llama-3.1-8b-instruct + # default: '1.8.5' + default_nimservice_image_tag: 1.8.5 + # default: 'vllm/vllm-openai' + default_vllm_image: vllm/vllm-openai + # default: 'v0.8.5' + default_vllm_image_tag: v0.8.5 + # default: 1000 + default_user_id: 1000 + # default: 2000 + default_group_id: 2000 + # default: 2000 + default_vllm_user_id: 2000 + # default: 0 + default_vllm_group_id: 0 + # default: 30 + peft_refresh_interval: 30 + # default: 'nmp-api' + lora_sidecar_image_name: nmp-api + lora_sidecar_command: + - nemo + - services + - run + - --sidecars + - adapters + lora_sidecar_args: [] + # BusyBox image repository for LoRA cache init containers. Fully qualified so it resolves on runtimes that block docker.io short names. | default: 'docker.io/library/busybox' + busybox_image: docker.io/library/busybox + # BusyBox image tag for LoRA cache init containers. | default: 'latest' + busybox_image_tag: latest + # default: 5.0 + delete_wait_seconds: 5.0 + # default: 0.5 + delete_poll_seconds: 0.5 + # default: False + enabled: false none: # Whether this backend is enabled | default: False enabled: false diff --git a/packages/nemo_platform/pyproject.toml b/packages/nemo_platform/pyproject.toml index 1147a71a50..8b6b43b356 100644 --- a/packages/nemo_platform/pyproject.toml +++ b/packages/nemo_platform/pyproject.toml @@ -214,6 +214,7 @@ models-service = [ "kubernetes>=31.0.0", "urllib3>=2.7.0", "nmp-common", + "nemo-platform-plugin", "docker>=7.1.0", "tenacity>=8.5.0", ] diff --git a/services/core/models/pyproject.toml b/services/core/models/pyproject.toml index 6dacc22d2a..dffae8ccc2 100644 --- a/services/core/models/pyproject.toml +++ b/services/core/models/pyproject.toml @@ -23,6 +23,8 @@ dependencies = [ "kubernetes>=31.0.0", "urllib3>=2.7.0", "nmp-common", + "nemo-deployments-plugin", + "nemo-platform-plugin", "docker>=7.1.0", "tenacity>=8.5.0", # Retry logic for Docker backend operations ] @@ -60,4 +62,6 @@ dev = [ [tool.uv.sources] nmp-common = { workspace = true } +nemo-deployments-plugin = { workspace = true } +nemo-platform-plugin = { workspace = true } models = { workspace = true } diff --git a/services/core/models/src/nmp/core/models/config.py b/services/core/models/src/nmp/core/models/config.py index b29eb23057..63bc96c3d2 100644 --- a/services/core/models/src/nmp/core/models/config.py +++ b/services/core/models/src/nmp/core/models/config.py @@ -8,6 +8,7 @@ from nmp.common.config import Runtime, create_service_config_class, get_platform_config, get_service_config from nmp.core.models.controllers.backends.registry import ( BackendConfig, + DeploymentsPluginBackendConfigModel, DockerBackendConfigModel, K8sNimOperatorBackendConfigModel, NoneBackendConfigModel, @@ -163,12 +164,14 @@ class ParallelismConfig(BaseModel): # Backend and controller configuration # ----------------------------------------------------------------------------- -BackendName = Literal["docker", "nim_operator"] +BackendName = Literal["docker", "nim_operator", "deployments_plugin", "none"] -# Map backend names to their config model classes +# Map backend names to their config model classes. +# ``none`` is the no-op substrate used when platform.runtime is ``none``. BACKEND_CONFIG_MODELS: dict[str, type[BackendConfig]] = { "docker": DockerBackendConfigModel, "nim_operator": K8sNimOperatorBackendConfigModel, + "deployments_plugin": DeploymentsPluginBackendConfigModel, "none": NoneBackendConfigModel, } diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/__init__.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/__init__.py new file mode 100644 index 0000000000..88c781b2ee --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/__init__.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Models backend configuration for the deployments-plugin substrate. + +Only the config model is exported from this package. The concrete +``DeploymentsPluginServiceBackend`` lives in ``backend`` and is imported +lazily from ``registry`` so the models service wheel does not require +``nemo_deployments_plugin`` unless that backend is selected. +""" + +from .config import DeploymentsPluginBackendConfigModel as DeploymentsPluginBackendConfigModel + +__all__ = ["DeploymentsPluginBackendConfigModel"] diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/backend.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/backend.py new file mode 100644 index 0000000000..0f75cef113 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/backend.py @@ -0,0 +1,241 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Models ServiceBackend backed by nemo-deployments plugin entities.""" + +import asyncio +import logging +import time +from typing import Any + +from nemo_deployments_plugin.entities import Deployment, DeploymentConfig, Prerequisite, Volume +from nemo_platform import AsyncNeMoPlatform +from nemo_platform.resources.entities import AsyncEntitiesResource +from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError +from nemo_platform_plugin.sdk_provider import get_async_platform_sdk +from nmp.common.config import Runtime +from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER +from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate, ServiceBackend +from nmp.core.models.controllers.backends.common import deployment_elapsed_seconds +from nmp.core.models.controllers.backends.deployments_plugin.compiler import compile_model_deployment +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig +from nmp.core.models.controllers.backends.deployments_plugin.executor import executor_for_runtime +from nmp.core.models.controllers.backends.deployments_plugin.naming import entity_names +from nmp.core.models.controllers.backends.deployments_plugin.resolve import resolve_plugin_deployment +from nmp.core.models.controllers.backends.deployments_plugin.status import aggregate_status, apply_pending_timeout +from nmp.core.models.controllers.backends.engine import ENGINE_GENERIC, config_engine +from nmp.core.models.controllers.context import ModelContext + +logger = logging.getLogger(__name__) + +_DEPLOYMENT_WORKSPACE_LABEL = "nmp.nvidia.com/deployment-workspace" +_DEPLOYMENT_NAME_LABEL = "nmp.nvidia.com/deployment-name" +_MODELS_ROLE_LABEL = "nmp.nvidia.com/models-role" + + +class DeploymentsPluginServiceBackend(ServiceBackend): + """Compile model deployments into Volume and Deployment plugin entities.""" + + def __init__(self, nmp_sdk: AsyncNeMoPlatform, config: dict[str, Any], huggingface_model_puller: str) -> None: + self._backend_config: DeploymentsPluginConfig | None = None + self._entities: NemoEntitiesClient | None = None + self._huggingface_model_puller = huggingface_model_puller + super().__init__(nmp_sdk, config) + + def init(self) -> None: + self._backend_config = DeploymentsPluginConfig(**self._config) + + def shutdown(self) -> None: + self._entities = None + + def _entity_client(self) -> NemoEntitiesClient: + if self._entities is None: + sdk = get_async_platform_sdk(as_service="models", internal=True) + self._entities = NemoEntitiesClient(AsyncEntitiesResource(sdk)) + return self._entities + + @property + def _cfg(self) -> DeploymentsPluginConfig: + assert self._backend_config is not None + return self._backend_config + + async def create_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate: + """Create plugin substrate entities in volume, puller, server order. + + Tears down any leftover substrate first so drift recovery (LOST → recreate) + does not collide with orphaned Volume / DeploymentConfig / Deployment entities. + """ + resolved = resolve_plugin_deployment(ctx, self._huggingface_model_puller) + if resolved.runtime == Runtime.NONE: + return DeploymentStatusUpdate( + status="UNKNOWN", status_message="Deployments plugin is unavailable for runtime none." + ) + lora_enabled = resolved.view.lora_enabled and config_engine(resolved.config) != ENGINE_GENERIC + if resolved.runtime == Runtime.DOCKER and lora_enabled: + # Fail fast: the plugin docker runtime is single-container today, so a + # LoRA deployment (server + adapters sidecar) cannot run there yet. + return DeploymentStatusUpdate( + status="ERROR", + status_message=( + "LoRA serving is not supported on the docker runtime yet " + "(deployments-plugin docker is single-container). Deploy LoRA " + "models on the kubernetes runtime instead." + ), + ) + teardown = await self.delete_model_deployment(resolved.deployment.workspace, resolved.deployment.name) + if teardown.status == "DELETING": + return DeploymentStatusUpdate( + status="PENDING", + status_message="Waiting for prior deployments-plugin substrate teardown before recreate.", + ) + executor = executor_for_runtime(self._cfg, resolved.runtime) + if executor is None: + return DeploymentStatusUpdate( + status="ERROR", + status_message=( + "No deployments-plugin executor configured for the current runtime. " + "Set docker_executor, k8s_executor, or default_executor under " + "models.controller.backends.deployments_plugin." + ), + error_details={ + "reason": "executor_not_configured", + "runtime": resolved.runtime.value, + }, + ) + try: + compiled = compile_model_deployment(resolved, self._cfg) + entities = self._entity_client() + if compiled.volume is not None: + await entities.create(compiled.volume) + if compiled.scratch_volume is not None: + await entities.create(compiled.scratch_volume) + if compiled.puller_config is not None: + await entities.create(compiled.puller_config) + await entities.create( + Deployment( + name=compiled.names.puller, + workspace=resolved.deployment.workspace, + deployment_config=compiled.names.puller, + executor=executor, + desired_state="READY", + status="PENDING", + ) + ) + await entities.create(compiled.server_config) + await entities.create( + Deployment( + name=compiled.names.server, + workspace=resolved.deployment.workspace, + deployment_config=compiled.names.server, + executor=executor, + desired_state="READY", + status="PENDING", + prerequisites=( + [Prerequisite(deployment_name=compiled.names.puller, condition="succeeded")] + if compiled.puller_prerequisite + else [] + ), + ) + ) + except Exception as exc: + await self._rollback_create(ctx) + return DeploymentStatusUpdate( + status="ERROR", + status_message=f"Unable to create deployments-plugin entities: {exc}", + error_details={"error": str(exc)}, + ) + return DeploymentStatusUpdate(status="PENDING", status_message="Created deployments-plugin entities.") + + async def _rollback_create(self, ctx: ModelContext) -> None: + """Best-effort controlled teardown after a partial create failure.""" + if ctx.model_deployment is None: + return + try: + await self.delete_model_deployment(ctx.model_deployment.workspace, ctx.model_deployment.name) + except Exception: + logger.warning( + "Failed to roll back deployments-plugin substrate after a create failure; orphaned entities may remain", + extra={ + "workspace": ctx.model_deployment.workspace, + "deployment_name": ctx.model_deployment.name, + }, + exc_info=True, + ) + + async def get_model_deployment_status(self, ctx: ModelContext) -> DeploymentStatusUpdate: + if ctx.model_deployment is None: + return DeploymentStatusUpdate(status="UNKNOWN", status_message="Model deployment unavailable.") + names = entity_names(ctx.model_deployment.name) + server = await self._get_optional(Deployment, ctx.model_deployment.workspace, names.server) + puller = await self._get_optional(Deployment, ctx.model_deployment.workspace, names.puller) + volume = await self._get_optional(Volume, ctx.model_deployment.workspace, names.volume) + result = aggregate_status(volume, puller, server, previously_ready=ctx.model_deployment.status == "READY") + elapsed = deployment_elapsed_seconds(ctx.model_deployment) + return apply_pending_timeout( + result, + elapsed_seconds=elapsed, + timeout_seconds=self._cfg.pending_timeout_seconds, + deployment_name=ctx.model_deployment.name, + ) + + async def update_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate: + del ctx + return DeploymentStatusUpdate(status="ERROR", status_message="Update via recreate not yet supported.") + + async def delete_model_deployment(self, workspace: str, name: str) -> DeploymentStatusUpdate: + """Stop deployments, wait for each to disappear, then remove their configs.""" + names = entity_names(name) + for deployment_name, config_name in ((names.server, names.server), (names.puller, names.puller)): + if not await self._delete_deployment_and_config(workspace, deployment_name, config_name): + return DeploymentStatusUpdate( + status="DELETING", status_message="Waiting for plugin deployment teardown." + ) + for volume_name in (names.scratch, names.volume): + try: + await self._entity_client().delete(Volume, name=volume_name, workspace=workspace) + except NemoEntityNotFoundError: + pass + return DeploymentStatusUpdate(status="DELETED", status_message="Deleted deployments-plugin entities.") + + async def _delete_deployment_and_config(self, workspace: str, deployment_name: str, config_name: str) -> bool: + deployment = await self._get_optional(Deployment, workspace, deployment_name) + if deployment is not None: + if deployment.status != "DELETING": + deployment.status = "DELETING" + deployment.desired_state = "STOPPED" + await self._entity_client().update(deployment) + if not await self._wait_for_deployment_gone(workspace, deployment_name): + return False + try: + await self._entity_client().delete(DeploymentConfig, name=config_name, workspace=workspace) + except NemoEntityNotFoundError: + pass + return True + + async def _wait_for_deployment_gone(self, workspace: str, name: str) -> bool: + deadline = time.monotonic() + self._cfg.delete_wait_seconds + while time.monotonic() < deadline: + if await self._get_optional(Deployment, workspace, name) is None: + return True + await asyncio.sleep(self._cfg.delete_poll_seconds) + return False + + async def _get_optional(self, entity_type: type[Any], workspace: str, name: str) -> Any | None: + try: + return await self._entity_client().get(entity_type, name=name, workspace=workspace) + except NemoEntityNotFoundError: + return None + + async def list_managed_deployment_names(self) -> list[str]: + # Labels live on immutable DeploymentConfig entities; deployments-plugin + # does not currently mirror them onto Deployment. + result = await self._entity_client().list(DeploymentConfig, workspace="-") + names = { + f"{config.labels[_DEPLOYMENT_WORKSPACE_LABEL]}/{config.labels[_DEPLOYMENT_NAME_LABEL]}" + for config in result.data + if config.labels.get(MODEL_MANAGED_BY_LABEL) == MODEL_MANAGED_BY_MODELS_CONTROLLER + and config.labels.get(_MODELS_ROLE_LABEL) == "server" + and _DEPLOYMENT_WORKSPACE_LABEL in config.labels + and _DEPLOYMENT_NAME_LABEL in config.labels + } + return sorted(names) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/compiler.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/compiler.py new file mode 100644 index 0000000000..549f2ac01d --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/compiler.py @@ -0,0 +1,265 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Compile ModelDeployments into deployments-plugin entity specifications.""" + +from dataclasses import dataclass + +from nemo_deployments_plugin.entities import ( + Container, + ContainerPort, + DeploymentConfig, + EnvVar, + HTTPGetAction, + K8sVolumeConfig, + Probe, + Volume, + VolumeBackendConfig, + VolumeMount, +) +from nemo_platform_plugin.jobs.image import get_qualified_image +from nmp.common.config import Runtime +from nmp.core.models.app import ModelWeightsType +from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig +from nmp.core.models.controllers.backends.deployments_plugin.naming import EntityNames, entity_names +from nmp.core.models.controllers.backends.deployments_plugin.resolve import ResolvedPluginDeployment +from nmp.core.models.controllers.backends.engine import ( + ENGINE_GENERIC, + ENGINE_NIM, + ENGINE_VLLM, + config_engine, + resolve_health_path, +) +from nmp.core.models.controllers.backends.generic_compiler import ( + compile_generic_args, + compile_generic_env_vars, + resolve_generic_image, +) +from nmp.core.models.controllers.backends.vllm_compiler import ( + MODEL_STORE_PATH, + compile_vllm_args, + compile_vllm_env_vars, + resolve_vllm_image, +) + +_WEIGHTS_MOUNT = "/model-store" +_SCRATCH_MOUNT = "/scratch" +_LORA_MOUNT = "/scratch/loras" +_SCRATCH_VOLUME_SIZE = "1Gi" + + +@dataclass(frozen=True) +class CompiledModelDeployment: + """The plugin entities and dependency metadata for one model deployment.""" + + names: EntityNames + volume: Volume | None + scratch_volume: Volume | None + puller_config: DeploymentConfig | None + server_config: DeploymentConfig + puller_prerequisite: bool + + +def _labels(resolved: ResolvedPluginDeployment, engine: str, role: str) -> dict[str, str]: + return { + MODEL_MANAGED_BY_LABEL: MODEL_MANAGED_BY_MODELS_CONTROLLER, + "nmp.nvidia.com/deployment-workspace": resolved.deployment.workspace, + "nmp.nvidia.com/deployment-name": resolved.deployment.name, + "nmp.nvidia.com/models-role": role, + "nmp.nvidia.com/engine": engine, + } + + +def _image(name: str, tag: str) -> str: + return name if "@" in name or name.endswith(f":{tag}") else f"{name}:{tag}" + + +def _busybox_image(config: DeploymentsPluginConfig) -> str: + return _image(config.busybox_image, config.busybox_image_tag) + + +def _weighted(resolved: ResolvedPluginDeployment, engine: str) -> bool: + is_files = resolved.weights_type == ModelWeightsType.FILES_SERVICE + if engine == ENGINE_VLLM: + return is_files or bool(resolved.model_namespace and resolved.model_name) + if engine == ENGINE_NIM: + return is_files + return is_files or bool(resolved.model_entity and resolved.model_entity.fileset) + + +def _env(values: dict[str, str]) -> list[EnvVar]: + return [EnvVar(name=name, value=value) for name, value in values.items()] + + +def _lora_sidecar( + resolved: ResolvedPluginDeployment, + *, + engine: str, + config: DeploymentsPluginConfig, + names: EntityNames, + weighted: bool, +) -> Container: + """Build the adapters sidecar with the same env contract as existing backends.""" + entity_workspace = resolved.model_entity.workspace if resolved.model_entity else resolved.deployment.workspace + entity_name = resolved.model_entity.name if resolved.model_entity else resolved.deployment.name + sidecar_env = { + "NIM_PEFT_SOURCE": _LORA_MOUNT, + "NIM_PEFT_REFRESH_INTERVAL": str(config.peft_refresh_interval), + "NMP_MODEL_ENTITY_WORKSPACE": entity_workspace, + "NMP_MODEL_ENTITY_NAME": entity_name, + } + if engine == ENGINE_VLLM: + sidecar_env["VLLM_LORA_BASE_MODEL_OVERRIDE"] = MODEL_STORE_PATH + mounts = [VolumeMount(name=names.scratch, mountPath=_SCRATCH_MOUNT)] + if weighted: + mounts.append(VolumeMount(name=names.volume, mountPath=_WEIGHTS_MOUNT, readOnly=True)) + return Container( + name="lora-adapters", + image=get_qualified_image(config.lora_sidecar_image_name), + command=config.lora_sidecar_command, + args=config.lora_sidecar_args, + env=_env(sidecar_env), + volumeMounts=mounts, + restartPolicy="Always", + ) + + +def compile_model_deployment( + resolved: ResolvedPluginDeployment, config: DeploymentsPluginConfig +) -> CompiledModelDeployment: + """Compile volume, puller, and always-on serving config specifications.""" + engine = config_engine(resolved.config) + if engine not in {ENGINE_NIM, ENGINE_VLLM, ENGINE_GENERIC}: + raise ValueError(f"Unsupported engine {engine!r}.") + names = entity_names(resolved.deployment.name) + weighted = _weighted(resolved, engine) + lora_enabled = resolved.view.lora_enabled and engine != ENGINE_GENERIC + volume = None + scratch_volume = None + puller_config = None + if weighted: + volume = Volume( + name=names.volume, + workspace=resolved.deployment.workspace, + size=resolved.view.disk_size or config.default_pvc_size, + backendConfig=VolumeBackendConfig(k8s=K8sVolumeConfig(storageClass=config.default_storage_class)), + ) + puller_env = {"HF_ENDPOINT": resolved.files_hf_url} + puller_args = ["download", f"{resolved.model_namespace}/{resolved.model_name}", "--local-dir", _WEIGHTS_MOUNT] + if resolved.model_revision: + puller_args.extend(["--revision", resolved.model_revision]) + puller = Container( + name="weight-puller", + image=resolved.huggingface_model_puller, + command=["hf"], + args=puller_args, + env=_env(puller_env), + volumeMounts=[VolumeMount(name=names.volume, mountPath=_WEIGHTS_MOUNT)], + ) + puller_config = DeploymentConfig( + name=names.puller, + workspace=resolved.deployment.workspace, + containers=[puller], + labels=_labels(resolved, engine, "puller"), + restartPolicy="OnFailure", + backoffLimit=config.max_restart_count, + ) + + if engine == ENGINE_VLLM: + image_name, image_tag = resolve_vllm_image( + resolved.view, config.default_vllm_image, config.default_vllm_image_tag + ) + args = compile_vllm_args(resolved.view, resolved.model_entity) + env = compile_vllm_env_vars(resolved.view) + elif engine == ENGINE_NIM: + image_name, image_tag = ( + resolved.view.image_name or config.default_nimservice_image, + resolved.view.image_tag or config.default_nimservice_image_tag, + ) + args = list(resolved.view.additional_args or []) + env = dict(resolved.view.additional_envs or {}) + env.update({"NIM_MODEL_NAME": _WEIGHTS_MOUNT, "NIM_MODEL_PATH": _WEIGHTS_MOUNT}) + if resolved.model_name: + env["NIM_SERVED_MODEL_NAME"] = ( + f"{resolved.model_namespace}/{resolved.model_name}" if resolved.model_namespace else resolved.model_name + ) + if lora_enabled: + env["NIM_PEFT_SOURCE"] = _LORA_MOUNT + env["NIM_PEFT_REFRESH_INTERVAL"] = str(config.peft_refresh_interval) + else: + image_name, image_tag = resolve_generic_image(resolved.view) + args = compile_generic_args(resolved.view) + env = compile_generic_env_vars(resolved.view) + + mounts: list[VolumeMount] = [] + if weighted: + mounts.append(VolumeMount(name=names.volume, mountPath=_WEIGHTS_MOUNT, readOnly=True)) + init_containers: list[Container] = [] + server_config_containers: list[Container] + if lora_enabled: + scratch_volume = Volume( + name=names.scratch, + workspace=resolved.deployment.workspace, + size=_SCRATCH_VOLUME_SIZE, + backendConfig=VolumeBackendConfig(k8s=K8sVolumeConfig(storageClass=config.default_storage_class)), + ) + mounts.append(VolumeMount(name=names.scratch, mountPath=_SCRATCH_MOUNT)) + # Ensure the LoRA cache dir exists before the server/sidecar start. + init_containers.append( + Container( + name="lora-cache-init", + image=_busybox_image(config), + command=["sh", "-c", f"mkdir -p {_LORA_MOUNT} && chmod -R 777 {_LORA_MOUNT}"], + volumeMounts=[VolumeMount(name=names.scratch, mountPath=_SCRATCH_MOUNT)], + ) + ) + lora = _lora_sidecar(resolved, engine=engine, config=config, names=names, weighted=weighted) + server = Container( + name="server", + image=_image(image_name, image_tag), + args=args, + env=_env(env), + ports=[ContainerPort(name="http", containerPort=8000)], + volumeMounts=mounts, + readinessProbe=Probe(httpGet=HTTPGetAction(path=resolve_health_path(engine, resolved.view), port=8000)), + ) + if resolved.runtime == Runtime.DOCKER: + # Docker v1 is single-container today; emit a second container so the + # shape matches the locked design for when the plugin docker backend + # accepts multi-container DeploymentConfigs. In practice the backend + # fails fast on docker + LoRA before reaching create (see + # DeploymentsPluginServiceBackend.create_model_deployment). + server_config_containers = [server, lora] + else: + server_config_containers = [server] + init_containers.append(lora) + else: + server_config_containers = [ + Container( + name="server", + image=_image(image_name, image_tag), + args=args, + env=_env(env), + ports=[ContainerPort(name="http", containerPort=8000)], + volumeMounts=mounts, + readinessProbe=Probe(httpGet=HTTPGetAction(path=resolve_health_path(engine, resolved.view), port=8000)), + ) + ] + + server_config = DeploymentConfig( + name=names.server, + workspace=resolved.deployment.workspace, + containers=server_config_containers, + initContainers=init_containers, + labels=_labels(resolved, engine, "server"), + restartPolicy="Always", + ) + return CompiledModelDeployment( + names=names, + volume=volume, + scratch_volume=scratch_volume, + puller_config=puller_config, + server_config=server_config, + puller_prerequisite=puller_config is not None, + ) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/config.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/config.py new file mode 100644 index 0000000000..7d03911285 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/config.py @@ -0,0 +1,51 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Configuration for the deployments-plugin models backend.""" + +from pydantic import BaseModel, Field + + +class DeploymentsPluginConfig(BaseModel): + default_executor: str | None = None + docker_executor: str | None = None + k8s_executor: str | None = None + pending_timeout_seconds: int = Field( + default=7200, + ge=60, + description="Maximum seconds a deployment may stay PENDING before ERROR.", + ) + max_restart_count: int = 5 + default_storage_class: str | None = None + default_pvc_size: str = "200Gi" + default_nimservice_image: str = "nvcr.io/nim/meta/llama-3.1-8b-instruct" + default_nimservice_image_tag: str = "1.8.5" + default_vllm_image: str = "vllm/vllm-openai" + default_vllm_image_tag: str = "v0.8.5" + default_user_id: int | None = 1000 + default_group_id: int | None = 2000 + default_vllm_user_id: int | None = 2000 + default_vllm_group_id: int | None = 0 + peft_refresh_interval: int = 30 + lora_sidecar_image_name: str = "nmp-api" + lora_sidecar_command: list[str] = Field( + default_factory=lambda: ["nemo", "services", "run", "--sidecars", "adapters"] + ) + lora_sidecar_args: list[str] = Field(default_factory=list) + busybox_image: str = Field( + default="docker.io/library/busybox", + description="BusyBox image repository for LoRA cache init containers. " + "Fully qualified so it resolves on runtimes that block docker.io short names.", + ) + busybox_image_tag: str = Field( + default="latest", + description="BusyBox image tag for LoRA cache init containers.", + ) + delete_wait_seconds: float = Field(default=5.0, gt=0) + delete_poll_seconds: float = Field(default=0.5, gt=0) + + +class DeploymentsPluginBackendConfigModel(DeploymentsPluginConfig): + """Flat registry configuration, including the enablement switch.""" + + enabled: bool = False diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/executor.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/executor.py new file mode 100644 index 0000000000..cf07d8f598 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/executor.py @@ -0,0 +1,24 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Executor selection for deployments-plugin model entities.""" + +from nmp.common.config import Runtime +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig + + +def executor_for_runtime(config: DeploymentsPluginConfig, runtime: Runtime) -> str | None: + """Resolve a configured executor name for the platform runtime. + + ``platform.runtime`` (docker/kubernetes) is global deployment topology; + executor names are per-backend config that map to entries in the + deployments-plugin executor registry. This returns ``None`` when the + operator enabled ``deployments_plugin`` but omitted ``docker_executor``, + ``k8s_executor``, and ``default_executor`` for the active runtime — + ``create_model_deployment`` surfaces that misconfiguration as ERROR. + """ + if runtime == Runtime.DOCKER: + return config.docker_executor or config.default_executor + if runtime == Runtime.KUBERNETES: + return config.k8s_executor or config.default_executor + return config.default_executor diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/naming.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/naming.py new file mode 100644 index 0000000000..e47297a455 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/naming.py @@ -0,0 +1,26 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Stable deployments-plugin entity names for a model deployment.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class EntityNames: + """Names of the substrate entities managed for one model deployment.""" + + volume: str + scratch: str + puller: str + server: str + + +def entity_names(name: str) -> EntityNames: + """Return names relative to the ModelDeployment name.""" + return EntityNames( + volume=f"{name}-weights", + scratch=f"{name}-scratch", + puller=f"{name}-puller", + server=f"{name}-server", + ) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/resolve.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/resolve.py new file mode 100644 index 0000000000..faa301bfd5 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/resolve.py @@ -0,0 +1,74 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Resolve models API objects into compiler inputs.""" + +from dataclasses import dataclass +from urllib.parse import urljoin + +from nemo_platform.types.inference.model_deployment import ModelDeployment +from nemo_platform.types.inference.model_deployment_config import ModelDeploymentConfig +from nemo_platform.types.models.model_entity import ModelEntity +from nmp.common.config import Runtime, get_platform_config +from nmp.core.models.app import ModelWeightsType, get_model_weights_type, parse_model_name_revision +from nmp.core.models.controllers.backends.common import DeploymentConfigView, deployment_config_view +from nmp.core.models.controllers.context import ModelContext + + +@dataclass(frozen=True) +class ResolvedPluginDeployment: + """All API-object data required to compile plugin entities.""" + + deployment: ModelDeployment + config: ModelDeploymentConfig + model_entity: ModelEntity | None + view: DeploymentConfigView + weights_type: ModelWeightsType + model_namespace: str | None + model_name: str | None + model_revision: str | None + files_hf_url: str + huggingface_model_puller: str + runtime: Runtime + + +def resolve_model_source( + model_entity: ModelEntity | None, view: DeploymentConfigView +) -> tuple[str | None, str | None, str | None]: + """Resolve file-set-backed model sources before config fallback.""" + namespace, name, revision = parse_model_name_revision( + model_namespace=view.model_namespace, model_name=view.model_name, model_revision=view.model_revision + ) + if model_entity and model_entity.fileset: + parts = str(model_entity.fileset).removeprefix("hf://").removeprefix("fileset://").split("/", 1) + if len(parts) == 2: + return parts[0], parts[1], revision + return namespace, name, revision + + +def resolve_plugin_deployment(ctx: ModelContext, huggingface_model_puller: str) -> ResolvedPluginDeployment: + """Build compiler input from a model reconciliation context.""" + if ctx.model_deployment is None or ctx.model_deployment_config is None: + raise ValueError("Model deployment and deployment config are required.") + view = deployment_config_view(ctx.model_deployment_config) + namespace, name, revision = resolve_model_source(ctx.model_entity, view) + platform_config = get_platform_config() + files_service_url = platform_config.service_discovery.get("files") or platform_config.base_url + files_hf_url = urljoin(files_service_url.rstrip("/") + "/", "apis/files/v2/hf") + return ResolvedPluginDeployment( + deployment=ctx.model_deployment, + config=ctx.model_deployment_config, + model_entity=ctx.model_entity, + view=view, + weights_type=get_model_weights_type( + model_deployment=ctx.model_deployment, + model_deployment_config=ctx.model_deployment_config, + model_entity=ctx.model_entity, + ), + model_namespace=namespace, + model_name=name, + model_revision=revision, + files_hf_url=files_hf_url, + huggingface_model_puller=huggingface_model_puller, + runtime=platform_config.runtime, + ) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/status.py b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/status.py new file mode 100644 index 0000000000..3335267a7d --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/deployments_plugin/status.py @@ -0,0 +1,146 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Project deployments-plugin substrate state onto ModelDeployment state.""" + +from typing import Any, Iterable + +from nemo_deployments_plugin.entities import Deployment, Volume +from nemo_deployments_plugin.types import Endpoint +from nemo_platform.types.inference import ModelDeploymentStatus +from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate +from nmp.core.models.controllers.backends.common import format_duration + +_STATUS_MAP: dict[str, ModelDeploymentStatus] = { + "PENDING": "PENDING", + "STARTING": "PENDING", + "READY": "READY", + "FAILED": "ERROR", + "LOST": "LOST", + "UNKNOWN": "UNKNOWN", + "DELETING": "DELETING", + "SUCCEEDED": "PENDING", +} + + +def map_status(status: str) -> ModelDeploymentStatus: + """Map a deployments-plugin status to a ModelDeployment status.""" + return _STATUS_MAP.get(status, "UNKNOWN") + + +def project_host_url(endpoints: Iterable[Endpoint]) -> str | None: + """Return the first HTTP(S) endpoint exposed by the plugin deployment.""" + return next( + (endpoint.url for endpoint in endpoints if endpoint.protocol in {"http", "https"} and endpoint.url), None + ) + + +def _substrate(entity: Deployment | Volume | None) -> dict[str, Any] | None: + if entity is None: + return None + return { + "status": entity.status, + "status_message": entity.status_message, + "error_details": entity.error_details, + } + + +# Substrate statuses that must not silently fall through to PENDING: FAILED/LOST +# are terminal (the dependent server Deployment can never satisfy its +# prerequisite), and UNKNOWN is indeterminate. Surfacing them avoids reporting a +# healthy-looking PENDING over a dead prerequisite. +_ATTENTION_SUBSTRATE_STATUSES = frozenset({"FAILED", "LOST", "UNKNOWN"}) + + +def _substrate_issue( + entity: Deployment | Volume | None, label: str, substrate: dict[str, Any] +) -> DeploymentStatusUpdate | None: + if entity is None or entity.status not in _ATTENTION_SUBSTRATE_STATUSES: + return None + status: ModelDeploymentStatus = "UNKNOWN" if entity.status == "UNKNOWN" else "ERROR" + return DeploymentStatusUpdate( + status=status, + status_message=entity.status_message or f"{label} is {entity.status}.", + error_details={"substrate": substrate}, + ) + + +def aggregate_status( + volume: Volume | None, + puller: Deployment | None, + server: Deployment | None, + *, + previously_ready: bool = False, +) -> DeploymentStatusUpdate: + """Project the three plugin entities, preferring the serving deployment.""" + substrate = {"volume": _substrate(volume), "puller": _substrate(puller), "server": _substrate(server)} + if server is not None: + status = map_status(server.status) + return DeploymentStatusUpdate( + status=status, + status_message=server.status_message or f"Server deployment is {server.status}.", + error_details={"substrate": substrate}, + host_url=project_host_url(server.endpoints) if status == "READY" else None, + ) + if previously_ready: + return DeploymentStatusUpdate( + status="LOST", + status_message="Serving deployment is missing after reporting READY.", + error_details={"substrate": substrate}, + ) + issue = _substrate_issue(puller, "Weight puller", substrate) or _substrate_issue( + volume, "Weights volume", substrate + ) + if issue is not None: + return issue + return DeploymentStatusUpdate( + status="PENDING", + status_message="Waiting for deployments-plugin substrate resources.", + error_details={"substrate": substrate}, + ) + + +def build_pending_timeout_error( + *, + deployment_name: str, + elapsed_seconds: float, + timeout_seconds: int, + substrate: dict[str, Any] | None = None, +) -> DeploymentStatusUpdate: + """Build ERROR status when a deployment exceeds ``pending_timeout_seconds``.""" + status_msg = ( + f"Deployment '{deployment_name}' timed out after {format_duration(elapsed_seconds)} waiting for " + f"deployments-plugin substrate to become READY (timeout: {format_duration(timeout_seconds)})." + ) + error_details: dict[str, Any] = { + "reason": "pending_timeout", + "elapsed_seconds": int(elapsed_seconds), + "timeout_seconds": timeout_seconds, + "deployment_name": deployment_name, + } + if substrate is not None: + error_details["substrate"] = substrate + return DeploymentStatusUpdate( + status="ERROR", + status_message=status_msg, + error_details=error_details, + ) + + +def apply_pending_timeout( + result: DeploymentStatusUpdate, + *, + elapsed_seconds: float, + timeout_seconds: int, + deployment_name: str, +) -> DeploymentStatusUpdate: + """Escalate a PENDING projection to ERROR once the deployment ages out.""" + if result.status != "PENDING" or elapsed_seconds < timeout_seconds: + return result + substrate = result.error_details.get("substrate") if result.error_details else None + return build_pending_timeout_error( + deployment_name=deployment_name, + elapsed_seconds=elapsed_seconds, + timeout_seconds=timeout_seconds, + substrate=substrate, + ) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/registry.py b/services/core/models/src/nmp/core/models/controllers/backends/registry.py index c37e4aa3d7..f17cc61009 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/registry.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/registry.py @@ -8,6 +8,12 @@ from nemo_platform import AsyncNeMoPlatform from nmp.core.models.controllers.backends.backends import ServiceBackend + +# NOTE: import the config model from the plugin-free `config` module (not the +# package __init__) so the registry does not eagerly import the optional +# `nemo_deployments_plugin` dependency. The backend class itself is resolved +# lazily in `from_config` only when the deployments_plugin backend is selected. +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginBackendConfigModel from nmp.core.models.controllers.backends.docker import DockerBackendConfig as DockerConfig from nmp.core.models.controllers.backends.docker import DockerServiceBackend from nmp.core.models.controllers.backends.k8s_nim_operator import K8sNimOperatorConfig, K8sNimOperatorServiceBackend @@ -30,25 +36,66 @@ class DockerBackendConfigModel(DockerConfig): class NoneBackendConfigModel(BaseModel): - """Configuration for None backend.""" + """Configuration for the ``none`` backend (no deployment substrate). + + Used when ``platform.runtime`` is ``none``. The backend is a deliberate + no-op: create/update/delete raise ``NotImplementedError``, status returns + ``UNKNOWN``, and orphan reconciliation sees no managed deployments. + """ enabled: bool = Field(default=False, description="Whether this backend is enabled") # Union of all backend configurations (no discriminator needed since dict key is the backend name) -BackendConfig = Union[DockerBackendConfigModel, K8sNimOperatorBackendConfigModel, NoneBackendConfigModel] +BackendConfig = Union[ + DockerBackendConfigModel, + K8sNimOperatorBackendConfigModel, + DeploymentsPluginBackendConfigModel, + NoneBackendConfigModel, +] # Type alias for the backend name BackendName = str -# Global registry of available backend implementations +# Global registry of always-importable backend implementations. The +# `deployments_plugin` backend is intentionally excluded here because it imports +# the optional `nemo_deployments_plugin` package; it is resolved lazily by +# `_resolve_backend_class` when selected. backend_classes: Dict[BackendName, type[ServiceBackend]] = { "docker": DockerServiceBackend, "nim_operator": K8sNimOperatorServiceBackend, "none": NoneServiceBackend, } +# Backends whose implementation lives behind an optional dependency and must be +# imported lazily. +_LAZY_BACKEND_NAMES = frozenset({"deployments_plugin"}) + +_DEPLOYMENTS_PLUGIN_IMPORT_ERROR = ( + "The deployments_plugin models backend requires the nemo-deployments-plugin " + "package. Install it (or include the deployments plugin in your platform " + "profile) before setting models.controller.backends.deployments_plugin.enabled." +) + + +def _resolve_backend_class( + name: BackendName, available_backends: Dict[BackendName, type[ServiceBackend]] +) -> type[ServiceBackend]: + """Return the backend class for ``name``, importing optional backends lazily.""" + if name in available_backends: + return available_backends[name] + if name == "deployments_plugin": + try: + from nmp.core.models.controllers.backends.deployments_plugin.backend import ( + DeploymentsPluginServiceBackend, + ) + except ImportError as exc: + raise ImportError(_DEPLOYMENTS_PLUGIN_IMPORT_ERROR) from exc + return DeploymentsPluginServiceBackend + available = ", ".join(sorted({*available_backends, *_LAZY_BACKEND_NAMES})) + raise KeyError(f"Unknown backend '{name}'. Available backends: {available}") + class BackendRegistry: """ @@ -115,15 +162,11 @@ def from_config( registry: Dict[BackendName, ServiceBackend] = {} for backend_name, backend_config in enabled_backends.items(): - if backend_name not in available_backends: - available = ", ".join(available_backends.keys()) - raise KeyError(f"Unknown backend '{backend_name}'. Available backends: {available}") - - backend_class = available_backends[backend_name] + backend_class = _resolve_backend_class(backend_name, available_backends) logger.info(f"Initializing backend: {backend_name}") config_dict = backend_config.model_dump(exclude={"enabled"}) - if backend_name == "nim_operator": + if backend_name in {"nim_operator", "deployments_plugin"}: registry[backend_name] = backend_class(nmp_sdk, config_dict, huggingface_model_puller) else: registry[backend_name] = backend_class(nmp_sdk, config_dict) diff --git a/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_backend.py b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_backend.py new file mode 100644 index 0000000000..9e99e0ad46 --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_backend.py @@ -0,0 +1,273 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from nemo_deployments_plugin.entities import Deployment, DeploymentConfig, Volume +from nemo_deployments_plugin.types import Endpoint +from nemo_platform_plugin.entity_client import NemoEntityNotFoundError +from nmp.common.config import Runtime +from nmp.core.models.app import ModelWeightsType +from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate +from nmp.core.models.controllers.backends.common import DeploymentConfigView +from nmp.core.models.controllers.backends.deployments_plugin.backend import DeploymentsPluginServiceBackend +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig +from nmp.core.models.controllers.backends.deployments_plugin.resolve import ResolvedPluginDeployment + + +def _ctx() -> SimpleNamespace: + return SimpleNamespace( + model_deployment=SimpleNamespace(name="my-dep", workspace="default", status="CREATED"), + model_deployment_config=SimpleNamespace(engine="vllm"), + model_entity=None, + ) + + +def _resolved() -> ResolvedPluginDeployment: + return ResolvedPluginDeployment( + deployment=SimpleNamespace(name="my-dep", workspace="default"), + config=SimpleNamespace(engine="vllm"), + model_entity=None, + view=DeploymentConfigView(model_namespace="org", model_name="model"), + weights_type=ModelWeightsType.FILES_SERVICE, + model_namespace="org", + model_name="model", + model_revision=None, + files_hf_url="http://files/hf", + huggingface_model_puller="puller:latest", + runtime=Runtime.KUBERNETES, + ) + + +@pytest.mark.asyncio +async def test_get_status_projects_ready_endpoint() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + backend._entities.get = AsyncMock( + side_effect=[ + Deployment( + name="my-dep-server", + workspace="default", + deployment_config="my-dep-server", + status="READY", + endpoints=[Endpoint(name="http", url="http://server", protocol="http")], + ), + NemoEntityNotFoundError("missing"), + NemoEntityNotFoundError("missing"), + ] + ) + result = await backend.get_model_deployment_status( + SimpleNamespace( + model_deployment=SimpleNamespace( + name="my-dep", + workspace="default", + status="PENDING", + created_at=datetime.now(timezone.utc), + ) + ) + ) + assert result.status == "READY" + assert result.host_url == "http://server" + + +@pytest.mark.asyncio +async def test_missing_ready_server_is_lost() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + backend._entities.get = AsyncMock(side_effect=NemoEntityNotFoundError("missing")) + result = await backend.get_model_deployment_status( + SimpleNamespace( + model_deployment=SimpleNamespace( + name="my-dep", + workspace="default", + status="READY", + created_at=datetime.now(timezone.utc), + ) + ) + ) + assert result.status == "LOST" + + +@pytest.mark.asyncio +async def test_create_order_volume_puller_server_with_prerequisite() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + created: list[object] = [] + + async def _create(entity: object) -> object: + created.append(entity) + return entity + + backend._entities.create = AsyncMock(side_effect=_create) + backend._entities.get = AsyncMock(side_effect=NemoEntityNotFoundError("missing")) + backend._entities.delete = AsyncMock(side_effect=NemoEntityNotFoundError("missing")) + with ( + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.resolve_plugin_deployment", + return_value=_resolved(), + ), + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.executor_for_runtime", + return_value="local-k8s", + ), + ): + result = await backend.create_model_deployment(_ctx()) + + assert result.status == "PENDING" + assert [type(item) for item in created] == [Volume, DeploymentConfig, Deployment, DeploymentConfig, Deployment] + puller_dep = created[2] + server_dep = created[4] + assert isinstance(puller_dep, Deployment) and puller_dep.name == "my-dep-puller" + assert isinstance(server_dep, Deployment) and server_dep.name == "my-dep-server" + assert server_dep.prerequisites[0].deployment_name == "my-dep-puller" + assert server_dep.prerequisites[0].condition == "succeeded" + + +def _resolved_docker_lora() -> ResolvedPluginDeployment: + return ResolvedPluginDeployment( + deployment=SimpleNamespace(name="my-dep", workspace="default"), + config=SimpleNamespace(engine="vllm"), + model_entity=None, + view=DeploymentConfigView(model_namespace="org", model_name="model", lora_enabled=True), + weights_type=ModelWeightsType.FILES_SERVICE, + model_namespace="org", + model_name="model", + model_revision=None, + files_hf_url="http://files/hf", + huggingface_model_puller="puller:latest", + runtime=Runtime.DOCKER, + ) + + +@pytest.mark.asyncio +async def test_docker_lora_fails_fast_before_touching_substrate() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + with ( + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.resolve_plugin_deployment", + return_value=_resolved_docker_lora(), + ), + patch.object(backend, "delete_model_deployment", AsyncMock()) as delete_mock, + ): + result = await backend.create_model_deployment(_ctx()) + assert result.status == "ERROR" + assert "docker" in result.status_message.lower() + assert "lora" in result.status_message.lower() + delete_mock.assert_not_called() + backend._entities.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_waits_when_prior_teardown_incomplete() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + with ( + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.resolve_plugin_deployment", + return_value=_resolved(), + ), + patch.object( + backend, + "delete_model_deployment", + AsyncMock(return_value=DeploymentStatusUpdate(status="DELETING", status_message="waiting")), + ), + ): + result = await backend.create_model_deployment(_ctx()) + assert result.status == "PENDING" + assert "teardown" in result.status_message.lower() + backend._entities.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_missing_executor_fails_fast_before_touching_substrate() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._entities = AsyncMock() + backend._entities.get = AsyncMock(side_effect=NemoEntityNotFoundError("missing")) + backend._entities.delete = AsyncMock(side_effect=NemoEntityNotFoundError("missing")) + with ( + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.resolve_plugin_deployment", + return_value=_resolved(), + ), + patch( + "nmp.core.models.controllers.backends.deployments_plugin.backend.executor_for_runtime", + return_value=None, + ), + ): + result = await backend.create_model_deployment(_ctx()) + assert result.status == "ERROR" + assert "executor" in result.status_message.lower() + assert result.error_details is not None + assert result.error_details["reason"] == "executor_not_configured" + backend._entities.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_pending_timeout_escalates_stuck_deployment() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend.init() + backend._backend_config = DeploymentsPluginConfig(pending_timeout_seconds=60) + backend._entities = AsyncMock() + server = Deployment( + name="my-dep-server", + workspace="default", + deployment_config="my-dep-server", + status="STARTING", + ) + backend._entities.get = AsyncMock( + side_effect=[ + server, + NemoEntityNotFoundError("missing"), + NemoEntityNotFoundError("missing"), + ] + ) + created_at = datetime.now(timezone.utc) - timedelta(seconds=120) + result = await backend.get_model_deployment_status( + SimpleNamespace( + model_deployment=SimpleNamespace( + name="my-dep", + workspace="default", + status="PENDING", + created_at=created_at, + ) + ) + ) + assert result.status == "ERROR" + assert result.error_details is not None + assert result.error_details["reason"] == "pending_timeout" + assert result.error_details["timeout_seconds"] == 60 + + +@pytest.mark.asyncio +async def test_delete_waits_for_server_before_config() -> None: + backend = DeploymentsPluginServiceBackend(AsyncMock(), {}, "puller:latest") + backend._backend_config = DeploymentsPluginConfig(delete_wait_seconds=0.02, delete_poll_seconds=0.005) + backend._entities = AsyncMock() + server = Deployment( + name="my-dep-server", + workspace="default", + deployment_config="my-dep-server", + status="READY", + ) + + async def _get(entity_type: type, name: str, workspace: str | None = None) -> Deployment: + del entity_type, name, workspace + return server + + backend._entities.get = AsyncMock(side_effect=_get) + backend._entities.update = AsyncMock(side_effect=lambda entity: entity) + backend._entities.delete = AsyncMock() + + result = await backend.delete_model_deployment("default", "my-dep") + assert result.status == "DELETING" + backend._entities.delete.assert_not_called() diff --git a/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_compiler.py b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_compiler.py new file mode 100644 index 0000000000..f6794faab0 --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_compiler.py @@ -0,0 +1,91 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from types import SimpleNamespace +from unittest.mock import patch + +from nmp.common.config import Runtime +from nmp.core.models.app import ModelWeightsType +from nmp.core.models.controllers.backends.common import DeploymentConfigView +from nmp.core.models.controllers.backends.deployments_plugin.compiler import compile_model_deployment +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig +from nmp.core.models.controllers.backends.deployments_plugin.resolve import ResolvedPluginDeployment + + +def _resolved(engine: str, *, lora: bool = False, runtime: Runtime = Runtime.KUBERNETES) -> ResolvedPluginDeployment: + return ResolvedPluginDeployment( + deployment=SimpleNamespace(name="my-dep", workspace="default"), + config=SimpleNamespace(engine=engine), + model_entity=None, + view=DeploymentConfigView(model_namespace="org", model_name="model", lora_enabled=lora), + weights_type=ModelWeightsType.FILES_SERVICE, + model_namespace="org", + model_name="model", + model_revision=None, + files_hf_url="http://files/hf", + huggingface_model_puller="puller:latest", + runtime=runtime, + ) + + +def test_vllm_weighted_chain_has_on_failure_puller_and_always_server() -> None: + compiled = compile_model_deployment(_resolved("vllm"), DeploymentsPluginConfig()) + assert compiled.volume is not None + assert compiled.puller_config is not None and compiled.puller_config.restart_policy == "OnFailure" + assert compiled.server_config.restart_policy == "Always" + assert compiled.puller_prerequisite is True + assert compiled.server_config.containers[0].volume_mounts[0].read_only is True + + +def test_nim_weighted_chain_sets_model_path_env() -> None: + compiled = compile_model_deployment(_resolved("nim"), DeploymentsPluginConfig()) + assert compiled.volume is not None + assert compiled.puller_config is not None + env = {item.name: item.value for item in compiled.server_config.containers[0].env} + assert env["NIM_MODEL_NAME"] == "/model-store" + assert env["NIM_MODEL_PATH"] == "/model-store" + assert env["NIM_SERVED_MODEL_NAME"] == "org/model" + + +def test_generic_weightless_is_server_only() -> None: + resolved = _resolved("generic") + resolved = ResolvedPluginDeployment( + deployment=resolved.deployment, + config=resolved.config, + model_entity=None, + view=DeploymentConfigView(image_name="custom/image", image_tag="1"), + weights_type=ModelWeightsType.BAKED_CONTAINER, + model_namespace=None, + model_name=None, + model_revision=None, + files_hf_url=resolved.files_hf_url, + huggingface_model_puller=resolved.huggingface_model_puller, + runtime=resolved.runtime, + ) + compiled = compile_model_deployment(resolved, DeploymentsPluginConfig()) + assert compiled.volume is None + assert compiled.puller_config is None + assert compiled.puller_prerequisite is False + assert compiled.server_config.containers[0].image == "custom/image:1" + + +def test_lora_uses_native_sidecar_on_k8s_and_container_on_docker() -> None: + config = DeploymentsPluginConfig() + with patch( + "nmp.core.models.controllers.backends.deployments_plugin.compiler.get_qualified_image", + return_value="registry/nmp-api:tag", + ): + k8s = compile_model_deployment(_resolved("vllm", lora=True), config) + docker = compile_model_deployment(_resolved("vllm", lora=True, runtime=Runtime.DOCKER), config) + + assert k8s.scratch_volume is not None + init = k8s.server_config.init_containers[0] + assert init.name == "lora-cache-init" + assert init.image == "docker.io/library/busybox:latest" + sidecar = k8s.server_config.init_containers[-1] + assert sidecar.restart_policy == "Always" + assert sidecar.image == "registry/nmp-api:tag" + env = {item.name: item.value for item in sidecar.env} + assert env["NIM_PEFT_SOURCE"] == "/scratch/loras" + assert env["VLLM_LORA_BASE_MODEL_OVERRIDE"] == "/model-store" + assert len(docker.server_config.containers) == 2 diff --git a/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_executor.py b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_executor.py new file mode 100644 index 0000000000..34f972f8c4 --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_executor.py @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from nmp.common.config import Runtime +from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginConfig +from nmp.core.models.controllers.backends.deployments_plugin.executor import executor_for_runtime + + +def test_executor_prefers_runtime_specific_value_then_default() -> None: + config = DeploymentsPluginConfig(default_executor="default", docker_executor="docker", k8s_executor="k8s") + assert executor_for_runtime(config, Runtime.DOCKER) == "docker" + assert executor_for_runtime(config, Runtime.KUBERNETES) == "k8s" + assert executor_for_runtime(DeploymentsPluginConfig(default_executor="default"), Runtime.DOCKER) == "default" diff --git a/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_naming.py b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_naming.py new file mode 100644 index 0000000000..4fb59a973c --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_naming.py @@ -0,0 +1,12 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from nmp.core.models.controllers.backends.deployments_plugin.naming import entity_names + + +def test_entity_names_have_expected_suffixes() -> None: + names = entity_names("my-dep") + assert names.volume == "my-dep-weights" + assert names.scratch == "my-dep-scratch" + assert names.puller == "my-dep-puller" + assert names.server == "my-dep-server" diff --git a/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_status.py b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_status.py new file mode 100644 index 0000000000..5429c5db4b --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/deployments_plugin/test_status.py @@ -0,0 +1,107 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest +from nemo_deployments_plugin.entities import Deployment, Volume +from nemo_deployments_plugin.types import Endpoint +from nmp.core.models.controllers.backends.deployments_plugin.status import ( + aggregate_status, + apply_pending_timeout, + build_pending_timeout_error, + map_status, + project_host_url, +) + + +@pytest.mark.parametrize( + ("plugin_status", "model_status"), + [ + ("PENDING", "PENDING"), + ("STARTING", "PENDING"), + ("READY", "READY"), + ("FAILED", "ERROR"), + ("LOST", "LOST"), + ("UNKNOWN", "UNKNOWN"), + ("DELETING", "DELETING"), + ("SUCCEEDED", "PENDING"), + ], +) +def test_map_status(plugin_status: str, model_status: str) -> None: + assert map_status(plugin_status) == model_status + + +def test_map_status_defaults_to_unknown() -> None: + assert map_status("SOMETHING_NEW") == "UNKNOWN" + + +def test_ready_projects_http_endpoint() -> None: + server = Deployment( + name="server", + workspace="default", + deployment_config="server", + status="READY", + endpoints=[Endpoint(name="http", url="https://server", protocol="https")], + ) + assert project_host_url(server.endpoints) == "https://server" + assert aggregate_status(None, None, server).host_url == "https://server" + + +def test_missing_ready_server_is_lost_and_failed_puller_is_error() -> None: + assert aggregate_status(None, None, None, previously_ready=True).status == "LOST" + puller = Deployment(name="puller", workspace="default", deployment_config="puller", status="FAILED") + assert aggregate_status(None, puller, None).status == "ERROR" + + +def test_no_substrate_yet_is_pending() -> None: + result = aggregate_status(None, None, None) + assert result.status == "PENDING" + + +def test_failed_volume_is_error() -> None: + volume = Volume(name="volume", workspace="default", size="1Gi", status="FAILED") + assert aggregate_status(volume, None, None).status == "ERROR" + + +def test_lost_puller_is_error() -> None: + puller = Deployment(name="puller", workspace="default", deployment_config="puller", status="LOST") + assert aggregate_status(None, puller, None).status == "ERROR" + + +def test_unknown_puller_surfaces_unknown() -> None: + puller = Deployment(name="puller", workspace="default", deployment_config="puller", status="UNKNOWN") + assert aggregate_status(None, puller, None).status == "UNKNOWN" + + +def test_apply_pending_timeout_escalates_pending_only() -> None: + pending = aggregate_status(None, None, None) + assert ( + apply_pending_timeout( + pending, + elapsed_seconds=30, + timeout_seconds=60, + deployment_name="my-dep", + ).status + == "PENDING" + ) + timed_out = apply_pending_timeout( + pending, + elapsed_seconds=90, + timeout_seconds=60, + deployment_name="my-dep", + ) + assert timed_out.status == "ERROR" + assert timed_out.error_details is not None + assert timed_out.error_details["reason"] == "pending_timeout" + + +def test_build_pending_timeout_error_includes_substrate() -> None: + substrate = {"server": {"status": "STARTING"}} + result = build_pending_timeout_error( + deployment_name="my-dep", + elapsed_seconds=120, + timeout_seconds=60, + substrate=substrate, + ) + assert result.status == "ERROR" + assert result.error_details is not None + assert result.error_details["substrate"] == substrate diff --git a/services/core/models/tests/unit/controllers/test_backend_registry.py b/services/core/models/tests/unit/controllers/test_backend_registry.py index 655e310212..5a1e6ecc66 100644 --- a/services/core/models/tests/unit/controllers/test_backend_registry.py +++ b/services/core/models/tests/unit/controllers/test_backend_registry.py @@ -244,3 +244,35 @@ def test_backend_registry_multiple_enabled_backends_raises_error(mock_nmp_sdk): backend_configs=config_with_multiple_enabled, huggingface_model_puller="nvcr.io/nvidia/nemo-microservices/nds-v2-huggingface-cli:25.10", ) + + +def test_deployments_plugin_missing_package_raises_guidance(mock_nmp_sdk): + """Missing nemo-deployments-plugin should surface install guidance.""" + import builtins + + from nmp.core.models.controllers.backends.deployments_plugin.config import DeploymentsPluginBackendConfigModel + from nmp.core.models.controllers.backends.registry import _resolve_backend_class, backend_classes + + real_import = builtins.__import__ + + def mock_import(name: str, *args: object, **kwargs: object) -> object: + if name == "nmp.core.models.controllers.backends.deployments_plugin.backend": + raise ImportError("No module named 'nemo_deployments_plugin'") + return real_import(name, *args, **kwargs) + + with ( + patch("builtins.__import__", side_effect=mock_import), + pytest.raises(ImportError, match="nemo-deployments-plugin"), + ): + _resolve_backend_class("deployments_plugin", backend_classes) + + # Enabled config path should fail the same way during registry init. + with ( + patch("builtins.__import__", side_effect=mock_import), + pytest.raises(ImportError, match="nemo-deployments-plugin"), + ): + BackendRegistry.from_config( + nmp_sdk=mock_nmp_sdk, + backend_configs={"deployments_plugin": DeploymentsPluginBackendConfigModel(enabled=True)}, + huggingface_model_puller="puller:latest", + ) diff --git a/uv.lock b/uv.lock index a708953e70..2a69108408 100644 --- a/uv.lock +++ b/uv.lock @@ -4844,6 +4844,7 @@ models-service = [ { name = "docker", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "fastapi", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "kubernetes", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nmp-common", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "pydantic", extra = ["email"], marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "pydantic-settings", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -5391,6 +5392,7 @@ requires-dist = [ { name = "nemo-platform-plugin", marker = "extra == 'all'", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-plugin", marker = "extra == 'core-service'", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-plugin", marker = "extra == 'inference-gateway-service'", editable = "packages/nemo_platform_plugin" }, + { name = "nemo-platform-plugin", marker = "extra == 'models-service'", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-plugin", marker = "extra == 'nemo-agents-plugin'", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-plugin", marker = "extra == 'nemo-anonymizer-plugin'", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-plugin", marker = "extra == 'nemo-auditor-plugin'", editable = "packages/nemo_platform_plugin" }, @@ -7389,6 +7391,8 @@ dependencies = [ { name = "docker", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "fastapi", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "kubernetes", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-deployments-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nmp-common", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "pydantic", extra = ["email"], marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "pydantic-settings", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -7418,6 +7422,8 @@ requires-dist = [ { name = "docker", specifier = ">=7.1.0" }, { name = "fastapi", specifier = ">=0.115.8" }, { name = "kubernetes", specifier = ">=31.0.0" }, + { name = "nemo-deployments-plugin", editable = "plugins/nemo-deployments" }, + { name = "nemo-platform-plugin", editable = "packages/nemo_platform_plugin" }, { name = "nmp-common", editable = "packages/nmp_common" }, { name = "pydantic", specifier = ">=2.10.6" }, { name = "pydantic-settings", specifier = ">=2.8.1" },