diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 27bf0c7..154a989 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -182,7 +182,10 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: paused = set(controller.paused_flows()) entries = list(controller.loaded.values()) errored = [e for e in entries if e.status is NodeStatus.ERROR] - unhealthy = [e for e in entries if e.health == "down"] + # Not `e.health == "down"` directly: a node has to have been down for + # `HEALTH_FLOOR` before it counts, so a device that misses one poll does + # not turn this screen amber. Same source the canvas marks nodes from. + unhealthy = controller.unhealthy_nodes() if quarantined: problems.append(f"{len(quarantined)} flow(s) quarantined") if errored: diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index e87584a..ce3f30d 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -113,6 +113,14 @@ REBUILD_WAIT = 15.0 # the whole-pipeline rebuild and would let a wedge sit unreported. FLOW_REBUILD_WAIT = 5.0 +# How long a node has to have been down before it is shown as down. A device +# that misses one poll and comes back is a flake, not a fault, and without this +# it turned the health badge amber and the node on the canvas red for a couple +# of seconds. Only going *down* is held back: recovery clears at once, since +# calling a working thing broken is the worse of the two mistakes to leave on +# screen. The stored health is never delayed — see `unhealthy_nodes`. +HEALTH_FLOOR = 60.0 + class RebuildBusy(RuntimeError): """A rebuild could not start because the one before it has not finished. @@ -142,6 +150,12 @@ class LoadedNode: error: str | None = None health: Health = "ok" health_detail: str | None = None + #: When `health` last became what it is, for the floor `unhealthy_nodes` + #: applies. A detail rotating under an unchanged status does not move it. + #: Zero reads as "for as long as anyone knows", so a node that starts out + #: down is shown straight away rather than waiting out a floor from a + #: transition that never happened. + health_since: float = 0.0 #: The last time this node raised while running, and what it said. Kept #: after it has run again — a failure nobody saw is the one worth keeping — #: so only an acknowledgement clears it, not a good run and not a rebuild. @@ -1240,6 +1254,12 @@ class FlowController: entry = self.loaded.get(node.id) if entry is None or (entry.health == status and entry.health_detail == detail): return + if entry.health != status: + # Only a real transition restarts the clock. A connector retrying + # with a different errno each poll rotates the detail while staying + # down, and moving the mark for that would keep the node forever + # under `HEALTH_FLOOR` — debouncing nothing at all. + entry.health_since = time.time() entry.health = cast(Health, status) entry.health_detail = detail self._publish( @@ -1252,6 +1272,24 @@ class FlowController: } ) + def unhealthy_nodes(self, flow: str | None = None) -> list[LoadedNode]: + """Nodes that have been down long enough to be worth telling anyone. + + The floor is on the surfacing, not on the transition: `entry.health` is + the truth the moment the node reports it, so a flow reading a node's + health is never told a stale story — only the screens wait. And only + for the way down; a node reading "ok" leaves this list immediately, + whenever it recovered. + """ + now = time.time() + return [ + entry + for entry in self.loaded.values() + if entry.health == "down" + and now - entry.health_since >= HEALTH_FLOOR + and (flow is None or entry.flow == flow) + ] + def _health_issues(self, flow: str | None = None) -> list[ValidationIssue]: """Nodes that are running but not working, as issues on their flow. @@ -1269,8 +1307,7 @@ class FlowController: flow=entry.flow, node=entry.id, ) - for entry in self.loaded.values() - if entry.health == "down" and (flow is None or entry.flow == flow) + for entry in self.unhealthy_nodes(flow) ] def flow_issues(self, flow: str) -> list[ValidationIssue]: diff --git a/backend/tests/flow/test_unhealthy_node_issue.py b/backend/tests/flow/test_unhealthy_node_issue.py index 45e44fa..7ef2420 100644 --- a/backend/tests/flow/test_unhealthy_node_issue.py +++ b/backend/tests/flow/test_unhealthy_node_issue.py @@ -5,8 +5,10 @@ and no screen ever asked. These cover the derivation that closes that gap. """ from pathlib import Path +from types import SimpleNamespace +from typing import Any -from fluksio.flow.controller import FlowController, LoadedNode +from fluksio.flow.controller import HEALTH_FLOOR, FlowController, LoadedNode from fluksio.flow.store import FlowStore @@ -40,3 +42,62 @@ def test_the_issue_clears_when_the_node_reports_itself_well(tmp_path: Path) -> N controller.loaded["house.owm"].health = "ok" assert controller.flow_issues("house") == [] + + +def _reporting_node(node_id: str, flow: str) -> Any: + """The bit of a node `_health_changed` reads: which node, on which flow.""" + return SimpleNamespace(id=node_id, flow=flow) + + +def a_running_controller(tmp_path: Path) -> tuple[FlowController, Any]: + """A controller holding one healthy node, plus its health reporter.""" + controller = FlowController(FlowStore(tmp_path / "flows")) + controller.loaded["house.owm"] = LoadedNode(id="house.owm", flow="house") + return controller, _reporting_node("house.owm", "house") + + +def test_a_node_that_drops_one_poll_never_surfaces(tmp_path: Path) -> None: + controller, node = a_running_controller(tmp_path) + + controller._health_changed(node, "down", "TimeoutError: no reply") + + # Stored truthfully — a flow reading node health sees it — but held back + # from the screens until the floor passes. + assert controller.loaded["house.owm"].health == "down" + assert controller.flow_issues("house") == [] + + controller._health_changed(node, "ok", None) + assert controller.flow_issues("house") == [] + + +def test_a_node_still_down_past_the_floor_surfaces(tmp_path: Path) -> None: + controller, node = a_running_controller(tmp_path) + + controller._health_changed(node, "down", "TimeoutError: no reply") + controller.loaded["house.owm"].health_since -= HEALTH_FLOOR + + assert [issue.code for issue in controller.flow_issues("house")] == [ + "node_unhealthy" + ] + + # Recovery is not debounced: the floor is only on the way down. + controller._health_changed(node, "ok", None) + assert controller.flow_issues("house") == [] + + +def test_a_rotating_error_detail_does_not_restart_the_floor(tmp_path: Path) -> None: + """The case that would quietly disable the debounce for real connectors. + + A connector retrying a dead device reports a different errno each poll. If + the detail moved the mark, the node would never reach the floor and would + never be reported down at all. + """ + controller, node = a_running_controller(tmp_path) + + controller._health_changed(node, "down", "OSError: [Errno 113] no route") + controller.loaded["house.owm"].health_since -= HEALTH_FLOOR + controller._health_changed(node, "down", "OSError: [Errno 110] timed out") + + issues = controller.flow_issues("house") + assert [issue.code for issue in issues] == ["node_unhealthy"] + assert "Errno 110" in issues[0].message