diff --git a/.github/actions/setup-kind-cluster/action.yaml b/.github/actions/setup-kind-cluster/action.yaml new file mode 100644 index 0000000000..961d6a8b09 --- /dev/null +++ b/.github/actions/setup-kind-cluster/action.yaml @@ -0,0 +1,174 @@ +name: Setup Kind cluster with NeMo Platform +description: > + Creates a Kind cluster, installs tooling (kind, kubectl, Helm, uv), + pre-pulls images, deploys NeMo Platform via Helm, and waits for the + API to become healthy. + +inputs: + kind_cluster_name: + description: Kind cluster name + required: true + kube_namespace: + description: Kubernetes namespace for NeMo Platform + required: false + default: nemo-platform + kube_gateway_name: + description: Gateway resource name + required: false + default: nmp-e2e-gateway + image_registry: + description: Container image registry + required: true + image_tag: + description: Container image tag + required: true + helm_values: + description: Helm values file path (relative to repo root) + required: false + default: e2e/k8s/values/kind.yaml + kind_image_pull_token: + description: Token for pulling images into Kind + required: true + kind_image_pull_user: + description: Username for pulling images into Kind + required: true + ngc_api_key: + description: NGC API key (can be a placeholder for CPU-only) + required: false + default: not-used + +runs: + using: composite + steps: + - name: Free disk space + uses: ./.github/actions/free-disk-space + with: + disable_swap: "true" + remove_haskell: "true" + remove_java: "true" + remove_ruby: "true" + remove_swift: "true" + prune_docker: "true" + + - name: Install kind + shell: bash + env: + KIND_VERSION: v0.32.0 + run: | + set -euo pipefail + + case "$(uname -m)" in + x86_64) arch=amd64 ;; + aarch64|arm64) arch=arm64 ;; + *) + echo "Unsupported architecture: $(uname -m)" >&2 + exit 1 + ;; + esac + + kind_url="https://github.com/kubernetes-sigs/kind/releases/download/${KIND_VERSION}/kind-linux-${arch}" + curl -fsSLo "${RUNNER_TEMP}/kind" "${kind_url}" + curl -fsSLo "${RUNNER_TEMP}/kind.sha256sum" "${kind_url}.sha256sum" + sed "s# kind-linux-${arch}# ${RUNNER_TEMP}/kind#" "${RUNNER_TEMP}/kind.sha256sum" | sha256sum -c - + sudo install -m 0755 "${RUNNER_TEMP}/kind" /usr/local/bin/kind + + - name: Install kubectl + shell: bash + env: + KUBECTL_VERSION: v1.33.7 + run: | + set -euo pipefail + + case "$(uname -m)" in + x86_64) arch=amd64 ;; + aarch64|arm64) arch=arm64 ;; + *) + echo "Unsupported architecture: $(uname -m)" >&2 + exit 1 + ;; + esac + + kubectl_url="https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/${arch}/kubectl" + curl -fsSLo "${RUNNER_TEMP}/kubectl" "${kubectl_url}" + curl -fsSLo "${RUNNER_TEMP}/kubectl.sha256" "${kubectl_url}.sha256" + echo "$(cat "${RUNNER_TEMP}/kubectl.sha256") ${RUNNER_TEMP}/kubectl" | sha256sum -c - + sudo install -m 0755 "${RUNNER_TEMP}/kubectl" /usr/local/bin/kubectl + + - name: Install Helm + uses: azure/setup-helm@dda3372f752e03dde6b3237bc9431cdc2f7a02a2 # v5.0.0 + + - name: Install uv + uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0 + with: + python-version: "3.13" + enable-cache: true + version-file: pyproject.toml + cache-dependency-glob: uv.lock + + - name: Start kind cluster + shell: bash + env: + KIND_CLUSTER_NAME: ${{ inputs.kind_cluster_name }} + KUBE_NAMESPACE: ${{ inputs.kube_namespace }} + NGC_API_KEY: ${{ inputs.ngc_api_key }} + run: bash e2e/k8s/scripts/setup_local_kind_cpu.sh + + - name: Set default kubectl namespace + shell: bash + env: + NAMESPACE: ${{ inputs.kube_namespace }} + run: kubectl config set-context --current --namespace="${NAMESPACE}" + + - name: Verify Gateway API setup + shell: bash + env: + NAMESPACE: ${{ inputs.kube_namespace }} + KUBE_GATEWAY_NAME: ${{ inputs.kube_gateway_name }} + run: | + set -euo pipefail + kubectl wait --for=condition=Established crd/gateways.gateway.networking.k8s.io --timeout=2m + kubectl wait --for=condition=Established crd/httproutes.gateway.networking.k8s.io --timeout=2m + kubectl get gatewayclass cloud-provider-kind + kubectl -n "${NAMESPACE}" get gateway "${KUBE_GATEWAY_NAME}" + + - name: Pre-pull GHCR images into kind + shell: bash + env: + KIND_IMAGE_PULL_TOKEN: ${{ inputs.kind_image_pull_token }} + KIND_IMAGE_PULL_USER: ${{ inputs.kind_image_pull_user }} + NMP_E2E_REGISTRY: ${{ inputs.image_registry }} + NMP_E2E_TAG: ${{ inputs.image_tag }} + run: | + e2e/k8s/scripts/prepull_kind_images.sh \ + "${NMP_E2E_REGISTRY}/nmp-api:${NMP_E2E_TAG}" \ + "${NMP_E2E_REGISTRY}/nmp-core:${NMP_E2E_TAG}" \ + "${NMP_E2E_REGISTRY}/nmp-cpu-tasks:${NMP_E2E_TAG}" + + - name: Install NeMo Platform + shell: bash + env: + NAMESPACE: ${{ inputs.kube_namespace }} + NMP_E2E_REGISTRY: ${{ inputs.image_registry }} + NMP_E2E_TAG: ${{ inputs.image_tag }} + HELM_VALUES: ${{ inputs.helm_values }} + REQUIRE_NMP_E2E_IMAGES: "true" + POSTGRES_IMAGE: docker.io/library/postgres + BUSYBOX_IMAGE: docker.io/library/busybox + run: | + if ! e2e/k8s/scripts/install_helm_e2e.sh; then + echo "--- helm list -A ---" + helm list -A || true + echo "--- helm status ${NAMESPACE}/nemo-platform ---" + helm status -n "${NAMESPACE}" nemo-platform || true + echo "--- kubectl get all -n ${NAMESPACE} ---" + kubectl get all -n "${NAMESPACE}" || true + exit 1 + fi + + - name: Wait for API + shell: bash + env: + NMP_E2E_CLUSTER_URL: ${{ env.NMP_E2E_CLUSTER_URL }} + run: | + test -n "${NMP_E2E_CLUSTER_URL}" + e2e/k8s/scripts/wait_for_api.sh "${NMP_E2E_CLUSTER_URL}/cluster-info" 120 diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index f0022991b6..afafbe5f9c 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -391,162 +391,125 @@ jobs: contents: read packages: read env: - BUSYBOX_IMAGE: docker.io/library/busybox - HELM_CHART: k8s/helm - K8S_E2E_SCRIPTS: e2e/k8s/scripts - K8S_E2E_VALUES: e2e/k8s/values KIND_CLUSTER_NAME: gha-${{ github.run_id }}-${{ github.run_attempt }}-kind-smoke - KUBE_GATEWAY_NAME: nmp-e2e-gateway - KUBE_NAMESPACE: nemo-platform - NAMESPACE: nemo-platform - NMP_E2E_CLUSTER_URL: "" - NMP_E2E_INTERNAL_HOST: nemo-platform-api:8080 - NMP_E2E_REGISTRY: ${{ needs.build-cpu-smoke-images.outputs.image_registry }} - NMP_E2E_TAG: ${{ needs.build-cpu-smoke-images.outputs.image_tag }} - POSTGRES_IMAGE: docker.io/library/postgres steps: - name: Checkout code uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 with: persist-credentials: false - - name: Free disk space - uses: ./.github/actions/free-disk-space + - name: Setup Kind cluster with NeMo Platform + uses: ./.github/actions/setup-kind-cluster with: - disable_swap: "true" - remove_haskell: "true" - remove_java: "true" - remove_ruby: "true" - remove_swift: "true" - prune_docker: "true" + kind_cluster_name: ${{ env.KIND_CLUSTER_NAME }} + image_registry: ${{ needs.build-cpu-smoke-images.outputs.image_registry }} + image_tag: ${{ needs.build-cpu-smoke-images.outputs.image_tag }} + kind_image_pull_token: ${{ github.token }} + kind_image_pull_user: ${{ github.actor }} - - name: Install kind - shell: bash - env: - KIND_VERSION: v0.32.0 - run: | - set -euo pipefail - - case "$(uname -m)" in - x86_64) arch=amd64 ;; - aarch64|arm64) arch=arm64 ;; - *) - echo "Unsupported architecture: $(uname -m)" >&2 - exit 1 - ;; - esac - - kind_url="https://github.com/kubernetes-sigs/kind/releases/download/${KIND_VERSION}/kind-linux-${arch}" - curl -fsSLo "${RUNNER_TEMP}/kind" "${kind_url}" - curl -fsSLo "${RUNNER_TEMP}/kind.sha256sum" "${kind_url}.sha256sum" - sed "s# kind-linux-${arch}# ${RUNNER_TEMP}/kind#" "${RUNNER_TEMP}/kind.sha256sum" | sha256sum -c - - sudo install -m 0755 "${RUNNER_TEMP}/kind" /usr/local/bin/kind - - - name: Install kubectl - shell: bash - env: - KUBECTL_VERSION: v1.33.7 - run: | - set -euo pipefail - - case "$(uname -m)" in - x86_64) arch=amd64 ;; - aarch64|arm64) arch=arm64 ;; - *) - echo "Unsupported architecture: $(uname -m)" >&2 - exit 1 - ;; - esac - - kubectl_url="https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/${arch}/kubectl" - curl -fsSLo "${RUNNER_TEMP}/kubectl" "${kubectl_url}" - curl -fsSLo "${RUNNER_TEMP}/kubectl.sha256" "${kubectl_url}.sha256" - echo "$(cat "${RUNNER_TEMP}/kubectl.sha256") ${RUNNER_TEMP}/kubectl" | sha256sum -c - - sudo install -m 0755 "${RUNNER_TEMP}/kubectl" /usr/local/bin/kubectl - - - name: Install Helm - uses: azure/setup-helm@dda3372f752e03dde6b3237bc9431cdc2f7a02a2 # v5.0.0 - - - name: Install uv - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0 - with: - python-version: "3.13" - enable-cache: true - version-file: pyproject.toml - cache-dependency-glob: uv.lock - - - name: Start kind cluster + - name: Run CPU job e2e smoke test shell: bash env: + _TYPER_FORCE_DISABLE_TERMINAL: "1" + E2E_SERVICES_LOG_DIR: ${{ runner.temp }}/e2e-services-logs NGC_API_KEY: not-used-for-ghcr-cpu-smoke - run: bash "${K8S_E2E_SCRIPTS}/setup_local_kind_cpu.sh" + run: | + test -n "${NMP_E2E_CLUSTER_URL}" + export NMP_BASE_URL="${NMP_E2E_CLUSTER_URL}" + uv run --frozen pytest \ + e2e/test_jobs.py::test_job_using_secret_environment_variable \ + -v \ + --run-e2e \ + --no-cov \ + --junitxml=report-kubernetes-smoke.xml - - name: Set default kubectl namespace + - name: Collect Kubernetes logs + if: always() shell: bash - run: kubectl config set-context --current --namespace="${NAMESPACE}" + run: e2e/k8s/scripts/collect_k8s_logs.sh - - name: Verify Gateway API setup + - name: Disk usage summary + if: always() shell: bash run: | - set -euo pipefail - kubectl wait --for=condition=Established crd/gateways.gateway.networking.k8s.io --timeout=2m - kubectl wait --for=condition=Established crd/httproutes.gateway.networking.k8s.io --timeout=2m - kubectl get gatewayclass cloud-provider-kind - kubectl -n "${NAMESPACE}" get gateway "${KUBE_GATEWAY_NAME}" + echo "=== Host disk ===" + df -h / + echo "=== Docker system ===" + docker system df + echo "=== kind node storage ===" + for node in $(kind get nodes --name "${KIND_CLUSTER_NAME}" 2>/dev/null); do + echo "--- ${node} ---" + docker exec "${node}" sh -c "du -sh /var/lib/containerd /var/lib/kubelet /var/log 2>/dev/null | sort -h" || true + done - - name: Pre-pull GHCR images into kind - shell: bash - env: - KIND_IMAGE_PULL_TOKEN: ${{ github.token }} - KIND_IMAGE_PULL_USER: ${{ github.actor }} - run: | - "${K8S_E2E_SCRIPTS}/prepull_kind_images.sh" \ - "${NMP_E2E_REGISTRY}/nmp-api:${NMP_E2E_TAG}" \ - "${NMP_E2E_REGISTRY}/nmp-core:${NMP_E2E_TAG}" \ - "${NMP_E2E_REGISTRY}/nmp-cpu-tasks:${NMP_E2E_TAG}" + - name: Upload Kubernetes artifacts + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: kind-smoke-kubernetes-artifacts + retention-days: 7 + if-no-files-found: ignore + path: | + k8s-logs/ + report-kubernetes-smoke.xml + ${{ runner.temp }}/e2e-services-logs/ - - name: Install NeMo Platform + - name: Delete kind cluster + if: always() shell: bash - env: - REQUIRE_NMP_E2E_IMAGES: "true" run: | - if ! HELM_VALUES="${K8S_E2E_VALUES}/kind.yaml" "${K8S_E2E_SCRIPTS}/install_helm_e2e.sh"; then - echo "--- helm list -A ---" - helm list -A || true - echo "--- helm status ${NAMESPACE}/nemo-platform ---" - helm status -n "${NAMESPACE}" nemo-platform || true - echo "--- kubectl get all -n ${NAMESPACE} ---" - kubectl get all -n "${NAMESPACE}" || true - exit 1 - fi + docker rm -f "cloud-provider-kind-${KIND_CLUSTER_NAME}" || true + kind delete cluster --name "${KIND_CLUSTER_NAME}" || true - - name: Wait for API - shell: bash - run: | - test -n "${NMP_E2E_CLUSTER_URL}" - "${K8S_E2E_SCRIPTS}/wait_for_api.sh" "${NMP_E2E_CLUSTER_URL}/cluster-info" 120 + kind-cpu-e2e: + name: Kind CPU e2e + needs: [changes, build-cpu-smoke-images] + if: > + !cancelled() && + needs.build-cpu-smoke-images.result == 'success' && + needs.build-cpu-smoke-images.outputs.publish_images == 'true' + runs-on: ubuntu-latest + timeout-minutes: 45 + permissions: + contents: read + packages: read + env: + KIND_CLUSTER_NAME: gha-${{ github.run_id }}-${{ github.run_attempt }}-kind-e2e + steps: + - name: Checkout code + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + persist-credentials: false - - name: Run CPU job e2e smoke test + - name: Setup Kind cluster with NeMo Platform + uses: ./.github/actions/setup-kind-cluster + with: + kind_cluster_name: ${{ env.KIND_CLUSTER_NAME }} + image_registry: ${{ needs.build-cpu-smoke-images.outputs.image_registry }} + image_tag: ${{ needs.build-cpu-smoke-images.outputs.image_tag }} + kind_image_pull_token: ${{ github.token }} + kind_image_pull_user: ${{ github.actor }} + + - name: Run jobs and data-designer e2e tests shell: bash env: _TYPER_FORCE_DISABLE_TERMINAL: "1" E2E_SERVICES_LOG_DIR: ${{ runner.temp }}/e2e-services-logs - NGC_API_KEY: not-used-for-ghcr-cpu-smoke + NGC_API_KEY: not-used-for-ghcr-cpu-e2e run: | test -n "${NMP_E2E_CLUSTER_URL}" export NMP_BASE_URL="${NMP_E2E_CLUSTER_URL}" uv run --frozen pytest \ - e2e/test_jobs.py::test_job_using_secret_environment_variable \ + e2e \ -v \ --run-e2e \ --no-cov \ - --junitxml=report-kubernetes-smoke.xml + --junitxml=report-kubernetes-e2e.xml - name: Collect Kubernetes logs if: always() shell: bash - run: | - "${K8S_E2E_SCRIPTS}/collect_k8s_logs.sh" + run: e2e/k8s/scripts/collect_k8s_logs.sh - name: Disk usage summary if: always() @@ -566,12 +529,12 @@ jobs: if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: - name: kind-smoke-kubernetes-artifacts + name: kind-e2e-kubernetes-artifacts retention-days: 7 if-no-files-found: ignore path: | k8s-logs/ - report-kubernetes-smoke.xml + report-kubernetes-e2e.xml ${{ runner.temp }}/e2e-services-logs/ - name: Delete kind cluster diff --git a/conftest.py b/conftest.py index f4deccde38..9c780c715f 100644 --- a/conftest.py +++ b/conftest.py @@ -287,6 +287,12 @@ def pytest_runtest_setup(item): if "e2e" in [marker.name for marker in item.iter_markers()]: if not item.config.getoption("--run-e2e"): skip_test("Skipping e2e test (use --run-e2e to run)") + if "subprocess_only" in [marker.name for marker in item.iter_markers()]: + if os.environ.get("NMP_BASE_URL"): + skip_test("Skipping subprocess-only test (NMP_BASE_URL is set)") + if "container_only" in [marker.name for marker in item.iter_markers()]: + if not os.environ.get("NMP_BASE_URL"): + skip_test("Skipping container-only test (requires NMP_BASE_URL)") from xdist.scheduler.loadscope import LoadScopeScheduling # noqa: E402 diff --git a/e2e/conftest.py b/e2e/conftest.py index 3a49a635a2..3826cb5bed 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -110,9 +110,14 @@ def pytest_collection_modifyitems(session: pytest.Session, config: pytest.Config @pytest.fixture def ngc_api_key() -> str: - """Return the NGC API key from the environment.""" - key = os.environ.get(NGC_API_KEY_ENV) - assert key, f"{NGC_API_KEY_ENV} must be set" + """Return the NGC API key from the environment. + + Skips the test when the key is missing or set to a CI placeholder + value (e.g. ``not-used-for-ghcr-cpu-*``). + """ + key = os.environ.get(NGC_API_KEY_ENV, "") + if not key or key.startswith("not-used"): + pytest.skip(f"{NGC_API_KEY_ENV} not set or is a placeholder") return key diff --git a/e2e/test_jobs.py b/e2e/test_jobs.py index c78d101a9d..012f4d74f7 100644 --- a/e2e/test_jobs.py +++ b/e2e/test_jobs.py @@ -13,6 +13,7 @@ import pytest from nemo_platform import NeMoPlatform, NotFoundError +from nemo_platform_plugin.jobs.constants import DEFAULT_JOB_STORAGE_PATH from nmp.testing.e2e import wait_for_job_logs, wait_for_platform_job JOB_SOURCE = "e2e-test-jobs" @@ -166,6 +167,10 @@ def test_job_config_is_readable(sdk: NeMoPlatform, workspace: str): def test_job_passing_data_between_steps(sdk: NeMoPlatform, workspace: str): """Test that data can be passed between job steps via persistent storage.""" + persistent_storage_env = { + "name": "NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH", + "value": DEFAULT_JOB_STORAGE_PATH, + } job = sdk.jobs.create( workspace=workspace, source=JOB_SOURCE, @@ -184,6 +189,7 @@ def test_job_passing_data_between_steps(sdk: NeMoPlatform, workspace: str): ], }, }, + "environment": [persistent_storage_env], }, { "name": "consume-data-step", @@ -197,6 +203,7 @@ def test_job_passing_data_between_steps(sdk: NeMoPlatform, workspace: str): ], }, }, + "environment": [persistent_storage_env], }, ], }, @@ -361,11 +368,15 @@ def test_job_cancel_once_active(sdk: NeMoPlatform, workspace: str): # --------------------------------------------------------------------------- -# Tests that require Docker backend +# Tests that require a container backend (Docker or Kubernetes) # --------------------------------------------------------------------------- -@pytest.mark.skip(reason="Subprocess backend does not support pause/resume (no SIGSTOP/SIGCONT handling)") +# AIRCORE-853: K8s reconciler checks for errored pods before checking if the +# job is suspended. When K8s kills pods during suspension, the terminated pod +# is misclassified as an error, causing the job to transition to 'error' +# instead of 'paused'. Re-enable once the reconciler is fixed. +@pytest.mark.skip(reason="AIRCORE-853: pause races with errored-pod detection in K8s reconciler") def test_job_pause_resume(sdk: NeMoPlatform, workspace: str): """Test that a job can be paused and then resumed after being paused.""" job = sdk.jobs.create( @@ -406,7 +417,7 @@ def test_job_pause_resume(sdk: NeMoPlatform, workspace: str): assert completed_job.status == "completed", f"Job failed with status: {completed_job.status}" -@pytest.mark.skip(reason="Subprocess backend does not support pause/resume (no SIGSTOP/SIGCONT handling)") +@pytest.mark.skip(reason="AIRCORE-853: pause races with errored-pod detection in K8s reconciler") def test_job_pause_and_cancel(sdk: NeMoPlatform, workspace: str): """Test that a job can be paused and then cancelled after being paused.""" job = sdk.jobs.create( @@ -442,7 +453,7 @@ def test_job_pause_and_cancel(sdk: NeMoPlatform, workspace: str): assert cancelled_job.status == "cancelled", f"Job should have been cancelled but has status: {cancelled_job.status}" -@pytest.mark.skip(reason="Docker-only: additional volumes require container volume mounts") +@pytest.mark.skip(reason="Requires additional_volumes configured in Helm chart storage config") def test_job_using_additional_volume(sdk: NeMoPlatform, workspace: str): """Test that a job can use an additional volume to store data between steps.""" job = sdk.jobs.create( @@ -493,10 +504,7 @@ def test_job_using_additional_volume(sdk: NeMoPlatform, workspace: str): assert "Successfully read data from persistent storage" in step_logs.data[2].message -@pytest.mark.skip( - reason="Docker-only: image validation is bypassed in subprocess mode " - "(cpu→subprocess translation discards the container image)" -) +@pytest.mark.container_only @pytest.mark.parametrize("bad_image", ["__invalid_ubuntu:image", "ubuntu:does-not-exist-1234"]) def test_job_invalid_image_format(sdk: NeMoPlatform, workspace: str, bad_image: str): """Test that a job with a bad image fails appropriately.""" diff --git a/e2e/test_jobs_auth.py b/e2e/test_jobs_auth.py index 79d4a096a8..ed83ebc225 100644 --- a/e2e/test_jobs_auth.py +++ b/e2e/test_jobs_auth.py @@ -28,6 +28,7 @@ logger = logging.getLogger(__name__) pytestmark = [ + pytest.mark.subprocess_only, pytest.mark.e2e_config("e2e/configs/local-subprocess.yaml", {"auth": {"enabled": True}}), ] diff --git a/e2e/test_smoke.py b/e2e/test_smoke.py index 322bd1efd3..9cca433fc0 100644 --- a/e2e/test_smoke.py +++ b/e2e/test_smoke.py @@ -10,14 +10,15 @@ def test_health_ready(sdk: NeMoPlatform): - """GET /health/ready returns 200 when all services are up.""" - resp = sdk._client.get("/health/ready") + """GET /status returns 200 with healthy status when all services are up.""" + resp = sdk._client.get("/status") assert resp.status_code == 200 + assert resp.json()["status"] == "healthy" def test_health_live(sdk: NeMoPlatform): - """GET /health/live returns 200 (liveness probe).""" - resp = sdk._client.get("/health/live") + """GET /status returns 200 (platform is reachable).""" + resp = sdk._client.get("/status") assert resp.status_code == 200 diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/job_context.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/job_context.py index 78b1019d56..9a5cbc0359 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/job_context.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/job_context.py @@ -45,7 +45,6 @@ def run(self, config: dict, *, ctx: JobContext, is_local: bool) -> dict: from nemo_platform_plugin.job_results import JobResults -@dataclass class StoragePaths: """Filesystem locations a job can read and write during execution. @@ -53,13 +52,29 @@ class StoragePaths: ephemeral: Scratch directory for working files and intermediate artifacts. No guarantees across steps or retries. Maps to ``NEMO_JOB_EPHEMERAL_TASK_STORAGE_PATH``. - persistent: Directory whose contents are preserved after the job - ends. Writes are slower than to ``ephemeral`` — keep final - outputs only. Maps to ``NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH``. + persistent: PVC-backed directory shared across steps within the + same job. Only available when the job step declares + ``NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH`` in its compile() + environment. Raises ``RuntimeError`` if accessed without + being provisioned. Maps to + ``NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH``. """ - ephemeral: Path - persistent: Path + def __init__(self, ephemeral: Path, persistent: Path | None = None) -> None: + self.ephemeral = ephemeral + self._persistent = persistent + + @property + def persistent(self) -> Path: + """Return the persistent storage path, or raise if not provisioned.""" + if self._persistent is None: + raise RuntimeError( + "This job did not request persistent storage. " + "Add NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH to the step's " + "environment list in compile() to enable it, or use " + "ctx.storage.ephemeral for scratch data." + ) + return self._persistent @dataclass(kw_only=True) diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/tasks/dispatcher.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/tasks/dispatcher.py index c2b7f24cd7..93d11a5b0d 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/tasks/dispatcher.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/tasks/dispatcher.py @@ -207,8 +207,6 @@ def _build_ctx_from_env(sdk: Any) -> JobContext: if not workspace: raise RuntimeError(f"{NEMO_JOB_WORKSPACE_ENVVAR} not set; running outside the platform?") persistent_str = os.environ.get(PERSISTENT_JOB_STORAGE_PATH_ENVVAR) - if not persistent_str: - raise RuntimeError(f"{PERSISTENT_JOB_STORAGE_PATH_ENVVAR} not set; running outside the platform?") ephemeral_str = os.environ.get(EPHEMERAL_TASK_STORAGE_PATH_ENVVAR) if not ephemeral_str: raise RuntimeError(f"{EPHEMERAL_TASK_STORAGE_PATH_ENVVAR} not set; running outside the platform?") @@ -217,7 +215,10 @@ def _build_ctx_from_env(sdk: Any) -> JobContext: raise RuntimeError(f"{NEMO_JOB_ID_ENVVAR} not set; running outside the platform?") return JobContext( workspace=workspace, - storage=StoragePaths(ephemeral=Path(ephemeral_str), persistent=Path(persistent_str)), + storage=StoragePaths( + ephemeral=Path(ephemeral_str), + persistent=Path(persistent_str) if persistent_str else None, + ), results=PlatformJobResults(job_name=job_id, workspace=workspace, sdk=sdk), job_id=job_id, ) diff --git a/packages/nemo_platform_plugin/tests/test_dispatcher.py b/packages/nemo_platform_plugin/tests/test_dispatcher.py index c8d0c68d47..f8bffcc820 100644 --- a/packages/nemo_platform_plugin/tests/test_dispatcher.py +++ b/packages/nemo_platform_plugin/tests/test_dispatcher.py @@ -666,17 +666,20 @@ def test_whitespace_workspace_raises(self, monkeypatch) -> None: with pytest.raises(RuntimeError, match="NEMO_JOB_WORKSPACE"): _build_ctx_from_env(_DEFAULT_SDK) - def test_missing_persistent_storage_raises(self, tmp_path: Path, monkeypatch) -> None: - # No silent fallback to ``/var/run/scratch/job``: that path only - # exists in container images, so a missing envvar in a subprocess - # executor must surface as a hard configuration error rather than - # masquerade as a working ctx. + def test_missing_persistent_storage_builds_ctx_but_access_raises(self, tmp_path: Path, monkeypatch) -> None: + # Persistent storage is optional — the ctx builds successfully + # without it, but accessing ctx.storage.persistent raises a clear + # RuntimeError so jobs that need it fail fast with guidance. monkeypatch.setenv("NEMO_JOB_WORKSPACE", "ws") monkeypatch.delenv("NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH", raising=False) monkeypatch.setenv("NEMO_JOB_EPHEMERAL_TASK_STORAGE_PATH", str(tmp_path / "e")) + monkeypatch.setenv("NEMO_JOB_ID", "test-job") - with pytest.raises(RuntimeError, match="NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH"): - _build_ctx_from_env(_DEFAULT_SDK) + ctx = _build_ctx_from_env(_DEFAULT_SDK) + assert ctx.storage.ephemeral == tmp_path / "e" + + with pytest.raises(RuntimeError, match="did not request persistent storage"): + _ = ctx.storage.persistent def test_missing_ephemeral_storage_raises(self, tmp_path: Path, monkeypatch) -> None: monkeypatch.setenv("NEMO_JOB_WORKSPACE", "ws") diff --git a/plugins/nemo-data-designer/src/nemo_data_designer_plugin/jobs/bridge.py b/plugins/nemo-data-designer/src/nemo_data_designer_plugin/jobs/bridge.py index 8d2a1fab5d..fb5e04a0dc 100644 --- a/plugins/nemo-data-designer/src/nemo_data_designer_plugin/jobs/bridge.py +++ b/plugins/nemo-data-designer/src/nemo_data_designer_plugin/jobs/bridge.py @@ -42,9 +42,10 @@ def _get_ctx(sdk: NeMoPlatform) -> JobContext: workspace = os.environ[NEMO_JOB_WORKSPACE_ENVVAR] job_name = os.environ[NEMO_JOB_ID_ENVVAR] + persistent_env = os.environ.get(PERSISTENT_JOB_STORAGE_PATH_ENVVAR) storage = StoragePaths( ephemeral=Path(os.environ[EPHEMERAL_TASK_STORAGE_PATH_ENVVAR]), - persistent=Path(os.environ[PERSISTENT_JOB_STORAGE_PATH_ENVVAR]), + persistent=Path(persistent_env) if persistent_env else None, ) results = PlatformJobResults( workspace=workspace, diff --git a/pytest.ini b/pytest.ini index fe374a65fb..a884ce87c2 100644 --- a/pytest.ini +++ b/pytest.ini @@ -60,6 +60,8 @@ markers = smoke_nmp_automodel_training: Import smoke tests for the nmp-automodel-training image e2e: End-to-end tests - test complete customer workflows on deployed infrastructure (Helm/Docker Compose) e2e_config(*layers): Ordered list of repo-root-relative config paths and/or inline dict overlays; empty means default local config + subprocess_only: Test only works in subprocess mode (not on Kubernetes); skipped when NMP_BASE_URL is set + container_only: Test requires a container backend (Docker or Kubernetes); skipped unless NMP_BASE_URL is set regression: Regression tests - test individual functional microservices for baseline functionality infrastructure: Infrastructure tests - ensure services are compatible with customer infrastructure canary: Canary tests - test deployed integration environments like top of tree diff --git a/services/core/jobs/jobs-launcher/cmd/root.go b/services/core/jobs/jobs-launcher/cmd/root.go index 5f43b433c0..0ba28d8786 100644 --- a/services/core/jobs/jobs-launcher/cmd/root.go +++ b/services/core/jobs/jobs-launcher/cmd/root.go @@ -23,4 +23,9 @@ func Execute() { logger.Printf("Command execution failed: %v", err) os.Exit(1) } + // os.Exit here instead of in the cobra Run callback so that deferred + // functions (OTEL shutdown / log flush) in runExecWithStdin run first. + if launcherExitCode != 0 { + os.Exit(launcherExitCode) + } } diff --git a/services/core/jobs/jobs-launcher/cmd/run.go b/services/core/jobs/jobs-launcher/cmd/run.go index 6cf9faddd2..48a1b470eb 100644 --- a/services/core/jobs/jobs-launcher/cmd/run.go +++ b/services/core/jobs/jobs-launcher/cmd/run.go @@ -16,11 +16,9 @@ import ( "strings" "sync" "syscall" - "time" "github.com/NVIDIA-NeMo/nemo-platform/services/core/jobs/jobs-launcher/nmpclient" "github.com/spf13/cobra" - "go.opentelemetry.io/otel/sdk/log" ) var runCmd = &cobra.Command{ @@ -32,10 +30,18 @@ var runCmd = &cobra.Command{ if err != nil { logger.Printf("Error: %v\n", err) } - os.Exit(exitCode) + // Stash exit code instead of calling os.Exit here. os.Exit skips + // deferred functions, including the OTEL shutdown in runExecWithStdin + // that flushes remaining log batches. Execute() calls os.Exit after + // cobra returns and all defers have run. + launcherExitCode = exitCode }, } +// launcherExitCode holds the subprocess exit code. Set by the run command, +// read by Execute() to exit after defers (including OTEL shutdown) complete. +var launcherExitCode int + func init() { rootCmd.AddCommand(runCmd) } @@ -130,7 +136,7 @@ func runExecWithStdin(args []string) (int, error) { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - otelShutdown, loggerProvider, err := setupOTELSDK(ctx) + otelShutdown, _, err := setupOTELSDK(ctx) if err != nil { return 1, err } @@ -139,11 +145,11 @@ func runExecWithStdin(args []string) (int, error) { err = errors.Join(err, otelShutdown(context.Background())) }() - return runExec(args, os.Stdin, loggerProvider) + return runExec(args, os.Stdin) } // runExec runs the specified command with arguments, injecting secrets as environment variables if specified -func runExec(args []string, stdinReader io.Reader, loggerProvider *log.LoggerProvider) (int, error) { +func runExec(args []string, stdinReader io.Reader) (int, error) { // Command and arguments cmdName := args[0] cmdArgs := []string{} @@ -253,23 +259,15 @@ func runExec(args []string, stdinReader io.Reader, loggerProvider *log.LoggerPro } }() - // Wait for the process to finish - err = cmd.Wait() - - // Wait for all output to be processed before returning - // This ensures logs are fully read and sent to OTEL before shutdown + // Wait for all output to be read before calling cmd.Wait(). + // cmd.Wait() closes stdout/stderr pipes, so readers must finish first. + // Once readers finish, all log records have been submitted to the OTEL + // batch processor. The deferred otelShutdown in runExecWithStdin flushes + // remaining batches before the process exits. wg.Wait() - // Force flush the OTEL pipeline to ensure all batched logs are exported - // This is especially important for short-lived jobs that fail quickly, where the - // batch processor may still have pending logs that haven't been exported yet - if loggerProvider != nil { - flushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if flushErr := loggerProvider.ForceFlush(flushCtx); flushErr != nil { - logger.Printf("Warning: failed to flush OTEL logs: %v\n", flushErr) - } - } + // Now that all output has been read, wait for the process to finish. + err = cmd.Wait() exitCode := cmd.ProcessState.ExitCode() if err != nil { diff --git a/services/core/jobs/jobs-launcher/cmd/run_test.go b/services/core/jobs/jobs-launcher/cmd/run_test.go index d4f5b8353e..f1e696a8aa 100644 --- a/services/core/jobs/jobs-launcher/cmd/run_test.go +++ b/services/core/jobs/jobs-launcher/cmd/run_test.go @@ -119,7 +119,7 @@ func TestRunExecWithStdinHelper(t *testing.T) { input := "test line 1\ntest line 2\n" var inputReader io.Reader = strings.NewReader(input) - exitCode, err := runExec([]string{"cat"}, inputReader, nil) + exitCode, err := runExec([]string{"cat"}, inputReader) if err != nil { t.Errorf("Unexpected error: %v", err) } @@ -224,10 +224,10 @@ func TestRunExecWithSecrets(t *testing.T) { if !tc.expectError { // Use sh to check that secrets are available in environment // This validates that secrets were actually injected - exitCode, err = runExec([]string{"sh", "-c", "env | grep -E '(TEST_SECRET|ANOTHER_SECRET)' || true"}, nil, nil) + exitCode, err = runExec([]string{"sh", "-c", "env | grep -E '(TEST_SECRET|ANOTHER_SECRET)' || true"}, nil) } else { // For error cases, use any simple command - exitCode, err = runExec([]string{"echo", "test"}, nil, nil) + exitCode, err = runExec([]string{"echo", "test"}, nil) } // Validate exit code @@ -267,7 +267,7 @@ func TestRunExecWithSecretsNotFound(t *testing.T) { os.Setenv("NMP_SECRETS_URL", mockServer.URL) os.Setenv("NMP_PRINCIPAL", `{"id":"test-principal"}`) - exitCode, err := runExec([]string{"echo", "test"}, nil, nil) + exitCode, err := runExec([]string{"echo", "test"}, nil) if exitCode != 1 { t.Errorf("Expected exit code 1 for secret not found, got %d", exitCode) @@ -295,7 +295,7 @@ func TestRunExecWithoutSecrets(t *testing.T) { }() // Should run normally without attempting secret fetching - exitCode, err := runExec([]string{"echo", "test without secrets"}, nil, nil) + exitCode, err := runExec([]string{"echo", "test without secrets"}, nil) if exitCode != 0 { t.Errorf("Expected exit code 0, got %d", exitCode) diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/common.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/common.py index 30f24e4991..a2b9a706d2 100644 --- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/common.py +++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/common.py @@ -991,7 +991,7 @@ def create_pod_template_spec( if step.step_spec.environment: for envvar in step.step_spec.environment: if envvar.value is not None: - # If the job has requested persistent job storage path, capture it for use when constructing the volume mount. + # Allow step to override the default persistent storage mount path. if envvar.name == PERSISTENT_JOB_STORAGE_PATH_ENVVAR: job_storage_mount = envvar.value @@ -1012,6 +1012,20 @@ def create_pod_template_spec( ), ] storage_config = config.storage + + # Persistent job storage (PVC mount) is only provisioned when the step + # explicitly declares NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH in its + # compile() environment. Jobs that don't declare it won't get a PVC + # mount — if they try to access ctx.storage.persistent at runtime, + # StoragePaths raises a clear RuntimeError guiding them to add it. + # + # TODO: Job authors should be able to declare whether they need + # persistent storage via a first-class field on the job spec (e.g. + # `requires_persistent_storage: bool` on NemoJob or PlatformJobStep), + # rather than the current mechanism of passing a magic env var in the + # step's environment list. This would make the contract between + # compile() and the runtime explicit. See AIRCORE-844 for context. + if storage_config.additional_volume_mounts: volume_mounts.extend(mount.to_k8s() for mount in storage_config.additional_volume_mounts)