From 2c5d3871660975401fb51398cdd5cd9ec948b3ed Mon Sep 17 00:00:00 2001 From: ditian12 Date: Tue, 15 Sep 2026 14:56:33 +0000 Subject: [PATCH 1/4] feat(umbp): serve metrics from a masterless node, starting with the standalone server A UMBP peer measured everything and, without a master, published none of it. MasterClient was the only metric destination in the process: PoolClient's CountMetric returned early when master_client_ was null, PublishComponentMetrics returned early for the same reason, and the component tick it drives was registered only through MasterClient::AddMetricsProvider. The standalone server -- the deployment sglang's direct external linker actually runs -- therefore had no /metrics endpoint, and no flag could give it one, because the repo's only MetricsServer instantiation is master_server.cpp's. Introduce MetricSink as the seam between measuring and publishing. MasterClient implements it unchanged, so with a master the metrics still go to the master -- the one behaviour this commit deliberately leaves alone. PrometheusMetricSink is the masterless counterpart: it owns a MetricsServer in this process and stamps the `node` label master used to add on ingest, so a dashboard query written for a master-backed cluster works against a standalone server, and two scraped servers do not collapse into one series. PoolClient picks its sink once in Init, and no emission site asks about masters again. Without a master it also starts its own tick thread, because the component counters are published by a timer that only MasterClient previously owned; Shutdown joins it before tearing down the backends it samples, with a final tick so a short-lived server still publishes. The endpoint is created only when there is no master. Its port comes from UMBP_STANDALONE_METRICS_PORT and defaults to 9092 rather than the master's 9091, so a node running both does not have two processes contending for one port. 0/off/false disables it; garbage is rejected loudly instead of being read as "off", because a typo discovered as a blank dashboard days later is the failure this is meant to prevent. A bind failure logs ERROR and continues: a server that stores KV but cannot be scraped beats one that will not start. The sink is injected after WithEmbeddedDefaults, not before. For a server started with no distributed environment -- the masterless default, and the case that needs a local endpoint most -- that function synthesizes `distributed` from a default-constructed struct and would have silently dropped the pointer. Tests cover the sink's counter/gauge/histogram forwarding, the node-label rules (stamp, do not overwrite, empty means none), and the port-resolution table. Co-Authored-By: Claude Opus 5 --- src/umbp/CMakeLists.txt | 1 + .../metrics/prometheus_metric_sink.cpp | 64 +++++ src/umbp/distributed/peer/peer_service.cpp | 42 ++- src/umbp/distributed/pool_client.cpp | 164 +++++++++--- src/umbp/include/umbp/common/config.h | 11 + src/umbp/include/umbp/distributed/config.h | 16 ++ .../umbp/distributed/master/master_client.h | 16 +- .../umbp/distributed/metrics/metric_sink.h | 67 +++++ .../metrics/prometheus_metric_sink.h | 83 ++++++ .../umbp/distributed/peer/peer_service.h | 8 +- .../include/umbp/distributed/pool_client.h | 37 ++- .../umbp/standalone/standalone_server.h | 27 ++ src/umbp/standalone/standalone_server.cpp | 104 +++++++- tests/cpp/umbp/distributed/CMakeLists.txt | 9 + .../test_prometheus_metric_sink.cpp | 250 ++++++++++++++++++ 15 files changed, 816 insertions(+), 83 deletions(-) create mode 100644 src/umbp/distributed/metrics/prometheus_metric_sink.cpp create mode 100644 src/umbp/include/umbp/distributed/metrics/metric_sink.h create mode 100644 src/umbp/include/umbp/distributed/metrics/prometheus_metric_sink.h create mode 100644 tests/cpp/umbp/distributed/test_prometheus_metric_sink.cpp diff --git a/src/umbp/CMakeLists.txt b/src/umbp/CMakeLists.txt index 1b62816aa..ec8fc82d8 100644 --- a/src/umbp/CMakeLists.txt +++ b/src/umbp/CMakeLists.txt @@ -384,6 +384,7 @@ add_library( distributed/routing/route_get_strategy.cpp distributed/routing/route_put_strategy.cpp distributed/metrics/component_metrics.cpp + distributed/metrics/prometheus_metric_sink.cpp distributed/peer/backend/page_backend.cpp distributed/peer/backend/hbm_backend.cpp distributed/peer/backend/ssd_backend.cpp diff --git a/src/umbp/distributed/metrics/prometheus_metric_sink.cpp b/src/umbp/distributed/metrics/prometheus_metric_sink.cpp new file mode 100644 index 000000000..30d5129ac --- /dev/null +++ b/src/umbp/distributed/metrics/prometheus_metric_sink.cpp @@ -0,0 +1,64 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +#include "umbp/distributed/metrics/prometheus_metric_sink.h" + +#include +#include + +#include "mori/metrics/prometheus_metrics_server.hpp" + +namespace mori::umbp { + +PrometheusMetricSink::PrometheusMetricSink(int port, std::string node_id) + : node_id_(std::move(node_id)), server_(std::make_unique(port)) {} + +PrometheusMetricSink::~PrometheusMetricSink() = default; + +int PrometheusMetricSink::port() const { return server_->port(); } + +MetricSink::Labels PrometheusMetricSink::WithNode(Labels labels) const { + if (node_id_.empty()) return labels; + // A component that already knows which node it speaks for keeps its own + // value: overwriting it here would be this sink inventing a fact about a + // series it is only forwarding. + const bool has_node = + std::any_of(labels.begin(), labels.end(), [](const auto& kv) { return kv.first == "node"; }); + if (!has_node) labels.emplace_back("node", node_id_); + return labels; +} + +void PrometheusMetricSink::AddCounter(std::string name, std::string help, Labels labels, + double delta) { + server_->addCounter(name, help, WithNode(std::move(labels)), delta); +} + +void PrometheusMetricSink::SetGauge(std::string name, std::string help, Labels labels, + double value) { + server_->setGauge(name, help, WithNode(std::move(labels)), value); +} + +void PrometheusMetricSink::Observe(std::string name, std::string help, Labels labels, + const std::vector& bounds, double value) { + server_->observe(name, help, WithNode(std::move(labels)), bounds, value); +} + +} // namespace mori::umbp diff --git a/src/umbp/distributed/peer/peer_service.cpp b/src/umbp/distributed/peer/peer_service.cpp index 7e68b9813..f4c57e1d3 100644 --- a/src/umbp/distributed/peer/peer_service.cpp +++ b/src/umbp/distributed/peer/peer_service.cpp @@ -32,8 +32,8 @@ #include "mori/utils/mori_log.hpp" #include "umbp/common/env_time.h" #include "umbp/common/grpc_limits.h" -#include "umbp/distributed/master/master_client.h" #include "umbp/distributed/master/master_metrics.h" +#include "umbp/distributed/metrics/metric_sink.h" #include "umbp/distributed/peer/backend/medium_backend.h" #include "umbp/distributed/peer/batch_resolve_codec.h" #include "umbp/distributed/pool/peer_pool.h" @@ -122,7 +122,7 @@ void FillPagesAndDescs(Response* resp, const std::vector& pages, u class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Service { public: - UMBPPeerServiceImpl(PeerPool* pool, MasterClient* master_client, + UMBPPeerServiceImpl(PeerPool* pool, MetricSink* metric_sink, const std::vector& engine_desc_bytes) : pool_(pool), registry_(pool == nullptr ? nullptr : pool->Backends()), @@ -133,7 +133,7 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se // starts this server). media_(registry_ == nullptr ? std::vector{} : registry_->All()), engine_desc_bytes_(engine_desc_bytes), - master_client_(master_client) {} + metric_sink_(metric_sink) {} grpc::Status GetPeerInfo(grpc::ServerContext* /*context*/, const ::umbp::GetPeerInfoRequest* /*request*/, @@ -440,46 +440,44 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se const std::vector& Media() const { return media_; } void RecordInboundPut(uint64_t bytes, const char* traffic) { - if (master_client_ == nullptr || bytes == 0) return; - MasterClient::Labels labels = {{"traffic", std::string(traffic)}}; - master_client_->AddCounter(MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL, - MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL_HELP, labels, - static_cast(bytes)); + if (metric_sink_ == nullptr || bytes == 0) return; + MetricSink::Labels labels = {{"traffic", std::string(traffic)}}; + metric_sink_->AddCounter(MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL_HELP, labels, + static_cast(bytes)); } void RecordInboundGet(uint64_t bytes, const char* traffic) { - if (master_client_ == nullptr || bytes == 0) return; - MasterClient::Labels labels = {{"traffic", std::string(traffic)}}; - master_client_->AddCounter(MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL, - MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL_HELP, labels, - static_cast(bytes)); + if (metric_sink_ == nullptr || bytes == 0) return; + MetricSink::Labels labels = {{"traffic", std::string(traffic)}}; + metric_sink_->AddCounter(MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL_HELP, labels, + static_cast(bytes)); } PeerPool* pool_; BackendRegistry* registry_; const std::vector media_; const std::vector& engine_desc_bytes_; - MasterClient* master_client_; + MetricSink* metric_sink_; }; PeerServiceServer::PeerServiceServer(PeerPool& pool, std::vector engine_desc_bytes, - MasterClient* master_client) - : pool_(&pool), - master_client_(master_client), - engine_desc_bytes_(std::move(engine_desc_bytes)) { - service_ = std::make_unique(pool_, master_client_, engine_desc_bytes_); + MetricSink* metric_sink) + : pool_(&pool), metric_sink_(metric_sink), engine_desc_bytes_(std::move(engine_desc_bytes)) { + service_ = std::make_unique(pool_, metric_sink_, engine_desc_bytes_); } PeerServiceServer::PeerServiceServer(BackendRegistry* registry, std::vector engine_desc_bytes, - MasterClient* master_client) + MetricSink* metric_sink) : owned_pool_(registry == nullptr ? nullptr : std::make_unique(registry, MakeSingleBackendPolicy())), pool_(owned_pool_.get()), - master_client_(master_client), + metric_sink_(metric_sink), engine_desc_bytes_(std::move(engine_desc_bytes)) { - service_ = std::make_unique(pool_, master_client_, engine_desc_bytes_); + service_ = std::make_unique(pool_, metric_sink_, engine_desc_bytes_); } PeerServiceServer::~PeerServiceServer() { Stop(); } diff --git a/src/umbp/distributed/pool_client.cpp b/src/umbp/distributed/pool_client.cpp index 7d87c6737..b250e7172 100644 --- a/src/umbp/distributed/pool_client.cpp +++ b/src/umbp/distributed/pool_client.cpp @@ -39,6 +39,7 @@ #include #include #include +#include #include #include #include @@ -116,17 +117,16 @@ BatchBandwidthSplit ComputeBatchBandwidthBytes(const std::vector& result return acc; } -// `master_client` is null on a node running without a master; the histogram -// simply has nowhere to go, and the caller should not have to know that. -void ObserveBatchBandwidth(MasterClient* master_client, double bytes, double seconds, - const char* metric_name, const char* metric_help, - std::string_view traffic) { - if (master_client == nullptr || bytes <= 0.0 || seconds <= 0.0) return; +// `sink` is null on a node that publishes nowhere — no master and no local +// metrics server — in which case the histogram has nowhere to go and the +// caller should not have to know that. +void ObserveBatchBandwidth(MetricSink* sink, double bytes, double seconds, const char* metric_name, + const char* metric_help, std::string_view traffic) { + if (sink == nullptr || bytes <= 0.0 || seconds <= 0.0) return; const double gibps = (bytes / seconds) / kGiB; if (gibps <= 0.0) return; - MasterClient::Labels labels = {{"traffic", std::string(traffic)}}; - master_client->Observe(metric_name, metric_help, std::move(labels), BatchBandwidthBucketsGiBps(), - gibps); + MetricSink::Labels labels = {{"traffic", std::string(traffic)}}; + sink->Observe(metric_name, metric_help, std::move(labels), BatchBandwidthBucketsGiBps(), gibps); } // --------------------------------------------------------------------------- @@ -453,10 +453,9 @@ ScopeExit(Fn) -> ScopeExit; // exit, so callers just add to them as keys succeed. class ScopedBatchBandwidth { public: - ScopedBatchBandwidth(MasterClient* master_client, const char* metric_name, - const char* metric_help, const double& local_bytes, - const double& remote_bytes) - : master_client_(master_client), + ScopedBatchBandwidth(MetricSink* sink, const char* metric_name, const char* metric_help, + const double& local_bytes, const double& remote_bytes) + : sink_(sink), metric_name_(metric_name), metric_help_(metric_help), local_bytes_(local_bytes), @@ -470,14 +469,12 @@ class ScopedBatchBandwidth { const double seconds = std::chrono::duration_cast>( std::chrono::steady_clock::now() - start_) .count(); - ObserveBatchBandwidth(master_client_, local_bytes_, seconds, metric_name_, metric_help_, - "local"); - ObserveBatchBandwidth(master_client_, remote_bytes_, seconds, metric_name_, metric_help_, - "remote"); + ObserveBatchBandwidth(sink_, local_bytes_, seconds, metric_name_, metric_help_, "local"); + ObserveBatchBandwidth(sink_, remote_bytes_, seconds, metric_name_, metric_help_, "remote"); } private: - MasterClient* master_client_; + MetricSink* sink_; const char* metric_name_; const char* metric_help_; const double& local_bytes_; @@ -997,21 +994,41 @@ bool PoolClient::Init() { if (master_client_) master_client_->SetBackendRegistry(®istry_); - // Every instrumented component on this node rides the existing metrics tick: - // the storage backend (generic slot-lifecycle series from the decorator, plus - // whatever the medium says about its own internals) and the transfer engine - // (per-engine bytes, plans and in-flight time). Backend-agnostic by - // construction — PoolClient forwards what each component samples and never - // learns which medium or transport produced it. - if (master_client_) master_client_->AddMetricsProvider([this] { PublishComponentMetrics(); }); + // Pick this node's metrics destination once, here, so that no emission site + // below ever has to ask whether there is a master. With a master the + // MasterClient IS the sink; without one the owner of this client may have + // supplied a local Prometheus sink (the standalone server does). Deliberately + // not both: the same series arriving at a master AND at a locally scraped + // endpoint would be double-counted by any Prometheus that scrapes the pair. + if (master_client_) { + metric_sink_ = master_client_.get(); + } else { + metric_sink_ = config_.metric_sink; + } + + // Every instrumented component on this node rides a metrics tick: the storage + // backend (generic slot-lifecycle series from the decorator, plus whatever the + // medium says about its own internals) and the transfer engine (per-engine + // bytes, plans and in-flight time). Backend-agnostic by construction — + // PoolClient forwards what each component samples and never learns which + // medium or transport produced it. + // + // Which thread drives that tick is the only difference between the two + // deployments: MasterClient already runs one, and a masterless node starts + // its own only when it actually has somewhere to publish. + if (master_client_) { + master_client_->AddMetricsProvider([this] { PublishComponentMetrics(); }); + } else if (metric_sink_ != nullptr) { + StartLocalMetricsReporting(); + } // Pack engine_desc for master registration. std::vector engine_desc_bytes; if (peer_directory_ != nullptr) engine_desc_bytes = peer_directory_->PackedLocalEngineDesc(); if (config_.peer_service_port > 0 || config_.auto_peer_service_port) { - peer_service_ = std::make_unique(*default_pool_, engine_desc_bytes, - master_client_.get()); + peer_service_ = + std::make_unique(*default_pool_, engine_desc_bytes, metric_sink_); if (!peer_service_->Start(config_.peer_service_port)) { MORI_UMBP_ERROR("[PoolClient] PeerService failed to start on port {}", config_.peer_service_port); @@ -1124,6 +1141,10 @@ void PoolClient::Shutdown() { MORI_UMBP_WARN("[PoolClient] UnregisterSelf failed: {}", status.error_message()); } } + // The masterless counterpart, joined at the same point in teardown and for + // the same reason: its final tick samples the backends and transfer engine, + // which the code below is about to destroy. + StopLocalMetricsReporting(); { std::lock_guard lock(peers_mutex_); @@ -1156,6 +1177,10 @@ void PoolClient::Shutdown() { hbm_engine_ = nullptr; transfer_engine_.reset(); + // Drop the sink before the object it may point at. In the master-backed + // deployment metric_sink_ aliases master_client_, so leaving it set here + // would leave a dangling pointer that any late CountMetric would follow. + metric_sink_ = nullptr; master_client_.reset(); } @@ -1187,10 +1212,10 @@ void PoolClient::RouteAllPutsLocally(size_t count, routes->assign(count, local); } -void PoolClient::CountMetric(std::string name, std::string help, MasterClient::Labels labels, +void PoolClient::CountMetric(std::string name, std::string help, MetricSink::Labels labels, double delta) { - if (master_client_ == nullptr) return; - master_client_->AddCounter(std::move(name), std::move(help), std::move(labels), delta); + if (metric_sink_ == nullptr) return; + metric_sink_->AddCounter(std::move(name), std::move(help), std::move(labels), delta); } BackendRegistry& PoolClient::Backends() { return registry_; } std::map PoolClient::TierReadHits() const { @@ -2236,10 +2261,10 @@ std::vector PoolClient::BatchPut(const std::vector& keys, std::chrono::duration_cast>(call_end - call_start).count(); if (seconds > 0.0) { auto split = ComputeBatchBandwidthBytes(outcomes, sizes, routes, config_.master_config.node_id); - ObserveBatchBandwidth(master_client_.get(), split.local, seconds, + ObserveBatchBandwidth(metric_sink_, split.local, seconds, MORI_UMBP_METRIC_CLIENT_BATCH_PUT_BANDWIDTH, MORI_UMBP_METRIC_CLIENT_BATCH_PUT_BANDWIDTH_HELP, "local"); - ObserveBatchBandwidth(master_client_.get(), split.remote, seconds, + ObserveBatchBandwidth(metric_sink_, split.remote, seconds, MORI_UMBP_METRIC_CLIENT_BATCH_PUT_BANDWIDTH, MORI_UMBP_METRIC_CLIENT_BATCH_PUT_BANDWIDTH_HELP, "remote"); } @@ -2960,10 +2985,10 @@ void PoolClient::ObserveBatchGetBandwidth(const std::vector& results, .count(); if (seconds <= 0.0) return; auto split = ComputeBatchBandwidthBytes(results, sizes, routes, config_.master_config.node_id); - ObserveBatchBandwidth(master_client_.get(), split.local, seconds, + ObserveBatchBandwidth(metric_sink_, split.local, seconds, MORI_UMBP_METRIC_CLIENT_BATCH_GET_BANDWIDTH, MORI_UMBP_METRIC_CLIENT_BATCH_GET_BANDWIDTH_HELP, "local"); - ObserveBatchBandwidth(master_client_.get(), split.remote, seconds, + ObserveBatchBandwidth(metric_sink_, split.remote, seconds, MORI_UMBP_METRIC_CLIENT_BATCH_GET_BANDWIDTH, MORI_UMBP_METRIC_CLIENT_BATCH_GET_BANDWIDTH_HELP, "remote"); } @@ -3122,9 +3147,9 @@ std::vector PoolClient::BatchGetRanges(const std::vector& key // Observed at scope exit for whichever phases ran; see ScopedBatchBandwidth. double local_bytes = 0.0; double remote_bytes = 0.0; - ScopedBatchBandwidth bandwidth( - master_client_.get(), MORI_UMBP_METRIC_CLIENT_BATCH_GET_RANGES_BANDWIDTH, - MORI_UMBP_METRIC_CLIENT_BATCH_GET_RANGES_BANDWIDTH_HELP, local_bytes, remote_bytes); + ScopedBatchBandwidth bandwidth(metric_sink_, MORI_UMBP_METRIC_CLIENT_BATCH_GET_RANGES_BANDWIDTH, + MORI_UMBP_METRIC_CLIENT_BATCH_GET_RANGES_BANDWIDTH_HELP, + local_bytes, remote_bytes); // Sub-timers. `dbg` is null when disabled, which makes every PhaseTimer inert. const bool ranged_debug = RangedDebugEnabled(); @@ -3560,9 +3585,9 @@ std::vector PoolClient::BatchPutRanges(const std::vector& key // Observed at scope exit for whichever phases ran; see ScopedBatchBandwidth. double local_bytes = 0.0; double remote_bytes = 0.0; - ScopedBatchBandwidth bandwidth( - master_client_.get(), MORI_UMBP_METRIC_CLIENT_BATCH_PUT_RANGES_BANDWIDTH, - MORI_UMBP_METRIC_CLIENT_BATCH_PUT_RANGES_BANDWIDTH_HELP, local_bytes, remote_bytes); + ScopedBatchBandwidth bandwidth(metric_sink_, MORI_UMBP_METRIC_CLIENT_BATCH_PUT_RANGES_BANDWIDTH, + MORI_UMBP_METRIC_CLIENT_BATCH_PUT_RANGES_BANDWIDTH_HELP, + local_bytes, remote_bytes); const bool ranged_debug = RangedDebugEnabled(); RangedPhases phases; @@ -4605,14 +4630,14 @@ bool PoolClient::EnsurePeerServiceConnection(PeerConnection& peer) { } void PoolClient::PublishComponentMetrics() { - if (!master_client_) return; + if (metric_sink_ == nullptr) return; MetricPublisher::Sink sink{ [this](const char* name, const char* help, const MetricLabels& labels, double delta) { CountMetric(name, help, labels, delta); }, [this](const char* name, const char* help, const MetricLabels& labels, double value) { - master_client_->SetGauge(name, help, labels, value); + metric_sink_->SetGauge(name, help, labels, value); }}; // Storage backends. tier= and backend= come from the component's own @@ -4668,4 +4693,59 @@ void PoolClient::PublishComponentMetrics() { } } +// --------------------------------------------------------------------------- +// Masterless metrics tick +// +// With a master, PublishComponentMetrics() rides MasterClient's existing +// metrics thread (AddMetricsProvider). Without one there is no such thread, +// so a node that publishes to a local metrics server needs its own — otherwise +// every counter a component keeps would sit in that component's atomics +// forever and /metrics would serve nothing but the handful of series the +// data-plane path writes directly. +// --------------------------------------------------------------------------- + +void PoolClient::StartLocalMetricsReporting() { + if (local_metrics_running_.load()) return; + local_metrics_interval_ms_ = static_cast( + GetEnvMilliseconds("UMBP_METRICS_REPORT_INTERVAL_MS", std::chrono::milliseconds(1000), + /*min_allowed=*/1) + .count()); + local_metrics_running_ = true; + try { + local_metrics_thread_ = std::thread(&PoolClient::LocalMetricsLoop, this); + } catch (const std::system_error& e) { + local_metrics_running_ = false; + MORI_UMBP_ERROR("[PoolClient] failed to start local metrics thread: {}", e.what()); + return; + } + MORI_UMBP_INFO("[PoolClient] local metrics reporting started (interval={}ms, no master)", + local_metrics_interval_ms_); +} + +void PoolClient::StopLocalMetricsReporting() { + if (!local_metrics_running_.load()) return; + local_metrics_running_ = false; + local_metrics_cv_.notify_one(); + if (local_metrics_thread_.joinable()) local_metrics_thread_.join(); + MORI_UMBP_INFO("[PoolClient] local metrics reporting stopped"); +} + +void PoolClient::LocalMetricsLoop() { + while (local_metrics_running_.load()) { + { + std::unique_lock lock(local_metrics_mu_); + local_metrics_cv_.wait_for(lock, std::chrono::milliseconds(local_metrics_interval_ms_), + [this] { return !local_metrics_running_.load(); }); + } + if (!local_metrics_running_.load()) break; + PublishComponentMetrics(); + } + // Final tick, mirroring MasterClient::MetricsLoop: without it the last + // sub-interval of every component's deltas would be lost at shutdown, and a + // short-lived server (a unit test, a one-shot benchmark) could publish + // nothing at all. Shutdown() calls this before tearing the components down, + // so they are still alive here. + PublishComponentMetrics(); +} + } // namespace mori::umbp diff --git a/src/umbp/include/umbp/common/config.h b/src/umbp/include/umbp/common/config.h index 147aabee4..35b801c35 100644 --- a/src/umbp/include/umbp/common/config.h +++ b/src/umbp/include/umbp/common/config.h @@ -30,6 +30,11 @@ namespace mori::umbp { +// Forward-declared rather than included: the dependency runs the other way +// (distributed/metrics/metric_sink.h is a distributed-layer header), and the +// config only ever stores a pointer to one. +class MetricSink; + enum class UMBPRole : int { Standalone = 0, SharedSSDLeader = 1, @@ -431,6 +436,12 @@ struct UMBPDistributedConfig { std::string workload_trace_path; uint32_t workload_trace_client_id = 0; uint64_t workload_trace_seed = 0; + + // Where a MASTERLESS node publishes its metrics; see + // PoolClientConfig::metric_sink, which this is lowered into verbatim. + // Caller-owned, must outlive the client, null disables. Ignored when + // master_config.master_address is set. + MetricSink* metric_sink = nullptr; }; // User-facing same-host standalone-process configuration. Set diff --git a/src/umbp/include/umbp/distributed/config.h b/src/umbp/include/umbp/distributed/config.h index 0c57f5e3c..346cc0e9d 100644 --- a/src/umbp/include/umbp/distributed/config.h +++ b/src/umbp/include/umbp/distributed/config.h @@ -32,6 +32,7 @@ #include "umbp/common/config.h" #include "umbp/common/env_time.h" +#include "umbp/distributed/metrics/metric_sink.h" #include "umbp/distributed/pool/policy_config.h" #include "umbp/distributed/types.h" @@ -339,6 +340,20 @@ struct PoolClientConfig { std::string workload_trace_path; uint32_t workload_trace_client_id = 0; uint64_t workload_trace_seed = 0; + + // Where this node's metrics go, for a node that has no master to send them + // to. Caller-owned and must outlive the PoolClient; null means metrics are + // measured and discarded, which is what every masterless deployment did + // before this seam existed. IGNORED when master_config.master_address is + // set: with a master, MasterClient is the sink and shipping the same series + // to both would double-count it on any Prometheus scraping the pair. + // + // PoolClient does not create one of these itself. A sink binds a TCP port, + // and PoolClient is also constructed in-process inside every sglang rank — + // eight of them per node — so making it implicit would turn one config flag + // into eight racing binds. The standalone server, which is one process per + // node, opts in explicitly. + MetricSink* metric_sink = nullptr; }; // Lower a user-facing UMBPDistributedConfig to the internal PoolClientConfig. @@ -380,6 +395,7 @@ inline PoolClientConfig ToPoolClientConfig(const UMBPDistributedConfig& dc, pc.workload_trace_path = dc.workload_trace_path; pc.workload_trace_client_id = dc.workload_trace_client_id; pc.workload_trace_seed = dc.workload_trace_seed; + pc.metric_sink = dc.metric_sink; pc.dram = std::move(dram); pc.ssd = std::move(ssd); // Unlike dram/ssd, UMBPHbmConfig carries no ownership knobs that live diff --git a/src/umbp/include/umbp/distributed/master/master_client.h b/src/umbp/include/umbp/distributed/master/master_client.h index 0b5eb3d7d..e1ca99f0c 100644 --- a/src/umbp/include/umbp/distributed/master/master_client.h +++ b/src/umbp/include/umbp/distributed/master/master_client.h @@ -42,6 +42,7 @@ #include #include "umbp/distributed/config.h" +#include "umbp/distributed/metrics/metric_sink.h" #include "umbp/distributed/peer/backend/medium_backend.h" #include "umbp/distributed/routing/route_put_strategy.h" #include "umbp/distributed/types.h" @@ -70,12 +71,15 @@ struct RouteGetResult { std::string logical_tier; }; -class MasterClient { +// Implements MetricSink: when a node has a master, that is where its metrics +// go. A masterless node is given a PrometheusMetricSink instead, and neither +// PoolClient nor any component publishing through it knows which one it has. +class MasterClient : public MetricSink { public: - using Labels = std::vector>; + using Labels = MetricSink::Labels; explicit MasterClient(const UMBPMasterClientConfig& config); - ~MasterClient(); + ~MasterClient() override; MasterClient(const MasterClient&) = delete; MasterClient& operator=(const MasterClient&) = delete; @@ -159,10 +163,10 @@ class MasterClient { bool ClearFullSync(); // --- Client-side metrics --- - void AddCounter(std::string name, std::string help, Labels labels, double delta); - void SetGauge(std::string name, std::string help, Labels labels, double value); + void AddCounter(std::string name, std::string help, Labels labels, double delta) override; + void SetGauge(std::string name, std::string help, Labels labels, double value) override; void Observe(std::string name, std::string help, Labels labels, const std::vector& bounds, - double value); + double value) override; // Register a callback run once per metrics flush tick in the existing metrics // thread (no new thread) so a component can publish counters/gauges from its diff --git a/src/umbp/include/umbp/distributed/metrics/metric_sink.h b/src/umbp/include/umbp/distributed/metrics/metric_sink.h new file mode 100644 index 000000000..9e5c39ad1 --- /dev/null +++ b/src/umbp/include/umbp/distributed/metrics/metric_sink.h @@ -0,0 +1,67 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +#pragma once + +#include +#include +#include + +// --------------------------------------------------------------------------- +// MetricSink — where a node's metrics go. +// +// Everything UMBP measures on a peer used to have exactly one destination: +// MasterClient buffered it and shipped it to the master, which merged it into +// the master's own MetricsServer. That made observability a property of +// cluster membership rather than of the process, so a node running WITHOUT a +// master — an embedded client, and every standalone server, which is the +// deployment sglang's direct external linker actually uses — measured +// everything and then dropped it on the floor, with no port to scrape and no +// flag that could change that. +// +// This interface is the seam that separates the two. A component reports +// into a MetricSink; whether that sink forwards over gRPC to a master +// (MasterClient) or serves the numbers locally in Prometheus text format +// (PrometheusMetricSink) is a deployment decision made once at wiring time. +// --------------------------------------------------------------------------- + +namespace mori::umbp { + +class MetricSink { + public: + using Labels = std::vector>; + + virtual ~MetricSink() = default; + + // Add `delta` to a monotonically-increasing counter series. + virtual void AddCounter(std::string name, std::string help, Labels labels, double delta) = 0; + + // Overwrite the current reading of a gauge series. + virtual void SetGauge(std::string name, std::string help, Labels labels, double value) = 0; + + // Record one histogram observation. `bounds` is an ascending list of finite + // upper bounds; the first write for a series wins the layout, matching + // MetricsServer::observe() so the two sinks stay interchangeable. + virtual void Observe(std::string name, std::string help, Labels labels, + const std::vector& bounds, double value) = 0; +}; + +} // namespace mori::umbp diff --git a/src/umbp/include/umbp/distributed/metrics/prometheus_metric_sink.h b/src/umbp/include/umbp/distributed/metrics/prometheus_metric_sink.h new file mode 100644 index 000000000..4fba84496 --- /dev/null +++ b/src/umbp/include/umbp/distributed/metrics/prometheus_metric_sink.h @@ -0,0 +1,83 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +#pragma once + +#include +#include +#include + +#include "umbp/distributed/metrics/metric_sink.h" + +namespace mori::metrics { +class MetricsServer; +} + +namespace mori::umbp { + +// --------------------------------------------------------------------------- +// PrometheusMetricSink — a node that serves its own numbers. +// +// The masterless counterpart to MasterClient: instead of buffering samples +// and shipping them to a master that owns the only MetricsServer in the +// deployment, this sink writes them straight into a MetricsServer running in +// THIS process, which a Prometheus job scrapes at http://:/metrics. +// +// It also stamps a `node` label onto every series it forwards, which is what +// master used to add on ingest. Keeping it means a dashboard query written +// against a master-backed cluster (`sum by (node) (...)`) works unchanged +// against a standalone server, and two servers scraped into one Prometheus do +// not silently collapse into one series. +// +// Owns the MetricsServer: construction binds the port and throws +// std::runtime_error if it cannot, so a misconfigured port fails loudly at +// startup instead of producing a server that is quietly unscrapeable. +// --------------------------------------------------------------------------- +class PrometheusMetricSink final : public MetricSink { + public: + // Binds `port` immediately. `node_id` is stamped as a `node` label on every + // series; an empty node_id stamps nothing. + PrometheusMetricSink(int port, std::string node_id); + ~PrometheusMetricSink() override; + + PrometheusMetricSink(const PrometheusMetricSink&) = delete; + PrometheusMetricSink& operator=(const PrometheusMetricSink&) = delete; + + void AddCounter(std::string name, std::string help, Labels labels, double delta) override; + void SetGauge(std::string name, std::string help, Labels labels, double value) override; + void Observe(std::string name, std::string help, Labels labels, const std::vector& bounds, + double value) override; + + int port() const; + const std::string& node_id() const { return node_id_; } + + // Exposed for tests that want to scrape without going over a socket. + mori::metrics::MetricsServer* server() { return server_.get(); } + + private: + // Appends {node=node_id_} unless the caller already supplied one. + Labels WithNode(Labels labels) const; + + std::string node_id_; + std::unique_ptr server_; +}; + +} // namespace mori::umbp diff --git a/src/umbp/include/umbp/distributed/peer/peer_service.h b/src/umbp/include/umbp/distributed/peer/peer_service.h index 184ab514b..3e2fe9f0c 100644 --- a/src/umbp/include/umbp/distributed/peer/peer_service.h +++ b/src/umbp/include/umbp/distributed/peer/peer_service.h @@ -30,7 +30,7 @@ namespace mori::umbp { class BackendRegistry; -class MasterClient; +class MetricSink; class PeerPool; class PeerServiceServer { @@ -43,11 +43,11 @@ class PeerServiceServer { // backend-agnostic refactor Phase 0 — SSD is unwired from the distributed // data plane (see design-backend-agnostic-refactor.md). PeerServiceServer(PeerPool& pool, std::vector engine_desc_bytes = {}, - MasterClient* master_client = nullptr); + MetricSink* metric_sink = nullptr); // Source-compatible adapter for embedders that have not created a Pool yet. // The server owns an implicit SingleBackendPolicy PeerPool over `registry`. PeerServiceServer(BackendRegistry* registry, std::vector engine_desc_bytes = {}, - MasterClient* master_client = nullptr); + MetricSink* metric_sink = nullptr); ~PeerServiceServer(); bool Start(uint16_t port); @@ -57,7 +57,7 @@ class PeerServiceServer { private: std::unique_ptr owned_pool_; PeerPool* pool_; - MasterClient* master_client_; + MetricSink* metric_sink_; std::vector engine_desc_bytes_; diff --git a/src/umbp/include/umbp/distributed/pool_client.h b/src/umbp/include/umbp/distributed/pool_client.h index cb678a7d6..3b5e2af80 100644 --- a/src/umbp/include/umbp/distributed/pool_client.h +++ b/src/umbp/include/umbp/distributed/pool_client.h @@ -253,9 +253,10 @@ class PoolClient { std::atomic initialized_{false}; // Sample every instrumented component on this node — each registered storage - // backend, plus the transfer engine — and ship the result. Registered as a - // MasterClient metrics provider, so it runs on the metrics thread and never - // on a data-plane path. + // backend, plus the transfer engine — and ship the result. With a master it + // is registered as a MasterClient metrics provider and runs on that client's + // metrics thread; without one it runs on local_metrics_thread_ below. Either + // way it is a tick, never a data-plane path. // // There is no per-component code here and no place to add any: a component is // a MetricSource, the labels that identify it come from Tier() plus the @@ -270,6 +271,27 @@ class PoolClient { std::unique_ptr master_client_; + // Where this node's metrics go. Points at master_client_ when this node has + // a master, otherwise at config_.metric_sink (a PrometheusMetricSink owned by + // whoever built this client), and is null when neither exists — the case + // where metrics are measured and discarded. Never owns. + MetricSink* metric_sink_ = nullptr; + + // Drives PublishComponentMetrics() when there is no MasterClient whose + // metrics thread could. Started by Init only when metric_sink_ is set and + // master_client_ is not; joined by Shutdown before the components it samples + // are torn down. Interval comes from UMBP_METRICS_REPORT_INTERVAL_MS, the + // same knob the master-backed path uses, so the two tick alike. + void LocalMetricsLoop(); + void StartLocalMetricsReporting(); + void StopLocalMetricsReporting(); + + std::thread local_metrics_thread_; + std::atomic local_metrics_running_{false}; + std::mutex local_metrics_mu_; + std::condition_variable local_metrics_cv_; + uint64_t local_metrics_interval_ms_ = 1000; + // Every storage medium live on this node. Owned here because PoolClient is // the natural lifetime anchor for the per-process IO engine + backend pools. // MasterClient borrows it for heartbeat aggregation; default_pool_ borrows it @@ -724,10 +746,11 @@ class PoolClient { const std::vector& sizes, std::vector* results); - // Counter sink that tolerates a node running without a master: metrics - // accumulate on the MasterClient and ride its flush tick, so with no master - // there is simply nowhere to put them. - void CountMetric(std::string name, std::string help, MasterClient::Labels labels, double delta); + // Counter sink that tolerates a node publishing nowhere: with a master the + // delta accumulates on the MasterClient and rides its flush tick, without one + // it goes straight into the local metrics server, and with neither it is + // dropped. + void CountMetric(std::string name, std::string help, MetricSink::Labels labels, double delta); // Placement when there is no master to ask: this node is the only candidate, // so every key is routed to it on the medium it serves. The put paths then diff --git a/src/umbp/include/umbp/standalone/standalone_server.h b/src/umbp/include/umbp/standalone/standalone_server.h index 9eba001e1..298187a2d 100644 --- a/src/umbp/include/umbp/standalone/standalone_server.h +++ b/src/umbp/include/umbp/standalone/standalone_server.h @@ -29,8 +29,27 @@ #include "umbp/common/config.h" +namespace mori::umbp { +class PrometheusMetricSink; +} + namespace mori::umbp::standalone { +// Port the standalone server serves Prometheus metrics on when nothing else +// says otherwise. Deliberately not the master's 9091: a node may run a master +// and a standalone server, and two processes silently contending for one port +// is a worse failure than a second number to remember. +inline constexpr int kDefaultStandaloneMetricsPort = 9092; + +// Resolve the metrics port from UMBP_STANDALONE_METRICS_PORT. +// unset -> kDefaultStandaloneMetricsPort +// 0 or "off"/"false" -> 0, meaning do not serve metrics +// 1..65535 -> that port +// Any other value is rejected: `*error` is set and 0 is NOT returned, because +// "the operator asked for something impossible" and "the operator asked for +// silence" must not look the same to the caller. +int ResolveMetricsPortFromEnv(bool* ok, std::string* error); + class StandaloneServer { public: StandaloneServer(UMBPConfig config, std::string address); @@ -42,11 +61,19 @@ class StandaloneServer { const std::string& address() const { return address_; } + // The port /metrics is served on, or 0 when this server publishes no metrics + // (explicitly disabled, a master is configured, or the bind failed). + int metrics_port() const; + private: class Impl; UMBPConfig config_; std::string address_; + // Declared BEFORE impl_ so it outlives it: the PoolClient inside impl_ holds + // a borrowed pointer to this sink and publishes into it until its Shutdown. + // Destruction runs in reverse declaration order, so impl_ dies first. + std::unique_ptr metric_sink_; std::unique_ptr impl_; }; diff --git a/src/umbp/standalone/standalone_server.cpp b/src/umbp/standalone/standalone_server.cpp index 0f68ceef1..ab7f03e31 100644 --- a/src/umbp/standalone/standalone_server.cpp +++ b/src/umbp/standalone/standalone_server.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -48,6 +49,7 @@ #include #include #include +#include #include #include #include @@ -59,6 +61,7 @@ #include "umbp/common/device_copy.h" #include "umbp/common/grpc_limits.h" #include "umbp/distributed/config.h" +#include "umbp/distributed/metrics/prometheus_metric_sink.h" #include "umbp/standalone/external_kv_identity_client.h" #include "umbp/standalone/ipc.h" #include "umbp/umbp_client.h" @@ -124,6 +127,11 @@ bool SetFdSocketTimeouts(int fd, std::chrono::milliseconds timeout) { setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) == 0; } +// Resolve the deployment the server will actually run. Called exactly once, +// by the StandaloneServer constructor, before the Impl (and therefore the +// PoolClient) is built: the node id the metrics `node` label needs does not +// exist in the config as given — it is synthesized here — so anything that +// wants it has to run after this and before the client. UMBPConfig NormalizeBackendConfig(UMBPConfig config) { // Older callers pass the worker-facing standalone_process field to the // server constructor. The server backend must never consume that field, @@ -479,8 +487,10 @@ class KeyHandleStats { class StandaloneServer::Impl final : public ::umbp::UMBPStandalone::Service { public: + // `config` must already be normalized (see NormalizeBackendConfig) — the + // constructor hands it straight to CreateUMBPClient. Impl(const UMBPConfig& config, std::string address) - : backend_config_(NormalizeBackendConfig(config)), + : backend_config_(config), client_(CreateUMBPClient(backend_config_)), // Concurrent reads for every medium but SSD, whose manager serializes // around its staging arena. Keyed on the LIVE medium rather than on @@ -1644,8 +1654,94 @@ class StandaloneServer::Impl final : public ::umbp::UMBPStandalone::Service { std::thread fd_thread_; }; +int ResolveMetricsPortFromEnv(bool* ok, std::string* error) { + if (ok != nullptr) *ok = true; + const char* raw = std::getenv("UMBP_STANDALONE_METRICS_PORT"); + if (raw == nullptr || raw[0] == '\0') return kDefaultStandaloneMetricsPort; + + std::string value(raw); + std::transform(value.begin(), value.end(), value.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + if (value == "off" || value == "false" || value == "no") return 0; + + try { + const long port = std::stol(value); + if (port == 0) return 0; + if (port < 0 || port > 65535) { + if (ok != nullptr) *ok = false; + if (error != nullptr) { + *error = "UMBP_STANDALONE_METRICS_PORT must be 0 (off) or 1..65535, got " + value; + } + return 0; + } + return static_cast(port); + } catch (const std::exception&) { + if (ok != nullptr) *ok = false; + if (error != nullptr) { + *error = "UMBP_STANDALONE_METRICS_PORT is not a number: " + std::string(raw); + } + return 0; + } +} + StandaloneServer::StandaloneServer(UMBPConfig config, std::string address) - : config_(std::move(config)), address_(std::move(address)) { + : config_(NormalizeBackendConfig(std::move(config))), address_(std::move(address)) { + // Metrics are only this process's job when it has no master. With one, the + // PoolClient below ships every series to that master, which already owns a + // MetricsServer and merges them there — binding a second endpoint here would + // publish the same numbers twice to anything scraping both. + const bool has_master = + config_.distributed.has_value() && !config_.distributed->master_config.master_address.empty(); + + bool port_ok = true; + std::string port_error; + const int metrics_port = ResolveMetricsPortFromEnv(&port_ok, &port_error); + if (!port_ok) { + // Refuse to quietly fall back to the default: an operator who set the + // variable wants a specific answer, and guessing one hides the typo until + // someone notices a blank dashboard. + MORI_UMBP_ERROR("[StandaloneServer] {}", port_error); + throw std::runtime_error(port_error); + } + + if (has_master) { + MORI_UMBP_INFO( + "[StandaloneServer] metrics endpoint not started: a master is configured ({}), so this " + "node's metrics are reported to it", + config_.distributed->master_config.master_address); + } else if (metrics_port == 0) { + MORI_UMBP_INFO( + "[StandaloneServer] metrics endpoint disabled by UMBP_STANDALONE_METRICS_PORT=0"); + } else if (!config_.distributed.has_value()) { + // Unreachable via NormalizeBackendConfig, which synthesizes a deployment + // for exactly this case. Handled rather than asserted because the + // alternative is dereferencing an empty optional to store the sink pointer, + // and a missing endpoint is a better failure than that. + MORI_UMBP_WARN( + "[StandaloneServer] metrics endpoint not started: no distributed deployment was resolved, " + "so there is no client to publish through"); + } else { + const std::string node_id = config_.distributed->master_config.node_id; + try { + metric_sink_ = std::make_unique(metrics_port, node_id); + // Borrowed by the PoolClient that Impl builds below. metric_sink_ is + // declared before impl_ in the header, so it outlives it. + config_.distributed->metric_sink = metric_sink_.get(); + MORI_UMBP_INFO("[StandaloneServer] metrics endpoint on http://0.0.0.0:{}/metrics (node='{}')", + metric_sink_->port(), node_id); + } catch (const std::exception& exc) { + // A busy port must not take the data plane down with it: a server that + // stores KV but cannot be scraped is strictly better than one that does + // not start. It is an ERROR because the operator asked for metrics and + // is not getting them. + MORI_UMBP_ERROR( + "[StandaloneServer] metrics endpoint failed to bind port {} ({}); continuing without " + "metrics", + metrics_port, exc.what()); + metric_sink_.reset(); + } + } + impl_ = std::make_unique(config_, address_); } @@ -1655,6 +1751,10 @@ bool StandaloneServer::Start() { return impl_->Start(); } void StandaloneServer::Run() { impl_->Run(); } +int StandaloneServer::metrics_port() const { + return metric_sink_ == nullptr ? 0 : metric_sink_->port(); +} + void StandaloneServer::Shutdown() { if (impl_) impl_->Shutdown(); } diff --git a/tests/cpp/umbp/distributed/CMakeLists.txt b/tests/cpp/umbp/distributed/CMakeLists.txt index 7af80982b..c0651b380 100644 --- a/tests/cpp/umbp/distributed/CMakeLists.txt +++ b/tests/cpp/umbp/distributed/CMakeLists.txt @@ -480,6 +480,15 @@ target_link_libraries(test_component_metrics PRIVATE umbp_common target_compile_features(test_component_metrics PRIVATE cxx_std_17) add_test(NAME test_component_metrics COMMAND test_component_metrics) +# test_prometheus_metric_sink — the masterless end of the same seam: the sink +# that serves a node's own metrics when there is no master to ship them to, and +# the rule deciding whether the standalone server binds a port for it at all. +add_executable(test_prometheus_metric_sink test_prometheus_metric_sink.cpp) +target_link_libraries(test_prometheus_metric_sink PRIVATE umbp_common + GTest::gtest_main) +target_compile_features(test_prometheus_metric_sink PRIVATE cxx_std_17) +add_test(NAME test_prometheus_metric_sink COMMAND test_prometheus_metric_sink) + # test_gds_engine — GdsEngine routing/planning (no GPU) plus a GPU + hipFile # read round trip (skipped without a device, or on a filesystem that rejects # O_DIRECT). Built only when the hipfile headers are installed; umbp_common diff --git a/tests/cpp/umbp/distributed/test_prometheus_metric_sink.cpp b/tests/cpp/umbp/distributed/test_prometheus_metric_sink.cpp new file mode 100644 index 000000000..732dcc7f1 --- /dev/null +++ b/tests/cpp/umbp/distributed/test_prometheus_metric_sink.cpp @@ -0,0 +1,250 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +// --------------------------------------------------------------------------- +// The masterless observability path. +// +// What used to happen: everything a peer measured went to MasterClient, which +// shipped it to a master. A node with no master measured all of it and then +// had nowhere to put it, so the standalone server — the deployment sglang's +// direct external linker actually runs — served no /metrics at all and no +// flag could make it. +// +// These tests cover the two halves of the fix that can be asserted without +// standing up a server: the sink that writes into a local MetricsServer, and +// the port-resolution rule that decides whether one is created. The wiring +// (PoolClient choosing a sink, StandaloneServer owning it) is exercised by +// the standalone server's own integration tests. +// --------------------------------------------------------------------------- + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "mori/metrics/prometheus_metrics_server.hpp" +#include "umbp/distributed/metrics/prometheus_metric_sink.h" +#include "umbp/standalone/standalone_server.h" + +namespace mori::umbp { +namespace { + +// Bind an ephemeral port, note it, and let it go: gives the test a port that +// is free right now without hard-coding one that a parallel ctest job may hold. +int PickFreePort() { + const int fd = ::socket(AF_INET, SOCK_STREAM, 0); + EXPECT_GE(fd, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = 0; + EXPECT_EQ(::bind(fd, reinterpret_cast(&addr), sizeof(addr)), 0); + socklen_t len = sizeof(addr); + EXPECT_EQ(::getsockname(fd, reinterpret_cast(&addr), &len), 0); + const int port = ntohs(addr.sin_port); + ::close(fd); + return port; +} + +std::string Scrape(int port) { + const int fd = ::socket(AF_INET, SOCK_STREAM, 0); + EXPECT_GE(fd, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons(static_cast(port)); + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (::connect(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + ::close(fd); + return {}; + } + const std::string req = "GET /metrics HTTP/1.0\r\n\r\n"; + EXPECT_GT(::send(fd, req.data(), req.size(), 0), 0); + std::string out; + char buf[4096]; + ssize_t n = 0; + while ((n = ::recv(fd, buf, sizeof(buf), 0)) > 0) out.append(buf, static_cast(n)); + ::close(fd); + return out; +} + +// RAII for an env var, so a failing assertion cannot leak it into the next test. +class ScopedEnv { + public: + ScopedEnv(const char* name, const char* value) : name_(name) { + const char* prev = std::getenv(name); + had_prev_ = prev != nullptr; + if (had_prev_) prev_ = prev; + if (value == nullptr) { + ::unsetenv(name); + } else { + ::setenv(name, value, 1); + } + } + ~ScopedEnv() { + if (had_prev_) { + ::setenv(name_, prev_.c_str(), 1); + } else { + ::unsetenv(name_); + } + } + + private: + const char* name_; + bool had_prev_ = false; + std::string prev_; +}; + +// -------------------------------------------------------------------------- +// PrometheusMetricSink +// -------------------------------------------------------------------------- + +TEST(PrometheusMetricSink, ServesCountersGaugesAndHistogramsItWasGiven) { + PrometheusMetricSink sink(PickFreePort(), "node-a"); + + sink.AddCounter("mori_umbp_test_bytes_total", "help", {{"op", "commit"}}, 4096); + sink.AddCounter("mori_umbp_test_bytes_total", "help", {{"op", "commit"}}, 1024); + sink.SetGauge("mori_umbp_test_pages", "help", {{"tier", "DRAM"}}, 17); + sink.Observe("mori_umbp_test_latency_seconds", "help", {{"rpc", "Get"}}, {0.1, 1.0}, 0.5); + + const std::string body = Scrape(sink.port()); + ASSERT_FALSE(body.empty()) << "metrics endpoint did not answer"; + + // Counter deltas accumulate rather than overwrite. + EXPECT_NE(body.find("mori_umbp_test_bytes_total"), std::string::npos); + EXPECT_NE(body.find("5120"), std::string::npos); + EXPECT_NE(body.find("mori_umbp_test_pages"), std::string::npos); + EXPECT_NE(body.find("17"), std::string::npos); + // Histograms land as a real Prometheus histogram, not a bare number. + EXPECT_NE(body.find("mori_umbp_test_latency_seconds_bucket"), std::string::npos); + EXPECT_NE(body.find("mori_umbp_test_latency_seconds_count"), std::string::npos); +} + +TEST(PrometheusMetricSink, StampsNodeLabelSoTwoServersDoNotCollapse) { + PrometheusMetricSink sink(PickFreePort(), "node-a"); + sink.AddCounter("mori_umbp_test_ops_total", "help", {{"op", "resolve"}}, 1); + + const std::string body = Scrape(sink.port()); + ASSERT_FALSE(body.empty()); + EXPECT_NE(body.find("node=\"node-a\""), std::string::npos) + << "the node label master used to add on ingest must survive the masterless path"; +} + +TEST(PrometheusMetricSink, DoesNotOverwriteANodeLabelTheComponentSupplied) { + PrometheusMetricSink sink(PickFreePort(), "node-a"); + // A series that already speaks for another node — the sink is forwarding it, + // not authoring it, so it must not relabel it as its own. + sink.SetGauge("mori_umbp_test_peer_pages", "help", {{"node", "node-b"}}, 3); + + const std::string body = Scrape(sink.port()); + ASSERT_FALSE(body.empty()); + EXPECT_NE(body.find("node=\"node-b\""), std::string::npos); + EXPECT_EQ(body.find("node=\"node-a\""), std::string::npos); +} + +TEST(PrometheusMetricSink, EmptyNodeIdStampsNothing) { + PrometheusMetricSink sink(PickFreePort(), ""); + sink.AddCounter("mori_umbp_test_ops_total", "help", {{"op", "evict"}}, 1); + + const std::string body = Scrape(sink.port()); + ASSERT_FALSE(body.empty()); + EXPECT_EQ(body.find("node=\"\""), std::string::npos); +} + +// -------------------------------------------------------------------------- +// Port resolution +// +// The distinction these pin down: "serve nothing" and "the operator typed +// something impossible" must not produce the same outcome, because the first +// is a choice and the second is a typo that would otherwise be discovered as +// a blank dashboard days later. +// -------------------------------------------------------------------------- + +TEST(StandaloneMetricsPort, UnsetUsesTheDefault) { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", nullptr); + bool ok = false; + std::string error; + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), + standalone::kDefaultStandaloneMetricsPort); + EXPECT_TRUE(ok); + EXPECT_TRUE(error.empty()); +} + +TEST(StandaloneMetricsPort, DefaultIsNotTheMasterPort) { + // A node may run a master and a standalone server; sharing 9091 would make + // one of them fail to bind for reasons neither reports clearly. + EXPECT_NE(standalone::kDefaultStandaloneMetricsPort, 9091); +} + +TEST(StandaloneMetricsPort, ExplicitPortIsHonored) { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "19099"); + bool ok = false; + std::string error; + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 19099); + EXPECT_TRUE(ok); +} + +TEST(StandaloneMetricsPort, ZeroAndWordsDisableIt) { + bool ok = false; + std::string error; + { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "0"); + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 0); + EXPECT_TRUE(ok); + } + { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "OFF"); + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 0); + EXPECT_TRUE(ok) << "case-insensitive: an operator writing OFF meant off"; + } + { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "false"); + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 0); + EXPECT_TRUE(ok); + } +} + +TEST(StandaloneMetricsPort, GarbageIsRejectedNotSilentlyDisabled) { + bool ok = true; + std::string error; + { + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "9O92"); // letter O, not zero + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 0); + EXPECT_FALSE(ok); + EXPECT_FALSE(error.empty()); + } + { + ok = true; + error.clear(); + ScopedEnv env("UMBP_STANDALONE_METRICS_PORT", "70000"); + EXPECT_EQ(standalone::ResolveMetricsPortFromEnv(&ok, &error), 0); + EXPECT_FALSE(ok); + EXPECT_FALSE(error.empty()); + } +} + +} // namespace +} // namespace mori::umbp From b99735899fe7d4009d52854b5e8e79537cf50e1c Mon Sep 17 00:00:00 2001 From: TianDi101 Date: Thu, 17 Sep 2026 03:05:54 +0000 Subject: [PATCH 2/4] fix(umbp): reject a mistyped metrics port, and make the masterless endpoint reachable Verifying the masterless metrics endpoint end to end turned up one real defect and two missing pieces of deployment glue. ResolveMetricsPortFromEnv used std::stol, which parses a numeric prefix and reports success for what it read. UMBP_STANDALONE_METRICS_PORT=9O92 (letter O) therefore resolved to port 9 -- privileged, so the bind fails, so the operator gets no metrics and an ERROR line about a port they never typed. That is exactly the "typo discovered as a blank dashboard days later" the resolver was written to prevent, so it now requires the whole string to be consumed. This is what the GarbageIsRejectedNotSilentlyDisabled test was already asserting and the only one of the nine that failed. Nothing told Prometheus the endpoint exists. examples/monitoring shipped one scrape fragment, for the master's 9091, so a masterless node published on 9092 and was never scraped -- the metrics reached a port and stopped there. Add the standalone fragment beside it, and say in the README which deployment wants which, with the point that the dashboards do not change: they are written against the series, and the server stamps the same `node` label the master used to add on ingest. UMBP_STANDALONE_METRICS_PORT was also absent from runtime-env-vars.md, where every other UMBP knob is documented. Verified on smci355-n09-29: a masterless standalone server (DRAM tier, no UMBP_MASTER_ADDRESS) serves /metrics on 9092, and after 64 x 1 MiB Put + Get through it reports backend bytes 67108864 committed and 67108864 resolved, every series carrying node="n09-29-test". 9/9 unit tests pass. Co-Authored-By: Claude Opus 5 --- examples/monitoring/README.md | 15 ++++++++++++++- examples/monitoring/umbp_standalone.scrape.yml | 16 ++++++++++++++++ src/umbp/doc/runtime-env-vars.md | 1 + src/umbp/standalone/standalone_server.cpp | 14 +++++++++++++- 4 files changed, 44 insertions(+), 2 deletions(-) create mode 100644 examples/monitoring/umbp_standalone.scrape.yml diff --git a/examples/monitoring/README.md b/examples/monitoring/README.md index 6600a0766..43d242691 100644 --- a/examples/monitoring/README.md +++ b/examples/monitoring/README.md @@ -1,10 +1,23 @@ # UMBP monitoring -A Prometheus scrape fragment and the Grafana dashboards for a UMBP deployment. +Prometheus scrape fragments and the Grafana dashboards for a UMBP deployment. Point Grafana at `grafana/dashboards/` and Prometheus at `umbp_master.scrape.yml`; `example/docker-compose.yml` is a working stack that does both. +Which fragment you want depends on where the metrics come out: + +| Deployment | Fragment | Port | +| --- | --- | --- | +| With a master | `umbp_master.scrape.yml` | `9091` | +| Masterless (standalone server) | `umbp_standalone.scrape.yml` | `9092` | + +A masterless node has no master to report to, so the standalone server serves +its own `/metrics` (see `UMBP_STANDALONE_METRICS_PORT`) and stamps the `node` +label the master used to add on ingest. **The dashboards below are the same +either way** — they are written against the series, not against the deployment +that published them. + ## The dashboards | Dashboard | Covers | diff --git a/examples/monitoring/umbp_standalone.scrape.yml b/examples/monitoring/umbp_standalone.scrape.yml new file mode 100644 index 000000000..00c28cef4 --- /dev/null +++ b/examples/monitoring/umbp_standalone.scrape.yml @@ -0,0 +1,16 @@ +# Prometheus scrape fragment for a MASTERLESS UMBP node, where the standalone +# server serves its own metrics instead of reporting them to a master. +# Drop this file into the directory pointed to by `scrape_config_files` +# in your prometheus.yml (requires Prometheus >= 2.43). +# +# The port is 9092, not the master's 9091, so a node running both is scraped +# twice without the two processes contending for one port. It follows +# UMBP_STANDALONE_METRICS_PORT; change the target here if you change that. +# +# Every series a standalone server publishes already carries a `node` label +# (the server stamps the one the master used to add on ingest), so listing +# several nodes here keeps them as separate series rather than collapsing them. +scrape_configs: + - job_name: umbp_standalone + static_configs: + - targets: ['127.0.0.1:9092'] diff --git a/src/umbp/doc/runtime-env-vars.md b/src/umbp/doc/runtime-env-vars.md index 384a58eb6..235a22db8 100644 --- a/src/umbp/doc/runtime-env-vars.md +++ b/src/umbp/doc/runtime-env-vars.md @@ -80,6 +80,7 @@ that has loaded `libmori_pybinds.so`). | `UMBP_RPC_SHUTDOWN_TIMEOUT_MS` | `3000` | ms | Deadline for `UnregisterClient` and the last `Heartbeat` in `~MasterClient`. Bounds `~MasterClient` worst-case at ≤ 2 × this value. | | `UMBP_GRPC_SHUTDOWN_DEADLINE_SEC` | `3` | sec | `server_->Shutdown(deadline)` budget, shared by master and peer service. | | `UMBP_METRICS_REPORT_INTERVAL_MS` | `1000` | ms | Cadence at which the pool client's `MasterClient` flushes buffered counters/gauges/histograms via `ReportMetrics`. | +| `UMBP_STANDALONE_METRICS_PORT` | `9092` | port | Port the **standalone server** serves `/metrics` on when it runs **without a master**, so a masterless node publishes what it measures instead of discarding it. Not the master's `9091`, so a node running both does not have two processes contending for one port. `0`/`off`/`false`/`no` (case-insensitive) disables the endpoint; anything else that is not a port number is rejected at startup rather than read as "off". Ignored when a master is configured — the metrics go to the master then, and a second endpoint would double-count on any Prometheus scraping both. A failed bind logs `ERROR` and the server still starts. Scrape fragment: `examples/monitoring/umbp_standalone.scrape.yml`. | | `UMBP_RELEASE_LEASE_MAX_RETRIES` | `2` | count | `ReleaseSsdLease` RPC attempt cap on the SSD read path. `min_allowed=1`. | | `UMBP_SSD_GET_MAX_ATTEMPTS` | `1` | count | Total remote SSD get attempts per key. `1` = no retry. Only NO_SLOT and a reader-local lease expiry retry; rpc failure / NOT_FOUND do not. Raise to absorb staging-slot contention. `min_allowed=1`. | | `UMBP_SSD_GET_RETRY_BACKOFF_MS` | `2` | ms | Sleep between remote SSD get retries (only applied when another attempt follows). `min_allowed=1`. | diff --git a/src/umbp/standalone/standalone_server.cpp b/src/umbp/standalone/standalone_server.cpp index ab7f03e31..46d73b8ca 100644 --- a/src/umbp/standalone/standalone_server.cpp +++ b/src/umbp/standalone/standalone_server.cpp @@ -1665,7 +1665,19 @@ int ResolveMetricsPortFromEnv(bool* ok, std::string* error) { if (value == "off" || value == "false" || value == "no") return 0; try { - const long port = std::stol(value); + std::size_t consumed = 0; + const long port = std::stol(value, &consumed); + // std::stol stops at the first character it cannot use and reports success + // for what it did read, so "9O92" (letter O) would resolve to port 9 and + // bind -- or fail to bind -- somewhere nobody is looking. Anything left + // over means the operator wrote something other than a port number. + if (consumed != value.size()) { + if (ok != nullptr) *ok = false; + if (error != nullptr) { + *error = "UMBP_STANDALONE_METRICS_PORT is not a number: " + std::string(raw); + } + return 0; + } if (port == 0) return 0; if (port < 0 || port > 65535) { if (ok != nullptr) *ok = false; From 35a5fe18be7907a1747d9fc627f4ac91dcfb45d5 Mon Sep 17 00:00:00 2001 From: TianDi101 Date: Thu, 17 Sep 2026 03:14:40 +0000 Subject: [PATCH 3/4] feat(umbp): report offload/load/evict/promote/demote in one periodic INFO line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answering "is this node evicting?" meant turning on DEBUG, which prints a line per key: enough volume under steady offload to change the timing of the thing being measured, and it drags every other DEBUG line in the process along with it. The numbers were all being measured already -- they just had nowhere to go in a run with no Prometheus attached, which is every run that is hard to debug: a ctest, a one-shot benchmark, an sglang rank someone is bisecting, a customer node. ActivitySummary renders one window of that. It owns no thread, reads no clock and calls no logger: it takes a cumulative snapshot and returns text, so the deltas, the units and the idle case are unit-tested without standing up a pool. PoolClient drives it on its own timer, in EVERY deployment -- master, masterless or embedded -- because it is precisely the no-destination case that needs it. UMBP_SUMMARY_INTERVAL_MS (default 60000, 0 disables); Shutdown joins it before tearing down the backends it samples, with a final render so a run shorter than one window still says what it did. The vocabulary is the deployment's, not the slot lifecycle's: offload = commit, load = resolve, evict = evict, with promote/demote the intra-pool transitions. They are read through SampleMetrics(), not new accessors, so a new medium lands in the summary with no code of its own; MetricPublisher is deliberately not shared, since its delta baselines belong to the Prometheus path and one consumer would steal the other's window. Local eviction turned out to be invisible to metrics entirely. A page-backed medium reclaiming to its low watermark frees pages inline instead of calling its own Evict(), so no interface call is ever made and op="evict" stayed at zero however much a masterless node reclaimed -- the DEBUG line was the only witness there has ever been. PageBackend now publishes it as a medium event, which also means it reaches Grafana for the first time. Removed, because the summary now says the same thing better: the per-round eviction line, and the eight per-key lines on the re-cache path. Their give-up reasons became counters instead, which is the part DEBUG was worst at -- every one of those paths is a best-effort return that leaves the read it was helping perfectly correct, so a dropped re-cache was invisible unless you were already printing a line per key. Each reason has a different fix (a bigger queue, a wider admission policy, a larger tier), so they stay distinguishable. An idle window still logs, carrying resident bytes and key count: a node holding 40 000 keys that moved none of them for a minute is the stall this is for, and is indistinguishable from a healthy quiet node unless something says how much is sitting there. Verified on smci355-n09-29, 400 x 1 MiB into a 256 MiB tier: 8.0s: offload 400.00 MiB/400 keys (50.0 MiB/s) · load 192.00 MiB/192 keys (24.0 MiB/s) miss=208 · evict 208.00 MiB/208 keys · promote 0 B · demote 0 B DRAM resident 192.00 MiB/256.00 MiB (75%) 192 keys 400 offloaded - 208 evicted = 192 resident = 192 loaded, and the 208 evicted are exactly the 208 misses. 13/13 new unit tests pass. ctest against the parent commit in the same container: 11 failures there, 10 here, the difference being test_prometheus_metric_sink, which the previous commit fixed. No new failures. The 10 are environmental -- they die in SetUp on "rsmi_num_monitor_devices reported 0 GPUs", this node's 8 GPUs being fully committed to another job. Co-Authored-By: Claude Opus 5 --- src/umbp/CMakeLists.txt | 1 + .../distributed/metrics/activity_summary.cpp | 275 ++++++++++++++++++ .../distributed/peer/backend/page_backend.cpp | 48 ++- src/umbp/distributed/pool_client.cpp | 176 ++++++++++- src/umbp/doc/runtime-env-vars.md | 1 + .../distributed/metrics/activity_summary.h | 132 +++++++++ .../distributed/metrics/component_metrics.h | 7 + .../distributed/peer/backend/page_backend.h | 18 ++ .../include/umbp/distributed/pool_client.h | 49 ++++ tests/cpp/umbp/distributed/CMakeLists.txt | 8 + .../distributed/test_activity_summary.cpp | 221 ++++++++++++++ 11 files changed, 922 insertions(+), 14 deletions(-) create mode 100644 src/umbp/distributed/metrics/activity_summary.cpp create mode 100644 src/umbp/include/umbp/distributed/metrics/activity_summary.h create mode 100644 tests/cpp/umbp/distributed/test_activity_summary.cpp diff --git a/src/umbp/CMakeLists.txt b/src/umbp/CMakeLists.txt index ec8fc82d8..9d223e1a4 100644 --- a/src/umbp/CMakeLists.txt +++ b/src/umbp/CMakeLists.txt @@ -385,6 +385,7 @@ add_library( distributed/routing/route_put_strategy.cpp distributed/metrics/component_metrics.cpp distributed/metrics/prometheus_metric_sink.cpp + distributed/metrics/activity_summary.cpp distributed/peer/backend/page_backend.cpp distributed/peer/backend/hbm_backend.cpp distributed/peer/backend/ssd_backend.cpp diff --git a/src/umbp/distributed/metrics/activity_summary.cpp b/src/umbp/distributed/metrics/activity_summary.cpp new file mode 100644 index 000000000..4bc44bfe2 --- /dev/null +++ b/src/umbp/distributed/metrics/activity_summary.cpp @@ -0,0 +1,275 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +#include "umbp/distributed/metrics/activity_summary.h" + +#include +#include +#include +#include + +namespace mori::umbp { + +namespace { + +// Counters only ever go up, but a snapshot can still come back smaller than +// its predecessor: a backend can be unregistered and re-registered, and a tier +// that disappears and returns starts from zero. Clamping at zero prints a +// quiet 0 for that window instead of an enormous number that reads like a +// catastrophic regression. +uint64_t Delta(uint64_t now, uint64_t before) { return now >= before ? now - before : now; } + +std::string Rate(uint64_t bytes, double seconds) { + if (seconds <= 0.0 || bytes == 0) return {}; + const double mib_per_s = (static_cast(bytes) / (1024.0 * 1024.0)) / seconds; + std::array buf{}; + std::snprintf(buf.data(), buf.size(), " (%.1f MiB/s)", mib_per_s); + return std::string(buf.data()); +} + +// "offload 1.19 GiB/1220 keys" — bytes first because that is the number that +// decides whether a tier is doing anything, key count second because that is +// the number that decides whether the batching is sane. +std::string Movement(const char* verb, uint64_t bytes, uint64_t keys, double seconds) { + std::array buf{}; + const std::string amount = ActivitySummary::FormatBytes(bytes); + std::snprintf(buf.data(), buf.size(), "%s %s/%llu keys%s", verb, amount.c_str(), + static_cast(keys), Rate(bytes, seconds).c_str()); + return std::string(buf.data()); +} + +bool Moved(const TierActivity& d) { + return d.offload_bytes != 0 || d.offload_keys != 0 || d.load_bytes != 0 || d.load_keys != 0 || + d.load_misses != 0 || d.evict_bytes != 0 || d.evict_keys != 0; +} + +TierActivity DiffTier(const TierActivity& now, const TierActivity& before) { + TierActivity d; + d.offload_keys = Delta(now.offload_keys, before.offload_keys); + d.offload_bytes = Delta(now.offload_bytes, before.offload_bytes); + d.load_keys = Delta(now.load_keys, before.load_keys); + d.load_bytes = Delta(now.load_bytes, before.load_bytes); + d.load_misses = Delta(now.load_misses, before.load_misses); + d.evict_keys = Delta(now.evict_keys, before.evict_keys); + d.evict_bytes = Delta(now.evict_bytes, before.evict_bytes); + // Gauges pass through untouched: differencing "how full is the tier" would + // turn the one reading that answers "are we wedged or just idle" into noise. + d.capacity_bytes = now.capacity_bytes; + d.used_bytes = now.used_bytes; + d.resident_keys = now.resident_keys; + return d; +} + +} // namespace + +std::string ActivitySummary::FormatBytes(uint64_t bytes) { + static constexpr std::array kUnits{"B", "KiB", "MiB", "GiB", "TiB"}; + if (bytes < 1024) { + std::array buf{}; + std::snprintf(buf.data(), buf.size(), "%llu B", static_cast(bytes)); + return std::string(buf.data()); + } + double value = static_cast(bytes); + size_t unit = 0; + while (value >= 1024.0 && unit + 1 < kUnits.size()) { + value /= 1024.0; + ++unit; + } + std::array buf{}; + std::snprintf(buf.data(), buf.size(), "%.2f %s", value, kUnits[unit]); + return std::string(buf.data()); +} + +std::vector ActivitySummary::Render(const ActivitySnapshot& now, + double window_seconds) { + // A named empty baseline rather than a temporary in the conditional: the + // ternary form materializes a full copy of previous_ every window (the two + // operands differ in value category), which is a map copy per tick to express + // "compare against zero". + static const ActivitySnapshot kNothingYet; + const ActivitySnapshot& before = has_previous_ ? previous_ : kNothingYet; + + // ---- aggregate across tiers ------------------------------------------- + TierActivity total; + std::map per_tier; + for (const auto& [tier, sample] : now.tiers) { + const auto it = before.tiers.find(tier); + const TierActivity baseline = it == before.tiers.end() ? TierActivity{} : it->second; + TierActivity d = DiffTier(sample, baseline); + total.offload_keys += d.offload_keys; + total.offload_bytes += d.offload_bytes; + total.load_keys += d.load_keys; + total.load_bytes += d.load_bytes; + total.load_misses += d.load_misses; + total.evict_keys += d.evict_keys; + total.evict_bytes += d.evict_bytes; + total.capacity_bytes += d.capacity_bytes; + total.used_bytes += d.used_bytes; + total.resident_keys += d.resident_keys; + per_tier.emplace(tier, std::move(d)); + } + + const uint64_t promoted = + Delta(now.transitions.promoted_bytes, before.transitions.promoted_bytes); + const uint64_t demoted = + Delta(now.transitions.offloaded_bytes, before.transitions.offloaded_bytes); + const uint64_t moves_ok = Delta(now.transitions.succeeded, before.transitions.succeeded); + const uint64_t moves_failed = Delta(now.transitions.failed, before.transitions.failed); + + const ReCacheActivity& rc_now = now.recache; + const ReCacheActivity& rc_old = before.recache; + const uint64_t rc_admitted = Delta(rc_now.admitted, rc_old.admitted); + const uint64_t rc_installed = Delta(rc_now.installed, rc_old.installed); + const uint64_t rc_rejected = Delta(rc_now.rejected, rc_old.rejected); + const uint64_t rc_queue_full = Delta(rc_now.queue_full, rc_old.queue_full); + const uint64_t rc_alloc_failed = Delta(rc_now.alloc_failed, rc_old.alloc_failed); + const uint64_t rc_install_failed = Delta(rc_now.install_failed, rc_old.install_failed); + const uint64_t rc_already = Delta(rc_now.already_present, rc_old.already_present); + const uint64_t rc_prefetched = Delta(rc_now.prefetched, rc_old.prefetched); + const uint64_t rc_fragmented = Delta(rc_now.fragmented, rc_old.fragmented); + const uint64_t rc_dropped = rc_rejected + rc_queue_full + rc_alloc_failed + rc_install_failed; + + // Adopt the new baseline only now that every delta above has been reduced to + // a value: `before` aliases previous_, so assigning earlier would leave the + // remaining reads pointing at the window that just ended. + previous_ = now; + has_previous_ = true; + + std::array window{}; + std::snprintf(window.data(), window.size(), "%.1fs", window_seconds); + + const bool anything = Moved(total) || promoted != 0 || demoted != 0 || moves_failed != 0 || + rc_admitted != 0 || rc_dropped != 0 || rc_prefetched != 0 || + rc_fragmented != 0 || rc_already != 0; + + std::vector lines; + + // ---- the idle case ----------------------------------------------------- + // Still a line, and still at INFO. A node holding 40 000 keys that moved + // none of them for a minute is the signature of the stall this summary was + // added to catch, and it is indistinguishable from a healthy quiet node + // unless something says how much is resident while nothing happens. + if (!anything) { + std::string line = std::string("[UMBP] ") + window.data() + ": idle — no offload/load/evict"; + if (total.resident_keys != 0 || total.used_bytes != 0) { + line += ", resident " + FormatBytes(total.used_bytes); + if (total.capacity_bytes > 0) { + std::array pct{}; + std::snprintf(pct.data(), pct.size(), " (%.0f%% of ", + 100.0 * static_cast(total.used_bytes) / + static_cast(total.capacity_bytes)); + line += std::string(pct.data()) + FormatBytes(total.capacity_bytes) + ")"; + } + std::array keys{}; + std::snprintf(keys.data(), keys.size(), " in %llu keys", + static_cast(total.resident_keys)); + line += keys.data(); + } else { + line += ", tiers empty"; + } + lines.push_back(std::move(line)); + return lines; + } + + // ---- what moved -------------------------------------------------------- + std::string head = std::string("[UMBP] ") + window.data() + ": " + + Movement("offload", total.offload_bytes, total.offload_keys, window_seconds) + + " · " + Movement("load", total.load_bytes, total.load_keys, window_seconds); + if (total.load_misses != 0) { + std::array miss{}; + std::snprintf(miss.data(), miss.size(), " miss=%llu", + static_cast(total.load_misses)); + head += miss.data(); + } + head += " · " + Movement("evict", total.evict_bytes, total.evict_keys, window_seconds); + + // Promotion and demotion are the SAME mechanism in opposite directions, so + // they are printed as a pair even when one of them is zero: "promote 0 B" + // next to a large demote is the shape of a tier graph that only sheds. + head += " · promote " + FormatBytes(promoted) + " · demote " + FormatBytes(demoted); + if (moves_ok != 0 || moves_failed != 0) { + std::array moves{}; + std::snprintf(moves.data(), moves.size(), " (%llu moved, %llu failed)", + static_cast(moves_ok), + static_cast(moves_failed)); + head += moves.data(); + } + lines.push_back(std::move(head)); + + // ---- the local-tier admission path ------------------------------------- + // Only when it did something, because a deployment that never re-caches + // (no remote peers, prefetch off) would otherwise carry a line of zeroes + // forever. + if (rc_admitted != 0 || rc_dropped != 0 || rc_prefetched != 0 || rc_already != 0 || + rc_fragmented != 0) { + std::array buf{}; + std::snprintf( + buf.data(), buf.size(), + "[UMBP] recache: %llu installed / %llu admitted (already=%llu, " + "prefetched=%llu); dropped %llu (rejected=%llu queue_full=%llu " + "no_memory=%llu install_failed=%llu fragmented=%llu)", + static_cast(rc_installed), static_cast(rc_admitted), + static_cast(rc_already), static_cast(rc_prefetched), + static_cast(rc_dropped), static_cast(rc_rejected), + static_cast(rc_queue_full), + static_cast(rc_alloc_failed), + static_cast(rc_install_failed), + static_cast(rc_fragmented)); + lines.emplace_back(buf.data()); + } + + // ---- per tier ---------------------------------------------------------- + // A single tier is already fully described by the total line above, so it is + // not repeated; with two or more, which tier absorbed the traffic is the + // whole question. + if (per_tier.size() > 1) { + for (const auto& [tier, d] : per_tier) { + std::array buf{}; + const std::string used = FormatBytes(d.used_bytes); + const std::string cap = FormatBytes(d.capacity_bytes); + const double pct = d.capacity_bytes == 0 ? 0.0 + : 100.0 * static_cast(d.used_bytes) / + static_cast(d.capacity_bytes); + std::snprintf(buf.data(), buf.size(), + "[UMBP] %s: offload %s load %s evict %s | resident %s/%s (%.0f%%) %llu keys", + tier.c_str(), FormatBytes(d.offload_bytes).c_str(), + FormatBytes(d.load_bytes).c_str(), FormatBytes(d.evict_bytes).c_str(), + used.c_str(), cap.c_str(), pct, + static_cast(d.resident_keys)); + lines.emplace_back(buf.data()); + } + } else if (per_tier.size() == 1) { + const auto& [tier, d] = *per_tier.begin(); + std::array buf{}; + const double pct = d.capacity_bytes == 0 ? 0.0 + : 100.0 * static_cast(d.used_bytes) / + static_cast(d.capacity_bytes); + std::snprintf(buf.data(), buf.size(), "[UMBP] %s resident %s/%s (%.0f%%) %llu keys", + tier.c_str(), FormatBytes(d.used_bytes).c_str(), + FormatBytes(d.capacity_bytes).c_str(), pct, + static_cast(d.resident_keys)); + lines.emplace_back(buf.data()); + } + + return lines; +} + +} // namespace mori::umbp diff --git a/src/umbp/distributed/peer/backend/page_backend.cpp b/src/umbp/distributed/peer/backend/page_backend.cpp index 7ade4ab15..91a6654e6 100644 --- a/src/umbp/distributed/peer/backend/page_backend.cpp +++ b/src/umbp/distributed/peer/backend/page_backend.cpp @@ -30,6 +30,7 @@ #include #include "mori/utils/mori_log.hpp" +#include "umbp/distributed/metrics/component_metrics.h" #include "umbp/distributed/transfer/transfer_engine.h" namespace mori::umbp { @@ -757,6 +758,42 @@ void PageBackend::SetEventPublishing(bool enabled) { // Local eviction (no-master deployments; see EnableLocalEviction) // --------------------------------------------------------------------------- +std::vector PageBackend::SampleMetrics() const { + // Only what the MediumBackend interface hides. Commits, resolves and + // master-driven evictions are all interface calls and are already measured by + // InstrumentedBackend; repeating them here would double-count them. + // + // Local watermark eviction is the exception that makes this function + // necessary: a round frees pages directly instead of calling Evict(), so from + // outside this class it is indistinguishable from keys that were never + // stored. On a masterless node that is ALL of the eviction there is. + std::vector out; + + auto event = [&out](const char* name, uint64_t v) { + if (v == 0) return; + out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_MEDIUM_EVENTS_TOTAL, + MORI_UMBP_METRIC_BACKEND_MEDIUM_EVENTS_TOTAL_HELP, + {{"event", name}}, + v}); + }; + + event(kPageBackendLocalEvictEvent, local_evict_keys_.load(std::memory_order_relaxed)); + event("local_evict_round", local_evict_rounds_.load(std::memory_order_relaxed)); + // A round that found nothing reclaimable: every candidate was under a read + // lease or a copy pin. Distinguishes a tier that cannot evict from one that + // has nothing to evict, which look the same in the freed-bytes counter. + event("local_evict_no_candidate", local_evict_no_candidate_.load(std::memory_order_relaxed)); + + const uint64_t freed_bytes = local_evict_bytes_.load(std::memory_order_relaxed); + if (freed_bytes != 0) { + out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL, + MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL_HELP, + {{"event", kPageBackendLocalEvictEvent}}, + freed_bytes}); + } + return out; +} + void PageBackend::EnableLocalEviction(double high_watermark, double low_watermark) { if (!(high_watermark > 0.0 && high_watermark <= 1.0 && low_watermark > 0.0 && low_watermark < high_watermark)) { @@ -860,13 +897,20 @@ size_t PageBackend::MaybeEvictToLowWatermark() { owned_.erase(owned_it); ++freed; } + local_evict_rounds_.fetch_add(1, std::memory_order_relaxed); if (freed > 0) { - MORI_UMBP_DEBUG("[PageBackend] local eviction freed {} key(s) ({} bytes) tier={}", freed, - selected_bytes, static_cast(tier_)); + // Counted, not logged. A round frees a handful of keys and runs whenever + // a commit finds the tier above its high watermark, so the per-round line + // this replaced could reach thousands per second under steady offload — + // and it was at DEBUG, so the volume also bought every other DEBUG line in + // the process. The periodic activity summary reports the window total. + local_evict_keys_.fetch_add(freed, std::memory_order_relaxed); + local_evict_bytes_.fetch_add(selected_bytes, std::memory_order_relaxed); local_evict_warned_ = false; } else if (!local_evict_warned_) { // Latched: a wedged pool would otherwise log this on every commit. local_evict_warned_ = true; + local_evict_no_candidate_.fetch_add(1, std::memory_order_relaxed); MORI_UMBP_WARN( "[PageBackend] local eviction found no reclaimable key at {}/{} bytes used " "(all leased or pinned) tier={}", diff --git a/src/umbp/distributed/pool_client.cpp b/src/umbp/distributed/pool_client.cpp index b250e7172..f59ac52ef 100644 --- a/src/umbp/distributed/pool_client.cpp +++ b/src/umbp/distributed/pool_client.cpp @@ -1022,6 +1022,10 @@ bool PoolClient::Init() { StartLocalMetricsReporting(); } + // Runs in EVERY deployment, master or not, sink or not: it is the readout + // for the case where no metrics destination exists at all. + StartActivitySummary(); + // Pack engine_desc for master registration. std::vector engine_desc_bytes; if (peer_directory_ != nullptr) engine_desc_bytes = peer_directory_->PackedLocalEngineDesc(); @@ -1145,6 +1149,9 @@ void PoolClient::Shutdown() { // the same reason: its final tick samples the backends and transfer engine, // which the code below is about to destroy. StopLocalMetricsReporting(); + // Joined here for the same reason and at the same point: its final render + // samples the backends and the pool, which the code below destroys. + StopActivitySummary(); { std::lock_guard lock(peers_mutex_); @@ -1929,8 +1936,7 @@ void PoolClient::MaybeReCacheAfterRemote(const std::string& key, const void* src // pure predicate, unit-tested in test_cache_remote_admission.cpp. if (!ShouldAdmitReCache(config_.cache_remote_fetches, config_.cache_remote_admission, config_.admission_max_block_bytes, size)) { - MORI_UMBP_DEBUG("[PoolClient] MaybeReCacheAfterRemote: key='{}' size={} not admitted", key, - size); + recache_stats_.rejected.fetch_add(1, std::memory_order_relaxed); return; } // ALWAYS and SIZE both delegate capacity enforcement to the peer allocator: @@ -1944,8 +1950,7 @@ void PoolClient::MaybeReCacheAfterRemote(const std::string& key, const void* src job.key = key; job.bytes = std::unique_ptr(new (std::nothrow) char[size]); if (!job.bytes) { - MORI_UMBP_DEBUG("[PoolClient] MaybeReCacheAfterRemote: allocation failed for key='{}' size={}", - key, size); + recache_stats_.alloc_failed.fetch_add(1, std::memory_order_relaxed); return; } job.size = size; @@ -1960,11 +1965,12 @@ void PoolClient::MaybeReCacheAfterRemote(const std::string& key, const void* src std::lock_guard lk(recache_mutex_); if (recache_stop_) return; if (recache_queue_.size() >= recache_queue_max_) { - MORI_UMBP_DEBUG("[PoolClient] MaybeReCacheAfterRemote: queue full, dropping key='{}'", key); + recache_stats_.queue_full.fetch_add(1, std::memory_order_relaxed); return; } recache_queue_.push_back(std::move(job)); } + recache_stats_.admitted.fetch_add(1, std::memory_order_relaxed); recache_cv_.notify_one(); } @@ -2017,12 +2023,13 @@ void PoolClient::MaybePrefetchWholeObject(const std::string& key, size_t object_ std::lock_guard lk(recache_mutex_); if (recache_stop_) return; if (recache_queue_.size() >= recache_queue_max_) { - MORI_UMBP_DEBUG("[PoolClient] MaybePrefetchWholeObject: queue full, dropping key='{}'", key); + recache_stats_.queue_full.fetch_add(1, std::memory_order_relaxed); return; } if (!prefetch_inflight_.insert(key).second) return; // already queued or running recache_queue_.push_back(std::move(job)); } + recache_stats_.admitted.fetch_add(1, std::memory_order_relaxed); recache_cv_.notify_one(); } @@ -2083,8 +2090,7 @@ void PoolClient::FetchWholeObjectsIntoMedium(std::vector& jobs) { TransferRef slot_buffer = contiguous ? backend->BufferRef(alloc.pages.front().buffer_index) : TransferRef{}; if (!contiguous || !slot_buffer.Valid()) { - MORI_UMBP_DEBUG("[PoolClient] prefetch: slot for key='{}' is not one addressable run", - job.key); + recache_stats_.fragmented.fetch_add(1, std::memory_order_relaxed); // Aborted here and never revisited: the commit/abort pass below walks // `planned`, which this key does not enter. backend->BatchAbort({alloc.slot_id}); @@ -2186,7 +2192,7 @@ void PoolClient::ReCacheWorkerLoop() { std::lock_guard lk(recache_mutex_); for (const auto& done : prefetch_batch) prefetch_inflight_.erase(done.key); } - MORI_UMBP_DEBUG("[PoolClient] ReCacheWorker: prefetched {} objects", prefetch_batch.size()); + recache_stats_.prefetched.fetch_add(prefetch_batch.size(), std::memory_order_relaxed); prefetch_batch.clear(); continue; } @@ -2197,14 +2203,14 @@ void PoolClient::ReCacheWorkerLoop() { // the same key. switch (ExecuteLocalPut(job.key, job.bytes.get(), job.size, medium_)) { case PutAttemptOutcome::kSuccess: - MORI_UMBP_DEBUG("[PoolClient] ReCacheWorker: re-cached key='{}' size={}", job.key, - job.size); + recache_stats_.installed.fetch_add(1, std::memory_order_relaxed); break; case PutAttemptOutcome::kSuccessAlreadyExists: + recache_stats_.already_present.fetch_add(1, std::memory_order_relaxed); break; case PutAttemptOutcome::kRetry: case PutAttemptOutcome::kFatal: - MORI_UMBP_DEBUG("[PoolClient] ReCacheWorker: local install failed for key='{}'", job.key); + recache_stats_.install_failed.fetch_add(1, std::memory_order_relaxed); break; } } @@ -4730,6 +4736,152 @@ void PoolClient::StopLocalMetricsReporting() { MORI_UMBP_INFO("[PoolClient] local metrics reporting stopped"); } +// --------------------------------------------------------------------------- +// The periodic activity summary +// +// Everything here is already in Prometheus. This exists for the runs that +// have no Prometheus: a ctest, a one-shot benchmark, an sglang rank on a +// customer node, a bisect. In those the counters lived in atomics nothing +// ever read, and the only way to see an eviction or a dropped re-cache was +// DEBUG, which prints per key. +// --------------------------------------------------------------------------- + +ActivitySnapshot PoolClient::CollectActivitySnapshot() const { + ActivitySnapshot snap; + + for (MediumBackend* backend : registry_.All()) { + if (backend == nullptr) continue; + TierActivity& tier = snap.tiers[TierTypeName(backend->Tier())]; + + // Read the same cumulative samples the metrics tick reads, and pick out the + // ops this summary speaks about. Going through SampleMetrics() rather than + // adding accessors keeps the rule the component layer is built on: a + // backend is observed through one seam, and a new medium lands here with no + // code of its own. MetricPublisher is deliberately NOT involved -- it owns + // the delta baselines for the Prometheus path, and sharing them would make + // each consumer steal the other's window. + for (const MetricSample& sample : backend->SampleMetrics()) { + const auto label = [&sample](const char* key) -> std::string { + for (const auto& kv : sample.labels) { + if (kv.first == key) return kv.second; + } + return {}; + }; + + if (std::strcmp(sample.name, MORI_UMBP_METRIC_BACKEND_OPS_TOTAL) == 0) { + const std::string op = label("op"); + const std::string status = label("status"); + // Key counts are the successful ones: an allocate that found no space + // is not an offloaded key, and counting it as one would make a tier + // that is refusing everything look busy. + if (op == "commit" && status == "ok") tier.offload_keys += sample.value; + if (op == "resolve" && status == "ok") tier.load_keys += sample.value; + if (op == "resolve" && status == "miss") tier.load_misses += sample.value; + if (op == "evict" && status == "ok") tier.evict_keys += sample.value; + } else if (std::strcmp(sample.name, MORI_UMBP_METRIC_BACKEND_BYTES_TOTAL) == 0) { + const std::string op = label("op"); + if (op == "commit") tier.offload_bytes += sample.value; + if (op == "resolve") tier.load_bytes += sample.value; + if (op == "evict") tier.evict_bytes += sample.value; + } else if (std::strcmp(sample.name, MORI_UMBP_METRIC_BACKEND_MEDIUM_EVENTS_TOTAL) == 0 && + label("event") == kPageBackendLocalEvictEvent) { + // Added to op="evict" rather than reported separately: to the operator + // asking "is this tier reclaiming", who initiated the reclaim is + // detail. The two are disjoint -- a local round never makes the + // interface call -- so this adds, it does not double-count. + tier.evict_keys += sample.value; + } else if (std::strcmp(sample.name, MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL) == 0 && + label("event") == kPageBackendLocalEvictEvent) { + tier.evict_bytes += sample.value; + } + } + + const TierCapacity capacity = backend->Capacity(); + tier.capacity_bytes += capacity.total_bytes; + tier.used_bytes += capacity.total_bytes >= capacity.available_bytes + ? capacity.total_bytes - capacity.available_bytes + : 0; + tier.resident_keys += backend->OwnedKeyCount(); + } + + if (default_pool_ != nullptr) snap.transitions = default_pool_->TransitionMetrics(); + + snap.recache.admitted = recache_stats_.admitted.load(std::memory_order_relaxed); + snap.recache.rejected = recache_stats_.rejected.load(std::memory_order_relaxed); + snap.recache.queue_full = recache_stats_.queue_full.load(std::memory_order_relaxed); + snap.recache.alloc_failed = recache_stats_.alloc_failed.load(std::memory_order_relaxed); + snap.recache.installed = recache_stats_.installed.load(std::memory_order_relaxed); + snap.recache.already_present = recache_stats_.already_present.load(std::memory_order_relaxed); + snap.recache.install_failed = recache_stats_.install_failed.load(std::memory_order_relaxed); + snap.recache.prefetched = recache_stats_.prefetched.load(std::memory_order_relaxed); + snap.recache.fragmented = recache_stats_.fragmented.load(std::memory_order_relaxed); + return snap; +} + +void PoolClient::EmitActivitySummary(double window_seconds) { + const auto lines = activity_summary_.Render(CollectActivitySnapshot(), window_seconds); + for (const std::string& line : lines) MORI_UMBP_INFO("{}", line); +} + +void PoolClient::StartActivitySummary() { + if (summary_running_.load()) return; + // 0 disables. Unlike the metrics port, a value this cannot parse falls back + // to the default rather than refusing to start: a wrong cadence announces + // itself in the very next line the summary prints, where a silently disabled + // metrics endpoint is discovered days later as a blank dashboard. + summary_interval_ms_ = static_cast(GetEnvMilliseconds("UMBP_SUMMARY_INTERVAL_MS", + std::chrono::milliseconds(60000), + /*min_allowed=*/0) + .count()); + if (summary_interval_ms_ == 0) { + MORI_UMBP_INFO("[PoolClient] periodic activity summary disabled (UMBP_SUMMARY_INTERVAL_MS=0)"); + return; + } + summary_last_ = std::chrono::steady_clock::now(); + summary_running_ = true; + try { + summary_thread_ = std::thread(&PoolClient::ActivitySummaryLoop, this); + } catch (const std::system_error& e) { + summary_running_ = false; + MORI_UMBP_ERROR("[PoolClient] failed to start activity summary thread: {}", e.what()); + return; + } + MORI_UMBP_INFO("[PoolClient] periodic activity summary every {}ms", summary_interval_ms_); +} + +void PoolClient::StopActivitySummary() { + if (!summary_running_.load()) return; + summary_running_ = false; + summary_cv_.notify_one(); + if (summary_thread_.joinable()) summary_thread_.join(); +} + +void PoolClient::ActivitySummaryLoop() { + while (summary_running_.load()) { + { + std::unique_lock lock(summary_mu_); + summary_cv_.wait_for(lock, std::chrono::milliseconds(summary_interval_ms_), + [this] { return !summary_running_.load(); }); + } + if (!summary_running_.load()) break; + const auto now = std::chrono::steady_clock::now(); + // The MEASURED window, not the configured one: a tick delayed behind a + // stalled backend should read as a long window rather than as a rate + // computed against a cadence that did not happen. + const double elapsed = + std::chrono::duration_cast>(now - summary_last_).count(); + summary_last_ = now; + EmitActivitySummary(elapsed); + } + // A final line before the backends are torn down, for the same reason the + // metrics tick has one: a run shorter than one window would otherwise report + // nothing at all, and that run is usually the unit test someone is trying to + // understand. + const auto now = std::chrono::steady_clock::now(); + EmitActivitySummary( + std::chrono::duration_cast>(now - summary_last_).count()); +} + void PoolClient::LocalMetricsLoop() { while (local_metrics_running_.load()) { { diff --git a/src/umbp/doc/runtime-env-vars.md b/src/umbp/doc/runtime-env-vars.md index 235a22db8..b625b9053 100644 --- a/src/umbp/doc/runtime-env-vars.md +++ b/src/umbp/doc/runtime-env-vars.md @@ -81,6 +81,7 @@ that has loaded `libmori_pybinds.so`). | `UMBP_GRPC_SHUTDOWN_DEADLINE_SEC` | `3` | sec | `server_->Shutdown(deadline)` budget, shared by master and peer service. | | `UMBP_METRICS_REPORT_INTERVAL_MS` | `1000` | ms | Cadence at which the pool client's `MasterClient` flushes buffered counters/gauges/histograms via `ReportMetrics`. | | `UMBP_STANDALONE_METRICS_PORT` | `9092` | port | Port the **standalone server** serves `/metrics` on when it runs **without a master**, so a masterless node publishes what it measures instead of discarding it. Not the master's `9091`, so a node running both does not have two processes contending for one port. `0`/`off`/`false`/`no` (case-insensitive) disables the endpoint; anything else that is not a port number is rejected at startup rather than read as "off". Ignored when a master is configured — the metrics go to the master then, and a second endpoint would double-count on any Prometheus scraping both. A failed bind logs `ERROR` and the server still starts. Scrape fragment: `examples/monitoring/umbp_standalone.scrape.yml`. | +| `UMBP_SUMMARY_INTERVAL_MS` | `60000` | ms | Cadence of the periodic **activity summary**: one INFO line per window saying what this process offloaded, loaded, evicted, promoted and demoted, plus what is resident. Runs in **every** deployment — master, masterless, embedded — because it is the readout for the runs that have no Prometheus at all (a ctest, a one-shot benchmark, an sglang rank being bisected). `0` disables it. A window in which nothing moved still logs one line, carrying resident bytes and key count, because a node holding keys and moving none of them is the stall this is meant to catch. It replaced the per-key `DEBUG` logs on the re-cache and local-eviction paths; an unparseable value falls back to the default rather than refusing to start. | | `UMBP_RELEASE_LEASE_MAX_RETRIES` | `2` | count | `ReleaseSsdLease` RPC attempt cap on the SSD read path. `min_allowed=1`. | | `UMBP_SSD_GET_MAX_ATTEMPTS` | `1` | count | Total remote SSD get attempts per key. `1` = no retry. Only NO_SLOT and a reader-local lease expiry retry; rpc failure / NOT_FOUND do not. Raise to absorb staging-slot contention. `min_allowed=1`. | | `UMBP_SSD_GET_RETRY_BACKOFF_MS` | `2` | ms | Sleep between remote SSD get retries (only applied when another attempt follows). `min_allowed=1`. | diff --git a/src/umbp/include/umbp/distributed/metrics/activity_summary.h b/src/umbp/include/umbp/distributed/metrics/activity_summary.h new file mode 100644 index 000000000..c6eb3b07c --- /dev/null +++ b/src/umbp/include/umbp/distributed/metrics/activity_summary.h @@ -0,0 +1,132 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +#pragma once + +#include +#include +#include +#include + +#include "umbp/distributed/types.h" + +// --------------------------------------------------------------------------- +// The periodic INFO line a UMBP process writes about itself. +// +// Why this exists at all: everything below is already measured and already +// published to Prometheus. But the deployments that are hardest to debug are +// exactly the ones with no Prometheus attached — a unit test, a one-off +// benchmark, a customer's node, an sglang rank someone is bisecting — and in +// those the numbers sat in atomics that nothing ever read. The alternative +// was DEBUG, which is per-key: turning it on to answer "is anything being +// evicted?" produced megabytes of lines per second and changed the timing of +// the thing under test. One aggregated line per window answers that question +// at the level it is actually asked, at a volume a long run can afford. +// +// RENDERING ONLY. This class owns no thread, reads no clock, opens no file +// and calls no logger: it is handed a cumulative snapshot and returns the +// text. That is what lets the interesting part — the deltas, the units, what +// an idle window says — be unit-tested without standing up a pool. +// --------------------------------------------------------------------------- + +namespace mori::umbp { + +// One tier's cumulative counters, summed over every backend of that tier. +// +// The three verbs are the storage-backend ops seen from the caller's side, and +// they are named here the way a UMBP deployment talks about them rather than +// the way the slot lifecycle does: +// +// offload = commit bytes written INTO this tier (sglang: KV leaving HBM) +// load = resolve bytes handed OUT to a reader (sglang: load-back) +// evict = evict bytes reclaimed +// +// `load_misses` is the resolve calls that found nothing. It is here because a +// tier that is loading nothing and a tier that is being asked for keys it does +// not have look identical in a bytes counter, and they are completely +// different bugs. +struct TierActivity { + uint64_t offload_keys = 0; + uint64_t offload_bytes = 0; + uint64_t load_keys = 0; + uint64_t load_bytes = 0; + uint64_t load_misses = 0; + uint64_t evict_keys = 0; + uint64_t evict_bytes = 0; + + // Live state, not a counter: sampled, never differenced. + uint64_t capacity_bytes = 0; + uint64_t used_bytes = 0; + uint64_t resident_keys = 0; +}; + +// The local-tier admission path: an object fetched from a remote node (or +// pulled ahead by locality prefetch) being installed into this node's medium. +// That is a promotion too — it just crosses a node boundary instead of a tier +// boundary — and it is the part of the system that fails silently, because +// every one of its give-up paths is a best-effort `return` that leaves the +// read it was helping perfectly correct. +struct ReCacheActivity { + uint64_t admitted = 0; // enqueued for install + uint64_t rejected = 0; // admission policy said no + uint64_t queue_full = 0; // dropped: install queue at its bound + uint64_t alloc_failed = 0; // dropped: could not stage the bytes + uint64_t installed = 0; // landed in the local medium + uint64_t already_present = 0; // a concurrent install won + uint64_t install_failed = 0; // local put refused it + uint64_t prefetched = 0; // whole objects pulled by locality prefetch + uint64_t fragmented = 0; // prefetch slot was not one addressable run +}; + +// Everything one window needs, all cumulative since process start. +struct ActivitySnapshot { + std::map tiers; // keyed by TierTypeName + TierTransitionMetrics transitions; // intra-pool promote / demote + ReCacheActivity recache; +}; + +class ActivitySummary { + public: + // Renders `now` against the previously observed snapshot and adopts it as the + // new baseline. The FIRST call has no baseline, so it reports `now` in full: + // for a process that started with empty tiers that is the same number, and + // for one that did not, a first line that undercounts would be worse than one + // that says "this is everything so far". + // + // `window_seconds` is what the caller measured between the two samples, not + // what it configured, so a tick delayed by a stalled backend says so. + // + // Returns one line per tier plus a leading total line, or a single line when + // nothing moved. Never empty: a window in which a busy node did nothing is + // itself the finding. + std::vector Render(const ActivitySnapshot& now, double window_seconds); + + // Byte counts as a human reads them ("1.19 GiB"). Exposed for the tests, and + // because every other UMBP log that prints bytes prints them raw, which is + // how "2147483648" ends up in a bug report nobody can size at a glance. + static std::string FormatBytes(uint64_t bytes); + + private: + ActivitySnapshot previous_; + bool has_previous_ = false; +}; + +} // namespace mori::umbp diff --git a/src/umbp/include/umbp/distributed/metrics/component_metrics.h b/src/umbp/include/umbp/distributed/metrics/component_metrics.h index 031b6c493..8cc175d67 100644 --- a/src/umbp/include/umbp/distributed/metrics/component_metrics.h +++ b/src/umbp/include/umbp/distributed/metrics/component_metrics.h @@ -107,6 +107,13 @@ namespace mori::umbp { "Medium-internal events a backend chose to publish (drive read outcomes, single-flight " \ "coalescing, eviction rounds, staging pressure), keyed by event" +// The one medium event with a reader outside the dashboards, so its spelling +// is shared rather than written twice. A page-backed medium reclaiming to its +// low watermark frees pages inline instead of calling its own Evict(), so this +// event -- and not op="evict" -- is the eviction that happened on a node with +// no master. The periodic activity summary adds the two together. +inline constexpr const char* kPageBackendLocalEvictEvent = "local_evict"; + #define MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL "mori_umbp_backend_medium_bytes_total" #define MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL_HELP \ "Medium-internal byte counters (e.g. bytes that actually reached the device, as opposed " \ diff --git a/src/umbp/include/umbp/distributed/peer/backend/page_backend.h b/src/umbp/include/umbp/distributed/peer/backend/page_backend.h index d0d8840c0..f73988840 100644 --- a/src/umbp/include/umbp/distributed/peer/backend/page_backend.h +++ b/src/umbp/include/umbp/distributed/peer/backend/page_backend.h @@ -355,6 +355,12 @@ class PageBackend : public MediumBackend { // local eviction off. void EnableLocalEviction(double high_watermark, double low_watermark) override; + // What this medium's interface cannot show from outside: the local + // watermark-eviction rounds, which never become an interface call. Follows + // the generic MORI_UMBP_METRIC_BACKEND_MEDIUM_* names with the specifics in + // an event= label, so it lands in the panel that is already there. + std::vector SampleMetrics() const override; + // Test seam: run one local-eviction round synchronously. Returns the number // of keys freed (0 when local eviction is off or usage is below the high // watermark). @@ -578,6 +584,18 @@ class PageBackend : public MediumBackend { // frees something. bool local_evict_warned_ = false; + // Local high-watermark eviction is the one reclaim path NOT visible to the + // instrumentation decorator: the rounds below free pages inline rather than + // calling this backend's own Evict(), so no interface call is ever made and + // op="evict" stays at zero however much a masterless node reclaims. These + // counters are what the medium publishes about itself for that reason, and + // they are what the periodic activity summary reports as eviction on a node + // with no master. Relaxed atomics, read once per tick from another thread. + std::atomic local_evict_keys_{0}; + std::atomic local_evict_bytes_{0}; + std::atomic local_evict_rounds_{0}; + std::atomic local_evict_no_candidate_{0}; + // False stops QueueEventLocked from recording anything (no master to ship // to). True is the cluster default. bool event_publishing_ = true; diff --git a/src/umbp/include/umbp/distributed/pool_client.h b/src/umbp/include/umbp/distributed/pool_client.h index 3b5e2af80..cd0eab215 100644 --- a/src/umbp/include/umbp/distributed/pool_client.h +++ b/src/umbp/include/umbp/distributed/pool_client.h @@ -42,6 +42,7 @@ #include "mori/io/engine.hpp" #include "umbp/distributed/config.h" #include "umbp/distributed/master/master_client.h" +#include "umbp/distributed/metrics/activity_summary.h" #include "umbp/distributed/metrics/component_metrics.h" #include "umbp/distributed/peer/backend/medium_backend.h" #include "umbp/distributed/transfer/transfer_engine.h" @@ -292,6 +293,34 @@ class PoolClient { std::condition_variable local_metrics_cv_; uint64_t local_metrics_interval_ms_ = 1000; + // ---- the periodic activity summary ------------------------------------ + // One INFO line per window saying what this process offloaded, loaded, + // evicted, promoted and demoted. Deliberately NOT tied to the metrics tick + // above: that one exists only when there is somewhere to publish (a master, + // or a local Prometheus endpoint), and the deployments where this summary + // earns its keep are exactly the ones with neither. It therefore runs in + // every deployment, on its own much slower timer. + // + // Cadence: UMBP_SUMMARY_INTERVAL_MS, default 60000, 0 disables. Joined by + // Shutdown before the backends it samples are destroyed, with a final render + // so a short run still reports what it did. + void ActivitySummaryLoop(); + void StartActivitySummary(); + void StopActivitySummary(); + // Reads every backend's cumulative counters plus the pool's transition + // totals. Cheap and lock-light — SampleMetrics() is relaxed atomic loads — + // but still a tick, never a data-plane path. + ActivitySnapshot CollectActivitySnapshot() const; + void EmitActivitySummary(double window_seconds); + + ActivitySummary activity_summary_; + std::thread summary_thread_; + std::atomic summary_running_{false}; + std::mutex summary_mu_; + std::condition_variable summary_cv_; + uint64_t summary_interval_ms_ = 60000; + std::chrono::steady_clock::time_point summary_last_; + // Every storage medium live on this node. Owned here because PoolClient is // the natural lifetime anchor for the per-process IO engine + backend pools. // MasterClient borrows it for heartbeat aggregation; default_pool_ borrows it @@ -637,6 +666,26 @@ class PoolClient { std::thread recache_worker_; bool recache_stop_ = false; size_t recache_queue_max_ = 1024; + + // Outcome counts for the local-tier admission path, reported by the periodic + // activity summary. These replaced per-key DEBUG lines at each of these + // sites: every one of them is a best-effort give-up that leaves the read it + // was helping perfectly correct, so the only way to see one was to turn on a + // level that prints a line per key and changes the timing of the thing being + // measured. Relaxed atomics — read once per summary window, sampled from + // another thread, never correctness state. + struct ReCacheCounters { + std::atomic admitted{0}; + std::atomic rejected{0}; + std::atomic queue_full{0}; + std::atomic alloc_failed{0}; + std::atomic installed{0}; + std::atomic already_present{0}; + std::atomic install_failed{0}; + std::atomic prefetched{0}; + std::atomic fragmented{0}; + }; + ReCacheCounters recache_stats_; struct BatchPutItem { size_t index; const std::string* key; diff --git a/tests/cpp/umbp/distributed/CMakeLists.txt b/tests/cpp/umbp/distributed/CMakeLists.txt index c0651b380..e43993b59 100644 --- a/tests/cpp/umbp/distributed/CMakeLists.txt +++ b/tests/cpp/umbp/distributed/CMakeLists.txt @@ -489,6 +489,14 @@ target_link_libraries(test_prometheus_metric_sink PRIVATE umbp_common target_compile_features(test_prometheus_metric_sink PRIVATE cxx_std_17) add_test(NAME test_prometheus_metric_sink COMMAND test_prometheus_metric_sink) +# test_activity_summary — the periodic INFO line that replaced the per-key DEBUG +# logs on the offload / load / evict / promote / demote paths. Pure rendering, +# so the deltas, units and the idle case are testable without a pool. +add_executable(test_activity_summary test_activity_summary.cpp) +target_link_libraries(test_activity_summary PRIVATE umbp_common GTest::gtest_main) +target_compile_features(test_activity_summary PRIVATE cxx_std_17) +add_test(NAME test_activity_summary COMMAND test_activity_summary) + # test_gds_engine — GdsEngine routing/planning (no GPU) plus a GPU + hipFile # read round trip (skipped without a device, or on a filesystem that rejects # O_DIRECT). Built only when the hipfile headers are installed; umbp_common diff --git a/tests/cpp/umbp/distributed/test_activity_summary.cpp b/tests/cpp/umbp/distributed/test_activity_summary.cpp new file mode 100644 index 000000000..2d701a189 --- /dev/null +++ b/tests/cpp/umbp/distributed/test_activity_summary.cpp @@ -0,0 +1,221 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +// What the periodic INFO summary says, tested without a pool. +// +// The summary replaced per-key DEBUG lines, so the thing worth pinning down is +// not that it prints, but that a reader can tell the same things from it that +// the DEBUG lines told them: that work happened, how much, and -- the part a +// counter makes easy and a log line never did -- that a window in which nothing +// happened is itself reported. + +#include + +#include +#include + +#include "umbp/distributed/metrics/activity_summary.h" + +namespace mori::umbp { +namespace { + +std::string Joined(const std::vector& lines) { + std::string all; + for (const auto& line : lines) { + all += line; + all += '\n'; + } + return all; +} + +ActivitySnapshot WithDram(uint64_t offload_bytes, uint64_t offload_keys, uint64_t load_bytes, + uint64_t load_keys, uint64_t evict_bytes, uint64_t evict_keys) { + ActivitySnapshot snap; + TierActivity& dram = snap.tiers["DRAM"]; + dram.offload_bytes = offload_bytes; + dram.offload_keys = offload_keys; + dram.load_bytes = load_bytes; + dram.load_keys = load_keys; + dram.evict_bytes = evict_bytes; + dram.evict_keys = evict_keys; + dram.capacity_bytes = 2ULL * 1024 * 1024 * 1024; + dram.used_bytes = 1ULL * 1024 * 1024 * 1024; + dram.resident_keys = 1024; + return snap; +} + +TEST(ActivitySummary, FormatsBytesTheWayAReaderSizesThem) { + EXPECT_EQ(ActivitySummary::FormatBytes(0), "0 B"); + EXPECT_EQ(ActivitySummary::FormatBytes(512), "512 B"); + EXPECT_EQ(ActivitySummary::FormatBytes(1024), "1.00 KiB"); + EXPECT_EQ(ActivitySummary::FormatBytes(1536), "1.50 KiB"); + EXPECT_EQ(ActivitySummary::FormatBytes(1024ULL * 1024), "1.00 MiB"); + EXPECT_EQ(ActivitySummary::FormatBytes(2147483648ULL), "2.00 GiB"); +} + +TEST(ActivitySummary, FirstWindowReportsEverythingSoFar) { + // No baseline exists yet. Reporting a delta against an implicit zero is the + // only honest option; the alternative (skip the first window) loses whatever + // a short-lived process did in its entire life. + ActivitySummary summary; + const auto lines = summary.Render(WithDram(1024 * 1024, 4, 2048 * 1024, 8, 0, 0), 60.0); + const std::string all = Joined(lines); + EXPECT_NE(all.find("offload 1.00 MiB/4 keys"), std::string::npos) << all; + EXPECT_NE(all.find("load 2.00 MiB/8 keys"), std::string::npos) << all; +} + +TEST(ActivitySummary, SecondWindowReportsOnlyWhatChanged) { + ActivitySummary summary; + summary.Render(WithDram(1024 * 1024, 4, 0, 0, 0, 0), 60.0); + // Cumulative counters: 3 MiB total means 2 MiB happened in THIS window. + const auto lines = summary.Render(WithDram(3 * 1024 * 1024, 12, 0, 0, 0, 0), 60.0); + const std::string all = Joined(lines); + EXPECT_NE(all.find("offload 2.00 MiB/8 keys"), std::string::npos) << all; + EXPECT_EQ(all.find("offload 3.00 MiB"), std::string::npos) << all; +} + +TEST(ActivitySummary, AnIdleWindowStillSaysWhatIsResident) { + // The point of the idle line: a node holding 1024 keys that moved none of + // them is the shape of a stall, and is indistinguishable from a healthy quiet + // node unless the line says how much is sitting there. + ActivitySummary summary; + summary.Render(WithDram(1024 * 1024, 4, 0, 0, 0, 0), 60.0); + const auto lines = summary.Render(WithDram(1024 * 1024, 4, 0, 0, 0, 0), 60.0); + const std::string all = Joined(lines); + EXPECT_NE(all.find("idle"), std::string::npos) << all; + EXPECT_NE(all.find("1024 keys"), std::string::npos) << all; + EXPECT_NE(all.find("1.00 GiB"), std::string::npos) << all; +} + +TEST(ActivitySummary, NeverReturnsNothing) { + // A caller logs whatever comes back; an empty result would make a quiet + // window look like a dead summary thread. + ActivitySummary summary; + EXPECT_FALSE(summary.Render(ActivitySnapshot{}, 60.0).empty()); +} + +TEST(ActivitySummary, ReportsEvictionSeparatelyFromOffload) { + // These two were the pair most often confused in the DEBUG era: bytes written + // into a tier and bytes reclaimed from it both looked like "the tier is busy". + ActivitySummary summary; + summary.Render(WithDram(0, 0, 0, 0, 0, 0), 60.0); + const auto lines = summary.Render(WithDram(4 * 1024 * 1024, 4, 0, 0, 1024 * 1024, 2), 60.0); + const std::string all = Joined(lines); + EXPECT_NE(all.find("offload 4.00 MiB/4 keys"), std::string::npos) << all; + EXPECT_NE(all.find("evict 1.00 MiB/2 keys"), std::string::npos) << all; +} + +TEST(ActivitySummary, ReportsPromotionAndDemotionAsAPairEvenWhenOneIsZero) { + // A tier graph that only sheds is a real misconfiguration, and it is visible + // only if the zero is printed next to the non-zero. + ActivitySummary summary; + ActivitySnapshot first = WithDram(0, 0, 0, 0, 0, 0); + summary.Render(first, 60.0); + + ActivitySnapshot second = WithDram(0, 0, 0, 0, 0, 0); + second.transitions.offloaded_bytes = 8ULL * 1024 * 1024; + second.transitions.succeeded = 4; + second.transitions.failed = 1; + const std::string all = Joined(summary.Render(second, 60.0)); + EXPECT_NE(all.find("promote 0 B"), std::string::npos) << all; + EXPECT_NE(all.find("demote 8.00 MiB"), std::string::npos) << all; + EXPECT_NE(all.find("1 failed"), std::string::npos) << all; +} + +TEST(ActivitySummary, SurfacesLoadMissesWhichBytesAloneHide) { + // A tier being asked for keys it does not have reads as zero load bytes, + // exactly like a tier nobody is asking. Different bugs, same byte count. + ActivitySummary summary; + summary.Render(WithDram(0, 0, 0, 0, 0, 0), 60.0); + ActivitySnapshot missing = WithDram(0, 0, 0, 0, 0, 0); + missing.tiers["DRAM"].load_misses = 77; + const std::string all = Joined(summary.Render(missing, 60.0)); + EXPECT_NE(all.find("miss=77"), std::string::npos) << all; +} + +TEST(ActivitySummary, ReCacheLineNamesWhyObjectsWereDropped) { + // This is the line that replaced six per-key DEBUG statements. Each drop + // reason has a different fix -- a bigger queue, a wider admission policy, a + // larger tier -- so the counts have to stay distinguishable. + ActivitySummary summary; + summary.Render(ActivitySnapshot{}, 60.0); + + ActivitySnapshot snap; + snap.recache.admitted = 100; + snap.recache.installed = 90; + snap.recache.queue_full = 7; + snap.recache.rejected = 3; + snap.recache.alloc_failed = 2; + snap.recache.install_failed = 1; + const std::string all = Joined(summary.Render(snap, 60.0)); + EXPECT_NE(all.find("90 installed / 100 admitted"), std::string::npos) << all; + EXPECT_NE(all.find("queue_full=7"), std::string::npos) << all; + EXPECT_NE(all.find("rejected=3"), std::string::npos) << all; + EXPECT_NE(all.find("no_memory=2"), std::string::npos) << all; + EXPECT_NE(all.find("install_failed=1"), std::string::npos) << all; + EXPECT_NE(all.find("dropped 13"), std::string::npos) << all; +} + +TEST(ActivitySummary, QuietReCachePathPrintsNoReCacheLine) { + // A deployment with no remote peers never re-caches. A line of zeroes every + // window would train the reader to skip the block that matters when it is + // finally non-zero. + ActivitySummary summary; + summary.Render(WithDram(0, 0, 0, 0, 0, 0), 60.0); + const std::string all = Joined(summary.Render(WithDram(1024, 1, 0, 0, 0, 0), 60.0)); + EXPECT_EQ(all.find("recache"), std::string::npos) << all; +} + +TEST(ActivitySummary, BreaksDownByTierOnlyWhenThereIsMoreThanOne) { + ActivitySummary summary; + + ActivitySnapshot two; + two.tiers["DRAM"].offload_bytes = 1024 * 1024; + two.tiers["DRAM"].capacity_bytes = 1024 * 1024 * 1024; + two.tiers["SSD"].offload_bytes = 4ULL * 1024 * 1024; + two.tiers["SSD"].capacity_bytes = 8ULL * 1024 * 1024 * 1024; + const std::string all = Joined(summary.Render(two, 60.0)); + EXPECT_NE(all.find("DRAM"), std::string::npos) << all; + EXPECT_NE(all.find("SSD"), std::string::npos) << all; + // The total line must still add them up, or a two-tier node needs mental + // arithmetic to answer "did anything move". + EXPECT_NE(all.find("offload 5.00 MiB"), std::string::npos) << all; +} + +TEST(ActivitySummary, ATierThatDisappearedDoesNotPrintAnEnormousDelta) { + // A backend can be unregistered and re-registered, restarting its counters. + // Without clamping, the next window subtracts a large baseline from a small + // reading and reports something that reads like data loss. + ActivitySummary summary; + summary.Render(WithDram(64ULL * 1024 * 1024, 64, 0, 0, 0, 0), 60.0); + const std::string all = Joined(summary.Render(WithDram(1024 * 1024, 1, 0, 0, 0, 0), 60.0)); + EXPECT_NE(all.find("offload 1.00 MiB/1 keys"), std::string::npos) << all; +} + +TEST(ActivitySummary, ReportsTheMeasuredWindowNotTheConfiguredOne) { + ActivitySummary summary; + const std::string all = Joined(summary.Render(WithDram(1024 * 1024, 1, 0, 0, 0, 0), 12.5)); + EXPECT_NE(all.find("12.5s"), std::string::npos) << all; +} + +} // namespace +} // namespace mori::umbp From 79293e41db08272c84307edfb493d4feae6dd508 Mon Sep 17 00:00:00 2001 From: TianDi101 Date: Fri, 9 Oct 2026 08:06:05 +0000 Subject: [PATCH 4/4] fix(umbp): create metric series at zero, and publish node state without a master Two gaps found verifying the masterless metrics endpoint against a live Prometheus + Grafana. Every counter series was born holding its first event. Sources skipped zero-valued counters and MetricPublisher only forwarded a strictly positive delta, so a series first appeared already carrying its first burst. Prometheus computes rate() and increase() between two samples of one series, so that burst was invisible: after 436 keys were locally evicted, increase(mori_umbp_backend_medium_events_total{event="local_evict"}) read 0 and the eviction panels were empty; increase() of committed keys read 1039 against 1280 actually committed. A steady counter loses one tick, but a rare one -- an eviction round, a no_space, a failed or rejected transfer -- loses the whole event, and those are the ones dashboards and alerts exist for. - MetricPublisher ships a counter the first time it sees it, even at zero. - InstrumentedBackend publishes the (op, status) pairs each op can produce, plus batches/seconds/bytes, from the first tick; impossible pairs stay out. - PageBackend publishes its local_evict counters from the first tick when local eviction is enabled. - CompositeTransferEngine publishes every direction of each registered engine and the rejected-items counter at zero. - PoolClient seeds the per-traffic byte counters and ranged install failures at zero, since those are written on the event itself. A masterless node exported no capacity, live-key or client-count series: the master derives them from heartbeats, so without one they did not exist and the capacity / utilization / live-keys panels were blank. PoolClient now publishes them itself when there is no master, under the master's names and labels, so the same panels read either deployment. Tests: a first-sight zero is shipped once and not re-shipped; a fresh backend exposes its produced pairs at zero but not impossible ones; an unused engine reports zero-valued series. Two tests that asserted the old "no series until first event" shape are updated to expect zero. Co-Authored-By: Claude Opus 5.5 --- .../distributed/metrics/component_metrics.cpp | 15 ++- .../peer/backend/instrumented_backend.cpp | 41 +++++--- .../distributed/peer/backend/page_backend.cpp | 18 ++-- src/umbp/distributed/pool_client.cpp | 97 +++++++++++++++++++ .../transfer/composite_transfer_engine.cpp | 58 +++++------ .../distributed/metrics/component_metrics.h | 4 +- .../include/umbp/distributed/pool_client.h | 13 +++ .../distributed/test_component_metrics.cpp | 59 ++++++++++- .../distributed/test_umbp_client_metrics.cpp | 12 ++- 9 files changed, 251 insertions(+), 66 deletions(-) diff --git a/src/umbp/distributed/metrics/component_metrics.cpp b/src/umbp/distributed/metrics/component_metrics.cpp index 6404d3e28..05db115a0 100644 --- a/src/umbp/distributed/metrics/component_metrics.cpp +++ b/src/umbp/distributed/metrics/component_metrics.cpp @@ -74,10 +74,19 @@ void MetricPublisher::Publish(const std::string& source_id, const MetricLabels& continue; } - uint64_t& last = last_[BaselineKey(source_id, s.name, source_labels, s.labels)]; - if (s.value > last && sink.counter) { + // A counter seen for the first time is shipped even at zero. Prometheus + // computes rate() and increase() from a series' own samples, so a series + // whose first sample already holds its first burst has no baseline to + // difference against and that burst never shows up. For a steady counter + // that loses one tick; for a rare one -- an eviction round, a no_space, a + // failed transfer -- it is the whole event. Creating the series at 0 + // before anything happens is what makes the first occurrence visible. + auto [it, first_sight] = + last_.try_emplace(BaselineKey(source_id, s.name, source_labels, s.labels), 0); + uint64_t& last = it->second; + if ((first_sight || s.value > last) && sink.counter) { sink.counter(s.name, s.help, MergeLabels(source_labels, s.labels), - static_cast(s.value - last) * s.scale); + s.value > last ? static_cast(s.value - last) * s.scale : 0.0); } // Updated even on a zero or negative delta so the next tick stays correct // against a counter that was rebuilt underneath us. diff --git a/src/umbp/distributed/peer/backend/instrumented_backend.cpp b/src/umbp/distributed/peer/backend/instrumented_backend.cpp index 8f0f6db47..0d5e35ae8 100644 --- a/src/umbp/distributed/peer/backend/instrumented_backend.cpp +++ b/src/umbp/distributed/peer/backend/instrumented_backend.cpp @@ -65,6 +65,24 @@ const char* StatusName(int status) { } } +// The (op, status) pairs each op can actually produce -- exactly the Add() +// calls below. They are published from the first tick, at zero, so every +// series exists before its first event: Prometheus can only see an increase +// between two samples of one series, so a series born holding its first +// count would hide that count from rate() and increase(). Pairs an op never +// produces (an abort has no "no_space") stay unpublished. +constexpr bool kProduces[5][5] = { + // ok exists no_space miss failed + {true, true, true, false, true}, // allocate + {true, false, false, false, true}, // commit + {true, false, false, false, true}, // abort + {true, false, false, true, false}, // resolve + {true, false, false, true, false}, // evict +}; + +// Ops that move bytes, so their bytes counter exists from the first tick. +constexpr bool kCarriesBytes[5] = {false, true, false, true, true}; + // Scoped timer that charges elapsed nanoseconds to an op on destruction, so an // early return inside a forwarded call still gets measured. class ScopedNanos { @@ -203,16 +221,16 @@ std::vector InstrumentedBackend::Evict(const std::vector InstrumentedBackend::SampleMetrics() const { std::vector out; + static_assert(kOpCount == 5 && kStatusCount == 5, + "kProduces / kCarriesBytes must list every op and status"); out.reserve(kOpCount * 4 + 8); for (int op = 0; op < kOpCount; ++op) { const char* op_name = OpName(op); for (int status = 0; status < kStatusCount; ++status) { + if (!kProduces[op][status]) continue; const uint64_t v = entries_[op][status].load(std::memory_order_relaxed); - // Skip combinations this op never produces (an abort has no "no_space") - // so the series set stays the shape of what actually happens. - if (v == 0) continue; out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_OPS_TOTAL, MORI_UMBP_METRIC_BACKEND_OPS_TOTAL_HELP, {{"op", op_name}, {"status", StatusName(status)}}, @@ -220,7 +238,6 @@ std::vector InstrumentedBackend::SampleMetrics() const { } const uint64_t batches = ops_[op].batches.load(std::memory_order_relaxed); - if (batches == 0) continue; out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_BATCHES_TOTAL, MORI_UMBP_METRIC_BACKEND_BATCHES_TOTAL_HELP, {{"op", op_name}}, @@ -230,17 +247,15 @@ std::vector InstrumentedBackend::SampleMetrics() const { // MetricSample::scale converts once, in the publisher, so the metric keeps // the seconds its name promises. const uint64_t nanos = ops_[op].nanos.load(std::memory_order_relaxed); - if (nanos > 0) { - out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_OP_SECONDS_TOTAL, - MORI_UMBP_METRIC_BACKEND_OP_SECONDS_TOTAL_HELP, - {{"op", op_name}}, - nanos, - MetricKind::kCounter, - 1.0 / kNanosPerSecond}); - } + out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_OP_SECONDS_TOTAL, + MORI_UMBP_METRIC_BACKEND_OP_SECONDS_TOTAL_HELP, + {{"op", op_name}}, + nanos, + MetricKind::kCounter, + 1.0 / kNanosPerSecond}); const uint64_t bytes = ops_[op].bytes.load(std::memory_order_relaxed); - if (bytes > 0) { + if (kCarriesBytes[op]) { out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_BYTES_TOTAL, MORI_UMBP_METRIC_BACKEND_BYTES_TOTAL_HELP, {{"op", op_name}}, diff --git a/src/umbp/distributed/peer/backend/page_backend.cpp b/src/umbp/distributed/peer/backend/page_backend.cpp index 56f86deaf..3a56b4d78 100644 --- a/src/umbp/distributed/peer/backend/page_backend.cpp +++ b/src/umbp/distributed/peer/backend/page_backend.cpp @@ -845,9 +845,14 @@ std::vector PageBackend::SampleMetrics() const { // outside this class it is indistinguishable from keys that were never // stored. On a masterless node that is ALL of the eviction there is. std::vector out; + // Only a node that evicts for itself has these counters. On one that does, + // they are published from the first tick, zero included: a series born + // already holding the first round's count hides that round from rate() and + // increase(), and eviction is rare enough that the first round is often the + // only one a dashboard window contains. + if (!local_evict_enabled_) return out; auto event = [&out](const char* name, uint64_t v) { - if (v == 0) return; out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_MEDIUM_EVENTS_TOTAL, MORI_UMBP_METRIC_BACKEND_MEDIUM_EVENTS_TOTAL_HELP, {{"event", name}}, @@ -861,13 +866,10 @@ std::vector PageBackend::SampleMetrics() const { // has nothing to evict, which look the same in the freed-bytes counter. event("local_evict_no_candidate", local_evict_no_candidate_.load(std::memory_order_relaxed)); - const uint64_t freed_bytes = local_evict_bytes_.load(std::memory_order_relaxed); - if (freed_bytes != 0) { - out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL, - MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL_HELP, - {{"event", kPageBackendLocalEvictEvent}}, - freed_bytes}); - } + out.push_back(MetricSample{MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL, + MORI_UMBP_METRIC_BACKEND_MEDIUM_BYTES_TOTAL_HELP, + {{"event", kPageBackendLocalEvictEvent}}, + local_evict_bytes_.load(std::memory_order_relaxed)}); return out; } diff --git a/src/umbp/distributed/pool_client.cpp b/src/umbp/distributed/pool_client.cpp index 6d8e7f221..88f7fb4b9 100644 --- a/src/umbp/distributed/pool_client.cpp +++ b/src/umbp/distributed/pool_client.cpp @@ -1169,6 +1169,7 @@ bool PoolClient::Init() { } else { metric_sink_ = config_.metric_sink; } + SeedEventCounters(); // Every instrumented component on this node rides a metrics tick: the storage // backend (generic slot-lifecycle series from the decorator, plus whatever the @@ -5112,6 +5113,102 @@ void PoolClient::PublishComponentMetrics() { } metric_publisher_.Publish("pool", {}, samples, sink); } + + if (!master_client_) PublishNodeStateGauges(); +} + +void PoolClient::PublishNodeStateGauges() { + // Same derivation MasterClient::SnapshotAndCacheTierCapacities feeds the + // master: every backend of a tier summed into one capacity. + std::map caps; + std::map live_keys; + for (MediumBackend* backend : registry_.All()) { + if (backend == nullptr) continue; + const TierCapacity c = backend->Capacity(); + TierCapacity& agg = caps[backend->Tier()]; + agg.total_bytes += c.total_bytes; + agg.available_bytes += c.available_bytes; + live_keys[backend->Tier()] += backend->OwnedKeyCount(); + } + + // A masterless node is the only client it can see. + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_COUNT, MORI_UMBP_METRIC_CLIENT_COUNT_HELP, {}, + 1.0); + + uint64_t total_keys = 0; + for (const auto& [tier, cap] : caps) { + const MetricSink::Labels labels = {{"tier", TierTypeName(tier)}}; + const uint64_t used = + cap.total_bytes >= cap.available_bytes ? cap.total_bytes - cap.available_bytes : 0; + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_CAPACITY_TOTAL, + MORI_UMBP_METRIC_CLIENT_CAPACITY_TOTAL_HELP, labels, + static_cast(cap.total_bytes)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_CAPACITY_AVAIL, + MORI_UMBP_METRIC_CLIENT_CAPACITY_AVAIL_HELP, labels, + static_cast(cap.available_bytes)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_CAPACITY_USED, + MORI_UMBP_METRIC_CLIENT_CAPACITY_USED_HELP, labels, + static_cast(used)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_CAPACITY_UTILIZATION, + MORI_UMBP_METRIC_CLIENT_CAPACITY_UTILIZATION_HELP, labels, + cap.total_bytes > 0 + ? static_cast(used) / static_cast(cap.total_bytes) + : 0.0); + const uint64_t keys = live_keys[tier]; + total_keys += keys; + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_KV_LIVE_COUNT, + MORI_UMBP_METRIC_CLIENT_KV_LIVE_COUNT_HELP, labels, + static_cast(keys)); + } + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_KV_LIVE_COUNT_TOTAL, + MORI_UMBP_METRIC_CLIENT_KV_LIVE_COUNT_TOTAL_HELP, {}, + static_cast(total_keys)); + + if (default_pool_ == nullptr) return; + for (const auto& [name, logical] : default_pool_->LogicalTierCapacities()) { + const auto& cap = logical.capacity; + const MetricSink::Labels labels = {{"logical_tier", name}, + {"tier", TierTypeName(logical.representative_tier)}}; + const uint64_t used = + cap.total_bytes >= cap.available_bytes ? cap.total_bytes - cap.available_bytes : 0; + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_TOTAL, + MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_TOTAL_HELP, labels, + static_cast(cap.total_bytes)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_AVAIL, + MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_AVAIL_HELP, labels, + static_cast(cap.available_bytes)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_USED, + MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_USED_HELP, labels, + static_cast(used)); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_UTILIZATION, + MORI_UMBP_METRIC_CLIENT_LOGICAL_CAPACITY_UTILIZATION_HELP, labels, + cap.total_bytes > 0 + ? static_cast(used) / static_cast(cap.total_bytes) + : 0.0); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_PUT_ELIGIBLE, + MORI_UMBP_METRIC_CLIENT_LOGICAL_PUT_ELIGIBLE_HELP, labels, + logical.put_eligible ? 1.0 : 0.0); + metric_sink_->SetGauge(MORI_UMBP_METRIC_CLIENT_LOGICAL_PEAK_UTILIZATION, + MORI_UMBP_METRIC_CLIENT_LOGICAL_PEAK_UTILIZATION_HELP, labels, + logical.peak_member_utilization); + } +} + +void PoolClient::SeedEventCounters() { + if (metric_sink_ == nullptr) return; + for (const char* traffic : {"local", "remote"}) { + const MetricSink::Labels labels = {{"traffic", traffic}}; + CountMetric(MORI_UMBP_METRIC_CLIENT_OUTBOUND_PUT_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_OUTBOUND_PUT_BYTES_TOTAL_HELP, labels, 0.0); + CountMetric(MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_INBOUND_PUT_BYTES_TOTAL_HELP, labels, 0.0); + CountMetric(MORI_UMBP_METRIC_CLIENT_OUTBOUND_GET_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_OUTBOUND_GET_BYTES_TOTAL_HELP, labels, 0.0); + CountMetric(MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL, + MORI_UMBP_METRIC_CLIENT_INBOUND_GET_BYTES_TOTAL_HELP, labels, 0.0); + } + CountMetric(MORI_UMBP_METRIC_RANGED_REMOTE_INSTALL_FAILURES_TOTAL, + MORI_UMBP_METRIC_RANGED_REMOTE_INSTALL_FAILURES_TOTAL_HELP, {}, 0.0); } // --------------------------------------------------------------------------- diff --git a/src/umbp/distributed/transfer/composite_transfer_engine.cpp b/src/umbp/distributed/transfer/composite_transfer_engine.cpp index c6434687e..15d111695 100644 --- a/src/umbp/distributed/transfer/composite_transfer_engine.cpp +++ b/src/umbp/distributed/transfer/composite_transfer_engine.cpp @@ -343,10 +343,14 @@ std::vector CompositeTransferEngine::SampleMetrics() const { for (size_t i = 0; i < engines_.size() && i < counters_.size(); ++i) { const char* engine_name = engines_[i]->Name(); + // Every direction of every registered engine is published from the first + // tick, at zero: Prometheus sees an increase only between two samples of + // one series, so a series born holding its first transfer -- or its first + // failure -- hides it from rate() and increase(). The cost is a few + // flat-zero series for directions an engine never carries. for (size_t d = 0; d < kDirectionCount; ++d) { const DirectionCounters& c = counters_[i]->by_direction[d]; const uint64_t plans = c.plans.load(std::memory_order_relaxed); - if (plans == 0) continue; const char* dir = DirectionName(static_cast(d)); const uint64_t failed = c.failed_plans.load(std::memory_order_relaxed); @@ -356,30 +360,21 @@ std::vector CompositeTransferEngine::SampleMetrics() const { // Plans posted but not yet settled count as ok until Wait says // otherwise; the correction lands on the next tick. plans >= failed ? plans - failed : 0}); - if (failed > 0) { - out.push_back( - MetricSample{MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, - MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL_HELP, - {{"engine", engine_name}, {"direction", dir}, {"status", "failed"}}, - failed}); - } - - const uint64_t bytes = c.bytes.load(std::memory_order_relaxed); - if (bytes > 0) { - out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_BYTES_TOTAL, - MORI_UMBP_METRIC_TRANSFER_BYTES_TOTAL_HELP, - {{"engine", engine_name}, {"direction", dir}}, - bytes}); - } - const uint64_t nanos = c.nanos.load(std::memory_order_relaxed); - if (nanos > 0) { - out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_SECONDS_TOTAL, - MORI_UMBP_METRIC_TRANSFER_SECONDS_TOTAL_HELP, - {{"engine", engine_name}, {"direction", dir}}, - nanos, - MetricKind::kCounter, - 1.0 / kNanosPerSecond}); - } + out.push_back( + MetricSample{MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, + MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL_HELP, + {{"engine", engine_name}, {"direction", dir}, {"status", "failed"}}, + failed}); + out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_BYTES_TOTAL, + MORI_UMBP_METRIC_TRANSFER_BYTES_TOTAL_HELP, + {{"engine", engine_name}, {"direction", dir}}, + c.bytes.load(std::memory_order_relaxed)}); + out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_SECONDS_TOTAL, + MORI_UMBP_METRIC_TRANSFER_SECONDS_TOTAL_HELP, + {{"engine", engine_name}, {"direction", dir}}, + c.nanos.load(std::memory_order_relaxed), + MetricKind::kCounter, + 1.0 / kNanosPerSecond}); } // Whatever the engine publishes about its own internals, stamped with the @@ -392,13 +387,12 @@ std::vector CompositeTransferEngine::SampleMetrics() const { // Items no engine would take. Not an engine failure — a routing one — so it // gets its own engine value rather than being blamed on whoever ran last. - const uint64_t rejected = rejected_items_.load(std::memory_order_relaxed); - if (rejected > 0) { - out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, - MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL_HELP, - {{"engine", "none"}, {"direction", "none"}, {"status", "rejected"}}, - rejected}); - } + // Published at zero too: it is a routing bug, so its first occurrence is + // the one that matters. + out.push_back(MetricSample{MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, + MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL_HELP, + {{"engine", "none"}, {"direction", "none"}, {"status", "rejected"}}, + rejected_items_.load(std::memory_order_relaxed)}); return out; } diff --git a/src/umbp/include/umbp/distributed/metrics/component_metrics.h b/src/umbp/include/umbp/distributed/metrics/component_metrics.h index 8cc175d67..526ecbb04 100644 --- a/src/umbp/include/umbp/distributed/metrics/component_metrics.h +++ b/src/umbp/include/umbp/distributed/metrics/component_metrics.h @@ -199,7 +199,9 @@ class MetricSource { class MetricPublisher { public: struct Sink { - // delta >= 0, already differenced. Not called for a zero delta. + // delta >= 0, already differenced. Not called for a zero delta, except + // once for a series seen for the first time, so the series exists at 0 + // before its first event (see Publish). std::function counter; // Current reading, shipped every tick. diff --git a/src/umbp/include/umbp/distributed/pool_client.h b/src/umbp/include/umbp/distributed/pool_client.h index 8865d8436..749aaa184 100644 --- a/src/umbp/include/umbp/distributed/pool_client.h +++ b/src/umbp/include/umbp/distributed/pool_client.h @@ -265,6 +265,19 @@ class PoolClient { // or a new transfer engine visible in Grafana without touching this file. void PublishComponentMetrics(); + // What a master derives from this node's heartbeats -- per-tier capacity, + // live key counts, logical-tier eligibility, the client count -- published + // directly when there is no master to derive it. Same metric names and + // labels as the master emits, so the same panels read either deployment. + void PublishNodeStateGauges(); + + // Creates at zero every counter the data path writes directly (the + // per-traffic byte counters, ranged install failures), once, before any + // traffic. Those are written on the event itself, so without this a series + // would be born holding its first event and rate()/increase() would never + // see it. + void SeedEventCounters(); + // Holds the delta baselines for every component published above, keyed by // (component, metric, labels). Touched once per tick; the counters // themselves live in the components. diff --git a/tests/cpp/umbp/distributed/test_component_metrics.cpp b/tests/cpp/umbp/distributed/test_component_metrics.cpp index b45f93d2b..1cecea4a0 100644 --- a/tests/cpp/umbp/distributed/test_component_metrics.cpp +++ b/tests/cpp/umbp/distributed/test_component_metrics.cpp @@ -119,6 +119,26 @@ TEST(MetricPublisher, ShipsDeltasNotAbsoluteValues) { EXPECT_DOUBLE_EQ(c.Counter(kName), 15.0); } +TEST(MetricPublisher, ShipsAFirstSightZeroSoTheSeriesExistsBeforeItsFirstEvent) { + FakeSource src; + src.samples = {MetricSample{kName, kHelp, {}, 0}}; + + MetricPublisher pub; + Collector c; + pub.Publish("src", {}, src, c.Sink()); + ASSERT_TRUE(c.Has(kName)) << "a counter seen for the first time must create its series"; + EXPECT_DOUBLE_EQ(c.Counter(kName), 0.0); + + // Only the first sight: an unchanged zero is not re-shipped every tick. + c.counters.clear(); + pub.Publish("src", {}, src, c.Sink()); + EXPECT_FALSE(c.Has(kName)); + + src.samples[0].value = 4; + pub.Publish("src", {}, src, c.Sink()); + EXPECT_DOUBLE_EQ(c.Counter(kName), 4.0); +} + TEST(MetricPublisher, RebasesInsteadOfShippingNegativeDeltas) { FakeSource src; src.samples = {MetricSample{kName, kHelp, {}, 100}}; @@ -341,7 +361,28 @@ TEST_F(InstrumentedBackendMetrics, EvictingAnAbsentKeyIsAMissNotAFailure) { backend_->Evict({"never-existed"}); Publish(); EXPECT_DOUBLE_EQ(Ops("evict", "miss"), 1.0); - EXPECT_DOUBLE_EQ(Ops("evict", "ok"), -1.0) << "no successful eviction should have been reported"; + EXPECT_DOUBLE_EQ(Ops("evict", "ok"), 0.0) << "no successful eviction should have been reported"; +} + +TEST_F(InstrumentedBackendMetrics, EverySeriesExistsAtZeroBeforeItsFirstEvent) { + // Prometheus sees an increase only between two samples of one series. A + // series born holding its first count hides that count from rate() and + // increase() -- for a rare event such as a no_space, the whole event. + Publish(); + EXPECT_DOUBLE_EQ(Ops("allocate", "no_space"), 0.0); + EXPECT_DOUBLE_EQ(Ops("commit", "failed"), 0.0); + EXPECT_DOUBLE_EQ(Ops("resolve", "miss"), 0.0); + EXPECT_DOUBLE_EQ(Ops("evict", "ok"), 0.0); + EXPECT_DOUBLE_EQ(Bytes("evict"), 0.0); + EXPECT_DOUBLE_EQ(Batches("abort"), 0.0); + // ...but only the pairs an op can produce: an abort never runs out of space. + EXPECT_DOUBLE_EQ(Ops("abort", "no_space"), -1.0); + EXPECT_DOUBLE_EQ(Bytes("allocate"), -1.0); + + // The first event then arrives as an increment on an existing series. + PutKey("k", 4096); + Publish(); + EXPECT_DOUBLE_EQ(Ops("commit", "ok"), 1.0); } TEST_F(InstrumentedBackendMetrics, RecordsTimeSpentInsideTheMedium) { @@ -458,15 +499,25 @@ TEST(TransferEngineMetrics, CountsItemsNoEngineWouldTake) { TEST(TransferEngineMetrics, AnEngineNeedsNoMetricsCodeToBeMeasured) { // The composite is the measurement point precisely so an engine can be added - // with AddEngine() alone. An engine registered but never used reports - // nothing rather than a row of zeros, keeping the series set honest. + // with AddEngine() alone. An engine registered but never used reports its + // series at zero, so its first transfer -- or first failure -- is an + // increment Prometheus can see rather than the value a series is born with. CompositeTransferEngine composite; composite.AddEngine(std::make_unique()); MetricPublisher pub; Collector c; pub.Publish("transfer", {}, composite, c.Sink()); - EXPECT_TRUE(c.counters.empty()); + ASSERT_FALSE(c.counters.empty()); + for (const auto& [identity, value] : c.counters) EXPECT_DOUBLE_EQ(value, 0.0) << identity; + + const MetricLabels failed = { + {"engine", "LocalCopyEngine"}, {"direction", "local"}, {"status", "failed"}}; + EXPECT_DOUBLE_EQ(c.Counter(Identity(MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, failed)), 0.0); + EXPECT_DOUBLE_EQ( + c.Counter(Identity(MORI_UMBP_METRIC_TRANSFER_OPS_TOTAL, + {{"engine", "none"}, {"direction", "none"}, {"status", "rejected"}})), + 0.0); } } // namespace diff --git a/tests/cpp/umbp/distributed/test_umbp_client_metrics.cpp b/tests/cpp/umbp/distributed/test_umbp_client_metrics.cpp index e01231f5a..bbbd3a988 100644 --- a/tests/cpp/umbp/distributed/test_umbp_client_metrics.cpp +++ b/tests/cpp/umbp/distributed/test_umbp_client_metrics.cpp @@ -521,14 +521,16 @@ TEST_F(PoolClientLocalByteTrackingTest, LocalPutGetBytesCounted) { EXPECT_GT(batch_get_local_count, 0.0) << "Expected local batch-get bandwidth count > 0; Prometheus shows " << batch_get_local_count; - // Single-node setup must not produce any remote traffic counters. - EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_outbound_put_bytes_total", "remote"), -1.0) + // Single-node setup must not move any remote bytes. The remote series + // exist from startup at zero (so a first remote transfer is an increment + // Prometheus can see), and must still read zero here. + EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_outbound_put_bytes_total", "remote"), 0.0) << "Unexpected remote outbound put bytes in single-node setup"; - EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_outbound_get_bytes_total", "remote"), -1.0) + EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_outbound_get_bytes_total", "remote"), 0.0) << "Unexpected remote outbound get bytes in single-node setup"; - EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_inbound_put_bytes_total", "remote"), -1.0) + EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_inbound_put_bytes_total", "remote"), 0.0) << "Unexpected remote inbound put bytes in single-node setup"; - EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_inbound_get_bytes_total", "remote"), -1.0) + EXPECT_EQ(ParseMetricValue(body, "mori_umbp_client_inbound_get_bytes_total", "remote"), 0.0) << "Unexpected remote inbound get bytes in single-node setup"; EXPECT_EQ(parse_hist_sum("mori_umbp_client_batch_put_bandwidth_gibps", "remote"), -1.0)