From 37129fb3d93ba5aff92c9468ae152c14d59beab8 Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Mon, 22 Jun 2026 18:44:55 +0200 Subject: [PATCH] fix(k8s): surface the real asset failure cause, not a bare "Job failed" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a child Job dies without streaming its terminal, the host authored a generic "Job failed" with no traceback. The child's events (logs and the rich asset_failed terminal) reach the DB only by being scraped off the pod's live log stream; that follow-stream has no reconnection and silently drops on long retries or API-server idle timeouts, so the real error — e.g. a persistent Amazon Ads 429 exhausting a 1h tenacity budget — was lost and the UI showed only "Job failed". On failure with no child terminal, recover the cause best-first: 1. re-read the pod's final logs (non-follow) and parse the child's own asset_failed/asset_exec_failed event for error + traceback; 2. fall back to the container termination state — reason/exit_code (so OOMKilled/137 is no longer hidden) and the log tail as the traceback. Set terminationMessagePolicy=FallbackToLogsOnError so Kubernetes captures that log tail even when the stream is already gone. Degrades to the original bare fallback when no job/client/pod is available, so nothing regresses. The frontend already renders error + traceback, so these surface automatically. --- .../src/interloper_k8s/runner.py | 148 +++++++++++++++++- .../tests/test_runner_terminal_events.py | 96 ++++++++++++ 2 files changed, 241 insertions(+), 3 deletions(-) diff --git a/packages/interloper-k8s/src/interloper_k8s/runner.py b/packages/interloper-k8s/src/interloper_k8s/runner.py index 093a3b38..f6bcc5fe 100644 --- a/packages/interloper-k8s/src/interloper_k8s/runner.py +++ b/packages/interloper-k8s/src/interloper_k8s/runner.py @@ -130,6 +130,11 @@ def _submit_asset( args=cmd[1:], env=env if env else None, resources=resources, + # On a non-zero exit with an empty termination-log, Kubernetes copies + # the tail of the container logs into the pod's termination state, so + # the host can recover a real cause even when the live log stream that + # carries child events has already dropped. + termination_message_policy="FallbackToLogsOnError", ) pod_spec = client.V1PodSpec( @@ -243,9 +248,10 @@ def _handle_completed(self, future: Future[Any], asset: Asset) -> None: try: future.result() except Exception as e: - self.state.mark_asset_failed(asset, str(e), emit=True) + error, tb = self._recover_asset_failure(job_name, asset.id, fallback=str(e)) + self.state.mark_asset_failed(asset, error, tb=tb, emit=True) if self.fail_fast or self.reraise: - raise RunnerError(f"Asset '{type(asset).key}' failed: {e}") from e + raise RunnerError(f"Asset '{type(asset).key}' failed: {error}") from e else: self.state.mark_asset_completed(asset, emit=True) @@ -265,7 +271,8 @@ def _handle_flushed_future(self, future: Future[Any], asset: Asset) -> None: try: future.result() except Exception as e: # noqa: BLE001 - self.state.mark_asset_failed(asset, str(e), emit=True) + error, tb = self._recover_asset_failure(job_name, asset.id, fallback=str(e)) + self.state.mark_asset_failed(asset, error, tb=tb, emit=True) else: self.state.mark_asset_completed(asset, emit=True) @@ -299,6 +306,141 @@ def _poll_job(self, job_name: str) -> None: raise RunnerError(f"Job {job_name} stopped (runner shutting down)") + # ------------------------------------------------------------------ + # Failure recovery + # ------------------------------------------------------------------ + + def _recover_asset_failure( + self, + job_name: str | None, + target_asset_id: str, + *, + fallback: str, + ) -> tuple[str, str | None]: + """Recover the real failure cause from a failed child Job's pod. + + The child streams its rich terminal (error + traceback) over the live + pod-log stream, but that follow-stream can drop while the pod is still + running (long retries, API-server idle timeouts), leaving the host with + only the Job status — a bare ``"Job ... failed"`` and no traceback. + + On failure we re-read the pod's final logs (non-follow) and, failing + that, its container termination state, to recover the real error rather + than the Job status alone. + + Returns: + ``(error, traceback)`` — the richest available; falls back to + ``fallback`` with no traceback when nothing can be recovered. + """ + if job_name is None or self._core_v1 is None: + return fallback, None + + pod = self._find_pod(job_name) + if pod is None: + return fallback, None + + pod_name = pod.metadata.name if pod.metadata else None + + # 1. Rich path: the child's own terminal event, re-read from the final + # logs the live stream missed. + if pod_name is not None: + error, tb = self._terminal_from_pod_logs(pod_name, target_asset_id) + if error is not None: + return error, tb + + # 2. Fallback: the container's termination state. With + # terminationMessagePolicy=FallbackToLogsOnError, ``message`` carries + # the tail of the logs; ``reason``/``exit_code`` flag OOMKills, signals. + return self._terminal_from_termination_state(pod, fallback) + + def _find_pod(self, job_name: str) -> Any | None: + """Return the (first) pod owned by ``job_name``, or ``None``.""" + if self._core_v1 is None: + return None + try: + pods = cast( + client.V1PodList, + self._core_v1.list_namespaced_pod( + namespace=self.namespace, + label_selector=f"job-name={job_name}", + ), + ) + except Exception: # noqa: BLE001 + return None + return pods.items[0] if pods.items else None + + def _terminal_from_pod_logs( + self, + pod_name: str, + target_asset_id: str, + ) -> tuple[str | None, str | None]: + """Re-read the pod's final logs and parse the child's terminal event. + + Returns the error/traceback of the last ``asset_failed`` / + ``asset_exec_failed`` event for the target asset, or ``(None, None)`` + when the logs hold no parseable terminal (e.g. the pod was SIGKILLed). + """ + if self._core_v1 is None: + return None, None + try: + logs = cast( + str, + self._core_v1.read_namespaced_pod_log( + name=pod_name, + namespace=self.namespace, + container="interloper", + tail_lines=2000, + ), + ) + except Exception: # noqa: BLE001 + return None, None + + error: str | None = None + tb: str | None = None + for line in logs.splitlines(): + line = line.rstrip() + if not line: + continue + try: + event = parse_event_from_log_line(line) + except Exception: # noqa: BLE001 + continue + if event is None or event.type not in (EventType.ASSET_FAILED, EventType.ASSET_EXEC_FAILED): + continue + event_asset_id = event.metadata.get("asset_id") + if event_asset_id and event_asset_id != target_asset_id: + continue + err = event.metadata.get("error") + if err: + error = err + tb = event.metadata.get("traceback") or tb + return error, tb + + def _terminal_from_termination_state(self, pod: Any, fallback: str) -> tuple[str, str | None]: + """Build a cause from the pod's terminated container state. + + ``reason``/``exit_code`` are appended to ``fallback`` (so an OOMKill or + signal is no longer hidden behind a bare "Job failed"); the termination + ``message`` — the log tail under FallbackToLogsOnError — becomes the + traceback shown in the UI. + """ + status = getattr(pod, "status", None) + if status is None: + return fallback, None + for cs in status.container_statuses or []: + term = cs.state.terminated if cs.state else None + if term is None: + continue + bits: list[str] = [] + if term.reason: + bits.append(f"reason={term.reason}") + if term.exit_code is not None: + bits.append(f"exit_code={term.exit_code}") + error = f"{fallback} ({', '.join(bits)})" if bits else fallback + tb = term.message or None + return error, tb + return fallback, None + # ------------------------------------------------------------------ # Job helpers # ------------------------------------------------------------------ diff --git a/packages/interloper-k8s/tests/test_runner_terminal_events.py b/packages/interloper-k8s/tests/test_runner_terminal_events.py index 7fc53df7..44543802 100644 --- a/packages/interloper-k8s/tests/test_runner_terminal_events.py +++ b/packages/interloper-k8s/tests/test_runner_terminal_events.py @@ -11,6 +11,8 @@ from collections.abc import Iterator from concurrent.futures import Future from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import MagicMock import interloper as il from interloper.errors import RunnerError @@ -135,3 +137,97 @@ def test_no_fail_fast_keeps_quiet_on_child_reported_failure() -> None: runner._handle_completed(future, asset) # must not raise assert not [e for e in events if e.type == EventType.ASSET_FAILED] + + +# ---------------------------------------------------------------------- +# Failure recovery: the host-authored terminal carries the pod's real cause +# (not a bare "Job ... failed") when the live event stream dropped before the +# child reported its terminal. +# ---------------------------------------------------------------------- + + +def _pod(name: str = "pod-x", *, terminated: SimpleNamespace | None = None) -> SimpleNamespace: + container_statuses = None + if terminated is not None: + container_statuses = [SimpleNamespace(state=SimpleNamespace(terminated=terminated))] + return SimpleNamespace( + metadata=SimpleNamespace(name=name), + status=SimpleNamespace(container_statuses=container_statuses), + ) + + +def _mock_core_v1(*, pod: SimpleNamespace | None = None, logs: str = "") -> MagicMock: + core = MagicMock() + core.list_namespaced_pod.return_value = SimpleNamespace(items=[pod] if pod is not None else []) + core.read_namespaced_pod_log.return_value = logs + return core + + +def _failed_event_line(asset_id: str, error: str, traceback: str | None = None) -> str: + metadata: dict[str, str] = {"asset_id": asset_id, "error": error} + if traceback is not None: + metadata["traceback"] = traceback + return Event(type=EventType.ASSET_FAILED, metadata=metadata).to_json() + + +def test_recover_returns_fallback_without_job_or_client() -> None: + """No job name (or no k8s client) → the bare fallback, no traceback.""" + runner, _ = _runner_with_asset("asset-1", "run-1") + assert runner._recover_asset_failure(None, "asset-1", fallback="Job x failed") == ("Job x failed", None) + + runner._core_v1 = _mock_core_v1() + assert runner._recover_asset_failure(None, "asset-1", fallback="Job x failed") == ("Job x failed", None) + + +def test_recover_reads_rich_terminal_from_pod_logs() -> None: + """The child's own error + traceback are recovered from the final logs.""" + runner, _ = _runner_with_asset("asset-1", "run-1") + line = _failed_event_line("asset-1", "429 Too Many Requests", "Traceback...\nHTTPStatusError") + runner._core_v1 = _mock_core_v1(pod=_pod(), logs=f"some debug log\n{line}\n") + + error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed") + + assert error == "429 Too Many Requests" + assert tb is not None and "HTTPStatusError" in tb + + +def test_recover_ignores_other_assets_terminal_in_logs() -> None: + """A terminal for a different asset must not be attributed to this one.""" + runner, _ = _runner_with_asset("asset-1", "run-1") + term = SimpleNamespace(reason="Error", exit_code=1, message="log tail") + runner._core_v1 = _mock_core_v1(pod=_pod(terminated=term), logs=_failed_event_line("other-asset", "not mine")) + + error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed") + + # No matching log terminal → falls through to the termination state. + assert error == "Job job-x failed (reason=Error, exit_code=1)" + assert tb == "log tail" + + +def test_recover_falls_back_to_termination_state_on_oomkill() -> None: + """A SIGKILLed pod has no terminal event; reason/exit_code surface instead.""" + runner, _ = _runner_with_asset("asset-1", "run-1") + term = SimpleNamespace(reason="OOMKilled", exit_code=137, message="killed log tail") + runner._core_v1 = _mock_core_v1(pod=_pod(terminated=term), logs="no events here\n") + + error, tb = runner._recover_asset_failure("job-x", "asset-1", fallback="Job job-x failed") + + assert error == "Job job-x failed (reason=OOMKilled, exit_code=137)" + assert tb == "killed log tail" + + +def test_handle_completed_emits_recovered_error_and_traceback() -> None: + """End to end: the host-authored ``asset_failed`` carries the recovered cause.""" + runner, asset = _runner_with_asset("asset-1", "run-1") + runner._core_v1 = _mock_core_v1(pod=_pod(), logs=_failed_event_line("asset-1", "429 Too Many Requests", "rich-tb")) + future: Future[None] = Future() + future.set_exception(RunnerError("Job interloper-run-x failed")) + runner._job_map[future] = "interloper-run-x" + + with _capture() as events: + runner._handle_completed(future, asset) + + failed = [e for e in events if e.type == EventType.ASSET_FAILED] + assert len(failed) == 1 + assert failed[0].metadata.get("error") == "429 Too Many Requests" + assert failed[0].metadata.get("traceback") == "rich-tb"