stroblmeandClaude Opus 5
9c149f9ed4
/observability/timeseries and /flows read every metric_minute row in the window
and folded them in Python, so the 7d preset pulled a week of rows on each 30 s
poll. date_bin() does the binning now — the row count drops to the slices asked
for, and to flows × 60 for the sparklines. A window of zero hours used to divide
by nothing and answer 500; windows are clamped to an hour at the low end and to
the retention period at the high end, past which there is nothing to find.
/observability/runs returns {data, count} rather than a bare list, so a minute
busier than the 200-row cap says so instead of quietly showing its newest 200.
The count is only queried when the page comes back full, which keeps the poll
from handing back what the fold just saved.
failures_24h leaves the summary — the Home tile counts errors over the selected
window from the rollups, and nothing had read the field since.
Deleting a flow now takes its Run rows and their nodes, metrics and artifacts
with it. This lives in the route rather than in forget_flow because renaming a
flow calls that too, and a rename must keep its history. The observability
rollups stay: they are the record of what ran, and retention already prunes them.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Uq8mtNb97A7praJLyeEYgs
303 lines
9.7 KiB
Python
303 lines
9.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session
|
|
|
|
from app.core.config import settings
|
|
from app.models import EngineEvent, FlowRun, MetricBucket
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/observability"
|
|
FLOW = "observability-test"
|
|
|
|
|
|
def _seed(db: Session) -> None:
|
|
now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
|
|
db.add(
|
|
MetricBucket(
|
|
flow=FLOW,
|
|
node=f"{FLOW}.calc",
|
|
bucket=now - timedelta(minutes=1),
|
|
executions=4,
|
|
errors=1,
|
|
messages=6,
|
|
duration_sum_ms=40.0,
|
|
duration_max_ms=25.0,
|
|
lag_sum_ms=100.0,
|
|
lag_max_ms=60.0,
|
|
items=4,
|
|
)
|
|
)
|
|
db.add(
|
|
EngineEvent(
|
|
ts=now,
|
|
type="node_error",
|
|
flow=FLOW,
|
|
node=f"{FLOW}.calc",
|
|
detail="ValueError: bad input\nTraceback",
|
|
)
|
|
)
|
|
db.add(
|
|
EngineEvent(
|
|
ts=now, type="audit", flow=FLOW, detail="published", actor="a@example.com"
|
|
)
|
|
)
|
|
db.add(
|
|
FlowRun(
|
|
id="9-0",
|
|
flow=FLOW,
|
|
source="external",
|
|
started_at=now,
|
|
finished_at=now,
|
|
status="ok",
|
|
nodes=3,
|
|
duration_ms=12.5,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_observability_requires_authentication(client: TestClient) -> None:
|
|
assert client.get(f"{PREFIX}/summary").status_code == 401
|
|
|
|
|
|
def test_the_summary_answers_even_when_degraded(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
response = client.get(f"{PREFIX}/summary", headers=superuser_token_headers)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] in {"ok", "degraded"}
|
|
assert set(body["flows"]) == {
|
|
"total",
|
|
"running",
|
|
"paused",
|
|
"quarantined",
|
|
"invalid",
|
|
}
|
|
assert "error" in body["nodes"]
|
|
# A windowed count comes from /flows?hours= now, which is what the tile
|
|
# that used to read this actually sums.
|
|
assert "failures_24h" not in body
|
|
|
|
|
|
def test_a_flow_that_cannot_run_makes_the_summary_degraded(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""A loop on the canvas has to reach the health screen.
|
|
|
|
Validation runs on a build, not on this request, so what the engine already
|
|
knows is what this reports — and reporting it is the whole point: a flow
|
|
with a dependency loop cannot run, and the screen used to say "ok".
|
|
"""
|
|
from app.flow.pipeline import ValidationIssue
|
|
|
|
controller = client.app.state.flow_controller
|
|
before = controller.issues
|
|
before_flows = controller.store.list_flows
|
|
# The looping flow has to be one the store lists, otherwise "running"
|
|
# counts nothing either way and the assertion below proves nothing.
|
|
controller.store.list_flows = lambda: ["looping"]
|
|
controller.issues = [
|
|
ValidationIssue(
|
|
code="cycle",
|
|
message="These nodes depend on each other in a loop",
|
|
flow="looping",
|
|
nodes=["looping.a", "looping.b"],
|
|
),
|
|
# Already counted as a node that failed to load, so not again here.
|
|
ValidationIssue(
|
|
code="node_error", message="boom", flow="broken", node="broken.x"
|
|
),
|
|
ValidationIssue(
|
|
code="unauthenticated_hook", message="open", flow="hooky", node="hooky.h"
|
|
),
|
|
]
|
|
try:
|
|
body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json()
|
|
finally:
|
|
controller.issues = before
|
|
controller.store.list_flows = before_flows
|
|
|
|
assert body["status"] == "degraded"
|
|
assert body["flows"]["invalid"] == 1
|
|
assert body["flows"]["running"] == 0
|
|
assert any("looping" in problem for problem in body["problems"])
|
|
assert not any("broken" in problem for problem in body["problems"])
|
|
assert not any("hooky" in problem for problem in body["problems"])
|
|
|
|
|
|
def test_the_history_reads_back(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
_seed(db)
|
|
|
|
points = client.get(f"{PREFIX}/timeseries", headers=superuser_token_headers).json()
|
|
mine = [point for point in points if point["executions"]]
|
|
assert mine and mine[-1]["avg_ms"] > 0
|
|
|
|
flows = client.get(f"{PREFIX}/flows", headers=superuser_token_headers).json()
|
|
row = next(entry for entry in flows if entry["flow"] == FLOW)
|
|
assert (row["executions"], row["errors"], row["messages"]) == (4, 1, 6)
|
|
assert len(row["spark"]) == 60
|
|
assert row["last_error_ts"] is not None
|
|
|
|
runs = client.get(f"{PREFIX}/runs", headers=superuser_token_headers).json()
|
|
assert any(run["id"] == "9-0" and run["status"] == "ok" for run in runs["data"])
|
|
|
|
failures = client.get(f"{PREFIX}/events", headers=superuser_token_headers).json()
|
|
assert any("Traceback" in event["detail"] for event in failures)
|
|
assert all(event["type"] != "audit" for event in failures)
|
|
|
|
audit = client.get(
|
|
f"{PREFIX}/events", headers=superuser_token_headers, params={"kind": "audit"}
|
|
).json()
|
|
assert any(event["actor"] == "a@example.com" for event in audit)
|
|
|
|
dead = client.get(f"{PREFIX}/dead-letter", headers=superuser_token_headers)
|
|
assert dead.status_code == 200
|
|
assert isinstance(dead.json(), list)
|
|
|
|
|
|
def test_the_timeseries_folds_into_the_requested_bucket(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
"""The fold moved into SQL, so the slices still have to line up as before.
|
|
|
|
Slots are aligned to the epoch, which is what the old modulo did, and a
|
|
slice only exists where a row does.
|
|
"""
|
|
flow = "bucket-fold-test"
|
|
# The start of the current quarter hour, half an hour back so both slices
|
|
# sit inside a one hour window.
|
|
now = datetime.now(timezone.utc).timestamp()
|
|
first = datetime.fromtimestamp(now // 900 * 900 - 1800, timezone.utc)
|
|
for offset, executions in ((0, 1), (2, 2), (5, 4), (15, 8)):
|
|
db.add(
|
|
MetricBucket(
|
|
flow=flow,
|
|
node=f"{flow}.calc",
|
|
bucket=first + timedelta(minutes=offset),
|
|
executions=executions,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
points = client.get(
|
|
f"{PREFIX}/timeseries",
|
|
headers=superuser_token_headers,
|
|
params={"flow": flow, "hours": 1, "bucket_s": 900},
|
|
).json()
|
|
|
|
assert len(points) == 2
|
|
assert points[0]["executions"] == 1 + 2 + 4
|
|
assert points[1]["executions"] == 8
|
|
assert points[0]["ts"] % 900 == 0
|
|
|
|
|
|
def test_a_zero_hour_window_is_still_an_hour(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""`hours=0` divided by nothing and answered 500."""
|
|
for path in ("timeseries", "flows"):
|
|
response = client.get(
|
|
f"{PREFIX}/{path}", headers=superuser_token_headers, params={"hours": 0}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_the_runs_page_carries_its_total(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
"""A full page says how much it left behind."""
|
|
minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
|
|
hours=5
|
|
)
|
|
for index in range(3):
|
|
db.add(
|
|
FlowRun(
|
|
id=f"paged-{index}",
|
|
flow=FLOW,
|
|
started_at=minute + timedelta(seconds=index),
|
|
status="ok",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
body = client.get(
|
|
f"{PREFIX}/runs",
|
|
headers=superuser_token_headers,
|
|
params={
|
|
"since": minute.isoformat(),
|
|
"until": (minute + timedelta(minutes=1)).isoformat(),
|
|
"limit": 2,
|
|
},
|
|
).json()
|
|
|
|
assert len(body["data"]) == 2
|
|
assert body["count"] == 3
|
|
|
|
|
|
def test_runs_narrow_to_one_minute(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
"""A minute picked off a chart reaches past what the recent list holds."""
|
|
minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
|
|
hours=3
|
|
)
|
|
db.add(FlowRun(id="minute-in", flow=FLOW, started_at=minute, status="ok"))
|
|
db.add(
|
|
FlowRun(
|
|
id="minute-after",
|
|
flow=FLOW,
|
|
started_at=minute + timedelta(minutes=1),
|
|
status="ok",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
runs = client.get(
|
|
f"{PREFIX}/runs",
|
|
headers=superuser_token_headers,
|
|
params={
|
|
"since": minute.isoformat(),
|
|
"until": (minute + timedelta(minutes=1)).isoformat(),
|
|
},
|
|
).json()
|
|
|
|
# The upper bound is exclusive, so the run starting the next minute is not
|
|
# in this one.
|
|
assert [run["id"] for run in runs["data"]] == ["minute-in"]
|
|
|
|
|
|
def test_events_narrow_to_one_minute(
|
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
|
) -> None:
|
|
"""The failures list reaches as far back as the charts beside it."""
|
|
minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
|
|
hours=4
|
|
)
|
|
db.add(EngineEvent(ts=minute, type="node_error", flow=FLOW, detail="minute-in"))
|
|
db.add(
|
|
EngineEvent(
|
|
ts=minute + timedelta(minutes=1),
|
|
type="node_error",
|
|
flow=FLOW,
|
|
detail="minute-after",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
events = client.get(
|
|
f"{PREFIX}/events",
|
|
headers=superuser_token_headers,
|
|
params={
|
|
"flow": FLOW,
|
|
"since": minute.isoformat(),
|
|
"until": (minute + timedelta(minutes=1)).isoformat(),
|
|
},
|
|
).json()
|
|
|
|
# The upper bound is exclusive, so the failure a minute later is not in it.
|
|
assert [event["detail"] for event in events] == ["minute-in"]
|