diff --git a/plugins/nemo-deployments/openapi/openapi.yaml b/plugins/nemo-deployments/openapi/openapi.yaml index bcca9974d8..1cb4462758 100644 --- a/plugins/nemo-deployments/openapi/openapi.yaml +++ b/plugins/nemo-deployments/openapi/openapi.yaml @@ -581,6 +581,15 @@ components: $ref: '#/components/schemas/Probe' readinessProbe: $ref: '#/components/schemas/Probe' + restartPolicy: + title: Restartpolicy + description: Per-container restart policy for init containers; Always enables + k8s native sidecar. + type: string + enum: + - Always + - OnFailure + - Never type: object required: - name @@ -625,6 +634,15 @@ components: $ref: '#/components/schemas/Probe' readinessProbe: $ref: '#/components/schemas/Probe' + restartPolicy: + title: Restartpolicy + description: Per-container restart policy for init containers; Always enables + k8s native sidecar. + type: string + enum: + - Always + - OnFailure + - Never type: object required: - name diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/backend.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/backend.py index 9e92977dc9..49fcc22ab9 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/backend.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/backend.py @@ -43,7 +43,7 @@ class K8sDeploymentBackend(DeploymentBackend): """Manage deployments and volumes as native Kubernetes objects. Job-backed deployments (``restart_policy`` Never/OnFailure) are implemented in phase 3. - Deployment + Service (Always) is implemented in phase 4; full PodSpec compilation lands in phase 5. + Deployment + Service (Always) is implemented in phase 4; full PodSpec compilation is phase 5. """ _clients: KubernetesClients @@ -146,7 +146,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate if config.restart_policy == "Always": try: - container = deployment_ops.validate_config_for_deployment(config) + deployment_ops.validate_config_for_deployment(config) except job_ops.DeploymentConfigError as exc: return BackendStatusUpdate(status="FAILED", status_message=str(exc)) return await deployment_ops.read_deployment_status( @@ -158,7 +158,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate config_name=config.name, restart_policy=config.restart_policy, backoff_limit=config.backoff_limit, - container=container, + containers=tuple(config.containers), ) return await job_ops.read_job_status( diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/compiler.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/compiler.py new file mode 100644 index 0000000000..01b5886d19 --- /dev/null +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/compiler.py @@ -0,0 +1,473 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Compile DeploymentConfig + K8sDeploymentConfig into Kubernetes PodSpec objects. + +Native sidecars require Kubernetes >= 1.29 (init container ``restartPolicy=Always``). +On older clusters, omit per-container restart policy on init containers. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from types import SimpleNamespace +from typing import Any + +from kubernetes.client.rest import ApiException +from nemo_deployments_plugin.backends.k8s.client import k8s_client_module +from nemo_deployments_plugin.backends.k8s.status import resource_labels_match +from nemo_deployments_plugin.backends.labels import k8s_deployment_configmap_name, k8s_volume_resource_name +from nemo_deployments_plugin.entities import ( + Affinity, + ConfigFile, + Container, + ContainerPort, + DeploymentConfig, + K8sDeploymentConfig, + PodSecurityContext, + Probe, + Toleration, + VolumeMount, +) +from nemo_deployments_plugin.types import RestartPolicy + +CONFIG_FILES_VOLUME = "config-files" +NATIVE_SIDECAR_RESTART_POLICY: RestartPolicy = "Always" + +logger = logging.getLogger(__name__) + + +class DeploymentConfigError(ValueError): + """Invalid deployment config for k8s workload compilation.""" + + +def merged_volume_mounts(config: DeploymentConfig, container: Container) -> list[VolumeMount]: + mounts_by_name: dict[str, VolumeMount] = {} + for mount in config.volume_mounts: + mounts_by_name[mount.name] = mount + for mount in container.volume_mounts: + mounts_by_name[mount.name] = mount + return list(mounts_by_name.values()) + + +def build_env_vars(container: Container) -> list[Any]: + k8s = k8s_client_module() + return [k8s.client.V1EnvVar(name=item.name, value=item.value) for item in container.env if item.value is not None] + + +def build_resource_requirements(container: Container) -> Any | None: + limits = container.resources.limits or None + requests = container.resources.requests or None + if not limits and not requests: + return None + k8s = k8s_client_module() + return k8s.client.V1ResourceRequirements(limits=limits, requests=requests) + + +def build_pod_volumes(*, workspace: str, mounts: list[VolumeMount]) -> list[Any]: + if not mounts: + return [] + k8s = k8s_client_module() + return [ + k8s.client.V1Volume( + name=mount.name, + persistent_volume_claim=k8s.client.V1PersistentVolumeClaimVolumeSource( + claim_name=k8s_volume_resource_name(workspace, mount.name), + ), + ) + for mount in mounts + ] + + +def build_volume_mounts(mounts: list[VolumeMount]) -> list[Any]: + if not mounts: + return [] + k8s = k8s_client_module() + return [ + k8s.client.V1VolumeMount( + name=mount.name, + mount_path=mount.mount_path, + read_only=mount.read_only, + sub_path=mount.sub_path, + ) + for mount in mounts + ] + + +def build_container_spec(container: Container, *, volume_mounts: list[VolumeMount] | None = None) -> Any: + k8s = k8s_client_module() + kwargs: dict[str, Any] = { + "name": container.name, + "image": container.image, + "env": build_env_vars(container) or None, + "resources": build_resource_requirements(container), + } + if container.command: + kwargs["command"] = list(container.command) + if container.args: + kwargs["args"] = list(container.args) + if volume_mounts: + kwargs["volume_mounts"] = build_volume_mounts(volume_mounts) + return k8s.client.V1Container(**kwargs) + + +@dataclass(frozen=True) +class CompiledWorkload: + """Kubernetes objects derived from a DeploymentConfig.""" + + pod_spec_kwargs: dict[str, Any] + configmap_body: Any | None + configmap_name: str | None + service_containers: tuple[Container, ...] + + +def _reraise_api_unless(exc: ApiException, *allowed_statuses: int) -> None: + if exc.status not in allowed_statuses: + raise exc + + +def _validate_port_names(config: DeploymentConfig) -> None: + seen_names: set[str] = set() + seen_ports: set[tuple[int, str]] = set() + for container in (*config.init_containers, *config.containers): + for port in container.ports: + port_name = port.name or f"port-{port.container_port}" + if port_name in seen_names: + raise DeploymentConfigError(f"duplicate container port name {port_name!r}") + seen_names.add(port_name) + port_key = (port.container_port, port.protocol) + if port_key in seen_ports: + raise DeploymentConfigError( + f"duplicate container port {port.container_port}/{port.protocol} across containers" + ) + seen_ports.add(port_key) + + +def validate_workload_config(config: DeploymentConfig) -> None: + """Validate container lists shared by Job and Deployment backends.""" + if not config.containers: + raise DeploymentConfigError("at least one container is required") + _validate_port_names(config) + for container in config.containers: + if container.restart_policy is not None: + raise DeploymentConfigError( + f"container {container.name} sets restart_policy; only init_containers may use per-container restart_policy" + ) + for init_container in config.init_containers: + if init_container.restart_policy not in (None, NATIVE_SIDECAR_RESTART_POLICY): + raise DeploymentConfigError( + f"init container {init_container.name} has unsupported restart_policy " + f"{init_container.restart_policy!r}; only Always (native sidecar) is supported" + ) + + +def validate_config_for_job(config: DeploymentConfig) -> None: + validate_workload_config(config) + if config.restart_policy == "Always": + raise DeploymentConfigError("restart_policy Always requires a Deployment workload, not a Job") + + +def validate_config_for_deployment(config: DeploymentConfig) -> None: + validate_workload_config(config) + if config.restart_policy != "Always": + raise DeploymentConfigError("restart_policy Always is required for Deployment + Service") + + +def configmap_data_key(path: str) -> str: + normalized = path if path.startswith("/") else f"/{path}" + key = normalized.lstrip("/").replace("/", "__") + return key or "config" + + +def _deserialize_k8s(data: dict[str, Any], klass: str) -> Any: + k8s = k8s_client_module() + response = SimpleNamespace(data=json.dumps(data)) + return k8s.client.ApiClient().deserialize(response=response, response_type=klass) + + +def _build_probe(probe: Probe | None) -> Any | None: + if probe is None: + return None + k8s = k8s_client_module() + kwargs: dict[str, Any] = { + "initial_delay_seconds": probe.initial_delay_seconds, + "period_seconds": probe.period_seconds, + "timeout_seconds": probe.timeout_seconds, + "failure_threshold": probe.failure_threshold, + } + if probe.exec_action is not None: + kwargs["exec"] = k8s.client.V1ExecAction(command=list(probe.exec_action.command)) + elif probe.http_get is not None: + kwargs["http_get"] = k8s.client.V1HTTPGetAction( + path=probe.http_get.path, + port=probe.http_get.port, + scheme=probe.http_get.scheme, + ) + elif probe.tcp_socket is not None: + kwargs["tcp_socket"] = k8s.client.V1TCPSocketAction(port=probe.tcp_socket.port) + else: + return None + return k8s.client.V1Probe(**kwargs) + + +def _build_container_ports(ports: list[ContainerPort]) -> list[Any]: + if not ports: + return [] + k8s = k8s_client_module() + return [ + k8s.client.V1ContainerPort( + name=port.name or f"port-{port.container_port}", + container_port=port.container_port, + protocol=port.protocol, + ) + for port in ports + ] + + +def _config_file_mounts(config_files: list[ConfigFile]) -> list[VolumeMount]: + return [ + VolumeMount( + name=CONFIG_FILES_VOLUME, + mountPath=config_file.path, + readOnly=True, + subPath=config_file.path.lstrip("/"), + ) + for config_file in config_files + ] + + +def _collect_pvc_mounts(config: DeploymentConfig) -> list[VolumeMount]: + mounts_by_name: dict[str, VolumeMount] = {} + for mount in config.volume_mounts: + mounts_by_name[mount.name] = mount + for container in (*config.init_containers, *config.containers): + for mount in container.volume_mounts: + mounts_by_name[mount.name] = mount + return list(mounts_by_name.values()) + + +def build_container( + container: Container, + *, + config: DeploymentConfig, + include_probes: bool, +) -> Any: + """Build a V1Container from a plugin Container.""" + k8s = k8s_client_module() + mounts = merged_volume_mounts(config, container) + if config.config_files: + mounts = [*mounts, *_config_file_mounts(config.config_files)] + base = build_container_spec(container, volume_mounts=mounts or None) + kwargs: dict[str, Any] = { + "name": base.name, + "image": base.image, + "command": base.command, + "args": base.args, + "env": base.env, + "resources": base.resources, + "volume_mounts": build_volume_mounts(mounts) if mounts else base.volume_mounts, + "ports": _build_container_ports(container.ports) or None, + } + if include_probes: + kwargs["liveness_probe"] = _build_probe(container.liveness_probe) + kwargs["readiness_probe"] = _build_probe(container.readiness_probe) + if container.restart_policy == NATIVE_SIDECAR_RESTART_POLICY: + kwargs["restart_policy"] = NATIVE_SIDECAR_RESTART_POLICY + return k8s.client.V1Container(**{key: value for key, value in kwargs.items() if value is not None}) + + +def _ordered_init_containers(config: DeploymentConfig) -> list[Container]: + sequential = [c for c in config.init_containers if c.restart_policy != NATIVE_SIDECAR_RESTART_POLICY] + sidecars = [c for c in config.init_containers if c.restart_policy == NATIVE_SIDECAR_RESTART_POLICY] + return [*sequential, *sidecars] + + +def build_tolerations(tolerations: list[Toleration]) -> list[Any]: + if not tolerations: + return [] + k8s = k8s_client_module() + return [k8s.client.V1Toleration(**item.model_dump(by_alias=False, exclude_none=True)) for item in tolerations] + + +def build_affinity(affinity: Affinity | None) -> Any | None: + if affinity is None: + return None + payload = affinity.model_dump(by_alias=True, exclude_none=True) + if not payload: + return None + return _deserialize_k8s(payload, "V1Affinity") + + +def build_pod_security_context(security_context: PodSecurityContext | None) -> Any | None: + if security_context is None: + return None + k8s = k8s_client_module() + payload = security_context.model_dump(by_alias=False, exclude_none=True) + if not payload: + return None + return k8s.client.V1PodSecurityContext(**payload) + + +def build_configmap_body( + *, + workspace: str, + deployment_name: str, + labels: dict[str, str], + config_files: list[ConfigFile], +) -> Any | None: + if not config_files: + return None + k8s = k8s_client_module() + data = {configmap_data_key(config_file.path): config_file.content for config_file in config_files} + return k8s.client.V1ConfigMap( + api_version="v1", + kind="ConfigMap", + metadata=k8s.client.V1ObjectMeta( + name=k8s_deployment_configmap_name(workspace, deployment_name), + labels=labels, + ), + data=data, + ) + + +def _build_config_file_volume(configmap_name: str, config_files: list[ConfigFile]) -> Any: + k8s = k8s_client_module() + items = [ + k8s.client.V1KeyToPath(key=configmap_data_key(config_file.path), path=config_file.path.lstrip("/")) + for config_file in config_files + ] + return k8s.client.V1Volume( + name=CONFIG_FILES_VOLUME, + config_map=k8s.client.V1ConfigMapVolumeSource(name=configmap_name, items=items), + ) + + +def compile_workload( + *, + config: DeploymentConfig, + workspace: str, + deployment_name: str, + labels: dict[str, str], + k8s_config: K8sDeploymentConfig | None, + pod_restart_policy: RestartPolicy, +) -> CompiledWorkload: + """Compile pod spec kwargs and optional ConfigMap for a Job or Deployment.""" + validate_workload_config(config) + pvc_mounts = _collect_pvc_mounts(config) + volumes = build_pod_volumes(workspace=workspace, mounts=pvc_mounts) + configmap_body = build_configmap_body( + workspace=workspace, + deployment_name=deployment_name, + labels=labels, + config_files=config.config_files, + ) + configmap_name = configmap_body.metadata.name if configmap_body is not None else None + if configmap_name is not None: + volumes = [*volumes, _build_config_file_volume(configmap_name, config.config_files)] + + init_containers = [ + build_container( + container, + config=config, + include_probes=container.restart_policy == NATIVE_SIDECAR_RESTART_POLICY, + ) + for container in _ordered_init_containers(config) + ] + main_containers = [ + build_container(container, config=config, include_probes=True) for container in config.containers + ] + + pod_spec_kwargs: dict[str, Any] = { + "restart_policy": pod_restart_policy, + "containers": main_containers, + } + if init_containers: + pod_spec_kwargs["init_containers"] = init_containers + if volumes: + pod_spec_kwargs["volumes"] = volumes + + if k8s_config is not None: + tolerations = build_tolerations(k8s_config.tolerations) + if tolerations: + pod_spec_kwargs["tolerations"] = tolerations + affinity = build_affinity(k8s_config.affinity) + if affinity is not None: + pod_spec_kwargs["affinity"] = affinity + security_context = build_pod_security_context(k8s_config.security_context) + if security_context is not None: + pod_spec_kwargs["security_context"] = security_context + if k8s_config.service_account: + pod_spec_kwargs["service_account_name"] = k8s_config.service_account + + return CompiledWorkload( + pod_spec_kwargs=pod_spec_kwargs, + configmap_body=configmap_body, + configmap_name=configmap_name, + service_containers=tuple(config.containers), + ) + + +def create_configmap( + core_v1: Any, + *, + namespace: str, + body: Any, + expected_labels: dict[str, str], + timeout: float | None, +) -> None: + try: + core_v1.create_namespaced_config_map(namespace=namespace, body=body, _request_timeout=timeout) + except ApiException as exc: + _reraise_api_unless(exc, 409) + existing = core_v1.read_namespaced_config_map( + name=body.metadata.name, + namespace=namespace, + _request_timeout=timeout, + ) + if not resource_labels_match(existing, expected_labels): + raise + + +def delete_configmap_best_effort( + core_v1: Any, + *, + namespace: str, + name: str | None, + expected_labels: dict[str, str], + timeout: float | None, +) -> None: + if name is None: + return + try: + delete_configmap( + core_v1, + namespace=namespace, + name=name, + expected_labels=expected_labels, + timeout=timeout, + ) + except ApiException: + logger.debug("Best-effort ConfigMap cleanup failed for %s", name, exc_info=True) + + +def delete_configmap( + core_v1: Any, + *, + namespace: str, + name: str, + expected_labels: dict[str, str], + timeout: float | None, +) -> None: + try: + configmap = core_v1.read_namespaced_config_map(name=name, namespace=namespace, _request_timeout=timeout) + except ApiException as exc: + _reraise_api_unless(exc, 404) + return + if not resource_labels_match(configmap, expected_labels): + return + try: + core_v1.delete_namespaced_config_map(name=name, namespace=namespace, _request_timeout=timeout) + except ApiException as exc: + _reraise_api_unless(exc, 404) diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/deployments.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/deployments.py index c5cb262f86..db72bfa2d8 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/deployments.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/deployments.py @@ -11,17 +11,22 @@ import asyncio import logging +from dataclasses import dataclass from typing import Any from kubernetes.client.rest import ApiException from nemo_deployments_plugin.backends.base import BackendStatusUpdate, LogResult from nemo_deployments_plugin.backends.k8s.client import KubernetesClients, k8s_client_module -from nemo_deployments_plugin.backends.k8s.jobs import ( +from nemo_deployments_plugin.backends.k8s.compiler import ( + CompiledWorkload, DeploymentConfigError, - build_container_spec, - build_pod_volumes, - build_volume_mounts, - merged_volume_mounts, + compile_workload, + create_configmap, + delete_configmap, + delete_configmap_best_effort, + validate_config_for_deployment, +) +from nemo_deployments_plugin.backends.k8s.jobs import ( newest_pod, resolve_deployment_namespace, resolve_k8s_deployment_config, @@ -38,93 +43,30 @@ DEPLOYMENT_WORKSPACE_LABEL, RESTART_POLICY_LABEL, deployment_identity_labels, + k8s_deployment_configmap_name, k8s_deployment_resource_name, managed_by_label_selector, ) -from nemo_deployments_plugin.entities import Container, ContainerPort, DeploymentConfig, Probe, VolumeMount +from nemo_deployments_plugin.entities import Container, DeploymentConfig, K8sDeploymentConfig from nemo_deployments_plugin.types import Endpoint, RestartPolicy logger = logging.getLogger(__name__) -APP_LABEL = "app" -DEFAULT_SERVICE_PORT = 8080 - -def app_selector_labels(resource_name: str) -> dict[str, str]: - return {APP_LABEL: resource_name} +@dataclass(frozen=True) +class BuiltDeployment: + """An apps/v1.Deployment plus the compiled workload used to build its pod template.""" - -def validate_config_for_deployment(config: DeploymentConfig) -> Container: - """Return the sole container spec for a Deployment-backed deployment.""" - if config.init_containers: - # User-specified initContainers are deferred to phase 5. Mesh sidecars (Istio, - # Linkerd) inject at admission time and do not appear in DeploymentConfig. - raise DeploymentConfigError("init_containers are not supported by the k8s Deployment backend in this phase") - if len(config.containers) != 1: - raise DeploymentConfigError( - f"k8s Deployment backend supports exactly one container; got {len(config.containers)}" - ) - if config.restart_policy != "Always": - raise DeploymentConfigError("restart_policy Always is required for Deployment + Service") - return config.containers[0] + deployment: Any + compiled: CompiledWorkload -def _build_probe(probe: Probe | None) -> Any | None: - if probe is None: - return None - k8s = k8s_client_module() - kwargs: dict[str, Any] = { - "initial_delay_seconds": probe.initial_delay_seconds, - "period_seconds": probe.period_seconds, - "timeout_seconds": probe.timeout_seconds, - "failure_threshold": probe.failure_threshold, - } - if probe.exec_action is not None: - kwargs["exec"] = k8s.client.V1ExecAction(command=list(probe.exec_action.command)) - elif probe.http_get is not None: - kwargs["http_get"] = k8s.client.V1HTTPGetAction( - path=probe.http_get.path, - port=probe.http_get.port, - scheme=probe.http_get.scheme, - ) - elif probe.tcp_socket is not None: - kwargs["tcp_socket"] = k8s.client.V1TCPSocketAction(port=probe.tcp_socket.port) - else: - return None - return k8s.client.V1Probe(**kwargs) - - -def _build_container_ports(ports: list[ContainerPort]) -> list[Any]: - if not ports: - return [] - k8s = k8s_client_module() - return [ - k8s.client.V1ContainerPort( - name=port.name or f"port-{port.container_port}", - container_port=port.container_port, - protocol=port.protocol, - ) - for port in ports - ] +APP_LABEL = "app" +DEFAULT_SERVICE_PORT = 8080 -def build_deployment_container_spec(container: Container, *, volume_mounts: list[VolumeMount] | None = None) -> Any: - k8s = k8s_client_module() - base = build_container_spec(container, volume_mounts=volume_mounts) - ports = _build_container_ports(container.ports) - kwargs: dict[str, Any] = { - "name": base.name, - "image": base.image, - "command": base.command, - "args": base.args, - "env": base.env, - "resources": base.resources, - "volume_mounts": build_volume_mounts(volume_mounts) if volume_mounts else base.volume_mounts, - "ports": ports or None, - "liveness_probe": _build_probe(container.liveness_probe), - "readiness_probe": _build_probe(container.readiness_probe), - } - return k8s.client.V1Container(**{key: value for key, value in kwargs.items() if value is not None}) +def app_selector_labels(resource_name: str) -> dict[str, str]: + return {APP_LABEL: resource_name} def build_deployment_body( @@ -132,22 +74,23 @@ def build_deployment_body( resource_name: str, labels: dict[str, str], config: DeploymentConfig, - container: Container, workspace: str, -) -> Any: - """Build an ``apps/v1.Deployment`` for create.""" + deployment_name: str, + k8s_config: K8sDeploymentConfig | None, +) -> BuiltDeployment: + """Build an ``apps/v1.Deployment`` for create and its compiled workload.""" k8s = k8s_client_module() selector_labels = app_selector_labels(resource_name) pod_labels = selector_labels | labels - mounts = merged_volume_mounts(config, container) - pod_spec_kwargs: dict[str, Any] = { - "containers": [build_deployment_container_spec(container, volume_mounts=mounts or None)], - } - volumes = build_pod_volumes(workspace=workspace, mounts=mounts) - if volumes: - pod_spec_kwargs["volumes"] = volumes - - return k8s.client.V1Deployment( + compiled = compile_workload( + config=config, + workspace=workspace, + deployment_name=deployment_name, + labels=labels, + k8s_config=k8s_config, + pod_restart_policy="Always", + ) + deployment = k8s.client.V1Deployment( api_version="apps/v1", kind="Deployment", metadata=k8s.client.V1ObjectMeta(name=resource_name, labels=labels), @@ -156,27 +99,34 @@ def build_deployment_body( selector=k8s.client.V1LabelSelector(match_labels=selector_labels), template=k8s.client.V1PodTemplateSpec( metadata=k8s.client.V1ObjectMeta(labels=pod_labels), - spec=k8s.client.V1PodSpec(**pod_spec_kwargs), + spec=k8s.client.V1PodSpec(**compiled.pod_spec_kwargs), ), ), ) + return BuiltDeployment(deployment=deployment, compiled=compiled) -def build_service_body(*, resource_name: str, labels: dict[str, str], container: Container) -> Any: +def build_service_body(*, resource_name: str, labels: dict[str, str], containers: tuple[Container, ...]) -> Any: """Build a ClusterIP ``v1.Service`` exposing container ports in-cluster.""" k8s = k8s_client_module() selector_labels = app_selector_labels(resource_name) service_ports: list[Any] = [] - for port in container.ports: - port_name = port.name or f"port-{port.container_port}" - service_ports.append( - k8s.client.V1ServicePort( - name=port_name, - port=port.container_port, - target_port=port_name, - protocol=port.protocol, + seen_ports: set[tuple[int, str]] = set() + for container in containers: + for port in container.ports: + port_key = (port.container_port, port.protocol) + if port_key in seen_ports: + continue + seen_ports.add(port_key) + port_name = port.name or f"port-{port.container_port}" + service_ports.append( + k8s.client.V1ServicePort( + name=port_name, + port=port.container_port, + target_port=port_name, + protocol=port.protocol, + ) ) - ) if not service_ports: service_ports.append( k8s.client.V1ServicePort( @@ -198,11 +148,16 @@ def build_service_body(*, resource_name: str, labels: dict[str, str], container: ) -def build_in_cluster_endpoints(*, resource_name: str, namespace: str, container: Container) -> list[Endpoint]: - """Build in-cluster HTTP endpoints for exposed container ports.""" +def build_in_cluster_endpoints( + *, + resource_name: str, + namespace: str, + containers: tuple[Container, ...], +) -> list[Endpoint]: + """Build in-cluster endpoints for exposed container ports.""" endpoints: list[Endpoint] = [] host = f"{resource_name}.{namespace}.svc.cluster.local" - if container.ports: + for container in containers: for port in container.ports: endpoint_name = port.name or f"port-{port.container_port}" is_udp = port.protocol == "UDP" @@ -215,6 +170,7 @@ def build_in_cluster_endpoints(*, resource_name: str, namespace: str, container: protocol=protocol, ) ) + if endpoints: return endpoints endpoints.append(Endpoint(name="http", url=f"http://{host}:{DEFAULT_SERVICE_PORT}", protocol="http")) return endpoints @@ -271,7 +227,7 @@ async def create_deployment( ) -> BackendStatusUpdate: resource_name = k8s_deployment_resource_name(workspace, name) try: - container = validate_config_for_deployment(config) + validate_config_for_deployment(config) k8s_config = resolve_k8s_deployment_config(backend_config) namespace = resolve_deployment_namespace(default_namespace=default_namespace, k8s_config=k8s_config) identity_labels = deployment_identity_labels( @@ -282,20 +238,56 @@ async def create_deployment( backoff_limit=config.backoff_limit, ) all_labels = {**labels, **config.labels, **identity_labels} - deployment_body = build_deployment_body( + built = build_deployment_body( resource_name=resource_name, labels=all_labels, config=config, - container=container, workspace=workspace, + deployment_name=name, + k8s_config=k8s_config, + ) + deployment_body = built.deployment + compiled = built.compiled + service_body = build_service_body( + resource_name=resource_name, + labels=all_labels, + containers=compiled.service_containers, ) - service_body = build_service_body(resource_name=resource_name, labels=all_labels, container=container) timeout = clients.request_timeout apps_v1 = clients.apps_v1 core_v1 = clients.core_v1 + def _rollback_partial_create(*, deployment_created: bool, delete_configmap: bool) -> None: + if deployment_created: + try: + apps_v1.delete_namespaced_deployment( + name=resource_name, + namespace=namespace, + propagation_policy="Background", + _request_timeout=timeout, + ) + except ApiException as cleanup_exc: + _log_cleanup_ignored(resource_name, cleanup_exc) + if delete_configmap: + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=compiled.configmap_name, + expected_labels=identity_labels, + timeout=timeout, + ) + def _create() -> Any: deployment_created = False + configmap_written = compiled.configmap_body is not None + if configmap_written: + create_configmap( + core_v1, + namespace=namespace, + body=compiled.configmap_body, + expected_labels=identity_labels, + timeout=timeout, + ) try: apps_v1.create_namespaced_deployment( namespace=namespace, @@ -305,6 +297,14 @@ def _create() -> Any: deployment_created = True except ApiException as exc: if exc.status != 409: + if configmap_written: + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=compiled.configmap_name, + expected_labels=identity_labels, + timeout=timeout, + ) raise deployment = apps_v1.read_namespaced_deployment( @@ -313,16 +313,7 @@ def _create() -> Any: _request_timeout=timeout, ) if not resource_labels_match(deployment, identity_labels): - if deployment_created: - try: - apps_v1.delete_namespaced_deployment( - name=resource_name, - namespace=namespace, - propagation_policy="Background", - _request_timeout=timeout, - ) - except ApiException as cleanup_exc: - _log_cleanup_ignored(resource_name, cleanup_exc) + _rollback_partial_create(deployment_created=deployment_created, delete_configmap=configmap_written) return deployment try: @@ -339,33 +330,25 @@ def _create() -> Any: _request_timeout=timeout, ) if not resource_labels_match(existing_service, identity_labels): - if deployment_created: - try: - apps_v1.delete_namespaced_deployment( - name=resource_name, - namespace=namespace, - propagation_policy="Background", - _request_timeout=timeout, - ) - except ApiException as cleanup_exc: - _log_cleanup_ignored(resource_name, cleanup_exc) + _rollback_partial_create( + deployment_created=deployment_created, + delete_configmap=deployment_created and configmap_written, + ) raise return deployment - if deployment_created: - try: - apps_v1.delete_namespaced_deployment( - name=resource_name, - namespace=namespace, - propagation_policy="Background", - _request_timeout=timeout, - ) - except ApiException as cleanup_exc: - _log_cleanup_ignored(resource_name, cleanup_exc) + _rollback_partial_create( + deployment_created=deployment_created, + delete_configmap=deployment_created and configmap_written, + ) raise return deployment deployment = await asyncio.to_thread(_create) - endpoints = build_in_cluster_endpoints(resource_name=resource_name, namespace=namespace, container=container) + endpoints = build_in_cluster_endpoints( + resource_name=resource_name, + namespace=namespace, + containers=compiled.service_containers, + ) pod = await _read_newest_pod(clients, namespace=namespace, match_labels=app_selector_labels(resource_name)) return status_from_deployment( deployment=deployment, @@ -391,7 +374,7 @@ async def read_deployment_status( config_name: str, restart_policy: RestartPolicy, backoff_limit: int, - container: Container, + containers: tuple[Container, ...], ) -> BackendStatusUpdate: resource_name = k8s_deployment_resource_name(workspace, name) expected_labels = deployment_identity_labels( @@ -421,7 +404,11 @@ def _read() -> Any: status="FAILED", status_message=f"Deployment {resource_name} is missing deployment config identity labels", ) - endpoints = build_in_cluster_endpoints(resource_name=resource_name, namespace=namespace, container=container) + endpoints = build_in_cluster_endpoints( + resource_name=resource_name, + namespace=namespace, + containers=containers, + ) pod = await _read_newest_pod(clients, namespace=namespace, match_labels=app_selector_labels(resource_name)) return status_from_deployment( deployment=deployment, @@ -448,6 +435,7 @@ async def delete_deployment( expected_labels: dict[str, str], ) -> BackendStatusUpdate: resource_name = k8s_deployment_resource_name(workspace, name) + configmap_name = k8s_deployment_configmap_name(workspace, name) try: k8s_config = resolve_k8s_deployment_config(backend_config) namespace = resolve_deployment_namespace(default_namespace=default_namespace, k8s_config=k8s_config) @@ -473,6 +461,13 @@ def _delete() -> str | None: except ApiException as service_read_exc: if service_read_exc.status != 404: raise + delete_configmap( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return None if resource_labels_match(service, expected_labels): try: @@ -484,6 +479,13 @@ def _delete() -> str | None: except ApiException as service_exc: if service_exc.status != 404: raise + delete_configmap( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return None raise if not resource_labels_match(deployment, expected_labels): @@ -503,10 +505,24 @@ def _delete() -> str | None: except ApiException as exc: if exc.status != 404: raise + delete_configmap( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return "deleted" result = await asyncio.to_thread(_delete) if result == "foreign": + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return BackendStatusUpdate( status="FAILED", status_message=f"Deployment {resource_name} exists but is not managed by this plugin", diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/jobs.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/jobs.py index 3435ba7f9d..df254b10ae 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/jobs.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/jobs.py @@ -7,11 +7,21 @@ import asyncio import logging +from dataclasses import dataclass from typing import Any from kubernetes.client.rest import ApiException from nemo_deployments_plugin.backends.base import BackendStatusUpdate, LogResult from nemo_deployments_plugin.backends.k8s.client import KubernetesClients, k8s_client_module +from nemo_deployments_plugin.backends.k8s.compiler import ( + CompiledWorkload, + DeploymentConfigError, + compile_workload, + create_configmap, + delete_configmap, + delete_configmap_best_effort, + validate_config_for_job, +) from nemo_deployments_plugin.backends.k8s.status import ( LOG_MAX_CHARS, missing_job_status, @@ -24,19 +34,23 @@ DEPLOYMENT_WORKSPACE_LABEL, MANAGED_BY_KEY, deployment_identity_labels, + k8s_deployment_configmap_name, k8s_deployment_resource_name, - k8s_volume_resource_name, managed_by_label_selector, ) from nemo_deployments_plugin.constants import MANAGED_BY_LABEL -from nemo_deployments_plugin.entities import Container, DeploymentConfig, K8sDeploymentConfig, VolumeMount +from nemo_deployments_plugin.entities import DeploymentConfig, K8sDeploymentConfig from nemo_deployments_plugin.types import RestartPolicy logger = logging.getLogger(__name__) -class DeploymentConfigError(ValueError): - """Invalid deployment config for k8s Job backend.""" +@dataclass(frozen=True) +class BuiltJob: + """A batch/v1.Job plus the compiled workload used to build its pod template.""" + + job: Any + compiled: CompiledWorkload def resolve_k8s_deployment_config(backend_config: dict[str, Any]) -> K8sDeploymentConfig | None: @@ -85,113 +99,32 @@ def trim_log_text(text: str) -> tuple[list[str], bool]: return lines, truncated -def validate_config_for_job(config: DeploymentConfig) -> Container: - """Return the sole container spec for a Job-backed deployment.""" - if config.init_containers: - raise DeploymentConfigError("init_containers are not supported by the k8s Job backend in this phase") - if len(config.containers) != 1: - raise DeploymentConfigError(f"k8s Job backend supports exactly one container; got {len(config.containers)}") - if config.restart_policy == "Always": - raise DeploymentConfigError("restart_policy Always uses Deployment, not Job") - return config.containers[0] - - -def merged_volume_mounts(config: DeploymentConfig, container: Container) -> list[VolumeMount]: - mounts_by_name: dict[str, VolumeMount] = {} - for mount in config.volume_mounts: - mounts_by_name[mount.name] = mount - for mount in container.volume_mounts: - mounts_by_name[mount.name] = mount - return list(mounts_by_name.values()) - - def job_backoff_limit(config: DeploymentConfig) -> int: if config.restart_policy == "Never": return 0 return config.backoff_limit -def build_env_vars(container: Container) -> list[Any]: - k8s = k8s_client_module() - return [k8s.client.V1EnvVar(name=item.name, value=item.value) for item in container.env if item.value is not None] - - -def build_resource_requirements(container: Container) -> Any | None: - limits = container.resources.limits or None - requests = container.resources.requests or None - if not limits and not requests: - return None - k8s = k8s_client_module() - return k8s.client.V1ResourceRequirements(limits=limits, requests=requests) - - -def build_pod_volumes(*, workspace: str, mounts: list[VolumeMount]) -> list[Any]: - if not mounts: - return [] - k8s = k8s_client_module() - return [ - k8s.client.V1Volume( - name=mount.name, - persistent_volume_claim=k8s.client.V1PersistentVolumeClaimVolumeSource( - claim_name=k8s_volume_resource_name(workspace, mount.name), - ), - ) - for mount in mounts - ] - - -def build_volume_mounts(mounts: list[VolumeMount]) -> list[Any]: - if not mounts: - return [] - k8s = k8s_client_module() - return [ - k8s.client.V1VolumeMount( - name=mount.name, - mount_path=mount.mount_path, - read_only=mount.read_only, - sub_path=mount.sub_path, - ) - for mount in mounts - ] - - -def build_container_spec(container: Container, *, volume_mounts: list[VolumeMount] | None = None) -> Any: - k8s = k8s_client_module() - kwargs: dict[str, Any] = { - "name": container.name, - "image": container.image, - "env": build_env_vars(container) or None, - "resources": build_resource_requirements(container), - } - if container.command: - kwargs["command"] = list(container.command) - if container.args: - kwargs["args"] = list(container.args) - if volume_mounts: - kwargs["volume_mounts"] = build_volume_mounts(volume_mounts) - return k8s.client.V1Container(**kwargs) - - def build_job_body( *, job_name: str, labels: dict[str, str], config: DeploymentConfig, - container: Container, workspace: str, -) -> Any: + deployment_name: str, + k8s_config: K8sDeploymentConfig | None, +) -> BuiltJob: """Build a ``batch/v1.Job`` for create.""" k8s = k8s_client_module() - mounts = merged_volume_mounts(config, container) - pod_spec_kwargs: dict[str, Any] = { - "restart_policy": config.restart_policy, - "containers": [build_container_spec(container, volume_mounts=mounts or None)], - } - volumes = build_pod_volumes(workspace=workspace, mounts=mounts) - if volumes: - pod_spec_kwargs["volumes"] = volumes - - return k8s.client.V1Job( + compiled = compile_workload( + config=config, + workspace=workspace, + deployment_name=deployment_name, + labels=labels, + k8s_config=k8s_config, + pod_restart_policy=config.restart_policy, + ) + job = k8s.client.V1Job( api_version="batch/v1", kind="Job", metadata=k8s.client.V1ObjectMeta(name=job_name, labels=labels), @@ -199,10 +132,11 @@ def build_job_body( backoff_limit=job_backoff_limit(config), template=k8s.client.V1PodTemplateSpec( metadata=k8s.client.V1ObjectMeta(labels=labels), - spec=k8s.client.V1PodSpec(**pod_spec_kwargs), + spec=k8s.client.V1PodSpec(**compiled.pod_spec_kwargs), ), ), ) + return BuiltJob(job=job, compiled=compiled) async def read_pod_exit_code( @@ -252,7 +186,7 @@ async def create_job( ) -> BackendStatusUpdate: job_name = k8s_deployment_resource_name(workspace, name) try: - container = validate_config_for_job(config) + validate_config_for_job(config) k8s_config = resolve_k8s_deployment_config(backend_config) namespace = resolve_deployment_namespace(default_namespace=default_namespace, k8s_config=k8s_config) identity_labels = deployment_identity_labels( @@ -263,17 +197,30 @@ async def create_job( backoff_limit=config.backoff_limit, ) all_labels = {**labels, **config.labels, **identity_labels} - body = build_job_body( + built = build_job_body( job_name=job_name, labels=all_labels, config=config, - container=container, workspace=workspace, + deployment_name=name, + k8s_config=k8s_config, ) + body = built.job + compiled = built.compiled timeout = clients.request_timeout batch_v1 = clients.batch_v1 + core_v1 = clients.core_v1 def _create() -> Any: + configmap_written = compiled.configmap_body is not None + if configmap_written: + create_configmap( + core_v1, + namespace=namespace, + body=compiled.configmap_body, + expected_labels=identity_labels, + timeout=timeout, + ) try: return batch_v1.create_namespaced_job( namespace=namespace, @@ -282,11 +229,28 @@ def _create() -> Any: ) except ApiException as exc: if exc.status == 409: - return batch_v1.read_namespaced_job( + job = batch_v1.read_namespaced_job( name=job_name, namespace=namespace, _request_timeout=timeout, ) + if not resource_labels_match(job, identity_labels) and configmap_written: + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=compiled.configmap_name, + expected_labels=identity_labels, + timeout=timeout, + ) + return job + if configmap_written: + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=compiled.configmap_name, + expected_labels=identity_labels, + timeout=timeout, + ) raise job = await asyncio.to_thread(_create) @@ -371,6 +335,8 @@ async def delete_job( namespace = resolve_deployment_namespace(default_namespace=default_namespace, k8s_config=k8s_config) timeout = clients.request_timeout batch_v1 = clients.batch_v1 + core_v1 = clients.core_v1 + configmap_name = k8s_deployment_configmap_name(workspace, name) def _delete() -> str | None: try: @@ -381,6 +347,13 @@ def _delete() -> str | None: ) except ApiException as exc: if exc.status == 404: + delete_configmap( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return None raise if not resource_labels_match(job, expected_labels): @@ -391,10 +364,24 @@ def _delete() -> str | None: propagation_policy="Background", _request_timeout=timeout, ) + delete_configmap( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return "deleted" result = await asyncio.to_thread(_delete) if result == "foreign": + delete_configmap_best_effort( + core_v1, + namespace=namespace, + name=configmap_name, + expected_labels=expected_labels, + timeout=timeout, + ) return BackendStatusUpdate( status="FAILED", status_message=f"Job {job_name} exists but is not managed by this plugin", diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/labels.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/labels.py index 430b0835c4..17068e7b4d 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/labels.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/labels.py @@ -58,6 +58,14 @@ def k8s_volume_resource_name(workspace: str, volume_name: str) -> str: ) +def k8s_deployment_configmap_name(workspace: str, deployment_name: str) -> str: + """Kubernetes ConfigMap name for deployment config files.""" + return k8s_safe_name( + f"dep-cm-{workspace}-{deployment_name}", + hash_input=f"{deployment_key(workspace, deployment_name)}/config", + ) + + def deployment_identity_labels( workspace: str, name: str, diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/entities.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/entities.py index c3b42d3fae..1c71cf6f74 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/entities.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/entities.py @@ -93,6 +93,11 @@ class Container(BaseModel): volume_mounts: list[VolumeMount] = Field(default_factory=list, alias="volumeMounts") liveness_probe: Probe | None = Field(default=None, alias="livenessProbe") readiness_probe: Probe | None = Field(default=None, alias="readinessProbe") + restart_policy: RestartPolicy | None = Field( + default=None, + alias="restartPolicy", + description="Per-container restart policy for init containers; Always enables k8s native sidecar.", + ) model_config = {"populate_by_name": True} @@ -263,11 +268,15 @@ class Deployment(NemoEntity, entity_type=ENTITY_TYPE_DEPLOYMENT): # Never → SUCCEEDED terminal; Always/OnFailure → READY while running. +def _default_volume_access_modes() -> list[AccessMode]: + return ["ReadWriteOnce"] + + class Volume(NemoEntity, entity_type=ENTITY_TYPE_VOLUME): """Persistent volume request and observed state.""" size: str = Field(default="1Gi", description="Requested storage size (Kubernetes quantity).") - access_modes: list[AccessMode] = Field(default_factory=lambda: ["ReadWriteOnce"]) + access_modes: list[AccessMode] = Field(default_factory=_default_volume_access_modes) backend_config: VolumeBackendConfig = Field(default_factory=VolumeBackendConfig, alias="backendConfig") status: VolumeStatus = Field(default="PENDING") status_message: str = Field(default="") diff --git a/plugins/nemo-deployments/tests/unit/backends/k8s/test_compiler.py b/plugins/nemo-deployments/tests/unit/backends/k8s/test_compiler.py new file mode 100644 index 0000000000..a58d8efc5b --- /dev/null +++ b/plugins/nemo-deployments/tests/unit/backends/k8s/test_compiler.py @@ -0,0 +1,258 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest +from backends.k8s.k8s_helpers import sample_always_config, sample_config +from kubernetes.client import ApiClient +from nemo_deployments_plugin.backends.k8s.compiler import ( + DeploymentConfigError, + build_configmap_body, + compile_workload, + configmap_data_key, + validate_config_for_deployment, + validate_config_for_job, +) +from nemo_deployments_plugin.backends.k8s.deployments import build_deployment_body +from nemo_deployments_plugin.backends.k8s.jobs import build_job_body +from nemo_deployments_plugin.entities import ( + ConfigFile, + Container, + ContainerPort, + K8sDeploymentConfig, +) + + +def _serialized(obj: object) -> dict: + return ApiClient().sanitize_for_serialization(obj) + + +def test_configmap_data_key_sanitizes_paths() -> None: + assert configmap_data_key("/etc/app/config.yaml") == "etc__app__config.yaml" + + +def test_compile_job_pod_spec_single_container() -> None: + config = sample_config(restart_policy="Never") + compiled = compile_workload( + config=config, + workspace="default", + deployment_name="task", + labels={"managed-by": "nemo-deployments"}, + k8s_config=None, + pod_restart_policy="Never", + ) + pod_spec = _serialized(compiled.pod_spec_kwargs) + assert pod_spec["restart_policy"] == "Never" + assert len(pod_spec["containers"]) == 1 + assert pod_spec["containers"][0]["name"] == "main" + assert compiled.configmap_body is None + + +def test_compile_deployment_includes_init_and_sidecar() -> None: + config = sample_always_config().model_copy( + update={ + "init_containers": [ + Container(name="bootstrap", image="busybox", command=["sh", "-c", "echo hi"]), + Container.model_validate( + { + "name": "sidecar", + "image": "nginx:alpine", + "restartPolicy": "Always", + "ports": [{"name": "proxy", "containerPort": 8081}], + "livenessProbe": {"httpGet": {"path": "/healthz", "port": 8081}}, + } + ), + ], + "containers": [ + Container(name="main", image="nginx:alpine", ports=[ContainerPort(name="http", containerPort=8080)]), + Container( + name="metrics", + image="prom/node-exporter", + ports=[ContainerPort(name="metrics", containerPort=9100)], + ), + ], + } + ) + compiled = compile_workload( + config=config, + workspace="default", + deployment_name="task", + labels={"managed-by": "nemo-deployments"}, + k8s_config=None, + pod_restart_policy="Always", + ) + pod_spec = _serialized(compiled.pod_spec_kwargs) + assert pod_spec == { + "restart_policy": "Always", + "init_containers": [ + { + "name": "bootstrap", + "image": "busybox", + "command": ["sh", "-c", "echo hi"], + }, + { + "name": "sidecar", + "image": "nginx:alpine", + "restartPolicy": "Always", + "ports": [{"name": "proxy", "containerPort": 8081, "protocol": "TCP"}], + "livenessProbe": { + "httpGet": {"path": "/healthz", "port": 8081, "scheme": "HTTP"}, + "initialDelaySeconds": 0, + "timeoutSeconds": 1, + "periodSeconds": 10, + "failureThreshold": 3, + }, + }, + ], + "containers": [ + { + "name": "main", + "image": "nginx:alpine", + "ports": [{"name": "http", "containerPort": 8080, "protocol": "TCP"}], + }, + { + "name": "metrics", + "image": "prom/node-exporter", + "ports": [{"name": "metrics", "containerPort": 9100, "protocol": "TCP"}], + }, + ], + } + assert len(compiled.service_containers) == 2 + + +def test_compile_applies_k8s_deployment_config() -> None: + config = sample_always_config() + k8s_config = K8sDeploymentConfig.model_validate( + { + "serviceAccount": "deploy-sa", + "tolerations": [{"key": "gpu", "operator": "Equal", "value": "true", "effect": "NoSchedule"}], + "affinity": {"nodeAffinity": {"requiredDuringSchedulingIgnoredDuringExecution": {"nodeSelectorTerms": []}}}, + "securityContext": {"runAsUser": 1000, "fsGroup": 2000}, + } + ) + compiled = compile_workload( + config=config, + workspace="default", + deployment_name="task", + labels={"managed-by": "nemo-deployments"}, + k8s_config=k8s_config, + pod_restart_policy="Always", + ) + pod_spec = _serialized(compiled.pod_spec_kwargs) + assert pod_spec["service_account_name"] == "deploy-sa" + assert pod_spec["tolerations"][0]["key"] == "gpu" + affinity = compiled.pod_spec_kwargs["affinity"] + assert affinity.node_affinity is not None + security_context = compiled.pod_spec_kwargs["security_context"] + assert security_context.run_as_user == 1000 + + +def test_compile_config_files_emit_configmap_and_mounts() -> None: + config = sample_always_config().model_copy( + update={"config_files": [ConfigFile(path="/etc/app/config.yaml", content="key: value")]} + ) + labels = {"managed-by": "nemo-deployments"} + compiled = compile_workload( + config=config, + workspace="default", + deployment_name="task", + labels=labels, + k8s_config=None, + pod_restart_policy="Always", + ) + assert compiled.configmap_body is not None + configmap = _serialized(compiled.configmap_body) + assert configmap["data"]["etc__app__config.yaml"] == "key: value" + pod_spec = _serialized(compiled.pod_spec_kwargs) + assert any(volume["name"] == "config-files" for volume in pod_spec["volumes"]) + main_container = compiled.pod_spec_kwargs["containers"][0] + mount_paths = [mount.mount_path for mount in main_container.volume_mounts or []] + assert "/etc/app/config.yaml" in mount_paths + + +def test_build_job_body_returns_compiled_workload() -> None: + config = sample_config(restart_policy="OnFailure") + built = build_job_body( + job_name="dep-default-task-abc", + labels={"managed-by": "nemo-deployments"}, + config=config, + workspace="default", + deployment_name="task", + k8s_config=None, + ) + assert built.job.kind == "Job" + assert built.compiled.pod_spec_kwargs["restart_policy"] == "OnFailure" + + +def test_build_deployment_body_returns_compiled_workload() -> None: + config = sample_always_config() + built = build_deployment_body( + resource_name="dep-default-task-abc", + labels={"managed-by": "nemo-deployments"}, + config=config, + workspace="default", + deployment_name="task", + k8s_config=None, + ) + assert built.deployment.kind == "Deployment" + assert built.compiled.service_containers[0].name == "main" + + +def test_validate_rejects_main_container_restart_policy() -> None: + config = sample_always_config().model_copy( + update={"containers": [Container.model_validate({"name": "main", "image": "nginx", "restartPolicy": "Always"})]} + ) + with pytest.raises(DeploymentConfigError, match="only init_containers"): + validate_config_for_deployment(config) + + +def test_validate_job_rejects_always() -> None: + with pytest.raises(DeploymentConfigError, match="Deployment"): + validate_config_for_job(sample_always_config()) + + +def test_validate_rejects_duplicate_port_names() -> None: + config = sample_always_config().model_copy( + update={ + "containers": [ + Container( + name="main", + image="nginx", + ports=[ContainerPort(name="http", containerPort=8080)], + ), + Container( + name="side", + image="nginx", + ports=[ContainerPort(name="http", containerPort=9090)], + ), + ], + } + ) + with pytest.raises(DeploymentConfigError, match="duplicate container port name"): + validate_config_for_deployment(config) + + +def test_validate_rejects_duplicate_listen_ports() -> None: + config = sample_always_config().model_copy( + update={ + "containers": [ + Container( + name="main", + image="nginx", + ports=[ContainerPort(name="http", containerPort=8080)], + ), + Container( + name="side", + image="nginx", + ports=[ContainerPort(name="alt", containerPort=8080)], + ), + ], + } + ) + with pytest.raises(DeploymentConfigError, match="duplicate container port 8080"): + validate_config_for_deployment(config) + + +def test_build_configmap_body_none_when_empty() -> None: + assert build_configmap_body(workspace="default", deployment_name="task", labels={}, config_files=[]) is None diff --git a/plugins/nemo-deployments/tests/unit/backends/k8s/test_deployments.py b/plugins/nemo-deployments/tests/unit/backends/k8s/test_deployments.py index 82c83c4d1c..03b24e39f6 100644 --- a/plugins/nemo-deployments/tests/unit/backends/k8s/test_deployments.py +++ b/plugins/nemo-deployments/tests/unit/backends/k8s/test_deployments.py @@ -3,6 +3,7 @@ from __future__ import annotations +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest @@ -17,12 +18,16 @@ from kubernetes.client.rest import ApiException from nemo_deployments_plugin.backends.k8s import deployments as deployment_ops from nemo_deployments_plugin.backends.k8s.client import KubernetesClients +from nemo_deployments_plugin.backends.k8s.compiler import validate_config_for_deployment from nemo_deployments_plugin.backends.k8s.deployments import ( build_in_cluster_endpoints, - validate_config_for_deployment, ) -from nemo_deployments_plugin.backends.labels import MANAGED_BY_KEY, k8s_deployment_resource_name -from nemo_deployments_plugin.entities import Container, ContainerPort +from nemo_deployments_plugin.backends.labels import ( + MANAGED_BY_KEY, + k8s_deployment_configmap_name, + k8s_deployment_resource_name, +) +from nemo_deployments_plugin.entities import ConfigFile, Container, ContainerPort from nemo_platform_plugin.entity_client import NemoEntityNotFoundError @@ -41,7 +46,9 @@ def deployment_ops_clients(mock_k8s_clients: MagicMock) -> MagicMock: def test_validate_config_for_deployment_rejects_never() -> None: - with pytest.raises(deployment_ops.DeploymentConfigError, match="Always"): + from nemo_deployments_plugin.backends.k8s.compiler import DeploymentConfigError + + with pytest.raises(DeploymentConfigError, match="Always"): validate_config_for_deployment(sample_always_config().model_copy(update={"restart_policy": "Never"})) @@ -50,7 +57,7 @@ def test_build_in_cluster_endpoints_uses_cluster_dns() -> None: endpoints = build_in_cluster_endpoints( resource_name=resource_name, namespace="nemo-deployments", - container=sample_always_config().containers[0], + containers=tuple(sample_always_config().containers), ) assert endpoints[0].url == f"http://{resource_name}.nemo-deployments.svc.cluster.local:8080" @@ -65,7 +72,7 @@ def test_build_in_cluster_endpoints_uses_tcp_scheme_for_udp() -> None: endpoints = build_in_cluster_endpoints( resource_name=resource_name, namespace="nemo-deployments", - container=container, + containers=(container,), ) assert endpoints[0].protocol == "tcp" assert endpoints[0].url == f"tcp://{resource_name}.nemo-deployments.svc.cluster.local:9090" @@ -195,14 +202,80 @@ async def test_create_deployment_rolls_back_when_service_create_fails( @pytest.mark.asyncio -async def test_read_status_rejects_unsupported_always_config(k8s_backend, mock_entities: AsyncMock) -> None: +async def test_create_deployment_rolls_back_configmap_when_service_create_fails( + deployment_ops_clients: MagicMock, mock_k8s_clients: MagicMock +) -> None: + config = sample_always_config().model_copy( + update={"config_files": [ConfigFile(path="/etc/app/config.yaml", content="key: value")]} + ) + mock_k8s_clients.apps_v1.create_namespaced_deployment.return_value = mock_deployment() + mock_k8s_clients.apps_v1.read_namespaced_deployment.return_value = mock_deployment() + mock_k8s_clients.core_v1.create_namespaced_service.side_effect = ApiException(status=500) + identity_labels = always_identity_labels(backoff_limit=config.backoff_limit) + mock_k8s_clients.core_v1.read_namespaced_config_map.return_value = SimpleNamespace( + metadata=SimpleNamespace(labels=identity_labels), + ) + + update = await deployment_ops.create_deployment( + deployment_ops_clients, + default_namespace="default", + workspace="default", + name="task", + config_name="config1", + labels={}, + backend_config={}, + config=config, + ) + + assert update.status == "FAILED" + mock_k8s_clients.core_v1.create_namespaced_config_map.assert_called_once() + mock_k8s_clients.core_v1.read_namespaced_config_map.assert_called_once_with( + name=k8s_deployment_configmap_name("default", "task"), + namespace="default", + _request_timeout=mock_k8s_clients.request_timeout, + ) + mock_k8s_clients.core_v1.delete_namespaced_config_map.assert_called_once() + + +@pytest.mark.asyncio +async def test_create_deployment_adopted_service_failure_keeps_configmap( + deployment_ops_clients: MagicMock, mock_k8s_clients: MagicMock +) -> None: + config = sample_always_config().model_copy( + update={"config_files": [ConfigFile(path="/etc/app/config.yaml", content="key: value")]} + ) + mock_k8s_clients.apps_v1.create_namespaced_deployment.side_effect = ApiException(status=409) + mock_k8s_clients.apps_v1.read_namespaced_deployment.return_value = mock_deployment() + mock_k8s_clients.core_v1.create_namespaced_service.side_effect = ApiException(status=500) + + update = await deployment_ops.create_deployment( + deployment_ops_clients, + default_namespace="default", + workspace="default", + name="task", + config_name="config1", + labels={}, + backend_config={}, + config=config, + ) + + assert update.status == "FAILED" + mock_k8s_clients.apps_v1.delete_namespaced_deployment.assert_not_called() + mock_k8s_clients.core_v1.delete_namespaced_config_map.assert_not_called() + + +@pytest.mark.asyncio +async def test_read_status_accepts_init_containers( + k8s_backend, mock_k8s_clients: MagicMock, mock_entities: AsyncMock +) -> None: config = sample_always_config().model_copy(update={"init_containers": [Container(name="init", image="busybox")]}) mock_entities.get.side_effect = [sample_deployment(), config] + mock_k8s_clients.apps_v1.read_namespaced_deployment.return_value = mock_deployment() + mock_k8s_clients.core_v1.list_namespaced_pod.return_value = MagicMock(items=[]) update = await k8s_backend.read_status(workspace="default", name="task") - assert update.status == "FAILED" - assert "init_containers" in update.status_message + assert update.status == "STARTING" @pytest.mark.asyncio @@ -267,6 +340,9 @@ async def test_delete_deployment_rejects_foreign( foreign = mock_deployment() foreign.metadata.labels = {MANAGED_BY_KEY: "other-plugin"} mock_k8s_clients.apps_v1.read_namespaced_deployment.return_value = foreign + mock_k8s_clients.core_v1.read_namespaced_config_map.return_value = SimpleNamespace( + metadata=SimpleNamespace(labels=always_identity_labels()), + ) update = await deployment_ops.delete_deployment( deployment_ops_clients, @@ -279,6 +355,11 @@ async def test_delete_deployment_rejects_foreign( assert update.status == "FAILED" mock_k8s_clients.apps_v1.delete_namespaced_deployment.assert_not_called() + mock_k8s_clients.core_v1.read_namespaced_config_map.assert_called_once_with( + name=k8s_deployment_configmap_name("default", "task"), + namespace="default", + _request_timeout=mock_k8s_clients.request_timeout, + ) @pytest.mark.asyncio diff --git a/plugins/nemo-deployments/tests/unit/backends/k8s/test_k8s_status_mapping.py b/plugins/nemo-deployments/tests/unit/backends/k8s/test_k8s_status_mapping.py index d3d58e82fe..f0939dd09e 100644 --- a/plugins/nemo-deployments/tests/unit/backends/k8s/test_k8s_status_mapping.py +++ b/plugins/nemo-deployments/tests/unit/backends/k8s/test_k8s_status_mapping.py @@ -72,7 +72,7 @@ def test_status_from_deployment(ready_replicas, deleting, pod, expected_status, endpoints = build_in_cluster_endpoints( resource_name=resource_name, namespace="default", - container=sample_always_config().containers[0], + containers=tuple(sample_always_config().containers), ) update = status_from_deployment( deployment=deployment,