From f7e5b899c68a37faed023e6ddd9e93bb852efc4e Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Thu, 25 Jun 2026 14:46:12 -0600 Subject: [PATCH 1/3] feat(models): generic engine support Signed-off-by: Ben McCown --- .../model_deployment_config_service.py | 30 +++ .../backends/docker/creation_reconciler.py | 39 +++- .../controllers/backends/generic_compiler.py | 57 +++++ .../backends/k8s_nim_operator/backend.py | 15 +- .../k8s_nim_operator/reconcilers/k8s.py | 194 +++++++++++++++++- .../k8s_nim_operator/vllm_k8s_compiler.py | 29 ++- .../backends/test_generic_compiler.py | 69 +++++++ .../backends/test_vllm_k8s_compiler.py | 28 +++ .../unit/controllers/test_docker_backend.py | 55 +++++ .../test_k8s_nim_operator_backend.py | 97 ++++++++- ...st_model_deployment_config_service_unit.py | 47 +++++ 11 files changed, 617 insertions(+), 43 deletions(-) create mode 100644 services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py create mode 100644 services/core/models/tests/unit/controllers/backends/test_generic_compiler.py diff --git a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py index b6d9d41e3c..f30733e4e2 100644 --- a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py +++ b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py @@ -15,7 +15,9 @@ from nmp.core.models.entities import ModelDeployment as ModelDeploymentEntity from nmp.core.models.entities import ModelDeploymentConfig as ModelDeploymentConfigEntity from nmp.core.models.schemas import ( + ContainerExecutorConfig, CreateModelDeploymentConfigRequest, + Engine, ModelDeploymentConfig, ModelDeploymentStatus, UpdateModelDeploymentConfigRequest, @@ -24,6 +26,30 @@ logger = logging.getLogger(__name__) +def _validate_engine_config(engine: Engine, executor_config: ContainerExecutorConfig) -> None: + """Validate engine-specific requirements on the executor config. + + The ``generic`` engine runs an arbitrary container with no inference-engine + compiler, so it has no platform-default image and no canonical health + endpoint. Both ``image_name`` and ``health_check_path`` must therefore be + supplied explicitly; the other engines fall back to their configured + defaults when these are unset. + """ + if engine != Engine.GENERIC: + return + missing: list[str] = [] + if not (executor_config.image_name and executor_config.image_name.strip()): + missing.append("image_name") + if not (executor_config.health_check_path and executor_config.health_check_path.strip()): + missing.append("health_check_path") + if missing: + raise ValueError( + "The 'generic' engine requires executor_config." + + " and executor_config.".join(missing) + + " to be set (no platform default exists for a generic container)." + ) + + class ReferentialIntegrityError(Exception): """Exception raised when trying to delete a resource that has dependencies.""" @@ -104,6 +130,8 @@ async def create_deployment_config( if existing is not None: raise ValueError(f"Deployment config with workspace '{workspace}' and name '{request.name}' already exists") + _validate_engine_config(request.engine, request.executor_config) + if not request.model_entity_id: try: model_workspace, model_name, _ = parse_model_name_revision( @@ -255,6 +283,8 @@ async def update_deployment_config( if not current: raise ValueError(f"Deployment config with workspace '{workspace}' and name '{name}' does not exist") + _validate_engine_config(request.engine, request.executor_config) + new_version = current.entity_version + 1 # Create new version entity diff --git a/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py b/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py index 4ddac4d850..b24557f57c 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py @@ -35,7 +35,7 @@ from nmp.core.models.app import ModelWeightsType, get_model_weights_type, is_multi_llm_image, parse_model_name_revision from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER from nmp.core.models.app.utils import _get_k8s_safe_name -from nmp.core.models.controllers.backends import vllm_compiler +from nmp.core.models.controllers.backends import generic_compiler, vllm_compiler from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate from nmp.core.models.controllers.backends.common import deployment_config_view from nmp.core.models.controllers.backends.docker.config import ( @@ -44,6 +44,7 @@ DockerBackendConfig, ) from nmp.core.models.controllers.backends.engine import ( + ENGINE_GENERIC, ENGINE_HEALTH_PATHS, ENGINE_LABEL, ENGINE_NIM, @@ -492,6 +493,10 @@ async def register_deployment( self._backend_config.default_vllm_image, self._backend_config.default_vllm_image_tag, ) + elif engine == ENGINE_GENERIC: + # Generic containers have no platform-default image; image_name is + # required (enforced at the API layer and again in the compiler). + image_name, image_tag = generic_compiler.resolve_generic_image(view) else: image_name = view.image_name or self._backend_config.default_nimservice_image image_tag = view.image_tag or self._backend_config.default_nimservice_image_tag @@ -893,11 +898,20 @@ async def _advance_creating_container( ), True state.tool_call_plugin_path = plugin_path - # Compile engine-specific environment variables (and serve args for vLLM). - vllm_serve_args: list[str] | None = None - if engine == ENGINE_VLLM: + # Compile engine-specific environment variables (and serve args for + # arg-configured engines: vLLM and generic). NIM is configured purely + # via env, so it leaves the container command unset. + serve_args: list[str] | None = None + if engine == ENGINE_GENERIC: + # Generic: run the image with the user's raw env + args verbatim. The + # platform synthesizes nothing (no served-model-name, no LoRA, etc.). + env_vars = generic_compiler.compile_generic_env_vars(view) + generic_args = generic_compiler.compile_generic_args(view) + # Only override the image's command when the user supplied args. + serve_args = generic_args or None + elif engine == ENGINE_VLLM: env_vars = vllm_compiler.compile_vllm_env_vars(view) - vllm_serve_args = vllm_compiler.compile_vllm_args(view, state.model_entity) + serve_args = vllm_compiler.compile_vllm_args(view, state.model_entity) if view.lora_enabled: # vLLM's lora_filesystem_resolver validates that # VLLM_LORA_RESOLVER_CACHE_DIR exists at startup, before the adapter @@ -1021,10 +1035,12 @@ async def cleanup_and_error(status_message: str, error_details: dict) -> tuple[D "restart_policy": {"Name": "unless-stopped"}, } - # vLLM serve args are passed as the container command (appended to the - # image's `vllm serve` entrypoint). NIM is configured purely via env. - if vllm_serve_args is not None: - create_args["command"] = vllm_serve_args + # Serve args are passed as the container command (appended to the + # image's entrypoint). vLLM uses its compiled `vllm serve` args; + # generic uses the user's raw additional_args. NIM is configured + # purely via env and leaves the command unset. + if serve_args is not None: + create_args["command"] = serve_args if nim_config.gpu > 1: fixed = MODELS_DOCKER_NIM_MULTI_GPU_SHM_SIZE or self._backend_config.nim_multi_gpu_shm_size @@ -1055,7 +1071,10 @@ async def cleanup_and_error(status_message: str, error_details: dict) -> tuple[D container_id = container.id[:12] logger.info("Container %s started successfully (ID: %s)", container_name, container_id) - if view.lora_enabled: + # The generic engine has no LoRA semantics (no engine compiler to + # wire the adapter sidecar against), so never attach the sidecar for + # it even if lora_enabled was set. + if view.lora_enabled and engine != ENGINE_GENERIC: cfg = get_platform_config() image = get_qualified_image(self._backend_config.lora_sidecar_image_name) sidecar_envs = cfg.to_shared_envvars() diff --git a/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py b/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py new file mode 100644 index 0000000000..5a45f89a39 --- /dev/null +++ b/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py @@ -0,0 +1,57 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Backend-agnostic compiler for the ``generic`` engine. + +The ``generic`` engine runs an arbitrary container as-is: there is no +inference-engine compiler synthesizing args or env. The deployment declares a +container image + tag, a readiness-probe path, and (optionally) raw args and +environment variables; the platform runs the image and probes it. + +Like :mod:`vllm_compiler`, these functions take a :class:`DeploymentConfigView` +and return plain data (image tuple, arg vector, env dict). They are NOT specific +to any service backend: the docker backend renders the result into a +``docker run`` container; the k8s backend renders it into a native Kubernetes +Deployment. Keep this module free of backend-specific imports so both can reuse +it. + +Unlike the vLLM/NIM engines there is no platform-default image for a generic +container, so :func:`resolve_generic_image` raises when ``image_name`` is unset. +The create/update API layer validates this up front; the compiler enforces it +again as a defensive backstop. +""" + +from nmp.core.models.controllers.backends.common import DeploymentConfigView + + +def resolve_generic_image(view: DeploymentConfigView) -> tuple[str, str]: + """Resolve the generic container image name and tag. + + There is no platform default for a generic image (it is an arbitrary + user-supplied container), so ``image_name`` is required. ``image_tag`` + defaults to ``latest`` when unset, mirroring Docker/Kubernetes conventions. + """ + if not (view.image_name and view.image_name.strip()): + raise ValueError("The 'generic' engine requires executor_config.image_name to be set (no platform default).") + image_tag = view.image_tag or "latest" + return view.image_name, image_tag + + +def compile_generic_args(view: DeploymentConfigView) -> list[str]: + """Return the container arg vector for a generic container. + + The platform synthesizes nothing for the generic engine: the user's + ``additional_args`` are the entire arg vector (appended to the image's own + entrypoint). Returns an empty list when none are supplied, in which case the + image's default command/args are used unchanged. + """ + return list(view.additional_args or []) + + +def compile_generic_env_vars(view: DeploymentConfigView) -> dict[str, str]: + """Return the environment variables for a generic container. + + Only the user's ``additional_envs`` are applied; the platform injects no + engine-specific environment for a generic container. + """ + return {str(k): str(v) for k, v in (view.additional_envs or {}).items()} diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py index 19197b89ce..4ae0195c6d 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py @@ -235,22 +235,21 @@ def _resolve(self, ctx: ModelContext) -> ResolvedDeployment: def _select_reconciler(self, engine: str) -> Optional[Reconciler]: """Select the reconciler for an engine. - Returns the vLLM reconciler for ``vllm``, the NIM-operator reconciler for - any other engine (the default), and ``None`` for ``generic`` -- which the - callers treat as the "unsupported engine" rejection (see - :meth:`_unsupported_engine`). + The direct-emission :class:`K8sReconciler` handles both ``vllm`` and + ``generic`` (it branches internally on the engine); every other engine + defaults to the NIM-operator reconciler. ``None`` is reserved for a + genuinely unknown engine, which the callers treat as the "unsupported + engine" rejection (see :meth:`_unsupported_engine`). """ - if engine == ENGINE_VLLM: + if engine in (ENGINE_VLLM, ENGINE_GENERIC): return self._k8s_reconciler - if engine == ENGINE_GENERIC: - return None return self._nim_reconciler @staticmethod def _unsupported_engine(engine: str) -> DeploymentStatusUpdate: return DeploymentStatusUpdate( status="ERROR", - status_message="The 'generic' engine is not yet supported on the k8s backend.", + status_message=f"The '{engine}' engine is not supported on the k8s backend.", error_details={"error": "unsupported_engine", "engine": engine}, host_url=None, ) diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py index 38059d28b1..bd98f2d385 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py @@ -27,10 +27,15 @@ from nmp.common.config import get_platform_config from nmp.core.models.app import get_deployment_resource_name from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER -from nmp.core.models.controllers.backends import vllm_compiler +from nmp.core.models.controllers.backends import generic_compiler, vllm_compiler from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate from nmp.core.models.controllers.backends.common import DeploymentConfigView -from nmp.core.models.controllers.backends.engine import ENGINE_VLLM, resolve_health_path +from nmp.core.models.controllers.backends.engine import ( + ENGINE_GENERIC, + ENGINE_VLLM, + config_engine, + resolve_health_path, +) from nmp.core.models.controllers.backends.k8s_nim_operator import vllm_k8s_compiler from nmp.core.models.controllers.backends.k8s_nim_operator.config import K8sNimOperatorConfig from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.base import ( @@ -44,12 +49,21 @@ class K8sReconciler(Reconciler): - """Reconciles a vLLM deployment by emitting native Kubernetes objects. + """Reconciles a deployment by emitting native Kubernetes objects directly. + + Handles two engines that share this direct-emission path: - Holds its own typed API clients (CoreV1 / AppsV1 / BatchV1), composes a - :class:`StatusProjector` (serving-pod readiness/diagnostics) and a - :class:`ResourceDeleter`, and drives the staged rollout itself, advancing - creation one phase at a time as it is polled via :meth:`get_status`. + * ``vllm`` -- a staged rollout (PVC + weight-puller Job -> serving Deployment + + Service), advanced one phase at a time as it is polled via + :meth:`get_status`. + * ``generic`` -- a self-contained container image with no model weights, so + it skips the PVC/puller entirely and emits the serving Deployment + + Service immediately at create. + + The engine is read from the resolved config (:func:`config_engine`) and + branches the create/update/status paths. Holds its own typed API clients + (CoreV1 / AppsV1 / BatchV1), composes a :class:`StatusProjector` (serving-pod + readiness/diagnostics) and a :class:`ResourceDeleter`. """ def __init__( @@ -76,11 +90,16 @@ def __init__( # ------------------------------------------------------------------ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Create phase P0: emit the PVC + weight-puller Job. + """Create the deployment's backend resources. - The Deployment + Service are created later by the status path once the Job - completes (controller-side weight-readiness gating). + For ``vllm`` this is phase P0: emit the PVC + weight-puller Job (the + Deployment + Service are created later by the status path once the Job + completes -- controller-side weight-readiness gating). For ``generic`` + there are no weights to pull, so the serving Deployment + Service are + emitted immediately. """ + if config_engine(resolved.config) == ENGINE_GENERIC: + return self._create_generic(resolved) deployment = resolved.deployment logger.info( f"Creating vLLM deployment: {deployment.workspace}/{deployment.name} (version: {deployment.entity_version})" @@ -148,7 +167,12 @@ async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: Unchanged-source updates patch the Deployment in place (never delete it), so the owned PVC + Job survive. A changed source deletes the Deployment (cascading PVC + Job) and drops back to the phased create. + + The ``generic`` engine has no weights, so its update simply re-applies the + serving Deployment + Service from the latest config. """ + if config_engine(resolved.config) == ENGINE_GENERIC: + return self._update_generic(resolved) deployment = resolved.deployment logger.info( f"Updating vLLM deployment: {deployment.workspace}/{deployment.name} (version: {deployment.entity_version})" @@ -191,7 +215,12 @@ async def get_status(self, resolved: ResolvedDeployment) -> DeploymentStatusUpda Reads the puller Job + (once created) the Deployment. When the Job has completed and the Deployment doesn't exist yet, this advances creation (phase P3) by emitting the Deployment + Service. + + The ``generic`` engine has no puller phase: there is always a Deployment + once create has run, so status is just its readiness. """ + if config_engine(resolved.config) == ENGINE_GENERIC: + return self._get_status_generic(resolved) deployment = resolved.deployment resource_name = resolved.resource_name view = resolved.view @@ -321,7 +350,7 @@ def _vllm_objects_exist(self, resource_name: str) -> bool: def _has_vllm_engine_label(obj) -> bool: labels = getattr(getattr(obj, "metadata", None), "labels", None) - return isinstance(labels, dict) and labels.get("nmp.nvidia.com/engine") == ENGINE_VLLM + return isinstance(labels, dict) and labels.get("nmp.nvidia.com/engine") in (ENGINE_VLLM, ENGINE_GENERIC) try: dep = self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) @@ -464,6 +493,149 @@ def _create_vllm_serving_objects( return DeploymentStatusUpdate(status="PENDING", status_message="Starting vLLM server", host_url=None) + # ------------------------------------------------------------------ + # generic-engine helpers (no model weights / no PVC / no puller Job) + # ------------------------------------------------------------------ + + def _create_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: + """Create a generic deployment: emit the Deployment + Service immediately. + + There are no model weights to pull, so there is no PVC and no puller Job; + the serving objects are created in one shot. + """ + deployment = resolved.deployment + logger.info( + f"Creating generic deployment: {deployment.workspace}/{deployment.name} " + f"(version: {deployment.entity_version})" + ) + try: + self._create_generic_serving_objects(deployment, resolved.resource_name, resolved.view) + return DeploymentStatusUpdate( + status="PENDING", + status_message="Starting container", + host_url=self._status.host_url(resolved.resource_name), + ) + except Exception as e: + logger.error(f"Failed to create generic deployment for {deployment.workspace}/{deployment.name}: {e}") + return DeploymentStatusUpdate( + status="ERROR", + status_message=f"Failed to create deployment {deployment.workspace}/{deployment.name} due to a service backend error", + error_details={"error": str(e), "error_type": type(e).__name__}, + host_url=None, + ) + + def _update_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: + """Update a generic deployment by re-applying its serving objects. + + Generic has no weight source, so there is no re-pull decision: if the + Deployment exists it is patched in place, otherwise it is (re)created. + """ + deployment = resolved.deployment + logger.info( + f"Updating generic deployment: {deployment.workspace}/{deployment.name} " + f"(version: {deployment.entity_version})" + ) + try: + self._create_generic_serving_objects(deployment, resolved.resource_name, resolved.view) + return DeploymentStatusUpdate( + status="PENDING", + status_message="Update accepted", + host_url=self._status.host_url(resolved.resource_name), + ) + except Exception as e: + logger.error(f"Failed to update generic deployment for {deployment.workspace}/{deployment.name}: {e}") + return DeploymentStatusUpdate( + status="ERROR", + status_message=f"Failed to update deployment {deployment.workspace}/{deployment.name} due to a service backend error", + error_details={"error": str(e), "error_type": type(e).__name__}, + host_url=None, + ) + + def _get_status_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: + """Project a generic deployment's status from its serving Deployment. + + The Deployment is created at create time (no staged rollout), so a 404 + means it was deleted externally -> LOST. + """ + resource_name = resolved.resource_name + try: + self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) + except k8s_client.exceptions.ApiException as e: + if e.status != 404: + raise + return DeploymentStatusUpdate( + status="LOST", + status_message="Serving Deployment not found; resources may have been deleted externally.", + host_url=None, + ) + return self._project_deployment_readiness(resource_name) + + def _create_generic_serving_objects( + self, + deployment: ModelDeployment, + resource_name: str, + view: DeploymentConfigView, + ) -> None: + """Create (or no-op if present) the generic Deployment + Service. + + Mirrors the vLLM serving-object creation but without any PVC/model-store + mount: a generic container runs purely from its image. uid/gid are left + unset so the image's own user runs (an arbitrary container should not be + forced into vLLM's or NIM's user). + """ + engine = ENGINE_GENERIC + health_path = resolve_health_path(engine, view) + image_name, image_tag = generic_compiler.resolve_generic_image(view) + args = generic_compiler.compile_generic_args(view) + env = generic_compiler.compile_generic_env_vars(view) + + startup_grace = self._backend_config.default_startup_probe_grace_period_seconds or 600 + + dep_obj = vllm_k8s_compiler.compile_deployment( + resource_name=resource_name, + workspace=deployment.workspace, + name=deployment.name, + engine=engine, + image=f"{image_name}:{image_tag}", + args=args, + health_path=health_path, + env=env, + gpu=view.gpu, + namespace=self._k8s_namespace, + service_account_name=self._backend_config.service_account_name, + user_id=None, + group_id=None, + shared_memory_size_limit=self._backend_config.default_shared_memory_size_limit, + startup_grace_seconds=startup_grace, + mount_model_store=False, + ) + svc_obj = vllm_k8s_compiler.compile_service( + resource_name=resource_name, + workspace=deployment.workspace, + name=deployment.name, + engine=engine, + namespace=self._k8s_namespace, + ) + + try: + created_dep = self._apps_v1.create_namespaced_deployment(namespace=self._k8s_namespace, body=dep_obj) + logger.info(f"Created generic Deployment {resource_name} in {self._k8s_namespace}") + except k8s_client.exceptions.ApiException as e: + if e.status != 409: + raise + created_dep = self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) + + owner_ref = k8s_client.V1OwnerReference( + api_version="apps/v1", + kind="Deployment", + name=created_dep.metadata.name, + uid=created_dep.metadata.uid, + controller=True, + block_owner_deletion=True, + ) + svc_obj.metadata.owner_references = [owner_ref] + self._create_or_skip(self._core_v1.create_namespaced_service, svc_obj, "Service") + def _delete_puller_job(self, resource_name: str) -> bool: """Delete the puller Job and confirm its pod is gone (releases RWO volume). diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/vllm_k8s_compiler.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/vllm_k8s_compiler.py index f981e298bc..32633a45dc 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/vllm_k8s_compiler.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/vllm_k8s_compiler.py @@ -307,6 +307,7 @@ def compile_deployment( init_containers: Optional[list[k8s_client.V1Container]] = None, sidecar_containers: Optional[list[k8s_client.V1Container]] = None, extra_labels: Optional[dict[str, str]] = None, + mount_model_store: bool = True, ) -> k8s_client.V1Deployment: """Compile the inference-server Deployment. @@ -316,6 +317,11 @@ def compile_deployment( startup/readiness probes. A ``dshm`` emptyDir is always mounted at ``/dev/shm`` (vLLM uses it for tensor-parallel NCCL); ``scratch`` is mounted for the LoRA cache dir. + + ``mount_model_store`` controls whether the ``model-store`` PVC volume + mount + are attached. The vLLM/NIM weight-pull paths set it ``True`` (the PVC holds + the pulled weights). The ``generic`` engine pulls no weights and has no PVC, + so it passes ``False`` -- the container runs purely from its image. """ selector_labels = {"app": resource_name} pod_labels = { @@ -330,10 +336,13 @@ def compile_deployment( failure_threshold = max(1, -(-startup_grace_seconds // period)) # ceil volume_mounts = [ - k8s_client.V1VolumeMount(name="model-store", mount_path=MODEL_STORE_PATH, read_only=True), k8s_client.V1VolumeMount(name="scratch", mount_path=SCRATCH_PATH), k8s_client.V1VolumeMount(name="dshm", mount_path=DSHM_PATH), ] + if mount_model_store: + volume_mounts.insert( + 0, k8s_client.V1VolumeMount(name="model-store", mount_path=MODEL_STORE_PATH, read_only=True) + ) container = k8s_client.V1Container( name=f"{resource_name}-ctr", @@ -352,19 +361,23 @@ def compile_deployment( containers.extend(sidecar_containers) volumes = [ - k8s_client.V1Volume( - name="model-store", - persistent_volume_claim=k8s_client.V1PersistentVolumeClaimVolumeSource( - claim_name=pvc_name(resource_name), - read_only=True, - ), - ), k8s_client.V1Volume(name="scratch", empty_dir=k8s_client.V1EmptyDirVolumeSource()), k8s_client.V1Volume( name="dshm", empty_dir=k8s_client.V1EmptyDirVolumeSource(medium="Memory", size_limit=shared_memory_size_limit), ), ] + if mount_model_store: + volumes.insert( + 0, + k8s_client.V1Volume( + name="model-store", + persistent_volume_claim=k8s_client.V1PersistentVolumeClaimVolumeSource( + claim_name=pvc_name(resource_name), + read_only=True, + ), + ), + ) pod_spec = k8s_client.V1PodSpec( service_account_name=service_account_name, diff --git a/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py b/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py new file mode 100644 index 0000000000..d544b8a589 --- /dev/null +++ b/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py @@ -0,0 +1,69 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the backend-agnostic generic-engine compiler.""" + +import pytest +from nmp.core.models.controllers.backends import generic_compiler +from nmp.core.models.controllers.backends.common import DeploymentConfigView + + +def _view(**kwargs) -> DeploymentConfigView: + return DeploymentConfigView(**kwargs) + + +# --------------------------------------------------------------------------- +# Image resolution +# --------------------------------------------------------------------------- + + +def test_resolve_generic_image_uses_config_image_and_tag(): + view = _view(gpu=0, image_name="nvcr.io/nim/nvidia/nemoguard-jailbreak-detect", image_tag="1.10.1") + name, tag = generic_compiler.resolve_generic_image(view) + assert name == "nvcr.io/nim/nvidia/nemoguard-jailbreak-detect" + assert tag == "1.10.1" + + +def test_resolve_generic_image_defaults_tag_to_latest(): + view = _view(gpu=0, image_name="my/container", image_tag=None) + name, tag = generic_compiler.resolve_generic_image(view) + assert name == "my/container" + assert tag == "latest" + + +def test_resolve_generic_image_requires_image_name(): + """There is no platform default for a generic image, so image_name is required.""" + view = _view(gpu=0, image_name=None) + with pytest.raises(ValueError, match="image_name"): + generic_compiler.resolve_generic_image(view) + + +def test_resolve_generic_image_rejects_blank_image_name(): + view = _view(gpu=0, image_name=" ") + with pytest.raises(ValueError, match="image_name"): + generic_compiler.resolve_generic_image(view) + + +# --------------------------------------------------------------------------- +# Args + env passthrough (the platform synthesizes nothing for generic) +# --------------------------------------------------------------------------- + + +def test_compile_generic_args_passthrough(): + view = _view(gpu=0, image_name="x", additional_args=["--port", "9000", "--foo"]) + assert generic_compiler.compile_generic_args(view) == ["--port", "9000", "--foo"] + + +def test_compile_generic_args_empty_when_none(): + view = _view(gpu=0, image_name="x", additional_args=None) + assert generic_compiler.compile_generic_args(view) == [] + + +def test_compile_generic_env_passthrough_stringifies(): + view = _view(gpu=0, image_name="x", additional_envs={"A": "1", "B": 2}) + assert generic_compiler.compile_generic_env_vars(view) == {"A": "1", "B": "2"} + + +def test_compile_generic_env_empty_when_none(): + view = _view(gpu=0, image_name="x", additional_envs=None) + assert generic_compiler.compile_generic_env_vars(view) == {} diff --git a/services/core/models/tests/unit/controllers/backends/test_vllm_k8s_compiler.py b/services/core/models/tests/unit/controllers/backends/test_vllm_k8s_compiler.py index 75a5488031..6b56d82a7e 100644 --- a/services/core/models/tests/unit/controllers/backends/test_vllm_k8s_compiler.py +++ b/services/core/models/tests/unit/controllers/backends/test_vllm_k8s_compiler.py @@ -187,6 +187,34 @@ def test_compile_deployment_basic(): assert vols["dshm"].empty_dir.medium == "Memory" +def test_compile_deployment_no_model_store_for_generic(): + """mount_model_store=False omits the model-store PVC volume + mount (generic engine).""" + dep = c.compile_deployment( + resource_name="md-default-jb", + workspace="default", + name="jb", + engine="generic", + image="nvcr.io/nim/nvidia/nemoguard-jailbreak-detect:1.10.1", + args=["--port", "8000"], + health_path="/v1/health/ready", + gpu=0, + mount_model_store=False, + ) + pod = dep.spec.template.spec + ctr = pod.containers[0] + + mount_names = {m.name for m in ctr.volume_mounts} + assert "model-store" not in mount_names + # scratch + dshm are still present (harmless emptyDirs). + assert "scratch" in mount_names + assert "dshm" in mount_names + + vol_names = {v.name for v in pod.volumes} + assert "model-store" not in vol_names + assert "scratch" in vol_names + assert "dshm" in vol_names + + def test_compile_deployment_cpu_only_no_gpu(): dep = c.compile_deployment( resource_name="r", diff --git a/services/core/models/tests/unit/controllers/test_docker_backend.py b/services/core/models/tests/unit/controllers/test_docker_backend.py index 63298357c1..1b3ee62800 100644 --- a/services/core/models/tests/unit/controllers/test_docker_backend.py +++ b/services/core/models/tests/unit/controllers/test_docker_backend.py @@ -40,6 +40,7 @@ "disk_size", "image_name", "image_tag", + "health_check_path", "additional_envs", "additional_args", "k8s_nim_operator_config", @@ -511,6 +512,60 @@ async def test_docker_backend_create_vllm_lora_sidecar(docker_backend, sample_de ) +@pytest.mark.asyncio +async def test_docker_backend_create_generic_deployment(docker_backend, sample_deployment, mock_docker_client): + """Engine=generic runs the user's image + raw args/env verbatim, no puller, no LoRA sidecar.""" + config = MagicMock() + set_deployment_config( + config, + engine="generic", + gpu=0, + image_name="nvcr.io/nim/nvidia/nemoguard-jailbreak-detect", + image_tag="1.10.1", + health_check_path="/v1/health/ready", + additional_args=["--port", "8000"], + additional_envs={"FOO": "bar"}, + ) + + mock_container = MagicMock() + mock_container.id = "generic12345678" + mock_container.start = MagicMock() + mock_docker_client.containers.create.return_value = mock_container + mock_docker_client.images.get.return_value = MagicMock() + mock_docker_client.containers.list.return_value = [] + + status_update = await drive_creation_to_completion_after_create( + docker_backend, sample_deployment, config, mock_docker_client + ) + + assert status_update.status == "PENDING" + + # Self-contained image: no model puller ran (generic has no model weights). + run_commands = [c.kwargs.get("command") for c in mock_docker_client.containers.run.call_args_list] + assert not any(isinstance(cmd, list) and any("download" in str(p) for p in cmd) for cmd in run_commands) + + # Exactly one container (no LoRA sidecar). + assert mock_docker_client.containers.create.call_count == 1 + create_args = mock_docker_client.containers.create.call_args_list[0][1] + assert create_args["image"] == "nvcr.io/nim/nvidia/nemoguard-jailbreak-detect:1.10.1" + # Raw additional_args become the container command verbatim. + assert create_args["command"] == ["--port", "8000"] + # Raw additional_envs become the env verbatim. + assert create_args["environment"]["FOO"] == "bar" + # Engine + explicit health-path labels recorded for status-time probe selection. + assert create_args["labels"]["nmp.nvidia.com/engine"] == "generic" + assert create_args["labels"]["nmp.nvidia.com/health-path"] == "/v1/health/ready" + + +async def drive_creation_to_completion_after_create(docker_backend, sample_deployment, config, mock_docker_client): + """Start creation for a no-puller config and drive it to completion.""" + await docker_backend.create_model_deployment( + ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) + ) + mock_docker_client.containers.get.side_effect = NotFound("Container not found") + return await drive_creation_to_completion(docker_backend, sample_deployment) + + def test_get_health_path_from_container_vllm(docker_backend): """vLLM containers probe /health.""" container = MagicMock() diff --git a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py index 369b6a578b..f7a161433e 100644 --- a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py +++ b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py @@ -1991,18 +1991,103 @@ async def test_vllm_create_emits_pvc_and_job_only(k8s_backend, sample_deployment assert job.spec.template.spec.containers[0].resources.requests["nvidia.com/gpu"] == "2" +def _generic_config(*, gpu: int = 0, image="nvcr.io/nim/nvidia/nemoguard-jailbreak-detect", tag="1.10.1"): + """A minimal generic ModelDeploymentConfig-like object (no model weights).""" + return SimpleNamespace( + engine="generic", + model_spec=SimpleNamespace( + model_type=None, + model_namespace=None, + model_name=None, + model_revision=None, + chat_template=None, + tool_call_config=None, + lora_enabled=False, + ), + executor_config=SimpleNamespace( + gpu=gpu, + disk_size="50Gi", + image_name=image, + image_tag=tag, + health_check_path="/v1/health/ready", + additional_envs={"FOO": "bar"}, + additional_args=["--port", "8000"], + k8s_nim_operator_config=None, + override_config=None, + ), + ) + + @pytest.mark.asyncio -async def test_generic_engine_rejected_on_k8s(k8s_backend, sample_deployment): - """The generic engine is explicitly unsupported on the k8s backend.""" +async def test_generic_create_emits_deployment_and_service_no_pvc(k8s_backend, sample_deployment): + """Generic create emits the Deployment + Service immediately, with no PVC/puller Job.""" backend = _vllm_backend(k8s_backend) - config = _vllm_config() - config.engine = "generic" + config = _generic_config() + + created_dep = MagicMock() + created_dep.metadata.name = backend._get_resource_name(sample_deployment) + created_dep.metadata.uid = "dep-uid" + backend._apps_v1.create_namespaced_deployment.return_value = created_dep + _sync_reconcilers(backend) result = await backend.create_model_deployment( ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) ) - assert result.status == "ERROR" - assert "generic" in result.status_message.lower() + + assert result.status == "PENDING" + backend._apps_v1.create_namespaced_deployment.assert_called_once() + backend._core_v1.create_namespaced_service.assert_called_once() + # No model weights for generic: no PVC, no puller Job. + backend._core_v1.create_namespaced_persistent_volume_claim.assert_not_called() + backend._batch_v1.create_namespaced_job.assert_not_called() + + # The container runs the user's image + raw args + env verbatim, with no + # model-store volume mounted. + dep_obj = backend._apps_v1.create_namespaced_deployment.call_args.kwargs["body"] + container = dep_obj.spec.template.spec.containers[0] + assert container.image == "nvcr.io/nim/nvidia/nemoguard-jailbreak-detect:1.10.1" + assert container.args == ["--port", "8000"] + assert {e.name: e.value for e in container.env} == {"FOO": "bar"} + volume_names = {v.name for v in dep_obj.spec.template.spec.volumes} + assert "model-store" not in volume_names + mount_names = {m.name for m in container.volume_mounts} + assert "model-store" not in mount_names + # Readiness/startup probes use the explicit health_check_path. + assert container.readiness_probe.http_get.path == "/v1/health/ready" + + +@pytest.mark.asyncio +async def test_generic_status_ready_when_deployment_ready(k8s_backend, sample_deployment): + """Generic status projects the serving Deployment's readiness directly (no Job/PVC).""" + backend = _vllm_backend(k8s_backend) + config = _generic_config() + + dep = MagicMock() + dep.status.ready_replicas = 1 + backend._apps_v1.read_namespaced_deployment.return_value = dep + + _sync_reconcilers(backend) + result = await backend.get_model_deployment_status( + ModelContext(model_deployment=sample_deployment, model_deployment_config=config) + ) + + assert result.status == "READY" + # Generic status never consults the puller Job. + backend._batch_v1.read_namespaced_job.assert_not_called() + + +@pytest.mark.asyncio +async def test_generic_status_lost_when_deployment_missing(k8s_backend, sample_deployment): + """Generic status reports LOST when the serving Deployment was deleted externally.""" + backend = _vllm_backend(k8s_backend) + config = _generic_config() + backend._apps_v1.read_namespaced_deployment.side_effect = _api_exception(404) + + _sync_reconcilers(backend) + result = await backend.get_model_deployment_status( + ModelContext(model_deployment=sample_deployment, model_deployment_config=config) + ) + assert result.status == "LOST" @pytest.mark.asyncio diff --git a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py index 4b293b1ddc..433bd8eb79 100644 --- a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py +++ b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py @@ -164,6 +164,53 @@ async def test_create_deployment_config_already_exists( mock_entity_client.create.assert_not_called() +@pytest.mark.asyncio +async def test_create_generic_config_requires_image_and_health_path(deployment_config_service, mock_entity_client): + """A generic config missing image_name or health_check_path is rejected at create.""" + mock_list_result = MagicMock() + mock_list_result.data = [] + mock_entity_client.list.return_value = mock_list_result + + request = CreateModelDeploymentConfigRequest( + name="generic-config", + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(), + executor_config=ContainerExecutorConfig(gpu=0), # no image_name / health_check_path + ) + + with pytest.raises(ValueError, match="image_name"): + await deployment_config_service.create_deployment_config(request, "default") + mock_entity_client.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_generic_config_succeeds_when_image_and_health_path_set( + deployment_config_service, mock_entity_client, sample_config_entity +): + """A generic config with image_name + health_check_path passes validation.""" + mock_list_result = MagicMock() + mock_list_result.data = [] + mock_entity_client.list.return_value = mock_list_result + mock_entity_client.create.return_value = sample_config_entity + + request = CreateModelDeploymentConfigRequest( + name="generic-config", + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(), + executor_config=ContainerExecutorConfig( + gpu=0, + image_name="nvcr.io/nim/nvidia/nemoguard-jailbreak-detect", + image_tag="1.10.1", + health_check_path="/v1/health/ready", + ), + model_entity_id="model-entity-123", + ) + + result = await deployment_config_service.create_deployment_config(request, "default") + assert result is not None + mock_entity_client.create.assert_called_once() + + @pytest.mark.asyncio async def test_get_deployment_config_found(deployment_config_service, mock_entity_client, sample_config_entity): """Test retrieving an existing deployment config.""" From f35c40dfd3725bc5b6ba9a4750351c71f89783f0 Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Thu, 25 Jun 2026 15:56:56 -0600 Subject: [PATCH 2/3] fix(models): weight-aware generic engine + PR review fixes Address CodeRabbit review feedback on generic engine support: - Make generic deployments weight-aware on both backends: when the config resolves to a fileset-backed model, pull weights (docker puller / k8s PVC+puller) and mount /model-store; otherwise run the image raw with no platform volumes. - k8s: refactor K8sReconciler to choose staged vs. immediate rollout by weight presence (not engine), patch serving Deployment/Service in place on update, and reject unknown engines explicitly instead of defaulting to NIM. - Validate + reject whitespace-padded generic image_name/health_check_path; trim defensively in the generic compiler. - Expand unit tests (validation, compiler, docker + k8s weight-aware paths, update patching). Signed-off-by: Ben McCown --- .../model_deployment_config_service.py | 19 +- .../backends/docker/creation_reconciler.py | 46 +- .../controllers/backends/generic_compiler.py | 8 +- .../backends/k8s_nim_operator/backend.py | 15 +- .../k8s_nim_operator/reconcilers/k8s.py | 513 ++++++++---------- .../backends/test_generic_compiler.py | 14 + .../unit/controllers/test_docker_backend.py | 36 ++ .../test_k8s_nim_operator_backend.py | 129 ++++- ...st_model_deployment_config_service_unit.py | 64 +++ 9 files changed, 518 insertions(+), 326 deletions(-) diff --git a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py index f30733e4e2..87e78d694e 100644 --- a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py +++ b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py @@ -34,20 +34,31 @@ def _validate_engine_config(engine: Engine, executor_config: ContainerExecutorCo endpoint. Both ``image_name`` and ``health_check_path`` must therefore be supplied explicitly; the other engines fall back to their configured defaults when these are unset. + + Values are also rejected when they contain surrounding whitespace: an + image reference or probe path is used verbatim downstream, where a + leading/trailing space would silently produce an invalid value. """ if engine != Engine.GENERIC: return missing: list[str] = [] - if not (executor_config.image_name and executor_config.image_name.strip()): - missing.append("image_name") - if not (executor_config.health_check_path and executor_config.health_check_path.strip()): - missing.append("health_check_path") + padded: list[str] = [] + for field in ("image_name", "health_check_path"): + value = getattr(executor_config, field) + if not (value and value.strip()): + missing.append(field) + elif value != value.strip(): + padded.append(field) if missing: raise ValueError( "The 'generic' engine requires executor_config." + " and executor_config.".join(missing) + " to be set (no platform default exists for a generic container)." ) + if padded: + raise ValueError( + "executor_config." + " and executor_config.".join(padded) + " must not have leading or trailing whitespace." + ) class ReferentialIntegrityError(Exception): diff --git a/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py b/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py index b24557f57c..4bade83418 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py @@ -503,20 +503,24 @@ async def register_deployment( full_image = f"{image_name}:{image_tag}" logger.info(f"Using image: {full_image} (engine={engine})") - # Create volumes for model cache + # Create volumes for model cache + scratch. A generic container that pulls + # no weights runs raw (no platform volumes mounted -- see container create + # below), so skip provisioning them; every other case mounts them. volume_name = self.get_volume_name(deployment.workspace, deployment.name) - try: - await asyncio.to_thread(self.create_volume, volume_name) - logger.info(f"Created volume: {volume_name}") - except Exception as e: - logger.warning(f"Failed to create volume {volume_name} (may already exist): {e}") - scratch_volume_name = volume_name + "-scratch" - try: - await asyncio.to_thread(self.create_volume, scratch_volume_name) - logger.info(f"Created volume: {scratch_volume_name}") - except Exception as e: - logger.warning(f"Failed to create volume {scratch_volume_name} (may already exist): {e}") + provision_volumes = engine != ENGINE_GENERIC or weights_from_files + if provision_volumes: + try: + await asyncio.to_thread(self.create_volume, volume_name) + logger.info(f"Created volume: {volume_name}") + except Exception as e: + logger.warning(f"Failed to create volume {volume_name} (may already exist): {e}") + + try: + await asyncio.to_thread(self.create_volume, scratch_volume_name) + logger.info(f"Created volume: {scratch_volume_name}") + except Exception as e: + logger.warning(f"Failed to create volume {scratch_volume_name} (may already exist): {e}") # Multi-LLM detection only applies to NIM images; vLLM/generic are never multi-LLM. is_multi_llm = False @@ -1015,16 +1019,26 @@ async def cleanup_and_error(status_message: str, error_details: dict) -> tuple[D try: logger.info("Creating container %s with image %s...", container_name, full_image) + # Platform volumes (/model-store, /scratch) hold pulled weights + scratch + # space. NIM/vLLM always mount them. A generic container runs the user's + # image as-is, so only mount them when the platform actually pulls weights + # for it (a fileset-backed model deployment); otherwise the mounts would + # shadow the image's own contents at those paths. + mount_platform_volumes = engine != ENGINE_GENERIC or self._needs_puller(state) + volumes: dict[str, Any] = {} + if mount_platform_volumes: + volumes = { + state.volume_name: {"bind": "/model-store", "mode": "rw"}, + state.scratch_volume_name: {"bind": "/scratch", "mode": "rw"}, + } + create_args: dict[str, Any] = { "image": full_image, "name": container_name, "environment": env_vars, "detach": True, "device_requests": device_requests, - "volumes": { - state.volume_name: {"bind": "/model-store", "mode": "rw"}, - state.scratch_volume_name: {"bind": "/scratch", "mode": "rw"}, - }, + "volumes": volumes, "labels": { "nmp.nvidia.com/deployment-workspace": deployment.workspace, "nmp.nvidia.com/deployment-name": deployment.name, diff --git a/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py b/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py index 5a45f89a39..2df6a85245 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/generic_compiler.py @@ -30,11 +30,15 @@ def resolve_generic_image(view: DeploymentConfigView) -> tuple[str, str]: There is no platform default for a generic image (it is an arbitrary user-supplied container), so ``image_name`` is required. ``image_tag`` defaults to ``latest`` when unset, mirroring Docker/Kubernetes conventions. + + Both values are trimmed: the API layer already rejects whitespace-padded + inputs for generic configs, but this stays robust to any other call path. """ if not (view.image_name and view.image_name.strip()): raise ValueError("The 'generic' engine requires executor_config.image_name to be set (no platform default).") - image_tag = view.image_tag or "latest" - return view.image_name, image_tag + image_name = view.image_name.strip() + image_tag = (view.image_tag or "").strip() or "latest" + return image_name, image_tag def compile_generic_args(view: DeploymentConfigView) -> list[str]: diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py index 4ae0195c6d..5d813aa24f 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py @@ -37,7 +37,7 @@ deployment_config_view, deployment_elapsed_seconds, ) -from nmp.core.models.controllers.backends.engine import ENGINE_GENERIC, ENGINE_VLLM, config_engine +from nmp.core.models.controllers.backends.engine import ENGINE_GENERIC, ENGINE_NIM, ENGINE_VLLM, config_engine from nmp.core.models.controllers.backends.k8s_nim_operator.config import K8sNimOperatorConfig from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.base import Reconciler, ResolvedDeployment from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.k8s import K8sReconciler @@ -235,15 +235,16 @@ def _resolve(self, ctx: ModelContext) -> ResolvedDeployment: def _select_reconciler(self, engine: str) -> Optional[Reconciler]: """Select the reconciler for an engine. - The direct-emission :class:`K8sReconciler` handles both ``vllm`` and - ``generic`` (it branches internally on the engine); every other engine - defaults to the NIM-operator reconciler. ``None`` is reserved for a - genuinely unknown engine, which the callers treat as the "unsupported - engine" rejection (see :meth:`_unsupported_engine`). + The direct-emission :class:`K8sReconciler` handles ``vllm`` and + ``generic``; ``nim`` uses the NIM-operator reconciler. Any other value is + unsupported and yields ``None``, which the callers turn into the + "unsupported engine" rejection (see :meth:`_unsupported_engine`). """ if engine in (ENGINE_VLLM, ENGINE_GENERIC): return self._k8s_reconciler - return self._nim_reconciler + if engine == ENGINE_NIM: + return self._nim_reconciler + return None @staticmethod def _unsupported_engine(engine: str) -> DeploymentStatusUpdate: diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py index bd98f2d385..787c91ebeb 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py @@ -1,21 +1,29 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Direct-emission Kubernetes reconciler for the vLLM engine. +"""Direct-emission Kubernetes reconciler for the vLLM and generic engines. Emits native Kubernetes objects (PVC / weight-puller Job / Deployment / Service) -directly -- there is no operator. Creation is staged and driven from -``get_status``: - -* P0 (``create``): emit the PVC + weight-puller Job. The serving Deployment + - Service are intentionally NOT created yet so the controller can gate on weight - readiness. -* P3 (in ``get_status``, once the puller Job succeeds): delete the completed - puller Job to release its ReadWriteOnce volume, then emit the serving - Deployment + Service with ownerReferences so a later delete cascades. - -Inputs arrive pre-resolved on a :class:`ResolvedDeployment` (the ServiceBackend -does the SDK / entity-shaping work); this reconciler talks only to Kubernetes. +directly -- there is no operator. Whether creation is staged depends on whether +the deployment has platform-managed weights (a fileset-backed model), NOT on the +engine: + +* **Weighted** (vLLM always; generic when a fileset is present): a staged rollout. + * P0 (``create``): emit the PVC + weight-puller Job. The serving Deployment + + Service are intentionally NOT created yet so the controller can gate on weight + readiness. + * P3 (in ``get_status``, once the puller Job succeeds): delete the completed + puller Job to release its ReadWriteOnce volume, then emit the serving + Deployment + Service with ownerReferences so a later delete cascades. +* **Weightless** (generic with no fileset): the serving Deployment + Service are + emitted immediately at ``create`` -- no PVC, no puller Job, no ``/model-store`` + mount; the container runs purely from its image. + +The engine selects only the compiler (image/args/env), the pod uid/gid, and +whether the LoRA sidecar is wired; the staged-vs-immediate decision is driven by +weight presence. Inputs arrive pre-resolved on a :class:`ResolvedDeployment` (the +ServiceBackend does the SDK / entity-shaping work); this reconciler talks only to +Kubernetes. """ from logging import getLogger @@ -25,11 +33,10 @@ from nemo_platform.types.inference.model_deployment import ModelDeployment from nemo_platform.types.models.model_entity import ModelEntity from nmp.common.config import get_platform_config -from nmp.core.models.app import get_deployment_resource_name +from nmp.core.models.app import ModelWeightsType, get_deployment_resource_name from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER from nmp.core.models.controllers.backends import generic_compiler, vllm_compiler from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate -from nmp.core.models.controllers.backends.common import DeploymentConfigView from nmp.core.models.controllers.backends.engine import ( ENGINE_GENERIC, ENGINE_VLLM, @@ -51,17 +58,18 @@ class K8sReconciler(Reconciler): """Reconciles a deployment by emitting native Kubernetes objects directly. - Handles two engines that share this direct-emission path: + Handles the ``vllm`` and ``generic`` engines, which share this direct-emission + path. The rollout shape is chosen by **weight presence**, not engine: - * ``vllm`` -- a staged rollout (PVC + weight-puller Job -> serving Deployment - + Service), advanced one phase at a time as it is polled via - :meth:`get_status`. - * ``generic`` -- a self-contained container image with no model weights, so - it skips the PVC/puller entirely and emits the serving Deployment + - Service immediately at create. + * Weighted (vLLM always; generic with a fileset) -- a staged rollout + (PVC + weight-puller Job -> serving Deployment + Service), advanced one phase + at a time as it is polled via :meth:`get_status`. + * Weightless (generic with no fileset) -- the serving Deployment + Service are + emitted immediately at create, with no PVC/puller and no ``/model-store`` + mount. - The engine is read from the resolved config (:func:`config_engine`) and - branches the create/update/status paths. Holds its own typed API clients + The engine (:func:`config_engine`) selects only the compiler, uid/gid, and + LoRA wiring (see :meth:`_serving_plan`). Holds its own typed API clients (CoreV1 / AppsV1 / BatchV1), composes a :class:`StatusProjector` (serving-pod readiness/diagnostics) and a :class:`ResourceDeleter`. """ @@ -89,20 +97,37 @@ def __init__( # Reconciler interface # ------------------------------------------------------------------ + @staticmethod + def _has_weights(resolved: ResolvedDeployment) -> bool: + """True when the platform pulls weights for this deployment. + + vLLM always pulls weights (it serves a model from the Files service). + Generic is weightless by default and only pulls weights when its config + resolves to a Files-service model (a fileset-backed entity). This -- not + the engine alone -- decides staged vs. immediate rollout. + """ + if config_engine(resolved.config) == ENGINE_GENERIC: + return resolved.weights_type == ModelWeightsType.FILES_SERVICE + # vLLM (the only other engine routed here) is always weighted. + return True + async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: """Create the deployment's backend resources. - For ``vllm`` this is phase P0: emit the PVC + weight-puller Job (the + Weighted deployments do phase P0: emit the PVC + weight-puller Job (the Deployment + Service are created later by the status path once the Job - completes -- controller-side weight-readiness gating). For ``generic`` - there are no weights to pull, so the serving Deployment + Service are + completes -- controller-side weight-readiness gating). Weightless generic + deployments have nothing to pull, so the serving Deployment + Service are emitted immediately. """ - if config_engine(resolved.config) == ENGINE_GENERIC: - return self._create_generic(resolved) + if not self._has_weights(resolved): + return self._create_serving_objects(resolved) + deployment = resolved.deployment + engine = config_engine(resolved.config) logger.info( - f"Creating vLLM deployment: {deployment.workspace}/{deployment.name} (version: {deployment.entity_version})" + f"Creating {engine} deployment: {deployment.workspace}/{deployment.name} " + f"(version: {deployment.entity_version})" ) try: resource_name = resolved.resource_name @@ -110,13 +135,14 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: model_repo, source_tag = self._model_source(resolved) disk_size = view.disk_size or self._backend_config.default_pvc_size if resolved.files_hf_url is None: - raise ValueError("Cannot create vLLM deployment: Files HF endpoint was not resolved") + raise ValueError(f"Cannot create {engine} deployment: Files HF endpoint was not resolved") + user_id, group_id = self._pod_user(engine) pvc = vllm_k8s_compiler.compile_pvc( resource_name=resource_name, workspace=deployment.workspace, name=deployment.name, - engine=ENGINE_VLLM, + engine=engine, disk_size=disk_size, storage_class=self._backend_config.default_storage_class, model_source=source_tag, @@ -127,7 +153,7 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: resource_name=resource_name, workspace=deployment.workspace, name=deployment.name, - engine=ENGINE_VLLM, + engine=engine, image=self._huggingface_model_puller, container_args=["download", model_repo, "--local-dir", vllm_k8s_compiler.MODEL_STORE_PATH], env={"HF_ENDPOINT": resolved.files_hf_url, "HF_TOKEN": "service:models"}, @@ -135,11 +161,8 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: namespace=self._k8s_namespace, service_account_name=self._backend_config.service_account_name, image_pull_secret=self._backend_config.huggingface_model_puller_image_pull_secret, - # Engine-specific uid/gid: vLLM uses 2000/0 (its image's user). A - # future NIM raw-object path must pass NIM's own uid/gid here, not - # these -- see the FUTURE note in vllm_k8s_compiler.py. - user_id=self._backend_config.default_vllm_user_id, - group_id=self._backend_config.default_vllm_group_id, + user_id=user_id, + group_id=group_id, model_source=source_tag, ) @@ -152,7 +175,7 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: host_url=self._status.host_url(resource_name), ) except Exception as e: - logger.error(f"Failed to create vLLM deployment for {deployment.workspace}/{deployment.name}: {e}") + logger.error(f"Failed to create {engine} deployment for {deployment.workspace}/{deployment.name}: {e}") return DeploymentStatusUpdate( status="ERROR", status_message=f"Failed to create deployment {deployment.workspace}/{deployment.name} due to a service backend error", @@ -161,39 +184,49 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: ) async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Update a vLLM deployment, applying the re-pull policy. + """Update a deployment. - Weights are only re-pulled when the model source (name/revision) changes. - Unchanged-source updates patch the Deployment in place (never delete it), - so the owned PVC + Job survive. A changed source deletes the Deployment - (cascading PVC + Job) and drops back to the phased create. + For weighted deployments, weights are only re-pulled when the model source + (name/revision) changes: a changed source deletes the objects (cascading + PVC + Job) and drops back to the phased create; an unchanged source patches + the serving Deployment + Service in place if they exist, else lets the + status path create them at P3. - The ``generic`` engine has no weights, so its update simply re-applies the - serving Deployment + Service from the latest config. + Weightless generic deployments have no weight source, so update just + patches (or creates) the serving Deployment + Service from the latest + config. """ - if config_engine(resolved.config) == ENGINE_GENERIC: - return self._update_generic(resolved) deployment = resolved.deployment + engine = config_engine(resolved.config) + resource_name = resolved.resource_name logger.info( - f"Updating vLLM deployment: {deployment.workspace}/{deployment.name} (version: {deployment.entity_version})" + f"Updating {engine} deployment: {deployment.workspace}/{deployment.name} " + f"(version: {deployment.entity_version})" ) try: - resource_name = resolved.resource_name - _, source_tag = self._model_source(resolved) + if not self._has_weights(resolved): + # No weights => no PVC/puller; the serving objects are the whole + # deployment. Re-apply them (patch in place if present). + self._apply_serving_objects(resolved) + return DeploymentStatusUpdate( + status="PENDING", + status_message="Update accepted", + host_url=self._status.host_url(resource_name), + ) + _, source_tag = self._model_source(resolved) existing_source = self._existing_model_source(resource_name) if existing_source is not None and existing_source != source_tag: logger.info( f"Model source changed ({existing_source} -> {source_tag}); re-pulling weights for {resource_name}" ) - self._delete_vllm_resources(resource_name) + self._delete_serving_resources(resource_name) return await self.create(resolved) - # Unchanged source: patch the Deployment + Service in place if present, - # else (still in the pull phase) recreate the puller objects if missing. - if self._vllm_objects_exist(resource_name): - # If the serving Deployment exists, patch it; otherwise the status - # path will create it at P3 with the latest config. + # Unchanged source: patch the serving Deployment + Service in place if + # present, else (still in the pull phase) the status path creates them. + if self._serving_deployment_exists(resource_name): + self._apply_serving_objects(resolved) return DeploymentStatusUpdate( status="PENDING", status_message="Update accepted", @@ -201,7 +234,7 @@ async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: ) return await self.create(resolved) except Exception as e: - logger.error(f"Failed to update vLLM deployment for {deployment.workspace}/{deployment.name}: {e}") + logger.error(f"Failed to update {engine} deployment for {deployment.workspace}/{deployment.name}: {e}") return DeploymentStatusUpdate( status="ERROR", status_message=f"Failed to update deployment {deployment.workspace}/{deployment.name} due to a service backend error", @@ -210,26 +243,18 @@ async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: ) async def get_status(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Drive the vLLM phased lifecycle and project status. + """Project status, driving the phased lifecycle for weighted deployments. - Reads the puller Job + (once created) the Deployment. When the Job has - completed and the Deployment doesn't exist yet, this advances creation - (phase P3) by emitting the Deployment + Service. + Weightless generic deployments have no puller phase: the Deployment is + created at ``create`` time, so a 404 means it was deleted externally (LOST). - The ``generic`` engine has no puller phase: there is always a Deployment - once create has run, so status is just its readiness. + Weighted deployments read the puller Job + (once created) the Deployment; + when the Job has completed and the Deployment doesn't exist yet, this + advances creation (phase P3) by emitting the Deployment + Service. """ - if config_engine(resolved.config) == ENGINE_GENERIC: - return self._get_status_generic(resolved) - deployment = resolved.deployment resource_name = resolved.resource_name - view = resolved.view - model_entity = resolved.model_entity - # The serving Deployment is the source of truth once it exists. We create - # it at P3 and delete the puller Job in the same step (to release the RWO - # volume), so a present Deployment means "past the pull phase" -- project - # its readiness and do NOT consult the (now-absent) Job. + # The serving Deployment is the source of truth once it exists. try: self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) deployment_exists = True @@ -241,7 +266,16 @@ async def get_status(self, resolved: ResolvedDeployment) -> DeploymentStatusUpda if deployment_exists: return self._project_deployment_readiness(resource_name) - # No Deployment yet: we're still in the pull phase. Consult the puller Job. + # No Deployment. For weightless deployments it should have been created at + # create-time, so its absence is external deletion -> LOST. + if not self._has_weights(resolved): + return DeploymentStatusUpdate( + status="LOST", + status_message="Serving Deployment not found; resources may have been deleted externally.", + host_url=None, + ) + + # Weighted: we're still in the pull phase. Consult the puller Job. job_name = vllm_k8s_compiler.pull_job_name(resource_name) try: job = self._batch_v1.read_namespaced_job(name=job_name, namespace=self._k8s_namespace) @@ -254,7 +288,7 @@ async def get_status(self, resolved: ResolvedDeployment) -> DeploymentStatusUpda # weights -> resume P3 by creating the serving objects); or # (b) genuine drift (PVC also gone -> LOST). if self._pvc_exists(resource_name): - return self._create_vllm_serving_objects(deployment, resource_name, view, model_entity) + return self._create_serving_objects(resolved) return DeploymentStatusUpdate( status="LOST", status_message="Weight-puller Job and PVC not found; resources may have been deleted externally.", @@ -277,7 +311,7 @@ async def get_status(self, resolved: ResolvedDeployment) -> DeploymentStatusUpda return DeploymentStatusUpdate(status="PENDING", status_message="Downloading model weights", host_url=None) # Job complete and no Deployment yet: phase P3 -- create the serving objects. - return self._create_vllm_serving_objects(deployment, resource_name, view, model_entity) + return self._create_serving_objects(resolved) async def delete(self, workspace: str, name: str) -> DeploymentStatusUpdate: """Delete the directly-emitted vLLM objects this reconciler owns. @@ -286,7 +320,7 @@ async def delete(self, workspace: str, name: str) -> DeploymentStatusUpdate: other reconciler's delete result. """ resource_name = get_deployment_resource_name(workspace, name) - errors = self._delete_vllm_resources(resource_name) + errors = self._delete_serving_resources(resource_name) if errors: summary = "; ".join(errors) return DeploymentStatusUpdate( @@ -339,32 +373,15 @@ def _model_source(resolved: ResolvedDeployment) -> tuple[str, str]: source_tag = f"{model_repo}@{revision}" if revision else model_repo return model_repo, source_tag - def _vllm_objects_exist(self, resource_name: str) -> bool: - """True if directly-emitted vLLM objects for this deployment exist. - - Checks the serving Deployment first (the puller Job is deleted once the - Deployment is created, so the Job alone is not a reliable marker), then the - puller Job for the pre-Deployment phase. Any lookup failure (including 404) - means "not (yet) a vLLM deployment". - """ - - def _has_vllm_engine_label(obj) -> bool: - labels = getattr(getattr(obj, "metadata", None), "labels", None) - return isinstance(labels, dict) and labels.get("nmp.nvidia.com/engine") in (ENGINE_VLLM, ENGINE_GENERIC) - + def _serving_deployment_exists(self, resource_name: str) -> bool: + """True if the serving Deployment for this resource already exists.""" try: - dep = self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) - if _has_vllm_engine_label(dep): - return True - except Exception: - pass - try: - job = self._batch_v1.read_namespaced_job( - name=vllm_k8s_compiler.pull_job_name(resource_name), namespace=self._k8s_namespace - ) - except Exception: - return False - return _has_vllm_engine_label(job) + self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) + return True + except k8s_client.exceptions.ApiException as e: + if e.status == 404: + return False + raise def _pvc_exists(self, resource_name: str) -> bool: """True if the model-weights PVC for this deployment exists.""" @@ -400,47 +417,54 @@ def _project_deployment_readiness(self, resource_name: str) -> DeploymentStatusU # Not ready yet: reuse the pod-drilldown (crash loop, image pull, events). return self._status.pod_status_from_deployment(resource_name) - def _create_vllm_serving_objects( - self, - deployment: ModelDeployment, - resource_name: str, - view: DeploymentConfigView, - model_entity: Optional[ModelEntity], - ) -> DeploymentStatusUpdate: - """Create the vLLM Deployment + Service after the puller Job has completed. - - Before creating the Deployment, the completed puller Job is deleted so its - pod releases the ReadWriteOnce PVC's volume attachment: a completed pod - keeps the volume attached to its node, which would otherwise block the - server pod from mounting the same RWO PVC if it schedules onto a different - node (Multi-Attach error). This runs only on the success path (the Job has - succeeded); a failed Job is left in place so the status path can read it + - its logs and report ERROR. - - Sets ownerReferences (PVC, Service -> Deployment) so deleting the - Deployment cascades the rest. + # ------------------------------------------------------------------ + # Engine-parameterized serving objects (shared by vLLM + generic) + # ------------------------------------------------------------------ + + def _pod_user(self, engine: str) -> tuple[Optional[int], Optional[int]]: + """Pod securityContext uid/gid for an engine. + + vLLM pins its image's user (2000/0); a generic container runs as its own + image's user (unset), since we can't assume an arbitrary image tolerates a + forced uid/gid. """ - # Release the RWO volume from the completed puller before the server needs - # it. Idempotent: if already deleted on a prior poll, _delete_puller_job - # treats NotFound as done. - if not self._delete_puller_job(resource_name): - return DeploymentStatusUpdate( - status="PENDING", - status_message="Releasing model weights volume", - host_url=self._status.host_url(resource_name), - ) + if engine == ENGINE_VLLM: + return self._backend_config.default_vllm_user_id, self._backend_config.default_vllm_group_id + return None, None - engine = ENGINE_VLLM + def _serving_spec( + self, + resolved: ResolvedDeployment, + *, + mount_model_store: bool, + ) -> tuple[k8s_client.V1Deployment, k8s_client.V1Service]: + """Compile the serving Deployment + Service for the deployment's engine. + + The engine selects the compiler (image/args/env), uid/gid, and LoRA wiring; + ``mount_model_store`` controls whether the model-weights PVC is mounted + (True for weighted deployments, False for a weightless generic container). + """ + deployment = resolved.deployment + view = resolved.view + engine = config_engine(resolved.config) + resource_name = resolved.resource_name health_path = resolve_health_path(engine, view) - image_name, image_tag = vllm_compiler.resolve_vllm_image( - view, self._backend_config.default_vllm_image, self._backend_config.default_vllm_image_tag - ) - args = vllm_compiler.compile_vllm_args(view, model_entity) - env = vllm_compiler.compile_vllm_env_vars(view) - startup_grace = self._backend_config.default_startup_probe_grace_period_seconds or 600 - - init_containers, sidecar_containers = self._build_lora_containers(deployment, view, model_entity) + user_id, group_id = self._pod_user(engine) + + if engine == ENGINE_GENERIC: + image_name, image_tag = generic_compiler.resolve_generic_image(view) + args = generic_compiler.compile_generic_args(view) + env = generic_compiler.compile_generic_env_vars(view) + init_containers: Optional[list] = None + sidecar_containers: Optional[list] = None + else: + image_name, image_tag = vllm_compiler.resolve_vllm_image( + view, self._backend_config.default_vllm_image, self._backend_config.default_vllm_image_tag + ) + args = vllm_compiler.compile_vllm_args(view, resolved.model_entity) + env = vllm_compiler.compile_vllm_env_vars(view) + init_containers, sidecar_containers = self._build_lora_containers(deployment, view, resolved.model_entity) dep_obj = vllm_k8s_compiler.compile_deployment( resource_name=resource_name, @@ -454,12 +478,13 @@ def _create_vllm_serving_objects( gpu=view.gpu, namespace=self._k8s_namespace, service_account_name=self._backend_config.service_account_name, - user_id=self._backend_config.default_vllm_user_id, - group_id=self._backend_config.default_vllm_group_id, + user_id=user_id, + group_id=group_id, shared_memory_size_limit=self._backend_config.default_shared_memory_size_limit, startup_grace_seconds=startup_grace, init_containers=init_containers, sidecar_containers=sidecar_containers, + mount_model_store=mount_model_store, ) svc_obj = vllm_k8s_compiler.compile_service( resource_name=resource_name, @@ -468,163 +493,45 @@ def _create_vllm_serving_objects( engine=engine, namespace=self._k8s_namespace, ) + return dep_obj, svc_obj - try: - created_dep = self._apps_v1.create_namespaced_deployment(namespace=self._k8s_namespace, body=dep_obj) - logger.info(f"Created vLLM Deployment {resource_name} in {self._k8s_namespace}") - except k8s_client.exceptions.ApiException as e: - if e.status != 409: - raise - created_dep = self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) - - # Owner reference -> Deployment, so PVC/Service cascade on delete. (The - # puller Job was already deleted above to release the RWO volume.) - owner_ref = k8s_client.V1OwnerReference( - api_version="apps/v1", - kind="Deployment", - name=created_dep.metadata.name, - uid=created_dep.metadata.uid, - controller=True, - block_owner_deletion=True, - ) - svc_obj.metadata.owner_references = [owner_ref] - self._create_or_skip(self._core_v1.create_namespaced_service, svc_obj, "Service") - self._set_owner_reference_on_pvc(resource_name, owner_ref) - - return DeploymentStatusUpdate(status="PENDING", status_message="Starting vLLM server", host_url=None) - - # ------------------------------------------------------------------ - # generic-engine helpers (no model weights / no PVC / no puller Job) - # ------------------------------------------------------------------ - - def _create_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Create a generic deployment: emit the Deployment + Service immediately. - - There are no model weights to pull, so there is no PVC and no puller Job; - the serving objects are created in one shot. - """ - deployment = resolved.deployment - logger.info( - f"Creating generic deployment: {deployment.workspace}/{deployment.name} " - f"(version: {deployment.entity_version})" - ) - try: - self._create_generic_serving_objects(deployment, resolved.resource_name, resolved.view) - return DeploymentStatusUpdate( - status="PENDING", - status_message="Starting container", - host_url=self._status.host_url(resolved.resource_name), - ) - except Exception as e: - logger.error(f"Failed to create generic deployment for {deployment.workspace}/{deployment.name}: {e}") - return DeploymentStatusUpdate( - status="ERROR", - status_message=f"Failed to create deployment {deployment.workspace}/{deployment.name} due to a service backend error", - error_details={"error": str(e), "error_type": type(e).__name__}, - host_url=None, - ) - - def _update_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Update a generic deployment by re-applying its serving objects. + def _create_serving_objects(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: + """Create the serving Deployment + Service for an engine. - Generic has no weight source, so there is no re-pull decision: if the - Deployment exists it is patched in place, otherwise it is (re)created. - """ - deployment = resolved.deployment - logger.info( - f"Updating generic deployment: {deployment.workspace}/{deployment.name} " - f"(version: {deployment.entity_version})" - ) - try: - self._create_generic_serving_objects(deployment, resolved.resource_name, resolved.view) - return DeploymentStatusUpdate( - status="PENDING", - status_message="Update accepted", - host_url=self._status.host_url(resolved.resource_name), - ) - except Exception as e: - logger.error(f"Failed to update generic deployment for {deployment.workspace}/{deployment.name}: {e}") - return DeploymentStatusUpdate( - status="ERROR", - status_message=f"Failed to update deployment {deployment.workspace}/{deployment.name} due to a service backend error", - error_details={"error": str(e), "error_type": type(e).__name__}, - host_url=None, - ) - - def _get_status_generic(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: - """Project a generic deployment's status from its serving Deployment. + For weighted deployments this is phase P3: the completed puller Job is + deleted first so its pod releases the ReadWriteOnce PVC's volume attachment + (a completed pod keeps the volume attached to its node, which would + otherwise block the server pod with a Multi-Attach error if it scheduled + onto a different node). For a weightless generic deployment there is no + Job/PVC, so this runs straight through. - The Deployment is created at create time (no staged rollout), so a 404 - means it was deleted externally -> LOST. + Sets ownerReferences (PVC + Service -> Deployment) so deleting the + Deployment cascades the rest. """ resource_name = resolved.resource_name - try: - self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) - except k8s_client.exceptions.ApiException as e: - if e.status != 404: - raise - return DeploymentStatusUpdate( - status="LOST", - status_message="Serving Deployment not found; resources may have been deleted externally.", - host_url=None, - ) - return self._project_deployment_readiness(resource_name) - - def _create_generic_serving_objects( - self, - deployment: ModelDeployment, - resource_name: str, - view: DeploymentConfigView, - ) -> None: - """Create (or no-op if present) the generic Deployment + Service. + has_weights = self._has_weights(resolved) - Mirrors the vLLM serving-object creation but without any PVC/model-store - mount: a generic container runs purely from its image. uid/gid are left - unset so the image's own user runs (an arbitrary container should not be - forced into vLLM's or NIM's user). - """ - engine = ENGINE_GENERIC - health_path = resolve_health_path(engine, view) - image_name, image_tag = generic_compiler.resolve_generic_image(view) - args = generic_compiler.compile_generic_args(view) - env = generic_compiler.compile_generic_env_vars(view) - - startup_grace = self._backend_config.default_startup_probe_grace_period_seconds or 600 + if has_weights: + # Release the RWO volume from the completed puller before the server + # needs it. Idempotent: a missing Job/pod counts as released. + if not self._delete_puller_job(resource_name): + return DeploymentStatusUpdate( + status="PENDING", + status_message="Releasing model weights volume", + host_url=self._status.host_url(resource_name), + ) - dep_obj = vllm_k8s_compiler.compile_deployment( - resource_name=resource_name, - workspace=deployment.workspace, - name=deployment.name, - engine=engine, - image=f"{image_name}:{image_tag}", - args=args, - health_path=health_path, - env=env, - gpu=view.gpu, - namespace=self._k8s_namespace, - service_account_name=self._backend_config.service_account_name, - user_id=None, - group_id=None, - shared_memory_size_limit=self._backend_config.default_shared_memory_size_limit, - startup_grace_seconds=startup_grace, - mount_model_store=False, - ) - svc_obj = vllm_k8s_compiler.compile_service( - resource_name=resource_name, - workspace=deployment.workspace, - name=deployment.name, - engine=engine, - namespace=self._k8s_namespace, - ) + dep_obj, svc_obj = self._serving_spec(resolved, mount_model_store=has_weights) try: created_dep = self._apps_v1.create_namespaced_deployment(namespace=self._k8s_namespace, body=dep_obj) - logger.info(f"Created generic Deployment {resource_name} in {self._k8s_namespace}") + logger.info(f"Created Deployment {resource_name} in {self._k8s_namespace}") except k8s_client.exceptions.ApiException as e: if e.status != 409: raise created_dep = self._apps_v1.read_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace) + # Owner reference -> Deployment, so PVC/Service cascade on delete. owner_ref = k8s_client.V1OwnerReference( api_version="apps/v1", kind="Deployment", @@ -635,6 +542,36 @@ def _create_generic_serving_objects( ) svc_obj.metadata.owner_references = [owner_ref] self._create_or_skip(self._core_v1.create_namespaced_service, svc_obj, "Service") + if has_weights: + self._set_owner_reference_on_pvc(resource_name, owner_ref) + + return DeploymentStatusUpdate(status="PENDING", status_message="Starting server", host_url=None) + + def _apply_serving_objects(self, resolved: ResolvedDeployment) -> None: + """Re-apply the serving Deployment + Service for an update. + + Patches the Deployment and Service when they already exist (so changed + image/args/env/health/gpu/labels actually take effect), and creates them + when they don't. Used by the update path; the PVC (if any) is unaffected. + """ + resource_name = resolved.resource_name + has_weights = self._has_weights(resolved) + dep_obj, svc_obj = self._serving_spec(resolved, mount_model_store=has_weights) + + if self._serving_deployment_exists(resource_name): + self._apps_v1.patch_namespaced_deployment(name=resource_name, namespace=self._k8s_namespace, body=dep_obj) + logger.info(f"Patched Deployment {resource_name} in {self._k8s_namespace}") + try: + self._core_v1.patch_namespaced_service(name=resource_name, namespace=self._k8s_namespace, body=svc_obj) + except k8s_client.exceptions.ApiException as e: + if e.status != 404: + raise + self._core_v1.create_namespaced_service(namespace=self._k8s_namespace, body=svc_obj) + return + + # Deployment absent: create both (idempotent on Service). + self._apps_v1.create_namespaced_deployment(namespace=self._k8s_namespace, body=dep_obj) + self._create_or_skip(self._core_v1.create_namespaced_service, svc_obj, "Service") def _delete_puller_job(self, resource_name: str) -> bool: """Delete the puller Job and confirm its pod is gone (releases RWO volume). @@ -763,11 +700,13 @@ def _existing_model_source(self, resource_name: str) -> str | None: annotations = (job.metadata.annotations or {}) if job.metadata else {} return annotations.get(vllm_k8s_compiler.MODEL_SOURCE_ANNOTATION) - def _delete_vllm_resources(self, resource_name: str) -> list[str]: - """Delete the directly-emitted vLLM objects by name (idempotent). + def _delete_serving_resources(self, resource_name: str) -> list[str]: + """Delete the directly-emitted objects by name (idempotent). - Returns a list of concise error strings for any real (non-404) failures; - empty when everything was deleted or already absent. + Covers the Deployment + Service plus the (optional) puller Job + PVC; a + weightless generic deployment simply has no Job/PVC, so those deletes are + 404-tolerant no-ops. Returns concise error strings for any real (non-404) + failures; empty when everything was deleted or already absent. """ deleters = [ (self._apps_v1.delete_namespaced_deployment, "Deployment", resource_name), diff --git a/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py b/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py index d544b8a589..c0c41ae0e3 100644 --- a/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py +++ b/services/core/models/tests/unit/controllers/backends/test_generic_compiler.py @@ -44,6 +44,20 @@ def test_resolve_generic_image_rejects_blank_image_name(): generic_compiler.resolve_generic_image(view) +def test_resolve_generic_image_trims_whitespace(): + """Defensive: surrounding whitespace is stripped from name and tag.""" + view = _view(gpu=0, image_name=" my/image ", image_tag=" v1 ") + name, tag = generic_compiler.resolve_generic_image(view) + assert name == "my/image" + assert tag == "v1" + + +def test_resolve_generic_image_blank_tag_falls_back_to_latest(): + view = _view(gpu=0, image_name="my/image", image_tag=" ") + _, tag = generic_compiler.resolve_generic_image(view) + assert tag == "latest" + + # --------------------------------------------------------------------------- # Args + env passthrough (the platform synthesizes nothing for generic) # --------------------------------------------------------------------------- diff --git a/services/core/models/tests/unit/controllers/test_docker_backend.py b/services/core/models/tests/unit/controllers/test_docker_backend.py index 1b3ee62800..0174e525b3 100644 --- a/services/core/models/tests/unit/controllers/test_docker_backend.py +++ b/services/core/models/tests/unit/controllers/test_docker_backend.py @@ -555,6 +555,8 @@ async def test_docker_backend_create_generic_deployment(docker_backend, sample_d # Engine + explicit health-path labels recorded for status-time probe selection. assert create_args["labels"]["nmp.nvidia.com/engine"] == "generic" assert create_args["labels"]["nmp.nvidia.com/health-path"] == "/v1/health/ready" + # Weightless generic runs raw: no platform volumes mounted (they'd shadow the image). + assert create_args["volumes"] == {} async def drive_creation_to_completion_after_create(docker_backend, sample_deployment, config, mock_docker_client): @@ -566,6 +568,40 @@ async def drive_creation_to_completion_after_create(docker_backend, sample_deplo return await drive_creation_to_completion(docker_backend, sample_deployment) +@pytest.mark.asyncio +async def test_docker_backend_create_generic_with_fileset_pulls_and_mounts( + docker_backend, sample_deployment, mock_docker_client +): + """Engine=generic with a fileset-backed model runs the puller and mounts platform volumes.""" + config = MagicMock() + set_deployment_config( + config, + engine="generic", + gpu=0, + image_name="my/custom-server", + image_tag="1.0", + health_check_path="/healthz", + model_namespace="default", + model_name="qwen-2-5-1-5b", + additional_args=["--model-dir", "/model-store"], + ) + + status_update = await _drive_vllm_with_puller(docker_backend, sample_deployment, mock_docker_client, config) + assert status_update.status == "PENDING" + + # Generic + fileset => the puller ran and the platform volumes are mounted so + # the pulled weights are available to the user's container at /model-store. + create_args = mock_docker_client.containers.create.call_args_list[0][1] + assert create_args["image"] == "my/custom-server:1.0" + assert create_args["command"] == ["--model-dir", "/model-store"] + assert create_args["volumes"][docker_backend._reconciler.get_volume_name("default", "test-deployment")] == { + "bind": "/model-store", + "mode": "rw", + } + assert create_args["labels"]["nmp.nvidia.com/engine"] == "generic" + assert create_args["labels"]["nmp.nvidia.com/health-path"] == "/healthz" + + def test_get_health_path_from_container_vllm(docker_backend): """vLLM containers probe /health.""" container = MagicMock() diff --git a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py index f7a161433e..b0d2d2ac76 100644 --- a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py +++ b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py @@ -37,12 +37,14 @@ def _nim_config(): """A minimal NIM-routing ModelDeploymentConfig-like object. - ``config_engine`` returns the NIM engine for anything that isn't explicitly - ``vllm``/``generic``, so a bare mock routes status/create/update to the NIM - reconciler. The NIM status path only reads ``resource_name``, so the resolved - model fields are irrelevant here. + Engine is explicitly ``nim`` so the backend routes status/create/update to the + NIM reconciler (the backend rejects unknown engine strings rather than + defaulting them to NIM). The NIM status path only reads ``resource_name``, so + the resolved model fields are irrelevant here. """ - return MagicMock() + config = MagicMock() + config.engine = "nim" + return config def _make_nimservice_mock(state: str, conditions: list | None = None): @@ -272,8 +274,15 @@ def sample_deployment(): @pytest.fixture def sample_config(): - """Create a sample ModelDeploymentConfig for testing.""" - return MagicMock() + """Create a sample ModelDeploymentConfig for testing. + + Engine is explicitly ``nim`` so the backend routes it to the NIM-operator + reconciler (the backend now rejects unknown engine strings rather than + defaulting them to NIM). + """ + config = MagicMock() + config.engine = "nim" + return config @pytest.mark.asyncio @@ -2090,6 +2099,102 @@ async def test_generic_status_lost_when_deployment_missing(k8s_backend, sample_d assert result.status == "LOST" +def _generic_weighted_config(*, gpu: int = 1): + """A generic config that also references a model (fileset-backed weights).""" + config = _generic_config(gpu=gpu) + config.model_spec.model_namespace = "default" + config.model_spec.model_name = "qwen" + return config + + +def _fileset_model_entity(): + """A model entity with a fileset -> resolves to FILES_SERVICE weights.""" + return SimpleNamespace( + workspace="default", name="qwen", spec=None, trust_remote_code=False, fileset="hf://default/qwen" + ) + + +@pytest.mark.asyncio +async def test_generic_with_fileset_runs_staged_puller(k8s_backend, sample_deployment): + """A generic deployment whose config resolves to a fileset pulls weights (staged).""" + backend = _vllm_backend(k8s_backend) + config = _generic_weighted_config(gpu=1) + + with patch.object(backend, "_resolve_model_source", return_value=("default", "qwen", None)): + _sync_reconcilers(backend) + result = await backend.create_model_deployment( + ModelContext( + model_deployment=sample_deployment, + model_deployment_config=config, + model_entity=_fileset_model_entity(), + ) + ) + + assert result.status == "PENDING" + # Weighted generic => staged rollout: PVC + puller Job, no Deployment yet. + backend._core_v1.create_namespaced_persistent_volume_claim.assert_called_once() + backend._batch_v1.create_namespaced_job.assert_called_once() + backend._apps_v1.create_namespaced_deployment.assert_not_called() + + +@pytest.mark.asyncio +async def test_generic_with_fileset_p3_mounts_model_store(k8s_backend, sample_deployment): + """At P3 the weighted generic serving Deployment mounts the model-store PVC + uses raw args.""" + backend = _vllm_backend(k8s_backend) + config = _generic_weighted_config(gpu=1) + + job = MagicMock() + job.status.failed = None + job.status.succeeded = 1 + backend._batch_v1.read_namespaced_job.return_value = job + backend._apps_v1.read_namespaced_deployment.side_effect = _api_exception(404) + backend._core_v1.list_namespaced_pod.return_value = MagicMock(items=[]) + created_dep = MagicMock() + created_dep.metadata.name = backend._get_resource_name(sample_deployment) + created_dep.metadata.uid = "dep-uid" + backend._apps_v1.create_namespaced_deployment.return_value = created_dep + + with patch.object(backend, "_resolve_model_source", return_value=("default", "qwen", None)): + _sync_reconcilers(backend) + result = await backend.get_model_deployment_status( + ModelContext( + model_deployment=sample_deployment, + model_deployment_config=config, + model_entity=_fileset_model_entity(), + ) + ) + + assert result.status == "PENDING" + backend._apps_v1.create_namespaced_deployment.assert_called_once() + dep_obj = backend._apps_v1.create_namespaced_deployment.call_args.kwargs["body"] + container = dep_obj.spec.template.spec.containers[0] + # Weighted: model-store PVC is mounted so the pulled weights are available. + volume_names = {v.name for v in dep_obj.spec.template.spec.volumes} + assert "model-store" in volume_names + # Still a generic container: runs the user's raw args (no vLLM serve synthesis). + assert container.args == ["--port", "8000"] + + +@pytest.mark.asyncio +async def test_generic_update_patches_deployment(k8s_backend, sample_deployment): + """Updating a (weightless) generic deployment patches the serving objects in place.""" + backend = _vllm_backend(k8s_backend) + config = _generic_config() + # Serving Deployment already exists -> patch, don't recreate. + backend._apps_v1.read_namespaced_deployment.return_value = MagicMock() + + _sync_reconcilers(backend) + result = await backend.update_model_deployment( + ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) + ) + + assert result.status == "PENDING" + backend._apps_v1.patch_namespaced_deployment.assert_called_once() + backend._core_v1.patch_namespaced_service.assert_called_once() + # Patched in place, not recreated. + backend._apps_v1.create_namespaced_deployment.assert_not_called() + + @pytest.mark.asyncio async def test_vllm_status_job_running_is_pending(k8s_backend, sample_deployment): """While the puller Job is running, status is PENDING.""" @@ -2298,7 +2403,7 @@ async def test_vllm_status_deployment_ready_is_ready(k8s_backend, sample_deploym @pytest.mark.asyncio async def test_vllm_update_unchanged_source_does_not_repull(k8s_backend, sample_deployment): - """Unchanged model source: no Job re-create, no resource deletion.""" + """Unchanged model source: no resource deletion; serving objects patched in place.""" backend = _vllm_backend(k8s_backend) config = _vllm_config() @@ -2306,9 +2411,11 @@ async def test_vllm_update_unchanged_source_does_not_repull(k8s_backend, sample_ existing_job.metadata.labels = {"nmp.nvidia.com/engine": "vllm"} existing_job.metadata.annotations = {"nmp.nvidia.com/model-source": "default/qwen"} backend._batch_v1.read_namespaced_job.return_value = existing_job + # Serving Deployment already exists -> update patches it in place. + backend._apps_v1.read_namespaced_deployment.return_value = MagicMock() with patch.object(backend, "_resolve_model_source", return_value=("default", "qwen", None)): - with patch.object(backend._k8s_reconciler, "_delete_vllm_resources") as mock_delete: + with patch.object(backend._k8s_reconciler, "_delete_serving_resources") as mock_delete: _sync_reconcilers(backend) result = await backend.update_model_deployment( ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) @@ -2316,6 +2423,8 @@ async def test_vllm_update_unchanged_source_does_not_repull(k8s_backend, sample_ mock_delete.assert_not_called() backend._batch_v1.create_namespaced_job.assert_not_called() + # Patched, not recreated. + backend._apps_v1.patch_namespaced_deployment.assert_called_once() assert result.status == "PENDING" @@ -2331,7 +2440,7 @@ async def test_vllm_update_changed_source_repulls(k8s_backend, sample_deployment backend._batch_v1.read_namespaced_job.return_value = existing_job with patch.object(backend, "_resolve_model_source", return_value=("default", "qwen", "v2")): - with patch.object(backend._k8s_reconciler, "_delete_vllm_resources") as mock_delete: + with patch.object(backend._k8s_reconciler, "_delete_serving_resources") as mock_delete: _sync_reconcilers(backend) result = await backend.update_model_deployment( ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) diff --git a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py index 433bd8eb79..d32425e052 100644 --- a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py +++ b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py @@ -183,6 +183,70 @@ async def test_create_generic_config_requires_image_and_health_path(deployment_c mock_entity_client.create.assert_not_called() +@pytest.mark.asyncio +async def test_create_generic_config_requires_health_check_path(deployment_config_service, mock_entity_client): + """A generic config with image_name set but health_check_path missing is rejected.""" + mock_list_result = MagicMock() + mock_list_result.data = [] + mock_entity_client.list.return_value = mock_list_result + + request = CreateModelDeploymentConfigRequest( + name="generic-config", + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(), + executor_config=ContainerExecutorConfig(gpu=0, image_name="my/image"), # no health_check_path + ) + + with pytest.raises(ValueError, match="health_check_path"): + await deployment_config_service.create_deployment_config(request, "default") + mock_entity_client.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_generic_config_rejects_whitespace_padded_fields(deployment_config_service, mock_entity_client): + """Whitespace-padded generic image/health-path values are rejected, not silently stored.""" + mock_list_result = MagicMock() + mock_list_result.data = [] + mock_entity_client.list.return_value = mock_list_result + + request = CreateModelDeploymentConfigRequest( + name="generic-config", + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(), + executor_config=ContainerExecutorConfig( + gpu=0, + image_name=" my/image ", + health_check_path=" /health ", + ), + ) + + with pytest.raises(ValueError, match="whitespace"): + await deployment_config_service.create_deployment_config(request, "default") + mock_entity_client.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_update_generic_config_requires_image_and_health_path( + deployment_config_service, mock_entity_client, sample_config_entity +): + """The update path enforces the same generic validation as create.""" + # Existing config so update proceeds to validation. + mock_list_result = MagicMock() + mock_list_result.data = [sample_config_entity] + mock_entity_client.list.return_value = mock_list_result + mock_entity_client.get.return_value = sample_config_entity + + request = UpdateModelDeploymentConfigRequest( + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(), + executor_config=ContainerExecutorConfig(gpu=0), # no image_name / health_check_path + ) + + with pytest.raises(ValueError, match="image_name"): + await deployment_config_service.update_deployment_config("default", "test-config", request) + mock_entity_client.create.assert_not_called() + + @pytest.mark.asyncio async def test_create_generic_config_succeeds_when_image_and_health_path_set( deployment_config_service, mock_entity_client, sample_config_entity From 07aafde762b1ccdd2998900ac9fece589f8f5457 Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Fri, 26 Jun 2026 12:50:50 -0600 Subject: [PATCH 3/3] fix(models): address generic engine PR review follow-ups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Reject engine=generic with model_spec.lora_enabled in the deployment config validation (no engine compiler to wire the adapter sidecar, so it would otherwise be silently ignored). - k8s: restore the mid-pull no-op on update — when the serving Deployment is absent but the puller Job is still present, accept the update without re-running create(), only falling back to create() on genuine drift. - Reword the create() docstring ("start at phase P0"). - Add unit tests for the generic+LoRA rejection and the mid-pull no-op update. Signed-off-by: Ben McCown --- .../model_deployment_config_service.py | 22 +++++++++--- .../k8s_nim_operator/reconcilers/k8s.py | 34 +++++++++++++++++-- .../test_k8s_nim_operator_backend.py | 30 ++++++++++++++++ ...st_model_deployment_config_service_unit.py | 24 +++++++++++++ 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py index 87e78d694e..95d599fb25 100644 --- a/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py +++ b/services/core/models/src/nmp/core/models/api/service/model_deployment_config_service.py @@ -19,6 +19,7 @@ CreateModelDeploymentConfigRequest, Engine, ModelDeploymentConfig, + ModelDeploymentConfigModelSpec, ModelDeploymentStatus, UpdateModelDeploymentConfigRequest, ) @@ -26,8 +27,12 @@ logger = logging.getLogger(__name__) -def _validate_engine_config(engine: Engine, executor_config: ContainerExecutorConfig) -> None: - """Validate engine-specific requirements on the executor config. +def _validate_engine_config( + engine: Engine, + executor_config: ContainerExecutorConfig, + model_spec: ModelDeploymentConfigModelSpec, +) -> None: + """Validate engine-specific requirements on the deployment config. The ``generic`` engine runs an arbitrary container with no inference-engine compiler, so it has no platform-default image and no canonical health @@ -38,6 +43,10 @@ def _validate_engine_config(engine: Engine, executor_config: ContainerExecutorCo Values are also rejected when they contain surrounding whitespace: an image reference or probe path is used verbatim downstream, where a leading/trailing space would silently produce an invalid value. + + LoRA is rejected for ``generic``: there is no engine compiler to wire the + adapter sidecar against, so ``lora_enabled`` would otherwise be silently + ignored. Reject it up front rather than accept a config that can't be honored. """ if engine != Engine.GENERIC: return @@ -59,6 +68,11 @@ def _validate_engine_config(engine: Engine, executor_config: ContainerExecutorCo raise ValueError( "executor_config." + " and executor_config.".join(padded) + " must not have leading or trailing whitespace." ) + if model_spec.lora_enabled: + raise ValueError( + "The 'generic' engine does not support LoRA (model_spec.lora_enabled); " + "there is no engine compiler to wire the adapter sidecar against." + ) class ReferentialIntegrityError(Exception): @@ -141,7 +155,7 @@ async def create_deployment_config( if existing is not None: raise ValueError(f"Deployment config with workspace '{workspace}' and name '{request.name}' already exists") - _validate_engine_config(request.engine, request.executor_config) + _validate_engine_config(request.engine, request.executor_config, request.model_spec) if not request.model_entity_id: try: @@ -294,7 +308,7 @@ async def update_deployment_config( if not current: raise ValueError(f"Deployment config with workspace '{workspace}' and name '{name}' does not exist") - _validate_engine_config(request.engine, request.executor_config) + _validate_engine_config(request.engine, request.executor_config, request.model_spec) new_version = current.entity_version + 1 diff --git a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py index 787c91ebeb..e87e3214b2 100644 --- a/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py +++ b/services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py @@ -114,8 +114,8 @@ def _has_weights(resolved: ResolvedDeployment) -> bool: async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: """Create the deployment's backend resources. - Weighted deployments do phase P0: emit the PVC + weight-puller Job (the - Deployment + Service are created later by the status path once the Job + Weighted deployments start at phase P0: emit the PVC + weight-puller Job + (the Deployment + Service are created later by the status path once the Job completes -- controller-side weight-readiness gating). Weightless generic deployments have nothing to pull, so the serving Deployment + Service are emitted immediately. @@ -224,7 +224,7 @@ async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: return await self.create(resolved) # Unchanged source: patch the serving Deployment + Service in place if - # present, else (still in the pull phase) the status path creates them. + # present. if self._serving_deployment_exists(resource_name): self._apply_serving_objects(resolved) return DeploymentStatusUpdate( @@ -232,6 +232,17 @@ async def update(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate: status_message="Update accepted", host_url=self._status.host_url(resource_name), ) + # No serving Deployment yet. If the puller Job is still present we're + # mid-pull; the status path will emit the serving objects at P3, so the + # update is a no-op (re-running create() here would re-assert the PVC + + # Job needlessly). Only fall back to create() if the pull objects are + # gone (genuine drift). + if self._puller_job_exists(resource_name): + return DeploymentStatusUpdate( + status="PENDING", + status_message="Update accepted", + host_url=self._status.host_url(resource_name), + ) return await self.create(resolved) except Exception as e: logger.error(f"Failed to update {engine} deployment for {deployment.workspace}/{deployment.name}: {e}") @@ -395,6 +406,23 @@ def _pvc_exists(self, resource_name: str) -> bool: return False raise + def _puller_job_exists(self, resource_name: str) -> bool: + """True if the weight-puller Job for this deployment still exists. + + Used by the update path to detect the mid-pull window (PVC + Job created, + serving Deployment not yet emitted) so an unchanged-source update is a + no-op rather than re-running create(). + """ + try: + self._batch_v1.read_namespaced_job( + name=vllm_k8s_compiler.pull_job_name(resource_name), namespace=self._k8s_namespace + ) + return True + except k8s_client.exceptions.ApiException as e: + if e.status == 404: + return False + raise + def _create_or_skip(self, create_fn, body, kind: str) -> None: """Create a namespaced object, tolerating 409 Conflict (already exists).""" try: diff --git a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py index b0d2d2ac76..0b483c7d56 100644 --- a/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py +++ b/services/core/models/tests/unit/controllers/test_k8s_nim_operator_backend.py @@ -2453,6 +2453,36 @@ async def test_vllm_update_changed_source_repulls(k8s_backend, sample_deployment assert result.status == "PENDING" +@pytest.mark.asyncio +async def test_vllm_update_during_pull_is_noop(k8s_backend, sample_deployment): + """Unchanged source, serving Deployment not yet created but puller Job present. + + Mid-pull updates must be a no-op (the status path emits the serving objects at + P3); we must not re-run create() or patch a non-existent Deployment. + """ + backend = _vllm_backend(k8s_backend) + config = _vllm_config() + + existing_job = MagicMock() + existing_job.metadata.labels = {"nmp.nvidia.com/engine": "vllm"} + existing_job.metadata.annotations = {"nmp.nvidia.com/model-source": "default/qwen"} + backend._batch_v1.read_namespaced_job.return_value = existing_job + # Serving Deployment does not exist yet (still pulling). + backend._apps_v1.read_namespaced_deployment.side_effect = _api_exception(404) + + with patch.object(backend, "_resolve_model_source", return_value=("default", "qwen", None)): + _sync_reconcilers(backend) + result = await backend.update_model_deployment( + ModelContext(model_deployment=sample_deployment, model_deployment_config=config, model_entity=None) + ) + + assert result.status == "PENDING" + # No-op: no re-create of PVC/Job, no Deployment patch. + backend._core_v1.create_namespaced_persistent_volume_claim.assert_not_called() + backend._batch_v1.create_namespaced_job.assert_not_called() + backend._apps_v1.patch_namespaced_deployment.assert_not_called() + + @pytest.mark.asyncio async def test_vllm_list_managed_unions_deployments(k8s_backend): """list_managed_deployment_names unions NIMServices and raw vLLM Deployments.""" diff --git a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py index d32425e052..0e958a9df1 100644 --- a/services/core/models/tests/unit/test_model_deployment_config_service_unit.py +++ b/services/core/models/tests/unit/test_model_deployment_config_service_unit.py @@ -275,6 +275,30 @@ async def test_create_generic_config_succeeds_when_image_and_health_path_set( mock_entity_client.create.assert_called_once() +@pytest.mark.asyncio +async def test_create_generic_config_rejects_lora_enabled(deployment_config_service, mock_entity_client): + """LoRA is unsupported for generic (no compiler to wire the sidecar) -> rejected.""" + mock_list_result = MagicMock() + mock_list_result.data = [] + mock_entity_client.list.return_value = mock_list_result + + request = CreateModelDeploymentConfigRequest( + name="generic-config", + engine="generic", + model_spec=ModelDeploymentConfigModelSpec(lora_enabled=True), + executor_config=ContainerExecutorConfig( + gpu=0, + image_name="my/image", + image_tag="1.0", + health_check_path="/healthz", + ), + ) + + with pytest.raises(ValueError, match="LoRA"): + await deployment_config_service.create_deployment_config(request, "default") + mock_entity_client.create.assert_not_called() + + @pytest.mark.asyncio async def test_get_deployment_config_found(deployment_config_service, mock_entity_client, sample_config_entity): """Test retrieving an existing deployment config."""