ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-07-30 18:40:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(telemetry): observe the write queue, read latency and stalls

Browse Source 
Adds the derived read and write means, the NuDB writer depth and insert
timings, and the seven acquisition counters to the existing
nodestore_state gauge. Every value multiplexes onto that one instrument
through its `metric` label, so no new instrument is created.

Means are omitted rather than reported as zero when their denominator is
zero, so a dashboard shows a gap instead of a plausible wrong number. All
four go through one new scaledMean() helper so the guard cannot be
forgotten at a future call site; it also saturates instead of wrapping,
because a wrapped gauge reads as a healthy-looking dip. A zero total over
real samples still reports zero, since a store fast enough to truncate
every sample must not look dead.

The NuDB write-path block is skipped entirely when getWriteStats() is
nullopt, which is every backend but NuDB, so absent labels distinguish
"not measured" from "measured, and idle". Writer depth is scaled by 100
and named accordingly, because it sits just above 1.0 and an integral
gauge would truncate the whole signal away.

The gauge callback body is split into four static helpers to stay inside
the per-function line budget and to make each domain testable with a
recording sink.

Also corrects nudb_bytes, which called getStoreSize() exactly as
node_written_bytes does, so the obvious write-amplification ratio was a
constant 1.0 and the old "on-disk size" comment was wrong. No file-size
accessor exists on Backend or Database, so the value is unchanged and the
comment now states what it really is.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Pratik Mankawde
2026-07-27 19:46:48 +01:00
parent c06a6586a9
commit 2f5f0944ab
4 changed files with 445 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/tests/libxrpl/nodestore/Database.cpp
View File

@@ -8,6 +8,7 @@
#include <xrpl/nodestore/DummyScheduler.h>
#include <xrpl/nodestore/Manager.h>
#include <xrpl/nodestore/Types.h>
#include <xrpl/nodestore/WriteStats.h>
#include <xrpl/protocol/SystemParameters.h>
#include <gtest/gtest.h>
@@ -142,6 +143,53 @@ TEST_P(NodeStoreDatabaseTest, fetch_missing)
fetchMissing(*db, batch_);
}
// Database::getWriteStats() forwards to the backend, and the telemetry
// exporter branches on the optional to decide whether to publish the NuDB
// write-queue labels at all. Backend.cpp covers the backend's own answer;
// this covers the forwarding, which is what the exporter actually calls.
TEST_P(NodeStoreDatabaseTest, write_stats_forwarded_from_backend)
{
auto db = makeDatabase();
// Before any write, only a measuring backend answers at all.
auto const initial = db->getWriteStats();
ASSERT_EQ(initial.has_value(), GetParam() == "nudb")
<< "only nudb measures its write path; backend=" << GetParam();
if (!initial)
{
// Negative path: a non-measuring backend must keep reporting absence
// after real writes too, so the exporter omits the labels rather
// than publishing zeros that would read as an idle write path.
storeBatch(*db, batch_);
EXPECT_FALSE(db->getWriteStats().has_value());
return;
}
// A freshly opened store has written nothing.
EXPECT_EQ(initial->insertCount, 0u);
EXPECT_EQ(initial->insertTotalUs, 0u);
EXPECT_EQ(initial->depthSum, 0u);
EXPECT_EQ(initial->concurrentWriters, 0u);
storeBatch(*db, batch_);
// Every object stored through the Database reaches the backend exactly
// once, so the forwarded count is the batch size and not, say, the
// number of store() batches.
auto const after = db->getWriteStats();
ASSERT_TRUE(after.has_value());
EXPECT_EQ(after->insertCount, batch_.size());
// One writer thread, so the depth recorded at each insert is exactly 1.
EXPECT_EQ(after->depthSum, batch_.size());
// No writer is left in flight once the calls have returned.
EXPECT_EQ(after->concurrentWriters, 0u);
EXPECT_GT(after->insertTotalUs, 0u);
// A maximum is never below the mean, which fails if the field held the
// minimum or the first sample instead of a running maximum.
EXPECT_GE(after->insertMaxUs * after->insertCount, after->insertTotalUs);
}
INSTANTIATE_TEST_SUITE_P(
NodeStoreBackends,
NodeStoreDatabaseTest,

136
src/tests/libxrpl/telemetry/MetricsRegistry.cpp
View File

@@ -1,7 +1,7 @@
/**
* GTest unit tests for MetricsRegistry.
*
* Two independent groups, split by what they can link:
* Three independent groups, split by what they can link:
*
* 1. sanitiseHandler() — the `handler` label sanitiser. Runs in **both**
* builds. sanitiseHandler() is a public static constexpr defined inline
@@ -10,7 +10,11 @@
* inside it would silently compile them out of the telemetry-enabled
* build, which is the build that actually exports the label.
*
* 2. The no-op / telemetry-disabled path — construction, start()/stop()
* 2. scaledMean() — the guarded-division helper behind every derived mean
* on the nodestore_state gauge. Also a public static constexpr inline,
* so it runs in both builds for the same reason.
*
* 3. The no-op / telemetry-disabled path — construction, start()/stop()
* lifecycle, and the synchronous record*() methods. Guarded, because
* when XRPL_ENABLE_TELEMETRY is defined MetricsRegistry.cpp is not
* compiled into this binary (see src/tests/libxrpl/CMakeLists.txt) and
@@ -23,6 +27,8 @@
#include <array>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <set>
#include <string_view>
@@ -287,6 +293,132 @@ TEST(MetricsRegistrySanitiseHandler, output_domain_is_exactly_44_values)
EXPECT_TRUE(domain.contains(name)) << "missing from domain: " << name;
}
// ---------------------------------------------------------------------------
// scaledMean() — the guarded division behind every derived mean published on
// the nodestore_state gauge.
//
// The property under test is not "it divides". It is that a zero denominator
// yields absence rather than a plausible zero, because a dashboard must show
// a gap instead of a number an operator would believe. Every expected value
// below is an independently computed constant, never a restatement of the
// implementation's own expression: `EXPECT_EQ(mean, total / count)` would
// pass even if both sides were wrong the same way.
// ---------------------------------------------------------------------------
namespace {
using Registry = MetricsRegistry;
// Compile-time checks first, so a regression is a build failure. Each
// expected value is written as a literal worked out by hand.
static_assert(Registry::scaledMean(500, 4) == 125); // 500/4 exactly
static_assert(Registry::scaledMean(9, 1) == 9); // single sample
static_assert(Registry::scaledMean(0, 4) == 0); // real zero mean
static_assert(Registry::scaledMean(7, 2) == 3); // 3.5 truncates
static_assert(Registry::scaledMean(7, 5, 100) == 140); // 1.4 scaled
static_assert(Registry::scaledMean(4, 4, 100) == 100); // depth exactly 1
static_assert(Registry::scaledMean(1, 3, 100) == 33); // 0.333 scaled
static_assert(!Registry::scaledMean(500, 0).has_value()); // no samples
static_assert(!Registry::scaledMean(0, 0).has_value()); // idle, not zero
static_assert(!Registry::scaledMean(500, 4, 0).has_value()); // scale 0
} // namespace
TEST(MetricsRegistryScaledMean, zero_count_reports_absence_not_zero)
{
// The whole point of the helper. A mean over no samples is undefined, and
// reporting it as 0 would draw a believable flat line at the bottom of a
// latency axis. Absence is the only honest answer.
EXPECT_FALSE(Registry::scaledMean(500, 0).has_value());
EXPECT_FALSE(Registry::scaledMean(0, 0).has_value());
EXPECT_FALSE(Registry::scaledMean(std::numeric_limits<std::uint64_t>::max(), 0).has_value());
// Cause, not just state: absence is specific to a zero denominator. The
// same numerator with one sample does produce a value, so the guard is
// keyed on the count and is not rejecting everything.
ASSERT_TRUE(Registry::scaledMean(500, 1).has_value());
EXPECT_EQ(*Registry::scaledMean(500, 1), 500);
}
TEST(MetricsRegistryScaledMean, zero_total_over_real_samples_is_a_genuine_zero)
{
// The negative counterpart of the test above, and the reason absence and
// zero must stay distinguishable: a store fast enough that every sample
// truncated to 0 us really does have a mean of 0. That must be reported,
// not suppressed, or a working fast path looks like a dead one.
auto const mean = Registry::scaledMean(0, 32);
ASSERT_TRUE(mean.has_value());
EXPECT_EQ(*mean, 0);
}
TEST(MetricsRegistryScaledMean, exact_means_are_computed_exactly)
{
// Independently computed expectations: 4800/32 is 150 by hand.
EXPECT_EQ(Registry::scaledMean(4800, 32), 150);
// A mean equal to its own total when there is one sample.
EXPECT_EQ(Registry::scaledMean(917, 1), 917);
// Truncation toward zero is the documented behaviour: 99/10 is 9.9.
EXPECT_EQ(Registry::scaledMean(99, 10), 9);
}
TEST(MetricsRegistryScaledMean, scale_recovers_the_fractional_digits)
{
// Mean writer depth is the reason `scale` exists. Unscaled, a depth of
// 1.4 truncates to 1 and is indistinguishable from an idle 1.0; the x100
// form must keep the fraction.
EXPECT_EQ(Registry::scaledMean(7, 5), 1); // the lost signal
EXPECT_EQ(Registry::scaledMean(7, 5, 100), 140); // the kept signal
// A pure fraction with no whole part must survive too. Without scaling
// the remainder this would read 0 and the metric would be useless.
EXPECT_EQ(Registry::scaledMean(1, 4, 100), 25);
EXPECT_EQ(Registry::scaledMean(3, 8, 100), 37); // 0.375 truncated
// Scaling must not invent precision where the value is already whole.
EXPECT_EQ(Registry::scaledMean(8, 4, 100), 200);
}
TEST(MetricsRegistryScaledMean, large_inputs_saturate_instead_of_wrapping)
{
constexpr auto kU64Max = std::numeric_limits<std::uint64_t>::max();
constexpr auto kI64Max = std::numeric_limits<std::int64_t>::max();
// A uint64 total that does not fit the signed gauge must clamp. Wrapping
// would surface as a sudden dip to a healthy-looking small number, which
// is the failure mode worth preventing.
auto const saturated = Registry::scaledMean(kU64Max, 1);
ASSERT_TRUE(saturated.has_value());
EXPECT_EQ(*saturated, kI64Max);
// Scaling must not overflow either: this quotient times 100 exceeds
// int64 range, so it clamps rather than wraps negative.
auto const scaled = Registry::scaledMean(kU64Max, 2, 100);
ASSERT_TRUE(scaled.has_value());
EXPECT_EQ(*scaled, kI64Max);
// Every result is non-negative; a negative latency or depth is
// meaningless and is the visible symptom of a wrap.
EXPECT_GE(*saturated, 0);
EXPECT_GE(*scaled, 0);
// Just below the boundary the value is exact, not clamped, so the clamp
// above is a real bound and not a blanket ceiling on everything.
auto const exact = Registry::scaledMean(static_cast<std::uint64_t>(kI64Max), 1);
ASSERT_TRUE(exact.has_value());
EXPECT_EQ(*exact, kI64Max);
auto const belowBoundary = Registry::scaledMean(1'000'000, 4, 100);
ASSERT_TRUE(belowBoundary.has_value());
EXPECT_EQ(*belowBoundary, 25'000'000);
}
TEST(MetricsRegistryScaledMean, default_scale_is_one)
{
// The two-argument form is the latency case and must not scale silently;
// if the default were 100 every published latency would be 100x wrong.
EXPECT_EQ(Registry::scaledMean(360, 8), Registry::scaledMean(360, 8, 1));
EXPECT_EQ(Registry::scaledMean(360, 8), 45);
}
// When telemetry is globally enabled, MetricsRegistry.cpp requires xrpld
// link dependencies we cannot satisfy in a standalone GTest binary.
#ifndef XRPL_ENABLE_TELEMETRY

164
src/xrpld/telemetry/MetricsRegistry.cpp
View File

@@ -24,6 +24,7 @@
#ifdef XRPL_ENABLE_TELEMETRY
#include <xrpld/app/ledger/AcquireStats.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/OpenLedger.h>
@@ -771,6 +772,90 @@ MetricsRegistry::registerLoadFactorGauge()
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->insertCount, 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()));
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()
{
@@ -779,8 +864,14 @@ MetricsRegistry::registerNodeStoreGauge()
// 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, queue depth, and write load");
"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);
@@ -792,55 +883,18 @@ MetricsRegistry::registerNodeStoreGauge()
{
auto& db = app.getNodeStore();
auto observe = [&](char const* name, int64_t value) {
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}});
};
// Cumulative counters (monotonically increasing).
observe("node_reads_total", static_cast<int64_t>(db.getFetchTotalCount()));
observe("node_reads_hit", static_cast<int64_t>(db.getFetchHitCount()));
observe("node_writes", static_cast<int64_t>(db.getStoreCount()));
observe("node_written_bytes", static_cast<int64_t>(db.getStoreSize()));
observe("node_read_bytes", static_cast<int64_t>(db.getFetchSize()));
// Cumulative I/O durations, read straight off the atomics.
observe("node_reads_duration_us", static_cast<int64_t>(db.getFetchDurationUs()));
observe("node_writes_duration_us", static_cast<int64_t>(db.getStoreDurationUs()));
// Write load score (instantaneous).
observe("write_load", static_cast<int64_t>(db.getWriteLoad()));
// Read queue depth (instantaneous). The JSON object is still
// needed for the queue and thread-pool values, which have no
// accessors.
json::Value obj(json::ValueType::Object);
db.getCountsJson(obj);
if (obj.isMember("read_queue"))
{
observe("read_queue", static_cast<int64_t>(obj["read_queue"].asUInt()));
}
// Read thread pool stats (native JSON ints, no jss:: constants).
if (obj.isMember("read_request_bundle"))
{
observe(
"read_request_bundle",
static_cast<int64_t>(obj["read_request_bundle"].asInt()));
}
if (obj.isMember("read_threads_running"))
{
observe(
"read_threads_running",
static_cast<int64_t>(obj["read_threads_running"].asInt()));
}
if (obj.isMember("read_threads_total"))
{
observe(
"read_threads_total",
static_cast<int64_t>(obj["read_threads_total"].asInt()));
}
// 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)
{
@@ -1362,7 +1416,9 @@ void
MetricsRegistry::registerStorageDetailGauge()
{
// --- Task 7.13: Storage detail gauges ---
// Reports NuDB on-disk size via the NodeStore JSON counters interface.
// 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.
storageDetailGauge_ =
meter_->CreateInt64ObservableGauge("storage_detail", "Storage detail metrics");
storageDetailGauge_->AddCallback(
@@ -1380,8 +1436,20 @@ MetricsRegistry::registerStorageDetailGauge()
->Observe(value, {{"metric", name}});
};
// NuDB on-disk size reported by the NodeStore backend.
// getStoreSize() returns the total bytes stored.
// Cumulative payload bytes handed to the NodeStore -- NOT
// on-disk file size, despite the name. 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.
//
// 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("nudb_bytes", static_cast<int64_t>(app.getNodeStore().getStoreSize()));
}
catch (...) // NOLINT(bugprone-empty-catch)

149
src/xrpld/telemetry/MetricsRegistry.h
View File

@@ -49,7 +49,8 @@
* +-- TxQ metrics
* +-- CountedObject counts
* +-- Load factor breakdown
* +-- NodeStore I/O gauges
* +-- NodeStore I/O gauges (totals, derived means, NuDB write queue,
* ledger-acquisition stall counters)
* +-- Server info (state, uptime, peers, consensus)
* +-- Build info (version label)
* +-- Complete ledger ranges (start/end pairs)
@@ -138,7 +139,10 @@
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <functional>
#include <limits>
#include <memory>
#include <optional>
#include <ranges>
#include <string>
#include <string_view>
@@ -155,6 +159,16 @@ namespace xrpl {
class ServiceRegistry;
// Defined in src/xrpld/app/ledger/AcquireStats.h. Forward-declared because
// only the gauge helpers in the .cpp touch it, and pulling an xrpld/app
// header in here would widen the dependencies of every file that includes
// this one.
class AcquireStats;
namespace node_store {
class Database;
} // namespace node_store
namespace telemetry {
/**
@@ -397,6 +411,79 @@ public:
return name;
}
/**
* Divide a cumulative total by its count, optionally scaled, reporting
* absence rather than zero when the count is zero.
*
* Every cumulative counter this registry publishes has a companion mean
* that is only defined once the counter has moved. Reporting such a mean
* as `0` is worse than not reporting it: `0` is a plausible reading, so a
* dashboard draws a flat line at the bottom of the axis and an operator
* concludes "reads are instant" when the truth is "nothing has been
* read". Returning std::nullopt makes the caller skip the observation, so
* the series has a genuine gap instead.
*
* @p scale exists because the gauge these feed is integral. A mean writer
* depth of 1.4 truncates to 1, which is indistinguishable from a healthy
* 1.0, so the caller scales by 100 and says so in the metric name.
*
* The arithmetic divides before scaling and scales the remainder
* separately, so a long-lived node cannot overflow the product. Should
* the result still exceed the gauge's range it saturates at
* INT64_MAX rather than wrapping, because a wrapped gauge reads as a
* sudden healthy-looking dip.
*
* Defined inline for the same reason as sanitiseHandler(): in a
* telemetry-enabled build MetricsRegistry.cpp is not compiled into the
* unit-test binary, so an out-of-line definition would be untestable.
* constexpr so the cases below are checked at compile time.
*
* @param total Cumulative numerator (e.g. summed microseconds).
* @param count Number of samples in @p total.
* @param scale Fixed-point multiplier applied to the quotient. Must be
* at least 1; 0 is meaningless and yields std::nullopt.
* @return The scaled mean, or std::nullopt when @p count is 0 (mean
* undefined) or @p scale is 0.
*
* @note Pure and reentrant: holds no state and performs no I/O.
* @note Truncates toward zero, like integer division. A mean of 9.9 us
* reads as 9 at @p scale 1 and as 990 at @p scale 100.
*
* Example:
* @code
* scaledMean(500, 4); // 125 -- mean microseconds
* scaledMean(7, 5, 100); // 140 -- mean 1.4, scaled by 100
* scaledMean(500, 0); // nullopt -- no samples, so no mean
* @endcode
*/
[[nodiscard]] static constexpr std::optional<std::int64_t>
scaledMean(std::uint64_t total, std::uint64_t count, std::uint64_t scale = 1) noexcept
{
if (count == 0 || scale == 0)
return std::nullopt;
constexpr auto kInt64Max =
static_cast<std::uint64_t>(std::numeric_limits<std::int64_t>::max());
auto const whole = total / count;
if (whole > kInt64Max / scale)
return static_cast<std::int64_t>(kInt64Max);
// Scale the remainder too, so `scale` recovers the fractional digits
// it exists for. Skipped when the product itself would overflow, at
// which point it is worth less than one part in 2^63 of the result.
auto const remainder = total % count;
std::uint64_t fraction = 0;
if (remainder <= std::numeric_limits<std::uint64_t>::max() / scale)
fraction = remainder * scale / count;
auto const scaled = whole * scale;
if (scaled > kInt64Max - fraction)
return static_cast<std::int64_t>(kInt64Max);
return static_cast<std::int64_t>(scaled + fraction);
}
/**
* Record a job enqueued event.
* @param jobType The job type name (e.g. "ledgerData").
@@ -639,7 +726,10 @@ private:
*/
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::ObservableInstrument> loadFactorGauge_;
/**
* Observable gauges for NodeStore write_load and read_queue.
* Observable gauge multiplexing every NodeStore value onto one
* instrument via its `metric` label: I/O totals, the read and write
* means derived from them, the NuDB write-queue detail, and the
* ledger-acquisition stall counters.
*/
opentelemetry::nostd::shared_ptr<opentelemetry::metrics::ObservableInstrument> nodeStoreGauge_;
/**
@@ -803,6 +893,61 @@ private:
registerLoadFactorGauge(); // Task 9.7
void
registerNodeStoreGauge(); // Task 9.1
/**
* Sink handed to the registerNodeStoreGauge() helpers below.
*
* Every value they publish multiplexes onto the single `nodestore_state`
* gauge through its `metric` label, so the helpers need no access to the
* OTel observer result -- just somewhere to put a name and a number.
* That also makes each helper directly unit-testable by passing a
* recording sink, which the real observer result is not.
*/
using ObserveFn = std::function<void(char const* name, std::int64_t value)>;
/**
* Observe the NodeStore I/O totals and the means derived from them.
*
* @param db NodeStore to read the counters from.
* @param observe Sink for one `metric`-labelled value.
*/
static void
observeNodeStoreTotals(node_store::Database& db, ObserveFn const& observe);
/**
* Observe the backend write-path detail, when the backend measures it.
*
* Publishes nothing for a backend whose getWriteStats() is std::nullopt,
* which is every backend except NuDB. Absent labels let a reader tell
* "not measured" from "measured, and idle"; zeros would read as a
* perfectly idle write path.
*
* @param db NodeStore whose writable backend is sampled.
* @param observe Sink for one `metric`-labelled value.
*/
static void
observeWritePathDetail(node_store::Database const& db, ObserveFn const& observe);
/**
* Observe the ledger-acquisition progress and stall counters.
*
* @param stats Process-wide acquisition counters.
* @param observe Sink for one `metric`-labelled value.
*/
static void
observeAcquireStats(AcquireStats const& stats, ObserveFn const& observe);
/**
* Observe the read queue depth and the read thread-pool counts.
*
* These four have no accessor on Database, so its JSON counters object
* is still the only way to reach them.
*
* @param db NodeStore to read the JSON counters from.
* @param observe Sink for one `metric`-labelled value.
*/
static void
observeReadQueue(node_store::Database& db, ObserveFn const& observe);
void
registerServerInfoGauge(); // Task 9.7a
void