Home charts execution time per node type on a log scale, from a /observability/node-types rollup

backend/fluksio/api/routes/observability.py

@@ -86,6 +86,17 @@ class FlowRollup(BaseModel):
     last_error_ts: float | None = None
 
 
+class NodeTypeTiming(BaseModel):
+    """How long one kind of node takes to run, across every node of it."""
+
+    type: str
+    #: How many published nodes of this type ran inside the window.
+    nodes: int
+    executions: int
+    avg_ms: float
+    max_ms: float
+
+
 class RunRow(BaseModel):
     id: str
     flow: str
@@ -428,6 +439,58 @@ def read_flow_rollups(
     ]
 
 
+@router.get("/node-types", response_model=list[NodeTypeTiming])
+def read_node_type_timings(
+    session: SessionDep,
+    controller: FlowControllerDep,
+    hours: Annotated[int, Query(ge=0, le=8760)] = 24,
+) -> Any:
+    """Execution time per node type, slowest first, to point at the hotspots.
+
+    The rollups are kept per node, so the type is joined on from the published
+    flows. A node deleted since it ran has no type left to count under.
+    """
+    hours = _window_hours(hours)
+    types = {
+        f"{flow.name}.{node.id}": node.type
+        for flow in controller.store.read_all()
+        for node in flow.nodes
+    }
+    statement = (
+        sa_select(
+            col(MetricBucket.node),
+            func.sum(col(MetricBucket.executions)).label("executions"),
+            func.sum(col(MetricBucket.duration_sum_ms)).label("duration_sum_ms"),
+            func.max(col(MetricBucket.duration_max_ms)).label("max_ms"),
+        )
+        .where(col(MetricBucket.bucket) >= _since(hours))
+        .group_by(col(MetricBucket.node))
+    )
+    totals: dict[str, dict[str, float]] = {}
+    for row in session.execute(statement):
+        kind = types.get(row.node)
+        if kind is None or not row.executions:
+            continue
+        entry = totals.setdefault(
+            kind, {"nodes": 0, "executions": 0, "duration_sum_ms": 0.0, "max_ms": 0.0}
+        )
+        entry["nodes"] += 1
+        entry["executions"] += row.executions
+        entry["duration_sum_ms"] += row.duration_sum_ms
+        entry["max_ms"] = max(entry["max_ms"], row.max_ms)
+    timings = [
+        NodeTypeTiming(
+            type=kind,
+            nodes=int(entry["nodes"]),
+            executions=int(entry["executions"]),
+            avg_ms=round(entry["duration_sum_ms"] / entry["executions"], 3),
+            max_ms=round(entry["max_ms"], 3),
+        )
+        for kind, entry in totals.items()
+    ]
+    return sorted(timings, key=lambda timing: timing.avg_ms, reverse=True)
+
+
 @router.get("/runs", response_model=RunPage)
 def read_runs(
     session: SessionDep,

backend/tests/api/routes/test_observability.py

@@ -512,3 +512,59 @@ def test_a_superuser_can_flush_the_collector_now(
         )
     ).all()
     assert len(written) == 1
+
+
+def test_execution_time_is_summed_per_node_type(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    from fluksio.flow.schemas import FlowDef, NodeDef
+
+    flow = "node_type_test"
+    controller = client.app.state.flow_controller
+    before = controller.store.read_all
+    controller.store.read_all = lambda: [
+        FlowDef(
+            name=flow,
+            nodes=[
+                NodeDef(id="a", type="python"),
+                NodeDef(id="b", type="python"),
+                NodeDef(id="tick", type="inject"),
+            ],
+        )
+    ]
+    now = datetime.now(UTC).replace(second=0, microsecond=0)
+    for node, executions, total, peak in (
+        ("a", 2, 30.0, 20.0),
+        ("b", 2, 10.0, 6.0),
+        ("tick", 4, 0.4, 0.2),
+        # Deleted since it ran: nothing to count it under.
+        ("gone", 1, 500.0, 500.0),
+    ):
+        db.add(
+            MetricBucket(
+                flow=flow,
+                node=f"{flow}.{node}",
+                bucket=now - timedelta(minutes=2),
+                executions=executions,
+                duration_sum_ms=total,
+                duration_max_ms=peak,
+            )
+        )
+    db.commit()
+    try:
+        rows = client.get(
+            f"{PREFIX}/node-types", headers=superuser_token_headers
+        ).json()
+    finally:
+        controller.store.read_all = before
+
+    # Slowest first.
+    assert [row["type"] for row in rows] == ["python", "inject"]
+    assert rows[0] == {
+        "type": "python",
+        "nodes": 2,
+        "executions": 4,
+        "avg_ms": 10.0,
+        "max_ms": 20.0,
+    }
+    assert rows[1]["avg_ms"] == 0.1