From 3d1e2078256359eace314d8f24c9d409291390bf Mon Sep 17 00:00:00 2001 From: Dmitry Sinyavin Date: Fri, 28 Aug 2026 11:50:03 +0200 Subject: [PATCH] Add events per node e2e support --- .claude/skills/tart-backend/SKILL.md | 1 + docs/grafana-guide.md | 56 +++ .../provisioning/dashboards/tart-global.json | 386 +++++++++++++++++- src/grafana.rs | 63 +++ src/grafana_store.rs | 155 ++++++- src/grafana_types.rs | 24 ++ tests/grafana_tests.rs | 144 +++++++ 7 files changed, 805 insertions(+), 24 deletions(-) diff --git a/.claude/skills/tart-backend/SKILL.md b/.claude/skills/tart-backend/SKILL.md index 9156ad2..35d5245 100644 --- a/.claude/skills/tart-backend/SKILL.md +++ b/.claude/skills/tart-backend/SKILL.md @@ -96,6 +96,7 @@ curl -s "$BASE/grafana/stats?start=$(date -u -d '5 min ago' +%FT%TZ)&end=$(date - `/grafana/validator-profiling` (+ `-timeseries`) — slow or failing guarantors vs the rest - `/grafana/validators/cores` — observed node→core mapping - `/grafana/nodes` — every node ever seen: version, last heard from +- `/grafana/events-by-node` — which nodes report the most of a given event type (top senders, with address and version) - `/grafana/node-stats` (+ `-aggregate`) — Status(10) snapshots: peers, DA store, guarantee pool **Which service is expensive?** diff --git a/docs/grafana-guide.md b/docs/grafana-guide.md index 9dda64d..528420f 100644 --- a/docs/grafana-guide.md +++ b/docs/grafana-guide.md @@ -905,6 +905,46 @@ curl 'http://localhost:8080/api/grafana/guarantee-discards?start=2025-01-15T00:0 --- +### 1.25 GET /api/grafana/events-by-node + +Per-node totals for a set of event types over a range, ranked by count — which nodes are the top senders of a given event. Each row carries the node's share of the network-wide total and its handshake identity (address, implementation, version, connection state, last seen) joined from the `nodes` table. + +**Query:** `EventsByNodeQuery` + +| Param | Type | Required | Default | Description | +|-------|------|----------|---------|-------------| +| `start` | ISO 8601 datetime | yes | - | Start of time range | +| `end` | ISO 8601 datetime | yes | - | End of time range | +| `event_types` | string | no | all types | Comma-separated list: numeric IDs, group names, or event names. Supports Grafana `{a,b}` braces. | +| `interval` | string | no | derived from range | Resolution hint (same values as `/timeseries`). Only selects the aggregate tier the totals are summed from: sub-minute values read the fresh 30 s counts, `1m` and up the `_1m` / `_1h` aggregates (which trail ingest by a few minutes). Omitted: range length / 300, mirroring Grafana's `$__interval`. | +| `limit` | u32 | no | 50 | Maximum number of nodes returned, highest count first. Capped at 2048. | + +**Aggregate Table Auto-Selection:** same rules as `/timeseries` (see 1.1), driven by `interval` (or the derived value) and the age of `start`. + +**Response:** Array sorted by descending `count`, then ascending `node_id`: + +```json +[{ + "node_id": "9d7aa2cb55242be29e4161d708fecf3a092255a4f748d2aa6f52c3b292c58f09", + "count": 1234, + "share": 0.182, + "address": "192.168.20.59:56318", + "implementation_name": "PolkaJam", + "implementation_version": "0.1.28", + "is_connected": true, + "last_seen_at": "2026-08-27T11:45:09Z" +}] +``` + +`share` is `count` divided by the total over **all** nodes in the range (computed before `limit` is applied), so the returned shares need not sum to 1. The node metadata fields are `null` if the node's handshake record is missing. + +```bash +# Who reported the most WorkPackageFailed(92) in the last hour? +curl 'http://localhost:8080/api/grafana/events-by-node?start=2025-01-15T00:00:00Z&end=2025-01-15T01:00:00Z&event_types=92&limit=10' +``` + +--- + ## 2. Shared Query Types All query parameter structs are defined in `src/grafana.rs`. @@ -1008,6 +1048,19 @@ struct EventsQuery { } ``` +### EventsByNodeQuery +Used by: `/events-by-node` + +```rust +struct EventsByNodeQuery { + start: DateTime, // required + end: DateTime, // required + event_types: Option, // comma-sep IDs/groups/names; all types if omitted + interval: Option, // resolution hint -> aggregate tier; derived from range if omitted + limit: Option, // default 50, max 2048 +} +``` + ### Special parsing - **`parse_service_ids()`**: Strips Grafana `{a,b}` braces, accepts decimal or `0x` hex, parses as u32 then casts to i32 @@ -1037,6 +1090,8 @@ All dashboards use the **Infinity** data source plugin (uid: `jamtart-api`, JSON | Nodes Active | stat | `/api/grafana/nodes` | — | | Live Events | timeseries | `/api/grafana/timeseries` | event_types=${event_group}, group_by=event_type, interval=1m | | Event Type Details | stat | `/api/grafana/timeseries` | event_types=${event_type}, group_by=event_type, interval=1m | +| Selected Events by Node | timeseries | `/api/grafana/timeseries` | event_types=${event_type}, group_by=node_id, interval=$__interval | +| Top Senders (Selected Events) | table | `/api/grafana/events-by-node` | event_types=${event_type}, interval=$__interval, limit=100 | | Failures Rate | stat | `/api/grafana/bottlenecks` | — | | WP Guarantee Rate | stat | `/api/grafana/timeseries` | event_types=guarantee_receiving | | Block Rate | timeseries | `/api/grafana/timeseries` | event_types=42 | @@ -1188,6 +1243,7 @@ Which endpoints are used by which dashboards: | Endpoint | Global | Blocks | Cores | Services | Node | DA | Connectivity | |----------|:------:|:------:|:-----:|:--------:|:----:|:--:|:------------:| | `/timeseries` | x | x | x | | x | | x | +| `/events-by-node` | x | | | | | | | | `/stats` | x | x | | | | | x | | `/nodes` | x | | | | x | | x | | `/cores` | | | x | | | | | diff --git a/grafana/provisioning/dashboards/tart-global.json b/grafana/provisioning/dashboards/tart-global.json index 2f4dd84..9958701 100644 --- a/grafana/provisioning/dashboards/tart-global.json +++ b/grafana/provisioning/dashboards/tart-global.json @@ -1889,13 +1889,379 @@ "description": "[JIP-3] All failure and discard events across the network: AuthoringFailed (41), BlockVerificationFailed (44), BlockExecutionFailed (46), BlockRequestFailed (65), TicketGenerationFailed (81), TicketTransferFailed (83), WorkPackageFailed (92), DuplicateWorkPackage (93), WorkPackageSharingFailed (99), GuaranteeSendFailed (107), GuaranteeReceiveFailed (111), GuaranteeDiscarded (113), ShardRequestFailed (122), AssuranceSendFailed (127), AssuranceReceiveFailed (130), and others. Grouped by type. (1-minute buckets from all_event_stats_1m)", "type": "table" }, + { + "datasource": { + "type": "yesoreyeram-infinity-datasource", + "uid": "jamtart-api" + }, + "description": "[JIP-3] Per-node counts of the event types picked in the Event Type variable \u2014 one line per reporting node, legend sorted by total so the top senders come first. Same data as Selected Events, split by node (group_by=node_id). Resolution follows the range: 30-second counts below 1-minute intervals, 1-minute / 1-hour aggregates above.", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 0, + "y": 32 + }, + "id": 42, + "options": { + "legend": { + "calcs": [ + "sum" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true, + "sortBy": "Total", + "sortDesc": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "columns": [ + { + "selector": "ts", + "text": "Time", + "type": "timestamp" + }, + { + "selector": "node_id", + "text": "Node ID", + "type": "string" + }, + { + "selector": "count", + "text": "Events", + "type": "number" + } + ], + "datasource": { + "type": "yesoreyeram-infinity-datasource", + "uid": "jamtart-api" + }, + "format": "table", + "parser": "backend", + "refId": "A", + "root_selector": "", + "source": "url", + "type": "json", + "url": "http://tart-backend:8080/api/grafana/timeseries?event_types=${event_type}&group_by=node_id&interval=${__interval}&start=${__from:date:iso}&end=${__to:date:iso}", + "url_options": { + "data": "", + "headers": [], + "method": "GET", + "params": [] + } + } + ], + "title": "Selected Events by Node", + "type": "timeseries", + "transformations": [ + { + "id": "partitionByValues", + "options": { + "fields": [ + "Node ID" + ], + "keepFields": false, + "naming": { + "asLabels": true + } + } + }, + { + "id": "renameByRegex", + "options": { + "regex": "^(.*?)([0-9a-f]{12})[0-9a-f]{52}(.*)$", + "renamePattern": "$1$2\u2026$3" + } + } + ], + "maxDataPoints": 300 + }, + { + "datasource": { + "type": "yesoreyeram-infinity-datasource", + "uid": "jamtart-api" + }, + "description": "[JIP-3] Nodes ranked by how many of the selected event types they reported in the time range (top 100), with each node's share of the network-wide total and its handshake identity: address, implementation, version, connection state, last seen. Click a Node ID to open the TART Node dashboard for it.", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "custom": { + "align": "auto", + "cellOptions": { + "type": "auto" + }, + "inspect": false + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "text", + "value": null + } + ] + } + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "Node ID" + }, + "properties": [ + { + "id": "links", + "value": [ + { + "title": "Open TART Node dashboard", + "url": "/d/tart-grafana-node?var-node_id=${__data.fields[\"Node ID\"]}&${__url_time_range}" + } + ] + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Events" + }, + "properties": [ + { + "id": "custom.cellOptions", + "value": { + "type": "gauge", + "mode": "gradient", + "valueDisplayMode": "text" + } + }, + { + "id": "color", + "value": { + "mode": "continuous-BlPu" + } + }, + { + "id": "custom.width", + "value": 160 + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Share" + }, + "properties": [ + { + "id": "unit", + "value": "percentunit" + }, + { + "id": "decimals", + "value": 1 + }, + { + "id": "custom.width", + "value": 80 + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "Connected" + }, + "properties": [ + { + "id": "mappings", + "value": [ + { + "options": { + "false": { + "color": "red", + "index": 1, + "text": "No" + }, + "true": { + "color": "green", + "index": 0, + "text": "Yes" + } + }, + "type": "value" + } + ] + }, + { + "id": "custom.cellOptions", + "value": { + "type": "color-text" + } + }, + { + "id": "custom.width", + "value": 90 + } + ] + } + ] + }, + "gridPos": { + "h": 10, + "w": 12, + "x": 12, + "y": 32 + }, + "id": 43, + "options": { + "cellHeight": "sm", + "footer": { + "countRows": false, + "fields": "", + "reducer": [ + "sum" + ], + "show": false + }, + "showHeader": true, + "sortBy": [ + { + "desc": true, + "displayName": "Events" + } + ] + }, + "targets": [ + { + "columns": [ + { + "selector": "node_id", + "text": "Node ID", + "type": "string" + }, + { + "selector": "count", + "text": "Events", + "type": "number" + }, + { + "selector": "share", + "text": "Share", + "type": "number" + }, + { + "selector": "address", + "text": "Address", + "type": "string" + }, + { + "selector": "implementation_name", + "text": "Impl", + "type": "string" + }, + { + "selector": "implementation_version", + "text": "Version", + "type": "string" + }, + { + "selector": "is_connected", + "text": "Connected", + "type": "string" + }, + { + "selector": "last_seen_at", + "text": "Last Seen", + "type": "timestamp" + } + ], + "datasource": { + "type": "yesoreyeram-infinity-datasource", + "uid": "jamtart-api" + }, + "format": "table", + "parser": "backend", + "refId": "A", + "root_selector": "", + "source": "url", + "type": "json", + "url": "http://tart-backend:8080/api/grafana/events-by-node?event_types=${event_type}&interval=${__interval}&limit=100&start=${__from:date:iso}&end=${__to:date:iso}", + "url_options": { + "data": "", + "headers": [], + "method": "GET", + "params": [] + } + } + ], + "title": "Top Senders (Selected Events)", + "type": "table", + "maxDataPoints": 300 + }, { "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, - "y": 32 + "y": 42 }, "id": 16, "title": "Blocks & Pipeline", @@ -1960,7 +2326,7 @@ "h": 8, "w": 8, "x": 0, - "y": 33 + "y": 43 }, "id": 14, "options": { @@ -2074,7 +2440,7 @@ "h": 8, "w": 8, "x": 8, - "y": 33 + "y": 43 }, "id": 15, "options": { @@ -2295,7 +2661,7 @@ "h": 8, "w": 8, "x": 16, - "y": 33 + "y": 43 }, "id": 17, "options": { @@ -2407,7 +2773,7 @@ "h": 6, "w": 12, "x": 0, - "y": 41 + "y": 51 }, "id": 18, "options": { @@ -2547,7 +2913,7 @@ "h": 6, "w": 12, "x": 12, - "y": 41 + "y": 51 }, "id": 19, "options": { @@ -2616,7 +2982,7 @@ "h": 1, "w": 24, "x": 0, - "y": 47 + "y": 57 }, "id": 23, "title": "Nodes & Data Availability", @@ -2665,7 +3031,7 @@ "h": 8, "w": 12, "x": 0, - "y": 48 + "y": 58 }, "id": 24, "options": { @@ -2782,7 +3148,7 @@ "h": 8, "w": 12, "x": 12, - "y": 48 + "y": 58 }, "id": 25, "options": { @@ -2901,7 +3267,7 @@ "h": 4, "w": 24, "x": 0, - "y": 48 + "y": 58 }, "id": 26, "options": { diff --git a/src/grafana.rs b/src/grafana.rs index eff38f0..81e43f9 100644 --- a/src/grafana.rs +++ b/src/grafana.rs @@ -23,6 +23,7 @@ use crate::onchain_types::*; #[openapi( paths( timeseries, + events_by_node, stats, cores_summary, core_detail, @@ -84,6 +85,7 @@ use crate::onchain_types::*; ), components(schemas( TimeseriesRow, + EventsByNodeRow, StatsResponse, CoreSummary, CoreDetail, @@ -189,6 +191,7 @@ pub struct GrafanaApiDoc; pub fn router() -> Router { Router::new() .route("/timeseries", get(timeseries)) + .route("/events-by-node", get(events_by_node)) .route("/stats", get(stats)) .route("/cores", get(cores_summary)) .route("/cores/:core_id", get(core_detail)) @@ -295,6 +298,21 @@ pub struct TimeseriesQuery { pub core: Option, } +/// Parameters for the per-node event totals endpoint. +#[derive(Deserialize, IntoParams)] +pub struct EventsByNodeQuery { + /// Start of time range (ISO 8601) + pub start: DateTime, + /// End of time range (ISO 8601) + pub end: DateTime, + /// Comma-separated event type codes, group names, or event names. Supports Grafana {a,b} syntax. Optional — if omitted, every type is counted. + pub event_types: Option, + /// Resolution hint (same values as /timeseries) that selects the aggregate tier the totals are summed from. Omitted: derived from the range length. + pub interval: Option, + /// Maximum number of nodes to return, highest count first (default: 50, max: 2048) + pub limit: Option, +} + /// Common time range + optional filters used by most endpoints. #[derive(Deserialize, IntoParams)] pub struct TimeRangeQuery { @@ -545,6 +563,51 @@ async fn timeseries( .map_err(|e| map_sqlx_error("grafana/timeseries", e)) } +/// Which nodes report the most of a given event: per-node totals over the range, +/// highest first, with each node's identity attached. +/// +/// Counts come from the same pre-aggregated tiers as `/timeseries`. The optional +/// `interval` is only a resolution hint that picks the tier: sub-minute values +/// read the fresh 30 s counts, anything from one minute up the 1 min or 1 h +/// aggregates, which trail ingestion by a few minutes. Omitted, it is derived +/// from the range length (about 300 buckets), so short ranges see the freshest +/// data and long ranges stay cheap. +/// +/// Answers: which nodes are the top senders of this event type, and who are they? +#[utoipa::path( + get, + path = "/api/grafana/events-by-node", + params(EventsByNodeQuery), + responses( + (status = 200, description = "Array of one row per reporting node, descending by count, each with the node's total for the selected event types, its share of the network-wide total and its handshake identity (address, implementation, version, connection state, last seen).", body = [EventsByNodeRow]), + (status = 500, description = "Database error"), + ), + tag = "grafana" +)] +async fn events_by_node( + Query(q): Query, + State(state): State, +) -> Result { + let event_types: Option> = q + .event_types + .map(|s| crate::event_type_meta::expand_event_types(&s)) + .filter(|v| !v.is_empty()); + let limit = i64::from(q.limit.unwrap_or(50).clamp(1, 2048)); + + state + .store + .grafana_events_by_node( + q.start, + q.end, + q.interval.as_deref(), + event_types.as_deref(), + limit, + ) + .await + .map(Json) + .map_err(|e| map_sqlx_error("grafana/events-by-node", e)) +} + /// Headline network counters for a dashboard summary row. /// /// Over the requested range: how many GuaranteeBuilt(105), WorkPackageFailed(92) diff --git a/src/grafana_store.rs b/src/grafana_store.rs index c83bd80..73fcbfa 100644 --- a/src/grafana_store.rs +++ b/src/grafana_store.rs @@ -101,6 +101,22 @@ fn snap_interval(input: &str) -> &'static str { "1d" } +/// Pick the event-count tier for a query: by resolution first (sub-minute → +/// 30 s raw counts, sub-hour → 1 m aggregates, otherwise 1 h), then bumped to +/// a coarser tier when the range starts beyond the finer tier's retention +/// (30 s counts keep 3 days, 1 m aggregates 30 days). The upgrade is silent: +/// the finer data simply no longer exists for that part of the range. +fn select_event_stats_table(interval_secs: i64, start: DateTime) -> &'static str { + let age = Utc::now() - start; + if interval_secs < 60 && age <= chrono::Duration::days(3) { + "all_event_stats_30s" + } else if interval_secs < 3600 && age <= chrono::Duration::days(30) { + "all_event_stats_1m" + } else { + "all_event_stats_1h" + } +} + impl EventStore { // ── 1. grafana_timeseries ────────────────────────────────────────── @@ -134,23 +150,10 @@ impl EventStore { let pg_interval = interval_to_pg(interval); // Select aggregate table (interval-based, then retention-aware upgrade) - let age = Utc::now() - start; let table = if group_by == Some("core") || core.is_some() { "all_core_stats_1m" - } else if interval_secs < 60 { - if age > chrono::Duration::days(3) { - "all_event_stats_1m" // 30s retention is 3 days, upgrade silently - } else { - "all_event_stats_30s" - } - } else if interval_secs < 3600 { - if age > chrono::Duration::days(30) { - "all_event_stats_1h" // 1m retention is 30 days, upgrade silently - } else { - "all_event_stats_1m" - } } else { - "all_event_stats_1h" + select_event_stats_table(interval_secs, start) }; // Safety: table is from a hardcoded set @@ -254,6 +257,91 @@ impl EventStore { Ok(results) } + // ── 1b. grafana_events_by_node ───────────────────────────────────── + + /// Per-node totals for a set of event types over a range, highest first, + /// joined with each node's handshake identity. + /// + /// Reads the same aggregate tiers as `grafana_timeseries`. `interval` is only + /// a resolution hint that picks the tier; `None` derives it from the range + /// length as roughly 300 buckets, mirroring Grafana's `$__interval`, so short + /// ranges read the fresh 30 s counts and long ranges the 1 m / 1 h aggregates. + /// `share` is computed over all nodes before `limit` is applied. + pub async fn grafana_events_by_node( + &self, + start: DateTime, + end: DateTime, + interval: Option<&str>, + event_types: Option<&[i16]>, + limit: i64, + ) -> Result, sqlx::Error> { + let interval_secs = match interval { + Some(i) => interval_to_seconds(snap_interval(i)).unwrap_or(60), + None => ((end - start).num_seconds() / 300).max(1), + }; + let table = select_event_stats_table(interval_secs, start); + + // Safety: table is from a hardcoded set + if !VALID_TABLES.contains(&table) { + return Err(sqlx::Error::Protocol(format!("invalid table: {table}"))); + } + + let (type_filter, limit_idx) = if event_types.is_some() { + ("AND event_type = ANY($3)", 4) + } else { + ("", 3) + }; + + // `share` is a window over the full per-node totals, so it is relative + // to every node in the range even when LIMIT trims the output. + let sql = format!( + r#" + WITH totals AS ( + SELECT node_id, SUM(event_count)::BIGINT AS count + FROM {table} + WHERE bucket >= $1 AND bucket < $2 {type_filter} + GROUP BY node_id + ) + SELECT + t.node_id, + t.count, + t.count::DOUBLE PRECISION + / NULLIF(SUM(t.count) OVER (), 0)::DOUBLE PRECISION AS share, + n.address, + n.implementation_name, + n.implementation_version, + n.is_connected, + n.last_seen_at + FROM totals t + LEFT JOIN nodes n ON n.node_id = t.node_id + ORDER BY t.count DESC, t.node_id ASC + LIMIT ${limit_idx} + "#, + ); + + let mut query = sqlx::query(&sql).bind(start).bind(end); + if let Some(types) = event_types { + query = query.bind(types.to_vec()); + } + query = query.bind(limit); + + let rows = query.fetch_all(self.pool()).await?; + + Ok(rows + .iter() + .map(|row| EventsByNodeRow { + node_id: row.get("node_id"), + count: row.get("count"), + share: row.get::, _>("share").unwrap_or(0.0), + address: row.get("address"), + implementation_name: row.get("implementation_name"), + implementation_version: row.get("implementation_version"), + is_connected: row.get("is_connected"), + last_seen_at: row.get("last_seen_at"), + }) + .collect()) + } + // ── 2. grafana_stats ─────────────────────────────────────────────── /// Dashboard summary stats: connected nodes, current slot, guarantees, failures, WP events. @@ -3843,4 +3931,43 @@ mod tests { assert!(p95 <= p99, "p95 ({p95}) <= p99 ({p99})"); assert!(p99 <= p100, "p99 ({p99}) <= p100 ({p100})"); } + + #[test] + fn event_stats_tier_by_resolution() { + let recent = Utc::now() - chrono::Duration::hours(1); + assert_eq!(select_event_stats_table(30, recent), "all_event_stats_30s"); + assert_eq!(select_event_stats_table(59, recent), "all_event_stats_30s"); + assert_eq!(select_event_stats_table(60, recent), "all_event_stats_1m"); + assert_eq!(select_event_stats_table(3599, recent), "all_event_stats_1m"); + assert_eq!(select_event_stats_table(3600, recent), "all_event_stats_1h"); + } + + #[test] + fn event_stats_tier_upgrades_past_retention() { + // 30 s counts keep 3 days: older ranges fall back to the 1 m aggregates. + let four_days = Utc::now() - chrono::Duration::days(4); + assert_eq!( + select_event_stats_table(30, four_days), + "all_event_stats_1m" + ); + assert_eq!( + select_event_stats_table(60, four_days), + "all_event_stats_1m" + ); + // 1 m aggregates keep 30 days: anything older reads the 1 h tier, whatever + // resolution was asked for. + let forty_days = Utc::now() - chrono::Duration::days(40); + assert_eq!( + select_event_stats_table(30, forty_days), + "all_event_stats_1h" + ); + assert_eq!( + select_event_stats_table(60, forty_days), + "all_event_stats_1h" + ); + assert_eq!( + select_event_stats_table(3600, forty_days), + "all_event_stats_1h" + ); + } } diff --git a/src/grafana_types.rs b/src/grafana_types.rs index 39172ee..b529b88 100644 --- a/src/grafana_types.rs +++ b/src/grafana_types.rs @@ -161,6 +161,30 @@ pub struct TimeseriesRow { pub node_id: Option, } +// ── /api/grafana/events-by-node ───────────────────────────────────────── + +/// One node's total for the selected event types over the range, with the node's +/// handshake identity attached — the "who reports most of this event" view. +#[derive(Debug, Serialize, ToSchema)] +pub struct EventsByNodeRow { + /// Node's JIP-3 peer ID — its Ed25519 public key, hex-encoded (64 characters) + pub node_id: String, + /// Events of the selected types this node reported in the range + pub count: i64, + /// `count` divided by the total over all nodes in the range (0–1) + pub share: f64, + /// Network address the node's telemetry session came from (`ip:port`) + pub address: Option, + /// Implementation name the node reported at handshake (e.g. "polkajam") + pub implementation_name: Option, + /// Implementation version the node reported at handshake + pub implementation_version: Option, + /// Whether the node is reporting telemetry right now + pub is_connected: Option, + /// When the node was last heard from + pub last_seen_at: Option>, +} + // ── /api/grafana/cores ────────────────────────────────────────────────── /// One core's work-package activity over the time range. diff --git a/tests/grafana_tests.rs b/tests/grafana_tests.rs index 6fc354c..bc1ca09 100644 --- a/tests/grafana_tests.rs +++ b/tests/grafana_tests.rs @@ -110,6 +110,10 @@ async fn test_grafana_all_endpoints_empty_200() { "/api/grafana/timeseries?{}&interval=1m", time_range_params() ), + format!( + "/api/grafana/events-by-node?{}&event_types=42", + time_range_params() + ), format!("/api/grafana/stats?{}", time_range_params()), format!("/api/grafana/cores?{}", time_range_params()), format!("/api/grafana/blocks/convergence?{}", time_range_params()), @@ -1240,6 +1244,146 @@ async fn test_grafana_timeseries_node_and_event_type_filters() { } } +// ───────────────────────────────────────────────────────────────────────────── +// Events by node: per-node totals ranked by count +// ───────────────────────────────────────────────────────────────────────────── + +fn assert_share(entry: &Value, expected: f64) { + let share = entry["share"].as_f64().expect("share should be a number"); + assert!( + (share - expected).abs() < 1e-9, + "expected share {expected}, got {share}" + ); +} + +#[tokio::test] +async fn test_grafana_events_by_node_ranking() { + let (server, telemetry, port, store) = setup_test_api().await; + let mut stream1 = connect_test_node(port, 1, &telemetry).await; + let mut stream2 = connect_test_node(port, 2, &telemetry).await; + + let ts = common::now_jce_micros(); + + // Node 1: three WorkPackageReceived(94) + one BestBlockChanged(11). + send_events( + &mut stream1, + &[ + common::wp_received_event(ts, 3000, 3), + common::wp_received_event(ts + 1000, 3001, 5), + common::wp_received_event(ts + 2000, 3002, 7), + common::best_block_event(ts + 3000, 100), + ], + ) + .await; + // Node 2: one WorkPackageReceived(94). + send_events( + &mut stream2, + &[common::wp_received_event(ts + 4000, 4000, 3)], + ) + .await; + common::flush_all(&telemetry).await; + common::refresh_aggregates(store.pool()).await; + + let node1_id = common::node_id_hex(1); + let node2_id = common::node_id_hex(2); + + // Both tiers must agree: 30s raw counts (sub-minute hint) and 1m aggregates. + for interval in ["30s", "1m"] { + let path = format!( + "/api/grafana/events-by-node?{}&event_types=94&interval={}", + time_range_params(), + interval + ); + let response = server.get(&path).await; + assert_eq!( + response.status_code(), + StatusCode::OK, + "interval={interval}" + ); + let json: Value = response.json(); + let arr = json + .as_array() + .expect("events-by-node should return an array"); + assert_eq!( + arr.len(), + 2, + "two nodes reported type 94 (interval={interval})" + ); + + // Ranked by count, highest first. + assert_eq!(arr[0]["node_id"], node1_id, "interval={interval}"); + assert_eq!(arr[0]["count"], 3, "interval={interval}"); + assert_share(&arr[0], 0.75); + assert_eq!(arr[1]["node_id"], node2_id, "interval={interval}"); + assert_eq!(arr[1]["count"], 1, "interval={interval}"); + assert_share(&arr[1], 0.25); + + // Node identity joined from the nodes table. + assert_eq!(arr[0]["implementation_name"], "test-node-1"); + assert_eq!(arr[1]["implementation_name"], "test-node-2"); + assert_eq!(arr[0]["is_connected"], true); + let address = arr[0]["address"] + .as_str() + .expect("address should be present for a connected node"); + assert!( + address.starts_with("127.0.0.1:"), + "address should be the telemetry session's peer address, got {address}" + ); + assert!(arr[0]["last_seen_at"].is_string(), "last_seen_at missing"); + } + + // event_types filter: only node 1 reported BestBlockChanged(11). + let path = format!( + "/api/grafana/events-by-node?{}&event_types=11", + time_range_params() + ); + let response = server.get(&path).await; + assert_eq!(response.status_code(), StatusCode::OK); + let json: Value = response.json(); + let arr = json.as_array().expect("should return array"); + assert_eq!(arr.len(), 1); + assert_eq!(arr[0]["node_id"], node1_id); + assert_eq!(arr[0]["count"], 1); + assert_share(&arr[0], 1.0); + + // limit keeps the top sender; share stays relative to all nodes. + let path = format!( + "/api/grafana/events-by-node?{}&event_types=94&limit=1", + time_range_params() + ); + let response = server.get(&path).await; + assert_eq!(response.status_code(), StatusCode::OK); + let json: Value = response.json(); + let arr = json.as_array().expect("should return array"); + assert_eq!(arr.len(), 1); + assert_eq!(arr[0]["node_id"], node1_id); + assert_share(&arr[0], 0.75); + + // Group names and Grafana braces work like in /timeseries: {wp_pipeline} covers type 94. + let path = format!( + "/api/grafana/events-by-node?{}&event_types=%7Bwp_pipeline%7D", + time_range_params() + ); + let response = server.get(&path).await; + assert_eq!(response.status_code(), StatusCode::OK); + let json: Value = response.json(); + let arr = json.as_array().expect("should return array"); + assert_eq!(arr.len(), 2); + assert_eq!(arr[0]["node_id"], node1_id); + assert_eq!(arr[0]["count"], 3); + + // No event_types: every type counts, node 1 (4 events) still ranks first. + let path = format!("/api/grafana/events-by-node?{}", time_range_params()); + let response = server.get(&path).await; + assert_eq!(response.status_code(), StatusCode::OK); + let json: Value = response.json(); + let arr = json.as_array().expect("should return array"); + assert_eq!(arr.len(), 2); + assert_eq!(arr[0]["node_id"], node1_id); + assert!(arr[0]["count"].as_i64().unwrap() >= 4); + assert_eq!(arr[1]["node_id"], node2_id); +} + // ───────────────────────────────────────────────────────────────────────────── // Event types endpoint with group filter // ─────────────────────────────────────────────────────────────────────────────