From 612c526c90b676eeb0ec09bc6d3baef80a6aa7dd Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Thu, 16 Jul 2026 14:10:41 -0600 Subject: [PATCH 1/3] test(e2e): run agent container-deploy e2e on kubernetes (kind) too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docker-mode agent deployment e2e (test_nemo_agents_docker.py) proved the container-mode chain (gateway -> agent container -> IGW -> mock provider) only for the Docker backend. Add the Kubernetes equivalent so the same chain is exercised on the kind CPU e2e CI job. - Extract the backend-agnostic core (mock-backed agent config, deploy -> wait running -> assert container-mode endpoint shape -> invoke -> assert -> cleanup) into e2e/agents_deploy_helpers.py, shared by both modules. A reap_backend_resources hook lets a backend reap leaked resources (docker container) during teardown. - Refactor test_nemo_agents_docker.py to call the shared core; behavior and markers (subprocess_only, needs_nmp_api_image, Linux-only) are unchanged. - Add test_nemo_agents_k8s.py: deployment_mode=k8s, marked container_only (runs only against an external cluster, i.e. the kind e2e job) + needs_nmp_api_image. A single dual-marked function isn't viable — the markers are mutually exclusive and the platform source / networking differ — so it's a thin wrapper over the shared core. - Wire a nemo-deployments k8s executor into the kind Helm values (e2e/k8s/values/kind.yaml); the nmp-core controller Role already grants the needed deployments/services/configmaps/pods permissions. default_namespace is the release namespace so the agent lands beside the platform and the injected in-cluster IGW URL is reachable. - Export NMP_E2E_IMAGE_REGISTRY/NMP_E2E_IMAGE_TAG in the kind-cpu-e2e job so the k8s test can compose the (node-pre-pulled) nmp-api image ref. Signed-off-by: Ben McCown --- .github/workflows/ci.yaml | 7 + e2e/agents_deploy_helpers.py | 255 +++++++++++++++++++++++++++++++++ e2e/k8s/values/kind.yaml | 19 +++ e2e/test_nemo_agents_docker.py | 200 ++------------------------ e2e/test_nemo_agents_k8s.py | 102 +++++++++++++ 5 files changed, 392 insertions(+), 191 deletions(-) create mode 100644 e2e/agents_deploy_helpers.py create mode 100644 e2e/test_nemo_agents_k8s.py diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 277e2d1141..03fd538bcc 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -525,6 +525,13 @@ jobs: _TYPER_FORCE_DISABLE_TERMINAL: "1" E2E_SERVICES_LOG_DIR: ${{ runner.temp }}/e2e-services-logs NGC_API_KEY: not-used-for-ghcr-cpu-e2e + # Let the k8s-mode agent deployment e2e (test_nemo_agents_k8s.py) learn + # the agent image ref. The Helm platform is configured with a k8s + # deployments executor (e2e/k8s/values/kind.yaml) and this same + # nmp-api image is pre-pulled into the kind nodes, so the agent pod + # resolves it node-locally under IfNotPresent. + NMP_E2E_IMAGE_REGISTRY: ${{ needs.build-cpu-smoke-images.outputs.image_registry }} + NMP_E2E_IMAGE_TAG: ${{ needs.build-cpu-smoke-images.outputs.image_tag }} run: | test -n "${NMP_E2E_CLUSTER_URL}" export NMP_BASE_URL="${NMP_E2E_CLUSTER_URL}" diff --git a/e2e/agents_deploy_helpers.py b/e2e/agents_deploy_helpers.py new file mode 100644 index 0000000000..6244cd9e47 --- /dev/null +++ b/e2e/agents_deploy_helpers.py @@ -0,0 +1,255 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared helpers for container-mode agent deployment e2e tests. + +Both the Docker (``test_nemo_agents_docker.py``) and Kubernetes +(``test_nemo_agents_k8s.py``) modules deploy a real agent **container** through +the nemo-deployments plugin and invoke it through the agents gateway. The end-to +-end chain they prove is identical apart from the backend:: + + sdk.agents.invoke (gateway proxy, container-mode endpoint resolution) + -> agent container (nat start fastapi) on docker | kubernetes + -> Inference Gateway /openai (base_url injected at deploy time) + -> mock provider short-circuit (no real upstream / no API key) + -> response back through the gateway + +This module holds the backend-agnostic core: the mock-provider-backed agent +config, the create -> wait-running -> assert-container-shape -> invoke -> assert +flow, and cleanup. The per-backend modules own only what genuinely differs +(pytest markers, how the deployment image ref is resolved, and best-effort +container cleanup). +""" + +import time +import uuid +from collections.abc import Callable +from typing import Any + +import httpx +import pytest +from nemo_platform import NeMoPlatform +from nmp.testing import MockProviderResponse, add_mock_provider + +# The mocked completion the deployed agent must round-trip back to the caller. +TEST_AGENT_RESPONSE = "The answer to your question is 42." + + +def unique_name(prefix: str) -> str: + return f"e2e-{prefix}-{uuid.uuid4().hex[:8]}" + + +def _chat_completion_response(content: str, model: str) -> dict[str, Any]: + return { + "id": "chatcmpl-agents-container-e2e", + "object": "chat.completion", + "model": model, + "choices": [ + { + "message": {"role": "assistant", "content": content}, + "finish_reason": "stop", + "index": 0, + } + ], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + } + + +def _mock_backed_agent_config(model_name: str) -> dict[str, Any]: + """A deterministic single-LLM workflow pointed at the mock model. + + ``base_url``/``api_key`` are intentionally omitted so the deployment injects + the Inference Gateway URL (and the mock provider short-circuits the call). + """ + return { + "llms": { + "main": { + "_type": "openai", + "model_name": model_name, + } + }, + "workflow": { + "_type": "chat_completion", + "llm_name": "main", + }, + } + + +def _page_data(page: Any) -> list[dict[str, Any]]: + if isinstance(page, dict): + data = page.get("data", []) + else: + data = getattr(page, "data", []) + assert isinstance(data, list) + return data + + +def delete_agent_if_exists(sdk: NeMoPlatform, *, workspace: str, name: str) -> None: + try: + sdk.agents.delete(name, workspace=workspace) + except httpx.HTTPStatusError as exc: + if exc.response.status_code != 404: + raise + + +def delete_deployment_if_exists(sdk: NeMoPlatform, *, workspace: str, name: str) -> None: + try: + sdk.agents.deployments.delete(name, workspace=workspace) + except httpx.HTTPStatusError as exc: + if exc.response.status_code != 404: + raise + + +def get_deployment_log_text(sdk: NeMoPlatform, *, workspace: str, name: str) -> str: + try: + response = sdk._client.get( + f"/apis/agents/v2/workspaces/{workspace}/deployments/{name}/logs", + params={"tail": 100}, + ) + response.raise_for_status() + except httpx.HTTPStatusError as exc: + return exc.response.text + + payload = response.json() + lines = _page_data(payload) + return "\n".join(str(line.get("message", line)) for line in lines) + + +def wait_for_deployment_deleted( + sdk: NeMoPlatform, + *, + workspace: str, + name: str, + timeout_seconds: float = 120, +) -> None: + deadline = time.monotonic() + timeout_seconds + last_status: str | None = None + while time.monotonic() < deadline: + try: + deployment = sdk.agents.deployments.get(name, workspace=workspace) + last_status = deployment.get("status") + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + return + raise + time.sleep(2) + pytest.fail(f"Deployment {name!r} was not deleted within {timeout_seconds}s; last status={last_status!r}") + + +def wait_for_deployment_running( + sdk: NeMoPlatform, + *, + workspace: str, + name: str, + timeout_seconds: float = 300, +) -> dict[str, Any]: + deadline = time.monotonic() + timeout_seconds + last_deployment: dict[str, Any] | None = None + while time.monotonic() < deadline: + deployment = sdk.agents.deployments.get(name, workspace=workspace) + last_deployment = deployment + status = deployment["status"] + if status == "running": + return deployment + if status == "failed": + logs = get_deployment_log_text(sdk, workspace=workspace, name=name) + pytest.fail(f"Deployment {name!r} failed: {deployment.get('error', '')}\n{logs}") + time.sleep(2) + pytest.fail(f"Deployment {name!r} did not reach running within {timeout_seconds}s: {last_deployment}") + + +def run_container_agent_deploy_and_invoke( + sdk: NeMoPlatform, + *, + workspace: str, + deployment_mode: str, + image: str, + running_timeout_seconds: float = 300, + reap_backend_resources: Callable[[str], None] | None = None, +) -> None: + """Deploy a mock-backed agent as a container and invoke it through the gateway. + + Backend-agnostic core shared by the docker and k8s e2e modules: + + 1. Register a mock inference provider + deterministic single-LLM agent. + 2. Deploy it with ``deployment_mode`` (``"docker"`` / ``"k8s"``) from ``image``. + 3. Wait for ``running`` and assert the container-mode endpoint shape (empty + scalar ``endpoint``, populated ``endpoints``) — this guards that the pass + came through the container path, not a subprocess fallback. + 4. Invoke through the gateway and assert the mocked completion round-trips. + 5. Clean up the deployment and agent (best-effort, isolated steps). + + ``reap_backend_resources``, if given, is called with the deployment name + during teardown (after the deployment is deleted) so a backend module can + best-effort reap leaked resources it uniquely knows about — e.g. a leftover + docker container. It must not raise; failures are swallowed like the rest of + teardown. + """ + agent_name = unique_name("calc-agent") + deployment_name = unique_name("calc-deployment") + model_name = unique_name("calc-model") + + add_mock_provider( + sdk, + workspace=workspace, + name=unique_name("calc-provider"), + mock_response_body_by_model={ + f"{workspace}/{model_name}": [ + MockProviderResponse(response_body=_chat_completion_response(TEST_AGENT_RESPONSE, model_name)), + ], + }, + served_models={model_name: model_name}, + ) + + sdk.agents.create( + workspace=workspace, + name=agent_name, + config=_mock_backed_agent_config(f"{workspace}/{model_name}"), + ) + + try: + created = sdk.agents.deployments.create( + workspace=workspace, + agent=agent_name, + name=deployment_name, + deployment_mode=deployment_mode, + image=image, + ) + assert created["deployment_mode"] == deployment_mode + + deployment = wait_for_deployment_running( + sdk, workspace=workspace, name=deployment_name, timeout_seconds=running_timeout_seconds + ) + assert deployment["agent"] == agent_name + assert deployment["deployment_mode"] == deployment_mode + + # Container-mode addressing: the loopback scalar ``endpoint`` is empty and + # the routable address lives in ``endpoints`` (this is what the gateway's + # container-mode resolution reads). Guarding this shape ensures a pass can + # only come through the container path, not a subprocess fallback. + assert deployment.get("endpoint", "") == "" + endpoints = deployment.get("endpoints") or [] + assert endpoints and endpoints[0]["url"], deployment + + response = sdk.agents.invoke( + workspace=workspace, + agent=agent_name, + input="What is 12 multiplied by 8?", + ) + content = response["choices"][0]["message"]["content"] + assert TEST_AGENT_RESPONSE in content, response + finally: + # Each step is isolated so a failure (e.g. a deployment-delete timeout) + # doesn't skip the remaining cleanup and leak resources. + _safe(delete_deployment_if_exists, sdk, workspace=workspace, name=deployment_name) + _safe(wait_for_deployment_deleted, sdk, workspace=workspace, name=deployment_name) + if reap_backend_resources is not None: + _safe(reap_backend_resources, deployment_name) + _safe(delete_agent_if_exists, sdk, workspace=workspace, name=agent_name) + + +def _safe(fn: Any, *args: Any, **kwargs: Any) -> None: + try: + fn(*args, **kwargs) + except Exception: + pass diff --git a/e2e/k8s/values/kind.yaml b/e2e/k8s/values/kind.yaml index 491c90dbb0..bdc241d935 100644 --- a/e2e/k8s/values/kind.yaml +++ b/e2e/k8s/values/kind.yaml @@ -33,3 +33,22 @@ platformConfig: backends: nim_operator: enabled: false + # Wire a nemo-deployments k8s executor so deployment_mode=k8s agent + # deployments (test_nemo_agents_k8s.py) run as real Deployment+Service + # workloads in the release namespace. The nmp-core controller ServiceAccount + # Role already grants apps/deployments, services, configmaps, and pods (see + # k8s/helm/templates/core/controller-role.yaml). default_namespace matches the + # Helm release namespace so the agent lands beside the platform and the + # in-cluster Inference Gateway URL injected into the pod is reachable. + agents: + deployments: + k8s_executor: kind-k8s + container_port: 8000 + deployments: + default_executor: kind-k8s + executors: + - name: kind-k8s + backend: k8s + config: + # Omit kubeconfig_path -> in-cluster ServiceAccount auth. + default_namespace: nemo-platform diff --git a/e2e/test_nemo_agents_docker.py b/e2e/test_nemo_agents_docker.py index 66ed24e6d7..4f60bea480 100644 --- a/e2e/test_nemo_agents_docker.py +++ b/e2e/test_nemo_agents_docker.py @@ -3,7 +3,9 @@ Unlike ``test_nemo_agents.py`` (backend-agnostic API/SDK surface, plus subprocess-mode deployment coverage), this module deploys an agent as a real **Docker container** through the nemo-deployments plugin and invokes it through -the agents gateway. +the agents gateway. The backend-agnostic deploy/invoke/assert core is shared +with the Kubernetes variant (``test_nemo_agents_k8s.py``) via +``e2e.agents_deploy_helpers``; this module owns only the docker-specific wiring. What it proves — the container-mode chain end to end:: @@ -43,15 +45,11 @@ import os import platform -import time -import uuid -from contextlib import suppress -from typing import Any -import httpx import pytest from nemo_platform import NeMoPlatform -from nmp.testing import MockProviderResponse, add_mock_provider + +from e2e.agents_deploy_helpers import run_container_agent_deploy_and_invoke # The docker bridge gateway. On Linux (incl. GitHub Actions ubuntu runners) this # address is reachable both from inside a container AND by the platform process @@ -92,131 +90,6 @@ ), ] -_TEST_AGENT_RESPONSE = "The answer to your question is 42." - - -def _unique_name(prefix: str) -> str: - return f"e2e-{prefix}-{uuid.uuid4().hex[:8]}" - - -def _chat_completion_response(content: str, model: str) -> dict[str, Any]: - return { - "id": "chatcmpl-agents-docker-e2e", - "object": "chat.completion", - "model": model, - "choices": [ - { - "message": {"role": "assistant", "content": content}, - "finish_reason": "stop", - "index": 0, - } - ], - "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, - } - - -def _mock_backed_agent_config(model_name: str) -> dict[str, Any]: - """A deterministic single-LLM workflow pointed at the mock model. - - ``base_url``/``api_key`` are intentionally omitted so the deployment injects - the Inference Gateway URL (and the mock provider short-circuits the call). - """ - return { - "llms": { - "main": { - "_type": "openai", - "model_name": model_name, - } - }, - "workflow": { - "_type": "chat_completion", - "llm_name": "main", - }, - } - - -def _page_data(page: Any) -> list[dict[str, Any]]: - if isinstance(page, dict): - data = page.get("data", []) - else: - data = getattr(page, "data", []) - assert isinstance(data, list) - return data - - -def _delete_agent_if_exists(sdk: NeMoPlatform, *, workspace: str, name: str) -> None: - try: - sdk.agents.delete(name, workspace=workspace) - except httpx.HTTPStatusError as exc: - if exc.response.status_code != 404: - raise - - -def _delete_deployment_if_exists(sdk: NeMoPlatform, *, workspace: str, name: str) -> None: - try: - sdk.agents.deployments.delete(name, workspace=workspace) - except httpx.HTTPStatusError as exc: - if exc.response.status_code != 404: - raise - - -def _get_deployment_log_text(sdk: NeMoPlatform, *, workspace: str, name: str) -> str: - try: - response = sdk._client.get( - f"/apis/agents/v2/workspaces/{workspace}/deployments/{name}/logs", - params={"tail": 100}, - ) - response.raise_for_status() - except httpx.HTTPStatusError as exc: - return exc.response.text - - payload = response.json() - lines = _page_data(payload) - return "\n".join(str(line.get("message", line)) for line in lines) - - -def _wait_for_deployment_deleted( - sdk: NeMoPlatform, - *, - workspace: str, - name: str, - timeout_seconds: float = 120, -) -> None: - deadline = time.monotonic() + timeout_seconds - last_status: str | None = None - while time.monotonic() < deadline: - try: - deployment = sdk.agents.deployments.get(name, workspace=workspace) - last_status = deployment.get("status") - except httpx.HTTPStatusError as exc: - if exc.response.status_code == 404: - return - raise - time.sleep(2) - pytest.fail(f"Deployment {name!r} was not deleted within {timeout_seconds}s; last status={last_status!r}") - - -def _wait_for_deployment_running( - sdk: NeMoPlatform, - *, - workspace: str, - name: str, - timeout_seconds: float = 300, -) -> dict[str, Any]: - deadline = time.monotonic() + timeout_seconds - last_deployment: dict[str, Any] | None = None - while time.monotonic() < deadline: - deployment = sdk.agents.deployments.get(name, workspace=workspace) - last_deployment = deployment - status = deployment["status"] - if status == "running": - return deployment - if status == "failed": - logs = _get_deployment_log_text(sdk, workspace=workspace, name=name) - pytest.fail(f"Deployment {name!r} failed: {deployment.get('error', '')}\n{logs}") - time.sleep(2) - pytest.fail(f"Deployment {name!r} did not reach running within {timeout_seconds}s: {last_deployment}") - @pytest.fixture(scope="session") def agent_deployment_image() -> str: @@ -268,65 +141,10 @@ def test_docker_agent_deploys_and_invokes_through_gateway( through the gateway and asserts the mocked completion round-trips from inside the container back to the caller. """ - agent_name = _unique_name("calc-agent") - deployment_name = _unique_name("calc-deployment") - model_name = _unique_name("calc-model") - - add_mock_provider( + run_container_agent_deploy_and_invoke( sdk, workspace=workspace, - name=_unique_name("calc-provider"), - mock_response_body_by_model={ - f"{workspace}/{model_name}": [ - MockProviderResponse(response_body=_chat_completion_response(_TEST_AGENT_RESPONSE, model_name)), - ], - }, - served_models={model_name: model_name}, + deployment_mode="docker", + image=agent_deployment_image, + reap_backend_resources=_remove_agent_container_if_present, ) - - sdk.agents.create( - workspace=workspace, - name=agent_name, - config=_mock_backed_agent_config(f"{workspace}/{model_name}"), - ) - - try: - created = sdk.agents.deployments.create( - workspace=workspace, - agent=agent_name, - name=deployment_name, - deployment_mode="docker", - image=agent_deployment_image, - ) - assert created["deployment_mode"] == "docker" - - deployment = _wait_for_deployment_running(sdk, workspace=workspace, name=deployment_name) - assert deployment["agent"] == agent_name - assert deployment["deployment_mode"] == "docker" - - # Container-mode addressing: the loopback scalar ``endpoint`` is empty and - # the routable address lives in ``endpoints`` (this is what the gateway's - # container-mode resolution reads). Guarding this shape ensures a pass can - # only come through the container path, not a subprocess fallback. - assert deployment.get("endpoint", "") == "" - endpoints = deployment.get("endpoints") or [] - assert endpoints and endpoints[0]["url"], deployment - - response = sdk.agents.invoke( - workspace=workspace, - agent=agent_name, - input="What is 12 multiplied by 8?", - ) - content = response["choices"][0]["message"]["content"] - assert _TEST_AGENT_RESPONSE in content, response - finally: - # Each step is isolated so a failure (e.g. a deployment-delete timeout) - # doesn't skip the remaining cleanup and leak resources. - with suppress(Exception): - _delete_deployment_if_exists(sdk, workspace=workspace, name=deployment_name) - with suppress(Exception): - _wait_for_deployment_deleted(sdk, workspace=workspace, name=deployment_name) - with suppress(Exception): - _remove_agent_container_if_present(deployment_name) - with suppress(Exception): - _delete_agent_if_exists(sdk, workspace=workspace, name=agent_name) diff --git a/e2e/test_nemo_agents_k8s.py b/e2e/test_nemo_agents_k8s.py new file mode 100644 index 0000000000..9fc0fa012f --- /dev/null +++ b/e2e/test_nemo_agents_k8s.py @@ -0,0 +1,102 @@ +"""E2E test for Kubernetes-mode agent deployments. + +The Kubernetes counterpart to ``test_nemo_agents_docker.py``: it deploys an +agent as a real **Kubernetes Deployment + Service** through the nemo-deployments +plugin and invokes it through the agents gateway. The backend-agnostic +deploy/invoke/assert core is shared via ``e2e.agents_deploy_helpers``; this +module owns only the k8s-specific wiring. + +What it proves — the container-mode chain end to end, on Kubernetes:: + + sdk.agents.invoke (gateway proxy, container-mode endpoint resolution) + -> k8s agent pod (nat start fastapi), fronted by a ClusterIP Service + -> Inference Gateway /openai (base_url injected at deploy time) + -> mock provider short-circuit (no real upstream / no API key) + -> response back through the gateway + +How it runs, and where: + +- This test only runs against an **external cluster** (``NMP_BASE_URL`` set) — + i.e. the Kind CPU e2e CI job, where a Helm-deployed platform is configured + with a nemo-deployments ``k8s`` executor (see ``e2e/k8s/values/kind.yaml``). + The ``container_only`` marker skips it for the subprocess harness (local / + plain e2e job), which is the inverse of the docker module's ``subprocess_only``. +- The agent runs from the platform's own ``nmp-api`` image, which already ships + the NAT runtime and agent components (see the docker module docstring). The + deployments k8s executor overrides the image entrypoint with ``nat start + fastapi``. In CI the image is pre-pulled into the kind nodes and referenced by + its commit-SHA tag, so the pod's default ``imagePullPolicy: IfNotPresent`` uses + the node-local image (the k8s backend does not use image pull secrets). +- The image ref is composed from ``NMP_E2E_IMAGE_REGISTRY`` / + ``NMP_E2E_IMAGE_TAG`` (the existing e2e image convention). The + ``needs_nmp_api_image`` marker skips the test unless both are set; the Kind + CPU e2e job exports them from the built image outputs. +- The agent is registered with a deterministic single-LLM ``chat_completion`` + workflow served by the e2e mock inference provider, so no ``NVIDIA_API_KEY`` + or model egress is needed; we assert the exact mocked completion round-trips. +- Gateway reachability is in-cluster: the agent Deployment/Service land in the + same namespace as the platform, and the Inference Gateway URL injected into + the agent pod is the platform's own in-cluster ``NMP_BASE_URL``, reachable pod + -> service. No docker-bridge / host-alias juggling is needed (that is a + docker-in-a-local-process concern only), so this module has no Linux/OS skip. +""" + +import os + +import pytest +from nemo_platform import NeMoPlatform + +from e2e.agents_deploy_helpers import run_container_agent_deploy_and_invoke + +# Platform image name to deploy the agent from (see module docstring). Registry +# and tag come from NMP_E2E_IMAGE_REGISTRY / NMP_E2E_IMAGE_TAG. +_AGENT_IMAGE_NAME = "nmp-api" + +# Markers: +# - ``container_only``: runs only against an external cluster (``NMP_BASE_URL`` +# set) — the Kind CPU e2e job, whose Helm platform is configured with a k8s +# deployments executor. Skipped on the subprocess harness (the inverse of the +# docker module's ``subprocess_only``), where no k8s executor exists. +# - ``needs_nmp_api_image``: skips unless NMP_E2E_IMAGE_REGISTRY + NMP_E2E_IMAGE_TAG +# are set, which is how the test learns the (node-pre-pulled) agent image ref. +pytestmark = [ + pytest.mark.container_only, + pytest.mark.needs_nmp_api_image, +] + + +@pytest.fixture(scope="session") +def agent_deployment_image() -> str: + """Return the prebuilt platform image ref to deploy the agent from. + + Composed from the e2e image convention (``NMP_E2E_IMAGE_REGISTRY`` / + ``NMP_E2E_IMAGE_TAG``) as ``{registry}/nmp-api:{tag}``. In the Kind e2e job + this exact ref is pre-pulled into the cluster nodes, so the pod resolves it + node-locally under ``IfNotPresent``. The ``needs_nmp_api_image`` marker + guarantees both env vars are set before this test runs; assert defensively. + """ + registry = os.environ.get("NMP_E2E_IMAGE_REGISTRY") + tag = os.environ.get("NMP_E2E_IMAGE_TAG") + assert registry and tag, "needs_nmp_api_image marker should have skipped when registry/tag are unset" + return f"{registry.rstrip('/')}/{_AGENT_IMAGE_NAME}:{tag}" + + +def test_k8s_agent_deploys_and_invokes_through_gateway( + sdk: NeMoPlatform, workspace: str, agent_deployment_image: str +) -> None: + """Deploy an agent as a k8s Deployment+Service from nmp-api and invoke it. + + Asserts the deployment reaches ``running`` with the container-mode endpoint + shape (empty scalar ``endpoint``, populated ``endpoints`` carrying the + in-cluster Service address), then invokes through the gateway and asserts the + mocked completion round-trips from inside the pod back to the caller. + """ + run_container_agent_deploy_and_invoke( + sdk, + workspace=workspace, + deployment_mode="k8s", + image=agent_deployment_image, + # Pod scheduling + (node-local) image resolution can take longer than the + # docker path's local container start. + running_timeout_seconds=420, + ) From 050911f198eec7beeb46ca45cba58d94d42312c6 Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Thu, 16 Jul 2026 16:02:38 -0600 Subject: [PATCH 2/3] fix(deployments): default k8s executor namespace to its own pod namespace MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The k8s deployments executor hardcoded default_namespace to "default" and had no way to self-locate, so the e2e kind values had to hardcode the release namespace (nemo-platform) — brittle and wrong if the namespace differs. Make default_namespace optional and resolve an effective namespace with precedence: explicit config value > POD_NAMESPACE (the controller pod's own namespace, injected via the downward API in the core controller deployment) > "default". A deployed platform can now omit the setting and agents land in its own namespace; an operator can still pin one, and per-deployment backend_config.k8s.namespace continues to override. Drop the hardcoded default_namespace from e2e/k8s/values/kind.yaml accordingly. Add unit tests covering the precedence (explicit wins, POD_NAMESPACE fallback, default fallback, blank POD_NAMESPACE treated as unset). Signed-off-by: Ben McCown --- e2e/k8s/values/kind.yaml | 17 ++++---- .../backends/k8s/backend.py | 32 +++++++-------- .../backends/k8s/config.py | 41 ++++++++++++++++--- .../tests/unit/backends/k8s/test_backend.py | 29 +++++++++++++ 4 files changed, 90 insertions(+), 29 deletions(-) diff --git a/e2e/k8s/values/kind.yaml b/e2e/k8s/values/kind.yaml index bdc241d935..b48cb55bc7 100644 --- a/e2e/k8s/values/kind.yaml +++ b/e2e/k8s/values/kind.yaml @@ -35,11 +35,13 @@ platformConfig: enabled: false # Wire a nemo-deployments k8s executor so deployment_mode=k8s agent # deployments (test_nemo_agents_k8s.py) run as real Deployment+Service - # workloads in the release namespace. The nmp-core controller ServiceAccount - # Role already grants apps/deployments, services, configmaps, and pods (see - # k8s/helm/templates/core/controller-role.yaml). default_namespace matches the - # Helm release namespace so the agent lands beside the platform and the - # in-cluster Inference Gateway URL injected into the pod is reachable. + # workloads. The nmp-core controller ServiceAccount Role already grants + # apps/deployments, services, configmaps, and pods (see + # k8s/helm/templates/core/controller-role.yaml). We omit default_namespace so + # the executor defaults to its own pod namespace (POD_NAMESPACE), i.e. the + # Helm release namespace — the agent lands beside the platform and the + # in-cluster Inference Gateway URL injected into the pod is reachable, without + # hardcoding the namespace here. agents: deployments: k8s_executor: kind-k8s @@ -49,6 +51,5 @@ platformConfig: executors: - name: kind-k8s backend: k8s - config: - # Omit kubeconfig_path -> in-cluster ServiceAccount auth. - default_namespace: nemo-platform + # No config needed: kubeconfig_path unset -> in-cluster ServiceAccount + # auth; default_namespace unset -> POD_NAMESPACE (release namespace). 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 49fcc22ab9..559442790b 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 @@ -62,7 +62,7 @@ def init(self) -> None: self._entities = NemoEntitiesClient(AsyncEntitiesResource(self._sdk)) logger.debug( "K8sDeploymentBackend initialized (default_namespace=%s)", - self._executor_config.default_namespace, + self._executor_config.effective_namespace, ) def shutdown(self) -> None: @@ -113,7 +113,7 @@ async def create_deployment( if config.restart_policy == "Always": return await deployment_ops.create_deployment( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, config_name=config_name, @@ -124,7 +124,7 @@ async def create_deployment( return await job_ops.create_job( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, config_name=config_name, @@ -151,7 +151,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate return BackendStatusUpdate(status="FAILED", status_message=str(exc)) return await deployment_ops.read_deployment_status( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -163,7 +163,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate return await job_ops.read_job_status( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -179,7 +179,7 @@ async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpd scope_labels = job_ops.deployment_scope_labels(workspace, name) deployment_result = await deployment_ops.delete_deployment( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config={}, @@ -187,7 +187,7 @@ async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpd ) job_result = await job_ops.delete_job( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config={}, @@ -207,7 +207,7 @@ async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpd if config.restart_policy == "Always": return await deployment_ops.delete_deployment( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -216,7 +216,7 @@ async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpd return await job_ops.delete_job( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -226,11 +226,11 @@ async def delete_deployment(self, workspace: str, name: str) -> BackendStatusUpd async def list_managed_deployment_names(self) -> list[str]: job_names = await job_ops.list_managed_job_names( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, ) deployment_names = await deployment_ops.list_managed_deployment_names( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, ) return sorted(set(job_names) | set(deployment_names)) @@ -251,7 +251,7 @@ async def get_logs( if config.restart_policy == "Always": return await deployment_ops.get_deployment_logs( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -260,7 +260,7 @@ async def get_logs( return await job_ops.get_job_logs( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -278,7 +278,7 @@ async def create_volume( ) -> VolumeStatusUpdate: return await volume_ops.create_volume( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, size=size, @@ -295,7 +295,7 @@ async def read_volume_status( ) -> VolumeStatusUpdate: return await volume_ops.read_volume_status( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, @@ -310,7 +310,7 @@ async def delete_volume( ) -> VolumeStatusUpdate: return await volume_ops.delete_volume( self._clients, - default_namespace=self._executor_config.default_namespace, + default_namespace=self._executor_config.effective_namespace, workspace=workspace, name=name, backend_config=backend_config, diff --git a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/config.py b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/config.py index 61d2abb0d0..8e91834fbc 100644 --- a/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/config.py +++ b/plugins/nemo-deployments/src/nemo_deployments_plugin/backends/k8s/config.py @@ -5,12 +5,22 @@ from __future__ import annotations +import os import re from pydantic import BaseModel, Field, field_validator _DNS_LABEL_PATTERN = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$") +# Downward-API env var carrying the controller pod's own namespace. The core +# controller deployment (where the deployments backend runs) injects this via +# ``fieldRef: metadata.namespace`` (see k8s/helm/templates/core/controller-deployment.yaml). +_POD_NAMESPACE_ENV = "POD_NAMESPACE" + +# Last-resort namespace when neither an explicit config value nor the downward +# API is available (e.g. local/out-of-cluster runs). +_FALLBACK_NAMESPACE = "default" + class K8sExecutorConfig(BaseModel): """Knobs for a named k8s executor instance (not entity backend_config).""" @@ -19,11 +29,16 @@ class K8sExecutorConfig(BaseModel): default=None, description="Path to kubeconfig file. When unset, uses in-cluster config or default kubeconfig.", ) - default_namespace: str = Field( - default="default", + default_namespace: str | None = Field( + default=None, min_length=1, max_length=63, - description="Namespace for resources when entity backend_config.k8s.namespace is unset.", + description=( + "Namespace for resources when entity backend_config.k8s.namespace is unset. When this is " + "itself unset, the executor defaults to its own pod namespace (POD_NAMESPACE, injected via " + "the downward API), so a deployed platform places agents beside itself; if that is also " + "unavailable it falls back to 'default'. See effective_namespace." + ), ) request_timeout: int = Field( default=60, @@ -33,7 +48,23 @@ class K8sExecutorConfig(BaseModel): @field_validator("default_namespace") @classmethod - def _validate_default_namespace(cls, value: str) -> str: - if not _DNS_LABEL_PATTERN.fullmatch(value): + def _validate_default_namespace(cls, value: str | None) -> str | None: + if value is not None and not _DNS_LABEL_PATTERN.fullmatch(value): raise ValueError("default_namespace must be a lowercase DNS-1123 label (alphanumeric, interior hyphens)") return value + + @property + def effective_namespace(self) -> str: + """Resolve the namespace to deploy into. + + Precedence: an explicit ``default_namespace`` config value wins; else the + controller's own pod namespace (``POD_NAMESPACE`` downward API); else + ``"default"``. This lets a deployed platform omit the setting and still + place agents in its own namespace, while an operator can always pin one. + """ + if self.default_namespace is not None: + return self.default_namespace + pod_namespace = os.environ.get(_POD_NAMESPACE_ENV, "").strip() + if pod_namespace: + return pod_namespace + return _FALLBACK_NAMESPACE diff --git a/plugins/nemo-deployments/tests/unit/backends/k8s/test_backend.py b/plugins/nemo-deployments/tests/unit/backends/k8s/test_backend.py index 30bab564b7..1ee4a2eb26 100644 --- a/plugins/nemo-deployments/tests/unit/backends/k8s/test_backend.py +++ b/plugins/nemo-deployments/tests/unit/backends/k8s/test_backend.py @@ -25,3 +25,32 @@ def test_shutdown_closes_kubernetes_clients(k8s_backend: K8sDeploymentBackend) - def test_default_namespace_rejects_invalid_dns_label() -> None: with pytest.raises(ValueError, match="default_namespace must be a lowercase DNS-1123 label"): K8sExecutorConfig(default_namespace="X") + + +def test_effective_namespace_prefers_explicit_config(monkeypatch: pytest.MonkeyPatch) -> None: + # An explicit config value wins even when POD_NAMESPACE is set. + monkeypatch.setenv("POD_NAMESPACE", "pod-ns") + config = K8sExecutorConfig(default_namespace="explicit-ns") + assert config.effective_namespace == "explicit-ns" + + +def test_effective_namespace_falls_back_to_pod_namespace(monkeypatch: pytest.MonkeyPatch) -> None: + # No explicit config -> the controller's own pod namespace (downward API). + monkeypatch.setenv("POD_NAMESPACE", "pod-ns") + config = K8sExecutorConfig() + assert config.default_namespace is None + assert config.effective_namespace == "pod-ns" + + +def test_effective_namespace_falls_back_to_default(monkeypatch: pytest.MonkeyPatch) -> None: + # Neither an explicit config value nor POD_NAMESPACE -> "default". + monkeypatch.delenv("POD_NAMESPACE", raising=False) + config = K8sExecutorConfig() + assert config.effective_namespace == "default" + + +def test_effective_namespace_ignores_blank_pod_namespace(monkeypatch: pytest.MonkeyPatch) -> None: + # A blank/whitespace POD_NAMESPACE is treated as unset. + monkeypatch.setenv("POD_NAMESPACE", " ") + config = K8sExecutorConfig() + assert config.effective_namespace == "default" From af7245123058aab1864788d7181eb08ad1063dba Mon Sep 17 00:00:00 2001 From: Ben McCown Date: Fri, 17 Jul 2026 09:41:26 -0600 Subject: [PATCH 3/3] CR feedback Signed-off-by: Ben McCown --- e2e/k8s/values/kind.yaml | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/e2e/k8s/values/kind.yaml b/e2e/k8s/values/kind.yaml index b48cb55bc7..cc6880fc6e 100644 --- a/e2e/k8s/values/kind.yaml +++ b/e2e/k8s/values/kind.yaml @@ -35,13 +35,8 @@ platformConfig: enabled: false # Wire a nemo-deployments k8s executor so deployment_mode=k8s agent # deployments (test_nemo_agents_k8s.py) run as real Deployment+Service - # workloads. The nmp-core controller ServiceAccount Role already grants - # apps/deployments, services, configmaps, and pods (see - # k8s/helm/templates/core/controller-role.yaml). We omit default_namespace so - # the executor defaults to its own pod namespace (POD_NAMESPACE), i.e. the - # Helm release namespace — the agent lands beside the platform and the - # in-cluster Inference Gateway URL injected into the pod is reachable, without - # hardcoding the namespace here. + # workloads. default_namespace is omitted so the executor uses its own pod + # namespace (the Helm release namespace), landing the agent beside the platform. agents: deployments: k8s_executor: kind-k8s