From 38713d9ba33df752b7986c4d2cf9c04b859c9d4e Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Tue, 29 Sep 2026 12:52:54 +0000 Subject: [PATCH 1/8] Bound GPU CI capacity waits and verify resource cleanup --- .github/workflows/trainer-rank-gpu.yml | 29 ++- scripts/ci/trainer-rank-gpu.py | 215 ++++++++++++++++++++- tests/unit/test_trainer_rank_gpu_ci.py | 249 ++++++++++++++++++++++++- 3 files changed, 474 insertions(+), 19 deletions(-) diff --git a/.github/workflows/trainer-rank-gpu.yml b/.github/workflows/trainer-rank-gpu.yml index 088b53122..82099c780 100644 --- a/.github/workflows/trainer-rank-gpu.yml +++ b/.github/workflows/trainer-rank-gpu.yml @@ -31,7 +31,12 @@ jobs: head_sha: ${{ steps.changes.outputs.head_sha }} reused: ${{ steps.reuse.outputs.reused }} reused_run: ${{ steps.reuse.outputs.run_id }} + admission_deadline: ${{ steps.admission.outputs.deadline }} steps: + - id: admission + name: Bound GPU queue admission + run: echo "deadline=$(( $(date +%s) + 1800 ))" >> "${GITHUB_OUTPUT}" + - uses: actions/checkout@v4 with: ref: ${{ github.event.pull_request.head.sha || github.sha }} @@ -144,14 +149,23 @@ jobs: if: needs.classify.outputs.required == 'true' && needs.classify.outputs.reused != 'true' runs-on: ubuntu-latest timeout-minutes: 45 + concurrency: + group: trainer-rank-gpu-capacity-${{ inputs.context || 'cks-wb3' }} + cancel-in-progress: false + queue: max environment: trainer-rank-gpu-validation env: SKY_INFRA: k8s/${{ inputs.context || 'cks-wb3' }} + EXPECTED_HEAD_SHA: ${{ needs.classify.outputs.head_sha }} + ADMISSION_DEADLINE: ${{ needs.classify.outputs.admission_deadline }} steps: - uses: actions/checkout@v4 with: ref: ${{ needs.classify.outputs.head_sha }} + - name: Check admission deadline + run: python3 scripts/ci/trainer-rank-gpu.py admit "${RUNNER_TEMP}/trainer-rank-result" + - name: Install launch dependencies run: | curl -LsSf https://astral.sh/uv/install.sh | sh @@ -221,16 +235,23 @@ jobs: set -euo pipefail sky_venv="${RUNNER_TEMP}/skypilot" uv venv --python 3.10 "${sky_venv}" + # Preserve cancelled requests before retry callbacks can reschedule them. uv pip install --python "${sky_venv}/bin/python" \ - 'skypilot[kubernetes,remote]==0.12.0' 'kubernetes==34.1.0' + 'skypilot[kubernetes,remote] @ git+https://github.com/OpenPipe/skypilot-modal.git@9792e52a66dbe8c1c2636476103ec90654ac71e3' \ + 'kubernetes==34.1.0' sky="${sky_venv}/bin/sky" - cluster="trainer-rank-gpu-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" - cleanup() { "${sky}" down -y "${cluster}" || true; } - trap cleanup EXIT "${sky}" check kubernetes "${sky_venv}/bin/python" scripts/ci/trainer-rank-gpu.py \ "${RUNNER_TEMP}/trainer-rank-result" + - name: Reconcile owned GPU resources + if: always() + run: | + python="${RUNNER_TEMP}/skypilot/bin/python" + if [ ! -x "${python}" ]; then python=python3; fi + "${python}" scripts/ci/trainer-rank-gpu.py cleanup \ + "${RUNNER_TEMP}/trainer-rank-result" + - name: Preserve remote result and diagnostic logs if: always() uses: actions/upload-artifact@v4 diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index 2ff17ed78..688aab3cb 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -1,8 +1,8 @@ """Run the CI job without making an attached log stream its result gate. -SkyPilot 0.12 returns an actual job ID from launch. Each SDK operation runs in a -bounded child; only that job's SUCCEEDED status passes. The workflow EXIT trap -still owns cluster teardown, including launch failures before a job ID is known. +SkyPilot returns an actual job ID from launch. Each SDK operation runs in a +bounded child; only that job's SUCCEEDED status passes. An always-run workflow +step cancels pending admission and verifies cleanup of owned resources. """ import json @@ -12,9 +12,11 @@ import subprocess import sys import time +import uuid NONTERMINAL = {"INIT", "PENDING", "SETTING_UP", "RUNNING"} INFRAS = {"k8s/cks-wb3", "k8s/ext-collab2"} +LABEL = "art-ci-execution" EXIT_CODES = { "SUCCEEDED": 0, "FAILED": 100, @@ -42,15 +44,96 @@ def read_bound(root, filename, owner): return data +def launch_task(root, owner): + import sky + from sky.provision.kubernetes import utils as kube_utils + + context = owner["infra"].split("/", 1)[1] + write_json( + root / "allocation.json", + {**owner, "namespace": kube_utils.get_namespace(context=context)}, + ) + task = sky.Task.from_yaml("scripts/ci/trainer-rank-gpu.sky.yaml") + task.set_resources_override( + { + "infra": owner["infra"], + "_cluster_config_overrides": { + "kubernetes": {"custom_metadata": {"labels": {LABEL: owner["label"]}}} + }, + } + ) + return task + + +def resources(root, owner, *, delete): + """Observe/delete only this attempt's nonce-labelled Pods and Services.""" + from kubernetes import client, config + + allocation = read_bound(root, "allocation.json", owner) + namespace = allocation["namespace"] + connection = config.new_client_from_config(context=owner["infra"].split("/", 1)[1]) + api = client.CoreV1Api(connection) + retained = [] + try: + while True: + found = [] + for kind in ("pod", "service"): + items = getattr(api, f"list_namespaced_{kind}")( + namespace, + label_selector=f"{LABEL}={owner['label']}", + _request_timeout=(3, 10), + ).items + for item in items: + metadata = item.metadata + if ( + metadata.labels.get(LABEL) != owner["label"] + or metadata.namespace != namespace + or not metadata.uid + ): + raise ValueError("Kubernetes returned an unowned resource") + identity = { + "kind": kind, + "name": metadata.name, + "uid": metadata.uid, + } + found.append(identity) + if identity not in retained: + retained.append(identity) + write_json( + root / ("resources-cleanup.json" if delete else "resources.json"), + {**allocation, "observed": retained, "remaining": found}, + ) + if not delete or not found: + return + for item in found: + try: + getattr(api, f"delete_namespaced_{item['kind']}")( + item["name"], + namespace, + body=client.V1DeleteOptions( + preconditions=client.V1Preconditions(uid=item["uid"]) + ), + _request_timeout=(3, 10), + ) + except client.ApiException as error: + if error.status != 404: + raise + time.sleep(1) + finally: + connection.close() + + def worker(root, operation): import sky owner = json.loads((root / "owner.json").read_text()) cluster = owner["cluster"] if operation == "launch": - task = sky.Task.from_yaml("scripts/ci/trainer-rank-gpu.sky.yaml") - task.set_resources_override({"infra": owner["infra"]}) - request_id = sky.launch(task, cluster_name=cluster, retry_until_up=False) + task = launch_task(root, owner) + if time.time() >= min(owner["admission_deadline"], owner["work_deadline"]): + raise TimeoutError("CI launch deadline reached") + write_json(root / "launch-attempt.json", owner) + request_id = sky.launch(task, cluster_name=cluster, retry_until_up=True) try: write_json(root / "request.json", {**owner, "request_id": request_id}) job_id, handle = sky.get(request_id) @@ -75,6 +158,41 @@ def worker(root, operation): if operation == "cancel_request": request = read_bound(root, "request.json", owner)["request_id"] sky.get(sky.api_cancel(request_ids=[request])) + while True: + records = sky.api_status(request_ids=[request]) + if len(records) != 1 or records[0].request_id != request: + raise ValueError("Sky did not return the exact launch request") + status = records[0].status + if status in {"SUCCEEDED", "FAILED", "CANCELLED"}: + write_json( + root / "request-terminal.json", + {**owner, "request_id": request, "status": status}, + ) + break + time.sleep(1) + return + if operation == "down": + clusters = sky.get(sky.status([cluster])) + if any(item["name"] != cluster for item in clusters): + raise ValueError("Sky returned a different cluster") + if clusters: + sky.get(sky.down(cluster)) + return + if operation == "stop_api": + # This workflow owns the entire ephemeral runner and its local API. + # Stop its workers before the final physical census, including when a + # cancelled launch has not yet unwound. Never stop a developer's API. + if ( + os.environ.get("GITHUB_ACTIONS") != "true" + or os.environ.get("RUNNER_ENVIRONMENT") != "github-hosted" + ): + raise ValueError("API cleanup requires an isolated GitHub-hosted runner") + sky.api_stop() # The maintained SDK rejects remote API endpoints. + if sky.api_status(): + raise RuntimeError("The CI API still reports requests after stopping") + return + if operation in {"resources", "remove_resources"}: + resources(root, owner, delete=operation == "remove_resources") return job_id = read_bound(root, "job.json", owner)["job_id"] if operation == "status": @@ -198,7 +316,11 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): job = None errors = 0 try: - run_worker(root, "launch", deadline - time.monotonic()) + run_worker( + root, + "launch", + min(deadline - time.monotonic(), owner["admission_deadline"] - time.time()), + ) job = read_bound(root, "job.json", owner) while True: try: @@ -231,12 +353,21 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): root / "failure.json", {**owner, "remote_status": remote_status, "error": repr(error)}, ) + if job is None: + write_json( + root / "result.json", + { + **owner, + "status": "UNCONFIRMED", + "reason": "launch_result_unavailable", + }, + ) except Exception as receipt_error: print(f"Failure receipt also failed: {receipt_error}", file=sys.stderr) raise finally: # Also recover the exact ID if a receipt read failed after launch. - # No guessed/latest job is ever cancelled. The outer trap handles unknown ID. + # No guessed/latest job is cancelled. The workflow reconciles unknown IDs. if job is None: try: job = read_bound(root, "job.json", owner) @@ -259,7 +390,7 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): print(f"::warning::Diagnostic receipt failed: {error}", flush=True) -def main(root): +def admit(root): root.mkdir(parents=True, exist_ok=False) run_id, attempt = os.environ["GITHUB_RUN_ID"], os.environ["GITHUB_RUN_ATTEMPT"] if not run_id.isdecimal() or not attempt.isdecimal(): @@ -280,18 +411,82 @@ def main(root): } if owner["head"] != os.environ["EXPECTED_HEAD_SHA"]: raise ValueError("Checkout differs from classified CI head") + now = time.time() + admission_deadline = int(os.environ["ADMISSION_DEADLINE"]) + owner.update( + label=uuid.uuid4().hex, + admitted_at=now, + admission_deadline=admission_deadline, + work_deadline=now + 35 * 60, + cleanup_deadline=now + 40 * 60, + ) write_json(root / "owner.json", owner) + if now >= admission_deadline: + write_json( + root / "result.json", + {**owner, "status": "NOT_RUN", "reason": "admission_expired"}, + ) + print("::error::GPU validation did not run: its queue admission expired.") + return 104 + return 0 + + +def cleanup(root): + if not (root / "launch-attempt.json").exists(): + return 0 + owner = json.loads((root / "owner.json").read_text()) + + read_bound(root, "launch-attempt.json", owner) + outcomes = {} + # A failed/unknown cancellation must not prevent best-effort physical cleanup. + for operation, limit in [ + ("cancel_request", 30), + ("resources", 20), + ("down", 90), + ("stop_api", 30), + ("remove_resources", 90), + ]: + try: + run_worker( + root, operation, min(limit, owner["cleanup_deadline"] - time.time()) + ) + outcomes[operation] = {"success": True} + except BaseException as error: + outcomes[operation] = {"success": False, "error": repr(error)} + confirmed = all(outcome["success"] for outcome in outcomes.values()) + write_json( + root / "cleanup.json", {**owner, "confirmed": confirmed, "operations": outcomes} + ) + return 0 if confirmed else 105 + + +def main(root): + if not (root / "owner.json").exists(): + code = admit(root) + if code: + return code + owner = json.loads((root / "owner.json").read_text()) + if time.time() >= owner["admission_deadline"]: + write_json( + root / "result.json", + {**owner, "status": "NOT_RUN", "reason": "admission_expired"}, + ) + return 104 def interrupted(signum, frame): raise InterruptedError(f"CI interrupted by signal {signum}") for signum in (signal.SIGINT, signal.SIGTERM): signal.signal(signum, interrupted) - return supervise(root, owner) + return supervise(root, owner, timeout=max(0, owner["work_deadline"] - time.time())) if __name__ == "__main__": if sys.argv[1] == "worker": worker(Path(sys.argv[2]), sys.argv[3]) + elif sys.argv[1] == "admit": + sys.exit(admit(Path(sys.argv[2]))) + elif sys.argv[1] == "cleanup": + sys.exit(cleanup(Path(sys.argv[2]))) else: sys.exit(main(Path(sys.argv[1]))) diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 2cefc9ed3..02e3d7b75 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -28,6 +28,10 @@ def owner(tmp_path): "head": "a" * 40, "attempt": "2", "infra": "k8s/cks-wb3", + "label": "a" * 32, + "admission_deadline": time.time() + 1800, + "work_deadline": time.time() + 2100, + "cleanup_deadline": time.time() + 2400, } ci.write_json(tmp_path / "owner.json", value) return value @@ -43,9 +47,16 @@ def sky(monkeypatch): job_status=Mock(return_value="status-request"), cancel=Mock(return_value="cancel-request"), api_cancel=Mock(return_value="api-cancel-request"), + api_status=Mock(return_value=[NS(request_id="request-17", status="CANCELLED")]), + api_stop=Mock(), tail_logs=Mock(return_value=0), ) monkeypatch.setitem(sys.modules, "sky", api) + monkeypatch.setitem( + sys.modules, + "sky.provision.kubernetes", + NS(utils=NS(get_namespace=lambda **_: "default")), + ) return api @@ -65,9 +76,14 @@ def get(request): sky.get.side_effect = get ci.worker(tmp_path, "launch") task = sky.Task.from_yaml.return_value - task.set_resources_override.assert_called_once_with({"infra": infra}) + assert task.set_resources_override.call_args.args[0] == { + "infra": infra, + "_cluster_config_overrides": { + "kubernetes": {"custom_metadata": {"labels": {ci.LABEL: owner["label"]}}} + }, + } sky.launch.assert_called_once_with( - task, cluster_name=owner["cluster"], retry_until_up=False + task, cluster_name=owner["cluster"], retry_until_up=True ) assert ci.read_bound(tmp_path, "job.json", owner)["job_id"] == 17 sky.tail_logs.assert_not_called() @@ -279,7 +295,14 @@ def test_launch_receipt_failure_cancels_request_and_preserves_error( tmp_path, owner, sky, monkeypatch ): primary = OSError("disk full") - monkeypatch.setattr(ci, "write_json", Mock(side_effect=primary)) + write = ci.write_json + + def fail_receipt(path, data): + if path.name == "request.json": + raise primary + write(path, data) + + monkeypatch.setattr(ci, "write_json", fail_receipt) sky.api_cancel.side_effect = OSError("cancel failed") with pytest.raises(OSError) as raised: ci.worker(tmp_path, "launch") @@ -313,6 +336,23 @@ def test_launch_timeout_cancels_only_recorded_request( ci.supervise(tmp_path, owner) assert raised.value is error assert calls == (["launch", "cancel_request"] if known_request else ["launch"]) + # A lost launch reply does not prove the remote test never started. + assert ci.read_bound(tmp_path, "result.json", owner)["status"] == "UNCONFIRMED" + + +def test_capacity_wait_uses_original_admission_expiry(tmp_path, owner, monkeypatch): + owner["admission_deadline"] = 107 + monkeypatch.setattr(ci.time, "time", lambda: 100) + calls = [] + + def run(root, operation, timeout): + calls.append((operation, timeout)) + raise TimeoutError("capacity wait expired") + + monkeypatch.setattr(ci, "run_worker", run) + with pytest.raises(TimeoutError): + ci.supervise(tmp_path, owner, timeout=2100) + assert calls == [("launch", 7)] @pytest.mark.parametrize( @@ -350,14 +390,16 @@ def test_real_worker_round_trip_with_fake_sdk(tmp_path, infra): """Exercise the actual direct Python parent/worker JSON transport, without Sky.""" (tmp_path / "sky.py").write_text(""" import os +import sys from pathlib import Path from types import SimpleNamespace as NS +sys.modules["sky.provision.kubernetes"] = NS(utils=NS(get_namespace=lambda **_: "default")) class Task: @classmethod def from_yaml(cls, path): return cls() - def set_resources_override(self, options): assert options == {"infra":os.environ["SKY_INFRA"]} + def set_resources_override(self, options): assert options["infra"] == os.environ["SKY_INFRA"] def launch(task, *, cluster_name, retry_until_up): - assert retry_until_up is False + assert retry_until_up is True Path(os.environ["FAKE_CLUSTER"]).write_text(cluster_name) return "launch-request" def get(value): @@ -386,6 +428,7 @@ def tail_logs(cluster, *, job_id, follow): "GITHUB_EVENT_NAME": "pull_request", "SKY_INFRA": infra, "EXPECTED_HEAD_SHA": head, + "ADMISSION_DEADLINE": str(int(time.time()) + 1800), } root = tmp_path / "evidence" result = subprocess.run( @@ -541,3 +584,199 @@ def interrupted(signum, frame): if process.poll() is None: os.killpg(process.pid, signal.SIGKILL) process.wait(timeout=2) + + +def admission_environment(monkeypatch, *, deadline): + for key, value in { + "GITHUB_RUN_ID": "42", + "GITHUB_RUN_ATTEMPT": "2", + "GITHUB_REPOSITORY": "OpenPipe/ART", + "GITHUB_EVENT_NAME": "pull_request", + "EXPECTED_HEAD_SHA": "a" * 40, + "SKY_INFRA": "k8s/cks-wb3", + "ADMISSION_DEADLINE": str(deadline), + }.items(): + monkeypatch.setenv(key, value) + monkeypatch.setattr(ci.subprocess, "check_output", lambda *a, **kw: "a" * 40) + + +def test_expired_queue_admission_cannot_later_launch(tmp_path, monkeypatch): + admission_environment(monkeypatch, deadline=100) + monkeypatch.setattr(ci.time, "time", lambda: 101) + launch = Mock() + monkeypatch.setattr(ci, "supervise", launch) + root = tmp_path / "expired" + assert ci.admit(root) == 104 + assert ci.main(root) == 104 + assert ci.cleanup(root) == 0 + launch.assert_not_called() + result = json.loads((root / "result.json").read_text()) + assert result["status"] == "NOT_RUN" and result["reason"] == "admission_expired" + + +def test_setup_consumes_original_admitted_work_window(tmp_path, monkeypatch): + admission_environment(monkeypatch, deadline=1300) + now = [100] + monkeypatch.setattr(ci.time, "time", lambda: now[0]) + root = tmp_path / "admitted" + assert ci.admit(root) == 0 + now[0] = 400 + run = Mock(return_value=0) + monkeypatch.setattr(ci, "supervise", run) + assert ci.main(root) == 0 + assert run.call_args.kwargs["timeout"] == 1800 + assert json.loads((root / "owner.json").read_text())["cleanup_deadline"] == 2500 + + +def test_admission_expiring_during_setup_does_not_launch(tmp_path, monkeypatch): + admission_environment(monkeypatch, deadline=300) + now = [100] + monkeypatch.setattr(ci.time, "time", lambda: now[0]) + root = tmp_path / "admitted" + assert ci.admit(root) == 0 + now[0] = 301 + run = Mock() + monkeypatch.setattr(ci, "supervise", run) + assert ci.main(root) == 104 + run.assert_not_called() + assert json.loads((root / "result.json").read_text())["status"] == "NOT_RUN" + + +@pytest.mark.parametrize("deadline", ["admission_deadline", "work_deadline"]) +def test_worker_does_not_submit_after_cutoff( + tmp_path, owner, sky, monkeypatch, deadline +): + monkeypatch.setattr(ci.time, "time", lambda: owner[deadline]) + with pytest.raises(TimeoutError, match="launch deadline"): + ci.worker(tmp_path, "launch") + sky.launch.assert_not_called() + assert not (tmp_path / "launch-attempt.json").exists() + + +def test_request_cancellation_waits_for_exact_terminal_state( + tmp_path, owner, sky, monkeypatch +): + ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request-17"}) + sky.api_status.side_effect = [ + [NS(request_id="request-17", status="PENDING")], + [NS(request_id="request-17", status="RUNNING")], + [NS(request_id="request-17", status="CANCELLED")], + ] + monkeypatch.setattr(ci.time, "sleep", lambda _: None) + ci.worker(tmp_path, "cancel_request") + assert sky.api_status.call_count == 3 + assert ( + ci.read_bound(tmp_path, "request-terminal.json", owner)["status"] == "CANCELLED" + ) + sky.launch.assert_not_called() + + +@pytest.mark.parametrize("records", [[], [NS(request_id="peer", status="CANCELLED")]]) +def test_request_cancellation_does_not_claim_unknown_terminal( + tmp_path, owner, sky, records +): + ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request-17"}) + sky.api_status.return_value = records + with pytest.raises(ValueError, match="exact launch request"): + ci.worker(tmp_path, "cancel_request") + assert not (tmp_path / "request-terminal.json").exists() + + +def test_cleanup_is_finite_and_reports_unknown_cancel(tmp_path, owner, monkeypatch): + ci.write_json(tmp_path / "launch-attempt.json", owner) + now = [owner["cleanup_deadline"] - 5] + monkeypatch.setattr(ci.time, "time", lambda: now[0]) + calls = [] + + def run(root, operation, timeout): + calls.append((operation, timeout)) + now[0] += 1 + if operation == "cancel_request": + raise TimeoutError("request state unknown") + + monkeypatch.setattr(ci, "run_worker", run) + assert ci.cleanup(tmp_path) == 105 + assert calls == [ + ("cancel_request", 5), + ("resources", 4), + ("down", 3), + ("stop_api", 2), + ("remove_resources", 1), + ] + receipt = ci.read_bound(tmp_path, "cleanup.json", owner) + assert receipt["confirmed"] is False + assert receipt["operations"]["remove_resources"]["success"] is True + + +def test_api_cleanup_is_limited_to_its_ephemeral_runner( + tmp_path, owner, sky, monkeypatch +): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("RUNNER_ENVIRONMENT", "self-hosted") + with pytest.raises(ValueError, match="isolated GitHub-hosted"): + ci.worker(tmp_path, "stop_api") + sky.api_stop.assert_not_called() + monkeypatch.setenv("RUNNER_ENVIRONMENT", "github-hosted") + sky.api_status.return_value = [] + ci.worker(tmp_path, "stop_api") + sky.api_stop.assert_called_once_with() + + +@pytest.mark.parametrize("foreign", [False, True]) +def test_cleanup_resources_records_uids_and_confines_deletion( + tmp_path, owner, monkeypatch, foreign +): + ci.write_json(tmp_path / "allocation.json", {**owner, "namespace": "ci"}) + pod = NS( + metadata=NS( + name="pod", + uid="original-uid", + namespace="ci", + labels={ci.LABEL: "peer" if foreign else owner["label"]}, + ) + ) + api = NS( + list_namespaced_pod=Mock(side_effect=[NS(items=[pod]), NS(items=[])]), + list_namespaced_service=Mock(return_value=NS(items=[])), + delete_namespaced_pod=Mock(), + ) + connection = NS(close=Mock()) + + class ApiException(Exception): + pass + + client = NS( + CoreV1Api=lambda c: api, + V1DeleteOptions=NS, + V1Preconditions=NS, + ApiException=ApiException, + ) + config = NS(new_client_from_config=Mock(return_value=connection)) + monkeypatch.setitem(sys.modules, "kubernetes", NS(client=client, config=config)) + monkeypatch.setattr(ci.time, "sleep", lambda _: None) + if foreign: + with pytest.raises(ValueError, match="unowned"): + ci.resources(tmp_path, owner, delete=True) + api.delete_namespaced_pod.assert_not_called() + else: + + def delete(name, namespace, *, body, _request_timeout): + assert (name, namespace, body.preconditions.uid) == ( + "pod", + "ci", + "original-uid", + ) + recorded = ci.read_bound(tmp_path, "resources-cleanup.json", owner) + assert recorded["remaining"][0]["uid"] == "original-uid" + + api.delete_namespaced_pod.side_effect = delete + ci.resources(tmp_path, owner, delete=True) + assert ( + ci.read_bound(tmp_path, "resources-cleanup.json", owner)["remaining"] == [] + ) + api.list_namespaced_pod.assert_called_with( + "ci", + label_selector=f"{ci.LABEL}={owner['label']}", + _request_timeout=(3, 10), + ) + connection.close.assert_called_once() From 446c667f95f0105dfa3b52e54df117c55fb538dc Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Tue, 29 Sep 2026 13:04:48 +0000 Subject: [PATCH 2/8] Clarify unsubmitted CI outcomes and pin merged Sky retry fix --- .github/workflows/trainer-rank-gpu.yml | 2 +- scripts/ci/trainer-rank-gpu.py | 9 ++++++--- tests/unit/test_trainer_rank_gpu_ci.py | 26 +++++++++++++++++++++++--- 3 files changed, 30 insertions(+), 7 deletions(-) diff --git a/.github/workflows/trainer-rank-gpu.yml b/.github/workflows/trainer-rank-gpu.yml index 82099c780..614636c3f 100644 --- a/.github/workflows/trainer-rank-gpu.yml +++ b/.github/workflows/trainer-rank-gpu.yml @@ -237,7 +237,7 @@ jobs: uv venv --python 3.10 "${sky_venv}" # Preserve cancelled requests before retry callbacks can reschedule them. uv pip install --python "${sky_venv}/bin/python" \ - 'skypilot[kubernetes,remote] @ git+https://github.com/OpenPipe/skypilot-modal.git@9792e52a66dbe8c1c2636476103ec90654ac71e3' \ + 'skypilot[kubernetes,remote] @ git+https://github.com/OpenPipe/skypilot-modal.git@924100745348cb0d880ba9e1f64b4e54cb1c0898' \ 'kubernetes==34.1.0' sky="${sky_venv}/bin/sky" "${sky}" check kubernetes diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index 688aab3cb..057429b9b 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -188,7 +188,7 @@ def worker(root, operation): ): raise ValueError("API cleanup requires an isolated GitHub-hosted runner") sky.api_stop() # The maintained SDK rejects remote API endpoints. - if sky.api_status(): + if sky.api_status(all_status=True, limit=1): raise RuntimeError("The CI API still reports requests after stopping") return if operation in {"resources", "remove_resources"}: @@ -354,12 +354,15 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): {**owner, "remote_status": remote_status, "error": repr(error)}, ) if job is None: + attempted = (root / "launch-attempt.json").exists() write_json( root / "result.json", { **owner, - "status": "UNCONFIRMED", - "reason": "launch_result_unavailable", + "status": "UNCONFIRMED" if attempted else "NOT_RUN", + "reason": "launch_result_unavailable" + if attempted + else "not_submitted", }, ) except Exception as receipt_error: diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 02e3d7b75..1a9ecc610 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -324,10 +324,14 @@ def test_cancel_request_requires_own_receipt(tmp_path, owner, sky): sky.api_cancel.assert_not_called() -@pytest.mark.parametrize("known_request", [False, True]) +@pytest.mark.parametrize( + "attempted,known_request", [(False, False), (True, False), (True, True)] +) def test_launch_timeout_cancels_only_recorded_request( - tmp_path, owner, monkeypatch, known_request + tmp_path, owner, monkeypatch, attempted, known_request ): + if attempted: + ci.write_json(tmp_path / "launch-attempt.json", owner) if known_request: ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request-17"}) error = TimeoutError("launch wait") @@ -337,7 +341,9 @@ def test_launch_timeout_cancels_only_recorded_request( assert raised.value is error assert calls == (["launch", "cancel_request"] if known_request else ["launch"]) # A lost launch reply does not prove the remote test never started. - assert ci.read_bound(tmp_path, "result.json", owner)["status"] == "UNCONFIRMED" + assert ci.read_bound(tmp_path, "result.json", owner)["status"] == ( + "UNCONFIRMED" if attempted else "NOT_RUN" + ) def test_capacity_wait_uses_original_admission_expiry(tmp_path, owner, monkeypatch): @@ -722,6 +728,20 @@ def test_api_cleanup_is_limited_to_its_ephemeral_runner( sky.api_stop.assert_called_once_with() +def test_api_cleanup_detects_surviving_server_with_only_terminal_requests( + tmp_path, owner, sky, monkeypatch +): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("RUNNER_ENVIRONMENT", "github-hosted") + sky.api_status.side_effect = lambda **options: ( + [NS(request_id="request-17", status="CANCELLED")] + if options.get("all_status") + else [] + ) + with pytest.raises(RuntimeError, match="still reports requests"): + ci.worker(tmp_path, "stop_api") + + @pytest.mark.parametrize("foreign", [False, True]) def test_cleanup_resources_records_uids_and_confines_deletion( tmp_path, owner, monkeypatch, foreign From 9f17d81fd004ba5768deabcaabee4806e417e2e2 Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 01:49:38 +0000 Subject: [PATCH 3/8] fix(ci): retain physical resource receipts before teardown --- scripts/ci/trainer-rank-gpu.py | 86 ++++++++++- tests/unit/test_trainer_rank_gpu_ci.py | 204 ++++++++++++++++++++++++- 2 files changed, 282 insertions(+), 8 deletions(-) diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index 057429b9b..e9b6716df 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -53,6 +53,10 @@ def launch_task(root, owner): root / "allocation.json", {**owner, "namespace": kube_utils.get_namespace(context=context)}, ) + write_json( + root / "cleanup.json", + {**owner, "confirmed": False, "physical_absence": "UNKNOWN"}, + ) task = sky.Task.from_yaml("scripts/ci/trainer-rank-gpu.sky.yaml") task.set_resources_override( { @@ -71,9 +75,21 @@ def resources(root, owner, *, delete): allocation = read_bound(root, "allocation.json", owner) namespace = allocation["namespace"] + retained = ( + read_bound(root, "resources.json", allocation)["observed"] + if (root / "resources.json").exists() + else [] + ) + target = root / ("resources-cleanup.json" if delete else "resources.json") + receipt = { + **allocation, + "observed": retained, + "remaining": None, + "exact_uid_absence": "UNKNOWN", + } + write_json(target, receipt) connection = config.new_client_from_config(context=owner["infra"].split("/", 1)[1]) api = client.CoreV1Api(connection) - retained = [] try: while True: found = [] @@ -95,14 +111,36 @@ def resources(root, owner, *, delete): "kind": kind, "name": metadata.name, "uid": metadata.uid, + "cloud_name": metadata.labels.get("skypilot-cluster-name"), } found.append(identity) if identity not in retained: retained.append(identity) - write_json( - root / ("resources-cleanup.json" if delete else "resources.json"), - {**allocation, "observed": retained, "remaining": found}, - ) + # Retain each UID even if the next resource query fails. + write_json(target, receipt) + receipt["remaining"] = found + write_json(target, receipt) + if delete and not found: + # A removed nonce label is not proof that the original UID is gone. + for item in retained: + try: + current = getattr(api, f"read_namespaced_{item['kind']}")( + item["name"], namespace, _request_timeout=(3, 10) + ).metadata + except client.ApiException as error: + if error.status != 404: + raise + else: + if ( + current.name != item["name"] + or current.namespace != namespace + or not current.uid + or current.uid == item["uid"] + ): + raise ValueError("Original resource absence is unproven") + if retained: + receipt["exact_uid_absence"] = "ABSENT" + write_json(target, receipt) if not delete or not found: return for item in found: @@ -145,6 +183,16 @@ def worker(root, operation): raise ValueError( "Sky launch returned a different cluster or invalid job ID" ) + cloud_name = getattr(handle, "cluster_name_on_cloud", None) + write_json( + root / "physical.json", + { + **read_bound(root, "allocation.json", owner), + "cloud_name": cloud_name if isinstance(cloud_name, str) else None, + "source": "sky.launch.handle", + "namespace_source": "prelaunch_context_configuration", + }, + ) write_json(root / "job.json", {**owner, "job_id": job_id}) except BaseException: try: @@ -322,6 +370,10 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): min(deadline - time.monotonic(), owner["admission_deadline"] - time.time()), ) job = read_bound(root, "job.json", owner) + try: + run_worker(root, "resources", min(20, deadline - time.monotonic())) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as error: + print(f"::warning::Initial resource receipt failed: {error}", flush=True) while True: try: run_worker(root, "status", min(30, deadline - time.monotonic())) @@ -379,6 +431,9 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): operations = (["cancel"] if not terminal else []) + ["logs"] if job else [] if job is None and (root / "request.json").exists(): operations = ["cancel_request"] + if not terminal and (root / "allocation.json").exists(): + # Cancellation may remove pending Pods before cleanup can observe them. + operations.insert(0, "resources") outcomes = {} for operation in operations: try: @@ -457,8 +512,27 @@ def cleanup(root): except BaseException as error: outcomes[operation] = {"success": False, "error": repr(error)} confirmed = all(outcome["success"] for outcome in outcomes.values()) + physical_absence = "UNKNOWN" + if outcomes["stop_api"]["success"] and outcomes["remove_resources"]["success"]: + try: + evidence = read_bound(root, "resources-cleanup.json", owner) + if ( + evidence.get("observed") + and evidence.get("remaining") == [] + and evidence.get("exact_uid_absence") == "ABSENT" + ): + physical_absence = "ABSENT" + except (OSError, ValueError, KeyError): + pass write_json( - root / "cleanup.json", {**owner, "confirmed": confirmed, "operations": outcomes} + root / "cleanup.json", + { + **owner, + "confirmed": confirmed, + "physical_absence": physical_absence, + "physical_absence_scope": "retained Pod and Service UIDs", + "operations": outcomes, + }, ) return 0 if confirmed else 105 diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 1a9ecc610..01e32d992 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -68,10 +68,20 @@ def test_submit_once_records_request_before_wait_and_actual_job( ci.write_json(tmp_path / "owner.json", owner) monkeypatch.setenv("SKY_INFRA", "k8s/unapproved") # Use recorded ownership. + def launch(*args, **kwargs): + receipt = ci.read_bound(tmp_path, "cleanup.json", owner) + assert receipt["physical_absence"] == "UNKNOWN" + assert receipt["confirmed"] is False + return "request-17" + + sky.launch.side_effect = launch + def get(request): assert request == "request-17" assert ci.read_bound(tmp_path, "request.json", owner)["request_id"] == request - return 17, NS(cluster_name=owner["cluster"]) + return 17, NS( + cluster_name=owner["cluster"], cluster_name_on_cloud="physical-suffix" + ) sky.get.side_effect = get ci.worker(tmp_path, "launch") @@ -86,6 +96,12 @@ def get(request): task, cluster_name=owner["cluster"], retry_until_up=True ) assert ci.read_bound(tmp_path, "job.json", owner)["job_id"] == 17 + assert ci.read_bound(tmp_path, "physical.json", owner)["cloud_name"] == ( + "physical-suffix" + ) + assert ci.read_bound(tmp_path, "cleanup.json", owner)["physical_absence"] == ( + "UNKNOWN" + ) sky.tail_logs.assert_not_called() @@ -247,7 +263,7 @@ def fail_read(root, filename, expected): def test_late_success_does_not_bypass_deadline(tmp_path, owner, monkeypatch): calls = scenario(monkeypatch, tmp_path, owner, ["SUCCEEDED"]) - clock = iter([0, 0, 0, 2]) + clock = iter([0, 0, 0, 0, 2]) monkeypatch.setattr(ci.time, "monotonic", lambda: next(clock)) with pytest.raises(TimeoutError): ci.supervise(tmp_path, owner, timeout=1) @@ -759,6 +775,9 @@ def test_cleanup_resources_records_uids_and_confines_deletion( list_namespaced_pod=Mock(side_effect=[NS(items=[pod]), NS(items=[])]), list_namespaced_service=Mock(return_value=NS(items=[])), delete_namespaced_pod=Mock(), + read_namespaced_pod=Mock( + return_value=NS(metadata=NS(name="pod", namespace="ci", uid="replacement")) + ), ) connection = NS(close=Mock()) @@ -800,3 +819,184 @@ def delete(name, namespace, *, body, _request_timeout): _request_timeout=(3, 10), ) connection.close.assert_called_once() + + +def test_handle_is_retained_before_job_receipt_failure( + tmp_path, owner, sky, monkeypatch +): + sky.get.return_value = ( + 17, + NS(cluster_name=owner["cluster"], cluster_name_on_cloud="actual-cloud-suffix"), + ) + write = ci.write_json + + def fail_job(path, data): + if path.name == "job.json": + raise OSError("interrupted after handle") + write(path, data) + + monkeypatch.setattr(ci, "write_json", fail_job) + with pytest.raises(OSError, match="after handle"): + ci.worker(tmp_path, "launch") + physical = ci.read_bound(tmp_path, "physical.json", owner) + assert physical["cloud_name"] == "actual-cloud-suffix" + assert physical["namespace"] == "default" + assert ( + ci.read_bound(tmp_path, "cleanup.json", owner)["physical_absence"] == "UNKNOWN" + ) + + +def test_initial_identity_capture_precedes_job_polling(tmp_path, owner, monkeypatch): + calls = scenario(monkeypatch, tmp_path, owner, ["SUCCEEDED"]) + assert ci.supervise(tmp_path, owner, poll_seconds=0) == 0 + assert calls[:3] == ["launch", "resources", "status"] + + +def test_initial_identity_failure_preserves_job_result(tmp_path, owner, monkeypatch): + scenario( + monkeypatch, + tmp_path, + owner, + ["SUCCEEDED"], + failures={"resources": subprocess.TimeoutExpired("resources", 20)}, + ) + assert ci.supervise(tmp_path, owner, poll_seconds=0) == 0 + + +def test_interrupted_provisioning_captures_ids_before_cancelling( + tmp_path, owner, monkeypatch +): + ci.write_json(tmp_path / "allocation.json", {**owner, "namespace": "ci"}) + ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request-17"}) + calls = scenario( + monkeypatch, tmp_path, owner, [], failures={"launch": TimeoutError("launch")} + ) + with pytest.raises(TimeoutError): + ci.supervise(tmp_path, owner) + assert calls == ["launch", "resources", "cancel_request"] + + +@pytest.fixture +def kube_receipts(tmp_path, owner, monkeypatch): + ci.write_json(tmp_path / "allocation.json", {**owner, "namespace": "ci"}) + pod = NS( + metadata=NS( + name="physical-head", + uid="original-uid", + namespace="ci", + labels={ + ci.LABEL: owner["label"], + "skypilot-cluster-name": "physical-suffix", + }, + ) + ) + + class ApiException(Exception): + def __init__(self, status): + self.status = status + + api = NS( + list_namespaced_pod=Mock(return_value=NS(items=[pod])), + list_namespaced_service=Mock(return_value=NS(items=[])), + read_namespaced_pod=Mock(side_effect=ApiException(404)), + delete_namespaced_pod=Mock(), + ) + connection = NS(close=Mock()) + monkeypatch.setitem( + sys.modules, + "kubernetes", + NS( + client=NS( + CoreV1Api=lambda _: api, + ApiException=ApiException, + V1DeleteOptions=NS, + V1Preconditions=NS, + ), + config=NS(new_client_from_config=lambda **_: connection), + ), + ) + return api, pod, ApiException + + +def test_partial_resource_query_retains_exact_pod(tmp_path, owner, kube_receipts): + api, pod, _ = kube_receipts + api.list_namespaced_service.side_effect = OSError("service query unavailable") + with pytest.raises(OSError): + ci.resources(tmp_path, owner, delete=False) + receipt = ci.read_bound(tmp_path, "resources.json", owner) + assert receipt["observed"][0]["uid"] == pod.metadata.uid + assert receipt["observed"][0]["cloud_name"] == "physical-suffix" + assert receipt["remaining"] is None + assert receipt["exact_uid_absence"] == "UNKNOWN" + + +@pytest.mark.parametrize("state", ["absent", "replaced", "relabelled", "forbidden"]) +def test_exact_uid_absence_requires_named_query(tmp_path, owner, kube_receipts, state): + api, pod, error = kube_receipts + ci.resources(tmp_path, owner, delete=False) + api.list_namespaced_pod.return_value = NS(items=[]) + # A later empty pre-cleanup query must not discard the live-job receipt. + ci.resources(tmp_path, owner, delete=False) + if state in {"replaced", "relabelled"}: + api.read_namespaced_pod.side_effect = None + api.read_namespaced_pod.return_value = NS( + metadata=NS( + name=pod.metadata.name, + namespace="ci", + uid="new-uid" if state == "replaced" else pod.metadata.uid, + ) + ) + elif state == "forbidden": + api.read_namespaced_pod.side_effect = error(403) + if state in {"relabelled", "forbidden"}: + with pytest.raises((ValueError, error)): + ci.resources(tmp_path, owner, delete=True) + else: + ci.resources(tmp_path, owner, delete=True) + receipt = ci.read_bound(tmp_path, "resources-cleanup.json", owner) + assert receipt["exact_uid_absence"] == ( + "ABSENT" if state in {"absent", "replaced"} else "UNKNOWN" + ) + api.read_namespaced_pod.assert_called_once_with( + "physical-head", "ci", _request_timeout=(3, 10) + ) + api.delete_namespaced_pod.assert_not_called() + + +def test_empty_resource_history_is_unknown(tmp_path, owner, kube_receipts): + api, _, _ = kube_receipts + api.list_namespaced_pod.return_value = NS(items=[]) + ci.resources(tmp_path, owner, delete=True) + assert ( + ci.read_bound(tmp_path, "resources-cleanup.json", owner)["exact_uid_absence"] + == "UNKNOWN" + ) + + +@pytest.mark.parametrize("stop_api", [True, False]) +@pytest.mark.parametrize("receipt", [True, False]) +def test_cleanup_absence_requires_stopped_creator_and_uid_receipt( + tmp_path, owner, monkeypatch, stop_api, receipt +): + ci.write_json(tmp_path / "launch-attempt.json", owner) + if receipt: + ci.write_json( + tmp_path / "resources-cleanup.json", + { + **owner, + "observed": [{"kind": "pod", "name": "pod", "uid": "original"}], + "remaining": [], + "exact_uid_absence": "ABSENT", + }, + ) + + def run(root, operation, timeout): + if operation == "stop_api" and not stop_api: + raise TimeoutError("creator still running") + + monkeypatch.setattr(ci, "run_worker", run) + assert ci.cleanup(tmp_path) == (0 if stop_api else 105) + assert ci.read_bound(tmp_path, "cleanup.json", owner)["physical_absence"] == ( + "ABSENT" if stop_api and receipt else "UNKNOWN" + ) From 423adbc17b12b5559f63b8dd8c7744a45edda884 Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 02:29:08 +0000 Subject: [PATCH 4/8] fix(ci): cancel before diagnostics and scope cleanup receipts --- .github/workflows/trainer-rank-gpu.yml | 3 +- scripts/ci/trainer-rank-gpu.py | 34 ++++++++----- tests/unit/test_trainer_rank_gpu_ci.py | 67 +++++++++++++++++++++----- 3 files changed, 79 insertions(+), 25 deletions(-) diff --git a/.github/workflows/trainer-rank-gpu.yml b/.github/workflows/trainer-rank-gpu.yml index a10aea9b9..04f43b5b3 100644 --- a/.github/workflows/trainer-rank-gpu.yml +++ b/.github/workflows/trainer-rank-gpu.yml @@ -34,7 +34,8 @@ jobs: admission_deadline: ${{ steps.admission.outputs.deadline }} steps: - id: admission - name: Bound GPU queue admission + name: Bound GPU client admission wait + # Preserve this classification's cutoff; rerun all jobs for a fresh window. run: echo "deadline=$(( $(date +%s) + 1800 ))" >> "${GITHUB_OUTPUT}" - uses: actions/checkout@v4 diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index e9b6716df..23d0a0223 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -2,7 +2,9 @@ SkyPilot returns an actual job ID from launch. Each SDK operation runs in a bounded child; only that job's SUCCEEDED status passes. An always-run workflow -step cancels pending admission and verifies cleanup of owned resources. +step requests cancellation and checks retained resource UIDs. The admission +deadline bounds the client wait, not provider creation. Successful API shutdown +does not prove creator retirement; cleanup receipts keep that state UNKNOWN. """ import json @@ -55,7 +57,12 @@ def launch_task(root, owner): ) write_json( root / "cleanup.json", - {**owner, "confirmed": False, "physical_absence": "UNKNOWN"}, + { + **owner, + "operations_succeeded": False, + "physical_absence": "UNKNOWN", + "creator_quiescence": "UNKNOWN", + }, ) task = sky.Task.from_yaml("scripts/ci/trainer-rank-gpu.sky.yaml") task.set_resources_override( @@ -228,8 +235,8 @@ def worker(root, operation): return if operation == "stop_api": # This workflow owns the entire ephemeral runner and its local API. - # Stop its workers before the final physical census, including when a - # cancelled launch has not yet unwound. Never stop a developer's API. + # Best-effort shutdown: the SDK snapshots descendants before killing + # the parent, so an absent API process does not prove creator retirement. if ( os.environ.get("GITHUB_ACTIONS") != "true" or os.environ.get("RUNNER_ENVIRONMENT") != "github-hosted" @@ -432,8 +439,8 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): if job is None and (root / "request.json").exists(): operations = ["cancel_request"] if not terminal and (root / "allocation.json").exists(): - # Cancellation may remove pending Pods before cleanup can observe them. - operations.insert(0, "resources") + # Request cancellation before diagnostics can prolong capacity wait. + operations.insert(1, "resources") outcomes = {} for operation in operations: try: @@ -475,6 +482,7 @@ def admit(root): label=uuid.uuid4().hex, admitted_at=now, admission_deadline=admission_deadline, + admission_deadline_scope="client_launch_wait", work_deadline=now + 35 * 60, cleanup_deadline=now + 40 * 60, ) @@ -484,7 +492,10 @@ def admit(root): root / "result.json", {**owner, "status": "NOT_RUN", "reason": "admission_expired"}, ) - print("::error::GPU validation did not run: its queue admission expired.") + print( + "::error::GPU validation did not run: its admission window expired. " + "Rerun all workflow jobs for a fresh window." + ) return 104 return 0 @@ -511,9 +522,9 @@ def cleanup(root): outcomes[operation] = {"success": True} except BaseException as error: outcomes[operation] = {"success": False, "error": repr(error)} - confirmed = all(outcome["success"] for outcome in outcomes.values()) + operations_succeeded = all(outcome["success"] for outcome in outcomes.values()) physical_absence = "UNKNOWN" - if outcomes["stop_api"]["success"] and outcomes["remove_resources"]["success"]: + if outcomes["remove_resources"]["success"]: try: evidence = read_bound(root, "resources-cleanup.json", owner) if ( @@ -528,13 +539,14 @@ def cleanup(root): root / "cleanup.json", { **owner, - "confirmed": confirmed, + "operations_succeeded": operations_succeeded, "physical_absence": physical_absence, "physical_absence_scope": "retained Pod and Service UIDs", + "creator_quiescence": "UNKNOWN", "operations": outcomes, }, ) - return 0 if confirmed else 105 + return 0 if operations_succeeded else 105 def main(root): diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 1d46909ea..d7e954ad3 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -44,6 +44,8 @@ def sky(monkeypatch): Task=NS(from_yaml=Mock(return_value=task)), launch=Mock(return_value="request-17"), get=Mock(), + status=Mock(return_value="cluster-status-request"), + down=Mock(return_value="down-request"), job_status=Mock(return_value="status-request"), cancel=Mock(return_value="cancel-request"), api_cancel=Mock(return_value="api-cancel-request"), @@ -71,7 +73,8 @@ def test_submit_once_records_request_before_wait_and_actual_job( def launch(*args, **kwargs): receipt = ci.read_bound(tmp_path, "cleanup.json", owner) assert receipt["physical_absence"] == "UNKNOWN" - assert receipt["confirmed"] is False + assert receipt["operations_succeeded"] is False + assert receipt["creator_quiescence"] == "UNKNOWN" return "request-17" sky.launch.side_effect = launch @@ -150,6 +153,25 @@ def test_cancel_and_logs_use_exact_job_and_never_follow(tmp_path, owner, sky): sky.tail_logs.assert_called_once_with(owner["cluster"], job_id=17, follow=False) +@pytest.mark.parametrize("present", [False, True]) +def test_down_checks_exact_cluster_before_teardown(tmp_path, owner, sky, present): + sky.get.side_effect = [[{"name": owner["cluster"]}] if present else [], None] + ci.worker(tmp_path, "down") + sky.status.assert_called_once_with([owner["cluster"]]) + if present: + sky.down.assert_called_once_with(owner["cluster"]) + assert sky.get.call_args.args == ("down-request",) + else: + sky.down.assert_not_called() + + +def test_down_rejects_foreign_cluster_before_teardown(tmp_path, owner, sky): + sky.get.return_value = [{"name": owner["cluster"]}, {"name": "peer"}] + with pytest.raises(ValueError, match="different cluster"): + ci.worker(tmp_path, "down") + sky.down.assert_not_called() + + def scenario(monkeypatch, root, owner, statuses, *, failures=None): iterator = iter(statuses) calls = [] @@ -679,7 +701,10 @@ def test_admission_expiring_during_setup_does_not_launch(tmp_path, monkeypatch): def test_worker_does_not_submit_after_cutoff( tmp_path, owner, sky, monkeypatch, deadline ): - monkeypatch.setattr(ci.time, "time", lambda: owner[deadline]) + owner.update(admission_deadline=200, work_deadline=200) + owner[deadline] = 100 + ci.write_json(tmp_path / "owner.json", owner) + monkeypatch.setattr(ci.time, "time", lambda: 100) with pytest.raises(TimeoutError, match="launch deadline"): ci.worker(tmp_path, "launch") sky.launch.assert_not_called() @@ -737,7 +762,8 @@ def run(root, operation, timeout): ("remove_resources", 1), ] receipt = ci.read_bound(tmp_path, "cleanup.json", owner) - assert receipt["confirmed"] is False + assert receipt["operations_succeeded"] is False + assert receipt["creator_quiescence"] == "UNKNOWN" assert receipt["operations"]["remove_resources"]["success"] is True @@ -874,18 +900,31 @@ def test_initial_identity_failure_preserves_job_result(tmp_path, owner, monkeypa assert ci.supervise(tmp_path, owner, poll_seconds=0) == 0 -def test_interrupted_provisioning_captures_ids_before_cancelling( +def test_interrupted_provisioning_cancels_before_diagnostics( tmp_path, owner, monkeypatch ): ci.write_json(tmp_path / "allocation.json", {**owner, "namespace": "ci"}) ci.write_json(tmp_path / "launch-attempt.json", owner) ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request-17"}) - calls = scenario( - monkeypatch, tmp_path, owner, [], failures={"launch": TimeoutError("launch")} - ) + calls, late_jobs = [], [] + cancelled = False + + def run(root, operation, timeout): + nonlocal cancelled + calls.append(operation) + if operation == "launch": + raise TimeoutError("client launch wait expired") + if operation == "cancel_request": + cancelled = True + if operation == "resources" and not cancelled: + # Capacity can become available while the diagnostic call is waiting. + late_jobs.append("submitted after client timeout") + + monkeypatch.setattr(ci, "run_worker", run) with pytest.raises(TimeoutError): ci.supervise(tmp_path, owner) - assert calls == ["launch", "resources", "cancel_request"] + assert late_jobs == [] + assert calls == ["launch", "cancel_request", "resources"] @pytest.fixture @@ -987,7 +1026,7 @@ def test_empty_resource_history_is_unknown(tmp_path, owner, kube_receipts): @pytest.mark.parametrize("stop_api", [True, False]) @pytest.mark.parametrize("receipt", [True, False]) -def test_cleanup_absence_requires_stopped_creator_and_uid_receipt( +def test_cleanup_distinguishes_uid_absence_from_unknown_creator_lifetime( tmp_path, owner, monkeypatch, stop_api, receipt ): ci.write_json(tmp_path / "launch-attempt.json", owner) @@ -1004,10 +1043,12 @@ def test_cleanup_absence_requires_stopped_creator_and_uid_receipt( def run(root, operation, timeout): if operation == "stop_api" and not stop_api: - raise TimeoutError("creator still running") + raise TimeoutError("API stop failed") monkeypatch.setattr(ci, "run_worker", run) assert ci.cleanup(tmp_path) == (0 if stop_api else 105) - assert ci.read_bound(tmp_path, "cleanup.json", owner)["physical_absence"] == ( - "ABSENT" if stop_api and receipt else "UNKNOWN" - ) + result = ci.read_bound(tmp_path, "cleanup.json", owner) + assert result["operations_succeeded"] is stop_api + assert result["creator_quiescence"] == "UNKNOWN" + assert result["physical_absence"] == ("ABSENT" if receipt else "UNKNOWN") + assert result["physical_absence_scope"] == "retained Pod and Service UIDs" From 9f69b806fd562efc5d077b066bf4b844ac7c35c3 Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 03:32:09 +0000 Subject: [PATCH 5/8] fix(ci): enclose Sky API creators in an owned cgroup --- .github/workflows/trainer-rank-gpu.yml | 28 +- scripts/ci/trainer-rank-gpu.py | 110 ++++-- scripts/ci/trainer_rank_api.py | 390 +++++++++++++++++++ tests/unit/test_trainer_rank_api.py | 388 ++++++++++++++++++ tests/unit/test_trainer_rank_gpu_ci.py | 174 +++++++-- tests/unit/test_trainer_rank_gpu_workflow.py | 37 ++ 6 files changed, 1048 insertions(+), 79 deletions(-) create mode 100644 scripts/ci/trainer_rank_api.py create mode 100644 tests/unit/test_trainer_rank_api.py diff --git a/.github/workflows/trainer-rank-gpu.yml b/.github/workflows/trainer-rank-gpu.yml index 04f43b5b3..cb324c1d7 100644 --- a/.github/workflows/trainer-rank-gpu.yml +++ b/.github/workflows/trainer-rank-gpu.yml @@ -144,9 +144,25 @@ jobs: echo "### TrainerRank GPU validation not required" >> "${GITHUB_STEP_SUMMARY}" fi + api-enclosure: + name: Qualify CI API enclosure (CPU only) + runs-on: ubuntu-latest + timeout-minutes: 3 + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + - name: Qualify detached-child retirement without Sky or credentials + run: python3 scripts/ci/trainer_rank_api.py probe "${RUNNER_TEMP}/api-enclosure-probe" + - uses: actions/upload-artifact@v4 + if: always() + with: + name: api-enclosure-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/api-enclosure-probe + validate: name: Run on 2x H200 - needs: classify + needs: [classify, api-enclosure] if: needs.classify.outputs.required == 'true' && needs.classify.outputs.reused != 'true' runs-on: ubuntu-latest timeout-minutes: 45 @@ -156,6 +172,7 @@ jobs: queue: max environment: trainer-rank-gpu-validation env: + SKYPILOT_DISABLE_LOCAL_API_SERVER: "1" SKY_INFRA: k8s/${{ inputs.context || vars.TRAINER_RANK_GPU_CONTEXT || 'cks-wb3' }} EXPECTED_HEAD_SHA: ${{ needs.classify.outputs.head_sha }} ADMISSION_DEADLINE: ${{ needs.classify.outputs.admission_deadline }} @@ -240,8 +257,6 @@ jobs: uv pip install --python "${sky_venv}/bin/python" \ 'skypilot[kubernetes,remote] @ git+https://github.com/OpenPipe/skypilot-modal.git@924100745348cb0d880ba9e1f64b4e54cb1c0898' \ 'kubernetes==34.1.0' - sky="${sky_venv}/bin/sky" - "${sky}" check kubernetes "${sky_venv}/bin/python" scripts/ci/trainer-rank-gpu.py \ "${RUNNER_TEMP}/trainer-rank-result" @@ -263,11 +278,12 @@ jobs: gate: name: trainer-rank-gpu-validation - needs: [classify, validate] + needs: [classify, api-enclosure, validate] if: always() runs-on: ubuntu-latest steps: - env: + ENCLOSURE_RESULT: ${{ needs.api-enclosure.result }} CLASSIFY_RESULT: ${{ needs.classify.result }} REQUIRED: ${{ needs.classify.outputs.required }} REUSED: ${{ needs.classify.outputs.reused }} @@ -275,6 +291,10 @@ jobs: RESULT: ${{ needs.validate.result }} run: | set -euo pipefail + if [ "${ENCLOSURE_RESULT}" != "success" ]; then + echo "CI API enclosure qualification failed." >&2 + exit 1 + fi if [ "${CLASSIFY_RESULT}" != "success" ]; then echo "TrainerRank GPU classification failed." >&2 exit 1 diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index 23d0a0223..cc7f4b115 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -3,8 +3,8 @@ SkyPilot returns an actual job ID from launch. Each SDK operation runs in a bounded child; only that job's SUCCEEDED status passes. An always-run workflow step requests cancellation and checks retained resource UIDs. The admission -deadline bounds the client wait, not provider creation. Successful API shutdown -does not prove creator retirement; cleanup receipts keep that state UNKNOWN. +deadline bounds the client wait, not provider creation. The owned API cgroup +must be retired before the final retained-UID reconciliation. """ import json @@ -16,6 +16,8 @@ import time import uuid +import trainer_rank_api as api_service + NONTERMINAL = {"INIT", "PENDING", "SETTING_UP", "RUNNING"} INFRAS = {"k8s/cks-wb3", "k8s/ext-collab2"} LABEL = "art-ci-execution" @@ -169,9 +171,21 @@ def resources(root, owner, *, delete): def worker(root, operation): + owner = json.loads((root / "owner.json").read_text()) + if operation == "stop_api": + api_service.stop(root, owner) + return + if operation in {"resources", "remove_resources"}: + if operation == "remove_resources": + api_service.require_retired(root, owner) + resources(root, owner, delete=operation == "remove_resources") + return + api_service.client_environment(root, owner) import sky - owner = json.loads((root / "owner.json").read_text()) + if operation == "check": + sky.get(sky.check(clouds=["kubernetes"])) + return cluster = owner["cluster"] if operation == "launch": task = launch_task(root, owner) @@ -233,22 +247,6 @@ def worker(root, operation): if clusters: sky.get(sky.down(cluster)) return - if operation == "stop_api": - # This workflow owns the entire ephemeral runner and its local API. - # Best-effort shutdown: the SDK snapshots descendants before killing - # the parent, so an absent API process does not prove creator retirement. - if ( - os.environ.get("GITHUB_ACTIONS") != "true" - or os.environ.get("RUNNER_ENVIRONMENT") != "github-hosted" - ): - raise ValueError("API cleanup requires an isolated GitHub-hosted runner") - sky.api_stop() # The maintained SDK rejects remote API endpoints. - if sky.api_status(all_status=True, limit=1): - raise RuntimeError("The CI API still reports requests after stopping") - return - if operation in {"resources", "remove_resources"}: - resources(root, owner, delete=operation == "remove_resources") - return job_id = read_bound(root, "job.json", owner)["job_id"] if operation == "status": statuses = sky.get(sky.job_status(cluster, job_ids=[job_id])) @@ -435,16 +433,26 @@ def supervise(root, owner, timeout=35 * 60, poll_seconds=10): job = read_bound(root, "job.json", owner) except Exception: pass - operations = (["cancel"] if not terminal else []) + ["logs"] if job else [] - if job is None and (root / "request.json").exists(): - operations = ["cancel_request"] - if not terminal and (root / "allocation.json").exists(): - # Request cancellation before diagnostics can prolong capacity wait. - operations.insert(1, "resources") + operations = ["logs"] if terminal else [] + if not terminal: + if job: + operations.append("cancel") + elif (root / "request.json").exists(): + operations.append("cancel_request") + # Retire creators before any census, logs, or teardown delays. A + # terminal cancellation receipt alone does not join SDK executors. + operations.append("stop_api") outcomes = {} for operation in operations: try: - run_worker(root, operation, 30) + run_worker( + root, + operation, + min( + 5 if operation.startswith("cancel") else 30, + owner["cleanup_deadline"] - time.time(), + ), + ) outcomes[operation] = {"success": True} except Exception as error: outcomes[operation] = {"success": False, "error": repr(error)} @@ -501,30 +509,51 @@ def admit(root): def cleanup(root): - if not (root / "launch-attempt.json").exists(): + if ( + not (root / "api-plan.json").exists() + and not (root / "launch-attempt.json").exists() + ): return 0 owner = json.loads((root / "owner.json").read_text()) - - read_bound(root, "launch-attempt.json", owner) + if (root / "launch-attempt.json").exists(): + read_bound(root, "launch-attempt.json", owner) + terminal = False + try: + result = read_bound(root, "result.json", owner) + terminal = result["status"] in EXIT_CODES and result["status"] is not None + read_bound(root, "job.json", owner) + except (OSError, ValueError, KeyError): + terminal = False + operations = [] + if not (root / "api-retired.json").exists(): + if terminal: + operations.append(("down", 90)) + elif (root / "request.json").exists(): + operations.append(("cancel_request", 5)) + elif (root / "job.json").exists(): + operations.append(("cancel", 5)) + operations.append(("stop_api", 30)) + if (root / "allocation.json").exists(): + operations.append(("remove_resources", 90)) outcomes = {} - # A failed/unknown cancellation must not prevent best-effort physical cleanup. - for operation, limit in [ - ("cancel_request", 30), - ("resources", 20), - ("down", 90), - ("stop_api", 30), - ("remove_resources", 90), - ]: + retired = False + for operation, limit in operations: try: + if operation == "remove_resources": + # Do not reconcile final absence while creators may still run. + api_service.require_retired(root, owner) run_worker( root, operation, min(limit, owner["cleanup_deadline"] - time.time()) ) + if operation == "stop_api": + api_service.require_retired(root, owner) + retired = True outcomes[operation] = {"success": True} except BaseException as error: outcomes[operation] = {"success": False, "error": repr(error)} operations_succeeded = all(outcome["success"] for outcome in outcomes.values()) physical_absence = "UNKNOWN" - if outcomes["remove_resources"]["success"]: + if outcomes.get("remove_resources", {}).get("success"): try: evidence = read_bound(root, "resources-cleanup.json", owner) if ( @@ -542,7 +571,8 @@ def cleanup(root): "operations_succeeded": operations_succeeded, "physical_absence": physical_absence, "physical_absence_scope": "retained Pod and Service UIDs", - "creator_quiescence": "UNKNOWN", + "creator_quiescence": "RETIRED" if retired else "UNKNOWN", + "creator_quiescence_scope": "owned local API cgroup; excludes provider in-flight requests", "operations": outcomes, }, ) @@ -567,6 +597,8 @@ def interrupted(signum, frame): for signum in (signal.SIGINT, signal.SIGTERM): signal.signal(signum, interrupted) + api_service.start(root, owner) + run_worker(root, "check", min(60, owner["admission_deadline"] - time.time())) return supervise(root, owner, timeout=max(0, owner["work_deadline"] - time.time())) diff --git a/scripts/ci/trainer_rank_api.py b/scripts/ci/trainer_rank_api.py new file mode 100644 index 000000000..62fda185a --- /dev/null +++ b/scripts/ci/trainer_rank_api.py @@ -0,0 +1,390 @@ +"""Private GitHub-hosted CI enclosure; never discover or stop another API server.""" + +import json +import math +import os +from pathlib import Path +import re +import socket +import subprocess +import sys +import time +import uuid + +CGROUPS = Path("/sys/fs/cgroup") +PROC = Path("/proc") +BOOT_ID = Path("/proc/sys/kernel/random/boot_id") +PROPERTIES = { + "Transient": "yes", + "Type": "exec", + "ExitType": "cgroup", + "KillMode": "control-group", + "SendSIGKILL": "yes", + "Restart": "no", + "NoNewPrivileges": "yes", + "ProtectControlGroups": "yes", +} + + +def write(path, data): + temporary = path.with_suffix(".tmp") + with temporary.open("w") as stream: + json.dump(data, stream) + stream.flush() + os.fsync(stream.fileno()) + temporary.replace(path) + + +def read(root, name, expected): + data = json.loads((root / name).read_text()) + if any(data.get(key) != value for key, value in expected.items()): + raise ValueError(f"Wrong CI API identity in {name}") + return data + + +def host(): + if ( + os.environ.get("GITHUB_ACTIONS") != "true" + or os.environ.get("RUNNER_ENVIRONMENT") != "github-hosted" + or os.getuid() == 0 + or not (CGROUPS / "cgroup.controllers").is_file() + ): + raise ValueError("CI API requires an isolated GitHub-hosted cgroup-v2 runner") + + +def unit_name(owner): + if not re.fullmatch(r"[0-9a-f]{32}", owner["label"]): + raise ValueError("Invalid API nonce") + return f"art-ci-{owner['label']}.service" + + +def identity(root, owner): + host() + plan = read(root, "api-plan.json", owner) + unit = unit_name(owner) + expected = { + **plan, + "unit": unit, + "cgroup": f"/system.slice/{unit}", + "boot_id": BOOT_ID.read_text().strip(), + "uid": os.getuid(), + "gid": os.getgid(), + } + receipt = read(root, "api-service.json", expected) + if not re.fullmatch(r"[0-9a-f]{32}", receipt["invocation"]): + raise ValueError("Invalid API invocation") + return receipt + + +def show(receipt): + result = subprocess.run( + [ + "systemctl", + "show", + receipt["unit"], + "--no-pager", + "--property=" + + ",".join( + [ + *PROPERTIES, + "LoadState", + "ActiveState", + "SubState", + "InvocationID", + "ControlGroup", + "User", + "Group", + "MainPID", + ] + ), + ], + capture_output=True, + text=True, + timeout=5, + ) + values = dict(line.split("=", 1) for line in result.stdout.splitlines()) + if result.returncode and not ( + result.returncode == 4 and values.get("LoadState") == "not-found" + ): + raise RuntimeError("Cannot inspect owned API unit") + return values + + +def matching(receipt, values): + expected = { + **PROPERTIES, + "InvocationID": receipt["invocation"], + "ControlGroup": receipt["cgroup"], + "User": str(receipt["uid"]), + "Group": str(receipt["gid"]), + } + if any(values.get(key) != value for key, value in expected.items()): + raise ValueError("Owned API unit changed identity or containment") + + +def empty(receipt): + path = CGROUPS / receipt["cgroup"].lstrip("/") + try: + events = dict( + line.split() for line in (path / "cgroup.events").read_text().splitlines() + ) + except FileNotFoundError: + # The kernel only removes an empty cgroup. Permission/read failures do + # not count, nor does a missing events file in a still-existing cgroup. + try: + path.stat() + except FileNotFoundError: + return True + raise RuntimeError("Owned API cgroup lacks population evidence") + if events.get("populated") not in {"0", "1"}: + raise RuntimeError("Owned API cgroup population is unknown") + return events["populated"] == "0" + + +def require_running(root, owner): + receipt = identity(root, owner) + values = show(receipt) + matching(receipt, values) + if values.get("ActiveState") != "active" or empty(receipt): + raise RuntimeError("Owned API service is not running") + return receipt + + +def client_environment(root, owner): + receipt = require_running(root, owner) + if not owns_listener(receipt): + raise RuntimeError("Owned API listener is unavailable") + os.environ["SKYPILOT_DISABLE_LOCAL_API_SERVER"] = "1" + os.environ["SKYPILOT_API_SERVER_ENDPOINT"] = f"http://127.0.0.1:{receipt['port']}" + os.environ["SKYPILOT_API_SERVER_LOCAL_PORT"] = str(receipt["port"]) + + +def owns_listener(receipt): + """Check only this cgroup's processes for the exact loopback listening socket.""" + address = f"0100007F:{receipt['port']:04X}" + sockets = { + f"socket:[{fields[9]}]" + for line in (PROC / "net/tcp").read_text().splitlines()[1:] + if (fields := line.split())[1] == address and fields[3] == "0A" + } + group = CGROUPS / receipt["cgroup"].lstrip("/") + for members in group.rglob("cgroup.procs"): + for pid in members.read_text().split(): + process = PROC / pid + try: + before = (process / "cgroup").read_text() + owned = f"0::{receipt['cgroup']}\n" + if before != owned: + continue + found = any( + os.readlink(fd) in sockets for fd in (process / "fd").iterdir() + ) + if found and (process / "cgroup").read_text() == before: + return True + except FileNotFoundError: + continue + return False + + +def start(root, owner, *, probe=False): + host() + unit = unit_name(owner) + # Do not retry/recreate an invocation, including an interrupted startup. + if (root / "api-plan.json").exists(): + raise ValueError("CI API startup was already attempted") + # Reserve 10s for systemd-run, 5s TERM grace, and 5s scheduling margin. + # This is never renewed, even when launch/setup consumes the work window. + remaining = math.floor(owner["cleanup_deadline"] - time.time()) - 20 + if remaining <= 0 or time.time() >= owner["admission_deadline"]: + raise TimeoutError("CI API lifetime expired") + with socket.socket() as reservation: + reservation.bind(("127.0.0.1", 0)) + port = reservation.getsockname()[1] + plan = { + **owner, + "unit": unit, + "port": port, + "runtime_seconds": remaining, + "probe": probe, + } + write(root / "api-plan.json", plan) + properties = { + **PROPERTIES, + "TimeoutStartSec": "10s", + "TimeoutStopSec": "5s", + "RuntimeMaxSec": f"{remaining}s", + } + command = [ + "sudo", + "-n", + "systemd-run", + f"--unit={plan['unit']}", + f"--uid={os.getuid()}", + f"--gid={os.getgid()}", + f"--working-directory={Path.cwd()}", + ] + command += [ + f"--property={key}={value}" + for key, value in properties.items() + if key != "Transient" + ] + # Workflow auth/config files live under HOME; task context/labels travel + # in the SDK payload. Retain the runner's executables and optional kubeconfig. + command += [ + f"--setenv={key}={os.environ[key]}" + for key in ("HOME", "PATH", "USER", "LOGNAME", "KUBECONFIG") + if key in os.environ + ] + command += [ + sys.executable, + str(Path(__file__).resolve()), + "serve", + str(root.resolve()), + ] + with (root / "api-start.log").open("ab") as log: + subprocess.run( + command, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=10 + ) + deadline = min(time.time() + 60, owner["admission_deadline"]) + while not (root / "api-service.json").exists(): + if time.time() >= deadline: + raise TimeoutError("CI API did not record its cgroup identity") + time.sleep(0.1) + receipt = require_running(root, owner) + if time.time() >= deadline: + raise TimeoutError("CI API startup missed admission") + write(root / "api-release.json", receipt) + while True: + require_running(root, owner) + if probe: + if (root / "api-probe.json").exists() and show(receipt).get( + "MainPID" + ) == "0": + read(root, "api-probe.json", receipt) + return + elif owns_listener(receipt): + return + if time.time() >= deadline: + raise TimeoutError("Owned API listener did not become ready") + time.sleep(0.1) + + +def stop(root, owner): + receipt = identity(root, owner) + if (root / "api-retired.json").exists(): + require_retired(root, owner) + return + values = show(receipt) + already_stopped = ( + values.get("ActiveState") in {"inactive", "failed"} + and values.get("InvocationID") == receipt["invocation"] + and values.get("ControlGroup") in {"", receipt["cgroup"]} + and empty(receipt) + ) + if values.get("LoadState") != "not-found" and not already_stopped: + matching(receipt, values) + # No asynchronous stop: systemd first sends TERM to the entire cgroup, + # then KILL after five seconds, including helpers which called setsid(). + subprocess.run( + ["sudo", "-n", "systemctl", "stop", receipt["unit"]], check=True, timeout=15 + ) + if not empty(receipt): + raise RuntimeError("Owned API cgroup retirement is unproven") + write(root / "api-retired.json", {**receipt, "creator_quiescence": "RETIRED"}) + + +def require_retired(root, owner): + receipt = identity(root, owner) + read(root, "api-retired.json", {**receipt, "creator_quiescence": "RETIRED"}) + if not empty(receipt): + raise RuntimeError("Owned API cgroup is populated after retirement") + + +def serve(root): + plan = json.loads((root / "api-plan.json").read_text()) + group = f"/system.slice/{plan['unit']}" + if (PROC / "self/cgroup").read_text() != f"0::{group}\n": + raise ValueError("API service did not enter the planned cgroup") + receipt = { + **plan, + "cgroup": group, + "invocation": os.environ["INVOCATION_ID"], + "boot_id": BOOT_ID.read_text().strip(), + "uid": os.getuid(), + "gid": os.getgid(), + } + write(root / "api-service.json", receipt) + deadline = min(time.time() + 15, plan["admission_deadline"]) + while not (root / "api-release.json").exists(): + if time.time() >= deadline: + raise TimeoutError("API service was not released by its owner") + time.sleep(0.1) + read(root, "api-release.json", receipt) + if time.time() >= plan["admission_deadline"]: + raise TimeoutError("CI API release missed admission") + if plan["probe"]: + # CPU-only hosted qualification: an ordinary detached child remains in + # this service cgroup after its leader exits. No SDK/network/GPU import. + subprocess.Popen( + [ + sys.executable, + "-c", + "import pathlib,signal,sys,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); pathlib.Path(sys.argv[1]).touch(); time.sleep(30)", + str(root / "api-probe-ready"), + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + deadline = time.time() + 5 + while not (root / "api-probe-ready").exists(): + if time.time() >= deadline: + raise TimeoutError("Detached probe did not start") + time.sleep(0.05) + write(root / "api-probe.json", receipt) + return + os.environ.update( + SKYPILOT_DISABLE_LOCAL_API_SERVER="1", + SKYPILOT_API_SERVER_ENDPOINT=f"http://127.0.0.1:{plan['port']}", + SKYPILOT_API_SERVER_LOCAL_PORT=str(plan["port"]), + IS_SKYPILOT_SERVER="true", + ) + # This is the exec target of the pinned SDK's api_start(foreground=True). + # Direct exec keeps the SDK autostart prohibition inherited by ALL children. + log = os.open( + root / "api-server.log", os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600 + ) + os.dup2(log, 1) + os.dup2(log, 2) + os.close(log) + os.execv( + sys.executable, + [ + sys.executable, + "-m", + "sky.server.server", + "--host=127.0.0.1", + f"--port={plan['port']}", + ], + ) + + +def probe(root): + root.mkdir(parents=True, exist_ok=False) + now = time.time() + owner = { + "label": uuid.uuid4().hex, + "admission_deadline": now + 30, + "cleanup_deadline": now + 60, + } + try: + start(root, owner, probe=True) + finally: + if (root / "api-service.json").exists(): + stop(root, owner) + require_retired(root, owner) + + +if __name__ == "__main__": + {"serve": serve, "probe": probe}[sys.argv[1]](Path(sys.argv[2])) diff --git a/tests/unit/test_trainer_rank_api.py b/tests/unit/test_trainer_rank_api.py new file mode 100644 index 000000000..5469980e1 --- /dev/null +++ b/tests/unit/test_trainer_rank_api.py @@ -0,0 +1,388 @@ +"""File-backed systemd/cgroup controls only: never start a service or Sky API.""" + +import importlib.util +import json +import os +from pathlib import Path +import subprocess +from types import SimpleNamespace as NS +from unittest.mock import MagicMock, Mock + +import pytest + +SCRIPT = Path(__file__).parents[2] / "scripts/ci/trainer_rank_api.py" +spec = importlib.util.spec_from_file_location("api_enclosure", SCRIPT) +api = importlib.util.module_from_spec(spec) +spec.loader.exec_module(api) + + +@pytest.fixture +def control(tmp_path, monkeypatch): + root = tmp_path / "evidence" + root.mkdir() + cgroups = tmp_path / "cgroups" + cgroups.mkdir() + (cgroups / "cgroup.controllers").touch() + boot = tmp_path / "boot-id" + boot.write_text("this-boot") + monkeypatch.setattr(api, "CGROUPS", cgroups) + monkeypatch.setattr(api, "BOOT_ID", boot) + monkeypatch.setenv("GITHUB_ACTIONS", "true") + monkeypatch.setenv("RUNNER_ENVIRONMENT", "github-hosted") + monkeypatch.setattr(api.os, "getuid", lambda: 1001) + monkeypatch.setattr(api.os, "getgid", lambda: 1002) + monkeypatch.setattr(api.time, "time", lambda: 100) + owner = {"label": "a" * 32, "admission_deadline": 1000, "cleanup_deadline": 2000} + unit = f"art-ci-{owner['label']}.service" + plan = { + **owner, + "unit": unit, + "port": 12345, + "runtime_seconds": 1880, + "probe": False, + } + receipt = { + **plan, + "cgroup": f"/system.slice/{unit}", + "invocation": "b" * 32, + "boot_id": "this-boot", + "uid": 1001, + "gid": 1002, + } + group = cgroups / receipt["cgroup"].lstrip("/") + group.mkdir(parents=True) + events = group / "cgroup.events" + events.write_text("populated 1\nfrozen 0\n") + api.write(root / "api-plan.json", plan) + api.write(root / "api-service.json", receipt) + values = { + **api.PROPERTIES, + "LoadState": "loaded", + "ActiveState": "active", + "SubState": "running", + "InvocationID": receipt["invocation"], + "ControlGroup": receipt["cgroup"], + "User": "1001", + "Group": "1002", + "MainPID": "42", + } + calls = [] + + def run(command, **kwargs): + calls.append((command, kwargs)) + if command[:2] == ["systemctl", "show"]: + assert command[2] == unit + return NS( + returncode=0, stdout="\n".join(f"{k}={v}" for k, v in values.items()) + ) + assert command == ["sudo", "-n", "systemctl", "stop", unit] + events.write_text("populated 0\nfrozen 0\n") + return NS(returncode=0) + + monkeypatch.setattr(api.subprocess, "run", run) + return NS( + root=root, + owner=owner, + receipt=receipt, + group=group, + events=events, + values=values, + calls=calls, + run=run, + ) + + +def test_stop_proves_whole_cgroup_empty_and_is_idempotent(control): + c = control + api.stop(c.root, c.owner) + api.require_retired(c.root, c.owner) + proof = json.loads((c.root / "api-retired.json").read_text()) + assert proof["creator_quiescence"] == "RETIRED" + assert proof["invocation"] == c.receipt["invocation"] + assert c.calls[-1][0] == ["sudo", "-n", "systemctl", "stop", c.receipt["unit"]] + assert c.calls[-1][1]["timeout"] == 15 + calls = list(c.calls) + api.stop(c.root, c.owner) + assert c.calls == calls + c.events.write_text("populated 1\n") + with pytest.raises(RuntimeError, match="populated"): + api.require_retired(c.root, c.owner) + + +@pytest.mark.parametrize( + "field,value", + [ + ("InvocationID", "c" * 32), + ("ControlGroup", "/system.slice/peer.service"), + ("User", "1003"), + ("Group", "1003"), + ("KillMode", "process"), + ("Restart", "always"), + ("ExitType", "main"), + ("ProtectControlGroups", "no"), + ], +) +def test_changed_unit_never_receives_stop(control, field, value): + c = control + c.values[field] = value + with pytest.raises(ValueError, match="changed identity"): + api.stop(c.root, c.owner) + assert all(command[0] == "systemctl" for command, _ in c.calls) + assert not (c.root / "api-retired.json").exists() + + +@pytest.mark.parametrize( + "field,value", + [("boot_id", "old-boot"), ("cgroup", "/peer"), ("uid", 123), ("invocation", "")], +) +def test_receipt_mismatch_never_controls_service(control, field, value): + c = control + api.write(c.root / "api-service.json", {**c.receipt, field: value}) + with pytest.raises(ValueError): + api.stop(c.root, c.owner) + assert c.calls == [] + + +def test_stop_success_or_terminal_cancellation_does_not_prove_retirement( + control, monkeypatch +): + c = control + + def no_retirement(command, **kwargs): + if command[0] == "systemctl": + return c.run(command, **kwargs) + return NS(returncode=0) # Detached late executor still populates cgroup. + + monkeypatch.setattr(api.subprocess, "run", no_retirement) + api.write(c.root / "request-terminal.json", {**c.owner, "status": "CANCELLED"}) + with pytest.raises(RuntimeError, match="retirement is unproven"): + api.stop(c.root, c.owner) + assert not (c.root / "api-retired.json").exists() + + +def test_missing_receipt_and_ambiguous_cgroup_are_unknown(control): + c = control + c.events.unlink() + with pytest.raises(RuntimeError, match="population evidence"): + api.empty(c.receipt) + (c.root / "api-service.json").unlink() + with pytest.raises(FileNotFoundError): + api.stop(c.root, c.owner) + assert c.calls == [] + + +def test_collected_unit_requires_recorded_cgroup_to_have_disappeared(control): + c = control + c.values.clear() + c.values["LoadState"] = "not-found" + with pytest.raises(RuntimeError, match="unproven"): + api.stop(c.root, c.owner) + c.events.unlink() + c.group.rmdir() + api.stop(c.root, c.owner) + api.require_retired(c.root, c.owner) + assert all(command[0] == "systemctl" for command, _ in c.calls) + + +def test_stop_timeout_cannot_emit_retirement(control, monkeypatch): + c = control + + def timeout(command, **kwargs): + if command[0] == "systemctl": + return c.run(command, **kwargs) + raise subprocess.TimeoutExpired(command, 15) + + monkeypatch.setattr(api.subprocess, "run", timeout) + with pytest.raises(subprocess.TimeoutExpired): + api.stop(c.root, c.owner) + assert not (c.root / "api-retired.json").exists() + + +def test_client_cannot_choose_another_endpoint_or_autostart(control, monkeypatch): + c = control + monkeypatch.setattr(api, "owns_listener", lambda _: True) + monkeypatch.setenv("SKYPILOT_API_SERVER_ENDPOINT", "https://peer") + monkeypatch.setenv("SKYPILOT_DISABLE_LOCAL_API_SERVER", "0") + api.client_environment(c.root, c.owner) + assert os.environ["SKYPILOT_API_SERVER_ENDPOINT"] == "http://127.0.0.1:12345" + assert os.environ["SKYPILOT_DISABLE_LOCAL_API_SERVER"] == "1" + c.values["ActiveState"] = "failed" + with pytest.raises(RuntimeError, match="not running"): + api.client_environment(c.root, c.owner) + + +def test_start_has_finite_runtime_and_receipt_gate_before_release(control, monkeypatch): + c = control + (c.root / "api-plan.json").unlink() + (c.root / "api-service.json").unlink() + reservation = Mock() + reservation.getsockname.return_value = ("127.0.0.1", 12345) + context = MagicMock() + context.__enter__.return_value = reservation + monkeypatch.setattr(api.socket, "socket", Mock(return_value=context)) + command_seen = [] + + def start(command, **kwargs): + if command[:3] == ["sudo", "-n", "systemd-run"]: + assert not (c.root / "api-release.json").exists() + command_seen.extend(command) + api.write(c.root / "api-service.json", c.receipt) + return NS(returncode=0) + return c.run(command, **kwargs) + + monkeypatch.setattr(api.subprocess, "run", start) + + def listener(receipt): + assert json.loads((c.root / "api-release.json").read_text()) == receipt + return True + + monkeypatch.setattr(api, "owns_listener", listener) + api.start(c.root, c.owner) + assert "--property=RuntimeMaxSec=1880s" in command_seen + assert "--property=TimeoutStopSec=5s" in command_seen + assert "--property=KillMode=control-group" in command_seen + assert "--property=ExitType=cgroup" in command_seen + assert "--uid=1001" in command_seen and "--gid=1002" in command_seen + with pytest.raises(ValueError, match="already attempted"): + api.start(c.root, c.owner) + + +@pytest.mark.parametrize("host", ["self-hosted", ""]) +def test_unsupported_host_is_rejected_without_service_control( + control, monkeypatch, host +): + c = control + monkeypatch.setenv("RUNNER_ENVIRONMENT", host) + with pytest.raises(ValueError, match="GitHub-hosted"): + api.start(c.root, c.owner) + assert c.calls == [] + + +def test_listener_must_belong_to_exact_cgroup(control, monkeypatch, tmp_path): + c = control + proc = tmp_path / "proc" + (proc / "net").mkdir(parents=True) + (proc / "net/tcp").write_text( + "header\n0: 0100007F:3039 00000000:0000 0A 0 0 0 0 0 123\n" + ) + (proc / "42/fd").mkdir(parents=True) + (proc / "42/fd/7").symlink_to("socket:[123]") + (proc / "42/cgroup").write_text("0::/peer\n") + (c.group / "cgroup.procs").write_text("42\n") + monkeypatch.setattr(api, "PROC", proc) + assert not api.owns_listener(c.receipt) + (proc / "42/cgroup").write_text(f"0::{c.receipt['cgroup']}\n") + assert api.owns_listener(c.receipt) + assert not api.owns_listener({**c.receipt, "port": 12346}) + + +def test_runtime_expiry_accepts_only_same_invocation_and_empty_cgroup(control): + c = control + c.values.update(ActiveState="failed", ControlGroup="") + with pytest.raises(ValueError): + api.stop(c.root, c.owner) + c.events.write_text("populated 0\n") + api.stop(c.root, c.owner) + api.require_retired(c.root, c.owner) + assert all(command[0] == "systemctl" for command, _ in c.calls) + + +def test_live_unit_without_owned_listener_cannot_receive_client_calls( + control, monkeypatch +): + monkeypatch.setattr(api, "owns_listener", lambda _: False) + with pytest.raises(RuntimeError, match="listener is unavailable"): + api.client_environment(control.root, control.owner) + + +@pytest.mark.parametrize("events", ["", "frozen 0\n", "populated unknown\n"]) +def test_unknown_population_blocks_start_and_retirement(control, events, monkeypatch): + c = control + monkeypatch.setattr( + api.subprocess, + "run", + lambda command, **kwargs: ( + c.run(command, **kwargs) if command[0] == "systemctl" else NS(returncode=0) + ), + ) + c.events.write_text(events) + with pytest.raises(RuntimeError, match="population is unknown"): + api.require_running(c.root, c.owner) + with pytest.raises(RuntimeError, match="population is unknown"): + api.stop(c.root, c.owner) + assert not (c.root / "api-retired.json").exists() + + +def test_invalid_nonce_is_rejected_before_any_service_creation(control): + c = control + (c.root / "api-plan.json").unlink() + with pytest.raises(ValueError, match="nonce"): + api.start(c.root, {**c.owner, "label": "../peer"}) + assert c.calls == [] + + +@pytest.mark.parametrize("released", [True, False]) +def test_foreground_server_inherits_autostart_prohibition_only_after_release( + control, monkeypatch, tmp_path, released +): + c = control + proc = tmp_path / "proc" + (proc / "self").mkdir(parents=True) + (proc / "self/cgroup").write_text(f"0::{c.receipt['cgroup']}\n") + monkeypatch.setattr(api, "PROC", proc) + monkeypatch.setenv("INVOCATION_ID", c.receipt["invocation"]) + monkeypatch.setenv("SKYPILOT_DISABLE_LOCAL_API_SERVER", "0") + api.write( + c.root / "api-release.json", + c.receipt if released else {**c.receipt, "invocation": "peer"}, + ) + execute = Mock(side_effect=RuntimeError("exec boundary")) + monkeypatch.setattr(api.os, "execv", execute) + monkeypatch.setattr(api.os, "dup2", Mock()) + with pytest.raises(RuntimeError if released else ValueError): + api.serve(c.root) + if released: + assert os.environ["SKYPILOT_DISABLE_LOCAL_API_SERVER"] == "1" + assert os.environ["IS_SKYPILOT_SERVER"] == "true" + assert execute.call_args.args[1][1:] == [ + "-m", + "sky.server.server", + "--host=127.0.0.1", + "--port=12345", + ] + else: + execute.assert_not_called() + assert c.calls == [] + + +def test_timed_out_start_cannot_release_a_later_service(control, monkeypatch, tmp_path): + c = control + (c.root / "api-plan.json").unlink() + (c.root / "api-service.json").unlink() + context = MagicMock() + context.__enter__.return_value.getsockname.return_value = ("127.0.0.1", 12345) + monkeypatch.setattr(api.socket, "socket", Mock(return_value=context)) + start = Mock(side_effect=subprocess.TimeoutExpired("systemd-run", 10)) + monkeypatch.setattr(api.subprocess, "run", start) + with pytest.raises(subprocess.TimeoutExpired): + api.start(c.root, c.owner) + assert "--property=RuntimeMaxSec=1880s" in start.call_args.args[0] + assert not (c.root / "api-release.json").exists() + assert not (c.root / "api-service.json").exists() + + # Simulate systemd accepting the timed-out request later. Its process has + # no permission receipt, independently times out, and never execs Sky. + proc = tmp_path / "proc" + (proc / "self").mkdir(parents=True) + (proc / "self/cgroup").write_text(f"0::{c.receipt['cgroup']}\n") + monkeypatch.setattr(api, "PROC", proc) + monkeypatch.setenv("INVOCATION_ID", c.receipt["invocation"]) + clock = iter([100, 116]) + monkeypatch.setattr(api.time, "time", lambda: next(clock)) + execute = Mock() + monkeypatch.setattr(api.os, "execv", execute) + with pytest.raises(TimeoutError, match="not released"): + api.serve(c.root) + assert (c.root / "api-service.json").exists() + assert not (c.root / "api-release.json").exists() + execute.assert_not_called() diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index d7e954ad3..6489aba92 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -15,12 +15,20 @@ import pytest SCRIPT = Path(__file__).parents[2] / "scripts/ci/trainer-rank-gpu.py" +sys.path.insert(0, str(SCRIPT.parent)) spec = importlib.util.spec_from_file_location("trainer_rank_gpu_ci", SCRIPT) assert spec is not None and spec.loader is not None ci = importlib.util.module_from_spec(spec) spec.loader.exec_module(ci) +@pytest.fixture(autouse=True) +def service_boundary(monkeypatch): + # Service controls have dedicated file-backed tests; never create a unit here. + for name in ("start", "client_environment", "require_retired"): + monkeypatch.setattr(ci.api_service, name, Mock()) + + @pytest.fixture def owner(tmp_path): value = { @@ -204,7 +212,7 @@ def test_remote_terminal_result_controls_exit( calls = scenario(monkeypatch, tmp_path, owner, ["SETTING_UP", "RUNNING", status]) assert ci.supervise(tmp_path, owner, poll_seconds=0) == code assert calls.count("launch") == 1 - assert calls[-1] == "logs" + assert calls[-1] == ("logs" if status is not None else "stop_api") assert ("cancel" in calls) is (status is None) assert json.loads((tmp_path / "result.json").read_text())["status"] == status @@ -243,7 +251,7 @@ def test_interruption_attempts_exact_cancel_and_logs( with pytest.raises(type(error)) as raised: ci.supervise(tmp_path, owner, poll_seconds=0) assert raised.value is error - assert calls[-2:] == ["cancel", "logs"] + assert calls[-2:] == ["cancel", "stop_api"] def test_repeated_query_failure_and_cleanup_errors_preserve_primary( @@ -261,7 +269,7 @@ def test_repeated_query_failure_and_cleanup_errors_preserve_primary( ci.supervise(tmp_path, owner, poll_seconds=0) assert raised.value is primary assert calls.count("status") == 3 - assert calls[-2:] == ["cancel", "logs"] + assert calls[-2:] == ["cancel", "stop_api"] def test_receipt_failure_still_cancels_and_preserves_error( @@ -280,7 +288,7 @@ def fail_read(root, filename, expected): with pytest.raises(OSError) as raised: ci.supervise(tmp_path, owner, poll_seconds=0) assert raised.value is primary - assert calls[-2:] == ["cancel", "logs"] + assert calls[-2:] == ["cancel", "stop_api"] def test_late_success_does_not_bypass_deadline(tmp_path, owner, monkeypatch): @@ -289,7 +297,7 @@ def test_late_success_does_not_bypass_deadline(tmp_path, owner, monkeypatch): monkeypatch.setattr(ci.time, "monotonic", lambda: next(clock)) with pytest.raises(TimeoutError): ci.supervise(tmp_path, owner, timeout=1) - assert calls[-2:] == ["cancel", "logs"] + assert calls[-2:] == ["cancel", "stop_api"] def test_real_bounded_child_kills_waiting_fork_descendant(tmp_path): @@ -377,7 +385,11 @@ def test_launch_timeout_cancels_only_recorded_request( with pytest.raises(TimeoutError) as raised: ci.supervise(tmp_path, owner) assert raised.value is error - assert calls == (["launch", "cancel_request"] if known_request else ["launch"]) + assert calls == [ + "launch", + *(["cancel_request"] if known_request else []), + "stop_api", + ] # A lost launch reply does not prove the remote test never started. assert ci.read_bound(tmp_path, "result.json", owner)["status"] == ( "UNCONFIRMED" if attempted else "NOT_RUN" @@ -396,7 +408,7 @@ def run(root, operation, timeout): monkeypatch.setattr(ci, "run_worker", run) with pytest.raises(TimeoutError): ci.supervise(tmp_path, owner, timeout=2100) - assert calls == [("launch", 7)] + assert calls == [("launch", 7), ("stop_api", 30)] @pytest.mark.parametrize( @@ -432,6 +444,13 @@ def test_main_rejects_invalid_scope_before_launch(tmp_path, monkeypatch, key, va @pytest.mark.parametrize("infra", ["k8s/cks-wb3", "k8s/ext-collab2"]) def test_real_worker_round_trip_with_fake_sdk(tmp_path, infra): """Exercise the actual direct Python parent/worker JSON transport, without Sky.""" + (tmp_path / "sitecustomize.py").write_text(f""" +import sys +sys.path.insert(0, {str(SCRIPT.parent)!r}) +import trainer_rank_api +trainer_rank_api.start = lambda *a, **kw: None +trainer_rank_api.client_environment = lambda *a, **kw: None +""") (tmp_path / "sky.py").write_text(""" import os import socket @@ -460,6 +479,7 @@ def get(value): if value == "launch-request": return 17, NS(cluster_name=Path(os.environ["FAKE_CLUSTER"]).read_text()) return value +def check(**kwargs): return "checked" def job_status(cluster, *, job_ids): assert job_ids == [17] return {17:NS(value="SUCCEEDED")} @@ -678,6 +698,7 @@ def test_setup_consumes_original_admitted_work_window(tmp_path, monkeypatch): now[0] = 400 run = Mock(return_value=0) monkeypatch.setattr(ci, "supervise", run) + monkeypatch.setattr(ci, "run_worker", Mock()) assert ci.main(root) == 0 assert run.call_args.kwargs["timeout"] == 1800 assert json.loads((root / "owner.json").read_text())["cleanup_deadline"] == 2500 @@ -742,6 +763,8 @@ def test_request_cancellation_does_not_claim_unknown_terminal( def test_cleanup_is_finite_and_reports_unknown_cancel(tmp_path, owner, monkeypatch): ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request"}) + ci.write_json(tmp_path / "allocation.json", owner) now = [owner["cleanup_deadline"] - 5] monkeypatch.setattr(ci.time, "time", lambda: now[0]) calls = [] @@ -756,43 +779,36 @@ def run(root, operation, timeout): assert ci.cleanup(tmp_path) == 105 assert calls == [ ("cancel_request", 5), - ("resources", 4), - ("down", 3), - ("stop_api", 2), - ("remove_resources", 1), + ("stop_api", 4), + ("remove_resources", 3), ] receipt = ci.read_bound(tmp_path, "cleanup.json", owner) assert receipt["operations_succeeded"] is False - assert receipt["creator_quiescence"] == "UNKNOWN" + assert receipt["creator_quiescence"] == "RETIRED" assert receipt["operations"]["remove_resources"]["success"] is True -def test_api_cleanup_is_limited_to_its_ephemeral_runner( +def test_api_cleanup_uses_owned_enclosure_without_sdk_shutdown( tmp_path, owner, sky, monkeypatch ): - monkeypatch.setenv("GITHUB_ACTIONS", "true") - monkeypatch.setenv("RUNNER_ENVIRONMENT", "self-hosted") - with pytest.raises(ValueError, match="isolated GitHub-hosted"): - ci.worker(tmp_path, "stop_api") - sky.api_stop.assert_not_called() - monkeypatch.setenv("RUNNER_ENVIRONMENT", "github-hosted") - sky.api_status.return_value = [] + stop = Mock() + monkeypatch.setattr(ci.api_service, "stop", stop) ci.worker(tmp_path, "stop_api") - sky.api_stop.assert_called_once_with() + stop.assert_called_once_with(tmp_path, owner) + sky.api_stop.assert_not_called() + sky.api_status.assert_not_called() -def test_api_cleanup_detects_surviving_server_with_only_terminal_requests( - tmp_path, owner, sky, monkeypatch -): - monkeypatch.setenv("GITHUB_ACTIONS", "true") - monkeypatch.setenv("RUNNER_ENVIRONMENT", "github-hosted") - sky.api_status.side_effect = lambda **options: ( - [NS(request_id="request-17", status="CANCELLED")] - if options.get("all_status") - else [] +def test_missing_enclosure_blocks_check_and_launch(tmp_path, owner, sky, monkeypatch): + monkeypatch.setattr( + ci.api_service, + "client_environment", + Mock(side_effect=ValueError("unowned API")), ) - with pytest.raises(RuntimeError, match="still reports requests"): - ci.worker(tmp_path, "stop_api") + for operation in ("check", "launch"): + with pytest.raises(ValueError, match="unowned API"): + ci.worker(tmp_path, operation) + sky.launch.assert_not_called() @pytest.mark.parametrize("foreign", [False, True]) @@ -924,7 +940,7 @@ def run(root, operation, timeout): with pytest.raises(TimeoutError): ci.supervise(tmp_path, owner) assert late_jobs == [] - assert calls == ["launch", "cancel_request", "resources"] + assert calls == ["launch", "cancel_request", "stop_api"] @pytest.fixture @@ -1026,10 +1042,15 @@ def test_empty_resource_history_is_unknown(tmp_path, owner, kube_receipts): @pytest.mark.parametrize("stop_api", [True, False]) @pytest.mark.parametrize("receipt", [True, False]) -def test_cleanup_distinguishes_uid_absence_from_unknown_creator_lifetime( +def test_cleanup_requires_creator_retirement_before_uid_reconciliation( tmp_path, owner, monkeypatch, stop_api, receipt ): ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "allocation.json", owner) + if not stop_api: + monkeypatch.setattr( + ci.api_service, "require_retired", Mock(side_effect=ValueError("unproven")) + ) if receipt: ci.write_json( tmp_path / "resources-cleanup.json", @@ -1049,6 +1070,87 @@ def run(root, operation, timeout): assert ci.cleanup(tmp_path) == (0 if stop_api else 105) result = ci.read_bound(tmp_path, "cleanup.json", owner) assert result["operations_succeeded"] is stop_api - assert result["creator_quiescence"] == "UNKNOWN" - assert result["physical_absence"] == ("ABSENT" if receipt else "UNKNOWN") + assert result["creator_quiescence"] == ("RETIRED" if stop_api else "UNKNOWN") + assert result["physical_absence"] == ( + "ABSENT" if receipt and stop_api else "UNKNOWN" + ) assert result["physical_absence_scope"] == "retained Pod and Service UIDs" + + +@pytest.mark.parametrize("status", ["SUCCEEDED", "FAILED", "CANCELLED"]) +def test_completed_launch_teardown_precedes_creator_retirement( + tmp_path, owner, monkeypatch, status +): + ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "job.json", {**owner, "job_id": 17}) + ci.write_json(tmp_path / "result.json", {**owner, "job_id": 17, "status": status}) + ci.write_json(tmp_path / "allocation.json", owner) + calls = [] + + def run(root, operation, timeout): + calls.append((operation, timeout)) + + monkeypatch.setattr(ci, "run_worker", run) + assert ci.cleanup(tmp_path) == 0 + assert calls == [("down", 90), ("stop_api", 30), ("remove_resources", 90)] + + +def test_lost_launch_and_cancel_failure_retire_before_any_diagnostics( + tmp_path, owner, monkeypatch +): + ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "request.json", {**owner, "request_id": "request"}) + ci.write_json(tmp_path / "allocation.json", owner) + calls = [] + + def run(root, operation, timeout): + calls.append(operation) + if operation in {"launch", "cancel_request"}: + raise TimeoutError(operation) + if operation == "stop_api": + ci.write_json(root / "api-retired.json", owner) + + monkeypatch.setattr(ci, "run_worker", run) + with pytest.raises(TimeoutError): + ci.supervise(tmp_path, owner) + assert calls == ["launch", "cancel_request", "stop_api"] + assert ci.cleanup(tmp_path) == 0 + assert calls[-2:] == ["stop_api", "remove_resources"] + + +def test_startup_failure_without_launch_still_retires_owned_service( + tmp_path, owner, monkeypatch +): + ci.write_json(tmp_path / "api-plan.json", owner) + run = Mock() + monkeypatch.setattr(ci, "run_worker", run) + assert ci.cleanup(tmp_path) == 0 + assert run.call_args.args[1:] == ("stop_api", 30) + + +def test_missing_retirement_receipt_blocks_direct_cleanup(tmp_path, owner, monkeypatch): + ci.write_json(tmp_path / "launch-attempt.json", owner) + ci.write_json(tmp_path / "allocation.json", owner) + monkeypatch.setattr( + ci.api_service, "require_retired", Mock(side_effect=FileNotFoundError) + ) + run = Mock() + monkeypatch.setattr(ci, "run_worker", run) + assert ci.cleanup(tmp_path) == 105 + assert [call.args[1] for call in run.call_args_list] == ["stop_api"] + receipt = ci.read_bound(tmp_path, "cleanup.json", owner) + assert receipt["creator_quiescence"] == receipt["physical_absence"] == "UNKNOWN" + + +def test_unsupported_enclosure_blocks_sky_check_and_submission( + tmp_path, owner, monkeypatch +): + monkeypatch.setattr( + ci.api_service, "start", Mock(side_effect=ValueError("unsupported enclosure")) + ) + run = Mock() + monkeypatch.setattr(ci, "run_worker", run) + with pytest.raises(ValueError, match="unsupported enclosure"): + ci.main(tmp_path) + run.assert_not_called() + assert not (tmp_path / "launch-attempt.json").exists() diff --git a/tests/unit/test_trainer_rank_gpu_workflow.py b/tests/unit/test_trainer_rank_gpu_workflow.py index 52a54aa6b..c546f425c 100644 --- a/tests/unit/test_trainer_rank_gpu_workflow.py +++ b/tests/unit/test_trainer_rank_gpu_workflow.py @@ -31,3 +31,40 @@ def test_ready_for_review_reuse_requires_exact_base_artifact_and_gpu_job() -> No "`trainer-rank-result-${run.id}-${run.run_attempt}-${pull.base.sha}`" in workflow ) + + +def test_cpu_enclosure_probe_gates_gpu_and_failure_gate(): + import os + import subprocess + + import yaml + + workflow = yaml.safe_load( + ( + Path(__file__).parents[2] / ".github/workflows/trainer-rank-gpu.yml" + ).read_text() + ) + jobs = workflow["jobs"] + probe = jobs["api-enclosure"] + assert probe["runs-on"] == "ubuntu-latest" and "environment" not in probe + assert probe["timeout-minutes"] == 3 + assert "api-enclosure" in jobs["validate"]["needs"] + assert jobs["validate"]["env"]["SKYPILOT_DISABLE_LOCAL_API_SERVER"] == "1" + gate = jobs["gate"]["steps"][0]["run"] + # Even a previously reusable success must fail if the CPU enclosure fails. + result = subprocess.run( + ["bash", "-c", gate], + env={ + **os.environ, + "ENCLOSURE_RESULT": "failure", + "CLASSIFY_RESULT": "success", + "REQUIRED": "true", + "REUSED": "true", + "REUSED_RUN": "42", + "RESULT": "success", + }, + capture_output=True, + timeout=5, + ) + assert result.returncode == 1 + assert b"enclosure qualification failed" in result.stderr From ca504f32424f69844c1d17493cc0ee8bdf19dbc6 Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 03:52:39 +0000 Subject: [PATCH 6/8] fix(ci): correct Sky preflight and guard cleanup retries --- scripts/ci/trainer-rank-gpu.py | 31 +++-- tests/unit/test_trainer_rank_api.py | 1 + tests/unit/test_trainer_rank_gpu_ci.py | 179 ++++++++++++++++++++++++- 3 files changed, 200 insertions(+), 11 deletions(-) diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index cc7f4b115..7f44e0935 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -84,11 +84,12 @@ def resources(root, owner, *, delete): allocation = read_bound(root, "allocation.json", owner) namespace = allocation["namespace"] - retained = ( - read_bound(root, "resources.json", allocation)["observed"] - if (root / "resources.json").exists() - else [] - ) + retained = [] + for filename in ("resources.json", "resources-cleanup.json"): + if (root / filename).exists(): + for identity in read_bound(root, filename, allocation)["observed"]: + if identity not in retained: + retained.append(identity) target = root / ("resources-cleanup.json" if delete else "resources.json") receipt = { **allocation, @@ -122,11 +123,16 @@ def resources(root, owner, *, delete): "uid": metadata.uid, "cloud_name": metadata.labels.get("skypilot-cluster-name"), } - found.append(identity) if identity not in retained: retained.append(identity) # Retain each UID even if the next resource query fails. write_json(target, receipt) + if delete: + version = getattr(metadata, "resource_version", None) + if not isinstance(version, str) or not version: + raise ValueError("Kubernetes resource version is missing") + identity = {**identity, "resource_version": version} + found.append(identity) receipt["remaining"] = found write_json(target, receipt) if delete and not found: @@ -158,12 +164,17 @@ def resources(root, owner, *, delete): item["name"], namespace, body=client.V1DeleteOptions( - preconditions=client.V1Preconditions(uid=item["uid"]) + preconditions=client.V1Preconditions( + uid=item["uid"], + resource_version=item["resource_version"], + ) ), _request_timeout=(3, 10), ) except client.ApiException as error: - if error.status != 404: + # A changed version (including a relabelled same UID) must + # be observed under the nonce selector again before delete. + if error.status not in {404, 409}: raise time.sleep(1) finally: @@ -184,7 +195,9 @@ def worker(root, operation): import sky if operation == "check": - sky.get(sky.check(clouds=["kubernetes"])) + from sky.client import sdk + + sky.get(sdk.check(infra_list=("kubernetes",), verbose=False)) return cluster = owner["cluster"] if operation == "launch": diff --git a/tests/unit/test_trainer_rank_api.py b/tests/unit/test_trainer_rank_api.py index 5469980e1..b2fa518c5 100644 --- a/tests/unit/test_trainer_rank_api.py +++ b/tests/unit/test_trainer_rank_api.py @@ -12,6 +12,7 @@ SCRIPT = Path(__file__).parents[2] / "scripts/ci/trainer_rank_api.py" spec = importlib.util.spec_from_file_location("api_enclosure", SCRIPT) +assert spec is not None and spec.loader is not None api = importlib.util.module_from_spec(spec) spec.loader.exec_module(api) diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 6489aba92..cd7c80443 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -10,7 +10,7 @@ import sys import time from types import SimpleNamespace as NS -from unittest.mock import Mock +from unittest.mock import Mock, create_autospec import pytest @@ -47,6 +47,9 @@ def owner(tmp_path): @pytest.fixture def sky(monkeypatch): + def check(infra_list, verbose, workspace=None): + return "check-request" + task = Mock() api = NS( Task=NS(from_yaml=Mock(return_value=task)), @@ -62,6 +65,11 @@ def sky(monkeypatch): tail_logs=Mock(return_value=0), ) monkeypatch.setitem(sys.modules, "sky", api) + monkeypatch.setitem( + sys.modules, + "sky.client", + NS(sdk=NS(check=create_autospec(check, side_effect=check))), + ) monkeypatch.setitem( sys.modules, "sky.provision.kubernetes", @@ -479,7 +487,13 @@ def get(value): if value == "launch-request": return 17, NS(cluster_name=Path(os.environ["FAKE_CLUSTER"]).read_text()) return value -def check(**kwargs): return "checked" +class SDK: + @staticmethod + def check(infra_list, verbose, workspace=None): + assert infra_list == ("kubernetes",) + assert verbose is False and workspace is None + return "checked" +sys.modules["sky.client"] = NS(sdk=SDK) def job_status(cluster, *, job_ids): assert job_ids == [17] return {17:NS(value="SUCCEEDED")} @@ -820,6 +834,7 @@ def test_cleanup_resources_records_uids_and_confines_deletion( metadata=NS( name="pod", uid="original-uid", + resource_version="opaque:version-1", namespace="ci", labels={ci.LABEL: "peer" if foreign else owner["label"]}, ) @@ -950,6 +965,7 @@ def kube_receipts(tmp_path, owner, monkeypatch): metadata=NS( name="physical-head", uid="original-uid", + resource_version="opaque:version-1", namespace="ci", labels={ ci.LABEL: owner["label"], @@ -1154,3 +1170,162 @@ def test_unsupported_enclosure_blocks_sky_check_and_submission( ci.main(tmp_path) run.assert_not_called() assert not (tmp_path / "launch-attempt.json").exists() + + +def test_check_uses_pinned_sdk_signature(tmp_path, owner, sky): + ci.worker(tmp_path, "check") + assert not hasattr(sky, "check") # The pinned package does not export sdk.check. + sys.modules["sky.client"].sdk.check.assert_called_once_with( + infra_list=("kubernetes",), verbose=False + ) + sky.get.assert_called_once_with("check-request") + sky.launch.assert_not_called() + + +def test_failed_sky_check_prevents_launch(tmp_path, owner, monkeypatch): + run = Mock(side_effect=subprocess.CalledProcessError(1, "check")) + supervise = Mock() + monkeypatch.setattr(ci, "run_worker", run) + monkeypatch.setattr(ci, "supervise", supervise) + with pytest.raises(subprocess.CalledProcessError): + ci.main(tmp_path) + assert run.call_args.args[1] == "check" + supervise.assert_not_called() + assert not (tmp_path / "launch-attempt.json").exists() + + +@pytest.mark.parametrize("state", ["relabelled", "absent", "replaced"]) +def test_failed_cleanup_uids_survive_retry(tmp_path, owner, kube_receipts, state): + api, pod, error = kube_receipts + initial = { + "kind": "pod", + "name": "earlier-pod", + "uid": "earlier-uid", + "cloud_name": None, + } + ci.write_json( + tmp_path / "resources.json", {**owner, "namespace": "ci", "observed": [initial]} + ) + api.list_namespaced_service.side_effect = OSError("interrupted census") + with pytest.raises(OSError, match="interrupted census"): + ci.worker(tmp_path, "remove_resources") + # This later UID only exists in the failed cleanup receipt. It no longer + # appears under the nonce label on retry, so an exact name/UID read is needed. + api.list_namespaced_service.side_effect = None + api.list_namespaced_pod.return_value = NS(items=[]) + + def read(name, namespace, **kwargs): + if name == initial["name"] or state == "absent": + raise error(404) + assert name == pod.metadata.name + return NS( + metadata=NS( + name=name, + namespace=namespace, + uid=pod.metadata.uid if state == "relabelled" else "new-uid", + ) + ) + + api.read_namespaced_pod.side_effect = read + if state == "relabelled": + with pytest.raises(ValueError, match="absence is unproven"): + ci.worker(tmp_path, "remove_resources") + else: + ci.worker(tmp_path, "remove_resources") + receipt = ci.read_bound(tmp_path, "resources-cleanup.json", owner) + assert {item["uid"] for item in receipt["observed"]} == { + "earlier-uid", + pod.metadata.uid, + } + assert receipt["exact_uid_absence"] == ( + "UNKNOWN" if state == "relabelled" else "ABSENT" + ) + assert {call.args[0] for call in api.read_namespaced_pod.call_args_list} == { + initial["name"], + pod.metadata.name, + } + api.delete_namespaced_pod.assert_not_called() + + +@pytest.mark.parametrize("relabel", [False, True]) +def test_delete_recensuses_version_conflict_and_preserves_relabelled_uid( + tmp_path, owner, kube_receipts, monkeypatch, relabel +): + api, pod, error = kube_receipts + state = {"present": True, "changed": False} + deleted = [] + api.list_namespaced_pod.side_effect = lambda *a, **k: NS( + items=[pod] + if state["present"] and pod.metadata.labels[ci.LABEL] == owner["label"] + else [] + ) + + def services(*args, **kwargs): + if not state["changed"]: + state["changed"] = True + pod.metadata.resource_version = "opaque:version-2" + if relabel: + pod.metadata.labels[ci.LABEL] = "peer" + return NS(items=[]) + + api.list_namespaced_service.side_effect = services + + def delete(name, namespace, *, body, **kwargs): + assert body.preconditions.uid == pod.metadata.uid + version = getattr(body.preconditions, "resource_version", None) + if version is not None and version != pod.metadata.resource_version: + raise error(409) + deleted.append(pod.metadata.labels[ci.LABEL]) + state["present"] = False + + api.delete_namespaced_pod.side_effect = delete + + def read(*args, **kwargs): + if not state["present"]: + raise error(404) + return pod + + api.read_namespaced_pod.side_effect = read + monkeypatch.setattr(ci.time, "sleep", lambda _: None) + try: + ci.worker(tmp_path, "remove_resources") + except ValueError: + assert relabel + if relabel: + assert deleted == [], "same UID was deleted after its ownership label changed" + assert state["present"] + else: + assert deleted == [owner["label"]] + assert ( + api.delete_namespaced_pod.call_count == 2 + ) # Conflict, then fresh version. + assert api.list_namespaced_pod.call_count >= 2 + + +@pytest.mark.parametrize("version", [None, "", 7]) +def test_cleanup_requires_opaque_nonempty_resource_version( + tmp_path, owner, kube_receipts, version +): + api, pod, _ = kube_receipts + pod.metadata.resource_version = version + with pytest.raises(ValueError, match="version is missing"): + ci.worker(tmp_path, "remove_resources") + api.delete_namespaced_pod.assert_not_called() + # Losing a deletion precondition must not erase the already observed UID. + receipt = ci.read_bound(tmp_path, "resources-cleanup.json", owner) + assert receipt["observed"][0]["uid"] == pod.metadata.uid + assert receipt["exact_uid_absence"] == "UNKNOWN" + + +def test_cleanup_rejects_foreign_retained_history_before_census( + tmp_path, owner, kube_receipts +): + api, _, _ = kube_receipts + ci.write_json( + tmp_path / "resources-cleanup.json", + {**owner, "namespace": "peer", "observed": []}, + ) + with pytest.raises(ValueError, match="Wrong CI identity"): + ci.worker(tmp_path, "remove_resources") + api.list_namespaced_pod.assert_not_called() + api.delete_namespaced_pod.assert_not_called() From e3b0ff002cf1f390b8ca1eec3d14e29351fd49dd Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 04:26:02 +0000 Subject: [PATCH 7/8] fix(ci): load API helper by sibling path --- scripts/ci/trainer-rank-gpu.py | 8 +++- tests/unit/test_trainer_rank_gpu_ci.py | 53 ++++++++++++++++++++++---- 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/scripts/ci/trainer-rank-gpu.py b/scripts/ci/trainer-rank-gpu.py index 0726bdc85..7291f4aa0 100644 --- a/scripts/ci/trainer-rank-gpu.py +++ b/scripts/ci/trainer-rank-gpu.py @@ -7,6 +7,7 @@ must be retired before the final retained-UID reconciliation. """ +import importlib.util import json import os from pathlib import Path @@ -16,7 +17,12 @@ import time import uuid -import trainer_rank_api as api_service +spec = importlib.util.spec_from_file_location( + "trainer_rank_api", Path(__file__).with_name("trainer_rank_api.py") +) +assert spec is not None and spec.loader is not None +api_service = importlib.util.module_from_spec(spec) +spec.loader.exec_module(api_service) NONTERMINAL = {"INIT", "PENDING", "SETTING_UP", "RUNNING"} INFRAS = {"k8s/cks-wb3", "k8s/ext-collab2"} diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index cd7c80443..1892ea337 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -15,13 +15,49 @@ import pytest SCRIPT = Path(__file__).parents[2] / "scripts/ci/trainer-rank-gpu.py" -sys.path.insert(0, str(SCRIPT.parent)) spec = importlib.util.spec_from_file_location("trainer_rank_gpu_ci", SCRIPT) assert spec is not None and spec.loader is not None ci = importlib.util.module_from_spec(spec) spec.loader.exec_module(ci) +@pytest.mark.parametrize("name", ["trainer-rank-gpu.py", "gpu-image-cleanup.py"]) +def test_file_import_resolves_sibling_without_search_path(tmp_path, name): + subprocess.run( + [ + sys.executable, + "-I", + "-S", + "-c", + """ +import importlib.util +from pathlib import Path +import sys + +script = Path(sys.argv[1]).resolve() +before = sys.path.copy() +assert str(script.parent) not in sys.path +assert "trainer_rank_api" not in sys.modules +spec = importlib.util.spec_from_file_location("isolated_ci_script", script) +assert spec is not None and spec.loader is not None +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) +driver = module.gpu_ci if script.name == "gpu-image-cleanup.py" else module +assert Path(driver.api_service.__file__).resolve() == script.with_name("trainer_rank_api.py") +assert callable(driver.run_child) +assert sys.path == before +assert "sky" not in sys.modules +""", + str(SCRIPT.with_name(name)), + ], + cwd=tmp_path, + check=True, + capture_output=True, + text=True, + timeout=10, + ) + + @pytest.fixture(autouse=True) def service_boundary(monkeypatch): # Service controls have dedicated file-backed tests; never create a unit here. @@ -452,12 +488,15 @@ def test_main_rejects_invalid_scope_before_launch(tmp_path, monkeypatch, key, va @pytest.mark.parametrize("infra", ["k8s/cks-wb3", "k8s/ext-collab2"]) def test_real_worker_round_trip_with_fake_sdk(tmp_path, infra): """Exercise the actual direct Python parent/worker JSON transport, without Sky.""" - (tmp_path / "sitecustomize.py").write_text(f""" -import sys -sys.path.insert(0, {str(SCRIPT.parent)!r}) -import trainer_rank_api -trainer_rank_api.start = lambda *a, **kw: None -trainer_rank_api.client_environment = lambda *a, **kw: None + (tmp_path / "sitecustomize.py").write_text(""" +from importlib.machinery import SourceFileLoader +exec_module = SourceFileLoader.exec_module +def offline_service(loader, module): + exec_module(loader, module) + if module.__name__ == "trainer_rank_api": + module.start = lambda *a, **kw: None + module.client_environment = lambda *a, **kw: None +SourceFileLoader.exec_module = offline_service """) (tmp_path / "sky.py").write_text(""" import os From 104ab691c2343c058fdb5b9c066a8ca6f608fb13 Mon Sep 17 00:00:00 2001 From: Brad Hilton Date: Sat, 3 Oct 2026 05:04:54 +0000 Subject: [PATCH 8/8] test(ci): isolate Kubernetes in every fake SDK worker --- tests/unit/test_trainer_rank_gpu_ci.py | 37 +++++++++++++++++++------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/tests/unit/test_trainer_rank_gpu_ci.py b/tests/unit/test_trainer_rank_gpu_ci.py index 1892ea337..99001b1b2 100644 --- a/tests/unit/test_trainer_rank_gpu_ci.py +++ b/tests/unit/test_trainer_rank_gpu_ci.py @@ -490,6 +490,25 @@ def test_real_worker_round_trip_with_fake_sdk(tmp_path, infra): """Exercise the actual direct Python parent/worker JSON transport, without Sky.""" (tmp_path / "sitecustomize.py").write_text(""" from importlib.machinery import SourceFileLoader +import os +from pathlib import Path +import socket +import sys +from types import SimpleNamespace as NS +def deny_network(*args, **kwargs): raise AssertionError("Fake SDK worker must stay offline") +socket.socket.connect = socket.socket.connect_ex = socket.getaddrinfo = deny_network +def census(kind, namespace, **kwargs): + assert namespace == "default" and "sky" not in sys.modules + with Path(os.environ["FAKE_CENSUS"]).open("a") as stream: + stream.write(kind + "\\n") + return NS(items=[]) +sys.modules["kubernetes"] = NS( + client=NS(CoreV1Api=lambda _: NS( + list_namespaced_pod=lambda *a, **kw: census("pod", *a, **kw), + list_namespaced_service=lambda *a, **kw: census("service", *a, **kw), + )), + config=NS(new_client_from_config=lambda **kw: NS(close=lambda: None)), +) exec_module = SourceFileLoader.exec_module def offline_service(loader, module): exec_module(loader, module) @@ -500,20 +519,10 @@ def offline_service(loader, module): """) (tmp_path / "sky.py").write_text(""" import os -import socket import sys from pathlib import Path from types import SimpleNamespace as NS -def deny_network(*args, **kwargs): raise AssertionError("Fake SDK worker must stay offline") -socket.socket.connect = socket.getaddrinfo = deny_network sys.modules["sky.provision.kubernetes"] = NS(utils=NS(get_namespace=lambda **_: "default")) -sys.modules["kubernetes"] = NS( - client=NS(CoreV1Api=lambda _: NS( - list_namespaced_pod=lambda *a, **kw: NS(items=[]), - list_namespaced_service=lambda *a, **kw: NS(items=[]), - )), - config=NS(new_client_from_config=lambda **kw: NS(close=lambda: None)), -) class Task: @classmethod def from_yaml(cls, path): return cls() @@ -545,10 +554,17 @@ def tail_logs(cluster, *, job_id, follow): head = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=repo, text=True ).strip() + kubeconfig = tmp_path / "empty-kubeconfig.yaml" + kubeconfig.write_text( + "apiVersion: v1\nkind: Config\nclusters: []\ncontexts: []\nusers: []\n" + ) + census = tmp_path / "census" env = { **os.environ, "PYTHONPATH": str(tmp_path), + "KUBECONFIG": str(kubeconfig), "FAKE_CLUSTER": str(tmp_path / "cluster"), + "FAKE_CENSUS": str(census), "GITHUB_RUN_ID": "42", "GITHUB_RUN_ATTEMPT": "2", "GITHUB_REPOSITORY": "OpenPipe/ART", @@ -570,6 +586,7 @@ def tail_logs(cluster, *, job_id, follow): assert json.loads((root / "result.json").read_text())["job_id"] == 17 assert json.loads((root / "result.json").read_text())["infra"] == infra assert json.loads((root / "resources.json").read_text())["remaining"] == [] + assert census.read_text().splitlines() == ["pod", "service"] assert "fake complete log" in (root / "logs.log").read_text()