ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-10-10 05:38:10 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
8e50c6f9004e119638c747cb89d8edfb48e51200
rippled/src/xrpld/telemetry/MetricsRegistry.cpp
Pratik Mankawde 31b58e6724 fix(telemetry): build the metrics pipeline at construction, before any producer 
MetricsRegistry created its provider and synchronous instruments in start(),
called from setup() after the node identity was read. Every XRPL_METRIC_*
call site creates its instrument on first use, so any site that ran before
that point found no meter and never recorded again. The start was moved three
times to chase the newest early caller; nothing guaranteed the order.

Build the pipeline in the constructor instead. ApplicationImp declares
metricsRegistry_ right after telemetry_ and before every subsystem, so
declaration order now guarantees the instruments exist before any producer.
start() is gone and its config parsing moves to makeMetricsRegistryOptions().

The registry now guarantees a meter whenever it is enabled: the real one, or
the OTel no-op meter if the pipeline failed to build. disablePipeline() owns
that fallback and its one error log, and the constructor routes both
std::exception and a non-std throw through it, because the SDK is third-party
code. So the macros shrink to one function-local static built from meter()
plus the record call: no once-flag, no null check, and no path for a call
that arrives before the meter, because that state no longer exists. The three
observable macros drop the same now-dead meter check.

The lifecycle is three explicit phases with a Phase enum: Ready at
construction, GaugesArmed by startAsyncGauges() once overlay_ exists, and
Stopped by stop(). startAsyncGauges() checks the phase before the pipeline,
so a second call and a call after stop() are each reported as what they are.
run() stops both observers (insight collector, registry) before any service,
and ~ApplicationImp repeats the stop for the setup() failure paths that never
reach run(). stop() already detaches the callbacks, so it is the only call.

Meter name and version come from kMeterName/kMeterVersion, and the endpoint
default from Telemetry::Setup, so the two metric pipelines share one source
for both.
2026-09-14 20:12:20 +01:00

1795 lines
76 KiB
C++
Raw Blame History

/**
* MetricsRegistry implementation — OpenTelemetry metric instruments for xrpld.
*
* This file contains:
* - Construction / destruction logic for the OTel MeterProvider pipeline.
* - Synchronous instrument creation (counters, histograms) for RPC, job
* queue, and NodeStore I/O metrics.
* - Observable gauge callback registration for cache hit rates, TxQ state,
* CountedObject instances, load factors, and NodeStore queue depth.
* - No-op stubs when XRPL_ENABLE_TELEMETRY is not defined.
*/
// On Windows, OTel's spin_lock_mutex.h (transitively included from
// MetricsRegistry.h) defines _WINSOCKAPI_ and includes <windows.h>.
// This poisons the include state for boost/asio/detail/socket_types.hpp,
// which requires winsock2.h to be included first. Pre-including the
// boost/asio socket types header gets winsock2.h in before the OTel
// headers can interfere.
#ifdef _MSC_VER
#include <boost/asio/detail/socket_types.hpp>
#endif
#include <xrpld/telemetry/MetricsRegistry.h>
// Unguarded because the constructor's `beast::Journal journal` parameter is
// declared in both configurations; only the member it initialises is guarded.
#include <xrpl/beast/utility/Journal.h>
#ifdef XRPL_ENABLE_TELEMETRY
// The app and overlay includes below are why
// .github/scripts/levelization/results/loops.txt records
// `xrpld.app <-> xrpld.telemetry` and `xrpld.overlay <-> xrpld.telemetry` as
// cycles, rather than an acyclic ordering.txt entry placing telemetry strictly
// below both. The observable gauges are pull-model: their callbacks sample live
// state when the reader thread fires, so they need the concrete types to call
// getJqTransOverflow(), size(), getPeerDisconnectCharges(), foreach() and
// txMetrics().
//
// The cycle is confined to this translation unit. No telemetry header includes
// app or overlay (MetricsRegistry.h forward-declares what it needs and takes a
// ServiceRegistry&), and all of src/xrpld builds into a single CMake target, so
// there is no header cycle and no link cycle to break.
//
// Inverting it properly means declaring a metrics-source interface below overlay
// and implementing it there, which is deliberately left as follow-up rather than
// widening this change. Note loops.txt is generated: it can only change as a
// consequence of changing these includes, never by editing the baseline.
#include <xrpld/app/ledger/AcquireStats.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/OpenLedger.h>
#include <xrpld/app/misc/TxQ.h>
#include <xrpld/app/misc/ValidatorList.h>
#include <xrpld/core/TimeKeeper.h>
#include <xrpld/overlay/Overlay.h>
#include <xrpl/basics/CountedObject.h>
#include <xrpl/basics/Log.h>
#include <xrpl/basics/UptimeClock.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/json/json_value.h>
#include <xrpl/nodestore/Database.h>
#include <xrpl/protocol/BuildInfo.h>
#include <xrpl/protocol/jss.h>
#include <xrpl/rdb/RelationalDatabase.h>
#include <xrpl/server/LoadFeeTrack.h>
#include <xrpl/server/NetworkOPs.h>
#include <xrpl/telemetry/GetObjectMetricNames.h>
#include <xrpl/telemetry/HistogramBuckets.h>
#include <xrpl/telemetry/RpcMetricNames.h>
#include <xrpl/telemetry/SpanNames.h>
// For networkTypeFromId(), the one xrpl.network.type mapping both export
// paths use. Adds no levelization edge: xrpld.telemetry > xrpl.telemetry
// already holds via SpanNames.h above.
#include <xrpl/telemetry/Telemetry.h>
#include <opentelemetry/context/context.h>
#include <opentelemetry/exporters/otlp/otlp_http_metric_exporter_factory.h>
#include <opentelemetry/exporters/otlp/otlp_http_metric_exporter_options.h>
#include <opentelemetry/metrics/noop.h>
#include <opentelemetry/metrics/observer_result.h>
#include <opentelemetry/nostd/shared_ptr.h>
#include <opentelemetry/nostd/variant.h>
#include <opentelemetry/sdk/metrics/aggregation/aggregation_config.h>
#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader_factory.h>
#include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader_options.h>
#include <opentelemetry/sdk/metrics/instruments.h>
#include <opentelemetry/sdk/metrics/meter_provider_factory.h>
#include <opentelemetry/sdk/metrics/view/instrument_selector_factory.h>
#include <opentelemetry/sdk/metrics/view/meter_selector_factory.h>
#include <opentelemetry/sdk/metrics/view/view_factory.h>
#include <opentelemetry/sdk/metrics/view/view_registry.h>
#include <opentelemetry/sdk/resource/resource.h>
#include <opentelemetry/semconv/incubating/service_attributes.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <sstream>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
namespace metric_sdk = opentelemetry::sdk::metrics;
namespace otlp_http = opentelemetry::exporter::otlp;
// Not `resource`: that would collide with xrpl::resource (the resource-accounting
// namespace), which encloses every use site below. Inner-scope lookup would find
// that namespace instead of this file-scope alias.
namespace otel_resource = opentelemetry::sdk::resource;
namespace {
// Microsecond-valued duration histogram instrument names. Each is
// referenced twice — once to register the explicit-bucket view and once
// to create the instrument — so they are named constants to keep the two
// sites in sync (a mismatch would silently drop the bucket override).
constexpr char kJobQueuedDurationUs[] = "job_queued_us";
constexpr char kJobRunningDurationUs[] = "job_running_us";
constexpr char kRpcMethodDurationUs[] = "rpc_method_us";
// Attribute (label) keys for the job instruments. Each is referenced from
// several record sites, and a counter and its histogram must carry exactly
// the same key spelling or the two series cannot be joined in a query.
constexpr char kJobTypeLabel[] = "job_type";
constexpr char kHandlerLabel[] = "handler";
/**
* Register an explicit-bucket histogram view.
*
* The SDK's default boundaries top out at 10,000, so any instrument whose
* values exceed that saturates and every quantile reads as the ceiling. The
* floor matters just as much and is easier to miss: a ladder whose first edge
* sits above the mass of the distribution makes every low quantile an
* interpolation inside bucket 0 -- a number derived from the bucket edge
* rather than from any sample. Both ends are chosen from measured
* distributions in HistogramBuckets.h.
*
* @param views The registry to add the view to.
* @param name Instrument name to match (e.g. "job_running_us").
* @param boundaries Bucket upper bounds, ascending.
*/
void
addHistogramView(
metric_sdk::ViewRegistry& views,
std::string const& name,
std::vector<double> boundaries)
{
auto config = std::make_shared<metric_sdk::HistogramAggregationConfig>();
config->boundaries_ = std::move(boundaries);
auto selector = metric_sdk::InstrumentSelectorFactory::Create(
metric_sdk::InstrumentType::kHistogram, name, "");
auto meterSelector = metric_sdk::MeterSelectorFactory::Create(
std::string(xrpl::telemetry::kMeterName), std::string(xrpl::telemetry::kMeterVersion), "");
auto view =
metric_sdk::ViewFactory::Create(name, "", metric_sdk::AggregationType::kHistogram, config);
views.AddView(std::move(selector), std::move(meterSelector), std::move(view));
}
/**
* Register the microsecond-ladder view for a duration instrument.
*
* Job wait/run times and RPC latencies routinely exceed the SDK default
* ceiling, so they all share `buckets::kMicrosecondBuckets`.
*
* @param views The registry to add the view to.
* @param name Instrument name to match.
*/
void
addMicrosecondHistogramView(metric_sdk::ViewRegistry& views, std::string const& name)
{
addHistogramView(
views,
name,
xrpl::telemetry::buckets::toVector(xrpl::telemetry::buckets::kMicrosecondBuckets));
}
} // namespace
#endif // XRPL_ENABLE_TELEMETRY
namespace xrpl::telemetry {
MetricsRegistry::MetricsRegistry(
[[maybe_unused]] bool enabled,
[[maybe_unused]] ServiceRegistry& app,
[[maybe_unused]] beast::Journal journal,
[[maybe_unused]] Options const& options)
: enabled_(enabled)
#ifdef XRPL_ENABLE_TELEMETRY
, app_(app)
, journal_(journal)
#endif
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_)
return;
// useTls is logged because a collector that requires TLS rejects a
// plaintext exporter with no local error. The paths are left out.
JLOG(journal_.info()) << "MetricsRegistry: starting, endpoint=" << options.endpoint
<< ", serviceName=" << options.serviceName
<< ", serviceVersion=" << options.serviceVersion
<< ", instanceId=" << options.serviceInstanceId
<< ", nodeId=" << options.nodeId << ", networkId=" << options.networkId
<< ", useTls=" << options.useTls;
// A broken pipeline must not stop the node. The SDK is third-party code,
// so the catch-all is deliberate, as in ~ApplicationImp.
try
{
initExporterAndProvider(options);
// Rule for anything added below: the constructor may create only
// instruments whose recording is PUSHED from app code -- counters and
// histograms. An instrument registered here is live immediately, and
// the reader thread may invoke a registered callback before the rest
// of the Application is built, so any observable whose callback reads
// an Application service belongs in startAsyncGauges(), not here.
// That includes observable COUNTERS, not just gauges:
// jq_trans_overflow_total was created here and its callback read
// getOverlay(), which asserts overlay_ is non-null.
initSyncInstruments();
}
catch (std::exception const& e)
{
disablePipeline(e.what());
return;
}
catch (...)
{
disablePipeline("unknown exception");
return;
}
JLOG(journal_.info()) << "MetricsRegistry: provider and instruments ready";
#endif // XRPL_ENABLE_TELEMETRY
}
#ifdef XRPL_ENABLE_TELEMETRY
void
MetricsRegistry::disablePipeline(std::string_view reason)
{
provider_.reset();
// meter_ becomes a no-op meter, which keeps the invariant the
// XRPL_METRIC_* macros rely on: an enabled registry always has a meter,
// so every call site gets an instrument (a no-op one here) with no check
// of its own. Through the base pointer, as Telemetry::getMeter() does:
// the no-op provider's override hides the base class's defaulted overload.
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::MeterProvider> const noop(
new opentelemetry::metrics::NoopMeterProvider());
meter_ = noop->GetMeter(std::string(kMeterName), std::string(kMeterVersion));
JLOG(journal_.error()) << "MetricsRegistry: metrics pipeline failed to initialise, "
"continuing without native metrics: "
<< reason;
}
#endif // XRPL_ENABLE_TELEMETRY
MetricsRegistry::~MetricsRegistry()
{
stop();
}
void
MetricsRegistry::startAsyncGauges()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_)
return;
// One arm per life. A second call would create a second set of
// same-named instruments, and a call after stop() would register on a
// provider that is gone. Checked before the pipeline, so a call after
// stop() is reported as what it is and not as a build failure.
if (phase_ != Phase::Ready)
{
JLOG(journal_.warn()) << "MetricsRegistry: startAsyncGauges() called "
<< (phase_ == Phase::Stopped ? "after stop()" : "twice")
<< "; ignored";
return;
}
// The pipeline failed to build: the meter is a no-op, so registering
// gauges on it would only log a success that is not one.
if (!provider_)
{
JLOG(journal_.warn()) << "MetricsRegistry: startAsyncGauges() without a pipeline; "
"no gauges registered";
return;
}
phase_ = Phase::GaugesArmed;
registerAsyncGauges();
JLOG(journal_.info()) << "MetricsRegistry: started successfully";
#endif // XRPL_ENABLE_TELEMETRY
}
#ifdef XRPL_ENABLE_TELEMETRY
void
MetricsRegistry::initExporterAndProvider(Options const& options)
{
// Configure OTLP/HTTP metric exporter. The TLS settings come from the one
// [telemetry] block that also drives the trace exporter in Telemetry.cpp,
// so both exporters reach the collector on the same terms.
otlp_http::OtlpHttpMetricExporterOptions exporterOpts;
exporterOpts.url = options.endpoint;
if (options.useTls)
{
exporterOpts.ssl_ca_cert_path = options.tlsCaCertPath;
exporterOpts.ssl_client_cert_path = options.tlsClientCertPath;
exporterOpts.ssl_client_key_path = options.tlsClientKeyPath;
}
auto exporter = otlp_http::OtlpHttpMetricExporterFactory::Create(exporterOpts);
// Configure periodic reader with 10-second export interval.
metric_sdk::PeriodicExportingMetricReaderOptions readerOpts;
readerOpts.export_interval_millis = std::chrono::milliseconds(10000);
readerOpts.export_timeout_millis = std::chrono::milliseconds(5000);
auto reader =
metric_sdk::PeriodicExportingMetricReaderFactory::Create(std::move(exporter), readerOpts);
// Stamp the same resource Telemetry::makeMetricsResource() builds for the
// trace pipeline. Both must agree: a node whose service.name or
// xrpl.network.type differs between the two pipelines splits its own
// series, and a dashboard filtering on either label shows only half.
//
// Use std::string, never a string literal: ResourceAttributes stores an
// OTel AttributeValue variant whose char-const* overload binds to bool,
// so a literal would be recorded as the boolean true.
otel_resource::ResourceAttributes attrs;
attrs[opentelemetry::semconv::service::kServiceName] = options.serviceName;
// int64_t, matching the trace resource. The same key with two types would
// give the two pipelines incompatible attribute values.
attrs[std::string(attr::networkId)] = static_cast<int64_t>(options.networkId);
// Derived here rather than passed in, so the id and the type label cannot
// disagree. Same helper the trace path uses.
attrs[std::string(attr::networkType)] = networkTypeFromId(options.networkId);
// The three below are left off when empty rather than stamped blank. An
// absent label reads as "not reported"; an empty one looks like a value.
if (!options.serviceVersion.empty())
attrs[opentelemetry::semconv::service::kServiceVersion] = options.serviceVersion;
if (!options.serviceInstanceId.empty())
attrs[opentelemetry::semconv::service::kServiceInstanceId] = options.serviceInstanceId;
// xrpl.node.id: the same per-node key the trace resource carries, so
// metrics and traces resolve to one node.
if (!options.nodeId.empty())
attrs[std::string(attr::nodeId)] = options.nodeId;
auto resourceAttrs = otel_resource::Resource::Create(attrs);
// Build a view registry with explicit microsecond buckets for the
// duration histograms. Without this they use the SDK default buckets
// (max 10,000 = 10 ms), saturating every quantile at 10 ms.
auto views = std::make_unique<metric_sdk::ViewRegistry>();
addMicrosecondHistogramView(*views, kJobQueuedDurationUs);
addMicrosecondHistogramView(*views, kJobRunningDurationUs);
addMicrosecondHistogramView(*views, kRpcMethodDurationUs);
// Recorded at its PeerImp.cpp call site, not created here, so the name
// comes from the shared constant both sites use.
addMicrosecondHistogramView(*views, kGetObjectLookupUs);
// The remaining two GetObject histograms are not durations, so the
// microsecond ladder above does not fit them. Both still need explicit
// boundaries: the SDK default stops at 10,000 and both ranges exceed it.
//
// Object counts run 1..kHardMaxReplyNodes (12288). The honest sync path
// asks for at most 8, so the low buckets are fine-grained and the upper
// ones follow the charge size bands (64, 1024) up to the hard cap.
addHistogramView(
*views, kGetObjectRequestObjects, buckets::toVector(buckets::kObjectCountBuckets));
// Charge values span 0 (free tier) to ~99k for a full-size all-miss
// request. Boundaries bracket the resource thresholds that decide a
// peer's fate -- kWarningThreshold (5000) and kDropThreshold (25000) --
// so a dashboard can show how close charges run to each.
addHistogramView(*views, kGetObjectCharge, buckets::toVector(buckets::kChargeBuckets));
// The two RPC request-count histograms are recorded at their ServerHandler
// and PathRequest call sites, so the names come from the shared constants
// all three sites use. Both are small counts, and the reason they need a
// view is the FLOOR rather than the ceiling: the SDK default edges start
// 0, 5, 10, 25, so a batch of one to five sub-requests -- the normal case --
// would land in a single bucket and every quantile over it would be an
// interpolation inside that bucket rather than a measurement.
//
// The object-count ladder is the fit: its 1, 2, 4, 8, 16 edges sit exactly
// where both distributions have their mass. Path counts are hard-bounded at
// kMaxPaths * kMaxAutoSrcCur = 352, well under its 12288 top. Batch sizes
// have no such cap; see the ceiling note in RpcMetricNames.h.
addHistogramView(*views, kRpcBatchSize, buckets::toVector(buckets::kObjectCountBuckets));
addHistogramView(
*views, kPathfindDiscoveredPaths, buckets::toVector(buckets::kObjectCountBuckets));
// Create MeterProvider with resource, then attach the metric reader.
provider_ = metric_sdk::MeterProviderFactory::Create(std::move(views), resourceAttrs);
provider_->AddMetricReader(std::move(reader));
// Get a meter for all xrpld instruments.
meter_ = provider_->GetMeter(std::string(kMeterName), std::string(kMeterVersion));
}
void
MetricsRegistry::initSyncInstruments()
{
// RPC per-method counters and histogram.
rpcStartedCounter_ =
meter_->CreateUInt64Counter("rpc_method_started_total", "Total RPC method calls started");
rpcFinishedCounter_ = meter_->CreateUInt64Counter(
"rpc_method_finished_total", "Total RPC method calls completed successfully");
rpcErroredCounter_ = meter_->CreateUInt64Counter(
"rpc_method_errored_total", "Total RPC method calls that errored");
rpcDurationHistogram_ = meter_->CreateDoubleHistogram(
kRpcMethodDurationUs, "RPC method execution time in microseconds");
// Job queue per-type counters and histograms.
jobQueuedCounter_ = meter_->CreateUInt64Counter("job_queued_total", "Total jobs enqueued");
jobStartedCounter_ = meter_->CreateUInt64Counter("job_started_total", "Total jobs started");
jobFinishedCounter_ = meter_->CreateUInt64Counter("job_finished_total", "Total jobs completed");
jobQueuedDurationHistogram_ = meter_->CreateDoubleHistogram(
kJobQueuedDurationUs, "Time jobs spent waiting in the queue (microseconds)");
jobRunningDurationHistogram_ =
meter_->CreateDoubleHistogram(kJobRunningDurationUs, "Job execution time in microseconds");
// --- External dashboard parity counters ---
ledgersClosedCounter_ =
meter_->CreateUInt64Counter("ledgers_closed_total", "Total ledgers closed by consensus");
validationsSentCounter_ = meter_->CreateUInt64Counter(
"validations_sent_total", "Total validations sent by this node");
validationsCheckedCounter_ = meter_->CreateUInt64Counter(
"validations_checked_total", "Total network validations received and checked");
stateChangesCounter_ =
meter_->CreateUInt64Counter("state_changes_total", "Total operating mode changes");
ledgerHistoryMismatchCounter_ = meter_->CreateUInt64Counter(
"ledger_history_mismatch_total", "Total built-vs-validated ledger mismatches by reason");
txqExpiredCounter_ = meter_->CreateUInt64Counter(
"txq_expired_total", "Total transactions expired out of the transaction queue");
txqDroppedCounter_ = meter_->CreateUInt64Counter(
"txq_dropped_total", "Total transactions refused admission to the queue by reason");
// Note: validation_agreements_total / validation_missed_total are monotonic
// ObservableCounters created in registerValidationTotalsCounters() (below).
}
#endif // XRPL_ENABLE_TELEMETRY
void
MetricsRegistry::detachCallbacks() noexcept
{
#ifdef XRPL_ENABLE_TELEMETRY
// Release so every subsequent callback acquire-load sees true.
callbacksDetached_.store(true, std::memory_order_release);
#endif // XRPL_ENABLE_TELEMETRY
}
void
MetricsRegistry::stop()
{
#ifdef XRPL_ENABLE_TELEMETRY
// Idempotent: the destructor calls this after run() or the Application
// destructor already did.
phase_ = Phase::Stopped;
if (!provider_)
return;
JLOG(journal_.info()) << "MetricsRegistry: stopping";
// Belt-and-suspenders: detachCallbacks() should have already been
// called by Application shutdown before any service the callbacks
// observe was stopped. Setting the flag here is redundant for a
// correct caller but protects against a future caller that forgets
// to detach first.
callbacksDetached_.store(true, std::memory_order_release);
// SDK teardown order: Shutdown() stops the PeriodicExportingMetricReader
// thread (so no further gauge callbacks fire) and performs the final
// collect-and-export drain itself. The trailing ForceFlush() is a
// redundant safety net (a no-op once the reader is shut down), then
// reset() destroys the provider.
provider_->Shutdown();
provider_->ForceFlush();
provider_.reset();
JLOG(journal_.info()) << "MetricsRegistry: stopped";
#endif // XRPL_ENABLE_TELEMETRY
}
// -----------------------------------------------------------------
// Synchronous instrument recording — RPC metrics
// -----------------------------------------------------------------
void
MetricsRegistry::recordRpcStarted([[maybe_unused]] std::string_view method)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !rpcStartedCounter_)
return;
rpcStartedCounter_->Add(1, {{"method", std::string(method)}});
#endif
}
void
MetricsRegistry::recordRpcFinished(
[[maybe_unused]] std::string_view method,
[[maybe_unused]] std::int64_t durationUs)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !rpcFinishedCounter_)
return;
rpcFinishedCounter_->Add(1, {{"method", std::string(method)}});
if (rpcDurationHistogram_)
{
rpcDurationHistogram_->Record(
static_cast<double>(durationUs),
{{"method", std::string(method)}},
opentelemetry::context::Context{});
}
#endif
}
void
MetricsRegistry::recordRpcErrored(
[[maybe_unused]] std::string_view method,
[[maybe_unused]] std::int64_t durationUs)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !rpcErroredCounter_)
return;
rpcErroredCounter_->Add(1, {{"method", std::string(method)}});
if (rpcDurationHistogram_)
{
rpcDurationHistogram_->Record(
static_cast<double>(durationUs),
{{"method", std::string(method)}},
opentelemetry::context::Context{});
}
#endif
}
// -----------------------------------------------------------------
// Synchronous instrument recording — Job Queue metrics
// -----------------------------------------------------------------
void
MetricsRegistry::recordJobQueued(
[[maybe_unused]] std::string_view jobType,
[[maybe_unused]] std::string_view jobName)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !jobQueuedCounter_)
return;
jobQueuedCounter_->Add(
1,
{{kJobTypeLabel, std::string(jobType)},
{kHandlerLabel, std::string(sanitiseHandler(jobName))}});
#endif
}
void
MetricsRegistry::recordJobStarted(
[[maybe_unused]] std::string_view jobType,
[[maybe_unused]] std::string_view jobName,
[[maybe_unused]] std::int64_t queuedDurUs)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !jobStartedCounter_)
return;
// Build the attribute pair once: both the counter and the histogram
// must carry the identical label set or they cannot be joined.
std::string const handler(sanitiseHandler(jobName));
jobStartedCounter_->Add(1, {{kJobTypeLabel, std::string(jobType)}, {kHandlerLabel, handler}});
if (jobQueuedDurationHistogram_ && queuedDurUs >= 0)
{
// Guard against negative queued durations: the caller derives this
// from a steady-clock delta that can go slightly negative under clock
// skew or reordering. The OTel SDK rejects negative histogram values
// (logging a warning per call), so skip them rather than spam.
jobQueuedDurationHistogram_->Record(
static_cast<double>(queuedDurUs),
{{kJobTypeLabel, std::string(jobType)}, {kHandlerLabel, handler}},
opentelemetry::context::Context{});
}
#endif
}
void
MetricsRegistry::recordJobFinished(
[[maybe_unused]] std::string_view jobType,
[[maybe_unused]] std::string_view jobName,
[[maybe_unused]] std::int64_t runningDurUs)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (!enabled_ || !jobFinishedCounter_)
return;
std::string const handler(sanitiseHandler(jobName));
jobFinishedCounter_->Add(1, {{kJobTypeLabel, std::string(jobType)}, {kHandlerLabel, handler}});
if (jobRunningDurationHistogram_)
{
jobRunningDurationHistogram_->Record(
static_cast<double>(runningDurUs),
{{kJobTypeLabel, std::string(jobType)}, {kHandlerLabel, handler}},
opentelemetry::context::Context{});
}
#endif
}
// -----------------------------------------------------------------
// Observable gauge callbacks
// -----------------------------------------------------------------
#ifdef XRPL_ENABLE_TELEMETRY
void
MetricsRegistry::registerAsyncGauges()
{
// Each helper creates one observable instrument and attaches one
// callback. Keeping the registration bodies in separate methods
// preserves the 80-line-per-function limit enforced by CLAUDE.md.
registerJqTransOverflowCounter();
registerCacheHitRateGauge();
registerTxqGauge();
registerObjectCountGauge();
registerLoadFactorGauge();
registerNodeStoreGauge();
registerServerInfoGauge();
registerBuildInfoGauge();
registerCompleteLedgersGauge();
registerDbMetricsGauge();
registerValidatorHealthGauge();
registerPeerQualityGauge();
registerReduceRelayGauge();
registerLedgerEconomyGauge();
registerStateTrackingGauge();
registerStorageDetailGauge();
registerValidationAgreementGauge();
registerValidationTotalsCounters();
}
void
MetricsRegistry::registerJqTransOverflowCounter()
{
// jq_trans_overflow_total is observed from Overlay's existing cumulative
// atomic (Overlay::getJqTransOverflow()) rather than pushed. The overlay
// owns the only increment site (PeerImp), so an ObservableCounter reads the
// live total each collection cycle without threading a push path through
// develop-owned overlay code.
//
// Registered with the gauges, not with the synchronous instruments: the
// callback reads getOverlay(), which asserts overlay_ is non-null. Arming
// it any earlier would let a reader tick fire before the overlay exists,
// and an assert is not caught by the try block below.
jqTransOverflowObservable_ = meter_->CreateInt64ObservableCounter(
"jq_trans_overflow_total", "Total job queue transaction overflows");
jqTransOverflowObservable_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
try
{
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(static_cast<int64_t>(self->app_.getOverlay().getJqTransOverflow()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
void
MetricsRegistry::registerCacheHitRateGauge()
{
// --- Cache hit rate and size gauges ---
cacheHitRateGauge_ =
meter_->CreateDoubleObservableGauge("cache_metrics", "Cache hit rates and sizes");
cacheHitRateGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
// SLE cache hit rate (0.0 - 1.0).
auto sleRate = app.getCachedSLEs().rate();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(sleRate, {{"metric", "SLE_hit_rate"}});
// Ledger cache hit rate.
// TaggedCache::getHitRate() returns 0-100; normalize to
// 0.0-1.0 so the Grafana panel using "percentunit" renders
// correctly.
auto ledgerRate = app.getLedgerMaster().getCacheHitRate() / 100.0;
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(ledgerRate, {{"metric", "ledger_hit_rate"}});
// AcceptedLedger cache hit rate (also 0-100 from
// TaggedCache; normalize to 0.0-1.0).
auto alRate = app.getAcceptedLedgerCache().getHitRate() / 100.0;
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(alRate, {{"metric", "AL_hit_rate"}});
// TreeNode cache size.
auto tnCacheSize = app.getNodeFamily().getTreeNodeCache()->getCacheSize();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(
static_cast<double>(tnCacheSize), {{"metric", "treenode_cache_size"}});
// TreeNode track size.
auto tnTrackSize = app.getNodeFamily().getTreeNodeCache()->getTrackSize();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(
static_cast<double>(tnTrackSize), {{"metric", "treenode_track_size"}});
// FullBelow cache size.
auto fbSize = app.getNodeFamily().getFullBelowCache()->size();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(static_cast<double>(fbSize), {{"metric", "fullbelow_size"}});
// AcceptedLedger cache size (entry count).
auto alSize = app.getAcceptedLedgerCache().size();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(static_cast<double>(alSize), {{"metric", "AL_size"}});
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerTxqGauge()
{
// --- TxQ metrics gauges ---
txqGauge_ = meter_->CreateDoubleObservableGauge("txq_metrics", "Transaction queue metrics");
txqGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto const metrics = app.getTxQ().getMetrics(*app.getOpenLedger().current());
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
observe("txq_count", static_cast<double>(metrics.txCount));
observe(
"txq_max_size",
metrics.txQMaxSize ? static_cast<double>(*metrics.txQMaxSize) : 0.0);
observe("txq_in_ledger", static_cast<double>(metrics.txInLedger));
observe("txq_per_ledger", static_cast<double>(metrics.txPerLedger));
observe(
"txq_reference_fee_level",
static_cast<double>(metrics.referenceFeeLevel.fee()));
observe(
"txq_min_processing_fee_level",
static_cast<double>(metrics.minProcessingFeeLevel.fee()));
observe("txq_med_fee_level", static_cast<double>(metrics.medFeeLevel.fee()));
observe(
"txq_open_ledger_fee_level",
static_cast<double>(metrics.openLedgerFeeLevel.fee()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if TxQ or OpenLedger are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerObjectCountGauge()
{
// --- Counted object instance gauges ---
objectCountGauge_ = meter_->CreateInt64ObservableGauge(
"object_count", "Live instance counts for key internal object types");
objectCountGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
try
{
// Iterate through all CountedObject types via the linked
// list in CountedObjects. We report all types with count
// > 0, filtering to the key types of interest.
auto counts = CountedObjects::getInstance().getCounts(0);
for (auto const& [name, count] : counts)
{
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(static_cast<int64_t>(count), {{"type", name}});
}
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
void
MetricsRegistry::registerLoadFactorGauge()
{
// --- Load factor breakdown gauges ---
loadFactorGauge_ =
meter_->CreateDoubleObservableGauge("load_factor_metrics", "Fee load factor breakdown");
loadFactorGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto& feeTrack = app.getFeeTrack();
auto const loadBase = static_cast<double>(feeTrack.getLoadBase());
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
// Combined load factor (server component).
observe(
"load_factor_server", static_cast<double>(feeTrack.getLoadFactor()) / loadBase);
// Individual factor components.
observe(
"load_factor_local", static_cast<double>(feeTrack.getLocalFee()) / loadBase);
observe("load_factor_net", static_cast<double>(feeTrack.getRemoteFee()) / loadBase);
observe(
"load_factor_cluster",
static_cast<double>(feeTrack.getClusterFee()) / loadBase);
// Fee escalation factors from TxQ.
auto const metrics = app.getTxQ().getMetrics(*app.getOpenLedger().current());
auto refLevel = static_cast<double>(metrics.referenceFeeLevel.fee());
if (refLevel > 0)
{
observe(
"load_factor_fee_escalation",
static_cast<double>(metrics.openLedgerFeeLevel.fee()) / refLevel);
observe(
"load_factor_fee_queue",
static_cast<double>(metrics.minProcessingFeeLevel.fee()) / refLevel);
}
// Combined load factor (max of server and fee escalation).
auto const loadFactorServer = feeTrack.getLoadFactor();
auto const loadBaseServer = feeTrack.getLoadBase();
double combined = static_cast<double>(loadFactorServer) / loadBase;
if (refLevel > 0)
{
double const feeEscalation =
static_cast<double>(metrics.openLedgerFeeLevel.fee()) * loadBaseServer /
refLevel;
if (feeEscalation > static_cast<double>(loadFactorServer))
{
combined = feeEscalation / loadBase;
}
}
observe("load_factor", combined);
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::observeNodeStoreTotals(node_store::Database& db, ObserveFn const& observe)
{
// Cumulative counters (monotonically increasing).
observe("node_reads_total", static_cast<std::int64_t>(db.getFetchTotalCount()));
observe("node_reads_hit", static_cast<std::int64_t>(db.getFetchHitCount()));
observe("node_writes", static_cast<std::int64_t>(db.getStoreCount()));
observe("node_written_bytes", static_cast<std::int64_t>(db.getStoreSize()));
observe("node_read_bytes", static_cast<std::int64_t>(db.getFetchSize()));
// Cumulative I/O durations, read straight off the atomics.
observe("node_reads_duration_us", static_cast<std::int64_t>(db.getFetchDurationUs()));
observe("node_writes_duration_us", static_cast<std::int64_t>(db.getStoreDurationUs()));
// Mean latencies. A cumulative total cannot separate "every read took
// 9 us" from "most took 2 and a few took 900", and the second is the
// cold-store signature: the hit rate reads the same either way, only the
// latency differs. Each mean is omitted rather than reported as zero
// when nothing has been read or written, so a dashboard shows a gap
// instead of a plausible wrong number.
if (auto const mean = scaledMean(db.getFetchDurationUs(), db.getFetchTotalCount()))
observe("read_mean_us", *mean);
if (auto const mean = scaledMean(db.getStoreDurationUs(), db.getStoreCount()))
observe("write_mean_us", *mean);
// Write load score (instantaneous).
observe("write_load", static_cast<std::int64_t>(db.getWriteLoad()));
}
void
MetricsRegistry::observeWritePathDetail(node_store::Database const& db, ObserveFn const& observe)
{
auto const ws = db.getWriteStats();
if (!ws)
return;
observe("nudb_writers_in_flight", static_cast<std::int64_t>(ws->concurrentWriters));
observe("nudb_insert_max_us", static_cast<std::int64_t>(ws->insertMaxUs));
if (auto const mean = scaledMean(ws->insertTotalUs, ws->insertCount))
observe("nudb_insert_mean_us", *mean);
// Mean writer depth times 100. NuDB serializes inserts behind one
// mutex, so this depth is the queue length at that mutex and sits just
// above 1.0 even under load. An integral gauge would truncate that to 1
// and lose the whole signal, hence the fixed-point scale -- which the
// name states, so nobody reads 140 as 140 writers.
if (auto const mean = scaledMean(ws->depthSum, ws->depthSamples, 100))
observe("nudb_writer_depth_x100", *mean);
}
void
MetricsRegistry::observeAcquireStats(AcquireStats const& stats, ObserveFn const& observe)
{
// Published unconditionally: for a counter, zero is the meaningful
// "no such event yet" reading, unlike for a mean. The diagnostic value
// is in the pairs -- deferrals rising while timeouts stay flat means the
// give-up path cannot fire, so an acquisition never ends.
observe("acquire_deferrals", static_cast<std::int64_t>(stats.getDeferrals()));
observe("acquire_timeouts", static_cast<std::int64_t>(stats.getTimeouts()));
// The same two events, narrowed to ledger acquisition. The pair above
// sums every TimeoutCounter subclass, so a busy replay lane can imitate
// a stalled ledger acquisition; compare these two instead when asking
// whether ledger acquisition's give-up path is advancing.
observe("acquire_ledger_deferrals", static_cast<std::int64_t>(stats.getLedgerDeferrals()));
observe("acquire_ledger_timeouts", static_cast<std::int64_t>(stats.getLedgerTimeouts()));
observe("acquire_give_ups", static_cast<std::int64_t>(stats.getGiveUps()));
observe("acquire_aborts", static_cast<std::int64_t>(stats.getAborts()));
observe("acquire_aborts_partial", static_cast<std::int64_t>(stats.getAbortsWithPartialWork()));
observe("acquire_completions", static_cast<std::int64_t>(stats.getCompletions()));
observe("acquire_sweep_evictions", static_cast<std::int64_t>(stats.getSweepEvictions()));
}
void
MetricsRegistry::observeReadQueue(node_store::Database& db, ObserveFn const& observe)
{
json::Value obj(json::ValueType::Object);
db.getCountsJson(obj);
if (obj.isMember("read_queue"))
observe("read_queue", static_cast<std::int64_t>(obj["read_queue"].asUInt()));
// Read thread pool stats (native JSON ints, no jss:: constants).
for (auto const* key : {"read_request_bundle", "read_threads_running", "read_threads_total"})
{
if (obj.isMember(key))
observe(key, static_cast<std::int64_t>(obj[key].asInt()));
}
}
void
MetricsRegistry::registerNodeStoreGauge()
{
// --- NodeStore I/O gauges ---
// The cumulative counters (reads, writes, bytes) are also exposed here
// as observable gauges. This avoids adding an xrpld dependency into the
// libxrpl nodestore code — the MetricsRegistry reads the existing atomic
// counters from Database via its public accessors.
//
// Every value multiplexes onto this one gauge through its `metric`
// label, so a new value needs no new instrument. The body is split
// across four helpers, one per domain, to stay inside the per-function
// line budget and to keep each domain testable on its own.
nodeStoreGauge_ = meter_->CreateInt64ObservableGauge(
"nodestore_state",
"NodeStore I/O counters, latencies, write-queue depth and acquisition stalls");
nodeStoreGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto& db = app.getNodeStore();
ObserveFn const observe = [&](char const* name, std::int64_t value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(value, {{"metric", name}});
};
// Qualified because the enclosing lambda captures nothing:
// these are static members, and the explicit scope says so.
MetricsRegistry::observeNodeStoreTotals(db, observe);
MetricsRegistry::observeWritePathDetail(db, observe);
MetricsRegistry::observeAcquireStats(app.getAcquireStats(), observe);
MetricsRegistry::observeReadQueue(db, observe);
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
void
MetricsRegistry::registerServerInfoGauge()
{
// --- Server info gauges ---
serverInfoGauge_ =
meter_->CreateInt64ObservableGauge("server_info", "Server-level health metrics");
serverInfoGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, int64_t value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(value, {{"metric", name}});
};
// Server operating mode (DISCONNECTED=0 .. FULL=4).
observe("server_state", static_cast<int64_t>(app.getOPs().getOperatingMode()));
// Uptime in seconds since server start.
observe(
"uptime", static_cast<int64_t>(UptimeClock::now().time_since_epoch().count()));
// Total peer count (inbound + outbound).
observe("peers", static_cast<int64_t>(app.getOverlay().size()));
// Validated ledger sequence (0 if none yet).
observe(
"validated_ledger_seq",
static_cast<int64_t>(app.getLedgerMaster().getValidLedgerIndex()));
// Current open ledger sequence.
observe(
"ledger_current_index",
static_cast<int64_t>(app.getLedgerMaster().getCurrentLedgerIndex()));
// Cumulative resource-related peer disconnects.
observe(
"peer_disconnects_resources",
static_cast<int64_t>(app.getOverlay().getPeerDisconnectCharges()));
// Last consensus round data (from JSON — only public API).
auto const consensusInfo = app.getOPs().getConsensusInfo();
if (consensusInfo.isMember("previous_proposers"))
{
observe(
"last_close_proposers",
static_cast<int64_t>(consensusInfo["previous_proposers"].asUInt()));
}
if (consensusInfo.isMember("previous_mseconds"))
{
observe(
"last_close_converge_time_ms",
static_cast<int64_t>(consensusInfo["previous_mseconds"].asUInt()));
}
// Network close time of the last closed ledger, as NetClock
// seconds since the XRPL epoch (2000-01-01). Unlike a span
// timestamp, a gauge value survives as a queryable time series,
// so dashboards can show last-close age (staleness) via
// now - value. The close interval comes from the
// ledgers_closed_total counter, not a delta of this gauge
// (a timestamp gauge's delta aliases to the scrape period).
// Skip until a ledger has closed.
if (auto const closed = app.getLedgerMaster().getClosedLedger())
{
observe(
"last_close_time",
static_cast<int64_t>(
closed->header().closeTime.time_since_epoch().count()));
}
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerBuildInfoGauge()
{
// --- Build info gauge ---
buildInfoGauge_ = meter_->CreateInt64ObservableGauge("build_info", "Build version information");
buildInfoGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* /* state */) {
try
{
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(1, {{"version", std::string(build_info::getVersionString())}});
}
catch (...) // NOLINT(bugprone-empty-catch)
{
}
},
nullptr);
}
void
MetricsRegistry::registerCompleteLedgersGauge()
{
// --- Complete ledgers range gauge ---
completeLedgersGauge_ = meter_->CreateInt64ObservableGauge(
"complete_ledgers", "Complete ledger range start/end pairs");
completeLedgersGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto const rangeStr = app.getLedgerMaster().getCompleteLedgers();
if (rangeStr.empty() || rangeStr == "empty")
return;
// Parse comma-separated ranges like
// "32570-50000,50005-75891421". A range of one ledger arrives
// as a bare sequence number, so parseLedgerRange() decides what
// a segment is; only genuinely unreadable ones are skipped.
std::size_t rangeIndex = 0;
std::istringstream stream(rangeStr);
std::string segment;
while (std::getline(stream, segment, ','))
{
auto const range = MetricsRegistry::parseLedgerRange(segment);
if (!range)
continue;
auto const idxStr = std::to_string(rangeIndex);
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(
static_cast<int64_t>(range->first),
{{"bound", "start"}, {"index", idxStr}});
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(
static_cast<int64_t>(range->second),
{{"bound", "end"}, {"index", idxStr}});
++rangeIndex;
}
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on parse error or if services not ready.
}
},
this);
}
void
MetricsRegistry::registerDbMetricsGauge()
{
// --- Database size and fetch rate gauges ---
dbMetricsGauge_ =
meter_->CreateInt64ObservableGauge("db_metrics", "Database storage sizes and fetch rates");
dbMetricsGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, int64_t value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(value, {{"metric", name}});
};
auto& rdb = app.getRelationalDatabase();
observe("db_kb_total", static_cast<int64_t>(rdb.getKBUsedAll()));
observe("db_kb_ledger", static_cast<int64_t>(rdb.getKBUsedLedger()));
observe("db_kb_transaction", static_cast<int64_t>(rdb.getKBUsedTransaction()));
// Historical ledger fetches per minute.
observe(
"historical_perminute",
static_cast<int64_t>(app.getInboundLedgers().fetchRate()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerValidatorHealthGauge()
{
// --- Validator health gauges ---
validatorHealthGauge_ =
meter_->CreateDoubleObservableGauge("validator_health", "Validator health indicators");
validatorHealthGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
observe("amendment_blocked", app.getOPs().isAmendmentBlocked() ? 1.0 : 0.0);
observe("unl_blocked", app.getOPs().isUNLBlocked() ? 1.0 : 0.0);
observe("validation_quorum", static_cast<double>(app.getValidators().quorum()));
// Days until UNL list expiry (-1 if no expiry known).
auto const expiry = app.getValidators().expires();
if (expiry)
{
auto const now = app.getTimeKeeper().closeTime();
auto const diffHours =
std::chrono::duration_cast<std::chrono::hours>(*expiry - now).count();
observe("unl_expiry_days", static_cast<double>(diffHours) / 24.0);
}
else
{
observe("unl_expiry_days", -1.0);
}
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerPeerQualityGauge()
{
// --- Peer quality gauges ---
// Uses Peer::json() to read latency and version since those accessors
// are not on the abstract Peer interface (they live on PeerImp).
peerQualityGauge_ =
meter_->CreateDoubleObservableGauge("peer_quality", "Peer network quality metrics");
peerQualityGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
// Collect latencies, version info, and tracking state from
// each peer's JSON.
std::vector<int> latencies;
int higherVersionCount = 0;
int totalPeers = 0;
int divergedCount = 0;
// Encode a version string into BuildInfo's comparable numeric
// form. Peers report the full "xrpld-3.3.0-b0" string while
// our baseline is the bare "3.3.0-b0", and SemanticVersion
// requires a leading digit, so strip any non-digit prefix
// first. Numeric encoding avoids the lexicographic bug where
// "2.3.0" > "2.10.0" and "xrpld-..." > "3...".
auto const encodeVersion = [](std::string_view v) -> std::uint64_t {
auto const firstDigit = v.find_first_of("0123456789");
if (firstDigit == std::string_view::npos)
return 0;
return build_info::encodeSoftwareVersion(v.substr(firstDigit));
};
auto const ownEncoded = encodeVersion(build_info::getVersionString());
app.getOverlay().foreach([&](std::shared_ptr<Peer> const& peer) {
++totalPeers;
auto const pj = peer->json();
if (pj.isMember(jss::latency))
{
latencies.push_back(pj[jss::latency].asInt());
}
if (pj.isMember(jss::version))
{
// Unparseable peer versions encode below ownEncoded, so
// the comparison correctly leaves them uncounted.
auto const pv = pj[jss::version].asString();
if (encodeVersion(pv) > ownEncoded)
++higherVersionCount;
}
// PeerImp::json() sets "track" to "diverged" when the peer's
// tracking state is Tracking::Diverged (i.e. it is following
// a different ledger chain than us).
if (pj.isMember(jss::track) && pj[jss::track].asString() == "diverged")
++divergedCount;
});
// P90 latency across connected peers.
if (!latencies.empty())
{
std::ranges::sort(latencies);
auto p90idx = static_cast<std::size_t>(latencies.size() * 0.9);
if (p90idx >= latencies.size())
p90idx = latencies.size() - 1;
observe("peer_latency_p90_ms", static_cast<double>(latencies[p90idx]));
}
else
{
observe("peer_latency_p90_ms", 0.0);
}
// Percentage of peers running a higher version.
double const higherPct = totalPeers > 0
? (static_cast<double>(higherVersionCount) / totalPeers * 100.0)
: 0.0;
observe("peers_higher_version_pct", higherPct);
// Count peers diverged from our ledger chain, read from the
// peer's "track" JSON field (set by PeerImp::json()). Diverged
// peers are following a different chain and are a leading
// indicator of local sync trouble.
observe("peers_insane_count", static_cast<double>(divergedCount));
// Binary flag: recommend upgrade if >60% run a newer version.
observe("upgrade_recommended", higherPct > 60.0 ? 1.0 : 0.0);
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerReduceRelayGauge()
{
// Transaction reduce-relay efficiency. Overlay::txMetrics() exposes the
// rolling averages as a JSON object with string values (std::to_string),
// so parse each field. A high suppressed:selected ratio proves the
// feature is saving bandwidth; a high not_enabled count means stale peers
// force full relay.
reduceRelayGauge_ = meter_->CreateInt64ObservableGauge(
"reduce_relay_metrics", "Transaction reduce-relay efficiency metrics");
reduceRelayGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto const tm = app.getOverlay().txMetrics();
auto observe = [&](char const* name, int64_t value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(value, {{"metric", name}});
};
// Each field is a decimal string; emit when present and parseable.
auto observeField = [&](auto const& field, char const* name) {
if (tm.isMember(field))
{
auto const s = tm[field].asString();
if (!s.empty())
observe(name, static_cast<int64_t>(std::stoll(s)));
}
};
observeField(jss::txr_selected_cnt, "selected_peers");
observeField(jss::txr_suppressed_cnt, "suppressed_peers");
observeField(jss::txr_not_enabled_cnt, "not_enabled_peers");
observeField(jss::txr_missing_tx_freq, "missing_tx_freq");
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready or a value is
// not parseable.
}
},
this);
}
void
MetricsRegistry::registerLedgerEconomyGauge()
{
// --- Ledger economy gauges ---
ledgerEconomyGauge_ =
meter_->CreateDoubleObservableGauge("ledger_economy", "Ledger fee and economy metrics");
ledgerEconomyGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
// Fee and reserve values from the validated ledger.
auto const ledger = app.getLedgerMaster().getValidatedLedger();
if (ledger)
{
auto const& fees = ledger->fees();
// Cost of a reference transaction (drops).
observe("base_fee_xrp", static_cast<double>(fees.base.drops()));
// Base reserve = one account, zero owned objects:
// accountReserve(ownerCount=0, accountCount=1) == reserve.
observe(
"reserve_base_xrp", static_cast<double>(fees.accountReserve(0, 1).drops()));
observe("reserve_inc_xrp", static_cast<double>(fees.increment.drops()));
}
// Seconds since the last validated ledger closed.
auto const age = app.getLedgerMaster().getValidatedLedgerAge();
observe("ledger_age_seconds", static_cast<double>(age.count()));
// Transaction rate from the open ledger's tx count.
// OpenView::txCount() tracks transactions in the current
// open ledger; dividing by the ledger age gives an
// approximate rate.
auto const& openLedger = app.getOpenLedger();
auto const txInLedger = openLedger.current()->txCount();
auto const ageVal = age.count();
if (ageVal > 0)
{
observe(
"transaction_rate",
static_cast<double>(txInLedger) / static_cast<double>(ageVal));
}
else
{
observe("transaction_rate", 0.0);
}
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerStateTrackingGauge()
{
// --- State tracking gauges ---
stateTrackingGauge_ =
meter_->CreateDoubleObservableGauge("state_tracking", "Node state and mode tracking");
stateTrackingGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
// State value: 0-4 from OperatingMode, 5=validating, 6=proposing.
auto const mode = app.getOPs().getOperatingMode();
auto stateValue = static_cast<double>(std::to_underlying(mode));
// If FULL, refine using consensus info for validating/proposing.
if (mode == OperatingMode::FULL)
{
auto const info = app.getOPs().getConsensusInfo();
if (info.isMember("proposing") && info["proposing"].asBool())
{
stateValue = 6.0;
}
else if (info.isMember("validating") && info["validating"].asBool())
{
stateValue = 5.0;
}
}
observe("state_value", stateValue);
// Time spent in the current operating mode, sourced from
// NetworkOPs' StateAccounting via a lightweight accessor.
auto const stateDurUs = app.getOPs().getServerStateDurationUs();
observe(
"time_in_current_state_seconds", static_cast<double>(stateDurUs.count()) / 1e6);
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip if services are not yet ready.
}
},
this);
}
void
MetricsRegistry::registerStorageDetailGauge()
{
// --- Storage detail gauges ---
// Reports the cumulative payload bytes handed to the NodeStore. See the
// note at the observe() call below: this is logical bytes stored, not
// on-disk file size, because no accessor for the latter exists. The label
// value names it that way so it is not read as a filesystem measurement.
storageDetailGauge_ =
meter_->CreateInt64ObservableGauge("storage_detail", "Storage detail metrics");
storageDetailGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
auto& app = self->app_;
try
{
auto observe = [&](char const* name, int64_t value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(value, {{"metric", name}});
};
// Cumulative payload bytes handed to the NodeStore. This is
// not an on-disk file size: getStoreSize() sums the object
// payloads this process has written, so it excludes NuDB's
// keys, bucket padding and log, and it resets with the
// process while the files do not. The value comes from
// Database, not from any backend, so it carries no nudb_
// prefix -- it reads the same on RocksDB.
//
// This is the same call node_written_bytes makes on the
// nodestore_state gauge, so the two series are equal by
// construction and their ratio is a constant 1.0. It is not
// a write-amplification measure. Backend exposes no file
// size accessor, so there is nothing better to read here;
// computing one would mean stat()ing the backend's files
// from the reader thread, which needs a new Backend method
// rather than a change at this call site.
observe(
"stored_object_bytes", static_cast<int64_t>(app.getNodeStore().getStoreSize()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
void
MetricsRegistry::registerValidationAgreementGauge()
{
// --- Validation agreement gauges ---
// Reports rolling-window agreement percentages and counts from
// ValidationTracker. reconcile() is called at the start of the
// callback so that pending ledger events are resolved before the
// window data is read (the callback fires every ~10 s from the
// PeriodicExportingMetricReader thread).
validationAgreementGauge_ = meter_->CreateDoubleObservableGauge(
"validation_agreement", "Validation agreement percentages and counts (1h/24h windows)");
validationAgreementGauge_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
try
{
// Reconcile pending events before reading window data.
self->validationTracker_.reconcile();
auto observe = [&](char const* name, double value) {
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<double>>>(result)
->Observe(value, {{"metric", name}});
};
observe("agreement_pct_1h", self->validationTracker_.agreementPct1h());
observe("agreement_pct_24h", self->validationTracker_.agreementPct24h());
observe(
"agreements_1h", static_cast<double>(self->validationTracker_.agreements1h()));
observe("missed_1h", static_cast<double>(self->validationTracker_.missed1h()));
observe(
"agreements_24h",
static_cast<double>(self->validationTracker_.agreements24h()));
observe("missed_24h", static_cast<double>(self->validationTracker_.missed24h()));
// 7-day window (matches external xrpl-validator-dashboard).
observe("agreement_pct_7d", self->validationTracker_.agreementPct7d());
observe(
"agreements_7d", static_cast<double>(self->validationTracker_.agreements7d()));
observe("missed_7d", static_cast<double>(self->validationTracker_.missed7d()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
void
MetricsRegistry::registerValidationTotalsCounters()
{
// Lifetime validation agreement/miss counters.
//
// These are monotonic ObservableCounters (not the sync Counters they used
// to be): a Prometheus _total must never decrease, but ValidationTracker's
// NET totals are non-monotonic (a late repair decrements the net miss
// count). We therefore observe the tracker's GROSS lifetime tallies, which
// count each ledger once at first classification and are never adjusted on
// repair (initial-classification semantics — see ValidationTracker). The
// repaired/agreement view remains available from validation_agreement.
//
// reconcile() is called first so pending events are resolved before the
// tallies are read; the callback fires every ~10 s from the
// PeriodicExportingMetricReader thread.
validationAgreementsObservable_ = meter_->CreateInt64ObservableCounter(
"validation_agreements_total",
"Lifetime validations that initially agreed with network consensus");
validationAgreementsObservable_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
try
{
self->validationTracker_.reconcile();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(static_cast<int64_t>(self->validationTracker_.totalAgreementsEver()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
validationMissedObservable_ = meter_->CreateInt64ObservableCounter(
"validation_missed_total", "Lifetime validations that initially missed network consensus");
validationMissedObservable_->AddCallback(
[](opentelemetry::metrics::ObserverResult result, void* state) {
auto* self = static_cast<MetricsRegistry*>(state);
if (self->callbacksDetached_.load(std::memory_order_acquire))
return;
try
{
self->validationTracker_.reconcile();
opentelemetry::nostd::get<opentelemetry::nostd::shared_ptr<
opentelemetry::metrics::ObserverResultT<int64_t>>>(result)
->Observe(static_cast<int64_t>(self->validationTracker_.totalMissedEver()));
}
catch (...) // NOLINT(bugprone-empty-catch)
{
// Silently skip on error.
}
},
this);
}
#endif // XRPL_ENABLE_TELEMETRY
// -----------------------------------------------------------------
// External dashboard parity counter increments
// -----------------------------------------------------------------
void
MetricsRegistry::incrementLedgersClosed()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && ledgersClosedCounter_)
ledgersClosedCounter_->Add(1);
#endif
}
void
MetricsRegistry::incrementValidationsSent()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && validationsSentCounter_)
validationsSentCounter_->Add(1);
#endif
}
void
MetricsRegistry::incrementValidationsChecked()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && validationsCheckedCounter_)
validationsCheckedCounter_->Add(1);
#endif
}
void
MetricsRegistry::incrementStateChanges()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && stateChangesCounter_)
stateChangesCounter_->Add(1);
#endif
}
void
MetricsRegistry::incrementLedgerHistoryMismatch(std::string_view reason)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && ledgerHistoryMismatchCounter_)
ledgerHistoryMismatchCounter_->Add(1, {{"reason", std::string(reason)}});
#endif
}
void
MetricsRegistry::incrementTxqExpired()
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && txqExpiredCounter_)
txqExpiredCounter_->Add(1);
#endif
}
void
MetricsRegistry::incrementTxqDropped(std::string_view reason)
{
#ifdef XRPL_ENABLE_TELEMETRY
if (enabled_ && txqDroppedCounter_)
txqDroppedCounter_->Add(1, {{"reason", std::string(reason)}});
#endif
}
} // namespace xrpl::telemetry
Reference in New Issue View Git Blame Copy Permalink