feat(observability): add sync pressure and source-cost flame graphs

.github/workflows/observability-artifacts.yml

@@ -23,6 +23,8 @@ on:
   pull_request:
     paths:
       - 'tools/observability/**'
+      # The pinned real-Loki result test mounts this exact single-binary config.
+      - 'tools/log-collection-poc/loki-config.yaml'
       # the gate must also verify itself when the gate changes
       - '.github/workflows/observability-artifacts.yml'
       # …and it must run when the SOURCES it mirrors change, not only when the
@@ -89,6 +91,43 @@ jobs:
           node tools/observability/verify-profile-render.mjs /tmp/render \
             --prom-node-label service_instance_id --vm-uid test-vm-uid --loki-uid test-loki-uid
 
+      # The Loki queries are valid even when a nested JSON field is misspelled;
+      # they simply return no rows. Run the generated dashboard expressions
+      # against the repository's Loki 3.3.2 configuration to pin actual
+      # scheduler/lane/phase/operation labels, moved-slot peaks, empty-slot
+      # handling, and the fixed-window resource bound. The multi-arch digest is
+      # the reproducibility contract; the tag is deliberately not used.
+      - name: Evaluate worker-pressure LogQL against pinned Loki
+        run: |
+          set -euo pipefail
+          container_id="$(docker run --rm -d \
+            -p 127.0.0.1:13100:3100 \
+            -v "${PWD}/tools/log-collection-poc/loki-config.yaml:/etc/loki/loki-config.yaml:ro" \
+            grafana/loki@sha256:8af2de1abbdd7aa92b27c9bcc96f0f4140c9096b507c77921ffddf1c6ad6c48f \
+            -config.file=/etc/loki/loki-config.yaml)"
+          cleanup() {
+            status=$?
+            trap - EXIT
+            if [ "$status" -ne 0 ]; then docker logs "$container_id"; fi
+            docker stop "$container_id" >/dev/null || true
+            exit "$status"
+          }
+          trap cleanup EXIT
+          ready=0
+          for _ in $(seq 1 30); do
+            if curl --fail --silent http://127.0.0.1:13100/ready >/dev/null; then
+              ready=1
+              break
+            fi
+            sleep 1
+          done
+          if [ "$ready" -ne 1 ]; then
+            echo 'Loki did not become ready' >&2
+            exit 1
+          fi
+          LOKI_URL=http://127.0.0.1:13100 \
+            node tools/observability/verify-backpressure-logql.mjs
+
       # W1 decision queries: the semantic contract behind the numbers, checked
       # per selector rather than by sampling. Run against BOTH renders so the
       # W1 queries are proven to follow --prom-node-label like every other
@@ -106,12 +145,12 @@ jobs:
       # contract. `--entrypoint promtool` is REQUIRED: the image entrypoint is
       # ["/bin/prometheus"], so the un-overridden form fails with
       # `prometheus: error: unexpected promtool`.
-      - name: Parse the W1 rule fixture (pinned promtool)
+      - name: Parse the W1 decision and Grafana-dashboard rule fixtures (pinned promtool)
         run: |
           set -euo pipefail
           docker run --rm -v "${PWD}/tools/observability:/w" --entrypoint promtool \
             prom/prometheus@sha256:6559acbd5d770b15bb3c954629ce190ac3cbbdb2b7f1c30f0385c4e05104e218 \
-            check rules /w/w1/w1-rules.yaml
+            check rules /w/w1/w1-rules.yaml /w/w1/w1-dashboard-rules.yaml
 
       # `check rules` proves the expressions PARSE. It is blind to what they
       # RETURN, and the defect that motivated these tests parsed perfectly:
@@ -147,3 +186,15 @@ jobs:
           docker run --rm -v /tmp/w1-promtool:/t:ro --entrypoint promtool \
             prom/prometheus@sha256:6559acbd5d770b15bb3c954629ce190ac3cbbdb2b7f1c30f0385c4e05104e218 \
             test rules /t/w1-rules.test.yaml
+
+      # The sync-cost fixture is expression-identical to the generated Grafana
+      # targets (verify-profile-render.mjs enforces that). Result assertions
+      # below are the second half of the contract: they pin label grouping,
+      # worker-equivalent and byte units, flame widths, histogram buckets/p95,
+      # zero-filling of request-only failures, and fail-closed evidence gates.
+      - name: Unit-test the generated sync-cost dashboard query results (pinned promtool)
+        run: |
+          set -euo pipefail
+          docker run --rm -v "${PWD}/tools/observability/w1:/t:ro" --entrypoint promtool \
+            prom/prometheus@sha256:6559acbd5d770b15bb3c954629ce190ac3cbbdb2b7f1c30f0385c4e05104e218 \
+            test rules /t/w1-dashboard-rules.test.yaml

docs/use-dkg/backpressure-observability.md

@@ -178,6 +178,92 @@ 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 and heatmaps
+
+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 a fixed most-recent one-hour window. The
+lookback is intentionally independent of the Grafana-selected range: Loki 3
+splits long-range instant metric queries internally, so letting an incident
+range of 6 or 24 hours flow into all five flamegraph targets can make the
+dashboard itself create backend pressure.
+
+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.
+
+Two heatmaps below the flame graph keep the active and queued meanings
+separate and retain the full Grafana-selected historical range. Each heatmap
+groups samples by scheduler/lane and shows the
+distribution over time of the peak sampled age in each Grafana resolution
+bucket:
+
+- **Active / admitted pressure age** shows how old the oldest work occupying a
+  worker slot became;
+- **Queued / waiting pressure age** shows how old the oldest work waiting for
+  admission became.
+
+The heatmaps retain the maximum observed age in each time bucket because the
+source is a sparse transition/summary stream. A darker cell means more samples
+fell into that time/age bucket; it does not mean more CPU was consumed. Empty
+periods mean Loki received no matching diagnostic sample, not necessarily that
+the scheduler was idle.
+
+## Grafana source-attributed sync cost
+
+The **DKG Nodes — Sync Cost** dashboard is the interactive Grafana companion
+to the W1 measurement contract introduced by PR #2033. It reads OpenTelemetry
+metrics from VictoriaMetrics and keeps the existing Loki pressure dashboard
+intact: logs answer what was present in a bounded pressure snapshot, while the
+metrics dashboard measures completed work, transferred payload, admission
+pressure, and catch-up outcomes over time.
+
+The dashboard covers every W1 instrument:
+
+| Instrument | Dashboard view |
+| --- | --- |
+| I1 physical sync attempts | Rate by source, outcome, transport, plane, and phase |
+| I2/I3 request and response payload bytes | Throughput by source and response outcome |
+| I4 logical operation duration | Source/lane/outcome flame graph, active-worker equivalents, and duration heatmap |
+| I5 rejected operations | Rate by source, lane, and bounded reason |
+| I6 single-flight joins | Rate by scope, owner source, and joining source |
+| I7 catch-up requests | Rate by route result and shared-memory request flag |
+| I8 catch-up jobs | Rate by terminal status and admission path |
+| I9 catch-up job duration | Walk-job p95 and duration heatmap |
+
+The flame-graph width is accumulated **active wall-clock occupancy** for
+completed logical sync operations in the selected Grafana range. It is not CPU
+time. Operations rejected before they start never receive a zero-duration I4
+sample; they appear in I5 instead.
+
+The two Prometheus heatmaps read real histogram buckets rather than estimating
+a distribution from log samples:
+
+- logical sync operation duration uses I4 and follows the selected
+  source/lane/outcome filters;
+- walk catch-up duration uses I9 and deliberately excludes synthetic
+  already-ready jobs, which perform no work.
+
+The dashboard also renders the W1 evidence gates and source-family shares over
+the selected Grafana range. Those interactive panels are for investigation;
+the generated `tools/observability/w1/w1-queries.md` packet remains the fixed
+1-hour/2-hour decision contract. Any failed evidence gate makes the window
+**inconclusive** rather than healthy, and the byte counters describe encoded
+application payload rather than network-wire bandwidth.
+
 ## Metrics
 
 The common OpenTelemetry instruments use bounded `scheduler` and `lane`

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);
+  };
+}

packages/cli/src/daemon/lifecycle.ts

@@ -151,6 +151,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';
@@ -1179,7 +1180,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, {

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"}',
+    );
+  });
+});

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',

tools/observability/RUNBOOK.md

@@ -15,6 +15,7 @@ import via `POST /api/dashboards/db {dashboard, folderUid, overwrite:true}`):
 | `dkg-fleet-logs` | `grafana-dashboard-dkg-fleet-logs.json` | logs (live) |
 | `dkg-node-logs` | `grafana-dashboard-dkg-node-logs.json` | logs (live) |
 | `dkg-node-metrics` | `grafana-dashboard-dkg-node-metrics.json` | node metrics endpoint + collector→VictoriaMetrics route (collector self-monitoring row is live already; the two raw-RPC panels additionally need nodes on a post-PR-#1409 build, which ships `dkg.chain.rpc.requests.total`) |
+| `dkg-sync-cost` | `grafana-dashboard-dkg-sync-cost.json` | node metrics endpoint + collector→VictoriaMetrics route + a node build containing PR #2033 W1 instruments I1–I9 |
 | `dkg-node-traces` | `grafana-dashboard-dkg-node-traces.json` | node traces endpoint + collector→Tempo route |
 
 Datasources are template variables (`loki` / `vm` / `tempo`) — the dashboards
@@ -34,6 +35,8 @@ channels): `example-alerts.md` (importable payloads: `alert-rules.provisioning.j
   split internally and fail with `maximum of series (500) reached` even at tiny
   stream counts. Use **range queries** + a Grafana reduce (`sum`/`last`) for
   totals; keep instant queries to short fixed windows (e.g. `[10m]`).
+  The worker-pressure flamegraph follows this rule with a fixed `[1h]`
+  snapshot; its adjacent heatmaps are the selected-range historical view.
 
 **To light up metrics + traces** (node side, per node): set
 `OTEL_EXPORTER_OTLP_ENDPOINT=http://<collector-host>:4318` (one env var

tools/observability/generate-observability.mjs

@@ -1,7 +1,7 @@
 #!/usr/bin/env node
 // Source of truth for the DKG observability Grafana artifacts.
 //
-// Emits the four dashboard JSONs, alert-rules.provisioning.json,
+// Emits the five dashboard JSONs, alert-rules.provisioning.json,
 // example-alerts.md and the W1 sync-measurement artifacts (w1/) in this
 // directory. Edit the lib/ sources, regenerate, commit both — never hand-edit
 // a rendered artifact.
@@ -53,7 +53,7 @@ assertPromLabel(PROM_NODE_LABEL, usage);
 const nodeProfile = promNodeProfile(PROM_NODE_LABEL);
 
 
-const { fleet, nodeLogs, metrics, traces } = buildDashboards({ nodeProfile });
+const { fleet, nodeLogs, metrics, syncCost, traces } = buildDashboards({ nodeProfile });
 const { alerts, specs, routes } = buildAlerts({ nodeProfile, VM_UID, LOKI_UID });
 const docs = buildDocs({ specs, routes });
 // W1 sync-measurement decision queries (lib/w1.mjs) render into a NESTED key
@@ -70,6 +70,7 @@ const rendered = new Map([
   ['grafana-dashboard-dkg-fleet-logs.json', JSON.stringify(fleet, null, 2) + '\n'],
   ['grafana-dashboard-dkg-node-logs.json', JSON.stringify(nodeLogs, null, 2) + '\n'],
   ['grafana-dashboard-dkg-node-metrics.json', JSON.stringify(metrics, null, 2) + '\n'],
+  ['grafana-dashboard-dkg-sync-cost.json', JSON.stringify(syncCost, null, 2) + '\n'],
   ['grafana-dashboard-dkg-node-traces.json', JSON.stringify(traces, null, 2) + '\n'],
   ['alert-rules.provisioning.json', JSON.stringify(alerts, null, 2) + '\n'],
   ...Object.entries(docs),