From c820d4b63774c6f0ada19e4a6e9d566aa440ff16 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 30 Oct 2026 02:01:16 +0800 Subject: [PATCH] fix(delivery): preserve bounded operator failure diagnostics --- .github/workflows/engineering.yml | 1 + backend/tests/test_operator_diagnostics.py | 227 +++++++++++++++++++++ scripts/release/README.md | 9 + scripts/release/lifecycle.py | 11 +- scripts/release/node_health.py | 22 +- scripts/release/operator_units.py | 168 ++++++++++++++- 6 files changed, 429 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_operator_diagnostics.py diff --git a/.github/workflows/engineering.yml b/.github/workflows/engineering.yml index ccc3de5..908391f 100644 --- a/.github/workflows/engineering.yml +++ b/.github/workflows/engineering.yml @@ -116,6 +116,7 @@ jobs: tests/test_delivery_restore_io.py \ tests/test_restore_io_contracts.py \ tests/test_restore_native_transport.py \ + tests/test_operator_diagnostics.py \ tests/test_operator_imports.py \ tests/test_health.py \ tests/test_identity.py \ diff --git a/backend/tests/test_operator_diagnostics.py b/backend/tests/test_operator_diagnostics.py new file mode 100644 index 0000000..2431196 --- /dev/null +++ b/backend/tests/test_operator_diagnostics.py @@ -0,0 +1,227 @@ +"""Ordinary controller/codec models; no service or isolation acceptance.""" + +import copy +import hashlib +import importlib +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +units = importlib.import_module("scripts.release.operator_units") +install = importlib.import_module("scripts.release.install") +service_install = importlib.import_module("scripts.release.service_install") + + +def record(): + return units.failure_record( + "factory-health", + "exit", + subprocess.CompletedProcess([], 1, b"private-output", b"private-error"), + units.DeliveryError("operator helper unavailable"), + ) + + +@pytest.fixture +def dispatched(monkeypatch): + selected = {"release_id": "sd-" + "a" * 12, "configuration_id": "cfg-" + "b" * 24} + payload = install.RELEASES / selected["release_id"] / "payload" + monkeypatch.setattr(units.os, "getuid", lambda: 0) + monkeypatch.setattr(units.os, "geteuid", lambda: 0) + monkeypatch.setattr(install, "selection", lambda **_: selected) + calls = [] + + def manager(*args): + calls.append(args) + return "LoadState=not-found\nActiveState=inactive\nMainPID=0\n" + + monkeypatch.setattr(service_install, "systemctl", manager) + return payload, calls + + +@pytest.mark.parametrize("failure", ["exit", "timeout", "decode", "stderr_bound"]) +def test_dispatch_diagnostics_keep_limits_cleanup_and_only_output_digests( + monkeypatch, dispatched, failure +): + payload, calls = dispatched + + def execute(command, **kwargs): + assert kwargs["timeout"] == 915 + assert "--property=MemoryMax=256M" in command + assert "--property=RuntimeMaxSec=900" in command + if failure == "timeout": + raise subprocess.TimeoutExpired( + command, 915, output=b"private-output", stderr=b"private-error" + ) + return subprocess.CompletedProcess( + command, + 1 if failure == "exit" else 0, + b"private-output", + b"private-error" * (400 if failure == "stderr_bound" else 1), + ) + + monkeypatch.setattr(units.subprocess, "run", execute) + with pytest.raises(units.OperatorExecutionError) as caught: + units.run("node-maintenance", payload, action="start") + error = caught.value + assert len(calls) == 2 + row = error.failures[0] + assert row["helper"] == "node-maintenance" + assert row["outcome"] == ("exit" if failure == "stderr_bound" else failure) + assert row["stdout_bytes"] == 14 + assert row["stdout_sha256"] == hashlib.sha256(b"private-output").hexdigest() + text = units.diagnostic_suffix(error) + assert "private-" not in text and len(text) <= 3072 + assert units.diagnostic_chain(text.encode()) == error.failures + + +def test_nested_failure_retains_inner_helper_and_outer_actual_exit(monkeypatch, dispatched): + payload, calls = dispatched + child = units.OperatorExecutionError("private-message", [record()]) + stderr = b"private-error\n" + units.diagnostic_suffix(child).encode() + monkeypatch.setattr( + units.subprocess, + "run", + lambda command, **_: subprocess.CompletedProcess(command, 1, b"", stderr), + ) + with pytest.raises(units.OperatorExecutionError) as caught: + units.run("node-maintenance", payload, action="start") + rows = caught.value.failures + assert [r["helper"] for r in rows] == ["factory-health", "node-maintenance"] + assert ( + rows[1]["returncode"] == 1 + and rows[1]["stderr_sha256"] == hashlib.sha256(stderr).hexdigest() + ) + assert len(calls) == 2 and "private-" not in units.diagnostic_suffix(caught.value) + + +@pytest.mark.parametrize( + "mutation", + [ + "unknown_field", + "helper", + "outcome", + "boolean_code", + "boolean_size", + "digest", + "unknown_readiness", + "deep", + "schema", + "duplicate", + "oversize", + "invalid_json", + ], +) +def test_untrusted_child_diagnostic_shape_is_discarded(mutation): + value = {"schema_version": 1, "failures": [record()]} + row = value["failures"][0] + if mutation == "unknown_field": + row["private"] = "must-not-escape" + elif mutation == "helper": + row["helper"] = "/private/command" + elif mutation == "outcome": + row["outcome"] = ["private-error"] + elif mutation == "boolean_code": + row["returncode"] = True + elif mutation == "boolean_size": + row["stdout_bytes"] = True + elif mutation == "digest": + row["stderr_sha256"] = "private-error" + elif mutation == "unknown_readiness": + row["readiness_errors"] = ["private-error"] + elif mutation == "deep": + value["failures"] *= 4 + elif mutation == "schema": + value["schema_version"] = True + text = (units.DIAGNOSTIC_PREFIX + json.dumps(value) + "\n").encode() + if mutation == "duplicate": + text *= 2 + elif mutation == "oversize": + text += b"x" * 4097 + elif mutation == "invalid_json": + text = units.DIAGNOSTIC_PREFIX.encode() + b"{broken" + assert units.diagnostic_chain(text) == [] + + +def test_readiness_preserves_two_inner_failures_and_only_fixed_codes(): + rows = [record(), copy.deepcopy(record()), copy.deepcopy(record())] + error = units.readiness_failure( + "preconditions unavailable", + { + "errors": ["factory_status_unavailable"], + "operator_failures": rows, + "password": "private-value", + }, + ) + assert len(error.failures) == 3 + assert error.failures[-1]["readiness_errors"] == ["factory_status_unavailable"] + assert "private-" not in units.diagnostic_suffix(error) + refused = units.readiness_failure( + "admission unavailable", {"errors": ["private-value"]}, admission=True + ) + assert refused.failures[0]["outcome"] == "admission" + assert "readiness_errors" not in refused.failures[0] + + +def test_readonly_health_keeps_failed_components_and_bounded_nested_causes(monkeypatch, tmp_path): + from concurrent.futures import Future + from contextlib import nullcontext + + health = importlib.import_module("scripts.release.node_health") + selected = {"release_id": "sd-" + "a" * 12, "configuration_id": "cfg-" + "b" * 24} + payload = tmp_path / "payload" + monkeypatch.setattr(units, "guarded", lambda _: None) + monkeypatch.setattr( + health, "binding", lambda: (selected, {"coordinator.json": Path("coordinator")}, payload) + ) + monkeypatch.setattr( + health, + "read_json", + lambda path, *args: ( + {"source_sha": "a" * 40} + if path.name == "manifest.json" + else {"platform_token": "private-token", "admission": {}} + ), + ) + monkeypatch.setattr(health.journal, "check_ready", lambda: {}) + monkeypatch.setattr(health.proxy_install, "check_binding", lambda *args: None) + monkeypatch.setattr(health.proxy_health, "observe", lambda _: {}) + monkeypatch.setattr(health, "inspect_unit", lambda *args: {}) + monkeypatch.setattr(health.dependency_install, "installed_endpoints", lambda _: {}) + + async def roles(_): + return {} + + monkeypatch.setattr(health, "roles", roles) + monkeypatch.setattr( + health, "http", lambda *args, **kwargs: {"status": "ok", "service": "platform"} + ) + monkeypatch.setattr(health, "validate_execution", lambda *args: {}) + monkeypatch.setattr(install, "selection", lambda **kwargs: selected) + pending = {} + for name in ("broker-health", "outbox-health", "import-health", "factory-health"): + future = Future() + row = record() + row["helper"] = name + future.set_exception(units.OperatorExecutionError("operator helper unavailable", [row])) + pending[name] = future + monkeypatch.setattr(health, "_status_helpers", lambda _: nullcontext(pending)) + observed = health.observe(probe=False) + assert observed["ready_for_admission"] is False + assert observed["service_preconditions_met"] is False + assert observed["errors"] == [ + "broker_metadata_unavailable", + "outbox_pipeline_unavailable", + "import_pipeline_unavailable", + "factory_status_unavailable", + ] + error = units.readiness_failure("service preconditions unavailable", observed) + assert [r["helper"] for r in error.failures] == [ + "broker-health", + "outbox-health", + "node-health", + ] + assert "private-" not in units.diagnostic_suffix(error) diff --git a/scripts/release/README.md b/scripts/release/README.md index b30c568..cc0a698 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -341,6 +341,15 @@ the outer operator also closes admission. If that record cannot be written, it stops only the verified API unit. A refusal before dispatch, such as an already active helper, does not change admission or stop another operation. +Dispatched failures append one `KF_OPERATOR_DIAGNOSTIC` JSON line to the CLI +error. Its versioned, bounded chain records fixed helper names, failure stages, +exit codes, output sizes and SHA-256 digests. Failed readiness includes only +allowlisted component error codes and at most two dependency helper records. +Raw child output, arguments, credentials and exception text are excluded from +the diagnostic line. Malformed, oversized or duplicate child diagnostic lines +are discarded. This observation does not change service limits, deadlines, +admission checks or cleanup requirements; a failed start remains a failure. + Drain keeps API available to accepted work and polls Coordinator, outbox, import, broker and factory within a finite deadline. Untruncated owner counts, stopped process groups, empty tasks/identity/capture resources and an absent factory socket diff --git a/scripts/release/lifecycle.py b/scripts/release/lifecycle.py index 5ee2544..88d0dec 100644 --- a/scripts/release/lifecycle.py +++ b/scripts/release/lifecycle.py @@ -639,12 +639,16 @@ class Maintenance: self.start_group(group) ready = operator_units.run("node-health", self.payload) if ready.get("service_preconditions_met") is not True: - raise DeliveryError("service preconditions unavailable; admission remains closed") + raise operator_units.readiness_failure( + "service preconditions unavailable; admission remains closed", ready + ) try: availability.write(self.selected, "open") ready = operator_units.run("node-health", self.payload) if ready.get("ready_for_admission") is not True: - raise DeliveryError("admission readiness readback unconfirmed") + raise operator_units.readiness_failure( + "admission readiness readback unconfirmed", ready, admission=True + ) self.save("accepting") except (OSError, ValueError, subprocess.SubprocessError): self.close_failed_admission() @@ -751,7 +755,8 @@ def main() -> None: 1, "Lifecycle refused: " + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) - + "\n", + + "\n" + + operator_units.diagnostic_suffix(error), ) diff --git a/scripts/release/node_health.py b/scripts/release/node_health.py index a03609c..1bc593f 100644 --- a/scripts/release/node_health.py +++ b/scripts/release/node_health.py @@ -525,6 +525,13 @@ def observe(*, probe: bool) -> dict: "errors": [], } components, errors = result["components"], result["errors"] + helper_failures = [] + + def remember_helper(error): + if isinstance(error, operator_units.OperatorExecutionError): + helper_failures.extend(error.failures[: max(0, 2 - len(helper_failures))]) + result["operator_failures"] = helper_failures + try: components["journal"] = journal.check_ready() except (OSError, ValueError, subprocess.SubprocessError): @@ -580,13 +587,15 @@ def observe(*, probe: bool) -> dict: with _status_helpers(payload) as pending: try: components["broker"] = pending["broker-health"].result() - except (OSError, ValueError, subprocess.SubprocessError): + except (OSError, ValueError, subprocess.SubprocessError) as error: + remember_helper(error) errors.append("broker_metadata_unavailable") try: state = pending["outbox-health"].result() validate_outbox(state) components["outbox_pipeline"] = state - except (OSError, ValueError, subprocess.SubprocessError): + except (OSError, ValueError, subprocess.SubprocessError) as error: + remember_helper(error) errors.append("outbox_pipeline_unavailable") try: state = pending["import-health"].result() @@ -617,13 +626,15 @@ def observe(*, probe: bool) -> dict: if type(state[key]) is not int or not 0 <= state[key] <= maximum: raise DeliveryError("import status bound differs") components["import_pipeline"] = state - except (OSError, ValueError, subprocess.SubprocessError): + except (OSError, ValueError, subprocess.SubprocessError) as error: + remember_helper(error) errors.append("import_pipeline_unavailable") try: status = pending["factory-health"].result() validate_factory(status, digest(paths["runtime.json"])) components["factory_status"] = status - except (OSError, ValueError, subprocess.SubprocessError): + except (OSError, ValueError, subprocess.SubprocessError) as error: + remember_helper(error) errors.append("factory_status_unavailable") if probe: try: @@ -723,7 +734,8 @@ def main() -> None: 1, "Node health refused: " + (str(error) if isinstance(error, DeliveryError) else type(error).__name__) - + "\n", + + "\n" + + operator_units.diagnostic_suffix(error), ) diff --git a/scripts/release/operator_units.py b/scripts/release/operator_units.py index 6f74868..3918a23 100644 --- a/scripts/release/operator_units.py +++ b/scripts/release/operator_units.py @@ -1,7 +1,9 @@ """Finite fixed-purpose operator helpers; no caller-controlled executable or identity.""" +import hashlib import json import os +import re import resource import subprocess from pathlib import Path @@ -13,6 +15,10 @@ from .policy import DeliveryError, invalid_constant, unique_object class OperatorExecutionError(DeliveryError): """A fixed helper was dispatched but its result or cleanup was unconfirmed.""" + def __init__(self, message: str, failures: list | None = None): + super().__init__(message) + self.failures = failures or [] + PROFILES = { "node-health": ("root", 0, 128, 16, 50, 300), @@ -28,6 +34,156 @@ PROFILES = { # OTP 27 reserves a 32 MiB dual-mapped JIT block before serving CLI queries. FILE_LIMIT_MIB = {"broker-health": 64} +DIAGNOSTIC_PREFIX = "KF_OPERATOR_DIAGNOSTIC " +READINESS_ERRORS = frozenset( + { + "bounded_journal_unavailable", + "https_entry_unavailable", + "selected_binding_changed", + "db_authority_unavailable", + "api_live_unavailable", + "execution_owner_unavailable", + "broker_metadata_unavailable", + "outbox_pipeline_unavailable", + "import_pipeline_unavailable", + "factory_status_unavailable", + "notification_delivery_unconfirmed", + *( + name + "_unavailable" + for name in ( + "api", + "coordinator", + "outbox", + "problem-import", + "worker", + "postgres", + "rabbitmq", + "sandbox-factory", + "proxy", + ) + ), + } +) + + +def failure_record(name: str, outcome: str, result=None, error=None) -> dict: + """Record only fixed identifiers, sizes and digests; never child output or argv.""" + record = {"helper": name, "outcome": outcome, "returncode": None} + if result is not None: + record["returncode"] = result.returncode + for stream in ("stdout", "stderr"): + value = getattr(result, stream, b"") if result is not None else b"" + if isinstance(error, subprocess.TimeoutExpired): + value = getattr(error, stream, None) or b"" + record[stream + "_bytes"] = len(value) + record[stream + "_sha256"] = hashlib.sha256(value).hexdigest() + record["error_sha256"] = hashlib.sha256( + (str(error) if isinstance(error, DeliveryError) else type(error).__name__).encode() + ).hexdigest() + return record + + +def diagnostic_chain(stderr: bytes) -> list: + """Accept one small, strictly shaped record from a fixed child CLI.""" + if len(stderr) > 4096: + return [] + lines = [line for line in stderr.splitlines() if line.startswith(DIAGNOSTIC_PREFIX.encode())] + try: + if len(lines) != 1: + return [] + value = decode(lines[0][len(DIAGNOSTIC_PREFIX) :], 3072) + if ( + set(value) != {"schema_version", "failures"} + or type(value["schema_version"]) is not int + or value["schema_version"] != 1 + ): + return [] + failures = value["failures"] + if not isinstance(failures, list) or not 1 <= len(failures) <= 3: + return [] + fields = { + "helper", + "outcome", + "returncode", + "stdout_bytes", + "stdout_sha256", + "stderr_bytes", + "stderr_sha256", + "error_sha256", + } + for row in failures: + if not isinstance(row, dict) or set(row) not in ( + fields, + fields | {"readiness_errors"}, + ): + return [] + if row["helper"] not in PROFILES or row["outcome"] not in { + "dispatch", + "exit", + "decode", + "cleanup", + "timeout", + "preconditions", + "admission", + }: + return [] + code = row["returncode"] + if code is not None and (type(code) is not int or not -255 <= code <= 255): + return [] + if any( + type(row[k]) is not int or not 0 <= row[k] <= 2**63 - 1 + for k in ("stdout_bytes", "stderr_bytes") + ): + return [] + if any( + not isinstance(row[k], str) or not re.fullmatch("[a-f0-9]{64}", row[k]) + for k in ("stdout_sha256", "stderr_sha256", "error_sha256") + ): + return [] + errors = row.get("readiness_errors", []) + if ( + not isinstance(errors, list) + or len(errors) > len(READINESS_ERRORS) + or any(not isinstance(code, str) or code not in READINESS_ERRORS for code in errors) + ): + return [] + return failures + except (TypeError, ValueError, KeyError): + return [] + + +def diagnostic_suffix(error: Exception) -> str: + if not isinstance(error, OperatorExecutionError) or not error.failures: + return "" + # Revalidate even internally built records before they cross a CLI boundary. + value = json.dumps({"schema_version": 1, "failures": error.failures}, separators=(",", ":")) + line = DIAGNOSTIC_PREFIX + value + if len(error.failures) > 4 or len(line) > 3072: + return "" + # The child decoder allows three ancestors; validate the fourth separately. + for row in error.failures: + single = DIAGNOSTIC_PREFIX + json.dumps({"schema_version": 1, "failures": [row]}) + if not diagnostic_chain(single.encode()): + return "" + return line + "\n" + + +def readiness_failure(message: str, result: dict, *, admission: bool = False): + error = OperatorExecutionError(message) + row = failure_record("node-health", "admission" if admission else "preconditions", error=error) + errors = result.get("errors") + if ( + isinstance(errors, list) + and len(errors) <= len(READINESS_ERRORS) + and all(isinstance(code, str) and code in READINESS_ERRORS for code in errors) + ): + row["readiness_errors"] = errors + child = DIAGNOSTIC_PREFIX + json.dumps( + {"schema_version": 1, "failures": result.get("operator_failures", [])} + ) + error.failures = diagnostic_chain(child.encode())[:2] + [row] + return error + def guarded(name: str) -> None: _, uid, memory, pids, cpu, _ = PROFILES[name] @@ -239,6 +395,8 @@ def run( command.append("--probe-notification") elif notification_probe: raise DeliveryError("unexpected broker probe override") + result = None + outcome = "dispatch" try: try: # Pipe output is bounded by the fixed helper and the service file/memory @@ -249,12 +407,15 @@ def run( timeout=seconds + 15, env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}, ) + outcome = "exit" if result.returncode or len(result.stderr) > 4096: raise DeliveryError("operator helper unavailable") + outcome = "decode" value = decode(result.stdout) if not isinstance(value, dict): raise DeliveryError("operator helper object required") finally: + previous_outcome, outcome = outcome, "cleanup" actual = dict( line.split("=", 1) for line in service_install.systemctl( @@ -277,7 +438,12 @@ def run( ) if actual.get("ActiveState") != "inactive" or actual.get("MainPID") != "0": raise DeliveryError("operator helper cleanup unconfirmed") + outcome = previous_outcome except (OSError, ValueError, subprocess.SubprocessError) as error: message = str(error) if isinstance(error, DeliveryError) else type(error).__name__ - raise OperatorExecutionError(message) from error + if isinstance(error, subprocess.TimeoutExpired) and outcome != "cleanup": + outcome = "timeout" + failures = diagnostic_chain(result.stderr) if result is not None else [] + failures.append(failure_record(name, outcome, result, error)) + raise OperatorExecutionError(message, failures) from error return value