Merge branch 'pratik/otel-phase3-tx-tracing' into pratik/otel-phase4-consensus-tracing

OpenTelemetryPlan/05-configuration-reference.md

@@ -53,7 +53,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).
 
 ---
 
@@ -67,7 +67,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
 
@@ -133,7 +133,7 @@ flowchart TB
     end
 
     subgraph init["Initialization"]
-        parse["setupTelemetry()"]
+        parse["makeTelemetrySetup()"]
         factory["makeTelemetry()"]
     end
 
@@ -167,7 +167,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

@@ -1141,6 +1141,22 @@ private:
     void
     startGenesisLedger();
 
+    /**
+     * Start the tracing pipeline.
+     *
+     * Called once from setup(), as soon as the node identity is known.
+     * Starting here rather than in start() means spans emitted during the
+     * rest of setup() are recorded: SpanGuard drops a span whenever the
+     * global Telemetry instance is not yet live, and the first consensus
+     * round runs inside setup().
+     *
+     * @pre nodeIdentity_ is populated, so setServiceInstanceId() has
+     * already supplied the service.instance.id resource attribute
+     * (the Telemetry resource is fixed once start() builds it).
+     */
+    void
+    startTelemetry();
+
     std::shared_ptr<Ledger>
     getLastFullLedger();
 
@@ -1227,6 +1243,27 @@ 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));
+
+    // Start telemetry here, not in start(). Spans are emitted during the rest
+    // of setup() — the first consensus round in beginConsensus() below — and
+    // are dropped unless the global Telemetry instance 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 trace stream for a run that never comes up.
+    //  - Before beginConsensus(): that call emits the first consensus spans.
+    startTelemetry();
+
     if (validatorKeys_.keys)
         setMaxDisallowedLedger();
 
@@ -1314,16 +1351,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));
-
     if (!cluster_->load(config().section(Sections::kClusterNodes)))
     {
         JLOG(journal_.fatal()) << "Invalid entry in cluster configuration.";
@@ -1536,6 +1563,11 @@ ApplicationImp::start(bool withTimers)
 
     ledgerCleaner_->start();
     perfLog_->start();
+}
+
+void
+ApplicationImp::startTelemetry()
+{
     telemetry_->start();
 }
 

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)