From 516f13af73fb44ca3a715fd705b2e06384220d07 Mon Sep 17 00:00:00 2001 From: Bojan Date: Fri, 31 Jul 2026 16:32:25 +0200 Subject: [PATCH 1/5] feat(observability): add worker pressure flame graph --- docs/use-dkg/backpressure-observability.md | 23 ++ packages/cli/src/daemon/backpressure-log.ts | 23 ++ packages/cli/src/daemon/lifecycle.ts | 6 +- packages/cli/test/backpressure-log.test.ts | 29 +++ packages/cli/vitest.unit.config.ts | 1 + .../grafana-dashboard-dkg-node-logs.json | 209 ++++++++++++++++++ tools/observability/lib/dashboards.mjs | 86 ++++++- tools/observability/lib/queries.mjs | 26 +++ tools/observability/verify-profile-render.mjs | 43 ++++ 9 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 packages/cli/src/daemon/backpressure-log.ts create mode 100644 packages/cli/test/backpressure-log.test.ts diff --git a/docs/use-dkg/backpressure-observability.md b/docs/use-dkg/backpressure-observability.md index 6ea8aec4a6..2e7eb15a8d 100644 --- a/docs/use-dkg/backpressure-observability.md +++ b/docs/use-dkg/backpressure-observability.md @@ -141,6 +141,29 @@ Per-item enqueue/start logs are deliberately avoided. Transition and periodic summary logging make sustained pressure visible without creating a log storm that competes with the overloaded scheduler. +The daemon routes these records through its structured log sink. That keeps a +full local copy and forwards the same redacted record to enabled syslog and +OTLP exporters, so the records reach the Loki-backed Grafana dashboards. + +## Grafana worker-pressure flame graph + +The **DKG Node — Logs** dashboard includes a **Scheduler pressure** row. Its +flame graph expands the bounded `activeOperations` and `queuedOperations` +arrays, groups them by scheduler, lane, and operation source, and shows the +largest sampled `oldestAgeMs` in the selected time range. + +The two top-level branches have distinct meanings: + +- **active / admitted**: work occupying a worker slot; +- **queued / waiting**: work still waiting for admission. + +Block width is elapsed pressure age in milliseconds. It answers which source +was present and how old its oldest observed work became. It is not CPU time, +request share, invocation count, or an exact completed-job duration. The +monitor intentionally emits transition, periodic summary, and recovery records +instead of a per-item event stream, so Grafana must not claim more precision +than the log contract provides. + ## Metrics The common OpenTelemetry instruments use bounded `scheduler` and `lane` diff --git a/packages/cli/src/daemon/backpressure-log.ts b/packages/cli/src/daemon/backpressure-log.ts new file mode 100644 index 0000000000..a454aadb47 --- /dev/null +++ b/packages/cli/src/daemon/backpressure-log.ts @@ -0,0 +1,23 @@ +import type { OperationContext } from '@origintrail-official/dkg-core'; + +export interface BackpressureLogWriter { + info(context: OperationContext, message: string): void; + warn(context: OperationContext, message: string): void; +} + +/** + * Route monitor output through the structured daemon logger. + * + * The structured logger is the single path that persists records to the + * dashboard database and forwards redacted copies to syslog/OTLP. Keeping this + * adapter separate makes the remote-observability wiring testable without + * starting a daemon. + */ +export function createBackpressureLogEmitter( + logger: BackpressureLogWriter, + context: OperationContext, +): (level: 'info' | 'warn', message: string) => void { + return (level, message) => { + logger[level](context, message); + }; +} diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index b0fae05696..6a54f04c53 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -150,6 +150,7 @@ import { resolveMetricsCollectorConfig, } from '../metrics-collector-config.js'; import { createDaemonLogSink } from './log-sink.js'; +import { createBackpressureLogEmitter } from './backpressure-log.js'; import { startRpcUsageTelemetry } from './rpc-usage-log.js'; import { startDashboardLogVolumePruner } from './dashboard-log-volume-pruner.js'; import { SqliteSnapshotPageIndexStore } from './snapshot-page-index-store.js'; @@ -1172,7 +1173,10 @@ export async function runDaemonInner( appendFile(logFile, line + "\n").catch(() => {}); } const backpressureMonitor = new BackpressureMonitor({ - emit: (level, message) => log(`[${level}] ${message}`), + emit: createBackpressureLogEmitter( + new Logger('backpressure'), + createOperationContext('system'), + ), }); configureApiQueryPriority(process.env.DKG_API_QUERY_PRIORITY, { diff --git a/packages/cli/test/backpressure-log.test.ts b/packages/cli/test/backpressure-log.test.ts new file mode 100644 index 0000000000..dc85cbf554 --- /dev/null +++ b/packages/cli/test/backpressure-log.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { OperationContext } from '@origintrail-official/dkg-core'; +import { createBackpressureLogEmitter } from '../src/daemon/backpressure-log.js'; + +describe('backpressure log bridge', () => { + it('routes warning and recovery records through the structured logger', () => { + const context: OperationContext = { + operationId: 'backpressure-monitor', + operationName: 'system', + }; + const logger = { + info: vi.fn(), + warn: vi.fn(), + }; + const emit = createBackpressureLogEmitter(logger, context); + + emit('warn', '[backpressure] {"event":"transition"}'); + emit('info', '[backpressure] {"event":"recovered"}'); + + expect(logger.warn).toHaveBeenCalledWith( + context, + '[backpressure] {"event":"transition"}', + ); + expect(logger.info).toHaveBeenCalledWith( + context, + '[backpressure] {"event":"recovered"}', + ); + }); +}); diff --git a/packages/cli/vitest.unit.config.ts b/packages/cli/vitest.unit.config.ts index c1dee3b3fe..ad87481df4 100644 --- a/packages/cli/vitest.unit.config.ts +++ b/packages/cli/vitest.unit.config.ts @@ -31,6 +31,7 @@ export default defineConfig({ 'test/config.test.ts', 'test/status-route-rpc.test.ts', 'test/backpressure-route.test.ts', + 'test/backpressure-log.test.ts', 'test/status-route-store-quads.test.ts', 'test/query-route-lifecycle.test.ts', 'test/status-command-store.test.ts', diff --git a/tools/observability/grafana-dashboard-dkg-node-logs.json b/tools/observability/grafana-dashboard-dkg-node-logs.json index 0bdccacf73..683bb98ced 100644 --- a/tools/observability/grafana-dashboard-dkg-node-logs.json +++ b/tools/observability/grafana-dashboard-dkg-node-logs.json @@ -204,6 +204,215 @@ "expr": "sum(sum_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `rpc_usage` | logfmt | method != `` | unwrap count [$__auto]))" } ] + }, + { + "title": "Scheduler pressure", + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 34 + }, + "type": "row", + "collapsed": false + }, + { + "title": "", + "gridPos": { + "h": 3, + "w": 24, + "x": 0, + "y": 35 + }, + "type": "text", + "options": { + "mode": "markdown", + "content": "**Worker queue diagnostics (PR #2003).** The flame graph reads structured `[backpressure]` records from Loki. A leaf answers **which scheduler/lane/operation produced the work** and shows the **largest sampled elapsed age** observed in the selected time range. Active and queued work are separate branches. These sparse transition/summary samples describe pressure; they are not CPU profiles, invocation counts, or exact end-to-end job durations." + }, + "transparent": true + }, + { + "title": "Worker queue pressure flame graph — $node", + "gridPos": { + "h": 14, + "w": 24, + "x": 0, + "y": 38 + }, + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "type": "flamegraph", + "description": "Each leaf is a PR #2003 worker source. Width is its peak sampled oldest elapsed age in the selected range (milliseconds), split between admitted/active work and queued work. This is pressure age, not CPU utilization, request share, or exact completed-job runtime.", + "options": { + "showFlameGraphOnly": false + }, + "fieldConfig": { + "defaults": { + "unit": "ms" + }, + "overrides": [] + }, + "targets": [ + { + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "refId": "A", + "expr": "sum((max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[0].operation\", age=\"activeOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[1].operation\", age=\"activeOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[2].operation\", age=\"activeOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[3].operation\", age=\"activeOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[4].operation\", age=\"activeOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[5].operation\", age=\"activeOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[6].operation\", age=\"activeOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[7].operation\", age=\"activeOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`7` | unwrap age | __error__ = `` [$__range])))) or (max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[0].operation\", age=\"queuedOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[1].operation\", age=\"queuedOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[2].operation\", age=\"queuedOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[3].operation\", age=\"queuedOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[4].operation\", age=\"queuedOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[5].operation\", age=\"queuedOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[6].operation\", age=\"queuedOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[7].operation\", age=\"queuedOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`7` | unwrap age | __error__ = `` [$__range])))))", + "legendFormat": "0|0|all sampled worker pressure", + "instant": true, + "queryType": "instant" + }, + { + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "refId": "B", + "expr": "sum(max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[0].operation\", age=\"activeOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[1].operation\", age=\"activeOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[2].operation\", age=\"activeOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[3].operation\", age=\"activeOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[4].operation\", age=\"activeOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[5].operation\", age=\"activeOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[6].operation\", age=\"activeOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[7].operation\", age=\"activeOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`7` | unwrap age | __error__ = `` [$__range]))))", + "legendFormat": "1|0|active / admitted", + "instant": true, + "queryType": "instant" + }, + { + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "refId": "C", + "expr": "max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[0].operation\", age=\"activeOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[1].operation\", age=\"activeOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[2].operation\", age=\"activeOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[3].operation\", age=\"activeOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[4].operation\", age=\"activeOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[5].operation\", age=\"activeOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[6].operation\", age=\"activeOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"activeOperations[7].operation\", age=\"activeOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`active`, slot=`7` | unwrap age | __error__ = `` [$__range])))", + "legendFormat": "2|1|{{scheduler}}/{{lane}}/{{operation}}", + "instant": true, + "queryType": "instant" + }, + { + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "refId": "D", + "expr": "sum(max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[0].operation\", age=\"queuedOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[1].operation\", age=\"queuedOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[2].operation\", age=\"queuedOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[3].operation\", age=\"queuedOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[4].operation\", age=\"queuedOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[5].operation\", age=\"queuedOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[6].operation\", age=\"queuedOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[7].operation\", age=\"queuedOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`7` | unwrap age | __error__ = `` [$__range]))))", + "legendFormat": "1|0|queued / waiting", + "instant": true, + "queryType": "instant" + }, + { + "datasource": { + "type": "loki", + "uid": "${loki}" + }, + "refId": "E", + "expr": "max by (scheduler, lane, phase, operation) ((max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[0].operation\", age=\"queuedOperations[0].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`0` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[1].operation\", age=\"queuedOperations[1].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`1` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[2].operation\", age=\"queuedOperations[2].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`2` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[3].operation\", age=\"queuedOperations[3].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`3` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[4].operation\", age=\"queuedOperations[4].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`4` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[5].operation\", age=\"queuedOperations[5].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`5` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[6].operation\", age=\"queuedOperations[6].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`6` | unwrap age | __error__ = `` [$__range])) or (max_over_time({service_name=\"dkg-node\", service_instance_id=\"$node\"} |= `[backpressure]` | regexp `\\[backpressure\\] (?P\\{.*\\})` | line_format `{{.payload}}` | json scheduler, lane, operation=\"queuedOperations[7].operation\", age=\"queuedOperations[7].oldestAgeMs\" | operation != `` | label_format phase=`queued`, slot=`7` | unwrap age | __error__ = `` [$__range])))", + "legendFormat": "2|1|{{scheduler}}/{{lane}}/{{operation}}", + "instant": true, + "queryType": "instant" + } + ], + "transformations": [ + { + "id": "seriesToRows", + "options": {} + }, + { + "id": "extractFields", + "options": { + "source": "Metric", + "format": "regexp", + "regexp": "(?\\d+)\\|(?[01])\\|(?