diff --git a/tools/olf/olf/e2e.py b/tools/olf/olf/e2e.py index 53905c7..ad81b63 100644 --- a/tools/olf/olf/e2e.py +++ b/tools/olf/olf/e2e.py @@ -12,6 +12,7 @@ import shutil import subprocess import time +import uuid from collections.abc import Callable, Mapping from dataclasses import dataclass from pathlib import Path @@ -82,6 +83,11 @@ ) DAGSTER_JOB_TIMEOUT_SECONDS = 1800 +READINESS_ATTEMPTS = 60 +READINESS_DELAY_SECONDS = 5 +DIAGNOSTIC_LOG_LINES = 80 +LAUNCH_RETRY_ATTEMPTS = 4 +LAUNCH_RETRY_DELAY_SECONDS = 3 class E2EError(RuntimeError): @@ -92,6 +98,75 @@ class DagsterTransientError(E2EError): pass +class DagsterHTTPError(E2EError): + """A non-transient Dagster HTTP response (normally a client error).""" + + +def workload_health_class(item: Mapping[str, Any], suite_jobs: tuple[str, ...] = PRODUCT_JOBS) -> str: + """Classify a Kubernetes workload for E2E readiness policy. + + ReplicaSet/StatefulSet pods and OpenLakeForge bootstrap Jobs are required. + Other Job pods are warnings unless their name/labels identify a job owned + by this suite. + """ + metadata = item.get("metadata", {}) + owner_kinds = {str(owner.get("kind")) for owner in metadata.get("ownerReferences", [])} + if "Job" not in owner_kinds: + return "required-service" + labels = metadata.get("labels", {}) + if labels.get("openlakeforge.io/job"): + return "platform-owned-job" + identity = " ".join( + str(value) + for value in ( + metadata.get("name", ""), + labels.get("job-name", ""), + labels.get("dagster/job", ""), + *(owner.get("name", "") for owner in metadata.get("ownerReferences", [])), + ) + ) + if any(job in identity for job in suite_jobs): + return "suite-owned-job" + return "independent-job" + + +def _bounded_pod_diagnostics(cfg: E2EConfig, names: list[str]) -> str: + lines: list[str] = [] + for name in names[:10]: + try: + output = kubectl( + cfg, + ["logs", f"pod/{name}", "--all-containers", f"--tail={DIAGNOSTIC_LOG_LINES}"], + capture=True, + ) + except E2EError as exc: + output = f"unable to collect logs: {exc}" + lines.append(f"{name}:\n{output[-6000:]}") + return "\n".join(lines) + + +def _bounded_job_diagnostics(cfg: E2EConfig, payload: Mapping[str, Any]) -> str: + """Collect only a small tail from failed independent Jobs.""" + diagnostics: list[str] = [] + for item in payload.get("items", []): + if workload_health_class(item) != "independent-job": + continue + status = item.get("status", {}).get("phase") + if status not in {"Failed", "Unknown"}: + continue + metadata = item.get("metadata", {}) + job_name = metadata.get("labels", {}).get("job-name") or next( + (owner.get("name") for owner in metadata.get("ownerReferences", []) if owner.get("kind") == "Job"), + metadata.get("name", "unknown"), + ) + try: + output = kubectl(cfg, ["logs", f"job/{job_name}", f"--tail={DIAGNOSTIC_LOG_LINES}"], capture=True) + except E2EError as exc: + output = f"unable to collect logs: {exc}" + diagnostics.append(f"{job_name} ({status}):\n{output[-6000:]}") + return "\n".join(diagnostics) + + @dataclass(frozen=True) class E2EConfig: env: Environment @@ -276,21 +351,54 @@ def run_full(cfg: E2EConfig) -> None: def check_pods_ready(cfg: E2EConfig) -> None: log.step("Checking pod health...") bad: list[str] = [] + warned: list[str] = [] last_error: E2EError | None = None - for _ in range(60): + for _ in range(READINESS_ATTEMPTS): try: payload = json.loads(kubectl(cfg, ["get", "pods", "-n", cfg.namespace, "-o", "json"], capture=True)) except E2EError as exc: last_error = exc time.sleep(5) continue - bad = unhealthy_pod_messages(payload) + bad, warned = classify_pod_health(payload, suite_jobs=PRODUCT_JOBS) + for message in warned: + log.warn(message) + if warned: + diagnostics = _bounded_job_diagnostics(cfg, payload) + if diagnostics: + log.warn(f"independent Job diagnostics:\n{diagnostics}") if not bad: return - time.sleep(5) + time.sleep(READINESS_DELAY_SECONDS) if last_error is not None and not bad: raise last_error - raise E2EError("unhealthy pods:\n" + "\n".join(bad)) + required_names = [message.split(":", 1)[0] for message in bad] + diagnostics = _bounded_pod_diagnostics(cfg, required_names) + raise E2EError("unhealthy required services:\n" + "\n".join(bad) + "\nDiagnostics:\n" + diagnostics) + + +def classify_pod_health( + payload: Mapping[str, Any], *, suite_jobs: tuple[str, ...] = PRODUCT_JOBS +) -> tuple[list[str], list[str]]: + """Return blocking service failures and bounded warning messages for Jobs.""" + bad: list[str] = [] + warned: list[str] = [] + for item in payload.get("items", []): + name = str(item.get("metadata", {}).get("name", "")) + phase = item.get("status", {}).get("phase") + health_class = workload_health_class(item, suite_jobs) + message = unhealthy_pod_messages({"items": [item]}) + if not message: + continue + if health_class != "independent-job": + bad.extend(message) + else: + labels = item.get("metadata", {}).get("labels", {}) + warned.append( + f"unrelated Job {name} ({labels.get('job-name', labels.get('dagster/pipeline', 'unknown owner'))}) " + f"is {phase}; continuing E2E readiness" + ) + return bad, warned def unhealthy_pod_messages(payload: Mapping[str, Any]) -> list[str]: @@ -406,11 +514,32 @@ def launch_and_poll_dagster_jobs(cfg: E2EConfig) -> None: client = DagsterClient(f"{base_url}/graphql") timeout_seconds = int(os.environ.get("DAGSTER_JOB_TIMEOUT_SECONDS", str(DAGSTER_JOB_TIMEOUT_SECONDS))) for job in PRODUCT_JOBS: - run_id = client.launch(job) + try: + run_id = client.launch(job) + except E2EError as exc: + diagnostics = _bounded_pod_diagnostics( + cfg, + ["dagster-dagster-webserver", *expected_user_code_pods(cfg)], + ) + raise E2EError(f"{exc}\nDagster diagnostics:\n{diagnostics}") from exc log.info(f"{job}: launched ({run_id})") client.poll(job, run_id, timeout_seconds=timeout_seconds) +def expected_user_code_pods(cfg: E2EConfig) -> list[str]: + """Discover user-code deployments for bounded failure diagnostics.""" + try: + raw = kubectl(cfg, ["get", "pods", "-n", cfg.namespace, "-o", "json"], capture=True) + payload = json.loads(raw) + except (E2EError, json.JSONDecodeError): + return [] + return [ + str(item.get("metadata", {}).get("name")) + for item in payload.get("items", []) + if "dagster-user-deployments" in str(item.get("metadata", {}).get("name", "")) + ][:5] + + class DagsterClient: def __init__( self, @@ -423,7 +552,25 @@ def __init__( def launch(self, job_name: str) -> str: location_name, repository_name = self.wait_for_repository(job_name) - result = self.graphql( + launch_key = f"openlakeforge-e2e-{job_name}-{uuid.uuid4().hex}" + variables = { + "executionParams": { + "selector": { + "repositoryLocationName": location_name, + "repositoryName": repository_name, + "pipelineName": job_name, + }, + "runConfigData": {}, + "mode": "default", + "executionMetadata": { + "tags": [{"key": "openlakeforge/e2e-key", "value": launch_key}], + }, + } + } + last_error: DagsterTransientError | None = None + for attempt in range(LAUNCH_RETRY_ATTEMPTS): + try: + result = self.graphql( """ mutation LaunchRun($executionParams: ExecutionParams!) { launchRun(executionParams: $executionParams) { @@ -434,22 +581,44 @@ def launch(self, job_name: str) -> str: } } """, - { - "executionParams": { - "selector": { - "repositoryLocationName": location_name, - "repositoryName": repository_name, - "pipelineName": job_name, - }, - "runConfigData": {}, - "mode": "default", - } - }, - )["launchRun"] + variables, + )["launchRun"] + break + except DagsterTransientError as exc: + last_error = exc + existing = self.find_run_by_tag(launch_key) + if existing: + return existing + if attempt + 1 == LAUNCH_RETRY_ATTEMPTS: + raise E2EError( + f"Dagster launch for {job_name} failed after {LAUNCH_RETRY_ATTEMPTS} attempts: {exc}" + ) from exc + time.sleep(LAUNCH_RETRY_DELAY_SECONDS) + else: # pragma: no cover - loop always breaks or raises + raise E2EError(f"Dagster launch failed: {last_error}") if result["__typename"] != "LaunchRunSuccess": raise E2EError(f"failed to launch {job_name}: {json.dumps(result, indent=2)}") return result["run"]["runId"] + def find_run_by_tag(self, launch_key: str) -> str | None: + """Find a run created before an ambiguous HTTP response.""" + try: + result = self.graphql( + """ + query ExistingRun($key: String!) { + runsOrError(filter: {tags: [{key: "openlakeforge/e2e-key", value: $key}]}, limit: 1) { + __typename + ... on Runs { results { runId } } + } + } + """, + {"key": launch_key}, + )["runsOrError"] + runs = result.get("results", []) + return str(runs[0]["runId"]) if runs else None + except E2EError: + return None + def wait_for_repository( self, job_name: str, @@ -469,7 +638,8 @@ def wait_for_repository( detail = f": {last_error}" if last_error else "" raise E2EError( f"Dagster repository for {job_name} did not become ready " - f"within {timeout_seconds} seconds{detail}" + f"within {timeout_seconds} seconds{detail}. Required code location may be unavailable; " + "inspect Dagster webserver and user-code logs." ) def try_reload_repository_location(self, location_name: str) -> None: @@ -601,7 +771,12 @@ def _requests_graphql(self, query: str, variables: Mapping[str, Any] | None) -> timeout=30, ) response.raise_for_status() - except requests.RequestException as exc: + except requests.HTTPError as exc: + status = exc.response.status_code if exc.response is not None else None + if status is not None and 500 <= status < 600: + raise DagsterTransientError(f"Dagster GraphQL HTTP {status}: {exc}") from exc + raise DagsterHTTPError(f"Dagster GraphQL HTTP {status}: {exc}") from exc + except (requests.ConnectionError, requests.Timeout) as exc: raise DagsterTransientError(f"Dagster GraphQL request failed: {exc}") from exc return response.json() def check_superset_dashboards(cfg: E2EConfig) -> None: diff --git a/tools/olf/tests/test_e2e.py b/tools/olf/tests/test_e2e.py index 509a15b..19c2159 100644 --- a/tools/olf/tests/test_e2e.py +++ b/tools/olf/tests/test_e2e.py @@ -50,6 +50,70 @@ def test_unhealthy_pod_messages_reports_unready_and_failed_pods() -> None: ] +def test_classify_pod_health_warns_for_unrelated_failed_job() -> None: + bad, warned = e2e.classify_pod_health( + { + "items": [ + { + "metadata": { + "name": "om-job-pod", + "ownerReferences": [{"kind": "Job", "name": "om-job-aws-glue-metadata-ingestion"}], + "labels": {"job-name": "om-job-aws-glue-metadata-ingestion"}, + }, + "status": {"phase": "Failed"}, + } + ] + } + ) + assert bad == [] + assert "unrelated Job om-job-pod" in warned[0] + + +def test_classify_pod_health_blocks_suite_owned_job() -> None: + bad, warned = e2e.classify_pod_health( + { + "items": [ + { + "metadata": { + "name": "run-pod", + "ownerReferences": [{"kind": "Job", "name": "sales_order_revenue_pipeline-run"}], + }, + "status": {"phase": "Failed"}, + } + ] + } + ) + assert warned == [] + assert bad == ["run-pod: Failed"] + + +def test_classify_pod_health_blocks_platform_bootstrap_job() -> None: + bad, warned = e2e.classify_pod_health( + { + "items": [ + { + "metadata": { + "name": "polaris-bootstrap-pod", + "ownerReferences": [{"kind": "Job", "name": "polaris-bootstrap"}], + "labels": { + "job-name": "polaris-bootstrap", + "openlakeforge.io/job": "catalog-bootstrap", + }, + }, + "status": {"phase": "Failed"}, + } + ] + } + ) + + assert warned == [] + assert bad == ["polaris-bootstrap-pod: Failed"] + + +def test_workload_health_classifies_required_service() -> None: + assert e2e.workload_health_class({"metadata": {"name": "dagster-webserver"}}) == "required-service" + + def test_check_pods_ready_retries_until_pods_are_ready( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -455,6 +519,54 @@ def request_json(_query: str, _variables: dict[str, Any] | None = None) -> dict[ assert responses == [] +def test_dagster_launch_retries_transient_failure_and_keeps_launch_tag(monkeypatch: pytest.MonkeyPatch) -> None: + responses: list[Exception | dict[str, Any]] = [ + e2e.DagsterTransientError("HTTP 503"), + {"data": {"launchRun": {"__typename": "LaunchRunSuccess", "run": {"runId": "run-1"}}}}, + ] + calls: list[dict[str, Any]] = [] + + def request_json(query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]: + if "query Workspace" in query: + return { + "data": { + "workspaceOrError": { + "__typename": "Workspace", + "locationEntries": [ + { + "name": "sales-dagster", + "locationOrLoadError": { + "__typename": "RepositoryLocation", + "repositories": [ + { + "name": "__repository__", + "jobs": [{"name": "sales_order_revenue_pipeline"}], + } + ], + }, + } + ], + } + } + } + if "ExistingRun" in query: + return {"data": {"runsOrError": {"__typename": "Runs", "results": []}}} + calls.append(variables or {}) + response = responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + monkeypatch.setattr(e2e, "LAUNCH_RETRY_DELAY_SECONDS", 0) + run_id = e2e.DagsterClient("http://dagster/graphql", request_json=request_json).launch( + "sales_order_revenue_pipeline" + ) + assert run_id == "run-1" + assert calls[0]["executionParams"]["executionMetadata"] == calls[1]["executionParams"]["executionMetadata"] + assert calls[0]["executionParams"]["executionMetadata"]["tags"][0]["key"] == "openlakeforge/e2e-key" + assert calls[0]["executionParams"].get("tags") is None + + def test_dagster_poll_times_out_quickly_for_non_terminal_runs() -> None: client = e2e.DagsterClient( "http://dagster/graphql",