feat(us-024): enhanced availability map with per-node health details
Coverage endpoint now skips node purge so dead nodes remain visible. Each band in /v1/coverage/<model> includes a `nodes` array with per-node health: alive status, last_seen_seconds_ago, heartbeat_success_rate, inference_success_rate, queue_depth, uptime_seconds, and endpoint. Updated tests to assert new band structure. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -244,6 +244,81 @@ def _coverage_map(
|
|||||||
return coverage
|
return coverage
|
||||||
|
|
||||||
|
|
||||||
|
def _node_health(node: "_NodeEntry", heartbeat_timeout: float) -> dict:
|
||||||
|
"""Per-node health detail for the availability map."""
|
||||||
|
age = time.monotonic() - node.last_heartbeat
|
||||||
|
alive = age <= heartbeat_timeout
|
||||||
|
hb_expected = max(1, round(node.uptime_seconds / 20.0)) # assume ~20s interval
|
||||||
|
hb_rate = round(min(1.0, node.heartbeats_received / hb_expected), 4) if node.heartbeats_received else 0.0
|
||||||
|
total = node.total_requests
|
||||||
|
inf_rate = round((total - node.failed_requests) / total, 4) if total > 0 else 1.0
|
||||||
|
return {
|
||||||
|
"node_id": node.node_id,
|
||||||
|
"endpoint": node.endpoint,
|
||||||
|
"alive": alive,
|
||||||
|
"last_seen_seconds_ago": round(age, 1),
|
||||||
|
"status": node.status,
|
||||||
|
"queue_depth": node.queue_depth,
|
||||||
|
"total_requests": node.total_requests,
|
||||||
|
"heartbeat_success_rate": hb_rate,
|
||||||
|
"inference_success_rate": inf_rate,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _coverage_map_detailed(
|
||||||
|
nodes: list["_NodeEntry"],
|
||||||
|
required_start: int,
|
||||||
|
required_end: int,
|
||||||
|
heartbeat_timeout: float,
|
||||||
|
) -> list[dict]:
|
||||||
|
"""Like _coverage_map but with per-node identity and health in each band.
|
||||||
|
|
||||||
|
Includes all nodes (alive and stale) so operators can see both coverage
|
||||||
|
holes and which dead nodes used to fill them.
|
||||||
|
"""
|
||||||
|
now = time.monotonic()
|
||||||
|
|
||||||
|
def covers(node: "_NodeEntry", layer: int) -> bool:
|
||||||
|
return (
|
||||||
|
node.shard_start is not None
|
||||||
|
and node.shard_end is not None
|
||||||
|
and node.shard_start <= layer <= node.shard_end
|
||||||
|
)
|
||||||
|
|
||||||
|
# Build per-layer list of nodes (with alive flag)
|
||||||
|
layer_nodes: list[list[tuple["_NodeEntry", bool]]] = []
|
||||||
|
for layer in range(required_start, required_end + 1):
|
||||||
|
ns = [
|
||||||
|
(n, (now - n.last_heartbeat) <= heartbeat_timeout)
|
||||||
|
for n in nodes
|
||||||
|
if covers(n, layer)
|
||||||
|
]
|
||||||
|
layer_nodes.append(ns)
|
||||||
|
|
||||||
|
# Merge consecutive layers with identical node-set (same node_ids, same alive)
|
||||||
|
coverage: list[dict] = []
|
||||||
|
for i, layer_idx in enumerate(range(required_start, required_end + 1)):
|
||||||
|
ns = layer_nodes[i]
|
||||||
|
node_ids = [n.node_id for n, _ in ns]
|
||||||
|
alive_flags = [a for _, a in ns]
|
||||||
|
# Compare with last band
|
||||||
|
if (
|
||||||
|
coverage
|
||||||
|
and coverage[-1]["end_layer"] == layer_idx - 1
|
||||||
|
and [nd["node_id"] for nd in coverage[-1]["nodes"]] == node_ids
|
||||||
|
and [nd["alive"] for nd in coverage[-1]["nodes"]] == alive_flags
|
||||||
|
):
|
||||||
|
coverage[-1]["end_layer"] = layer_idx
|
||||||
|
else:
|
||||||
|
coverage.append({
|
||||||
|
"start_layer": layer_idx,
|
||||||
|
"end_layer": layer_idx,
|
||||||
|
"node_count": len(ns),
|
||||||
|
"nodes": [_node_health(n, heartbeat_timeout) for n, _ in ns],
|
||||||
|
})
|
||||||
|
return coverage
|
||||||
|
|
||||||
|
|
||||||
def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]:
|
def _coverage_gaps(coverage: list[dict]) -> list[tuple[int, int]]:
|
||||||
return [
|
return [
|
||||||
(segment["start_layer"], segment["end_layer"])
|
(segment["start_layer"], segment["end_layer"])
|
||||||
@@ -624,19 +699,20 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
|
|||||||
|
|
||||||
def _handle_coverage(self, model: str):
|
def _handle_coverage(self, model: str):
|
||||||
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
|
||||||
|
# Do NOT purge before coverage — dead nodes are included with alive=false
|
||||||
|
# so operators can see what was covering each layer band before failure.
|
||||||
with server.lock:
|
with server.lock:
|
||||||
self._purge_expired_nodes()
|
|
||||||
resolved = _nodes_and_bounds_for_model(server, model)
|
resolved = _nodes_and_bounds_for_model(server, model)
|
||||||
if resolved is None:
|
if resolved is None:
|
||||||
self._send_json(404, {"error": f"no nodes registered for model {model!r}"})
|
self._send_json(404, {"error": f"no nodes registered for model {model!r}"})
|
||||||
return
|
return
|
||||||
alive, required_start, required_end = resolved
|
all_nodes, required_start, required_end = resolved
|
||||||
if server.contracts is not None:
|
if server.contracts is not None:
|
||||||
alive = [
|
all_nodes = [
|
||||||
node for node in alive
|
node for node in all_nodes
|
||||||
if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned
|
if not node.wallet_address or not server.contracts.registry.get_wallet(node.wallet_address).banned
|
||||||
]
|
]
|
||||||
coverage = _coverage_map(alive, required_start, required_end)
|
coverage = _coverage_map_detailed(all_nodes, required_start, required_end, server.heartbeat_timeout)
|
||||||
self._send_json(200, {"model": model, "coverage": coverage})
|
self._send_json(200, {"model": model, "coverage": coverage})
|
||||||
|
|
||||||
def _handle_tracker_nodes(self, model: str):
|
def _handle_tracker_nodes(self, model: str):
|
||||||
|
|||||||
@@ -243,10 +243,24 @@ def test_tracker_coverage_endpoint_reports_uncovered_ranges():
|
|||||||
)
|
)
|
||||||
|
|
||||||
coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model")
|
coverage_resp = _get_json(f"http://127.0.0.1:{tracker_port}/v1/coverage/tiny-model")
|
||||||
assert coverage_resp["coverage"] == [
|
bands = coverage_resp["coverage"]
|
||||||
{"start_layer": 0, "end_layer": 2, "node_count": 1},
|
assert len(bands) == 2
|
||||||
{"start_layer": 3, "end_layer": 5, "node_count": 0},
|
# Covered band: layers 0-2, one node
|
||||||
]
|
assert bands[0]["start_layer"] == 0
|
||||||
|
assert bands[0]["end_layer"] == 2
|
||||||
|
assert bands[0]["node_count"] == 1
|
||||||
|
assert len(bands[0]["nodes"]) == 1
|
||||||
|
node_info = bands[0]["nodes"][0]
|
||||||
|
assert node_info["endpoint"] == "http://127.0.0.1:9001"
|
||||||
|
assert node_info["alive"] is True
|
||||||
|
assert "last_seen_seconds_ago" in node_info
|
||||||
|
assert "heartbeat_success_rate" in node_info
|
||||||
|
assert "inference_success_rate" in node_info
|
||||||
|
# Gap band: layers 3-5, no nodes
|
||||||
|
assert bands[1]["start_layer"] == 3
|
||||||
|
assert bands[1]["end_layer"] == 5
|
||||||
|
assert bands[1]["node_count"] == 0
|
||||||
|
assert bands[1]["nodes"] == []
|
||||||
|
|
||||||
try:
|
try:
|
||||||
_get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
_get_json(f"http://127.0.0.1:{tracker_port}/v1/route?model=tiny-model")
|
||||||
@@ -283,11 +297,12 @@ def test_tracker_coverage_endpoint_accepts_registered_hf_repo_or_short_name():
|
|||||||
)
|
)
|
||||||
|
|
||||||
assert by_repo["model"] == "Qwen/Qwen2.5-0.5B-Instruct"
|
assert by_repo["model"] == "Qwen/Qwen2.5-0.5B-Instruct"
|
||||||
assert by_repo["coverage"] == [
|
bands = by_repo["coverage"]
|
||||||
{"start_layer": 0, "end_layer": 19, "node_count": 1},
|
assert len(bands) == 3
|
||||||
{"start_layer": 20, "end_layer": 21, "node_count": 2},
|
assert bands[0]["start_layer"] == 0 and bands[0]["end_layer"] == 19 and bands[0]["node_count"] == 1
|
||||||
{"start_layer": 22, "end_layer": 23, "node_count": 1},
|
assert bands[1]["start_layer"] == 20 and bands[1]["end_layer"] == 21 and bands[1]["node_count"] == 2
|
||||||
]
|
assert bands[2]["start_layer"] == 22 and bands[2]["end_layer"] == 23 and bands[2]["node_count"] == 1
|
||||||
|
assert all("nodes" in b for b in bands)
|
||||||
assert by_short_name["coverage"] == by_repo["coverage"]
|
assert by_short_name["coverage"] == by_repo["coverage"]
|
||||||
finally:
|
finally:
|
||||||
tracker.stop()
|
tracker.stop()
|
||||||
|
|||||||
Reference in New Issue
Block a user