From 4115617eb9f6c4c54105915a8494e17cd0a5ee38 Mon Sep 17 00:00:00 2001 From: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com> Date: Sat, 25 Jul 2026 13:46:03 +0100 Subject: [PATCH] feat(telemetry): add job-queue occupancy and saturation gauges (WP-A4) Sync-critical job types run at very low concurrency limits (ledgerRequest and ledgerData allow 3 each), so a node can stall simply because those jobs are held back behind other work. Nothing exposed that until now: the existing job metrics are rates and quantiles of jobs that already moved, or a single queue-wide depth. - jobq_backlog{metric,job_type}: instantaneous waiting, running and deferred counts per job type. Deferred is the starvation signal and had no exposure anywhere; it is set when a type is at its concurrency limit. - jobq_saturation{metric}: running tasks, worker-thread count and total waiting, so a slowdown spanning several subsystems can be attributed to worker-pool exhaustion instead of being diagnosed once per victim. Both read through two new const accessors on JobQueue that take the existing mutex once and copy integers, so a single reading is internally consistent and no per-job cost is added. The job_type label reuses the same JobTypes name helper the existing job counters use, so the two label sets join. Co-Authored-By: Claude Opus 5 (1M context) --- .../09-data-collection-reference.md | 48 ++- .../dashboards/ledger-sync-health.json | 408 ++++++++++++++++++ .../telemetry/workload/expected_metrics.json | 10 +- .../telemetry/workload/validate_telemetry.py | 8 + docs/telemetry-glossary.md | 30 ++ docs/telemetry-runbook.md | 36 ++ include/xrpl/core/JobQueue.h | 105 +++++ src/libxrpl/core/detail/JobQueue.cpp | 43 ++ src/test/core/JobQueue_test.cpp | 134 ++++++ src/tests/libxrpl/telemetry/MetricMacros.cpp | 199 +++++++++ .../libxrpl/telemetry/MetricsRegistry.cpp | 28 +- src/xrpld/telemetry/MetricsRegistry.cpp | 97 +++++ src/xrpld/telemetry/MetricsRegistry.h | 76 ++++ 13 files changed, 1190 insertions(+), 32 deletions(-) diff --git a/OpenTelemetryPlan/09-data-collection-reference.md b/OpenTelemetryPlan/09-data-collection-reference.md index 1a82a21b03..813965ced7 100644 --- a/OpenTelemetryPlan/09-data-collection-reference.md +++ b/OpenTelemetryPlan/09-data-collection-reference.md @@ -1401,26 +1401,28 @@ signal as it lands. `Type` is the instrument kind (counter / gauge / histogram / span / span attr), `Emit site` the owning source file, and `Panel` the dashboard panel that renders it. -| Signal | Type | Emit site | Panel | Meaning | -| --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | -------------------------------------------------------------------- | ----------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `dns_resolve_total` (`outcome` = `resolved` \| `empty`) | counter | `OverlayImpl.cpp` — `OverlayImpl::reportDnsResolve` | DNS Resolve Outcome Rate | Peer hostname resolutions. `empty` means a configured bootstrap or `[ips_fixed]` name returned no address, so that peer is never dialled. | -| `dns_resolve_latency_ms` | histogram | `OverlayImpl.cpp` — `OverlayImpl::reportDnsResolve` | DNS Resolve Latency (p95) | Time to resolve a configured peer hostname. Seconds-scale values mean the resolver is timing out ahead of every dial. | -| `overlay_connect_total` (`outcome` = `connected` \| `tcp_fail` \| `tls_fail` \| `upgrade_fail` \| `timeout`) | counter | `ConnectAttempt.cpp` — `ConnectAttempt::reportOutcome` | Outbound Dial Outcome Rate | Outbound peer connection attempts by terminal outcome. The outcome names the stage that broke: TCP, TLS, HTTP upgrade, or no terminal state in time. | -| `overlay_dial_latency_ms` | histogram | `ConnectAttempt.cpp` — `ConnectAttempt::reportOutcome` | Outbound Dial Latency (p95) | Time from starting an outbound dial to its terminal outcome, successes and failures together. A p95 near the dial timeout means peers accept TCP but never finish the handshake. | -| `handshake_negotiation_fail_total` (`reason`, 14 values incl. `wrong_network`, `invalid_network_id`, `clock_skew`, `self_connection`, `session_verify_failed`) | counter | `Handshake.cpp` — `throwNegotiationFailure` (from `verifyHandshake`) | Handshake Negotiation Failures by Reason | Peer handshakes rejected after TLS while checking network id, clock, keys and addresses. `reason` names the failing check. | -| `unl_fetch_total` (`site` = configured UNL URI; `outcome` = the 9 `ListDisposition` strings `accepted` \| `expired` \| `same_sequence` \| `pending` \| `known_sequence` \| `unsupported_version` \| `untrusted` \| `stale` \| `invalid`, plus `fetch_error` \| `bad_status` \| `parse_error`) | counter | `ValidatorSite.cpp` — `ValidatorSite::reportFetchOutcome` | UNL Fetch Rate by Site & Outcome | Validator-list fetches per site. `accepted` is the only success; `same_sequence` and `known_sequence` are normal no-op refreshes; the three literals are transport or content faults. | -| `unl_quorum` (`metric` = `trusted_keys` \| `quorum`) | observable gauge | `MetricsRegistry.cpp` — `registerUnlQuorumGauge` | UNL Trusted Keys vs Quorum; UNL Quorum Headroom | Trusted UNL key count against the validations a ledger needs. `trusted_keys` at or below `quorum` means the node can never declare a ledger validated. | -| `clock_close_offset_seconds` (`metric` = `offset`) | observable gauge | `MetricsRegistry.cpp` — `registerClockSkewGauge` | Clock Close Offset | Network close time offset from the local clock. Negative means the local clock runs ahead. `server_info` only surfaces `close_time_offset` at 60 s or more, so this gauge sees skew far earlier. | -| `state_changes_total` (`from`, `to` = `disconnected` \| `connected` \| `syncing` \| `tracking` \| `full`) | counter | `NetworkOPs.cpp` — `NetworkOPsImp::setMode` | Mode Transitions by Edge | Operating-mode transitions keyed on the (`from`, `to`) edge. The edge is what separates a clean `disconnected`→`connected`→`syncing`→`tracking`→`full` climb from `full`→`connected` flapping; an unlabelled total cannot tell them apart. | -| `sync_state` (`metric` = `initial_full_duration_us`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Time to First FULL | Microseconds from process start to the first `full` transition, sourced from `NetworkOPs::getInitialSyncDurationUs()`. Stays 0 until `full` is reached, so a flat 0 is itself the "never synced" signal; once set it never changes. | -| `sync_state` (`metric` = `network_ledger_gate`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Network Ledger Gate | 1 while the node is still waiting to see a full network ledger (`NetworkOPs::isNeedNetworkLedger()`), else 0. A persistent 1 blocks transaction submission and `full`, whatever the rest of the pipeline shows. | -| `sync_state` (`metric` = `server_stall_seconds`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Server Stall | Current main-loop stall duration from `LoadManager::getCurrentStallSeconds()`, 0 when healthy. Same duration the load monitor logs as "Server stalled for N seconds", which previously existed only in that log line. | -| `sync_state` (`metric` = `ledgers_behind`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Ledgers Behind Network | Peer-reported network tip minus our validated sequence, floored at 0 (`NetworkOPs::getLedgersBehindNetwork()`). Reads each peer's already cached ledger range, so no new network round trip. | -| `server_stall_events_total` | observable counter | `MetricsRegistry.cpp` — `registerStallEventsCounter` | Server Stall Event Rate | Distinct stall episodes since process start, counted once per episode rather than per stalled second. A rising rate is repeated fresh stalls; a flat rate with a large `server_stall_seconds` is one long stall. | -| `sync_acquire` (`metric` = `missing_state_nodes_max` \| `missing_tx_nodes_max`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Missing SHAMap Nodes per Acquire (state/tx) | Largest outstanding SHAMap node count across in-flight acquires, split by tree, from the count `getMissingNodes()` already produces during its sweep (`InboundLedger.cpp` — `InboundLedger::trigger`). **The headline stuck-sync signal:** flat and non-zero across ticks means the acquire will never finish; shrinking means slow but alive. Aggregated as a max rather than labelled per ledger, because a `ledger_seq` label would mint one series per ledger acquired — per-ledger identity stays on the `ledger.acquire` span. | -| `sync_acquire` (`metric` = `received_data_depth`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Received-Data Stash Depth & In-Flight Acquires | Peer packets stashed across all in-flight acquires waiting to be applied, summed because it measures one shared processing backlog. A growing depth means arriving node data outpaces processing, so the limit is the job queue or disk rather than peer supply. | -| `sync_acquire` (`metric` = `in_flight`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Received-Data Stash Depth & In-Flight Acquires | Number of ledger acquires currently running. Exported so the three values above can be read in context: all zero with `in_flight` zero is an idle node, not a healthy one. | -| `shamap_cache_hit_rate` (`metric` = `treenode`) | observable gauge | `MetricsRegistry.cpp` — `registerCacheHitRateDetailGauge` | SHAMap TreeNode Cache Hit Rate | Share of SHAMap tree-node lookups served from memory, from the previously-uncalled `TaggedCache::getHitRate()`, normalized from 0-100 to 0.0-1.0. Distinct from `nodestore_state`-derived NuDB Cache Hit Ratio on the Ledger Data Sync dashboard: this is the in-memory layer **above** the node store, so a miss here is what causes a read there. The full-below cache is not reported — it is a `KeyCache` whose only lookup path increments `stats_.hits`/`stats_.misses` while `getHitRate()` reads the separate `hits_`/`misses_` members, so its rate is hard-wired to 0 until that accounting is fixed. | -| `sync_acquire_no_progress_total` | counter | `InboundLedger.cpp` — `InboundLedger::onTimer` | Acquire Stall Rate (no progress) | Acquire timeouts where not one new node arrived since the previous timeout, from the `progress_` flag that was previously log-only. Fires on the 3 s acquire timer, never per node. A sustained rate together with a flat missing-node count is the definitive "stuck, not slow" signature. | -| `sync_addnode_total` (`outcome` = `good` \| `duplicate` \| `invalid`) | counter | `InboundLedger.cpp` — `InboundLedger::recordBatchOutcome` | Add-Node Outcomes | SHAMap nodes received during acquire, split by result. Emitted once per received packet from the aggregated batch tally the trace log already printed — never inside the per-node `receiveNode()` loop. Separates real progress (`good`) from wasted bandwidth (`duplicate`) and a misbehaving peer (`invalid`), all three of which look like healthy throughput in traffic metrics. | -| `sync_acquire_source_total` (`source` = `local` \| `network`) | counter | `InboundLedger.cpp` — `InboundLedger::init` | Acquire Source (local vs network) | Whether an acquire was satisfied entirely from the local node store or needed peers, emitted once per new acquire after the first local lookup. Sustained `network` on a node that should already hold the range means sync is disk-bound rather than peer-bound. | +| Signal | Type | Emit site | Panel | Meaning | +| --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | -------------------------------------------------------------------- | --------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `dns_resolve_total` (`outcome` = `resolved` \| `empty`) | counter | `OverlayImpl.cpp` — `OverlayImpl::reportDnsResolve` | DNS Resolve Outcome Rate | Peer hostname resolutions. `empty` means a configured bootstrap or `[ips_fixed]` name returned no address, so that peer is never dialled. | +| `dns_resolve_latency_ms` | histogram | `OverlayImpl.cpp` — `OverlayImpl::reportDnsResolve` | DNS Resolve Latency (p95) | Time to resolve a configured peer hostname. Seconds-scale values mean the resolver is timing out ahead of every dial. | +| `overlay_connect_total` (`outcome` = `connected` \| `tcp_fail` \| `tls_fail` \| `upgrade_fail` \| `timeout`) | counter | `ConnectAttempt.cpp` — `ConnectAttempt::reportOutcome` | Outbound Dial Outcome Rate | Outbound peer connection attempts by terminal outcome. The outcome names the stage that broke: TCP, TLS, HTTP upgrade, or no terminal state in time. | +| `overlay_dial_latency_ms` | histogram | `ConnectAttempt.cpp` — `ConnectAttempt::reportOutcome` | Outbound Dial Latency (p95) | Time from starting an outbound dial to its terminal outcome, successes and failures together. A p95 near the dial timeout means peers accept TCP but never finish the handshake. | +| `handshake_negotiation_fail_total` (`reason`, 14 values incl. `wrong_network`, `invalid_network_id`, `clock_skew`, `self_connection`, `session_verify_failed`) | counter | `Handshake.cpp` — `throwNegotiationFailure` (from `verifyHandshake`) | Handshake Negotiation Failures by Reason | Peer handshakes rejected after TLS while checking network id, clock, keys and addresses. `reason` names the failing check. | +| `unl_fetch_total` (`site` = configured UNL URI; `outcome` = the 9 `ListDisposition` strings `accepted` \| `expired` \| `same_sequence` \| `pending` \| `known_sequence` \| `unsupported_version` \| `untrusted` \| `stale` \| `invalid`, plus `fetch_error` \| `bad_status` \| `parse_error`) | counter | `ValidatorSite.cpp` — `ValidatorSite::reportFetchOutcome` | UNL Fetch Rate by Site & Outcome | Validator-list fetches per site. `accepted` is the only success; `same_sequence` and `known_sequence` are normal no-op refreshes; the three literals are transport or content faults. | +| `unl_quorum` (`metric` = `trusted_keys` \| `quorum`) | observable gauge | `MetricsRegistry.cpp` — `registerUnlQuorumGauge` | UNL Trusted Keys vs Quorum; UNL Quorum Headroom | Trusted UNL key count against the validations a ledger needs. `trusted_keys` at or below `quorum` means the node can never declare a ledger validated. | +| `clock_close_offset_seconds` (`metric` = `offset`) | observable gauge | `MetricsRegistry.cpp` — `registerClockSkewGauge` | Clock Close Offset | Network close time offset from the local clock. Negative means the local clock runs ahead. `server_info` only surfaces `close_time_offset` at 60 s or more, so this gauge sees skew far earlier. | +| `state_changes_total` (`from`, `to` = `disconnected` \| `connected` \| `syncing` \| `tracking` \| `full`) | counter | `NetworkOPs.cpp` — `NetworkOPsImp::setMode` | Mode Transitions by Edge | Operating-mode transitions keyed on the (`from`, `to`) edge. The edge is what separates a clean `disconnected`→`connected`→`syncing`→`tracking`→`full` climb from `full`→`connected` flapping; an unlabelled total cannot tell them apart. | +| `sync_state` (`metric` = `initial_full_duration_us`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Time to First FULL | Microseconds from process start to the first `full` transition, sourced from `NetworkOPs::getInitialSyncDurationUs()`. Stays 0 until `full` is reached, so a flat 0 is itself the "never synced" signal; once set it never changes. | +| `sync_state` (`metric` = `network_ledger_gate`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Network Ledger Gate | 1 while the node is still waiting to see a full network ledger (`NetworkOPs::isNeedNetworkLedger()`), else 0. A persistent 1 blocks transaction submission and `full`, whatever the rest of the pipeline shows. | +| `sync_state` (`metric` = `server_stall_seconds`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Server Stall | Current main-loop stall duration from `LoadManager::getCurrentStallSeconds()`, 0 when healthy. Same duration the load monitor logs as "Server stalled for N seconds", which previously existed only in that log line. | +| `sync_state` (`metric` = `ledgers_behind`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncStateGauge` | Ledgers Behind Network | Peer-reported network tip minus our validated sequence, floored at 0 (`NetworkOPs::getLedgersBehindNetwork()`). Reads each peer's already cached ledger range, so no new network round trip. | +| `server_stall_events_total` | observable counter | `MetricsRegistry.cpp` — `registerStallEventsCounter` | Server Stall Event Rate | Distinct stall episodes since process start, counted once per episode rather than per stalled second. A rising rate is repeated fresh stalls; a flat rate with a large `server_stall_seconds` is one long stall. | +| `sync_acquire` (`metric` = `missing_state_nodes_max` \| `missing_tx_nodes_max`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Missing SHAMap Nodes per Acquire (state/tx) | Largest outstanding SHAMap node count across in-flight acquires, split by tree, from the count `getMissingNodes()` already produces during its sweep (`InboundLedger.cpp` — `InboundLedger::trigger`). **The headline stuck-sync signal:** flat and non-zero across ticks means the acquire will never finish; shrinking means slow but alive. Aggregated as a max rather than labelled per ledger, because a `ledger_seq` label would mint one series per ledger acquired — per-ledger identity stays on the `ledger.acquire` span. | +| `sync_acquire` (`metric` = `received_data_depth`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Received-Data Stash Depth & In-Flight Acquires | Peer packets stashed across all in-flight acquires waiting to be applied, summed because it measures one shared processing backlog. A growing depth means arriving node data outpaces processing, so the limit is the job queue or disk rather than peer supply. | +| `sync_acquire` (`metric` = `in_flight`) | observable gauge | `MetricsRegistry.cpp` — `registerSyncAcquireGauge` | Received-Data Stash Depth & In-Flight Acquires | Number of ledger acquires currently running. Exported so the three values above can be read in context: all zero with `in_flight` zero is an idle node, not a healthy one. | +| `shamap_cache_hit_rate` (`metric` = `treenode`) | observable gauge | `MetricsRegistry.cpp` — `registerCacheHitRateDetailGauge` | SHAMap TreeNode Cache Hit Rate | Share of SHAMap tree-node lookups served from memory, from the previously-uncalled `TaggedCache::getHitRate()`, normalized from 0-100 to 0.0-1.0. Distinct from `nodestore_state`-derived NuDB Cache Hit Ratio on the Ledger Data Sync dashboard: this is the in-memory layer **above** the node store, so a miss here is what causes a read there. The full-below cache is not reported — it is a `KeyCache` whose only lookup path increments `stats_.hits`/`stats_.misses` while `getHitRate()` reads the separate `hits_`/`misses_` members, so its rate is hard-wired to 0 until that accounting is fixed. | +| `sync_acquire_no_progress_total` | counter | `InboundLedger.cpp` — `InboundLedger::onTimer` | Acquire Stall Rate (no progress) | Acquire timeouts where not one new node arrived since the previous timeout, from the `progress_` flag that was previously log-only. Fires on the 3 s acquire timer, never per node. A sustained rate together with a flat missing-node count is the definitive "stuck, not slow" signature. | +| `sync_addnode_total` (`outcome` = `good` \| `duplicate` \| `invalid`) | counter | `InboundLedger.cpp` — `InboundLedger::recordBatchOutcome` | Add-Node Outcomes | SHAMap nodes received during acquire, split by result. Emitted once per received packet from the aggregated batch tally the trace log already printed — never inside the per-node `receiveNode()` loop. Separates real progress (`good`) from wasted bandwidth (`duplicate`) and a misbehaving peer (`invalid`), all three of which look like healthy throughput in traffic metrics. | +| `sync_acquire_source_total` (`source` = `local` \| `network`) | counter | `InboundLedger.cpp` — `InboundLedger::init` | Acquire Source (local vs network) | Whether an acquire was satisfied entirely from the local node store or needed peers, emitted once per new acquire after the first local lookup. Sustained `network` on a node that should already hold the range means sync is disk-bound rather than peer-bound. | +| `jobq_backlog` (`metric` = `waiting` \| `running` \| `deferred`; `job_type` = the `JobTypes::name()` string) | observable gauge | `MetricsRegistry.cpp` — `registerJobQueueBacklogGauge` | Deferred Jobs by Type (starvation); Job Queue Occupancy by Type (waiting/running) | Instantaneous per-job-type queue occupancy, from `JobQueue::getJobTypeCounts()` (one mutex acquire per ~10 s tick). **`deferred` is the signal this adds:** jobs the queue accepted but withheld because the type is already at its concurrency limit, which is counted in neither `waiting` nor `running` and had no exposure anywhere before. The sync-critical types are capped at 3 (`JtLedgerReq`, `JtLedgerData` in `JobTypes.h`), so they starve first. Distinct from the existing `job_queued_total` / `job_started_total` / `job_finished_total` counters and `job_queued_us` / `job_running_us` histograms, which are event-driven from PerfLogImp and describe jobs that already moved, and from the StatsD `jobq_job_count`, which is queue-wide with no per-type split. Cardinality is bounded by the JobType enum (~46 values); every type is observed every tick, so an idle type reports 0 rather than dropping its series. | +| `jobq_saturation` (`metric` = `running_tasks` \| `worker_threads` \| `total_waiting`) | observable gauge | `MetricsRegistry.cpp` — `registerJobQueueSaturationGauge` | Worker Pool Saturation; Worker Pool Capacity & Total Backlog | Global worker-pool saturation from `JobQueue::getWorkerSaturation()`: tasks in flight, threads the pool is configured to run, and jobs queued across all types, all from one reading so the ratio and the backlog describe the same instant. `worker_threads` is exported rather than hardcoded in the dashboard because it is derived at startup from `[workers]`, node size and hardware concurrency. Exists separately from `jobq_backlog` because a pool-wide slowdown otherwise appears as an independent fault in every subsystem queued behind it; a `running_tasks / worker_threads` ratio at 1.0 **with** a non-zero `total_waiting` attributes it to pool exhaustion once. | diff --git a/docker/telemetry/grafana/dashboards/ledger-sync-health.json b/docker/telemetry/grafana/dashboards/ledger-sync-health.json index 2dd8cbb880..16d023db5e 100644 --- a/docker/telemetry/grafana/dashboards/ledger-sync-health.json +++ b/docker/telemetry/grafana/dashboards/ledger-sync-health.json @@ -2059,6 +2059,374 @@ ], "title": "Received-Data Stash Depth & In-Flight Acquires", "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "###### What this is:\n*Jobs held back because their job type is already running at its concurrency limit. A held-back job is not \"waiting\" and not \"running\" \u2014 it exists and is being denied a worker thread, which is starvation rather than idleness.*\n\n###### How it's computed:\n*jobq_backlog series deferred, per job_type, read from the JobQueue's own per-type deferred counter on each collection tick.*\n\n###### Reading it:\n*0 everywhere is healthy. Any sustained non-zero value names the job type whose concurrency limit is the bottleneck.*\n\n###### Healthy range:\n*0.*\n\n###### Watch for:\n*ledgerData or ledgerRequest deferred above zero during a fresh sync. Both run at a limit of 3, so they are the first types to starve, and this is the only place that state is visible \u2014 the job counters and queue-wait histograms cannot show it.*\n\n###### Keywords:\n- **Deferred job** *(per node)* \u2014 a job the queue accepted but withheld from a worker because its type is at its concurrency limit.\n\n###### Computation boundary:\n*Result: Per node \u2014 each series is one server's own value.*\n*Computed in xrpld code (MetricsRegistry, OpenTelemetry SDK) and exported as a metric; the collector only forwards it; the Grafana query selects and aggregates it.*\n\n###### Source:\n[MetricsRegistry.cpp](https://github.com/XRPLF/rippled/blob/develop/src/xrpld/telemetry/MetricsRegistry.cpp)\n\n###### Function:\n`registerJobQueueBacklogGauge`\n\n###### References:\n[Telemetry glossary](https://github.com/XRPLF/rippled/blob/develop/docs/telemetry-glossary.md#deferred-job)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Jobs", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 30, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 3, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": 1800000, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "line" + } + }, + "displayName": "${__field.labels.series} ${__field.labels.xrpl_ident}", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1 + } + ] + }, + "unit": "short" + } + }, + "gridPos": { + "h": 12, + "w": 12, + "x": 0, + "y": 134 + }, + "id": 24, + "options": { + "annotations": { + "clustering": -1, + "multiLane": false + }, + "legend": { + "calcs": [], + "displayMode": "list", + "enableFacetedFilter": false, + "overflow": "ellipsis", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "maxHeight": 600, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "13.2.0-28926505616", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "label_replace(label_join(label_replace(jobq_backlog{service_instance_id=~\"$node\", deployment_environment=~\"$deployment_environment\", xrpl_network_type=~\"$xrpl_network_type\", service_name=~\"$service_name\", xrpl_work_item=~\"$xrpl_work_item\", xrpl_branch=~\"$xrpl_branch\", xrpl_node_role=~\"$xrpl_node_role\", metric=\"deferred\", job_type=~\"$job_type\"}, \"series\", \"$1 deferred\", \"job_type\", \"(.*)\"), \"xrpl_ident\", \", \", \"service_instance_id\", \"xrpl_branch\", \"xrpl_work_item\"), \"xrpl_ident\", \"[$1]\", \"xrpl_ident\", \"(?:, )*(.*[^, ])(?:, )*\")", + "refId": "A" + } + ], + "title": "Deferred Jobs by Type (starvation)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "###### What this is:\n*Instantaneous per-job-type queue occupancy: how many jobs of each type are queued (waiting) and how many are executing (running) right now.*\n\n###### How it's computed:\n*jobq_backlog series waiting and running, per job_type, sampled from the JobQueue under one lock acquire so the two agree on the same instant.*\n\n###### Reading it:\n*running at a type's concurrency limit with waiting above zero means that type is the constraint. Use the Job Type variable to isolate one type.*\n\n###### Healthy range:\n*waiting near 0; running low single digits per type.*\n\n###### Watch for:\n*A waiting count that climbs while running sits flat at the limit \u2014 pair with the Deferred Jobs panel, which shows how much of that backlog the limit is actively withholding. Distinct from Job Queue Wait p95: that measures how long jobs already waited, this measures how many are waiting now.*\n\n###### Keywords:\n- **Job queue occupancy** *(per node)* \u2014 the number of jobs of a type queued or executing at the moment of sampling.\n\n###### Computation boundary:\n*Result: Per node \u2014 each series is one server's own value.*\n*Computed in xrpld code (MetricsRegistry, OpenTelemetry SDK) and exported as a metric; the collector only forwards it; the Grafana query selects and aggregates it.*\n\n###### Source:\n[MetricsRegistry.cpp](https://github.com/XRPLF/rippled/blob/develop/src/xrpld/telemetry/MetricsRegistry.cpp)\n\n###### Function:\n`registerJobQueueBacklogGauge`\n\n###### References:\n[Telemetry glossary](https://github.com/XRPLF/rippled/blob/develop/docs/telemetry-glossary.md#job-queue-occupancy)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Jobs", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 3, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": 1800000, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "displayName": "${__field.labels.series} ${__field.labels.xrpl_ident}", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "short" + } + }, + "gridPos": { + "h": 12, + "w": 12, + "x": 12, + "y": 134 + }, + "id": 25, + "options": { + "annotations": { + "clustering": -1, + "multiLane": false + }, + "legend": { + "calcs": [], + "displayMode": "list", + "enableFacetedFilter": false, + "overflow": "ellipsis", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "maxHeight": 600, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "13.2.0-28926505616", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "label_replace(label_join(label_replace(jobq_backlog{service_instance_id=~\"$node\", deployment_environment=~\"$deployment_environment\", xrpl_network_type=~\"$xrpl_network_type\", service_name=~\"$service_name\", xrpl_work_item=~\"$xrpl_work_item\", xrpl_branch=~\"$xrpl_branch\", xrpl_node_role=~\"$xrpl_node_role\", metric=~\"waiting|running\", job_type=~\"$job_type\"}, \"series\", \"$1\", \"job_type\", \"(.*)\"), \"xrpl_ident\", \", \", \"service_instance_id\", \"xrpl_branch\", \"xrpl_work_item\"), \"xrpl_ident\", \"[$1]\", \"xrpl_ident\", \"(?:, )*(.*[^, ])(?:, )*\")", + "refId": "A" + } + ], + "title": "Job Queue Occupancy by Type (waiting/running)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "###### What this is:\n*Share of the worker-thread pool currently executing a job. This is the pool-wide view: when the pool itself is exhausted, every subsystem queued behind it looks independently slow, and this panel attributes the whole slowdown once.*\n\n###### How it's computed:\n*jobq_saturation running_tasks divided by worker_threads (the denominator is clamped to at least 1). The thread count is exported rather than hardcoded because it is derived at startup from [workers], node size and hardware concurrency.*\n\n###### Reading it:\n*Below 80% (green) means the pool has spare capacity, so a slow stage is that stage's own fault. At 100% every worker is busy \u2014 read Total Jobs Queued next: 100% with a queue is an exhausted pool, 100% with an empty queue is merely busy.*\n\n###### Healthy range:\n*< 80%.*\n\n###### Watch for:\n*A sustained 100% together with a non-zero backlog. Every job type is then starved by the pool, so fix pool capacity or the long-running jobs holding it, not the individual victim subsystems.*\n\n###### Keywords:\n- **Worker-pool saturation** *(per node)* \u2014 worker threads executing a job as a share of the threads the pool is configured to run.\n\n###### Computation boundary:\n*Result: Per node \u2014 each series is one server's own value.*\n*Computed in xrpld code (MetricsRegistry, OpenTelemetry SDK) and exported as a metric; the collector only forwards it; the Grafana query selects and aggregates it.*\n\n###### Source:\n[MetricsRegistry.cpp](https://github.com/XRPLF/rippled/blob/develop/src/xrpld/telemetry/MetricsRegistry.cpp)\n\n###### Function:\n`registerJobQueueSaturationGauge`\n\n###### References:\n[Telemetry glossary](https://github.com/XRPLF/rippled/blob/develop/docs/telemetry-glossary.md#worker-pool-saturation)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 0.8 + }, + { + "color": "red", + "value": 1 + } + ] + }, + "unit": "percentunit" + } + }, + "gridPos": { + "h": 12, + "w": 12, + "x": 0, + "y": 146 + }, + "id": 26, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "center", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "value_and_name", + "wideLayout": true + }, + "pluginVersion": "13.2.0-28926505616", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "label_replace(label_join(label_replace(jobq_saturation{service_instance_id=~\"$node\", deployment_environment=~\"$deployment_environment\", xrpl_network_type=~\"$xrpl_network_type\", service_name=~\"$service_name\", xrpl_work_item=~\"$xrpl_work_item\", xrpl_branch=~\"$xrpl_branch\", xrpl_node_role=~\"$xrpl_node_role\", metric=\"running_tasks\"} / ignoring(metric) clamp_min(jobq_saturation{service_instance_id=~\"$node\", deployment_environment=~\"$deployment_environment\", xrpl_network_type=~\"$xrpl_network_type\", service_name=~\"$service_name\", xrpl_work_item=~\"$xrpl_work_item\", xrpl_branch=~\"$xrpl_branch\", xrpl_node_role=~\"$xrpl_node_role\", metric=\"worker_threads\"}, 1), \"series\", \"Worker saturation\", \"\", \"\"), \"xrpl_ident\", \", \", \"service_instance_id\", \"xrpl_branch\", \"xrpl_work_item\"), \"xrpl_ident\", \"[$1]\", \"xrpl_ident\", \"(?:, )*(.*[^, ])(?:, )*\")", + "refId": "A" + } + ], + "title": "Worker Pool Saturation", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "###### What this is:\n*The three raw worker-pool numbers behind the saturation ratio: tasks in flight, threads configured, and total jobs queued across every job type.*\n\n###### How it's computed:\n*jobq_saturation series running_tasks, worker_threads and total_waiting, all read from one JobQueue sample so the ratio and the backlog describe the same instant.*\n\n###### Reading it:\n*running_tasks tracking worker_threads means the pool is fully committed. total_waiting is what makes that legible: queued work behind a fully committed pool is exhaustion, no queued work is just a busy moment.*\n\n###### Healthy range:\n*running_tasks below worker_threads; total_waiting near 0.*\n\n###### Watch for:\n*total_waiting climbing while running_tasks is pinned at worker_threads. Distinct from the jobq_job_count depth panel on the Ledger Data Sync dashboard, which has no capacity term at all, so a depth reading there cannot say whether the pool is the cause.*\n\n###### Keywords:\n- **Worker thread pool** *(per node)* \u2014 the fixed set of threads that execute all job-queue work, sized at startup.\n\n###### Computation boundary:\n*Result: Per node \u2014 each series is one server's own value.*\n*Computed in xrpld code (MetricsRegistry, OpenTelemetry SDK) and exported as a metric; the collector only forwards it; the Grafana query selects and aggregates it.*\n\n###### Source:\n[MetricsRegistry.cpp](https://github.com/XRPLF/rippled/blob/develop/src/xrpld/telemetry/MetricsRegistry.cpp)\n\n###### Function:\n`registerJobQueueSaturationGauge`\n\n###### References:\n[Telemetry glossary](https://github.com/XRPLF/rippled/blob/develop/docs/telemetry-glossary.md#worker-pool-saturation)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Count", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 3, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": 1800000, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "displayName": "${__field.labels.series} ${__field.labels.xrpl_ident}", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "short" + } + }, + "gridPos": { + "h": 12, + "w": 12, + "x": 12, + "y": 146 + }, + "id": 27, + "options": { + "annotations": { + "clustering": -1, + "multiLane": false + }, + "legend": { + "calcs": [], + "displayMode": "list", + "enableFacetedFilter": false, + "overflow": "ellipsis", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "maxHeight": 600, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "13.2.0-28926505616", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "label_replace(label_join(label_replace(jobq_saturation{service_instance_id=~\"$node\", deployment_environment=~\"$deployment_environment\", xrpl_network_type=~\"$xrpl_network_type\", service_name=~\"$service_name\", xrpl_work_item=~\"$xrpl_work_item\", xrpl_branch=~\"$xrpl_branch\", xrpl_node_role=~\"$xrpl_node_role\", metric=~\"$saturation_metric\"}, \"series\", \"$1\", \"metric\", \"(.*)\"), \"xrpl_ident\", \", \", \"service_instance_id\", \"xrpl_branch\", \"xrpl_work_item\"), \"xrpl_ident\", \"[$1]\", \"xrpl_ident\", \"(?:, )*(.*[^, ])(?:, )*\")", + "refId": "A" + } + ], + "title": "Worker Pool Capacity & Total Backlog", + "type": "timeseries" } ], "schemaVersion": 39, @@ -2397,6 +2765,46 @@ "multi": true, "refresh": 2, "sort": 1 + }, + { + "name": "job_type", + "label": "Job Type", + "description": "Filter the job-queue backlog by JobType name [ledgerData / ledgerRequest / ...]", + "type": "query", + "query": "label_values(jobq_backlog, job_type)", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "includeAll": true, + "allValue": ".*", + "current": { + "text": "All", + "value": "$__all" + }, + "multi": true, + "refresh": 2, + "sort": 1 + }, + { + "name": "saturation_metric", + "label": "Saturation Metric", + "description": "Filter the worker-pool gauge sub-series [running_tasks / worker_threads / total_waiting]", + "type": "query", + "query": "label_values(jobq_saturation, metric)", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "includeAll": true, + "allValue": ".*", + "current": { + "text": "All", + "value": "$__all" + }, + "multi": true, + "refresh": 2, + "sort": 1 } ] }, diff --git a/docker/telemetry/workload/expected_metrics.json b/docker/telemetry/workload/expected_metrics.json index d47f8affd6..476d6c1400 100644 --- a/docker/telemetry/workload/expected_metrics.json +++ b/docker/telemetry/workload/expected_metrics.json @@ -149,9 +149,17 @@ "sync_acquire{metric=\"missing_tx_nodes_max\"}", "sync_acquire{metric=\"received_data_depth\"}", "sync_acquire{metric=\"in_flight\"}", - "shamap_cache_hit_rate{metric=\"treenode\"}" + "shamap_cache_hit_rate{metric=\"treenode\"}", + "jobq_backlog{metric=\"waiting\",job_type=\"ledgerData\"}", + "jobq_backlog{metric=\"running\",job_type=\"ledgerData\"}", + "jobq_backlog{metric=\"deferred\",job_type=\"ledgerData\"}", + "jobq_backlog{metric=\"deferred\",job_type=\"ledgerRequest\"}", + "jobq_saturation{metric=\"running_tasks\"}", + "jobq_saturation{metric=\"worker_threads\"}", + "jobq_saturation{metric=\"total_waiting\"}" ], "_acquire_note": "The four sync_acquire sub-series and shamap_cache_hit_rate are unconditional: both are observable gauges whose callbacks observe every series on each collection tick, so each is present even when the value is 0 (an idle node reports in_flight=0 and missing_state_nodes_max=0, and a cold cache reports a 0.0 hit rate). Absence, not a zero, is the regression. The three WP-A3 counters (sync_acquire_source_total, sync_addnode_total, sync_acquire_no_progress_total) are deliberately NOT asserted here: all three are emitted only from InboundLedger, which runs only when a node must fetch a ledger it lacks. expected_spans.json already marks the ledger.acquire span optional for exactly this reason (\"A healthy local cluster rarely back-fills history\"), and the metric validator has no per-metric optional flag, so listing them would fail the harness red on a healthy run. They are covered by exact-value unit tests in src/tests/libxrpl/telemetry/MetricMacros.cpp and by the ledger-sync-health panels; add them here only alongside a harness step that forces a real acquire (e.g. starting a node against an existing ledger history).", + "_jobq_note": "The jobq_backlog and jobq_saturation series are unconditional: both are observable gauges whose callbacks iterate EVERY registered JobType (jobData_ is populated from JobTypes at JobQueue construction) and observe all three fields on each collection tick, so a series exists even when the value is 0. That is why an idle-but-registered type like ledgerData is safe to assert by name here — a fresh harness node that never defers a single job still reports jobq_backlog{metric=\"deferred\",job_type=\"ledgerData\"} = 0, and absence, not the zero, is the regression. Two job_type values are asserted (ledgerData and ledgerRequest) because they are the sync-critical types capped at concurrency 3 in JobTypes.h, so they are the ones whose deferred series must never silently vanish. Only deferred is asserted for ledgerRequest to keep the list short: the three-field fan-out is already proven by ledgerData. worker_threads is asserted because it is the denominator of the dashboard saturation ratio, and it is always at least 1 (the JobQueue ctor gives standalone mode exactly one worker), so a zero or missing reading there means the accessor regressed rather than the node being idle.", "_conditional_note": "handshake_negotiation_fail_total and unl_fetch_total are conditional under the local harness: the first only exists once a handshake is rejected, and the second needs a [validator_list_sites] entry (run-full-validation.sh generates a static [validators] file instead). The validator has no per-metric optional flag, so if either reports 0 series in a harness run, move it out of this group rather than weakening the check.", "_sync_state_note": "The four sync_state sub-series are unconditional: the gauge observes all four on every collection tick, so each is present as a series even when its value is 0 (a node that never reached FULL reports initial_full_duration_us=0, and a healthy node reports server_stall_seconds=0). The check asserts series presence, not a non-zero value, which is exactly right here — a zero is a meaningful reading for these signals, and absence is the regression. server_stall_events_total is likewise always present because the observable counter reports the tally (0 or more) every tick. state_changes_total is asserted here with a from!=\"\",to!=\"\" selector rather than bare (parity_counters already asserts the bare name): the selector is what proves the WP-A2 {from,to} label dimension actually reached Prometheus, so a regression to the old unlabelled counter fails this check instead of silently passing on the bare name. It needs at least one real mode transition, which any node reaching connected/syncing produces during startup." }, diff --git a/docker/telemetry/workload/validate_telemetry.py b/docker/telemetry/workload/validate_telemetry.py index 3cc2d72604..b44c22853c 100644 --- a/docker/telemetry/workload/validate_telemetry.py +++ b/docker/telemetry/workload/validate_telemetry.py @@ -608,6 +608,14 @@ async def validate_metrics( "sync_acquire", "sync_addnode_total", "shamap_cache_hit_rate", + # JobQueue saturation signals. jobq_backlog carries the + # per-type waiting/running/deferred sub-series and + # jobq_saturation the pool running_tasks/worker_threads/ + # total_waiting; both are asserted. No new prefix entry + # is needed -- the "jobq_" prefix above already matches + # them (it was added for the StatsD jobq_job_count + # gauge), so listing them again would only duplicate + # the diagnostic output. ) ) ] diff --git a/docs/telemetry-glossary.md b/docs/telemetry-glossary.md index c5af73fc51..4693723756 100644 --- a/docs/telemetry-glossary.md +++ b/docs/telemetry-glossary.md @@ -519,6 +519,16 @@ Turning each configured peer hostname into IP addresses, which happens before an **See also:** [Outbound dial latency](#outbound-dial-latency) · [Peer protocol on xrpl.org](https://xrpl.org/docs/concepts/networks-and-servers/peer-protocol) + + +### Deferred job + +A job the queue accepted but withheld from a worker thread because its job type is already running at that type's concurrency limit. This is a third state alongside waiting and running, and it is counted in neither: the work exists and is being actively denied a thread, which is starvation rather than idleness or overload. The distinction matters because the sync-critical types run at very small limits — ledger requests and inbound ledger data are each capped at three concurrent jobs — so during a fresh sync those types routinely have work withheld while the queue looks shallow and the queue-wait quantiles look unremarkable. A sustained non-zero deferred count names the job type whose limit is the bottleneck. + +**Scope:** per node — measured on and specific to this individual server. + +**See also:** [Job queue occupancy](#job-queue-occupancy) · [Worker-pool saturation](#worker-pool-saturation) + ### Acquire source @@ -577,6 +587,16 @@ The rate at which the node fetches older ledgers to extend or repair its stored **Scope:** per node — measured on and specific to this individual server. + + +### Job queue occupancy + +How many jobs of a given type are queued or executing at the instant the queue is sampled, as opposed to how many passed through it over a period. Occupancy answers "what is sitting there now"; the job counters and queue-wait quantiles answer "what already moved and how long it had waited". The two can disagree in the way that matters most: a type whose jobs are all still queued produces no completed-job samples at all, so a latency quantile can look healthy precisely because nothing is finishing. + +**Scope:** per node — measured on and specific to this individual server. + +**See also:** [Deferred job](#deferred-job) · [Worker-pool saturation](#worker-pool-saturation) + ### Ledger acquire (inbound fetch) @@ -695,6 +715,16 @@ The trusted UNL key count minus the quorum a ledger needs, so it reads as spare **See also:** [UNL fetch outcome](#unl-fetch-outcome) · [Validation quorum](#validation-quorum) · [Validation quorum on xrpl.org](https://xrpl.org/docs/concepts/consensus-protocol/negative-unl) + + +### Worker-pool saturation + +The share of the job-queue worker threads currently executing a job. The thread count is fixed at startup from configuration, node size and hardware concurrency, so it is a real ceiling rather than an elastic one. Saturation is read together with the total number of jobs queued, because the ratio alone is ambiguous: every thread busy with nothing queued is a busy instant, while every thread busy with work piling up behind them is an exhausted pool. The distinction is what makes this a pool-level signal — when the pool is exhausted, every subsystem whose jobs are queued behind it slows at the same time, so each one appears to have its own independent fault. Reading saturation first attributes that whole pattern once, instead of once per victim. + +**Scope:** per node — measured on and specific to this individual server. + +**See also:** [Deferred job](#deferred-job) · [Job queue occupancy](#job-queue-occupancy) · [Server stall](#server-stall) + ## Peer & Overlay Networking diff --git a/docs/telemetry-runbook.md b/docs/telemetry-runbook.md index a909538807..1132ef03ee 100644 --- a/docs/telemetry-runbook.md +++ b/docs/telemetry-runbook.md @@ -2274,6 +2274,42 @@ each step gates the next: stop at the first one that is wrong. This is also the pairing that explains a large existing database syncing slower than a fresh one. +10. **Is the work arriving but never getting a worker thread?** + Steps 6 to 9 all assume the node is at least trying to process what it + receives. This step covers the case where it is not: the data arrived, the + job was queued, and no thread ever ran it. Read the two levels in order, + because the pool-wide answer supersedes the per-type one. + - **Pool first** — _Worker Pool Saturation_ (`jobq_saturation`, + `running_tasks / worker_threads`) with _Worker Pool Capacity & Total + Backlog_ beside it. Below 80% the pool has spare capacity, so skip to the + per-type read. At 100% the answer depends on `total_waiting`: + - 100% with `total_waiting` near zero — the pool is merely busy at this + instant. Not a fault. + - 100% with `total_waiting` climbing — the pool is **exhausted**. Every + job type is starved by it, so every other stage in this row will look + slow at once. Stop here: fixing an individual subsystem cannot help + while no thread is free to run its jobs. Look at what is holding the + threads (long-running jobs, disk waits from step 9) or at the + `[workers]` setting for the node size. + - **Then per type** — _Deferred Jobs by Type (starvation)_ + (`jobq_backlog`, `metric=deferred`). This is the signal that exists + nowhere else. A deferred job is one the queue **accepted and then + withheld** because its type is already running at its concurrency limit, + so it appears in neither `waiting` nor `running`, and neither the job + counters nor the queue-wait histograms can show it. Any sustained + non-zero value names the job type whose limit is the bottleneck. During a + fresh sync watch `job_type=ledgerData` and `job_type=ledgerRequest` + first: both run at a limit of 3, so they are the types that starve + soonest, and starved `ledgerData` is exactly why the received-data stash + in step 8 grows while the missing-node count in step 6 stays flat. + _Job Queue Occupancy by Type (waiting/running)_ gives the context — + `running` pinned at the limit with `waiting` above zero confirms the + limit, not the supply, is what is holding the type back. + Note the difference from _Job Queue Wait p95 By Type_ on the Ledger Data + Sync dashboard: that panel measures how long jobs that already ran had + waited, so it reports the past; these gauges report what is sitting in + the queue right now, including the part being actively withheld. + ## Performance Tuning | Scenario | Recommendation | diff --git a/include/xrpl/core/JobQueue.h b/include/xrpl/core/JobQueue.h index 0c9fc76357..8f35775878 100644 --- a/include/xrpl/core/JobQueue.h +++ b/include/xrpl/core/JobQueue.h @@ -32,6 +32,7 @@ #include #include #include +#include namespace xrpl { @@ -215,6 +216,110 @@ public: int getJobCountGE(JobType t) const; + /** + * Occupancy snapshot for a single JobType. + * + * A plain value type so it can cross the libxrpl/xrpld boundary: xrpld + * telemetry observes queue occupancy without libxrpl gaining any + * dependency on the telemetry code. + * + * `deferred` is the field with no other exposure anywhere. The + * sync-critical types run at tiny concurrency limits (`JtLedgerReq` and + * `JtLedgerData` are capped at 3 in JobTypes.h), so a job of those types + * is commonly held back rather than merely queued, and a held-back job is + * invisible in `waiting` and `running` alike. + */ + struct JobTypeCount + { + /** + * The job type these counts describe. The caller turns this into a + * label via JobTypes::name(), so the name is not duplicated here. + */ + JobType type{JtInvalid}; + + /** + * Jobs enqueued and not yet dispatched to a worker thread. + */ + int waiting{0}; + + /** + * Jobs currently executing on a worker thread. + */ + int running{0}; + + /** + * Jobs held back because this type is already at its concurrency + * limit. A non-zero value means work of this type exists and is + * being denied a worker: starvation, not idleness. + */ + int deferred{0}; + }; + + /** + * Snapshot the occupancy of every registered job type. + * + * One mutex acquire copies three integers per type, which is the same + * lock and the same fields getJobCount() already reads — the counting + * logic is not duplicated, only batched, so the caller does not have to + * take the lock once per type to build a full picture. + * + * @return One JobTypeCount per registered JobType, in JobType order. + * + * @note Thread-safe; takes the internal mutex briefly. The values are a + * point-in-time reading and are mutually consistent with each other + * because they come from one acquire. + * @note Intended for a periodic observer (the telemetry reader ticks + * every ~10 s). It is not free enough to call from a hot path. + */ + [[nodiscard]] std::vector + getJobTypeCounts() const; + + /** + * Worker-pool saturation reading: work in flight against capacity. + * + * Answers "is the whole pool exhausted?" in one place. Without it, a + * pool-wide slowdown shows up separately in every subsystem whose jobs + * are queued behind it, and each one looks like its own fault. + */ + struct WorkerSaturation + { + /** + * Calls to processTask() executing right now across all job types. + */ + int runningTasks{0}; + + /** + * Worker threads the pool is configured to run — the ceiling + * `runningTasks` is measured against. + */ + int workerThreads{0}; + + /** + * Jobs queued and not yet dispatched, summed over all job types. + */ + int totalWaiting{0}; + }; + + /** + * Read the global worker-pool saturation. + * + * All three fields are produced under a single mutex acquire so the + * running/threads ratio and the backlog describe the same instant; read + * separately they could disagree and imply a saturation that never + * existed. + * + * @return The current saturation reading. + * + * @note Thread-safe with respect to the queue counters, which are read + * under the internal mutex. The configured thread count is a plain int + * that only changes at pool construction and at stop(); telemetry never + * races that write, because Application detaches the metric callbacks + * before it stops the JobQueue. + * @note Intended for a periodic observer, not a hot path. + */ + [[nodiscard]] WorkerSaturation + getWorkerSaturation() const; + /** * Return a scoped LoadEvent. */ diff --git a/src/libxrpl/core/detail/JobQueue.cpp b/src/libxrpl/core/detail/JobQueue.cpp index 8f95a8a5fa..a12b4a3531 100644 --- a/src/libxrpl/core/detail/JobQueue.cpp +++ b/src/libxrpl/core/detail/JobQueue.cpp @@ -18,6 +18,7 @@ #include #include #include +#include namespace xrpl { @@ -153,6 +154,48 @@ JobQueue::getJobCountGE(JobType t) const return ret; } +std::vector +JobQueue::getJobTypeCounts() const +{ + std::vector out; + + std::scoped_lock const lock(mutex_); + + // Reserve once so the loop itself cannot reallocate while the lock is + // held; the body is then three integer reads per type. + out.reserve(jobData_.size()); + for (auto const& [type, data] : jobData_) + { + out.push_back( + JobTypeCount{ + .type = type, + .waiting = data.waiting, + .running = data.running, + .deferred = data.deferred}); + } + + return out; +} + +JobQueue::WorkerSaturation +JobQueue::getWorkerSaturation() const +{ + // Read the configured thread count and the in-flight task count before + // taking the mutex: neither is guarded by it (numberOfThreads_ is set at + // construction, runningTaskCount_ is an atomic), and holding the queue + // lock across them would add contention for no benefit. + WorkerSaturation out; + out.runningTasks = workers_.numberOfCurrentlyRunningTasks(); + out.workerThreads = workers_.getNumberOfThreads(); + + std::scoped_lock const lock(mutex_); + + for (auto const& entry : jobData_) + out.totalWaiting += entry.second.waiting; + + return out; +} + std::unique_ptr JobQueue::makeLoadEvent(JobType t, std::string const& name) { diff --git a/src/test/core/JobQueue_test.cpp b/src/test/core/JobQueue_test.cpp index aa6a0d60a8..35167c2e16 100644 --- a/src/test/core/JobQueue_test.cpp +++ b/src/test/core/JobQueue_test.cpp @@ -3,9 +3,13 @@ #include #include #include +#include +#include #include #include +#include +#include namespace xrpl::test { @@ -130,12 +134,142 @@ class JobQueue_test : public beast::unit_test::Suite } } + /** + * The telemetry accessors added for the job-queue saturation gauges. + * + * These feed `jobq_backlog{metric,job_type}` and `jobq_saturation{metric}`, + * which are polled from an xrpld observable-gauge callback. The values are + * asserted exactly, because the whole point of the signals is that a + * specific count (especially `deferred`) is correct -- a plausible-looking + * number would misreport starvation as health. + */ + void + testTelemetryAccessors() + { + testcase("telemetry occupancy accessors"); + + jtx::Env env{*this}; + JobQueue& jQueue = env.app().getJobQueue(); + + // --- Every registered type is present, and an idle one reads zero --- + // Absence and zero must be distinguishable: the gauge observes every + // type on every tick, so a missing type would be an exporter bug, not + // an idle queue. + auto const counts = jQueue.getJobTypeCounts(); + + // One entry per registered JobType. JobTypes is the registry the + // JobQueue constructor populates jobData_ from, so the sizes must + // agree exactly -- a mismatch means a type is silently unreported. + BEAST_EXPECT(counts.size() == JobTypes::instance().size()); + + auto findType = [&counts](JobType t) { + return std::ranges::find_if(counts, [t](auto const& c) { return c.type == t; }); + }; + + // A sync-critical type is present even though nothing has been + // enqueued for it, and reads exactly zero on all three fields. + auto const ledgerData = findType(JtLedgerData); + BEAST_EXPECT(ledgerData != counts.end()); + if (ledgerData != counts.end()) + { + BEAST_EXPECT(ledgerData->waiting == 0); + BEAST_EXPECT(ledgerData->running == 0); + BEAST_EXPECT(ledgerData->deferred == 0); + } + + // JtInvalid is NOT a registered type: jobData_ is built from + // JobTypes, whose map excludes it. So the snapshot must not carry it. + BEAST_EXPECT(findType(JtInvalid) == counts.end()); + + // --- The per-type snapshot agrees with the existing accessor --- + // getJobTypeCounts() must not re-implement counting: `waiting` is the + // same field getJobCount() returns, and `waiting + running` is what + // getJobCountTotal() returns. Asserted on every type so a divergence + // anywhere fails, not just on the one type a test happens to poke. + for (auto const& count : counts) + { + BEAST_EXPECT(count.waiting == jQueue.getJobCount(count.type)); + BEAST_EXPECT(count.waiting + count.running == jQueue.getJobCountTotal(count.type)); + } + + // --- Saturation: the pool reports its own capacity --- + auto const saturation = jQueue.getWorkerSaturation(); + + // jtx::Env runs standalone, which the JobQueue ctor maps to exactly + // one worker thread. This is the dashboard ratio's denominator, so it + // must be the real configured count and never zero (a zero would make + // the ratio undefined). + BEAST_EXPECT(saturation.workerThreads == 1); + + // totalWaiting is the sum of the per-type waiting counts from the same + // fields, so the two accessors must agree. + int const summedWaiting = + std::accumulate(counts.begin(), counts.end(), 0, [](int acc, auto const& c) { + return acc + c.waiting; + }); + BEAST_EXPECT(saturation.totalWaiting == summedWaiting); + + // An in-flight task count can never exceed the configured pool size. + BEAST_EXPECT(saturation.runningTasks >= 0); + BEAST_EXPECT(saturation.runningTasks <= saturation.workerThreads); + + // --- A queued job is actually observed --- + // Block one job inside its handler so the queue provably holds work + // while it is sampled: without this the sample could race the job to + // completion and read zeros, which would pass vacuously. + std::atomic release{false}; + std::atomic started{false}; + BEAST_EXPECT(jQueue.addJob(JtClient, "OccupancyBlocker", [&release, &started]() { + started = true; + while (!release) + std::this_thread::yield(); + })); + + while (!started) + std::this_thread::yield(); + + // With the single standalone worker occupied, the type reports exactly + // one job running. + auto const busy = jQueue.getJobTypeCounts(); + auto const busyClient = + std::ranges::find_if(busy, [](auto const& c) { return c.type == JtClient; }); + BEAST_EXPECT(busyClient != busy.end()); + if (busyClient != busy.end()) + BEAST_EXPECT(busyClient->running == 1); + + // And the pool reports exactly one task in flight out of one thread: + // a fully saturated pool, which is the reading the gauge exists for. + auto const busySaturation = jQueue.getWorkerSaturation(); + BEAST_EXPECT(busySaturation.runningTasks == 1); + BEAST_EXPECT(busySaturation.workerThreads == 1); + + release = true; + jQueue.rendezvous(); + + // After draining, the same type reads zero again -- the counters are + // live readings, not a high-water mark. rendezvous() returns with the + // queue mutex having seen finishJob(), so the running count is settled + // by here. (Workers::runningTaskCount_ is decremented only after + // processTask returns, which rendezvous does not wait for, so it is + // deliberately not asserted at this point.) + auto const drained = jQueue.getJobTypeCounts(); + auto const drainedClient = + std::ranges::find_if(drained, [](auto const& c) { return c.type == JtClient; }); + BEAST_EXPECT(drainedClient != drained.end()); + if (drainedClient != drained.end()) + { + BEAST_EXPECT(drainedClient->waiting == 0); + BEAST_EXPECT(drainedClient->running == 0); + } + } + public: void run() override { testAddJob(); testPostCoro(); + testTelemetryAccessors(); } }; diff --git a/src/tests/libxrpl/telemetry/MetricMacros.cpp b/src/tests/libxrpl/telemetry/MetricMacros.cpp index c05312dd8c..7904b4d632 100644 --- a/src/tests/libxrpl/telemetry/MetricMacros.cpp +++ b/src/tests/libxrpl/telemetry/MetricMacros.cpp @@ -25,6 +25,10 @@ #include +#include +#include +#include + #include #include #include @@ -48,6 +52,7 @@ #include #include #include +#include using namespace xrpl; @@ -1584,4 +1589,198 @@ TEST(MetricMacros, acquire_counters_emit_nothing_when_registry_disabled) EXPECT_EQ(app.registry().meterCalls(), 0); } +// ----------------------------------------------------------------- +// JobQueue saturation diagnostics (WP-A4). +// +// Asserts the EXACT values and label shapes of the two gauges: +// jobq_backlog{metric,job_type} MetricsRegistry::registerJobQueueBacklogGauge +// waiting / running / deferred, per type +// jobq_saturation{metric} MetricsRegistry::registerJobQueueSaturationGauge +// running_tasks / worker_threads / total_waiting +// +// Both are observable instruments registered directly on the SDK meter, +// mirroring the production callback shape, because the real MetricsRegistry's +// enabled path cannot be linked into this standalone binary (see the file +// header). The snapshot types are the REAL JobQueue::JobTypeCount and +// JobQueue::WorkerSaturation, and the label values come from the real +// JobTypes::name(), so a rename or reorder on either side breaks these tests +// instead of silently drifting from production. +// ----------------------------------------------------------------- + +// jobq_backlog must keep waiting, running and deferred on separate series per +// job type. `deferred` is the reason this gauge exists: a job held back by its +// type's concurrency limit is counted in neither of the other two fields, and +// appears in no other metric at all. The values chosen are a starved +// JtLedgerData -- limit 3, so 3 running and the rest deferred. +TEST(MetricMacros, jobq_backlog_gauge_separates_waiting_running_and_deferred) +{ + CollectingProvider const provider; + + // The real snapshot type the production callback iterates. JtLedgerData is + // at its limit of 3 with 5 more jobs held back; JtLedgerReq has one job + // merely waiting; JtSweep is registered but idle. + std::vector observed{ + JobQueue::JobTypeCount{.type = JtLedgerData, .waiting = 5, .running = 3, .deferred = 5}, + JobQueue::JobTypeCount{.type = JtLedgerReq, .waiting = 1, .running = 0, .deferred = 0}, + JobQueue::JobTypeCount{.type = JtSweep, .waiting = 0, .running = 0, .deferred = 0}}; + + // Keep the instrument alive for the whole test: destroying the handle + // deregisters the callback, which is why the real registry holds a member. + auto gauge = provider.meter()->CreateInt64ObservableGauge( + "jobq_backlog", "JobQueue occupancy per job type (waiting/running/deferred)"); + gauge->AddCallback( + [](opentelemetry::metrics::ObserverResult result, void* state) { + auto const* counts = static_cast const*>(state); + // Same two-label Observe() form the production callback uses. + auto observe = [&](char const* field, std::string const& jobType, std::int64_t value) { + opentelemetry::nostd::get>>(result) + ->Observe(value, {{"metric", field}, {"job_type", jobType}}); + }; + for (auto const& count : *counts) + { + // The same name helper production uses -- never a literal. + auto const& jobType = JobTypes::name(count.type); + observe("waiting", jobType, count.waiting); + observe("running", jobType, count.running); + observe("deferred", jobType, count.deferred); + } + }, + &observed); + + auto const starved = provider.collect(); + + // Three types x three fields, every one its own series: no field and no + // type collapses into another. + ASSERT_EQ(starved.at("jobq_backlog").size(), 9u); + + // The starved type, exactly as configured. The limit of 3 is visible as + // running=3, and the 5 jobs the limit is denying are the deferred series. + EXPECT_EQ( + gaugeValue(starved, "jobq_backlog", attrs("metric", "waiting", "job_type", "ledgerData")), + 5); + EXPECT_EQ( + gaugeValue(starved, "jobq_backlog", attrs("metric", "running", "job_type", "ledgerData")), + 3); + EXPECT_EQ( + gaugeValue(starved, "jobq_backlog", attrs("metric", "deferred", "job_type", "ledgerData")), + 5); + + // A type that is queued but NOT deferred reads deferred=0 while waiting=1. + // This is the distinction the gauge exists to make: "queued" and "denied a + // worker" are different states, and only the latter is starvation. + EXPECT_EQ( + gaugeValue( + starved, "jobq_backlog", attrs("metric", "waiting", "job_type", "ledgerRequest")), + 1); + EXPECT_EQ( + gaugeValue( + starved, "jobq_backlog", attrs("metric", "deferred", "job_type", "ledgerRequest")), + 0); + + // An idle registered type reports zeros rather than dropping out. Absence + // would be indistinguishable from a broken exporter, so every type is + // observed on every tick. + ASSERT_EQ( + starved.at("jobq_backlog").count(attrs("metric", "waiting", "job_type", "sweep")), 1u); + EXPECT_EQ( + gaugeValue(starved, "jobq_backlog", attrs("metric", "waiting", "job_type", "sweep")), 0); + + // Exactly two label keys, in the documented order, on every series. A + // third label would multiply the series count per job type. + for (auto const& [labels, point] : starved.at("jobq_backlog")) + { + ASSERT_EQ(labels.size(), 2u); + EXPECT_EQ(labels.count("metric"), 1u); + EXPECT_EQ(labels.count("job_type"), 1u); + } + + // NEGATIVE: a type never present in the snapshot has no series, so the + // readings above are not an artifact of a catch-all series. + EXPECT_EQ( + starved.at("jobq_backlog").count(attrs("metric", "waiting", "job_type", "transaction")), + 0u); + // NEGATIVE: the label VALUE is the JobTypes name, not the enum spelling. + EXPECT_EQ( + starved.at("jobq_backlog").count(attrs("metric", "waiting", "job_type", "JtLedgerData")), + 0u); + + // The starvation clearing is the recovery reading: deferred drains to 0 + // while running stays at the limit, so the panel shows work flowing again. + observed[0] = + JobQueue::JobTypeCount{.type = JtLedgerData, .waiting = 0, .running = 3, .deferred = 0}; + auto const draining = provider.collect(); + EXPECT_EQ( + gaugeValue(draining, "jobq_backlog", attrs("metric", "deferred", "job_type", "ledgerData")), + 0); + EXPECT_EQ( + gaugeValue(draining, "jobq_backlog", attrs("metric", "running", "job_type", "ledgerData")), + 3); +} + +// jobq_saturation exports the worker-thread count alongside the in-flight +// count so a dashboard can form the ratio without hardcoding a denominator +// that is derived at startup. The values chosen are a fully exhausted pool. +TEST(MetricMacros, jobq_saturation_gauge_observes_exact_pool_exhaustion_values) +{ + CollectingProvider const provider; + + // The real reading type the production callback consumes: every one of 6 + // workers busy, with 12 jobs queued behind them. + JobQueue::WorkerSaturation observed{.runningTasks = 6, .workerThreads = 6, .totalWaiting = 12}; + + auto gauge = provider.meter()->CreateInt64ObservableGauge( + "jobq_saturation", "Worker-pool saturation: tasks in flight, worker threads, jobs queued"); + gauge->AddCallback( + [](opentelemetry::metrics::ObserverResult result, void* state) { + auto const* self = static_cast(state); + auto observe = [&](char const* field, std::int64_t value) { + opentelemetry::nostd::get>>(result) + ->Observe(value, {{"metric", field}}); + }; + observe("running_tasks", self->runningTasks); + observe("worker_threads", self->workerThreads); + observe("total_waiting", self->totalWaiting); + }, + &observed); + + auto const exhausted = provider.collect(); + + // Exactly three series, one per `metric` value. + ASSERT_EQ(exhausted.at("jobq_saturation").size(), 3u); + EXPECT_EQ(gaugeValue(exhausted, "jobq_saturation", attrs("metric", "running_tasks")), 6); + EXPECT_EQ(gaugeValue(exhausted, "jobq_saturation", attrs("metric", "worker_threads")), 6); + EXPECT_EQ(gaugeValue(exhausted, "jobq_saturation", attrs("metric", "total_waiting")), 12); + + // The label key is exactly "metric" and it is the only label present, so + // this gauge stays a single fixed-cardinality group. + auto const& firstKey = exhausted.at("jobq_saturation").begin()->first; + ASSERT_EQ(firstKey.size(), 1u); + EXPECT_EQ(firstKey.begin()->first, "metric"); + + // A busy-but-not-exhausted pool: same 1.0 ratio, but nothing is queued. + // These two readings are what the ratio alone cannot separate, which is + // why total_waiting is exported next to it. + observed = JobQueue::WorkerSaturation{.runningTasks = 6, .workerThreads = 6, .totalWaiting = 0}; + auto const busy = provider.collect(); + EXPECT_EQ(gaugeValue(busy, "jobq_saturation", attrs("metric", "running_tasks")), 6); + EXPECT_EQ(gaugeValue(busy, "jobq_saturation", attrs("metric", "total_waiting")), 0); + + // An idle pool reads zero in flight with the thread count still reported. + // A zero denominator would make the dashboard ratio undefined, so the + // thread count must never drop out with the load. + observed = JobQueue::WorkerSaturation{.runningTasks = 0, .workerThreads = 6, .totalWaiting = 0}; + auto const idle = provider.collect(); + EXPECT_EQ(gaugeValue(idle, "jobq_saturation", attrs("metric", "running_tasks")), 0); + EXPECT_EQ(gaugeValue(idle, "jobq_saturation", attrs("metric", "worker_threads")), 6); + + // Standalone mode runs a single worker: the denominator is genuinely + // node-specific, which is exactly why it is exported and not hardcoded. + observed = JobQueue::WorkerSaturation{.runningTasks = 1, .workerThreads = 1, .totalWaiting = 3}; + auto const standalone = provider.collect(); + EXPECT_EQ(gaugeValue(standalone, "jobq_saturation", attrs("metric", "worker_threads")), 1); + EXPECT_EQ(gaugeValue(standalone, "jobq_saturation", attrs("metric", "total_waiting")), 3); +} + #endif // XRPL_ENABLE_TELEMETRY diff --git a/src/tests/libxrpl/telemetry/MetricsRegistry.cpp b/src/tests/libxrpl/telemetry/MetricsRegistry.cpp index 008c599ea8..862d46faef 100644 --- a/src/tests/libxrpl/telemetry/MetricsRegistry.cpp +++ b/src/tests/libxrpl/telemetry/MetricsRegistry.cpp @@ -18,7 +18,8 @@ * * CONSEQUENCE for the sync-diagnostics gauges (`unl_quorum`, * `clock_close_offset_seconds`, `sync_state`, - * `server_stall_events_total`, `sync_acquire`, `shamap_cache_hit_rate`): + * `server_stall_events_total`, `sync_acquire`, `shamap_cache_hit_rate`, + * `jobq_backlog`, `jobq_saturation`): * this file CANNOT assert an observed gauge * value, because on this build the gauges do not exist -- their registration * methods and the OTel instrument members are inside @@ -26,7 +27,8 @@ * is provable here, and what the tests below assert, is the complementary * half: that nothing is registered and no service is consulted. The exact * observed values (trusted_keys=5, quorum=4, offset=-3, the sync_state / - * stall-episode values, and the acquire-progress / cache-hit-rate values) are + * stall-episode values, the acquire-progress / cache-hit-rate values, and the + * per-type backlog / pool-saturation values) are * asserted in MetricMacros.cpp, which is the file compiled when telemetry IS * enabled. */ @@ -382,10 +384,11 @@ TEST_F(MetricsRegistryTest, destructor_calls_stop) // `clock_close_offset_seconds` reads TimeKeeper::closeOffset(); `sync_state` and // `server_stall_events_total` read NetworkOPs and LoadManager; `sync_acquire` // reads InboundLedgers::acquireProgress() and `shamap_cache_hit_rate` reads the -// node Family's tree-node cache. All are +// node Family's tree-node cache; `jobq_backlog` and `jobq_saturation` read +// JobQueue::getJobTypeCounts() / getWorkerSaturation(). All are // reached through the ServiceRegistry, and MockServiceRegistry::getValidators() // / getTimeKeeper() / getOPs() / getLoadManager() / getInboundLedgers() / -// getNodeFamily() THROW std::logic_error. So "no +// getNodeFamily() / getJobQueue() THROW std::logic_error. So "no // gauge callback ran" is directly observable here: had registerAsyncGauges() run // and had a callback fired, one of those accessors would have thrown. // @@ -425,7 +428,9 @@ TEST_F(MetricsRegistryTest, disabled_lifecycle_never_consults_gauge_services) // start() is where registerAsyncGauges() -- and with it // registerUnlQuorumGauge() / registerClockSkewGauge() / // registerSyncStateGauge() / registerStallEventsCounter() / - // registerSyncAcquireGauge() / registerCacheHitRateDetailGauge() -- would run. + // registerSyncAcquireGauge() / registerCacheHitRateDetailGauge() / + // registerJobQueueBacklogGauge() / registerJobQueueSaturationGauge() -- + // would run. EXPECT_NO_THROW(registry.start("http://localhost:4318/v1/metrics")); // detachCallbacks() is the shutdown hook the real gauges honour. It must be @@ -454,6 +459,11 @@ TEST_F(MetricsRegistryTest, disabled_lifecycle_never_consults_gauge_services) // Family's tree-node cache. Neither was consulted above. EXPECT_THROW(mockApp_.getInboundLedgers(), std::logic_error); EXPECT_THROW(mockApp_.getNodeFamily(), std::logic_error); + // The service both WP-A4 job-queue gauges read: jobq_backlog polls + // getJobTypeCounts() and jobq_saturation polls getWorkerSaturation(), + // both on the JobQueue. Neither was consulted above, so neither gauge + // took the JobQueue mutex on a telemetry-off build. + EXPECT_THROW(mockApp_.getJobQueue(), std::logic_error); } // Even asking for enabled=true registers no sync-diagnostics gauge on a @@ -473,9 +483,11 @@ TEST_F(MetricsRegistryTest, enabled_flag_alone_registers_no_gauges_when_compiled // Yet the whole lifecycle stays inert. If registerAsyncGauges() had run and // registered registerUnlQuorumGauge()/registerClockSkewGauge()/ // registerSyncStateGauge()/registerStallEventsCounter()/ - // registerSyncAcquireGauge()/registerCacheHitRateDetailGauge(), a callback - // would reach getValidators()/getTimeKeeper()/getOPs()/getLoadManager()/ - // getInboundLedgers()/getNodeFamily() and throw std::logic_error. + // registerSyncAcquireGauge()/registerCacheHitRateDetailGauge()/ + // registerJobQueueBacklogGauge()/registerJobQueueSaturationGauge(), a + // callback would reach getValidators()/getTimeKeeper()/getOPs()/ + // getLoadManager()/getInboundLedgers()/getNodeFamily()/getJobQueue() and + // throw std::logic_error. EXPECT_NO_THROW(enabledRequest.start("http://localhost:4318/v1/metrics")); EXPECT_NO_THROW(enabledRequest.detachCallbacks()); EXPECT_NO_THROW(enabledRequest.stop()); diff --git a/src/xrpld/telemetry/MetricsRegistry.cpp b/src/xrpld/telemetry/MetricsRegistry.cpp index 5a3ec3699b..f6c684628c 100644 --- a/src/xrpld/telemetry/MetricsRegistry.cpp +++ b/src/xrpld/telemetry/MetricsRegistry.cpp @@ -37,6 +37,8 @@ #include #include #include +#include +#include #include #include #include @@ -490,6 +492,8 @@ MetricsRegistry::registerAsyncGauges() registerStallEventsCounter(); registerSyncAcquireGauge(); registerCacheHitRateDetailGauge(); + registerJobQueueBacklogGauge(); + registerJobQueueSaturationGauge(); } void @@ -1721,6 +1725,99 @@ MetricsRegistry::registerCacheHitRateDetailGauge() this); } +void +MetricsRegistry::registerJobQueueBacklogGauge() +{ + // --- Sync diagnostics: which job types are starved right now? --- + // The existing job_* counters and histograms describe jobs that already + // moved. This is instantaneous occupancy, and `deferred` in particular + // has no other exposure: a job held back by its type's concurrency limit + // counts as neither waiting nor running anywhere else. + jobQueueBacklogGauge_ = meter_->CreateInt64ObservableGauge( + "jobq_backlog", "JobQueue occupancy per job type (waiting/running/deferred)"); + jobQueueBacklogGauge_->AddCallback( + [](opentelemetry::metrics::ObserverResult result, void* state) { + auto* self = static_cast(state); + if (self->callbacksDetached_.load(std::memory_order_acquire)) + return; + auto& app = self->app_; + + try + { + auto observe = [&](char const* field, std::string const& jobType, int64_t value) { + opentelemetry::nostd::get>>(result) + ->Observe(value, {{"metric", field}, {"job_type", jobType}}); + }; + + // One snapshot under one lock acquire, so the three fields of + // a type are mutually consistent rather than read at three + // different instants. + for (auto const& count : app.getJobQueue().getJobTypeCounts()) + { + // The name helper is the single source of the label value, + // the same one the job_*_total counters already use, so the + // two label sets join. + auto const& jobType = JobTypes::name(count.type); + observe("waiting", jobType, count.waiting); + observe("running", jobType, count.running); + observe("deferred", jobType, count.deferred); + } + } + catch (...) // NOLINT(bugprone-empty-catch) + { + // Silently skip if services are not yet ready. + } + }, + this); +} + +void +MetricsRegistry::registerJobQueueSaturationGauge() +{ + // --- Sync diagnostics: is the whole worker pool exhausted? --- + // Attributes a broad multi-stage slowdown to the pool once, instead of + // leaving it to look like an independent fault in every subsystem whose + // jobs are queued behind it. + jobQueueSaturationGauge_ = meter_->CreateInt64ObservableGauge( + "jobq_saturation", "Worker-pool saturation: tasks in flight, worker threads, jobs queued"); + jobQueueSaturationGauge_->AddCallback( + [](opentelemetry::metrics::ObserverResult result, void* state) { + auto* self = static_cast(state); + if (self->callbacksDetached_.load(std::memory_order_acquire)) + return; + auto& app = self->app_; + + try + { + auto observe = [&](char const* field, int64_t value) { + opentelemetry::nostd::get>>(result) + ->Observe(value, {{"metric", field}}); + }; + + // One reading feeds all three series so the ratio and the + // backlog describe the same instant. + auto const saturation = app.getJobQueue().getWorkerSaturation(); + observe("running_tasks", saturation.runningTasks); + + // The denominator for the ratio panel. Derived at startup from + // [workers], node size and hardware concurrency, so it cannot + // be hardcoded in a dashboard. + observe("worker_threads", saturation.workerThreads); + + // Ratio at 1.0 alone is a busy pool; ratio at 1.0 with a + // non-zero backlog is an exhausted one. + observe("total_waiting", saturation.totalWaiting); + } + catch (...) // NOLINT(bugprone-empty-catch) + { + // Silently skip if services are not yet ready. + } + }, + this); +} + #endif // XRPL_ENABLE_TELEMETRY // ----------------------------------------------------------------- diff --git a/src/xrpld/telemetry/MetricsRegistry.h b/src/xrpld/telemetry/MetricsRegistry.h index 7f411983e5..2a10e4a3d4 100644 --- a/src/xrpld/telemetry/MetricsRegistry.h +++ b/src/xrpld/telemetry/MetricsRegistry.h @@ -64,6 +64,8 @@ * +-- Clock close offset (local clock skew) * +-- Sync state (time to first FULL, network-ledger gate, * | server stall seconds, ledgers behind network) + * +-- JobQueue backlog (waiting/running/deferred per job type) + * +-- JobQueue saturation (running tasks vs worker threads vs backlog) * +-- jq_trans_overflow_total (observed from Overlay) * +-- server_stall_events_total (observed from LoadManager) * @@ -582,6 +584,18 @@ private: */ opentelemetry::nostd::shared_ptr shamapCacheHitRateGauge_; + /** + * Observable gauge for per-job-type JobQueue occupancy: waiting, running + * and deferred counts, keyed by job type. + */ + opentelemetry::nostd::shared_ptr + jobQueueBacklogGauge_; + /** + * Observable gauge for global worker-pool saturation: tasks in flight, + * configured worker threads, and total jobs queued. + */ + opentelemetry::nostd::shared_ptr + jobQueueSaturationGauge_; /** * Observable gauge for build version info (label-based, value=1). */ @@ -902,6 +916,68 @@ private: */ void registerCacheHitRateDetailGauge(); // sync diagnostics: treenode cache + + /** + * Register the `jobq_backlog` gauge. + * + * Three series per job type, from one JobQueue::getJobTypeCounts() + * snapshot, under the `metric` and `job_type` attributes: + * + * `waiting` — jobs enqueued and not yet dispatched to a worker. + * `running` — jobs executing on a worker. + * `deferred` — **the signal this gauge exists for.** Jobs held back + * because the type is already at its concurrency limit. The + * sync-critical types run at limits of 3 (`JtLedgerReq`, + * `JtLedgerData` in JobTypes.h), so during a fresh sync those types + * routinely have work denied a worker, and that state appears in + * neither `waiting` nor `running`. + * + * Distinct from the job metrics that already exist. `job_queued_total` / + * `job_started_total` / `job_finished_total` and the `job_queued_us` / + * `job_running_us` histograms are all event-driven and come from + * PerfLogImp: they describe jobs that already moved. This gauge is + * instantaneous occupancy — what is sitting in the queue right now, which + * a rate or a latency quantile cannot express. The StatsD + * `jobq_job_count` gauge is queue-wide only, with no per-type split and + * no deferred count at all. + * + * `job_type` is the JobTypes::name() string, matching the label the job + * counters already use so the two can be joined. Cardinality is bounded + * by the JobType enum (~46 values), and every type is observed on every + * tick, so an idle type reports 0 rather than dropping its series. + * + * @note Pulled on the OTel reader thread (~10 s tick), never on a hot + * path. Takes the JobQueue mutex once per tick for three integer reads + * per type; no per-job cost is added anywhere. + */ + void + registerJobQueueBacklogGauge(); // sync diagnostics: per-type backlog + + /** + * Register the `jobq_saturation` gauge. + * + * Three series under the `metric` attribute, from one + * JobQueue::getWorkerSaturation() reading: + * + * `running_tasks` — worker threads currently executing a job. + * `worker_threads` — threads the pool is configured to run, the + * denominator that makes `running_tasks` legible. Exported rather + * than hardcoded in the dashboard because it is derived at startup + * from `[workers]`, node size and hardware concurrency. + * `total_waiting` — jobs queued across all types. + * + * The reason this is separate from `jobq_backlog`: when the pool itself + * is exhausted, every subsystem waiting behind it looks independently + * slow, and each per-type panel invites the wrong conclusion. A + * `running_tasks / worker_threads` ratio at 1.0 with a non-zero + * `total_waiting` attributes the whole slowdown to pool exhaustion once. + * + * @note Pulled on the OTel reader thread (~10 s tick). One atomic load, + * one plain int read, and one pass over the per-type counters under the + * JobQueue mutex. + */ + void + registerJobQueueSaturationGauge(); // sync diagnostics: pool saturation #endif // XRPL_ENABLE_TELEMETRY };