feat(deployments): implement DockerDeploymentBackend (AIRCORE-756)

Add Docker backend for nemo-deployments-plugin with single-container v1
lifecycle, entity-store config fetch, port allocation, exec/http probes,
plugin-local GPU pool, and unit plus integration test coverage.

Stacks on #315 (AIRCORE-758 reconciler prerequisite DAG).

Signed-off-by: Tyler Bray <tbray@nvidia.com>

Author: tylersbray

plugins/nemo-deployments/pyproject.toml

@@ -6,11 +6,15 @@ readme = "README.md"
 requires-python = ">=3.11,<3.14"
 dependencies = [
     "fastapi>=0.115",
+    "httpx>=0.27",
     "nemo-platform",
     "nemo-platform-plugin",
     "pydantic>=2.10.6",
 ]
 
+[project.optional-dependencies]
+docker = ["docker>=7.0"]
+
 [project.entry-points."nemo.services"]
 deployments = "nemo_deployments_plugin.service:DeploymentsService"
 
@@ -29,7 +33,7 @@ nemo-platform = { workspace = true }
 nemo-platform-plugin = { workspace = true }
 
 [dependency-groups]
-dev = ["pytest>=8.3.4", "pytest-asyncio>=0.25.3", "httpx>=0.27", "fastapi>=0.115"]
+dev = ["pytest>=8.3.4", "pytest-asyncio>=0.25.3", "httpx>=0.27", "fastapi>=0.115", "docker>=7.0"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py

@@ -0,0 +1,16 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Executor-level Docker backend configuration."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class DockerExecutorConfig(BaseModel):
+    """Knobs for a named docker executor instance (not entity backend_config)."""
+
+    docker_host: str | None = Field(default=None, description="Override DOCKER_HOST for this executor.")
+    docker_timeout: int = Field(default=60, ge=1)
+    pull_images: bool = Field(default=True, description="Pull container images before run when missing locally.")

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/config.py

@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Compile DeploymentConfig into docker.containers.run kwargs."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from nemo_deployments_plugin.backends.docker.labels import docker_volume_name
+from nemo_deployments_plugin.entities import Container, DeploymentConfig, DockerDeploymentConfig, VolumeMount
+from nemo_deployments_plugin.types import RestartPolicy
+
+
+class DeploymentConfigError(ValueError):
+    """Invalid deployment config for docker backend."""
+
+
+def parse_docker_backend_config(backend_config: dict[str, Any]) -> DockerDeploymentConfig:
+    docker_section = backend_config.get("docker") or {}
+    return DockerDeploymentConfig.model_validate(docker_section)
+
+
+def validate_config_for_docker(config: DeploymentConfig) -> Container:
+    if config.init_containers:
+        raise DeploymentConfigError("init_containers are not supported by the docker backend in v1")
+    if len(config.containers) != 1:
+        raise DeploymentConfigError(f"docker backend v1 supports exactly one container; got {len(config.containers)}")
+    return config.containers[0]
+
+
+def restart_policy_kwargs(restart_policy: RestartPolicy, backoff_limit: int) -> dict[str, Any]:
+    if restart_policy == "Always":
+        return {"restart_policy": {"Name": "always"}}
+    if restart_policy == "OnFailure":
+        return {"restart_policy": {"Name": "on-failure", "MaximumRetryCount": backoff_limit}}
+    return {}
+
+
+def env_dict(container: Container) -> dict[str, str]:
+    result: dict[str, str] = {}
+    for item in container.env:
+        if item.value is not None:
+            result[item.name] = item.value
+    return result
+
+
+def merged_volume_mounts(config: DeploymentConfig, container: Container) -> list[VolumeMount]:
+    by_name: dict[str, VolumeMount] = {}
+    for mount in config.volume_mounts:
+        by_name[mount.name] = mount
+    for mount in container.volume_mounts:
+        by_name[mount.name] = mount
+    return list(by_name.values())
+
+
+def build_volume_bindings(
+    workspace: str,
+    mounts: list[VolumeMount],
+) -> dict[str, dict[str, str]]:
+    bindings: dict[str, dict[str, str]] = {}
+    for mount in mounts:
+        vol_name = docker_volume_name(workspace, mount.name)
+        bindings[vol_name] = {
+            "bind": mount.mount_path,
+            "mode": "ro" if mount.read_only else "rw",
+        }
+    return bindings
+
+
+def build_port_bindings(
+    container: Container,
+    host_ports: dict[int, int],
+) -> dict[str, int | list[tuple[str, int]] | None]:
+    ports: dict[str, int | list[tuple[str, int]] | None] = {}
+    for port_spec in container.ports:
+        container_port = port_spec.container_port
+        protocol = port_spec.protocol.lower()
+        key = f"{container_port}/{protocol}"
+        host_port = host_ports.get(container_port)
+        if host_port is not None:
+            ports[key] = host_port
+        else:
+            ports[key] = container_port
+    return ports
+
+
+def gpu_count_from_container(container: Container) -> int:
+    limit = container.resources.limits.get("nvidia.com/gpu")
+    if not limit:
+        return 0
+    try:
+        return int(limit)
+    except ValueError:
+        return 0
+
+
+def device_requests_for_gpus(gpu_ids: list[int]) -> list[dict[str, Any]]:
+    if not gpu_ids:
+        return []
+    return [
+        {
+            "Driver": "nvidia",
+            "Count": 0,
+            "DeviceIDs": [str(gpu_id) for gpu_id in gpu_ids],
+            "Capabilities": [["gpu"]],
+        }
+    ]

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/containers.py

@@ -0,0 +1,102 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Thread-safe GPU pool for Docker deployments (plugin-local; not shared with models).
+
+During the 759 cutover both pools may coexist briefly — consolidate into
+nemo_platform_plugin when models docker backend is removed.
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+import threading
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+
+class GPUAllocationError(Exception):
+    """Raised when GPU allocation fails due to insufficient resources."""
+
+
+@dataclass
+class GPUPoolStatus:
+    total: int
+    available: int
+    allocated: int
+    allocations: dict[str, list[int]] = field(default_factory=dict)
+    gpu_state: dict[int, str | None] = field(default_factory=dict)
+
+
+class DockerGPUPool:
+    """Thread-safe pool of GPU device IDs for Docker device_requests."""
+
+    def __init__(self, reserved_gpu_device_ids: list[int]) -> None:
+        self.num_reserved_gpus = len(reserved_gpu_device_ids)
+        self.gpu_to_workload_id: dict[int, str | None] = {gpu_id: None for gpu_id in reserved_gpu_device_ids}
+        self._mutex = threading.Lock()
+
+    def allocate_gpu(self, workload_id: str, num_requested: int = 1) -> list[int]:
+        with self._mutex:
+            if num_requested <= 0:
+                raise GPUAllocationError(f"Invalid GPU request: {num_requested}. Must be a positive integer.")
+            available_gpus = {gpu for gpu, workload in self.gpu_to_workload_id.items() if workload is None}
+            if len(available_gpus) < num_requested:
+                raise GPUAllocationError(
+                    f"Not enough GPUs available. Requested {num_requested}, "
+                    f"available {len(available_gpus)} out of {self.num_reserved_gpus} total."
+                )
+            gpu_ids: list[int] = []
+            for _ in range(num_requested):
+                gpu_id = available_gpus.pop()
+                gpu_ids.append(gpu_id)
+                self.gpu_to_workload_id[gpu_id] = workload_id
+            logger.info("DockerGPUPool: allocated gpu_ids %s to workload %s", gpu_ids, workload_id)
+            return gpu_ids
+
+    def release_gpu(self, workload_id: str) -> list[int]:
+        with self._mutex:
+            gpu_ids = [gpu for gpu, workload in self.gpu_to_workload_id.items() if workload == workload_id]
+            if gpu_ids:
+                logger.info("DockerGPUPool: releasing gpu_ids %s from workload %s", gpu_ids, workload_id)
+            for gpu_id in gpu_ids:
+                self.gpu_to_workload_id[gpu_id] = None
+            return gpu_ids
+
+
+_pool: DockerGPUPool | None = None
+_pool_lock = threading.Lock()
+
+
+def detect_gpu_device_ids() -> list[int]:
+    """Return GPU indices from nvidia-smi when available."""
+    try:
+        result = subprocess.run(
+            ["nvidia-smi", "--query-gpu=index", "--format=csv,noheader"],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=10,
+        )
+    except (FileNotFoundError, subprocess.SubprocessError, OSError):
+        return []
+    ids: list[int] = []
+    for line in result.stdout.splitlines():
+        stripped = line.strip()
+        if stripped.isdigit():
+            ids.append(int(stripped))
+    return ids
+
+
+def get_shared_gpu_pool() -> DockerGPUPool | None:
+    """Lazy singleton GPU pool shared across docker executor instances in this process."""
+    global _pool
+    with _pool_lock:
+        if _pool is None:
+            device_ids = detect_gpu_device_ids()
+            if not device_ids:
+                return None
+            _pool = DockerGPUPool(reserved_gpu_device_ids=device_ids)
+        return _pool

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/gpu.py

@@ -0,0 +1,83 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Docker resource naming and identity labels for orphan cleanup."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+
+from nemo_deployments_plugin.constants import MANAGED_BY_LABEL
+
+MANAGED_BY_KEY = "managed-by"
+DEPLOYMENT_WORKSPACE_LABEL = "nmp.nvidia.com/deployment-workspace"
+DEPLOYMENT_NAME_LABEL = "nmp.nvidia.com/deployment-name"
+RESTART_POLICY_LABEL = "nmp.nvidia.com/restart-policy"
+CONFIG_NAME_LABEL = "nmp.nvidia.com/deployment-config"
+VOLUME_WORKSPACE_LABEL = "nmp.nvidia.com/volume-workspace"
+VOLUME_NAME_LABEL = "nmp.nvidia.com/volume-name"
+
+
+def k8s_safe_name(base_name: str, *, max_length: int = 63, suffix: str = "") -> str:
+    """Generate a DNS-label-safe name (RFC 1035) from arbitrary input."""
+    hash_suffix = hashlib.sha256(base_name.encode()).hexdigest()[:8]
+    normalized = re.sub(r"[^a-z0-9-]", "-", base_name.lower())
+    normalized = re.sub(r"[-]+", "-", normalized)
+    if normalized and not normalized[0].isalpha():
+        normalized = f"x{normalized}"
+    normalized = normalized.rstrip("-")
+
+    reserved = len(suffix) + len(hash_suffix) + 1
+    if len(normalized) + reserved > max_length:
+        trim = max_length - reserved
+        normalized = normalized[:trim].rstrip("-")
+        normalized = f"{normalized}-{hash_suffix}{suffix}"
+    elif suffix:
+        normalized = f"{normalized}{suffix}"
+    return normalized
+
+
+def container_name(workspace: str, deployment_name: str) -> str:
+    return k8s_safe_name(f"dep-{workspace}-{deployment_name}")
+
+
+def docker_volume_name(workspace: str, volume_name: str) -> str:
+    return k8s_safe_name(f"dep-vol-{workspace}-{volume_name}")
+
+
+def deployment_key(workspace: str, name: str) -> str:
+    return f"{workspace}/{name}"
+
+
+BACKOFF_LIMIT_LABEL = "nmp.nvidia.com/backoff-limit"
+
+
+def deployment_identity_labels(
+    workspace: str,
+    name: str,
+    restart_policy: str,
+    *,
+    config_name: str,
+    backoff_limit: int = 6,
+) -> dict[str, str]:
+    return {
+        MANAGED_BY_KEY: MANAGED_BY_LABEL,
+        DEPLOYMENT_WORKSPACE_LABEL: workspace,
+        DEPLOYMENT_NAME_LABEL: name,
+        RESTART_POLICY_LABEL: restart_policy,
+        CONFIG_NAME_LABEL: config_name,
+        BACKOFF_LIMIT_LABEL: str(backoff_limit),
+    }
+
+
+def volume_identity_labels(workspace: str, name: str) -> dict[str, str]:
+    return {
+        MANAGED_BY_KEY: MANAGED_BY_LABEL,
+        VOLUME_WORKSPACE_LABEL: workspace,
+        VOLUME_NAME_LABEL: name,
+    }
+
+
+def managed_by_filter() -> dict[str, str]:
+    return {"label": f"{MANAGED_BY_KEY}={MANAGED_BY_LABEL}"}