diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 3af0e52..a24db46 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -180,6 +180,14 @@ def _aware(when: datetime) -> datetime: return when if when.tzinfo else when.replace(tzinfo=UTC) +#: How long after its last failure a node still counts as failing, and as +#: having failed. A node that stopped being called is not kept in the summary +#: forever, and a failure someone read is cleared with `acknowledge_error`. +FAILING_WINDOW_S = 900.0 +#: Failures in a row, with no good run between, that make a node "failing". +FAILING_AFTER = 2 + + @router.get("/summary", response_model=HealthSummary) async def read_summary(request: Request, controller: FlowControllerDep) -> Any: """How the engine is doing right now. Always 200, degraded or not.""" @@ -215,14 +223,17 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: f"{', '.join(sorted(e.id for e in unhealthy))}" ) # A node that loads and then fails on every call is not in `errored`, and - # until this it read as healthy. ponytail: one failure reads degraded for - # 15 minutes; a counter with decay if that proves noisy. + # would read as healthy. It reads degraded once it has failed twice running + # with no good run between, and only while the last failure is recent: one + # bad input, such as a visitor's, is a warning, and the next good run ends it. now = time.time() - failing = [ + recent = [ e for e in entries - if e.last_error_ts is not None and now - e.last_error_ts < 900 + if e.last_error_ts is not None and now - e.last_error_ts < FAILING_WINDOW_S ] + failing = [e for e in recent if e.failures_in_a_row >= FAILING_AFTER] + failed_once = [e for e in recent if 0 < e.failures_in_a_row < FAILING_AFTER] if failing: problems.append( f"{len(failing)} node(s) failing: " @@ -256,6 +267,11 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: if degraded else [] ) + if failed_once: + warnings.append( + f"{len(failed_once)} node(s) failed: " + f"{', '.join(sorted(e.id for e in failed_once))}" + ) return HealthSummary( status="degraded" if problems else "ok", diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 194429c..1d16dbb 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -177,12 +177,26 @@ class LoadedNode: #: so only an acknowledgement clears it, not a good run and not a rebuild. last_error: str = "" last_error_ts: float | None = None + #: Failures since it last ran well, so one bad input reads differently from + #: a node that fails every time. The summary goes by this, not by the age + #: of `last_error_ts`. + failures_in_a_row: int = 0 + #: When it last ran well after failing: the recovery the counter above + #: was reset by. + last_ok_ts: float | None = None #: Published, and with no body of its own — so what runs is the new-node #: template, which returns nothing at all. missing_source: bool = False #: The worker this node names although its type never runs on one. ignored_worker: str = "" + def inherit_failure(self, previous: LoadedNode) -> None: + """Keep what the node this one replaces last failed with.""" + self.last_error = previous.last_error + self.last_error_ts = previous.last_error_ts + self.failures_in_a_row = previous.failures_in_a_row + self.last_ok_ts = previous.last_ok_ts + @dataclass class RunContext: @@ -597,13 +611,23 @@ class FlowController: async with self.events.subscribe() as queue: while True: event = await queue.get() - if event.get("type") != "node_error": + kind = event.get("type") + if kind not in ("node_error", "node_executed"): continue entry = self.loaded.get(str(event.get("node") or "")) if entry is None: continue + ts = float(event.get("ts") or time.time()) + if kind == "node_executed": + # Almost every event here, so only a node with a failure + # standing is touched: running well is how it recovers. + if entry.failures_in_a_row: + entry.failures_in_a_row = 0 + entry.last_ok_ts = ts + continue entry.last_error = str(event.get("error") or "") - entry.last_error_ts = float(event.get("ts") or time.time()) + entry.last_error_ts = ts + entry.failures_in_a_row += 1 def acknowledge_error(self, node_id: str) -> None: """Forget what a node's last failure was. The only thing that clears it. @@ -616,6 +640,7 @@ class FlowController: raise KeyError(node_id) entry.last_error = "" entry.last_error_ts = None + entry.failures_in_a_row = 0 async def set_enabled(self, flow: str, enabled: bool) -> None: """Stop or start one flow, without building anything. @@ -721,8 +746,7 @@ class FlowController: for node_id, entry in loaded.items(): previous = self.loaded.get(node_id) if previous is not None and previous.last_error: - entry.last_error = previous.last_error - entry.last_error_ts = previous.last_error_ts + entry.inherit_failure(previous) self.loaded = loaded self.pipeline = Pipeline( @@ -820,8 +844,7 @@ class FlowController: for node_id, entry in loaded.items(): previous = self.loaded.get(node_id) if previous is not None and previous.last_error: - entry.last_error = previous.last_error - entry.last_error_ts = previous.last_error_ts + entry.inherit_failure(previous) # Rebound rather than mutated: request threads and the failure # watcher read these without holding anything. diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 6a2984f..fd123c8 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -456,7 +456,8 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded( ) -> None: """A node that loads and then raises each time it runs is not in `error`. - It is named while its last failure is recent, and drops out once it is not. + Failing twice running makes the instance degraded, once is a warning, and + a failure that is no longer recent counts for neither. """ import time @@ -464,15 +465,28 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded( controller = client.app.state.flow_controller before = controller.loaded + now = time.time() controller.loaded = { "house.calc": LoadedNode( - id="house.calc", flow="house", last_error="boom", last_error_ts=time.time() + id="house.calc", + flow="house", + last_error="boom", + last_error_ts=now, + failures_in_a_row=2, + ), + "house.once": LoadedNode( + id="house.once", + flow="house", + last_error="boom", + last_error_ts=now, + failures_in_a_row=1, ), "house.old": LoadedNode( id="house.old", flow="house", last_error="boom", - last_error_ts=time.time() - 3600, + last_error_ts=now - 3600, + failures_in_a_row=5, ), } try: @@ -483,7 +497,38 @@ def test_a_node_failing_on_every_call_makes_the_summary_degraded( assert body["status"] == "degraded" assert body["nodes"]["failing"] == 1 (problem,) = [p for p in body["problems"] if "failing" in p] - assert "house.calc" in problem and "house.old" not in problem + assert "house.calc" in problem + assert "house.once" not in problem and "house.old" not in problem + (warning,) = [w for w in body["warnings"] if "failed" in w] + assert "house.once" in warning and "house.calc" not in warning + + +def test_one_failure_is_a_warning_and_leaves_the_status_alone( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + import time + + from fluksio.flow.controller import LoadedNode + + controller = client.app.state.flow_controller + before = controller.loaded + controller.loaded = { + "house.once": LoadedNode( + id="house.once", + flow="house", + last_error="boom", + last_error_ts=time.time(), + failures_in_a_row=1, + ) + } + try: + body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json() + finally: + controller.loaded = before + + assert body["status"] == "ok" + assert body["nodes"]["failing"] == 0 + assert body["warnings"] == ["1 node(s) failed: house.once"] def test_a_superuser_can_flush_the_collector_now( diff --git a/backend/tests/flow/test_node_failures.py b/backend/tests/flow/test_node_failures.py new file mode 100644 index 0000000..0ce695f --- /dev/null +++ b/backend/tests/flow/test_node_failures.py @@ -0,0 +1,58 @@ +"""A node's failures in a row, which the health summary reads.""" + +import asyncio +from pathlib import Path + +from fluksio.flow.controller import FlowController, LoadedNode +from fluksio.flow.events import EventBus +from fluksio.flow.store import FlowStore + +NODE = "house.calc" + + +def _failure(ts: float) -> dict: + return {"type": "node_error", "node": NODE, "error": "boom", "ts": ts} + + +def test_failures_count_until_the_node_runs_well(tmp_path: Path) -> None: + async def scenario() -> LoadedNode: + events = EventBus() + events.bind(asyncio.get_running_loop()) + controller = FlowController(FlowStore(tmp_path / "flows"), events=events) + controller.loaded[NODE] = LoadedNode(id=NODE, flow="house") + watcher = asyncio.create_task(controller._watch_failures()) + await asyncio.sleep(0) + + events.publish(_failure(1.0)) + events.publish(_failure(2.0)) + await asyncio.sleep(0.05) + assert controller.loaded[NODE].failures_in_a_row == 2 + + events.publish({"type": "node_executed", "node": NODE, "ts": 3.0}) + await asyncio.sleep(0.05) + watcher.cancel() + return controller.loaded[NODE] + + entry = asyncio.run(scenario()) + + assert entry.failures_in_a_row == 0 + assert entry.last_ok_ts == 3.0 + # What it failed with stays for the canvas until someone acknowledges it. + assert entry.last_error == "boom" + + +def test_a_rebuilt_node_keeps_its_failures() -> None: + old = LoadedNode( + id=NODE, + flow="house", + last_error="boom", + last_error_ts=5.0, + failures_in_a_row=3, + last_ok_ts=1.0, + ) + new = LoadedNode(id=NODE, flow="house") + + new.inherit_failure(old) + + assert (new.last_error, new.last_error_ts) == ("boom", 5.0) + assert (new.failures_in_a_row, new.last_ok_ts) == (3, 1.0) diff --git a/docs/interface/index.md b/docs/interface/index.md index 0151090..1cb7bba 100644 --- a/docs/interface/index.md +++ b/docs/interface/index.md @@ -70,6 +70,11 @@ After you have used batch runs, it also shows running and queued runs. A degraded instance lists the problems that need attention. Warnings, such as a device that is offline, are listed in amber and leave the status as it is. +A node that raises is a warning after its first failure and a problem when it +fails twice in a row with no good run between. A good run ends either, and so +does acknowledging the failure on the canvas. Without one, a failure counts for +15 minutes after the last time it happened. + ### Execution time per node type One bar per node type at its average time per run, slowest first, on a log