Merge branch 'pratik/otel-phase8-log-correlation' into pratik/otel-phase9-metric-gap-fill

Resolves the telemetry-startup conflict between the two branches. Both
sides move the telemetry start earlier in setup(); they disagree only on
how far the pipeline had been split at that point.

phase-1b (arriving) moved nodeIdentity_, setServiceInstanceId() and the
telemetry start up to just after the wallet DB is proven usable. phase-9
had split the metrics pipeline in two and left its copy of that block at
the old, later position.

Kept both intentions: the block stays at phase-1b's early position, and
metricsRegistry_ construction moves up with it so it precedes
startTelemetry() -- the metrics half is guarded on the registry existing,
so leaving the construction behind would have started tracing while
silently skipping metrics. phase-9's later copy is dropped as the stale
duplicate. The two-phase split is preserved: startTelemetryGauges() still
runs after overlay_ is constructed, because the observable callbacks read
it and getOverlay() asserts.

Net effect is that the metrics provider now starts earlier than on either
branch, and still before beginConsensus() emits the first spans and the
only operating-mode transition.

OpenTelemetryPlan/05-configuration-reference.md

@@ -55,7 +55,7 @@ phases. They will be added as the corresponding subsystems are instrumented:
 
 > **TxQ** = Transaction Queue
 
-The parser `setupTelemetry()` in `src/libxrpl/telemetry/TelemetryConfig.cpp` reads the `[telemetry]` `Section` and populates a `Telemetry::Setup` struct, applying the defaults listed in Section 5.1.2 via `section.value_or(...)`. It derives `serviceInstanceId` from the node public key when not overridden, selects the exporter endpoint default by exporter type, and leaves the sampling ratio at its fixed 1.0 default (not read from config — see Section 7.4.2).
+The parser `makeTelemetrySetup()` in `src/libxrpl/telemetry/TelemetryConfig.cpp` reads the `[telemetry]` `Section` and populates a `Telemetry::Setup` struct, applying the defaults listed in Section 5.1.2 via `section.value_or(...)`. It derives `serviceInstanceId` from the node public key when not overridden, selects the exporter endpoint default by exporter type, and leaves the sampling ratio at its fixed 1.0 default (not read from config — see Section 7.4.2).
 
 ---
 
@@ -69,7 +69,7 @@ The parser `setupTelemetry()` in `src/libxrpl/telemetry/TelemetryConfig.cpp` rea
 > constructed with an empty `serviceInstanceId` and patched via
 > `setServiceInstanceId()` once `setup()` has called `getNodeIdentity()`.
 
-`ApplicationImp` (in `src/xrpld/app/main/Application.cpp`) owns a `std::unique_ptr<telemetry::Telemetry> telemetry_`. It is built in the member initializer list via `makeTelemetry(setupTelemetry(...))` with an empty `serviceInstanceId`, then patched in `setup()` by calling `setServiceInstanceId()` with the Base58 node public key (unless the user supplied a custom `service_instance_id`). `start()` and `run()` forward to `telemetry_->start()` / `telemetry_->stop()`, and `getTelemetry()` returns the owned instance.
+`ApplicationImp` (in `src/xrpld/app/main/Application.cpp`) owns a `std::unique_ptr<telemetry::Telemetry> telemetry_`. It is built in the member initializer list via `makeTelemetry(makeTelemetrySetup(...))` with an empty `serviceInstanceId`, then patched in `setup()` by calling `setServiceInstanceId()` with the Base58 node public key (unless the user supplied a custom `service_instance_id`). `start()` and `run()` forward to `telemetry_->start()` / `telemetry_->stop()`, and `getTelemetry()` returns the owned instance.
 
 ### 5.3.2 ServiceRegistry Interface Addition
 
@@ -135,7 +135,7 @@ flowchart TB
     end
 
     subgraph init["Initialization"]
-        parse["setupTelemetry()"]
+        parse["makeTelemetrySetup()"]
         factory["makeTelemetry()"]
     end
 
@@ -169,7 +169,7 @@ flowchart TB
 **Reading the diagram:**
 
 - **Configuration Sources**: `xrpld.cfg` provides runtime settings (endpoint, per-component trace toggles) while the CMake flag controls whether telemetry is compiled in at all. Head sampling is fixed at 1.0 and is not a config option; volume reduction happens via tail sampling in the collector.
-- **Initialization**: `setupTelemetry()` parses config values, then `makeTelemetry()` constructs the provider, processor, and exporter objects.
+- **Initialization**: `makeTelemetrySetup()` parses config values, then `makeTelemetry()` constructs the provider, processor, and exporter objects.
 - **Runtime Components**: The `TracerProvider` creates spans, the `BatchProcessor` buffers them, and the `OTLP Exporter` serializes and sends them over the wire.
 - **OTLP arrow to Collector**: Trace data leaves the xrpld process via OTLP/HTTP and enters the external Collector pipeline. (OTLP/gRPC is future work — see design decisions §2.2.2.)
 - **Collector Pipeline**: `Receivers` ingest OTLP data, `Processors` apply sampling/filtering/enrichment, and `Exporters` forward traces to storage backends (Tempo, etc.).

OpenTelemetryPlan/Phase2_taskList.md

@@ -67,7 +67,7 @@
 
 - `src/tests/libxrpl/telemetry/TelemetryConfig.cpp`:
   - Test Setup defaults (all fields have correct initial values)
-  - Test `setupTelemetry` config parser (empty section, full section, edge cases)
+  - Test `makeTelemetrySetup` config parser (empty section, full section, edge cases)
   - Test `samplingRatio` clamping (values outside 0.0-1.0)
 
 - `src/tests/libxrpl/telemetry/SpanGuardFactory.cpp`:

src/xrpld/app/main/Application.cpp

@@ -1312,6 +1312,39 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
         return false;
     }
 
+    nodeIdentity_ = getNodeIdentity(*this, cmdline);
+
+    // Now that the node identity is known, inject it into the telemetry
+    // resource attributes — but only if the user didn't already set a
+    // custom service_instance_id in [telemetry].  The Telemetry object
+    // was constructed with an empty serviceInstanceId because
+    // nodeIdentity_ is not available in the member initializer list.
+    if (!config_->section("telemetry").exists("service_instance_id"))
+        telemetry_->setServiceInstanceId(toBase58(TokenType::NodePublic, nodeIdentity_->first));
+
+    // Create the OTel MetricsRegistry for gap-fill metrics (counters,
+    // histograms, observable gauges). It must exist before startTelemetry(),
+    // which starts the metrics half of the pipeline.
+    metricsRegistry_ = std::make_unique<telemetry::MetricsRegistry>(
+        telemetry_->isEnabled(), *this, logs_->journal("MetricsRegistry"));
+
+    // Start telemetry here, not in start(). Spans and metrics are both emitted
+    // during the rest of setup() — the first consensus round in
+    // beginConsensus() below emits spans and records the process's only
+    // operating-mode transition — and both are dropped unless the pipeline is
+    // already live.
+    //
+    // The position is bounded on both sides:
+    //  - After initRelationalDatabase(): the wallet DB must exist for the node
+    //    identity above, and a DB failure aborts setup(), so starting earlier
+    //    would export a partial stream for a run that never comes up.
+    //  - Before beginConsensus(): that call emits the first consensus spans
+    //    and the only mode-transition counter increment.
+    //
+    // Only the observable instruments have to wait for their subsystems; they
+    // are registered separately by startTelemetryGauges() once overlay_ exists.
+    startTelemetry();
+
     if (validatorKeys_.keys)
         setMaxDisallowedLedger();
 
@@ -1399,29 +1432,6 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
 
     orderBookDB_->setup(getLedgerMaster().getCurrentLedger());
 
-    nodeIdentity_ = getNodeIdentity(*this, cmdline);
-
-    // Now that the node identity is known, inject it into the telemetry
-    // resource attributes — but only if the user didn't already set a
-    // custom service_instance_id in [telemetry].  The Telemetry object
-    // was constructed with an empty serviceInstanceId because
-    // nodeIdentity_ is not available in the member initializer list.
-    if (!config_->section("telemetry").exists("service_instance_id"))
-        telemetry_->setServiceInstanceId(toBase58(TokenType::NodePublic, nodeIdentity_->first));
-
-    // Create the OTel MetricsRegistry for gap-fill metrics (counters,
-    // histograms, observable gauges).
-    metricsRegistry_ = std::make_unique<telemetry::MetricsRegistry>(
-        telemetry_->isEnabled(), *this, logs_->journal("MetricsRegistry"));
-
-    // Start tracing and the metrics provider right away, so the meter exists
-    // before anything records a metric. beginConsensus() below emits a
-    // mode-transition counter, and it is the only mode transition the process
-    // ever makes — a meter created after it would lose that series entirely.
-    // Only the observable gauges have to wait; they are registered by
-    // startTelemetryGauges() once overlay_ exists.
-    startTelemetry();
-
     if (!cluster_->load(config().section(Sections::kClusterNodes)))
     {
         JLOG(journal_.fatal()) << "Invalid entry in cluster configuration.";

src/xrpld/rpc/detail/PathFindSpanNames.h

@@ -65,10 +65,13 @@ inline constexpr auto discover = makeStr("discover");
 }  // namespace op
 
 // ===== Attribute keys ======================================================
-//
-// All pathfind attributes are namespaced under `pathfind_*` (underscore form,
-// per Phase 1c naming spec rule 5). Avoids collisions with bare keys like
-// `fast` or `num_paths` that other subsystems may introduce.
+
+/**
+ * All pathfind attributes are namespaced under `pathfind_*`, in underscore
+ * form, per the span attribute naming convention in CONTRIBUTING.md. Avoids
+ * collisions with bare keys like `fast` or `num_paths` that other subsystems
+ * may introduce.
+ */
 
 namespace attr {
 /**

src/xrpld/rpc/detail/PathRequest.cpp

@@ -35,6 +35,7 @@
 #include <xrpl/resource/Consumer.h>
 #include <xrpl/server/InfoSub.h>
 #include <xrpl/server/LoadFeeTrack.h>
+#include <xrpl/telemetry/Redaction.h>
 #include <xrpl/telemetry/SpanGuard.h>
 #include <xrpl/tx/paths/RippleCalc.h>
 
@@ -587,9 +588,13 @@ PathRequest::findPaths(
     // that a single RPC call produces one discover span instead of N (one per
     // candidate source asset). Trade-off: per-asset discovery/ranking timing
     // is no longer split into individual spans — span count and Tempo storage
-    // are bounded per RPC at the cost of per-asset visibility. If per-asset
-    // breakdown is needed in the future, add child spans inside the loop body
-    // (`Pathfinder::findPaths`/`computePathRanks`) parented off this span.
+    // are bounded per RPC at the cost of per-asset visibility.
+    //
+    // This is an unscoped guard: it takes the ambient span as its own parent,
+    // but does not itself become the ambient parent. Adding per-asset child
+    // spans inside the loop body therefore requires either making this a
+    // ScopedSpanGuard or passing its context explicitly via childSpan(); a
+    // child created here today would parent to this span's parent, not to it.
     using namespace telemetry;
     auto span = SpanGuard::span(
         TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::discover);
@@ -744,7 +749,20 @@ PathRequest::doUpdate(
     auto span = ScopedSpanGuard(
         TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::compute);
     span.setAttribute(pathfind_span::attr::fast, fast);
-    span.setAttribute(pathfind_span::attr::destCurrency, to_string(saDstAmount_.asset()).c_str());
+    // to_string(Issue) renders a non-XRP asset as "<issuer>/<currency>" with the
+    // issuer as a plaintext Base58 address, so it cannot be emitted as-is: every
+    // account reaching a span is hashed first. Redact just the issuer and keep
+    // the currency, which is what this attribute is for. An MPT asset renders as
+    // its issuance ID and carries no address, so it needs no redaction.
+    span.setAttribute(
+        pathfind_span::attr::destCurrency,
+        saDstAmount_.asset().visit(
+            [](Issue const& issue) {
+                return isXRP(issue.account)
+                    ? to_string(issue.currency)
+                    : redactAccount(toBase58(issue.account)) + "/" + to_string(issue.currency);
+            },
+            [](MPTIssue const& mpt) { return to_string(mpt.getMptID()); }));
 
     JLOG(journal_.debug()) << iIdentifier_ << " update " << (fast ? "fast" : "normal");
 

src/xrpld/rpc/detail/PathRequestManager.cpp

@@ -23,6 +23,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <utility>
 #include <vector>
 
@@ -77,15 +78,21 @@ PathRequestManager::updateAll(std::shared_ptr<ReadView const> const& inLedger)
     using namespace telemetry;
     // updateAll runs on every ledger close. Skip span emission when there are
     // no active path subscriptions, to avoid a steady stream of empty spans at
-    // mainnet close cadence. A null guard is used in that case; all other work
-    // still runs unchanged (notably the isNewPathRequest() flag reset below),
-    // so behaviour matches the pre-span code path.
-    auto span = requests.empty()
-        ? SpanGuard{}
-        : SpanGuard::span(
-              TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::updateAll);
-    span.setAttribute(pathfind_span::attr::ledgerIndex, static_cast<int64_t>(inLedger->seq()));
-    span.setAttribute(pathfind_span::attr::numRequests, static_cast<int64_t>(requests.size()));
+    // mainnet close cadence. All other work still runs unchanged (notably the
+    // isNewPathRequest() flag reset below), so behaviour matches the pre-span
+    // code path.
+    //
+    // Scoped, so the pathfind.compute spans that doUpdate() creates below on
+    // this thread nest under it. std::optional because ScopedSpanGuard is
+    // deliberately non-movable, so it cannot be produced by a ternary.
+    std::optional<ScopedSpanGuard> span;
+    if (!requests.empty())
+    {
+        span.emplace(
+            TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::updateAll);
+        span->setAttribute(pathfind_span::attr::ledgerIndex, static_cast<int64_t>(inLedger->seq()));
+        span->setAttribute(pathfind_span::attr::numRequests, static_cast<int64_t>(requests.size()));
+    }
 
     bool newRequests = app_.getLedgerMaster().isNewPathRequest();
     bool mustBreak = false;

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -4,6 +4,7 @@
 #include <xrpld/overlay/Overlay.h>
 #include <xrpld/rpc/RPCHandler.h>
 #include <xrpld/rpc/Role.h>
+#include <xrpld/rpc/detail/Handler.h>
 #include <xrpld/rpc/detail/RpcSpanNames.h>
 #include <xrpld/rpc/detail/Tuning.h>
 #include <xrpld/rpc/detail/WSInfoSub.h>
@@ -107,6 +108,42 @@ statusRequestResponse(http_request_type const& request, boost::beast::http::stat
     return handoff;
 }
 
+/**
+ * Resolve the command attribute for a WebSocket message to a bounded value.
+ *
+ * The command/method field is client-supplied and is promoted to a Prometheus
+ * label by the spanmetrics connector, so the raw string must never be emitted:
+ * arbitrary request input would drive unbounded label cardinality. Resolving
+ * against the handler registry keeps per-command attribution for real commands
+ * and collapses everything else to a single "unknown" series.
+ *
+ * A request naming both fields with different values is not a real command
+ * (processSession rejects it below), so it also resolves to "unknown".
+ *
+ * @param jv      The parsed WebSocket request object.
+ * @param config  Node config, for the beta-RPC-API flag used to look up the
+ *                handler for the request's API version.
+ * @return The canonical handler name, or "unknown" for an unrecognized,
+ *         missing, non-string, or self-contradictory command.
+ */
+static std::string_view
+resolveWsCommandSpanName(json::Value const& jv, Config const& config)
+{
+    bool const hasCommand = jv.isMember(jss::command) && jv[jss::command].isString();
+    bool const hasMethod = jv.isMember(jss::method) && jv[jss::method].isString();
+    if (!hasCommand && !hasMethod)
+        return rpc_span::val::unknownCommand;
+
+    std::string const cmd = hasCommand ? jv[jss::command].asString() : jv[jss::method].asString();
+    if (hasCommand && hasMethod && cmd != jv[jss::method].asString())
+        return rpc_span::val::unknownCommand;
+
+    auto const* handler =
+        RPC::getHandler(RPC::getAPIVersionNumber(jv, config.betaRpcApi), config.betaRpcApi, cmd);
+    return (handler != nullptr) ? std::string_view{handler->name}
+                                : std::string_view{rpc_span::val::unknownCommand};
+}
+
 // VFALCO TODO Rewrite to use boost::beast::http::fields
 static bool
 authorized(Port const& port, std::map<std::string, std::string> const& h)
@@ -434,14 +471,13 @@ ServerHandler::processSession(
     // Fresh root so each WS message is its own trace.
     auto span = ScopedSpanGuard::freshRoot(
         TraceCategory::Rpc, rpc_span::prefix::rpc, rpc_span::op::wsMessage);
-    if (jv.isMember(jss::command) && jv[jss::command].isString())
-    {
-        span.setAttribute(rpc_span::attr::command, jv[jss::command].asString().c_str());
-    }
-    else if (jv.isMember(jss::method) && jv[jss::method].isString())
-    {
-        span.setAttribute(rpc_span::attr::command, jv[jss::method].asString().c_str());
-    }
+    // The command is client-supplied and becomes a Prometheus label via the
+    // spanmetrics connector, so it is resolved against the handler registry
+    // before emission: a recognized command keeps its canonical name, anything
+    // else collapses to "unknown". Emitting the raw string would let request
+    // input drive unbounded label cardinality. Mirrors the HTTP path's
+    // resolveCommandSpanName().
+    span.setAttribute(rpc_span::attr::command, resolveWsCommandSpanName(jv, app_.config()));
 
     auto is = std::static_pointer_cast<WSInfoSub>(session->appDefined);
     if (is->getConsumer().disconnect(journal_))

src/xrpld/rpc/handlers/orderbook/PathFind.cpp

@@ -13,6 +13,8 @@
 #include <xrpl/telemetry/Redaction.h>
 #include <xrpl/telemetry/SpanGuard.h>
 
+#include <utility>
+
 namespace xrpl {
 
 json::Value
@@ -23,10 +25,15 @@ doPathFind(RPC::JsonContext& context)
     // thread) nest under it. doPathFind does not yield, so scoping is safe.
     auto span = ScopedSpanGuard(
         TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::request);
-    // Addresses are hashed before emission for privacy.
-    if (auto const& src = context.params[jss::source_account]; src.isString())
+    // Addresses are hashed before emission for privacy. Read through a const
+    // reference: the non-const json::Value::operator[] inserts a null for a
+    // missing key, which would make PathRequest::parseJson's isMember() checks
+    // see an absent field as present and return Malformed instead of Missing.
+    // Reading for telemetry must not alter what the request looks like.
+    auto const& params = std::as_const(context.params);
+    if (auto const& src = params[jss::source_account]; src.isString())
         span.setAttribute(pathfind_span::attr::sourceAccount, redactAccount(src.asString()));
-    if (auto const& dst = context.params[jss::destination_account]; dst.isString())
+    if (auto const& dst = params[jss::destination_account]; dst.isString())
         span.setAttribute(pathfind_span::attr::destAccount, redactAccount(dst.asString()));
 
     if (context.app.config().pathSearchMax == 0)

src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp

@@ -34,10 +34,15 @@ doRipplePathFind(RPC::JsonContext& context)
     // span's log lines stay trace-correlated.
     auto span = ScopedSpanGuard(
         TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::request);
-    // Addresses are hashed before emission for privacy.
-    if (auto const& src = context.params[jss::source_account]; src.isString())
+    // Addresses are hashed before emission for privacy. Read through a const
+    // reference: the non-const json::Value::operator[] inserts a null for a
+    // missing key, which would make PathRequest::parseJson's isMember() checks
+    // see an absent field as present and return Malformed instead of Missing.
+    // Reading for telemetry must not alter what the request looks like.
+    auto const& params = std::as_const(context.params);
+    if (auto const& src = params[jss::source_account]; src.isString())
         span.setAttribute(pathfind_span::attr::sourceAccount, redactAccount(src.asString()));
-    if (auto const& dst = context.params[jss::destination_account]; dst.isString())
+    if (auto const& dst = params[jss::destination_account]; dst.isString())
         span.setAttribute(pathfind_span::attr::destAccount, redactAccount(dst.asString()));
 
     if (context.app.config().pathSearchMax == 0)