ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-08-21 14:20:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Pratik Mankawde
2026-07-29 16:05:43 +01:00
parent 9a3f68355a e404d7412f
commit 2edec71e18
6 changed files with 104 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
OpenTelemetryPlan/03-implementation-strategy.md
View File

@@ -335,78 +335,6 @@ Instrumentation is gated on two levels. A compile-time feature flag (`XRPL_ENABL
This section provides a detailed assessment of how intrusive the OpenTelemetry integration is to the existing xrpld codebase.
### 3.9.1 Files Modified Summary
| Component | Files Modified | Architectural Impact |
| --------------------- | -------------- | -------------------- |
| **Core Telemetry** | 10 new files | None (new module) |
| **Application Init** | 2 files | Minimal |
| **RPC Layer** | 3 files | Minimal |
| **Transaction Relay** | 4 files | Low |
| **Consensus** | 3 files | Low-Medium |
| **Protocol Buffers** | 1 file | Low |
| **CMake/Build** | 3 files | Minimal |
| **PathFinding** | 2 | Minimal |
| **TxQ/Fee** | 2 | Minimal |
| **Validator/Amend** | 3 | Minimal |
| **Total** | **~33 files** | **Low** |
### 3.9.2 Detailed File Impact
```mermaid
pie title Code Changes by Component
"New Telemetry Module" : 800
"Transaction Relay" : 160
"Consensus" : 130
"RPC Layer" : 100
"PathFinding" : 80
"TxQ/Fee" : 60
"Validator/Amendment" : 40
"Application Init" : 35
"Protocol Buffers" : 25
"Build System" : 60
```
#### New Files (No Impact on Existing Code)
| File | Purpose |
| ------------------------------------------- | ------------------------- |
| `include/xrpl/telemetry/Telemetry.h` | Main interface |
| `include/xrpl/telemetry/TelemetryConfig.h` | Configuration structures |
| `include/xrpl/telemetry/TraceContext.h` | Context propagation |
| `include/xrpl/telemetry/SpanGuard.h` | RAII wrapper |
| `include/xrpl/telemetry/DiscardFlag.h` | Thread-local discard flag |
| `include/xrpl/telemetry/SpanAttributes.h` | Attribute helpers |
| `src/libxrpl/telemetry/Telemetry.cpp` | Implementation |
| `src/libxrpl/telemetry/TelemetryConfig.cpp` | Config parsing |
| `src/libxrpl/telemetry/TraceContext.cpp` | Context serialization |
| `src/libxrpl/telemetry/NullTelemetry.cpp` | No-op implementation |
#### Modified Files (Existing Xrpld Code)
| File | Risk Level |
| ------------------------------------------------- | ---------- |
| `src/xrpld/app/main/Application.cpp` | Low |
| `include/xrpl/core/ServiceRegistry.h` | Low |
| `src/xrpld/rpc/detail/ServerHandler.cpp` | Low |
| `src/xrpld/rpc/handlers/*.cpp` | Low |
| `src/xrpld/overlay/detail/PeerImp.cpp` | Medium |
| `src/xrpld/overlay/detail/OverlayImpl.cpp` | Medium |
| `src/xrpld/app/consensus/RCLConsensus.cpp` | Medium |
| `src/xrpld/app/consensus/RCLConsensusAdaptor.cpp` | Medium |
| `src/xrpld/core/JobQueue.cpp` | Low |
| `src/xrpld/app/paths/PathRequest.cpp` | Low |
| `src/xrpld/app/paths/Pathfinder.cpp` | Low |
| `src/xrpld/app/misc/TxQ.cpp` | Low |
| `src/xrpld/app/main/LoadManager.cpp` | Low |
| `src/xrpld/app/misc/ValidatorList.cpp` | Low |
| `src/xrpld/app/misc/AmendmentTable.cpp` | Low |
| `src/xrpld/app/misc/Manifest.cpp` | Low |
| `src/xrpld/shamap/SHAMap.cpp` | Low |
| `src/xrpld/overlay/detail/ripple.proto` | Low |
| `CMakeLists.txt` | Low |
| `cmake/FindOpenTelemetry.cmake` | None (new) |
### 3.9.3 Risk Assessment by Component
<div align="center">

45
src/libxrpl/telemetry/SpanGuard.cpp
View File

@@ -57,6 +57,7 @@
#include <cstdint>
#include <cstring>
#include <exception>
#include <format>
#include <initializer_list>
#include <memory>
#include <optional>
@@ -198,6 +199,34 @@ categoryToSpanKind(TraceCategory cat)
return otel_trace::SpanKind::kInternal; // unreachable
}
/**
* Join a span-name prefix and suffix into the dotted full name.
*
* Wraps std::format because the callers are noexcept: std::format can throw
* (std::bad_alloc, or std::format_error on a malformed spec) and an escaping
* exception would terminate the process. Telemetry must never take the node
* down, so a failure yields std::nullopt and the caller returns a null guard —
* the same degrade-to-no-op path already used when telemetry is disabled.
*
* @param prefix Segment before the dot (e.g. "consensus").
* @param name Segment after the dot (e.g. "round").
* @return The joined name, or std::nullopt if formatting failed.
*/
[[nodiscard]] std::optional<std::string>
joinSpanName(std::string_view prefix, std::string_view name) noexcept
{
try
{
return std::format("{}.{}", prefix, name);
}
catch (std::exception const&)
{
// Out of memory or a bad format spec. Drop the span rather than
// propagate out of a noexcept factory.
return std::nullopt;
}
}
} // namespace
SpanGuard
@@ -206,10 +235,10 @@ SpanGuard::span(TraceCategory cat, std::string_view prefix, std::string_view nam
auto* tel = Telemetry::getInstance();
if ((tel == nullptr) || !tel->isEnabled() || !isCategoryEnabled(*tel, cat))
return {};
std::string fullName;
fullName.reserve(prefix.size() + 1 + name.size());
fullName.append(prefix).append(1, '.').append(name);
return SpanGuard(std::make_unique<Impl>(tel->startSpan(fullName, categoryToSpanKind(cat))));
auto const fullName = joinSpanName(prefix, name);
if (!fullName)
return {};
return SpanGuard(std::make_unique<Impl>(tel->startSpan(*fullName, categoryToSpanKind(cat))));
}
SpanGuard
@@ -218,13 +247,13 @@ SpanGuard::freshRoot(TraceCategory cat, std::string_view prefix, std::string_vie
auto* tel = Telemetry::getInstance();
if ((tel == nullptr) || !tel->isEnabled() || !isCategoryEnabled(*tel, cat))
return {};
std::string fullName;
fullName.reserve(prefix.size() + 1 + name.size());
fullName.append(prefix).append(1, '.').append(name);
auto const fullName = joinSpanName(prefix, name);
if (!fullName)
return {};
// Force a fresh trace root: do NOT inherit this thread's active span.
auto rootCtx = opentelemetry::context::Context{otel_trace::kIsRootSpanKey, true};
return SpanGuard(
std::make_unique<Impl>(tel->startSpan(fullName, rootCtx, categoryToSpanKind(cat))));
std::make_unique<Impl>(tel->startSpan(*fullName, rootCtx, categoryToSpanKind(cat))));
}
// ===== Child / linked span creation ========================================

15
src/xrpld/app/main/GRPCServer.cpp
View File

@@ -243,8 +243,19 @@ GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::C
{
std::pair<Response, grpc::Status> result = handler_(context);
setIsUnlimited(result.first, isUnlimited);
span.setAttribute(grpc_span::attr::grpcStatus, grpc_span::val::success);
span.setOk();
// The handler can return a non-OK status without throwing, so
// the span status must follow result.second rather than assume
// success — otherwise every failed call traces as OK.
if (result.second.ok())
{
span.setAttribute(grpc_span::attr::grpcStatus, grpc_span::val::success);
span.setOk();
}
else
{
span.setAttribute(grpc_span::attr::grpcStatus, grpc_span::val::error);
span.setError(result.second.error_message());
}
responder_.Finish(result.first, result.second, this);
}
}

4
src/xrpld/rpc/ServerHandler.h
View File

@@ -194,10 +194,8 @@ private:
/**
* Process an RPC request and write the reply to `output`.
* @return false if the request resulted in an error response, true
* otherwise. Lets the caller's enclosing span reflect the outcome.
*/
bool
void
processRequest(
Port const& port,
std::string const& request,

15
src/xrpld/rpc/detail/RPCHandler.cpp
View File

@@ -239,6 +239,14 @@ resolveCommandSpanName(JsonContext const& context)
if (!context.params.isMember(jss::command) && !context.params.isMember(jss::method))
return rpc_span::val::unknownCommand;
// fillHandler() rejects a request that supplies both fields with differing
// values as rpcUNKNOWN_COMMAND. Mirror that here, or the span would be
// labelled with one of the two names and misattribute the error to a
// command that was never dispatched.
if (context.params.isMember(jss::command) && context.params.isMember(jss::method) &&
context.params[jss::command].asString() != context.params[jss::method].asString())
return rpc_span::val::unknownCommand;
std::string const cmd = context.params.isMember(jss::command)
? context.params[jss::command].asString()
: context.params[jss::method].asString();
@@ -262,6 +270,13 @@ doCommand(RPC::JsonContext& context, json::Value& result)
auto const cmdName = resolveCommandSpanName(context);
auto span = ScopedSpanGuard(TraceCategory::Rpc, rpc_span::prefix::command, cmdName);
span.setAttribute(rpc_span::attr::command, cmdName);
// Mirror the attribute set callMethod() puts on a successful command
// span, so error spans stay filterable by API version and role.
span.setAttribute(rpc_span::attr::version, static_cast<int64_t>(context.apiVersion));
span.setAttribute(
rpc_span::attr::rpcRole,
context.role == Role::ADMIN ? std::string_view(rpc_span::val::admin)
: std::string_view(rpc_span::val::user));
span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
span.setError(getErrorInfo(error).token.cStr());

69
src/xrpld/rpc/detail/ServerHandler.cpp
View File

@@ -606,7 +606,7 @@ ServerHandler::processSession(
auto const requestBody = ::xrpl::buffersToString(session->request().body().data());
span.setAttribute(rpc_span::attr::requestPayloadSize, static_cast<int64_t>(requestBody.size()));
bool const ok = processRequest(
processRequest(
session->port(),
requestBody,
session->remoteAddress().atPort(0),
@@ -628,15 +628,11 @@ ServerHandler::processSession(
{
session->close(true);
}
// Reflect the request outcome on the wrapper span instead of always OK.
if (ok)
{
span.setOk();
}
else
{
span.setError(rpc_span::val::error);
}
// Status is left unset: the OTel spec says instrumentation should leave it
// unset unless the operation itself errored, and reserves Ok for an
// operator asserting success. This span only delimits the HTTP request and
// has no error of its own to report — the outcome is determined inside
// rpc.process, which sets its own status.
}
static json::Value
@@ -655,7 +651,7 @@ constexpr json::Int kServerOverloaded = -32604;
constexpr json::Int kForbidden = -32605;
constexpr json::Int kWrongVersion = -32606;
bool
void
ServerHandler::processRequest(
Port const& port,
std::string const& request,
@@ -684,6 +680,7 @@ ServerHandler::processRequest(
// (the span would otherwise end UNSET, invisible to {status.code=error}).
auto httpReplyError = [&](int status, std::string const& message) {
spanHadError = true;
span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
span.setError(message);
httpReply(status, message, output, rpcJ);
};
@@ -695,7 +692,7 @@ ServerHandler::processRequest(
!jsonOrig || !jsonOrig.isObject())
{
httpReplyError(400, "Unable to parse request: " + reader.getFormattedErrorMessages());
return false;
return;
}
}
@@ -707,7 +704,7 @@ ServerHandler::processRequest(
if (!jsonOrig.isMember(jss::params) || !jsonOrig[jss::params].isArray())
{
httpReplyError(400, "Malformed batch request");
return false;
return;
}
size = jsonOrig[jss::params].size();
}
@@ -716,6 +713,17 @@ ServerHandler::processRequest(
span.setAttribute(rpc_span::attr::batchSize, static_cast<int64_t>(size));
json::Value reply(batch ? json::ValueType::Array : json::ValueType::Object);
// Append a per-request error item and record that the request failed.
// Batch responses (and ripplerpc < 3.0) always carry HTTP 200, so the
// span status can only learn about these failures through spanHadError.
// Every per-item error path must go through here, or an entirely failed
// batch would end its span as successful.
auto appendItemError = [&](json::Value&& item) {
spanHadError = true;
reply.append(std::move(item));
};
auto const start(std::chrono::high_resolution_clock::now());
for (unsigned i = 0; i < size; ++i)
{
@@ -726,7 +734,7 @@ ServerHandler::processRequest(
json::Value r(json::ValueType::Object);
r[jss::request] = jsonRPC;
r[jss::error] = makeJsonError(kMethodNotFound, "Method not found");
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -749,12 +757,12 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(400, jss::invalid_API_version.cStr());
return false;
return;
}
json::Value r(json::ValueType::Object);
r[jss::request] = jsonRPC;
r[jss::error] = makeJsonError(kWrongVersion, jss::invalid_API_version.cStr());
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -792,11 +800,11 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(503, "Server is overloaded");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kServerOverloaded, "Server is overloaded");
reply.append(r);
appendItemError(std::move(r));
continue;
}
}
@@ -807,11 +815,11 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(403, "Forbidden");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kForbidden, "Forbidden");
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -821,11 +829,11 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(400, "Null method");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kMethodNotFound, "Null method");
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -836,11 +844,11 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(400, "method is not string");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kMethodNotFound, "method is not string");
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -851,11 +859,11 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(400, "method is empty");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kMethodNotFound, "method is empty");
reply.append(r);
appendItemError(std::move(r));
continue;
}
@@ -877,7 +885,7 @@ ServerHandler::processRequest(
{
usage.charge(Resource::kFeeMalformedRpc);
httpReplyError(400, "params unparsable");
return false;
return;
}
else
{
@@ -886,7 +894,7 @@ ServerHandler::processRequest(
{
usage.charge(Resource::kFeeMalformedRpc);
httpReplyError(400, "params unparsable");
return false;
return;
}
}
}
@@ -904,12 +912,12 @@ ServerHandler::processRequest(
if (!batch)
{
httpReplyError(400, "ripplerpc is not a string");
return false;
return;
}
json::Value r = jsonRPC;
r[jss::error] = makeJsonError(kMethodNotFound, "ripplerpc is not a string");
reply.append(r);
appendItemError(std::move(r));
continue;
}
ripplerpc = params[jss::ripplerpc].asString();
@@ -1110,7 +1118,6 @@ ServerHandler::processRequest(
span.setOk();
}
httpReply(httpStatus, response, output, rpcJ);
return !spanHadError;
}
//------------------------------------------------------------------------------