From b296fcaf1cb41cbabd6d24170c579f948b055895 Mon Sep 17 00:00:00 2001 From: Ben Copeland Date: Tue, 1 Sep 2026 15:45:09 +0100 Subject: [PATCH] mcp: read job logs from Maestro when the dashboard has none A job that fails before producing results never reaches the dashboard, so get_log could not fetch its log at all: get_log and get_build both return "not found", while Maestro holds it as an incomplete job with its logs attached. Read those instead when the dashboard has no test for the id. The lava_log artifact is a plain gzipped log, so the existing streaming path reads it with the bounds and deadline it already applies. Where there is no lava_log the LAVA callback carries the log as a JSON field, which has to be read whole: that path is capped on both the download and the decompressed size. Sampling incomplete jobs, the two together cover every one that has any log artifact at all, and the pipeline treats the callback's log field as temporary, so it is the fallback rather than the first choice. Only bare or maestro-prefixed ids are looked up. Stripping any origin would let a missing other-origin: resolve to an unrelated Maestro and return the wrong log. Reported-by: Yogesh Lal Signed-off-by: Ben Copeland --- kcidev/api.py | 148 ++++++++++++++++++++++- kcidev/mcp/tools_dashboard.py | 6 +- tests/test_api.py | 221 ++++++++++++++++++++++++++++++++++ 3 files changed, 370 insertions(+), 5 deletions(-) diff --git a/kcidev/api.py b/kcidev/api.py index 7b1f620..d269b85 100644 --- a/kcidev/api.py +++ b/kcidev/api.py @@ -9,6 +9,7 @@ """ import ipaddress +import json import socket import zlib from datetime import datetime, timezone @@ -117,6 +118,7 @@ def _as_library_error(action, func, *args, **kwargs): ) MAX_LOG_BYTES = 1 << 20 +MAX_CALLBACK_BYTES = 8 << 20 LOG_SCAN_LIMIT = 64 << 20 LOG_DEADLINE_SECONDS = 60 _LOG_CHUNK = 1 << 16 @@ -194,6 +196,27 @@ def _gunzip_iter(raw_iter): raise KciDevError("Log gzip stream is incomplete or malformed") +def _maestro_node_id(node_id): + """Return the Maestro node id, or None when the id is not Maestro's. + + The dashboard prefixes ids with their origin while Maestro takes the + bare hex, so only a bare id or an explicitly maestro-prefixed one may + be looked up. Stripping any prefix would let a missing + other-origin: resolve to an unrelated Maestro . + """ + if ":" not in node_id: + return node_id + origin, _, rest = node_id.partition(":") + return rest if origin == "maestro" else None + + +def _bounded_text(raw, max_bytes, tail): + """Apply the head/tail bound to text already held in memory.""" + if len(raw) <= max_bytes: + return raw, len(raw) + return (raw[-max_bytes:] if tail else raw[:max_bytes]), len(raw) + + def _pick_log_url(test): if not isinstance(test, dict): return None @@ -496,6 +519,15 @@ def get_log(self, test_id, max_bytes=16384, tail=True): ``total_bytes`` is likewise a floor. The log URL is validated (scheme and resolved address) before fetching, though a DNS rebind between that check and the request remains a residual gap. + + A job that failed before producing results has no dashboard test, + so its log is read from Maestro instead: the ``lava_log`` + artifact where present, which is a plain log this same path + streams, otherwise the log field of the LAVA callback. The + returned ``source`` says which was used, and for the callback + ``log_url`` is that document rather than a log file. Only bare + or maestro-prefixed ids are looked up, so a missing test from + another origin keeps its dashboard error. """ if isinstance(max_bytes, bool) or not isinstance(max_bytes, int): raise KciDevError("max_bytes must be a positive integer") @@ -503,10 +535,21 @@ def get_log(self, test_id, max_bytes=16384, tail=True): raise KciDevError("max_bytes must be a positive integer") max_bytes = min(max_bytes, MAX_LOG_BYTES) - test = self.get_test(test_id) - log_url = _pick_log_url(test) - if not log_url: - raise KciDevError(f"No log available for test {test_id}") + try: + test = self.get_test(test_id) + except KciDevError as exc: + if "not found" not in str(exc).lower(): + raise + if _maestro_node_id(test_id) is None: + raise + log_url, log_source, callback = self._job_log_source(test_id, exc) + if callback is not None: + return self._callback_log(test_id, callback, max_bytes, tail) + else: + log_url = _pick_log_url(test) + log_source = "dashboard" + if not log_url: + raise KciDevError(f"No log available for test {test_id}") buf = bytearray() total = 0 @@ -583,6 +626,103 @@ def raw_iter(): "scan_limited": scan_limited, "deadline_exceeded": deadline_exceeded, "tail": tail, + "source": log_source, + "text": returned.decode("utf-8", errors="replace"), + } + + def _job_log_source(self, test_id, dashboard_error): + """Resolve a Maestro job's log, preferring a plain log artifact. + + lava_log is a gzipped log file the normal streaming path can read. + The LAVA callback carries the log as a JSON field instead, which + has to be read whole, and the pipeline treats that field as + temporary, so it is only the fallback. + """ + try: + node = self.get_node(_maestro_node_id(test_id)) + except KciDevError as exc: + raise KciDevError( + f"{dashboard_error}; the Maestro fallback also failed: {exc}" + ) from exc + artifacts = node.get("artifacts") or {} + if artifacts.get("lava_log"): + return artifacts["lava_log"], "maestro-log", None + if artifacts.get("callback_data"): + return None, "maestro-callback", artifacts["callback_data"] + raise KciDevError(f"No log available for {test_id}") + + def _callback_log(self, test_id, callback_url, max_bytes, tail): + """Fall back to the Maestro job callback when the dashboard has none. + + A job that fails before producing results never reaches the + dashboard, so its log is only in the LAVA callback Maestro keeps. + That callback is a JSON document rather than a log file, so it has + to be read whole before the log field can be taken out; it is + capped at ``MAX_CALLBACK_BYTES`` rather than streamed. + """ + response = None + try: + response = _stream_public_get(callback_url) + response.raise_for_status() + body = bytearray() + deadline = monotonic() + LOG_DEADLINE_SECONDS + for chunk in response.iter_content(_LOG_CHUNK): + if not chunk: + continue + if monotonic() > deadline: + raise KciDevError( + f"Job callback download for {test_id} passed the " + f"{LOG_DEADLINE_SECONDS}s deadline" + ) + body += chunk + if len(body) > MAX_CALLBACK_BYTES: + raise KciDevError( + f"Job callback for {test_id} exceeds " + f"{MAX_CALLBACK_BYTES} bytes" + ) + raw = bytes(body) + if raw[:2] == b"\x1f\x8b": + expanded = bytearray() + for piece in _gunzip_iter(iter([raw])): + expanded += piece + if len(expanded) > MAX_CALLBACK_BYTES: + raise KciDevError( + f"Job callback for {test_id} exceeds " + f"{MAX_CALLBACK_BYTES} bytes decompressed" + ) + raw = bytes(expanded) + callback = json.loads(raw.decode("utf-8", errors="replace")) + except KciDevError: + raise + except ( + requests.exceptions.RequestException, + zlib.error, + OSError, + ValueError, + ) as exc: + raise KciDevError( + f"Job callback download failed for {test_id}: {exc}" + ) from exc + finally: + if response is not None: + response.close() + + log = callback.get("log") + if not isinstance(log, str): + raise KciDevError(f"Job callback for {test_id} carries no log") + + encoded = log.encode("utf-8", errors="replace") + returned, total = _bounded_text(encoded, max_bytes, tail) + return { + "test_id": test_id, + "log_url": callback_url, + "total_bytes": total, + "returned_bytes": len(returned), + "truncated": total > len(returned), + "scan_limited": False, + "deadline_exceeded": False, + "tail": tail, + "source": "maestro-callback", "text": returned.decode("utf-8", errors="replace"), } diff --git a/kcidev/mcp/tools_dashboard.py b/kcidev/mcp/tools_dashboard.py index 07a3259..9978f0a 100644 --- a/kcidev/mcp/tools_dashboard.py +++ b/kcidev/mcp/tools_dashboard.py @@ -248,7 +248,11 @@ def get_log(test_id: str, max_bytes: int = 16384, tail: bool = True): if needed, up to a 1 MiB ceiling: asking for more returns that ceiling rather than the whole log, so compare returned_bytes with total_bytes rather than retrying the same call. A download that runs - past about a minute stops early and sets deadline_exceeded. Use + past about a minute stops early and sets deadline_exceeded. Jobs that + failed before producing results never reach the dashboard, so for + those the log comes from Maestro instead and 'source' says which was + used; get_node carries the infra diagnosis itself, which is usually + the better answer. Use get_test first for the shorter log_excerpt. """ return _current_client().get_log(test_id, max_bytes=max_bytes, tail=tail) diff --git a/tests/test_api.py b/tests/test_api.py index 87940cb..9205f05 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -741,3 +741,224 @@ def test_get_log_deadline_returns_partial_gzip(monkeypatch): assert out["deadline_exceeded"] is True assert out["truncated"] is True + + +def _job_node(callback_url="https://files.kernelci.org/cb.json.gz"): + return { + "id": "6a90b63cc5867fba9468b9b1", + "kind": "job", + "result": "incomplete", + "data": {"error_code": "Infrastructure", "error_msg": "Unable to flash"}, + "artifacts": {"callback_data": callback_url} if callback_url else {}, + } + + +def _mock_node(monkeypatch, node): + response = Mock(status_code=200) + response.json.return_value = node + monkeypatch.setattr( + maestro_common.kcidev_session, "get", Mock(return_value=response) + ) + + +def _no_dashboard_test(monkeypatch): + def boom(self, tid): + raise api.KciDevError("Dashboard test request failed: Test not found") + + monkeypatch.setattr(KernelCIClient, "get_test", boom) + + +def test_get_log_falls_back_to_the_maestro_job_callback(monkeypatch): + import gzip + + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + payload = gzip.compress(json.dumps({"log": "flash failed\nboom\n"}).encode()) + _mock_stream(monkeypatch, [payload]) + + out = _client().get_log("6a90b63cc5867fba9468b9b1") + + assert "flash failed" in out["text"] + assert out["source"] == "maestro-callback" + + +def test_get_log_fallback_accepts_the_prefixed_id(monkeypatch): + import gzip + + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [gzip.compress(json.dumps({"log": "hello"}).encode())]) + + out = _client().get_log("maestro:6a90b63cc5867fba9468b9b1") + + assert out["text"] == "hello" + + +def test_get_log_fallback_without_a_callback_is_a_clean_error(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node(callback_url=None)) + + with pytest.raises(api.KciDevError, match="No log available"): + _client().get_log("6a90b63cc5867fba9468b9b1") + + +def test_get_log_dashboard_path_is_still_preferred(monkeypatch): + monkeypatch.setattr(KernelCIClient, "get_test", lambda self, tid: _log_test()) + _mock_stream(monkeypatch, [b"from the dashboard"]) + + out = _client().get_log("maestro:abc") + + assert out["source"] == "dashboard" + assert out["text"] == "from the dashboard" + + +def test_get_log_fallback_caps_an_oversized_callback(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + huge = b"x" * (api.MAX_CALLBACK_BYTES + 1) + _mock_stream(monkeypatch, [huge]) + + with pytest.raises(api.KciDevError, match="exceeds"): + _client().get_log("6a90b63cc5867fba9468b9b1") + + +def test_get_log_fallback_bounds_the_extracted_log(monkeypatch): + import gzip + + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + body = json.dumps({"log": "A" * 100 + "TAILEND"}).encode() + _mock_stream(monkeypatch, [gzip.compress(body)]) + + out = _client().get_log("6a90b63cc5867fba9468b9b1", max_bytes=7, tail=True) + + assert out["text"] == "TAILEND" + assert out["truncated"] is True + assert out["total_bytes"] == 107 + + +def test_get_log_fallback_refuses_a_decompression_bomb(monkeypatch): + import gzip + + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + bomb = gzip.compress(b"\0" * (api.MAX_CALLBACK_BYTES * 8)) + assert len(bomb) < api.MAX_CALLBACK_BYTES + _mock_stream(monkeypatch, [bomb]) + + with pytest.raises(api.KciDevError, match="exceeds"): + _client().get_log("6a90b63cc5867fba9468b9b1") + + +def test_get_log_does_not_fall_back_on_a_non_not_found_error(monkeypatch): + def boom(self, tid): + raise api.KciDevError("Dashboard test request failed: upstream exploded") + + monkeypatch.setattr(KernelCIClient, "get_test", boom) + node = Mock(side_effect=AssertionError("fallback must not run")) + monkeypatch.setattr(KernelCIClient, "get_node", node) + + with pytest.raises(api.KciDevError, match="upstream exploded"): + _client().get_log("maestro:abc") + + +def test_get_log_reports_both_errors_when_the_fallback_also_fails(monkeypatch): + _no_dashboard_test(monkeypatch) + + def no_node(self, nid): + raise api.KciDevError("No Maestro api URL configured") + + monkeypatch.setattr(KernelCIClient, "get_node", no_node) + + with pytest.raises(api.KciDevError) as excinfo: + _client().get_log("maestro:abc") + + assert "Test not found" in str(excinfo.value) + assert "No Maestro api URL configured" in str(excinfo.value) + + +def test_get_log_fallback_reads_an_uncompressed_callback(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [json.dumps({"log": "plain json"}).encode()]) + + assert _client().get_log("6a90b63c")["text"] == "plain json" + + +def test_get_log_fallback_without_a_log_field_is_a_clean_error(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [json.dumps({"results": {}}).encode()]) + + with pytest.raises(api.KciDevError, match="carries no log"): + _client().get_log("6a90b63c") + + +def test_get_log_fallback_head_bounding(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [json.dumps({"log": "HEADSTART" + "z" * 90}).encode()]) + + out = _client().get_log("6a90b63c", max_bytes=9, tail=False) + assert out["text"] == "HEADSTART" + + +def test_get_log_fallback_stops_at_the_deadline(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [b"x" * 1000 for _ in range(50)]) + ticks = iter([0.0] + [api.LOG_DEADLINE_SECONDS + 1] * 200) + monkeypatch.setattr(api, "monotonic", lambda: next(ticks)) + + with pytest.raises(api.KciDevError, match="deadline|too long|timed out"): + _client().get_log("6a90b63c") + + +def test_get_log_does_not_fall_back_for_a_non_maestro_origin(monkeypatch): + _no_dashboard_test(monkeypatch) + monkeypatch.setattr( + KernelCIClient, + "get_node", + Mock(side_effect=AssertionError("must not query Maestro")), + ) + + with pytest.raises(api.KciDevError, match="Test not found"): + _client().get_log("redhat:6a90b63cc5867fba9468b9b1") + + +def test_get_log_fallback_malformed_gzip_is_a_clean_error(monkeypatch): + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [b"\x1f\x8b\x08\x00" + b"\xff" * 200]) + + with pytest.raises(api.KciDevError): + _client().get_log("6a90b63cc5867fba9468b9b1") + + +def test_get_log_prefers_the_lava_log_artifact(monkeypatch): + import gzip + + node = _job_node() + node["artifacts"]["lava_log"] = "https://files.kernelci.org/log.txt.gz" + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, node) + _mock_stream(monkeypatch, [gzip.compress(b"from the lava_log artifact\n")]) + + out = _client().get_log("6a90b63cc5867fba9468b9b1") + + assert "from the lava_log artifact" in out["text"] + assert out["source"] == "maestro-log" + assert out["log_url"] == "https://files.kernelci.org/log.txt.gz" + + +def test_get_log_uses_the_callback_when_there_is_no_lava_log(monkeypatch): + import gzip + + _no_dashboard_test(monkeypatch) + _mock_node(monkeypatch, _job_node()) + _mock_stream(monkeypatch, [gzip.compress(json.dumps({"log": "from cb"}).encode())]) + + out = _client().get_log("6a90b63cc5867fba9468b9b1") + + assert out["text"] == "from cb" + assert out["source"] == "maestro-callback"