refactor(models): address PR review feedback

Composition over inheritance for the k8s reconcilers (review #2/#3):
* Extract StatusProjector (pod-status projection, crash-loop/pending-timeout
  error builders, host URL) and ResourceDeleter (idempotent 404-tolerant
  delete) as standalone collaborators.
* Reconciler is now a pure interface (the 5 verbs); NimOperatorReconciler and
  K8sReconciler compose the projector + deleter instead of inheriting them.
  The backend builds both collaborators in init() and injects them.

Thread the reconcile context through the backend interface (review #19):
* create/update/get_model_deployment_status now take a single
  ctx: ModelContext instead of (deployment, config, model_entity); applied
  across the ServiceBackend ABC and the docker / none / k8s backends, the
  deployment reconciler call sites, and the test mocks. delete stays
  (workspace, name).

Fixes + nits:
* Harden NIMService status read against a null status/state (review #15):
  (nim_status.get("state") or "").lower() can no longer raise.
* Convert nim_operator logging to structured extra={} (review #13); avoid the
  reserved LogRecord 'name' key (use resource_name / deployment_name).
* Flatten the Files-service create/update branches into a guard-clause helper
  (review #14).
* compile_puller_job: rename args -> container_args (review #17).
* Reconciler nits: import the vllm_k8s_compiler module under its full name
  (review #7), reflow the P3 (a)/(b) comment (review #8), quote values in the
  model-source error (review #9), drop the _ = image_pull_secrets dance
  (review #12), name the event-message cap MAX_EVENT_MESSAGE_CHARS (review #6),
  and document the _select_reconciler None contract (review #16).

Signed-off-by: Ben McCown <[email protected]>

Author: benmccown

services/core/inference-gateway/tests/integration/conftest.py

@@ -87,31 +87,19 @@ def init(self) -> None:
         """No-op init for mock backend."""
         pass
 
-    async def create_model_deployment(
-        self,
-        deployment: Any,
-        config: Any,
-        model_entity: Any = None,
-    ) -> DeploymentStatusUpdate:
+    async def create_model_deployment(self, ctx: Any) -> DeploymentStatusUpdate:
         """Record call and return configured response."""
-        self.create_calls.append((deployment, config, model_entity))
+        self.create_calls.append((ctx.model_deployment, ctx.model_deployment_config, ctx.model_entity))
         return self.create_response
 
-    async def update_model_deployment(
-        self,
-        deployment: Any,
-        config: Any,
-        model_entity: Any = None,
-    ) -> DeploymentStatusUpdate:
+    async def update_model_deployment(self, ctx: Any) -> DeploymentStatusUpdate:
         """Record call and return configured response."""
-        self.update_calls.append((deployment, config, model_entity))
+        self.update_calls.append((ctx.model_deployment, ctx.model_deployment_config, ctx.model_entity))
         return self.create_response
 
-    async def get_model_deployment_status(
-        self, deployment: Any, config: Any = None, model_entity: Any = None
-    ) -> DeploymentStatusUpdate:
+    async def get_model_deployment_status(self, ctx: Any) -> DeploymentStatusUpdate:
         """Record call and return configured response."""
-        self.status_calls.append(deployment)
+        self.status_calls.append(ctx.model_deployment)
         return self.status_response
 
     async def delete_model_deployment(self, deployment: Any) -> DeploymentStatusUpdate:

services/core/models/src/nmp/core/models/controllers/backends/backends.py

@@ -4,13 +4,11 @@
 """Base backend interface for Models Controller service."""
 
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 
 from nemo_platform import AsyncNeMoPlatform
 from nemo_platform.types.inference import ModelDeploymentStatus
-from nemo_platform.types.inference.model_deployment import ModelDeployment
-from nemo_platform.types.inference.model_deployment_config import ModelDeploymentConfig
-from nemo_platform.types.models.model_entity import ModelEntity
+from nmp.core.models.controllers.context import ModelContext
 from pydantic import BaseModel
 
 
@@ -67,15 +65,12 @@ def shutdown(self) -> None:
         ...
 
     @abstractmethod
-    async def create_model_deployment(
-        self, deployment: ModelDeployment, config: ModelDeploymentConfig, model_entity: Optional[ModelEntity] = None
-    ) -> DeploymentStatusUpdate:
+    async def create_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Create a new model deployment.
 
         Args:
-            deployment: The ModelDeployment object to create
-            config: The ModelDeploymentConfig for this deployment
-            model_entity: Optional Model entity from Entity Store (contains peft, artifact, etc.)
+            ctx: The reconciliation context bundling the ModelDeployment, its
+                ModelDeploymentConfig, and the optional Model entity.
 
         Returns:
             DeploymentStatusUpdate with the current status after creation attempt
@@ -86,15 +81,13 @@ async def create_model_deployment(
         ...
 
     @abstractmethod
-    async def update_model_deployment(
-        self, deployment: ModelDeployment, config: ModelDeploymentConfig, model_entity: Optional[ModelEntity] = None
-    ) -> DeploymentStatusUpdate:
+    async def update_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Update an existing model deployment.
 
         Args:
-            deployment: The ModelDeployment object with updated configuration
-            config: The ModelDeploymentConfig for this deployment (may be a new version)
-            model_entity: Optional Model entity from Entity Store (contains peft, artifact, etc.)
+            ctx: The reconciliation context bundling the ModelDeployment, its
+                (possibly new-version) ModelDeploymentConfig, and the optional
+                Model entity.
 
         Returns:
             DeploymentStatusUpdate with the current status after update attempt
@@ -105,20 +98,14 @@ async def update_model_deployment(
         ...
 
     @abstractmethod
-    async def get_model_deployment_status(
-        self,
-        deployment: ModelDeployment,
-        config: Optional[ModelDeploymentConfig] = None,
-        model_entity: Optional[ModelEntity] = None,
-    ) -> DeploymentStatusUpdate:
+    async def get_model_deployment_status(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Get the current status of a model deployment.
 
         Args:
-            deployment: The ModelDeployment object to check
-            config: The ModelDeploymentConfig for this deployment. Some backends
-                need it to advance creation (e.g. the k8s vLLM path emits the
-                serving Deployment once the weight-puller Job completes).
-            model_entity: Optional Model entity from Entity Store.
+            ctx: The reconciliation context bundling the ModelDeployment, its
+                ModelDeploymentConfig, and the optional Model entity. Some backends
+                need the config to advance creation (e.g. the k8s vLLM path emits
+                the serving Deployment once the weight-puller Job completes).
 
         Returns:
             DeploymentStatusUpdate with the current deployment status

services/core/models/src/nmp/core/models/controllers/backends/docker/backend.py

@@ -12,14 +12,12 @@
 import asyncio
 import os
 from logging import getLogger
-from typing import Any, Optional
+from typing import Any
 
 import httpx
 from docker.errors import APIError, NotFound
 from nemo_platform import NotFoundError
 from nemo_platform.types.inference.model_deployment import ModelDeployment
-from nemo_platform.types.inference.model_deployment_config import ModelDeploymentConfig
-from nemo_platform.types.models.model_entity import ModelEntity
 from nmp.common.config import get_platform_config
 from nmp.common.docker.gpu_pool import DockerGPUPool
 from nmp.common.resources import SharedResourceManager
@@ -37,6 +35,7 @@
     NGC_IMAGE_REGISTRY_USER_NAME,
     DockerDeploymentCreationReconciler,
 )
+from nmp.core.models.controllers.context import ModelContext
 from requests.exceptions import ConnectionError as RequestsConnectionError
 from requests.exceptions import ReadTimeout
 from urllib3.exceptions import ReadTimeoutError as Urllib3ReadTimeoutError
@@ -184,17 +183,15 @@ async def _ensure_ngc_login(self, ngc_api_key: str | None) -> None:
     # ServiceBackend CRUD interface
     # ==================================================================
 
-    async def create_model_deployment(
-        self,
-        deployment: ModelDeployment,
-        config: ModelDeploymentConfig,
-        model_entity: Optional[ModelEntity] = None,
-    ) -> DeploymentStatusUpdate:
+    async def create_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Create a new model deployment as a Docker container.
 
         Resolves NGC credentials and delegates the multi-stage creation
         pipeline to :class:`DockerDeploymentCreationReconciler`.
         """
+        deployment = ctx.model_deployment
+        config = ctx.model_deployment_config
+        model_entity = ctx.model_entity
         resolved_ngc_key = await self._resolve_ngc_api_key()
         await self._ensure_ngc_login(resolved_ngc_key)
 
@@ -205,30 +202,22 @@ async def create_model_deployment(
             resolved_ngc_key,
         )
 
-    async def update_model_deployment(
-        self,
-        deployment: ModelDeployment,
-        config: ModelDeploymentConfig,
-        model_entity: Optional[ModelEntity] = None,
-    ) -> DeploymentStatusUpdate:
+    async def update_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Update a model deployment by recreating the container."""
+        deployment = ctx.model_deployment
         logger.info(f"Updating Docker deployment: {deployment.workspace}/{deployment.name}")
         delete_result = await self.delete_model_deployment(deployment.workspace, deployment.name)
         if delete_result.status == "ERROR":
             return delete_result
-        return await self.create_model_deployment(deployment, config, model_entity)
+        return await self.create_model_deployment(ctx)
 
-    async def get_model_deployment_status(
-        self,
-        deployment: ModelDeployment,
-        config: Optional[ModelDeploymentConfig] = None,
-        model_entity: Optional[ModelEntity] = None,
-    ) -> DeploymentStatusUpdate:
+    async def get_model_deployment_status(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Get the status of a Docker model deployment.
 
         While the deployment is still progressing through the creation
         pipeline this delegates to the reconciler's ``advance`` method.
         """
+        deployment = ctx.model_deployment
         if self._reconciler.is_deploying(deployment.workspace, deployment.name):
             deployment_key = self._reconciler.get_deployment_key(deployment.workspace, deployment.name)
             return await self._reconciler.advance(deployment_key)

services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/backend.py

@@ -23,7 +23,6 @@
 from kubernetes import config as k8s_config
 from kubernetes.dynamic import DynamicClient
 from nemo_platform.types.inference.model_deployment import ModelDeployment
-from nemo_platform.types.inference.model_deployment_config import ModelDeploymentConfig
 from nemo_platform.types.models.model_entity import ModelEntity
 from nmp.common.config import get_platform_config
 from nmp.core.models.app import (
@@ -40,11 +39,14 @@
 )
 from nmp.core.models.controllers.backends.engine import ENGINE_GENERIC, 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 ResolvedDeployment
+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
 from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.nim_operator import (
     NimOperatorReconciler,
 )
+from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.resource_deleter import ResourceDeleter
+from nmp.core.models.controllers.backends.k8s_nim_operator.reconcilers.status_projector import StatusProjector
+from nmp.core.models.controllers.context import ModelContext
 
 logger = getLogger(__name__)
 
@@ -62,6 +64,8 @@ def __init__(self, nmp_sdk, config, huggingface_model_puller: str):
         self._k8s_namespace: str | None = None
         self._backend_config: K8sNimOperatorConfig | None = None
         self._huggingface_model_puller = huggingface_model_puller
+        self._status_projector: StatusProjector | None = None
+        self._resource_deleter: ResourceDeleter | None = None
         self._nim_reconciler: NimOperatorReconciler | None = None
         self._k8s_reconciler: K8sReconciler | None = None
         super().__init__(nmp_sdk, config)
@@ -88,18 +92,30 @@ def init(self) -> None:
         self._k8s_namespace = self._get_current_namespace()
         logger.info(f"Models controller will deploy models to namespace: {self._k8s_namespace}")
 
-        self._nim_reconciler = NimOperatorReconciler(
+        # Shared collaborators composed into both reconcilers (and used directly
+        # by the PENDING-timeout policy below).
+        self._status_projector = StatusProjector(
             k8s_client_=self._k8s_client,
+            backend_config=self._backend_config,
+            k8s_namespace=self._k8s_namespace,
+        )
+        self._resource_deleter = ResourceDeleter(k8s_namespace=self._k8s_namespace)
+
+        self._nim_reconciler = NimOperatorReconciler(
             dynamic_client=self._dynamic_client,
             backend_config=self._backend_config,
             k8s_namespace=self._k8s_namespace,
             huggingface_model_puller=self._huggingface_model_puller,
+            status=self._status_projector,
+            deleter=self._resource_deleter,
         )
         self._k8s_reconciler = K8sReconciler(
             k8s_client_=self._k8s_client,
             backend_config=self._backend_config,
             k8s_namespace=self._k8s_namespace,
             huggingface_model_puller=self._huggingface_model_puller,
+            status=self._status_projector,
+            deleter=self._resource_deleter,
         )
 
     def shutdown(self) -> None:
@@ -189,13 +205,11 @@ def _remote_files_hf_url(self) -> str:
         files_url = platform_config.service_discovery.get("files") or platform_config.base_url
         return urljoin(files_url.rstrip("/") + "/", "apis/files/v2/hf")
 
-    def _resolve(
-        self,
-        deployment: ModelDeployment,
-        config: ModelDeploymentConfig,
-        model_entity: Optional[ModelEntity],
-    ) -> ResolvedDeployment:
+    def _resolve(self, ctx: ModelContext) -> ResolvedDeployment:
         """Resolve everything a reconciler needs from the API object + SDK state."""
+        deployment = ctx.model_deployment
+        config = ctx.model_deployment_config
+        model_entity = ctx.model_entity
         view = deployment_config_view(config)
         model_namespace, model_name, model_revision = self._resolve_model_source(model_entity, view)
         weights_type = get_model_weights_type(
@@ -218,8 +232,14 @@ def _resolve(
             huggingface_model_puller=self._huggingface_model_puller,
         )
 
-    def _select_reconciler(self, engine: str):
-        """Select the reconciler for an engine, rejecting the unsupported one."""
+    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`).
+        """
         if engine == ENGINE_VLLM:
             return self._k8s_reconciler
         if engine == ENGINE_GENERIC:
@@ -239,48 +259,42 @@ def _unsupported_engine(engine: str) -> DeploymentStatusUpdate:
     # ServiceBackend interface (resolve + select + delegate)
     # ------------------------------------------------------------------
 
-    async def create_model_deployment(
-        self, deployment: ModelDeployment, config: ModelDeploymentConfig, model_entity: Optional[ModelEntity] = None
-    ) -> DeploymentStatusUpdate:
-        """Create a new model deployment (dispatches on ``config.engine``)."""
-        engine = config_engine(config)
+    async def create_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
+        """Create a new model deployment (dispatches on the config's engine)."""
+        engine = config_engine(ctx.model_deployment_config)
         reconciler = self._select_reconciler(engine)
         if reconciler is None:
             return self._unsupported_engine(engine)
-        resolved = self._resolve(deployment, config, model_entity)
+        resolved = self._resolve(ctx)
         return await reconciler.create(resolved)
 
-    async def update_model_deployment(
-        self, deployment: ModelDeployment, config: ModelDeploymentConfig, model_entity: Optional[ModelEntity] = None
-    ) -> DeploymentStatusUpdate:
-        """Update an existing model deployment (dispatches on ``config.engine``)."""
-        engine = config_engine(config)
+    async def update_model_deployment(self, ctx: ModelContext) -> DeploymentStatusUpdate:
+        """Update an existing model deployment (dispatches on the config's engine)."""
+        engine = config_engine(ctx.model_deployment_config)
         reconciler = self._select_reconciler(engine)
         if reconciler is None:
             return self._unsupported_engine(engine)
-        resolved = self._resolve(deployment, config, model_entity)
+        resolved = self._resolve(ctx)
         return await reconciler.update(resolved)
 
-    async def get_model_deployment_status(
-        self,
-        deployment: ModelDeployment,
-        config: Optional[ModelDeploymentConfig] = None,
-        model_entity: Optional[ModelEntity] = None,
-    ) -> DeploymentStatusUpdate:
+    async def get_model_deployment_status(self, ctx: ModelContext) -> DeploymentStatusUpdate:
         """Get the current status of a model deployment.
 
-        The engine is taken from ``config`` (same selection as create/update), so a
-        config is required. When ``config`` is ``None`` (e.g. the controller failed
-        to fetch it this cycle) the backend cannot determine the deployment's state
-        and returns ``UNKNOWN``; the controller retries on the next poll (which
-        normally has a config) and escalates to ERROR after its retry budget.
+        The engine is taken from the config (same selection as create/update), so a
+        config is required. When ``ctx.model_deployment_config`` is ``None`` (e.g.
+        the controller failed to fetch it this cycle) the backend cannot determine
+        the deployment's state and returns ``UNKNOWN``; the controller retries on
+        the next poll (which normally has a config) and escalates to ERROR after
+        its retry budget.
 
         In addition to the reconciler's status, this method enforces the PENDING
         timeout policy: if the deployment has been alive longer than
         ``pending_timeout_seconds`` and is still PENDING, transition to ERROR with
         diagnostic information. (Crash-loop detection is handled inside the
         reconciler's pod drill-down.)
         """
+        deployment = ctx.model_deployment
+        config = ctx.model_deployment_config
         logger.debug(
             f"Checking deployment status: {deployment.workspace}/{deployment.name} "
             f"(version: {deployment.entity_version})"
@@ -305,15 +319,15 @@ async def get_model_deployment_status(
                 return self._unsupported_engine(engine)
             # A reconciler MAY advance creation in get_status; it needs the
             # resolved config to compile the serving spec.
-            resolved = self._resolve(deployment, config, model_entity)
+            resolved = self._resolve(ctx)
             result = await reconciler.get_status(resolved)
 
             if result.status == "PENDING":
                 elapsed = deployment_elapsed_seconds(deployment)
 
                 if elapsed >= self._backend_config.pending_timeout_seconds:
-                    pod_name = self._nim_reconciler._find_pod_name(resource_name)
-                    return self._nim_reconciler._build_pending_timeout_error(resource_name, elapsed, pod_name)
+                    pod_name = self._status_projector.find_pod_name(resource_name)
+                    return self._status_projector.build_pending_timeout_error(resource_name, elapsed, pod_name)
 
                 # Use a stable message (no elapsed/timeout) so we don't create a new history entry every poll