Merge branch 'pratik/otel-phase5-docs-deployment' into pratik/otel-phase6-statsd

Conflict resolved in PeerImp.cpp, proposal receive handler: keep this
branch's proposalSpan name, and take the arriving change that
allocates the holder only for a live span.
This commit is contained in:
Pratik Mankawde committed 2026-09-30 23:05:38 +01:00
commit 2feb874df2
35 files changed
+1991 -326

No files matched your search

+6 -4
View File
@@ -59,10 +59,12 @@ ignore:
- "include/xrpl/beast/test/"
- "include/xrpl/beast/unit_test/"
# Telemetry modules. Telemetry compiles in by default (conanfile.py,
# CMakeLists.txt), but the unit-test suite never starts an exporter, so these
# files record no coverage. This block belongs on the earliest branch that
# adds telemetry code — codecov config only flows child-ward, so an ignore
# added on a later branch can never cover the branches before it.
# CMakeLists.txt). SpanGuardScope.cpp sends spans to an in-memory exporter,
# but no unit test sends spans through the real OTLP export path, so these
# files are left out of the coverage report. This block belongs on the
# earliest branch that adds telemetry code — codecov config only flows
# child-ward, so an ignore added on a later branch can never cover the
# branches before it.
- "src/xrpld/telemetry/"
- "src/libxrpl/telemetry/"
- "include/xrpl/telemetry/"
+4
View File
@@ -148,6 +148,10 @@ endif()
# There is no CMake option. The one switch is `conan install -o telemetry=`,
# which decides whether opentelemetry-cpp is fetched and sets the variable read
# below through the generated toolchain.
#
# The define is set for every target in this project, and also PUBLIC on
# xrpl.libxrpl.telemetry (cmake/XrplCore.cmake), so a parent project that
# adds this one with add_subdirectory() and links that module gets it too.
if(telemetry)
find_package(opentelemetry-cpp CONFIG REQUIRED)
add_compile_definitions(XRPL_ENABLE_TELEMETRY)
+2
View File
@@ -1698,6 +1698,8 @@ validators.txt
# enabled=0
#
# Enable or disable telemetry at runtime. Default: 0 (disabled).
# This key, use_tls and the trace_* keys take 0, 1, true or false, in
# any case. Any other value is a startup error that names the key.
#
# service_name=xrpld
#
+8
View File
@@ -229,6 +229,14 @@ if(telemetry)
xrpl.libxrpl.telemetry
PUBLIC opentelemetry-cpp::opentelemetry-cpp
)
# PUBLIC, so a parent project that adds this one with add_subdirectory()
# and links this module sees the same class layouts it was built with.
# CMakeLists.txt also sets the define for every target in this project.
# Conan consumers get it from conanfile.py instead.
target_compile_definitions(
xrpl.libxrpl.telemetry
PUBLIC XRPL_ENABLE_TELEMETRY
)
endif()
# Links xrpl.libxrpl.telemetry for the consensus tracing spans declared in
-3
View File
@@ -71,9 +71,6 @@ opentelemetry-cpp/*:tools.build:cxxflags+={{ otel_sanitizer_flags }}
{# Scoped to boost/* since it is the only gap. #}
boost/*:MACOSX_DEPLOYMENT_TARGET={{ min_macos_version }}
{% endif %}
{% if compiler == "gcc" and compiler_version < 13 %}
tools.build:cxxflags+=['-Wno-restrict']
{% endif %}
{% if os == "Windows" %}
# opentelemetry-cpp's recipe removes the `shared` option on Windows and never
# sets BUILD_SHARED_LIBS, so its upstream CMake defaults the protobuf-generated
+1 -1
View File
@@ -151,7 +151,7 @@ class Xrpl(ConanFile):
# OpenTelemetry C++ SDK for distributed tracing (optional).
# Provides OTLP/HTTP exporter, batch span processor, and trace API.
if self.options.telemetry:
self.requires("opentelemetry-cpp/1.28.0")
self.requires("opentelemetry-cpp/1.28.0", transitive_headers=True)
self.requires("xxhash/0.8.3", transitive_headers=True)
exports_sources = (
+8 -8
View File
@@ -113,14 +113,14 @@ hash); `tx.preflight` is stateless and omits both.
### Transaction Queue Spans
| Span Name | Source File | Attributes | Description |
| ------------------ | ----------- | ----------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `txq.enqueue` | TxQ.cpp | `tx_hash`, `tx_type`, `current_ledger_seq`, `current_ledger_hash` | Enqueue decision; parents to `tx.process` on the submission path (explicit context), a root on the open-ledger rebuild path — `current_ledger_seq` correlates it to the ledger in both cases |
| `txq.apply_direct` | TxQ.cpp | -- | Direct apply attempt (bypassing queue) |
| `txq.batch_clear` | TxQ.cpp | -- | Batch clear of queued transactions for an account |
| `txq.accept` | TxQ.cpp | `queue_size`, `ledger_changed` | Ledger-close accept loop over queued transactions |
| `txq.accept_tx` | TxQ.cpp | `tx_hash`, `retries_remaining`, `ter_code`, `txq_status` | Per-transaction apply during accept |
| `txq.cleanup` | TxQ.cpp | `ledger_seq` | Post-close cleanup of expired queue entries |
| Span Name | Source File | Attributes | Description |
| ------------------ | ----------- | ----------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `txq.enqueue` | TxQ.cpp | `tx_hash`, `tx_type`, `current_ledger_seq`, `current_ledger_hash` | Admission decision on every `TxQ::apply` (`txq_status` = outcome); child of `tx.process` on submit, of `consensus.accept.apply` on a rebuild in `doAccept`, else a root |
| `txq.apply_direct` | TxQ.cpp | -- | Direct apply attempt (bypassing queue) |
| `txq.batch_clear` | TxQ.cpp | -- | Batch clear of queued transactions for an account |
| `txq.accept` | TxQ.cpp | `queue_size`, `ledger_changed` | Ledger-close accept loop over queued transactions |
| `txq.accept_tx` | TxQ.cpp | `tx_hash`, `retries_remaining`, `ter_code`, `txq_status` | Per-transaction apply during accept |
| `txq.cleanup` | TxQ.cpp | `ledger_seq`, `expired_count` | Once per closed ledger, even when nothing expired: fee-metric update, queue resize, and expiry of entries past `LastLedgerSequence` (`expired_count`) |
### PathFinding Spans
-6
View File
@@ -24,9 +24,6 @@ class Manager;
namespace perf {
class PerfLog;
} // namespace perf
namespace telemetry {
class Telemetry;
} // namespace telemetry
// This is temporary until we migrate all code to use ServiceRegistry.
class Application;
@@ -227,9 +224,6 @@ public:
virtual perf::PerfLog&
getPerfLog() = 0;
virtual telemetry::Telemetry&
getTelemetry() = 0;
// Configuration and state
[[nodiscard]] virtual bool
isStopping() const = 0;
+1 -1
View File
@@ -39,7 +39,7 @@
* @note Thread safety: the flag is thread-local, so each thread observes only
* its own discard signal — no synchronization is required.
*
* @see SpanGuard::discard(), FilteringSpanProcessor (Telemetry.cpp)
* @see SpanGuard::discard(), FilteringSpanProcessor (FilteringSpanProcessor.h)
*/
namespace xrpl::telemetry {
@@ -0,0 +1,138 @@
#pragma once
#ifdef XRPL_ENABLE_TELEMETRY
#include <opentelemetry/sdk/trace/processor.h>
#include <opentelemetry/sdk/trace/recordable.h>
#include <opentelemetry/trace/span_context.h>
#include <chrono>
#include <memory>
namespace xrpl::telemetry {
/**
* Span processor that sits in front of the batch processor. It drops
* discarded spans, and it makes every span export each attribute key once.
*
* The OTLP recordable appends every SetAttribute() call without looking up
* the key. Without this processor a key set twice is exported twice, and a
* reader that takes the first value misses the update. So the processor
* wraps each recordable it hands to the SDK. The wrapper keeps the last value
* for each key and writes each key once, in first-set order, when the span
* ends. Every other call goes straight to the delegate's recordable.
*
* Dependency diagram:
*
* +------------------------+ delegate_ +--------------------+
* | FilteringSpanProcessor |------------->| BatchSpanProcessor |
* +------------------------+ +--------------------+
* | reads | makes
* v v
* DiscardScope (DiscardFlag.h) OtlpRecordable
*
* Data flow for one span:
*
* MakeRecordable() --> wrapper around the delegate's recordable
* SetAttribute() --> wrapper keeps an owned copy, last value wins
* other calls --> forwarded to the delegate's recordable at once
* OnEnd() --> discarded? drop the span
* --> else write each kept key once, then pass the
* delegate's recordable to the delegate
*
* The discard check reads a thread-local flag, not the span's attributes,
* because the recordable type depends on the exporter and has no common
* getter. The flag works because Span::End() calls OnEnd() on its own thread.
*
* @code
* // Production: wrap the batch processor before building the provider.
* auto processor = std::make_unique<FilteringSpanProcessor>(std::move(batchProcessor));
*
* // A key set twice is exported once, as "accepted".
* auto span = processor->MakeRecordable();
* span->SetAttribute("consensus_phase", "open");
* span->SetAttribute("consensus_phase", "accepted");
* processor->OnEnd(std::move(span));
*
* // Edge case: a span ended inside a DiscardScope never reaches the delegate.
* {
* DiscardScope discardScope;
* processor->OnEnd(processor->MakeRecordable());
* }
* @endcode
*
* @note Thread safety: OnStart() and OnEnd() may run on many threads at once.
* Each wrapper belongs to one span, and the SDK span serializes calls to it.
* The discard flag is thread-local.
* @note Limitations: every distinct key is kept until the span ends, so memory
* grows with the number of distinct keys, and the delegate's attribute count
* limit applies when the keys are written. Attributes inside one event or link
* are forwarded as given. An attribute a delegate sets on the recordable in its
* own OnStart() skips the merge.
*/
class FilteringSpanProcessor final : public opentelemetry::sdk::trace::SpanProcessor
{
/**
* Receives every span that is not discarded. The batch processor in
* production.
*/
std::unique_ptr<opentelemetry::sdk::trace::SpanProcessor> delegate_;
public:
/**
* @param delegate Processor that receives every span that is not discarded.
*/
explicit FilteringSpanProcessor(
std::unique_ptr<opentelemetry::sdk::trace::SpanProcessor> delegate);
/**
* Asks the delegate for a recordable and wraps it.
*
* @return the wrapper, or nullptr when the delegate made none. The SDK then
* records nothing for the span.
*/
std::unique_ptr<opentelemetry::sdk::trace::Recordable>
MakeRecordable() noexcept override;
/**
* Tells the delegate a span started, passing the delegate's own recordable.
*
* @param span Recordable from MakeRecordable(). Any other recordable is
* passed on as it is.
* @param parentContext Context of the span's parent.
*/
void
OnStart(
opentelemetry::sdk::trace::Recordable& span,
opentelemetry::trace::SpanContext const& parentContext) noexcept override;
/**
* Drops the span if it was discarded. Otherwise writes each kept attribute
* once and passes the delegate's own recordable to the delegate.
*
* @param span Recordable from MakeRecordable(). Any other recordable is
* passed on as it is.
*/
void
OnEnd(std::unique_ptr<opentelemetry::sdk::trace::Recordable>&& span) noexcept override;
/**
* @param timeout Longest time to wait for the delegate.
* @return the delegate's result.
*/
bool
ForceFlush(
std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override;
/**
* @param timeout Longest time to wait for the delegate.
* @return the delegate's result.
*/
bool
Shutdown(
std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override;
};
} // namespace xrpl::telemetry
#endif // XRPL_ENABLE_TELEMETRY
+24 -7
View File
@@ -673,14 +673,18 @@ public:
/**
* RAII guard that activates a span on the current context store (scoped).
*
* Wraps a SpanGuard (which owns the span) plus an OTel Scope that
* Holds a SpanGuard (which owns the span) plus an OTel Scope that
* pushes the span onto the active context store's stack for the
* guard's lifetime, so child spans created under that store inherit
* it as parent. On destruction the Scope pops BEFORE the span ends
* (member order: guard first, scope second). Non-copyable and
* (member order: guard_ first, impl_ second). Non-copyable and
* non-movable — factories return unnamed temporaries, so guaranteed
* copy elision (C++17) covers `auto s = ScopedSpanGuard::freshRoot(...)`.
*
* The Scope and the store it was pushed onto sit behind the pimpl, which
* is allocated only for a live span. A null guard, such as one made while
* telemetry is off, allocates nothing here.
*
* When the span must outlive the current store (e.g. handed to a job
* queue), convert to a plain SpanGuard with `operator SpanGuard() &&`:
* that pops the Scope eagerly under the constructing store and yields a
@@ -688,10 +692,12 @@ public:
*
* ScopedSpanGuard dependency diagram:
*
* @code
* +--------------------------------------------+
* | ScopedSpanGuard |
* | (scoped, store-bound) |
* +--------------------------------------------+
* | - guard_ : SpanGuard (inline) |
* | - impl_ : unique_ptr<ScopedImpl> (pimpl) |
* +--------------------------------------------+
* | + (cat, prefix, name) [ctor] |
@@ -703,14 +709,14 @@ public:
* | + spanContext() / discard() |
* | + operator bool() |
* +--------------------------------------------+
* | hides (pimpl)
* +------+--------------------+
* | |
* | | hides (pimpl)
* | owns |
* +----------+ +-----------------------------+
* | SpanGuard| | optional<Scope> |
* | SpanGuard| | ScopedImpl: Scope + owner |
* | (span) | | (OTel, store-bound; |
* | | | present : span active) |
* | | | null : guard_ is null) |
* +----------+ +-----------------------------+
* @endcode
*
* Usage examples:
*
@@ -758,7 +764,18 @@ public:
*/
class ScopedSpanGuard
{
/**
* Owns the span. Declared before impl_ so it is destroyed after it: the
* Scope pops before the span ends.
*/
SpanGuard guard_;
struct ScopedImpl;
/**
* The active OTel Scope and the store it was pushed onto. Null when
* guard_ is null, and after the handoff or discard() pops the Scope.
*/
std::unique_ptr<ScopedImpl> impl_;
explicit ScopedSpanGuard(SpanGuard&& guard) noexcept;
@@ -0,0 +1,360 @@
/**
* Implementation of FilteringSpanProcessor.
*
* The processor gives the SDK a UniqueAttributeRecordable in place of the
* delegate's recordable. The wrapper keeps one owned value per attribute key
* and writes the keys once, in first-set order, when the span ends.
*
* @see FilteringSpanProcessor (FilteringSpanProcessor.h)
*/
#ifdef XRPL_ENABLE_TELEMETRY
#include <xrpl/telemetry/FilteringSpanProcessor.h>
#include <xrpl/telemetry/DiscardFlag.h>
#include <opentelemetry/common/attribute_value.h>
#include <opentelemetry/common/key_value_iterable.h>
#include <opentelemetry/common/timestamp.h>
#include <opentelemetry/nostd/span.h>
#include <opentelemetry/nostd/string_view.h>
#include <opentelemetry/nostd/variant.h>
#include <opentelemetry/sdk/common/attribute_utils.h>
#include <opentelemetry/sdk/instrumentationscope/instrumentation_scope.h>
#include <opentelemetry/sdk/resource/resource.h>
#include <opentelemetry/sdk/trace/processor.h>
#include <opentelemetry/sdk/trace/recordable.h>
#include <opentelemetry/sdk/trace/span_limits.h>
#include <opentelemetry/trace/span_context.h>
#include <opentelemetry/trace/span_id.h>
#include <opentelemetry/trace/span_metadata.h>
#include <opentelemetry/trace/trace_flags.h>
#include <algorithm>
#include <chrono>
#include <memory>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
#include <vector>
namespace xrpl::telemetry {
namespace {
namespace trace_api = opentelemetry::trace;
namespace trace_sdk = opentelemetry::sdk::trace;
/**
* Writes one owned attribute value to a recordable, as the view type the SDK
* takes. The view lasts only for the call, and the recordable copies it.
*
* @code
* opentelemetry::nostd::visit(ForwardOwnedValue{recordable, key}, ownedValue);
* @endcode
*
* @note The key and the owned value must outlive the call.
*/
struct ForwardOwnedValue
{
/**
* Recordable that receives the attribute.
*/
trace_sdk::Recordable& target;
/**
* Attribute key, pointing into the caller's owned copy.
*/
opentelemetry::nostd::string_view key;
template <typename T>
requires std::is_arithmetic_v<T>
void
operator()(T value) const noexcept
{
target.SetAttribute(key, value);
}
void
operator()(std::string const& value) const noexcept
{
target.SetAttribute(key, opentelemetry::nostd::string_view{value.data(), value.size()});
}
template <typename T>
void
operator()(std::vector<T> const& values) const noexcept
{
target.SetAttribute(key, opentelemetry::nostd::span<T const>{values.data(), values.size()});
}
// std::vector<bool> packs its bits and has no data(), so copy to plain bools.
void
operator()(std::vector<bool> const& values) const noexcept
{
auto const flags = std::make_unique<bool[]>(values.size());
std::ranges::copy(values, flags.get());
target.SetAttribute(
key, opentelemetry::nostd::span<bool const>{flags.get(), values.size()});
}
// A string array is sent as views into the owned strings.
void
operator()(std::vector<std::string> const& values) const noexcept
{
std::vector<opentelemetry::nostd::string_view> views;
views.reserve(values.size());
for (auto const& value : values)
{
views.emplace_back(value.data(), value.size());
}
target.SetAttribute(
key,
opentelemetry::nostd::span<opentelemetry::nostd::string_view const>{
views.data(), views.size()});
}
};
/**
* Recordable the processor gives the SDK in place of the delegate's.
*
* SetAttribute() keeps an owned copy of each value and replaces any earlier
* value for the same key. The first set fixes the key's position. Every other
* call goes to the delegate's recordable at once.
*
* SDK span --SetAttribute()--> attributes_ (one entry per key)
* --other calls-----> inner_ (the delegate's recordable)
* flushAndRelease(): attributes_ --> inner_, then inner_ is handed back
*
* @code
* UniqueAttributeRecordable wrapper{delegate.MakeRecordable()};
* wrapper.SetAttribute("txq_status", "rejected");
* wrapper.SetAttribute("txq_status", "queued");
* delegate.OnEnd(wrapper.flushAndRelease()); // one txq_status, "queued"
* @endcode
*
* @note Not thread-safe. The SDK span serializes calls to its recordable.
* @note operator SpanData*() keeps its default of nullptr, because the
* delegate's recordable does not hold the kept attributes until the span ends.
*/
class UniqueAttributeRecordable final : public trace_sdk::Recordable
{
/**
* One kept attribute: an owned key and an owned copy of its last value.
*/
using Attribute = std::pair<std::string, opentelemetry::sdk::common::OwnedAttributeValue>;
/**
* The delegate's recordable. Null only after flushAndRelease().
*/
std::unique_ptr<trace_sdk::Recordable> inner_;
/**
* Kept attributes, in the order each key was first set.
*/
std::vector<Attribute> attributes_;
public:
/**
* @param inner The delegate's recordable. Must not be null.
*/
explicit UniqueAttributeRecordable(std::unique_ptr<trace_sdk::Recordable> inner) noexcept
: inner_(std::move(inner))
{
}
/**
* @return the delegate's recordable.
*/
[[nodiscard]] trace_sdk::Recordable&
inner() const noexcept
{
return *inner_;
}
/**
* Writes each kept attribute to the delegate's recordable once, in
* first-set order. The delegate's recordable applies its own limits here.
*
* @return the delegate's recordable. Do not use this wrapper afterwards.
*/
[[nodiscard]] std::unique_ptr<trace_sdk::Recordable>
flushAndRelease() noexcept
{
for (auto const& [key, value] : attributes_)
{
opentelemetry::nostd::visit(
ForwardOwnedValue{*inner_, {key.data(), key.size()}}, value);
}
return std::move(inner_);
}
void
SetAttribute(
opentelemetry::nostd::string_view key,
opentelemetry::common::AttributeValue const& value) noexcept override
{
auto owned = opentelemetry::sdk::common::VisitVariant(
opentelemetry::sdk::common::AttributeConverter{}, value);
if (!owned.second)
{
return;
}
std::string_view const name{key.data(), key.size()};
auto const found = std::ranges::find_if(
attributes_, [name](Attribute const& attribute) { return attribute.first == name; });
if (found != attributes_.end())
{
found->second = std::move(owned.first);
}
else
{
attributes_.emplace_back(std::string{name}, std::move(owned.first));
}
}
void
SetIdentity(trace_api::SpanContext const& spanContext, trace_api::SpanId parentSpanId) noexcept
override
{
inner_->SetIdentity(spanContext, parentSpanId);
}
void
AddEvent(
opentelemetry::nostd::string_view name,
opentelemetry::common::SystemTimestamp timestamp,
opentelemetry::common::KeyValueIterable const& attributes) noexcept override
{
inner_->AddEvent(name, timestamp, attributes);
}
void
AddLink(
trace_api::SpanContext const& spanContext,
opentelemetry::common::KeyValueIterable const& attributes) noexcept override
{
inner_->AddLink(spanContext, attributes);
}
void
SetStatus(trace_api::StatusCode code, opentelemetry::nostd::string_view description) noexcept
override
{
inner_->SetStatus(code, description);
}
void
SetName(opentelemetry::nostd::string_view name) noexcept override
{
inner_->SetName(name);
}
void
SetTraceFlags(trace_api::TraceFlags flags) noexcept override
{
inner_->SetTraceFlags(flags);
}
void
SetSpanKind(trace_api::SpanKind spanKind) noexcept override
{
inner_->SetSpanKind(spanKind);
}
void
SetResource(opentelemetry::sdk::resource::Resource const& resource) noexcept override
{
inner_->SetResource(resource);
}
void
SetStartTime(opentelemetry::common::SystemTimestamp startTime) noexcept override
{
inner_->SetStartTime(startTime);
}
void
SetDuration(std::chrono::nanoseconds duration) noexcept override
{
inner_->SetDuration(duration);
}
void
SetSpanLimits(trace_sdk::SpanLimits const& limits) noexcept override
{
inner_->SetSpanLimits(limits);
}
void
SetInstrumentationScope(
opentelemetry::sdk::instrumentationscope::InstrumentationScope const& scope) noexcept
override
{
inner_->SetInstrumentationScope(scope);
}
};
} // namespace
FilteringSpanProcessor::FilteringSpanProcessor(std::unique_ptr<trace_sdk::SpanProcessor> delegate)
: delegate_(std::move(delegate))
{
}
std::unique_ptr<trace_sdk::Recordable>
FilteringSpanProcessor::MakeRecordable() noexcept
{
auto inner = delegate_->MakeRecordable();
if (inner == nullptr)
{
return nullptr;
}
return std::make_unique<UniqueAttributeRecordable>(std::move(inner));
}
void
FilteringSpanProcessor::OnStart(
trace_sdk::Recordable& span,
trace_api::SpanContext const& parentContext) noexcept
{
auto* const wrapper = dynamic_cast<UniqueAttributeRecordable*>(&span);
delegate_->OnStart(wrapper != nullptr ? wrapper->inner() : span, parentContext);
}
void
FilteringSpanProcessor::OnEnd(std::unique_ptr<trace_sdk::Recordable>&& span) noexcept
{
if (DiscardScope::isActive())
{
// SpanGuard::discard() is inside a DiscardScope on this thread,
// which it entered just before calling Span::End() — and End()
// invokes OnEnd() synchronously. Drop the span.
return;
}
auto* const wrapper = dynamic_cast<UniqueAttributeRecordable*>(span.get());
if (wrapper == nullptr)
{
delegate_->OnEnd(std::move(span));
return;
}
delegate_->OnEnd(wrapper->flushAndRelease());
}
bool
FilteringSpanProcessor::ForceFlush(std::chrono::microseconds timeout) noexcept
{
return delegate_->ForceFlush(timeout);
}
bool
FilteringSpanProcessor::Shutdown(std::chrono::microseconds timeout) noexcept
{
return delegate_->Shutdown(timeout);
}
} // namespace xrpl::telemetry
#endif // XRPL_ENABLE_TELEMETRY
+54 -69
View File
@@ -9,13 +9,13 @@
* - SpanGuard::Impl holds ONLY the OTel Span (shared_ptr). It never
* touches the thread-local context stack, so a SpanGuard is
* thread-free and may be moved to and destroyed on any thread.
* - ScopedSpanGuard::ScopedImpl holds a SpanGuard plus an optional
* Scope. The Scope pushes the span onto the constructing context
* store's stack; member order (guard first, scope second) ensures
* the Scope pops BEFORE the span ends on destruction. It is
* store-bound: construct and destroy while the same LocalValue
* context store is active (the same thread, or the same JobQueue
* coroutine even if it resumes on another worker).
* - ScopedSpanGuard holds a SpanGuard inline plus a ScopedImpl, which
* holds the Scope and is allocated only for a live span. The Scope
* pushes the span onto the constructing context store's stack; member
* order (guard_ first, impl_ second) ensures the Scope pops BEFORE the
* span ends on destruction. It is store-bound: construct and destroy
* while the same LocalValue context store is active (the same thread,
* or the same JobQueue coroutine even if it resumes on another worker).
*
* Static factory methods access the global Telemetry instance via
* Telemetry::getInstance(), check whether the requested TraceCategory
@@ -23,7 +23,7 @@
* null guard whose methods are all no-ops.
*
* @see SpanGuard, ScopedSpanGuard (SpanGuard.h), Telemetry (Telemetry.h),
* FilteringSpanProcessor (Telemetry.cpp)
* FilteringSpanProcessor (FilteringSpanProcessor.h)
*/
#ifdef XRPL_ENABLE_TELEMETRY
@@ -677,63 +677,48 @@ SpanGuard::discard() noexcept
struct ScopedSpanGuard::ScopedImpl
{
/**
* The unscoped guard that owns the span. Declared FIRST so it is
* destroyed AFTER `scope`: the Scope must pop before the span ends.
* Active OTel Scope binding the guard's span to the active context
* store's stack. Destroying this ScopedImpl pops it; operator
* SpanGuard() and discard() do that eagerly under the constructing
* store.
*/
SpanGuard guard;
/**
* Active OTel Scope binding the span to the active context store's
* stack. Present while the guard is active; reset() (via operator
* SpanGuard) pops it eagerly under the constructing store. Declared
* AFTER `guard` so destruction order pops the scope before the span
* ends.
*/
std::optional<otel_trace::Scope> scope;
otel_trace::Scope scope;
/**
* Identity of the LocalValue store the Scope was pushed onto (the
* coroutine's store on a coro, else the thread's own store). Popping
* the Scope (via ~Scope or reset()) must happen while the SAME store
* is active, else a different stack is corrupted. Coroutine-aware
* storage lets a coro legitimately resume on another worker while
* keeping the same store, so store-identity — not thread-id — is the
* correct invariant. Checked in the destructor and operator
* SpanGuard() to turn a silent cross-store pop into an assertion
* failure in debug/test/fuzzing builds.
* the Scope must happen while the SAME store is active, else a
* different stack is corrupted. Coroutine-aware storage lets a coro
* legitimately resume on another worker while keeping the same store,
* so store-identity — not thread-id — is the correct invariant.
* Checked in the destructor, operator SpanGuard() and discard() to
* turn a silent cross-store pop into an assertion failure in
* debug/test/fuzzing builds.
*
* Assigned in the constructor body AFTER the Scope push, not as a
* member initializer: the push is the first LocalValue touch on a
* fresh thread for a root or explicit-parent span (the OTel SDK skips
* the ambient-context read in those cases), so it materializes the
* store. Capturing before the push would record nullptr while the
* destructor sees the now-materialized store. For a null guard no
* Scope is pushed and the assertions short-circuit, so this holds
* whatever store is active then.
* Initialized AFTER `scope` in the member init list: members
* initialize in declaration order, so the Scope's push runs first. On a
* thread with no store yet, the push creates it; GetCurrent() only peeks
* and never creates one. Capturing before the push would record nullptr
* while the destructor sees the new store.
*/
void const* owner = nullptr;
void const* owner;
/**
* Wrap a SpanGuard and activate its span on this thread. If the
* guard is active, push its span onto the thread-local context
* stack; a null guard leaves `scope` empty.
* @param g The span-owning guard to activate.
* Push a live guard's span onto the current context store.
* @param span The span to activate. The owning SpanGuard keeps it alive.
*/
explicit ScopedImpl(SpanGuard g) : guard(std::move(g))
explicit ScopedImpl(opentelemetry::nostd::shared_ptr<otel_trace::Span> const& span)
: scope(span), owner(detail::getLocalValues().get())
{
if (guard)
scope.emplace(guard.impl_->span);
// Capture after the push so `owner` names the store the Scope
// actually lives on, even when the push materialized it.
owner = detail::getLocalValues().get();
}
};
// ===== ScopedSpanGuard core lifecycle ======================================
ScopedSpanGuard::ScopedSpanGuard(SpanGuard&& guard) noexcept
: impl_(std::make_unique<ScopedImpl>(std::move(guard)))
ScopedSpanGuard::ScopedSpanGuard(SpanGuard&& guard) noexcept : guard_(std::move(guard))
{
if (guard_)
impl_ = std::make_unique<ScopedImpl>(guard_.impl_->span);
}
ScopedSpanGuard::~ScopedSpanGuard()
@@ -743,7 +728,7 @@ ScopedSpanGuard::~ScopedSpanGuard()
// store's context stack. A null guard holds no Scope, so the check is
// skipped.
XRPL_ASSERT(
!impl_ || !impl_->scope.has_value() || impl_->owner == detail::getLocalValues().get(),
!impl_ || impl_->owner == detail::getLocalValues().get(),
"xrpl::telemetry::ScopedSpanGuard::~ScopedSpanGuard : destroyed on the "
"constructing context store");
}
@@ -772,7 +757,7 @@ ScopedSpanGuard::freshRoot(
ScopedSpanGuard
ScopedSpanGuard::childSpan(std::string_view name) const noexcept
{
return ScopedSpanGuard(impl_->guard.childSpan(name));
return ScopedSpanGuard(guard_.childSpan(name));
}
ScopedSpanGuard
@@ -784,7 +769,7 @@ ScopedSpanGuard::childSpan(std::string_view name, SpanContext const& parentCtx)
ScopedSpanGuard
ScopedSpanGuard::linkedSpan(std::string_view name) const noexcept
{
return ScopedSpanGuard(impl_->guard.linkedSpan(name));
return ScopedSpanGuard(guard_.linkedSpan(name));
}
ScopedSpanGuard
@@ -802,11 +787,11 @@ operator SpanGuard() && noexcept
// active; handing off under a different store would pop the wrong stack.
// A null guard holds no Scope, so the check is skipped.
XRPL_ASSERT(
!impl_->scope.has_value() || impl_->owner == detail::getLocalValues().get(),
!impl_ || impl_->owner == detail::getLocalValues().get(),
"xrpl::telemetry::ScopedSpanGuard::operator SpanGuard : handoff on the "
"constructing context store");
impl_->scope.reset(); // eager pop, on the origin store
return std::move(impl_->guard);
impl_.reset(); // eager pop, on the origin store
return std::move(guard_);
}
// ===== ScopedSpanGuard context capture =====================================
@@ -814,7 +799,7 @@ operator SpanGuard() && noexcept
SpanContext
ScopedSpanGuard::spanContext() const noexcept
{
return impl_->guard.spanContext();
return guard_.spanContext();
}
// ===== ScopedSpanGuard forwarding methods ==================================
@@ -822,49 +807,49 @@ ScopedSpanGuard::spanContext() const noexcept
void
ScopedSpanGuard::setAttribute(std::string_view key, std::string_view value) noexcept
{
impl_->guard.setAttribute(key, value);
guard_.setAttribute(key, value);
}
void
ScopedSpanGuard::setAttribute(std::string_view key, char const* value) noexcept
{
impl_->guard.setAttribute(key, value);
guard_.setAttribute(key, value);
}
void
ScopedSpanGuard::setAttribute(std::string_view key, std::int64_t value) noexcept
{
impl_->guard.setAttribute(key, value);
guard_.setAttribute(key, value);
}
void
ScopedSpanGuard::setAttribute(std::string_view key, double value) noexcept
{
impl_->guard.setAttribute(key, value);
guard_.setAttribute(key, value);
}
void
ScopedSpanGuard::setAttribute(std::string_view key, bool value) noexcept
{
impl_->guard.setAttribute(key, value);
guard_.setAttribute(key, value);
}
void
ScopedSpanGuard::setOk() noexcept
{
impl_->guard.setOk();
guard_.setOk();
}
void
ScopedSpanGuard::setError(std::string_view description) noexcept
{
impl_->guard.setError(description);
guard_.setError(description);
}
void
ScopedSpanGuard::addEvent(std::string_view name) noexcept
{
impl_->guard.addEvent(name);
guard_.addEvent(name);
}
void
@@ -872,13 +857,13 @@ ScopedSpanGuard::addEvent(
std::string_view name,
std::initializer_list<EventAttribute> attrs) noexcept
{
impl_->guard.addEvent(name, attrs);
guard_.addEvent(name, attrs);
}
void
ScopedSpanGuard::recordException(std::exception const& e) noexcept
{
impl_->guard.recordException(e);
guard_.recordException(e);
}
void
@@ -888,19 +873,19 @@ ScopedSpanGuard::discard() noexcept
// active; discarding under a different store corrupts that store's
// context stack. A null guard holds no Scope, so the check is skipped.
XRPL_ASSERT(
!impl_ || !impl_->scope.has_value() || impl_->owner == detail::getLocalValues().get(),
!impl_ || impl_->owner == detail::getLocalValues().get(),
"xrpl::telemetry::ScopedSpanGuard::discard : discarded on the "
"constructing context store");
// Pop the scope first (under the constructing store) so no span stays
// active on the stack, then discard the span via the owned guard.
impl_->scope.reset();
impl_->guard.discard();
impl_.reset();
guard_.discard();
}
ScopedSpanGuard::
operator bool() const noexcept
{
return impl_ && static_cast<bool>(impl_->guard);
return static_cast<bool>(guard_);
}
// ===== ScopedActivation ====================================================
+37 -101
View File
@@ -4,8 +4,6 @@
* Compiled only when XRPL_ENABLE_TELEMETRY is defined (via CMake
* telemetry=ON). Contains:
*
* - FilteringSpanProcessor: decorator that drops spans marked with
* kDiscardedAttr before they enter the batch export queue.
* - TelemetryImpl: configures the OTel SDK with an OTLP/HTTP exporter,
* FilteringSpanProcessor wrapping a batch span processor, the head
* sampler from makeHeadSampler(), and resource attributes.
@@ -23,7 +21,7 @@
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/telemetry/CoroAwareContextStorage.h>
#include <xrpl/telemetry/DeterministicIdGenerator.h>
#include <xrpl/telemetry/DiscardFlag.h>
#include <xrpl/telemetry/FilteringSpanProcessor.h>
#include <xrpl/telemetry/HeadSampler.h>
#include <xrpl/telemetry/SpanNames.h>
@@ -32,10 +30,10 @@
#include <opentelemetry/exporters/otlp/otlp_http_exporter_factory.h>
#include <opentelemetry/exporters/otlp/otlp_http_exporter_options.h>
#include <opentelemetry/nostd/shared_ptr.h>
#include <opentelemetry/nostd/string_view.h>
#include <opentelemetry/sdk/resource/resource.h>
#include <opentelemetry/sdk/trace/batch_span_processor_factory.h>
#include <opentelemetry/sdk/trace/batch_span_processor_options.h>
#include <opentelemetry/sdk/trace/processor.h>
#include <opentelemetry/sdk/trace/sampler.h>
#include <opentelemetry/sdk/trace/samplers/always_off.h>
#include <opentelemetry/sdk/trace/samplers/always_on.h>
@@ -69,89 +67,19 @@ namespace otlp_http = opentelemetry::exporter::otlp;
namespace resource = opentelemetry::sdk::resource;
/**
* SpanProcessor decorator that drops discarded spans.
* View a std::string_view as the SDK's string view, without a copy.
*
* Wraps a delegate processor (typically BatchSpanProcessor). In OnEnd(),
* calls DiscardScope::isActive(). If the calling thread is inside a
* DiscardScope (entered by SpanGuard::discard()), the span is silently
* dropped — never entering the batch queue, never sent over the network,
* never stored.
* Without the SDK's STL option, nostd::string_view is a separate type with
* no constructor from std::string_view.
*
* Uses a thread-local flag rather than inspecting Recordable attributes
* because the Recordable type varies by exporter (SpanData for simple
* exporters, OtlpRecordable for OTLP) and none expose a uniform getter.
* The flag is safe because Span::End() calls OnEnd() synchronously on
* the same thread.
*
* All other methods delegate directly to the wrapped processor.
*
* Dependency diagram:
*
* +---------------------------+
* | FilteringSpanProcessor |
* +---------------------------+
* | - delegate_ : unique_ptr |
* | <SpanProcessor> |
* +---------------------------+
* | wraps
* +---------+-----------+
* | BatchSpanProcessor |
* +---------------------+
*
* @note Thread safety: OnEnd() may be called concurrently from multiple
* threads. The discard flag behind DiscardScope is thread-local, so each
* thread's discard state is independent — no synchronization needed.
* @param text Characters that must outlive the returned view.
* @return A view of the same characters.
*/
class FilteringSpanProcessor : public trace_sdk::SpanProcessor
[[nodiscard]] opentelemetry::nostd::string_view
toOtelView(std::string_view text) noexcept
{
std::unique_ptr<trace_sdk::SpanProcessor> delegate_;
public:
explicit FilteringSpanProcessor(std::unique_ptr<trace_sdk::SpanProcessor> delegate)
: delegate_(std::move(delegate))
{
}
std::unique_ptr<trace_sdk::Recordable>
MakeRecordable() noexcept override
{
return delegate_->MakeRecordable();
}
void
OnStart(
trace_sdk::Recordable& span,
opentelemetry::trace::SpanContext const& parentContext) noexcept override
{
delegate_->OnStart(span, parentContext);
}
void
OnEnd(std::unique_ptr<trace_sdk::Recordable>&& span) noexcept override
{
if (DiscardScope::isActive())
{
// SpanGuard::discard() is inside a DiscardScope on this thread,
// which it entered just before calling Span::End() — and End()
// invokes OnEnd() synchronously. Drop the span.
return;
}
delegate_->OnEnd(std::move(span));
}
bool
ForceFlush(
std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override
{
return delegate_->ForceFlush(timeout);
}
bool
Shutdown(std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override
{
return delegate_->Shutdown(timeout);
}
};
return {text.data(), text.size()};
}
/**
* No-op implementation used when XRPL_ENABLE_TELEMETRY is defined but
@@ -282,6 +210,17 @@ class TelemetryImpl : public Telemetry
*/
std::shared_ptr<trace_sdk::TracerProvider> sdkProvider_;
/**
* Tracer for kTracerName, cached so a span does not take the provider's
* lock.
*
* Starts as the global provider's tracer. start() sets it to the SDK
* tracer before setInstance() publishes this object, and stop() never
* writes it, so readers need no lock.
*/
opentelemetry::nostd::shared_ptr<trace_api::Tracer> tracer_ =
trace_api::Provider::GetTracerProvider()->GetTracer(toOtelView(kTracerName));
/**
* Coroutine-aware runtime-context storage, installed globally so the OTel
* ambient context follows JobQueue coroutines. Held for the process
@@ -318,8 +257,8 @@ public:
auto batchProcessor =
trace_sdk::BatchSpanProcessorFactory::Create(std::move(exporter), processorOpts);
// Wrap batch processor with filtering processor that drops spans
// marked with kDiscardedAttr (via SpanGuard::discard()).
// Wrap the batch processor. FilteringSpanProcessor drops discarded
// spans and exports each attribute key once, with its last value.
auto processor = std::make_unique<FilteringSpanProcessor>(std::move(batchProcessor));
// Configure resource attributes
@@ -364,6 +303,10 @@ public:
trace_api::Provider::SetTracerProvider(
opentelemetry::nostd::shared_ptr<trace_api::TracerProvider>(sdkProvider_));
// GetTracer() locks the provider, so fetch the tracer once here rather
// than per span. It must be set before setInstance() publishes it.
tracer_ = sdkProvider_->GetTracer(toOtelView(kTracerName));
// Register as the global Telemetry instance so SpanGuard factory
// methods can access it without callers passing a reference.
Telemetry::setInstance(this);
@@ -388,13 +331,10 @@ public:
// Force flush with timeout to avoid blocking indefinitely
// when the OTLP endpoint is unreachable.
sdkProvider_->ForceFlush(std::chrono::milliseconds(5000));
// TODO: sdkProvider_ is not thread-safe. This reset() races with
// getTracer() if any thread is still calling startSpan().
// Currently safe because Application::stop() shuts down
// serverHandler_, overlay_, and jobQueue_ before calling
// telemetry_->stop() — so no callers should remain. If the
// shutdown order ever changes, add an std::atomic<bool> stopped_
// flag checked in getTracer() to make this robust.
// startSpan() and getTracer() never read sdkProvider_, so this
// reset() cannot race with them. A span started after stop()
// still goes through tracer_, into the shut-down pipeline, and is
// dropped rather than exported.
sdkProvider_.reset();
trace_api::Provider::SetTracerProvider(
opentelemetry::nostd::shared_ptr<trace_api::TracerProvider>(
@@ -448,20 +388,17 @@ public:
[[nodiscard]] opentelemetry::nostd::shared_ptr<trace_api::Tracer>
getTracer(std::string_view name = kTracerName) override
{
if (!sdkProvider_)
{
return trace_api::Provider::GetTracerProvider()->GetTracer(std::string(name));
}
return sdkProvider_->GetTracer(std::string(name));
if (name == kTracerName)
return tracer_;
return trace_api::Provider::GetTracerProvider()->GetTracer(toOtelView(name));
}
[[nodiscard]] opentelemetry::nostd::shared_ptr<trace_api::Span>
startSpan(std::string_view name, trace_api::SpanKind kind) override
{
auto tracer = getTracer();
trace_api::StartSpanOptions opts;
opts.kind = kind;
return tracer->StartSpan(std::string(name), opts);
return tracer_->StartSpan(toOtelView(name), opts);
}
[[nodiscard]] opentelemetry::nostd::shared_ptr<trace_api::Span>
@@ -470,11 +407,10 @@ public:
opentelemetry::context::Context const& parentContext,
trace_api::SpanKind kind) override
{
auto tracer = getTracer();
trace_api::StartSpanOptions opts;
opts.kind = kind;
opts.parent = parentContext;
return tracer->StartSpan(std::string(name), opts);
return tracer_->StartSpan(toOtelView(name), opts);
}
};
+39 -7
View File
@@ -9,6 +9,7 @@
*/
#include <xrpl/basics/contract.h>
#include <xrpl/beast/core/LexicalCast.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/telemetry/Telemetry.h>
@@ -138,6 +139,37 @@ readBounded(
return static_cast<std::uint32_t>(*parsed);
}
/**
* Read an on/off key. Accepts 0, 1, true or false, in any case.
*
* Reading it as a number would throw boost::bad_lexical_cast for "true",
* naming no key. This reads the text and names the key instead.
*
* @param section The [telemetry] section to read from.
* @param name Key to read, as documented in cfg/xrpld-example.cfg.
* @param absentValue Value returned when the key is absent.
* @return The configured value, or absentValue if the key is absent.
* @throws std::runtime_error For any other value, with a message naming the
* key.
*/
[[nodiscard]] bool
readFlag(Section const& section, std::string_view name, bool absentValue)
{
std::string const keyName{name};
auto const text = section.get(keyName);
if (!text)
return absentValue;
bool flag = false;
if (!beast::lexicalCastChecked(flag, *text))
{
Throw<std::runtime_error>(
"Invalid value '" + keyName + "' in " + kSectionLabel +
": must be 0, 1, true or false.");
}
return flag;
}
/**
* Derive a human-readable network type label from the numeric network ID.
* @param networkId The network identifier from [network_id] config.
@@ -272,14 +304,14 @@ makeTelemetrySetup(
{
Telemetry::Setup setup;
setup.enabled = section.valueOr<int>(key::enabled, 0) != 0;
setup.enabled = readFlag(section, key::enabled, false);
setup.serviceName = section.valueOr<std::string>(key::serviceName, dflt::serviceName);
setup.serviceVersion = version;
setup.serviceInstanceId = section.valueOr<std::string>(key::serviceInstanceId, nodePublicKey);
setup.tracesEndpoint = section.valueOr<std::string>(key::tracesEndpoint, dflt::tracesEndpoint);
setup.useTls = section.valueOr<int>(key::useTls, 0) != 0;
setup.useTls = readFlag(section, key::useTls, false);
setup.tlsCertPath = section.valueOr<std::string>(key::tlsCaCert, "");
setup.tlsClientCertPath = section.valueOr<std::string>(key::tlsClientCert, "");
setup.tlsClientKeyPath = section.valueOr<std::string>(key::tlsClientKey, "");
@@ -366,11 +398,11 @@ makeTelemetrySetup(
setup.networkId = networkId;
setup.networkType = networkTypeFromId(networkId);
setup.traceTransactions = section.valueOr<int>(key::traceTransactions, 1) != 0;
setup.traceConsensus = section.valueOr<int>(key::traceConsensus, 1) != 0;
setup.traceRpc = section.valueOr<int>(key::traceRpc, 1) != 0;
setup.tracePeer = section.valueOr<int>(key::tracePeer, 1) != 0;
setup.traceLedger = section.valueOr<int>(key::traceLedger, 1) != 0;
setup.traceTransactions = readFlag(section, key::traceTransactions, true);
setup.traceConsensus = readFlag(section, key::traceConsensus, true);
setup.traceRpc = readFlag(section, key::traceRpc, true);
setup.tracePeer = readFlag(section, key::tracePeer, true);
setup.traceLedger = readFlag(section, key::traceLedger, true);
setup.consensusTraceStrategy =
readConsensusTraceStrategy(section.valueOr<std::string>(key::consensusTraceStrategy, ""));
@@ -9,7 +9,6 @@
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/ledger/PendingSaves.h>
#include <xrpl/server/LoadFeeTrack.h>
#include <xrpl/telemetry/Telemetry.h>
#include <boost/asio/io_context.hpp>
@@ -338,12 +337,6 @@ public:
throw std::logic_error("TestServiceRegistry::getPerfLog() not implemented");
}
telemetry::Telemetry&
getTelemetry() override
{
throw std::logic_error("TestServiceRegistry::getTelemetry() not implemented");
}
// Configuration and state
bool
isStopping() const override
@@ -0,0 +1,812 @@
// Tests for FilteringSpanProcessor.
//
// The OTLP recordable appends every SetAttribute() call. Without the
// processor a key set twice is exported twice, and a reader that takes the
// first value misses the update. These tests put the processor in front of a
// fake delegate whose recordable logs every call it receives. They check
// that each key arrives once, with its last value, in first-set order; that
// values are copied when set; that every other call is forwarded at once and
// unchanged; that the delegate gets its own recordable back; and that
// discarded spans are still dropped. One test uses the real OTLP recordable
// to show its attribute count limit still applies.
//
// The whole file is telemetry-only: without XRPL_ENABLE_TELEMETRY the SDK
// headers are unavailable, so the translation unit compiles empty.
#ifdef XRPL_ENABLE_TELEMETRY
#include <xrpl/telemetry/FilteringSpanProcessor.h>
#include <xrpl/telemetry/DiscardFlag.h>
#include <gtest/gtest.h>
#include <opentelemetry/common/attribute_value.h>
#include <opentelemetry/common/key_value_iterable.h>
#include <opentelemetry/common/key_value_iterable_view.h>
#include <opentelemetry/common/timestamp.h>
#include <opentelemetry/exporters/otlp/otlp_recordable.h>
#include <opentelemetry/nostd/span.h>
#include <opentelemetry/nostd/string_view.h>
#include <opentelemetry/nostd/variant.h>
#include <opentelemetry/sdk/instrumentationscope/instrumentation_scope.h>
#include <opentelemetry/sdk/resource/resource.h>
#include <opentelemetry/sdk/trace/processor.h>
#include <opentelemetry/sdk/trace/recordable.h>
#include <opentelemetry/sdk/trace/samplers/always_on_factory.h>
#include <opentelemetry/sdk/trace/span_limits.h>
#include <opentelemetry/sdk/trace/tracer_provider.h>
#include <opentelemetry/sdk/trace/tracer_provider_factory.h>
#include <opentelemetry/trace/span.h>
#include <opentelemetry/trace/span_context.h>
#include <opentelemetry/trace/span_id.h>
#include <opentelemetry/trace/span_metadata.h>
#include <opentelemetry/trace/trace_flags.h>
#include <opentelemetry/trace/trace_id.h>
#include <opentelemetry/trace/tracer.h>
#include <algorithm>
#include <array>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <format>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <utility>
#include <vector>
namespace xrpl::telemetry {
namespace {
namespace otel_common = opentelemetry::common;
namespace otel_nostd = opentelemetry::nostd;
namespace otel_trace = opentelemetry::trace;
namespace otel_sdk_trace = opentelemetry::sdk::trace;
/**
* @return the text of an SDK string view.
*/
std::string
toString(otel_nostd::string_view text)
{
return {text.data(), text.size()};
}
/**
* @return a span id as 16 lowercase hex digits.
*/
std::string
toHex(otel_trace::SpanId const& id)
{
std::array<char, 2 * otel_trace::SpanId::kSize> text{};
id.ToLowerBase16(text);
return {text.data(), text.size()};
}
/**
* @return a span id holding these bytes.
*/
otel_trace::SpanId
spanId(std::array<std::uint8_t, otel_trace::SpanId::kSize> const& bytes)
{
return otel_trace::SpanId{otel_nostd::span<std::uint8_t const, otel_trace::SpanId::kSize>{
bytes.data(), bytes.size()}};
}
/**
* @return a trace id holding these bytes.
*/
otel_trace::TraceId
traceId(std::array<std::uint8_t, otel_trace::TraceId::kSize> const& bytes)
{
return otel_trace::TraceId{otel_nostd::span<std::uint8_t const, otel_trace::TraceId::kSize>{
bytes.data(), bytes.size()}};
}
/**
* Renders an attribute value with its type, such as `int64:7` or
* `string:"open"`, so one string comparison checks both. Array types the
* tests never set render as `unrendered`.
*/
struct RenderValue
{
std::string
operator()(bool value) const
{
return std::format("bool:{}", value);
}
std::string
operator()(std::int32_t value) const
{
return std::format("int32:{}", value);
}
std::string
operator()(std::uint32_t value) const
{
return std::format("uint32:{}", value);
}
std::string
operator()(std::int64_t value) const
{
return std::format("int64:{}", value);
}
std::string
operator()(std::uint64_t value) const
{
return std::format("uint64:{}", value);
}
std::string
operator()(double value) const
{
return std::format("double:{}", value);
}
std::string
operator()(char const* value) const
{
return std::format("string:\"{}\"", value);
}
std::string
operator()(otel_nostd::string_view value) const
{
return std::format("string:\"{}\"", toString(value));
}
std::string
operator()(otel_nostd::span<bool const> values) const
{
std::string text;
for (bool const value : values)
{
text += std::format("{}{}", text.empty() ? "" : ",", value);
}
return std::format("bool[]:[{}]", text);
}
std::string
operator()(otel_nostd::span<std::int64_t const> values) const
{
std::string text;
for (std::int64_t const value : values)
{
text += std::format("{}{}", text.empty() ? "" : ",", value);
}
return std::format("int64[]:[{}]", text);
}
std::string
operator()(otel_nostd::span<otel_nostd::string_view const> values) const
{
std::string text;
for (auto const& value : values)
{
text += std::format("{}\"{}\"", text.empty() ? "" : ",", toString(value));
}
return std::format("string[]:[{}]", text);
}
template <typename T, std::size_t N>
std::string
operator()(otel_nostd::span<T, N> const&) const
{
return "unrendered";
}
};
/**
* @return the attributes of an event or link as `key=value` pairs.
*/
std::string
render(otel_common::KeyValueIterable const& attributes)
{
std::string text;
attributes.ForEachKeyValue(
[&text](otel_nostd::string_view key, otel_common::AttributeValue value) noexcept {
text += std::format(
"{}{}={}",
text.empty() ? "" : ",",
toString(key),
otel_nostd::visit(RenderValue{}, value));
return true;
});
return text;
}
/**
* Recordable that stands in for the exporter's. It logs every call it
* receives, one line per call, so a test can compare the whole sequence.
*
* @note Test-only. Not thread-safe; each test drives it from one thread.
*/
class RecordingRecordable final : public otel_sdk_trace::Recordable
{
public:
/**
* One line per call, in the order received.
*/
std::vector<std::string> calls;
/**
* The last resource passed in, to check the same object arrives.
*/
opentelemetry::sdk::resource::Resource const* resource = nullptr;
/**
* The last scope passed in, to check the same object arrives.
*/
opentelemetry::sdk::instrumentationscope::InstrumentationScope const* scope = nullptr;
void
SetIdentity(
otel_trace::SpanContext const& spanContext,
otel_trace::SpanId parentSpanId) noexcept override
{
calls.push_back(
std::format(
"SetIdentity {} parent {}", toHex(spanContext.span_id()), toHex(parentSpanId)));
}
void
SetAttribute(otel_nostd::string_view key, otel_common::AttributeValue const& value) noexcept
override
{
calls.push_back(
std::format(
"SetAttribute {}={}", toString(key), otel_nostd::visit(RenderValue{}, value)));
}
void
AddEvent(
otel_nostd::string_view name,
otel_common::SystemTimestamp timestamp,
otel_common::KeyValueIterable const& attributes) noexcept override
{
calls.push_back(
std::format(
"AddEvent {} at {} {{{}}}",
toString(name),
timestamp.time_since_epoch().count(),
render(attributes)));
}
void
AddLink(
otel_trace::SpanContext const& spanContext,
otel_common::KeyValueIterable const& attributes) noexcept override
{
calls.push_back(
std::format("AddLink {} {{{}}}", toHex(spanContext.span_id()), render(attributes)));
}
void
SetStatus(otel_trace::StatusCode code, otel_nostd::string_view description) noexcept override
{
calls.push_back(
std::format("SetStatus {} {}", std::to_underlying(code), toString(description)));
}
void
SetName(otel_nostd::string_view name) noexcept override
{
calls.push_back(std::format("SetName {}", toString(name)));
}
void
SetTraceFlags(otel_trace::TraceFlags flags) noexcept override
{
calls.push_back(std::format("SetTraceFlags {}", static_cast<unsigned>(flags.flags())));
}
void
SetSpanKind(otel_trace::SpanKind kind) noexcept override
{
calls.push_back(std::format("SetSpanKind {}", std::to_underlying(kind)));
}
void
SetResource(opentelemetry::sdk::resource::Resource const& value) noexcept override
{
resource = &value;
calls.emplace_back("SetResource");
}
void
SetStartTime(otel_common::SystemTimestamp startTime) noexcept override
{
calls.push_back(std::format("SetStartTime {}", startTime.time_since_epoch().count()));
}
void
SetDuration(std::chrono::nanoseconds duration) noexcept override
{
calls.push_back(std::format("SetDuration {}", duration.count()));
}
void
SetSpanLimits(otel_sdk_trace::SpanLimits const& limits) noexcept override
{
calls.push_back(std::format("SetSpanLimits {}", limits.attribute_count_limit));
}
void
SetInstrumentationScope(
opentelemetry::sdk::instrumentationscope::InstrumentationScope const& value) noexcept
override
{
scope = &value;
calls.emplace_back("SetInstrumentationScope");
}
};
/**
* @return the calls a recordable received, or one line saying it is not a
* RecordingRecordable, so a wrong type shows up in the assertion message.
*/
std::vector<std::string>
callsOf(otel_sdk_trace::Recordable const* recordable)
{
auto const* recording = dynamic_cast<RecordingRecordable const*>(recordable);
if (recording == nullptr)
{
return {"not a RecordingRecordable"};
}
return recording->calls;
}
/**
* @return only the SetAttribute lines a recordable received.
*/
std::vector<std::string>
attributeCallsOf(otel_sdk_trace::Recordable const* recordable)
{
auto calls = callsOf(recordable);
std::erase_if(
calls, [](std::string const& call) { return !call.starts_with("SetAttribute "); });
return calls;
}
/**
* What the fake delegate saw. Kept outside the delegate so a test can read it
* after the processor that owns the delegate is gone.
*/
struct DelegateLog
{
/**
* Every recordable the delegate made, in order. Compared by address only.
*/
std::vector<otel_sdk_trace::Recordable const*> made;
/**
* Every recordable passed to OnStart(), in order. Compared by address only.
*/
std::vector<otel_sdk_trace::Recordable const*> started;
/**
* Every recordable passed to OnEnd(), kept alive for inspection.
*/
std::vector<std::unique_ptr<otel_sdk_trace::Recordable>> ended;
};
/**
* Span processor that stands in for the batch processor. It makes recordables
* with a factory and keeps every span it is handed in a DelegateLog.
*
* FilteringSpanProcessor
* | delegate_
* v
* RecordingProcessor --make_--> RecordingRecordable (logs each call)
* |
* +--> DelegateLog: made, started, ended
*
* @note Test-only. The log must outlive the processor.
*/
class RecordingProcessor final : public otel_sdk_trace::SpanProcessor
{
public:
/**
* Makes the recordable the delegate returns from MakeRecordable().
*/
using Factory = std::function<std::unique_ptr<otel_sdk_trace::Recordable>()>;
/**
* @param log Where to record what the delegate sees.
* @param make Factory for the delegate's recordables.
*/
RecordingProcessor(DelegateLog& log, Factory make) : log_(log), make_(std::move(make))
{
}
std::unique_ptr<otel_sdk_trace::Recordable>
MakeRecordable() noexcept override
{
auto recordable = make_();
log_.made.push_back(recordable.get());
return recordable;
}
void
OnStart(otel_sdk_trace::Recordable& span, otel_trace::SpanContext const&) noexcept override
{
log_.started.push_back(&span);
}
void
OnEnd(std::unique_ptr<otel_sdk_trace::Recordable>&& span) noexcept override
{
log_.ended.push_back(std::move(span));
}
bool
ForceFlush(std::chrono::microseconds) noexcept override
{
return true;
}
bool
Shutdown(std::chrono::microseconds) noexcept override
{
return true;
}
private:
DelegateLog& log_;
Factory make_;
};
/**
* @return a RecordingRecordable, as the delegate's recordable.
*/
std::unique_ptr<otel_sdk_trace::Recordable>
makeRecording()
{
return std::make_unique<RecordingRecordable>();
}
/**
* @return a recording delegate that writes to @p log.
*/
std::unique_ptr<RecordingProcessor>
recordingDelegate(DelegateLog& log, RecordingProcessor::Factory make = makeRecording)
{
return std::make_unique<RecordingProcessor>(log, std::move(make));
}
TEST(FilteringSpanProcessor, repeated_key_is_exported_once_with_its_last_value)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute("consensus_phase", "open");
span->SetAttribute("consensus_phase", "establish");
span->SetAttribute("consensus_phase", "accepted");
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{R"(SetAttribute consensus_phase=string:"accepted")"}));
}
TEST(FilteringSpanProcessor, distinct_keys_are_exported_once_each_in_first_set_order)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
// Not alphabetical, and the first key is set again last: a sorted
// container or a move-to-back on overwrite both give another order.
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute("zeta", std::int64_t{1});
span->SetAttribute("alpha", std::int64_t{2});
span->SetAttribute("mid", std::int64_t{3});
span->SetAttribute("zeta", std::int64_t{4});
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{
"SetAttribute zeta=int64:4",
"SetAttribute alpha=int64:2",
"SetAttribute mid=int64:3"}));
}
TEST(FilteringSpanProcessor, string_value_is_copied_when_set)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
std::string status{"accepted"};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute("tx_status", otel_nostd::string_view{status.data(), status.size()});
std::ranges::fill(status, 'x');
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{R"(SetAttribute tx_status=string:"accepted")"}));
}
TEST(FilteringSpanProcessor, key_is_copied_when_set)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
std::string key{"tx_type"};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute(otel_nostd::string_view{key.data(), key.size()}, true);
std::ranges::fill(key, 'x');
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{"SetAttribute tx_type=bool:true"}));
}
TEST(FilteringSpanProcessor, scalar_values_keep_their_type)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute("flag", true);
span->SetAttribute("count32", std::int32_t{-5});
span->SetAttribute("ucount32", std::uint32_t{5});
span->SetAttribute("count64", std::int64_t{-7});
span->SetAttribute("ucount64", std::uint64_t{7});
span->SetAttribute("ratio", 1.5);
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{
"SetAttribute flag=bool:true",
"SetAttribute count32=int32:-5",
"SetAttribute ucount32=uint32:5",
"SetAttribute count64=int64:-7",
"SetAttribute ucount64=uint64:7",
"SetAttribute ratio=double:1.5"}));
}
TEST(FilteringSpanProcessor, array_values_are_copied_and_exported_intact)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
std::array<bool, 3> flags{true, false, true};
std::array<std::int64_t, 2> counts{3, 4};
std::string first{"a"};
std::string second{"b"};
std::array<otel_nostd::string_view, 2> const names{
otel_nostd::string_view{first}, otel_nostd::string_view{second}};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetAttribute("flags", otel_nostd::span<bool const>{flags.data(), flags.size()});
span->SetAttribute(
"counts", otel_nostd::span<std::int64_t const>{counts.data(), counts.size()});
span->SetAttribute(
"names", otel_nostd::span<otel_nostd::string_view const>{names.data(), names.size()});
flags.fill(false);
counts.fill(0);
std::ranges::fill(first, 'x');
std::ranges::fill(second, 'y');
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(
callsOf(log.ended.front().get()),
(std::vector<std::string>{
"SetAttribute flags=bool[]:[true,false,true]",
"SetAttribute counts=int64[]:[3,4]",
R"(SetAttribute names=string[]:["a","b"])"}));
}
TEST(FilteringSpanProcessor, other_calls_are_forwarded_at_once_and_unchanged)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
otel_trace::SpanContext const context{
traceId({1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}),
spanId({1, 2, 3, 4, 5, 6, 7, 8}),
otel_trace::TraceFlags{otel_trace::TraceFlags::kIsSampled},
false};
otel_trace::SpanContext const linked{
traceId({16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1}),
spanId({9, 9, 9, 9, 9, 9, 9, 9}),
otel_trace::TraceFlags{otel_trace::TraceFlags::kIsSampled},
true};
otel_sdk_trace::SpanLimits limits;
limits.attribute_count_limit = 7;
auto const resource = opentelemetry::sdk::resource::Resource::Create({});
auto const scope =
opentelemetry::sdk::instrumentationscope::InstrumentationScope::Create("scope");
std::map<std::string, std::int64_t> const eventAttributes{{"ledger_seq", 9}};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetSpanLimits(limits);
span->SetName("tx.process");
span->SetInstrumentationScope(*scope);
span->SetIdentity(context, spanId({8, 7, 6, 5, 4, 3, 2, 1}));
span->SetTraceFlags(otel_trace::TraceFlags{otel_trace::TraceFlags::kIsSampled});
span->SetSpanKind(otel_trace::SpanKind::kServer);
span->SetStartTime(otel_common::SystemTimestamp{std::chrono::nanoseconds{100}});
span->SetResource(resource);
span->AddEvent(
"ledger.closed",
otel_common::SystemTimestamp{std::chrono::nanoseconds{150}},
otel_common::KeyValueIterableView<std::map<std::string, std::int64_t>>{eventAttributes});
span->AddLink(linked);
span->SetStatus(otel_trace::StatusCode::kError, "boom");
span->SetDuration(std::chrono::nanoseconds{250});
std::vector<std::string> const expected{
"SetSpanLimits 7",
"SetName tx.process",
"SetInstrumentationScope",
"SetIdentity 0102030405060708 parent 0807060504030201",
"SetTraceFlags 1",
"SetSpanKind 1",
"SetStartTime 100",
"SetResource",
"AddEvent ledger.closed at 150 {ledger_seq=int64:9}",
"AddLink 0909090909090909 {}",
"SetStatus 2 boom",
"SetDuration 250"};
// Every call has reached the delegate's recordable before the span ends.
ASSERT_EQ(log.made.size(), 1U);
EXPECT_EQ(callsOf(log.made.front()), expected);
auto const* recording = dynamic_cast<RecordingRecordable const*>(log.made.front());
ASSERT_NE(recording, nullptr);
EXPECT_EQ(recording->resource, &resource);
EXPECT_EQ(recording->scope, scope.get());
// Ending the span adds nothing when no attribute was set.
processor.OnEnd(std::move(span));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(callsOf(log.ended.front().get()), expected);
}
TEST(FilteringSpanProcessor, span_ended_inside_discard_scope_is_dropped)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
auto discarded = processor.MakeRecordable();
ASSERT_NE(discarded, nullptr);
discarded->SetAttribute("tx_status", "rejected");
{
DiscardScope const discardScope;
processor.OnEnd(std::move(discarded));
}
EXPECT_TRUE(log.ended.empty());
// Control: the same processor still exports a span ended outside the scope.
auto kept = processor.MakeRecordable();
ASSERT_NE(kept, nullptr);
processor.OnEnd(std::move(kept));
EXPECT_EQ(log.ended.size(), 1U);
}
TEST(FilteringSpanProcessor, delegate_gets_its_own_recordable_at_start_and_end)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
ASSERT_EQ(log.made.size(), 1U);
otel_sdk_trace::Recordable const* const inner = log.made.front();
EXPECT_NE(span.get(), inner);
processor.OnStart(*span, otel_trace::SpanContext::GetInvalid());
processor.OnEnd(std::move(span));
EXPECT_EQ(log.started, (std::vector<otel_sdk_trace::Recordable const*>{inner}));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(log.ended.front().get(), inner);
}
TEST(FilteringSpanProcessor, recordable_it_did_not_make_is_passed_on_unchanged)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log)};
auto foreign = std::make_unique<RecordingRecordable>();
otel_sdk_trace::Recordable const* const address = foreign.get();
processor.OnStart(*foreign, otel_trace::SpanContext::GetInvalid());
processor.OnEnd(std::move(foreign));
EXPECT_EQ(log.started, (std::vector<otel_sdk_trace::Recordable const*>{address}));
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(log.ended.front().get(), address);
}
TEST(FilteringSpanProcessor, null_recordable_from_the_delegate_stays_null)
{
DelegateLog log;
FilteringSpanProcessor processor{
recordingDelegate(log, [] { return std::unique_ptr<otel_sdk_trace::Recordable>{}; })};
EXPECT_EQ(processor.MakeRecordable(), nullptr);
EXPECT_EQ(log.made.size(), 1U);
}
TEST(FilteringSpanProcessor, otlp_recordable_gets_each_key_once_within_its_count_limit)
{
DelegateLog log;
FilteringSpanProcessor processor{recordingDelegate(log, [] {
return std::unique_ptr<otel_sdk_trace::Recordable>{
std::make_unique<opentelemetry::exporter::otlp::OtlpRecordable>()};
})};
otel_sdk_trace::SpanLimits limits;
limits.attribute_count_limit = 2;
auto span = processor.MakeRecordable();
ASSERT_NE(span, nullptr);
span->SetSpanLimits(limits);
span->SetAttribute("first", std::int64_t{1});
span->SetAttribute("second", std::int64_t{2});
span->SetAttribute("first", std::int64_t{3});
span->SetAttribute("third", std::int64_t{4});
processor.OnEnd(std::move(span));
// Two keys fit the limit. "first" keeps its place and its last value, and
// only "third" is dropped.
ASSERT_EQ(log.ended.size(), 1U);
auto const* otlp =
dynamic_cast<opentelemetry::exporter::otlp::OtlpRecordable const*>(log.ended.front().get());
ASSERT_NE(otlp, nullptr);
auto const& exported = otlp->span();
ASSERT_EQ(exported.attributes_size(), 2);
EXPECT_EQ(exported.attributes(0).key(), "first");
EXPECT_EQ(exported.attributes(0).value().int_value(), 3);
EXPECT_EQ(exported.attributes(1).key(), "second");
EXPECT_EQ(exported.attributes(1).value().int_value(), 2);
EXPECT_EQ(exported.dropped_attributes_count(), 1U);
}
TEST(FilteringSpanProcessor, sdk_span_that_sets_a_key_twice_exports_it_once)
{
DelegateLog log;
auto const provider = otel_sdk_trace::TracerProviderFactory::Create(
std::make_unique<FilteringSpanProcessor>(recordingDelegate(log)),
opentelemetry::sdk::resource::Resource::Create({}),
otel_sdk_trace::AlwaysOnSamplerFactory::Create());
auto const tracer = provider->GetTracer("filtering-span-processor-test");
auto const span = tracer->StartSpan("txq.enqueue");
span->SetAttribute("txq_status", "rejected");
span->SetAttribute("txq_status", "queued");
span->End();
ASSERT_EQ(log.made.size(), 1U);
EXPECT_EQ(log.started, log.made);
ASSERT_EQ(log.ended.size(), 1U);
EXPECT_EQ(log.ended.front().get(), log.made.front());
EXPECT_EQ(
attributeCallsOf(log.ended.front().get()),
(std::vector<std::string>{R"(SetAttribute txq_status=string:"queued")"}));
}
} // namespace
} // namespace xrpl::telemetry
#endif // XRPL_ENABLE_TELEMETRY
@@ -27,8 +27,12 @@ TEST(SpanGuardFactory, null_guard_methods_are_safe)
span.addEvent("event");
}
TEST(SpanGuardFactory, category_span_returns_null_when_disabled)
// No Telemetry is installed here, so span() returns a null guard before it
// reads the category. SpanGuardScope.cpp covers a category that is off.
TEST(SpanGuardFactory, category_span_returns_null_when_telemetry_absent)
{
ASSERT_EQ(Telemetry::getInstance(), nullptr);
auto span = SpanGuard::span(TraceCategory::Rpc, "rpc", "test");
EXPECT_FALSE(span);
+162 -11
View File
@@ -12,6 +12,10 @@
// pops that Scope eagerly on the origin store and yields a thread-free
// SpanGuard; destroying it while a different store is active trips an
// owner-store assertion.
// - A category that is off gets a null guard from SpanGuard::span,
// SpanGuard::freshRoot, the ScopedSpanGuard constructor and
// ScopedSpanGuard::freshRoot, and the methods called on a null
// ScopedSpanGuard are safe no-ops.
// - DeterministicIdGenerator (installed by the test TracerProvider) mints a
// caller-pinned trace_id for a forced-root span. PendingTraceId pins the id
// for one root span; an ambient child under a live parent never adopts it.
@@ -63,6 +67,8 @@
#include <cstdint>
#include <cstring>
#include <memory>
#include <set>
#include <stdexcept>
#include <string>
#include <string_view>
#include <thread>
@@ -79,12 +85,12 @@ namespace otel_memory = opentelemetry::exporter::memory;
/**
* In-memory Telemetry backing for SpanGuard scope tests.
*
* Reports every trace category as enabled and creates spans through an SDK
* TracerProvider whose SimpleSpanProcessor forwards ended spans to an
* InMemorySpanExporter, so a test can read the exact exported SpanData
* (trace id, span id, parent id, name). The provider is built with a
* DeterministicIdGenerator so PendingTraceId can pin the trace_id of a
* forced-root span.
* Reports every trace category as enabled until a test turns some off with
* setDisabledCategories(). Creates spans through an SDK TracerProvider whose
* SimpleSpanProcessor forwards ended spans to an InMemorySpanExporter, so a
* test can read the exact exported SpanData (trace id, span id, parent id,
* name). The provider is built with a DeterministicIdGenerator so
* PendingTraceId can pin the trace_id of a forced-root span.
*
* Inheritance:
*
@@ -129,6 +135,18 @@ public:
return spanData_;
}
/**
* Turn trace categories off, replacing the set from any earlier call.
* Call it only while no other thread is creating spans.
* @param categories Categories whose shouldTrace*() returns false. Every
* other category returns true.
*/
void
setDisabledCategories(std::set<TraceCategory> categories)
{
disabled_ = std::move(categories);
}
void
start() override
{
@@ -146,27 +164,27 @@ public:
[[nodiscard]] bool
shouldTraceTransactions() const override
{
return true;
return isCategoryOn(TraceCategory::Transactions);
}
[[nodiscard]] bool
shouldTraceConsensus() const override
{
return true;
return isCategoryOn(TraceCategory::Consensus);
}
[[nodiscard]] bool
shouldTraceRpc() const override
{
return true;
return isCategoryOn(TraceCategory::Rpc);
}
[[nodiscard]] bool
shouldTracePeer() const override
{
return true;
return isCategoryOn(TraceCategory::Peer);
}
[[nodiscard]] bool
shouldTraceLedger() const override
{
return true;
return isCategoryOn(TraceCategory::Ledger);
}
/**
@@ -206,6 +224,16 @@ public:
}
private:
/**
* @param cat Category to look up.
* @return false if a test turned cat off, else true.
*/
[[nodiscard]] bool
isCategoryOn(TraceCategory cat) const
{
return !disabled_.contains(cat);
}
/**
* SDK provider owning the export pipeline.
*/
@@ -215,6 +243,11 @@ private:
* Shared buffer that receives ended spans from the exporter.
*/
std::shared_ptr<otel_memory::InMemorySpanData> spanData_;
/**
* Categories a test turned off. Empty by default, so every category is on.
*/
std::set<TraceCategory> disabled_;
};
/**
@@ -556,6 +589,124 @@ TEST_F(SpanGuardScopeTest, scoped_guard_survives_localvalue_store_swap)
EXPECT_EQ(countSpans(spanData()->GetSpans(), "rpc.process"), 1u);
}
// A category that is off gets a null guard from SpanGuard::span,
// SpanGuard::freshRoot, the ScopedSpanGuard constructor and
// ScopedSpanGuard::freshRoot, while a category that is on keeps exporting.
// Each category is off once, with the next one as the control, so each is also
// seen exporting.
TEST_F(SpanGuardScopeTest, disabled_category_gets_null_guard_from_span_factories)
{
constexpr std::array cases{
std::pair{TraceCategory::Rpc, TraceCategory::Transactions},
std::pair{TraceCategory::Transactions, TraceCategory::Consensus},
std::pair{TraceCategory::Consensus, TraceCategory::Peer},
std::pair{TraceCategory::Peer, TraceCategory::Ledger},
std::pair{TraceCategory::Ledger, TraceCategory::Rpc},
};
for (auto const& [off, on] : cases)
{
SCOPED_TRACE(::testing::Message() << "category off: " << std::to_underlying(off));
telemetry_->setDisabledCategories({off});
{
auto const plain = SpanGuard::span(off, "off", "span");
auto const root = SpanGuard::freshRoot(off, "off", "fresh_root");
ScopedSpanGuard const scoped(off, "off", "scoped");
auto const scopedRoot = ScopedSpanGuard::freshRoot(off, "off", "scoped_fresh_root");
EXPECT_FALSE(static_cast<bool>(plain));
EXPECT_FALSE(static_cast<bool>(root));
EXPECT_FALSE(static_cast<bool>(scoped));
EXPECT_FALSE(static_cast<bool>(scopedRoot));
auto const control = SpanGuard::span(on, "control", "span");
EXPECT_TRUE(static_cast<bool>(control));
}
// Only the control span reached the exporter.
auto const spans = spanData()->GetSpans();
ASSERT_EQ(spans.size(), 1u);
EXPECT_EQ(nameOf(*spans[0]), "control.span");
}
}
/**
* Build null ScopedSpanGuards in cat and call the ScopedSpanGuard methods
* below on them.
*
* Every guard a method returns must be null, and the context invalid. The
* caller first makes guards in cat null, by removing the Telemetry or by
* turning cat off.
*
* @param cat Category to build the guards in.
*/
void
callMethodsOnNullScopedGuard(TraceCategory cat)
{
{
ScopedSpanGuard guard(cat, "null", "scoped");
ASSERT_FALSE(static_cast<bool>(guard));
char const* const noText = nullptr;
guard.setAttribute("string_key", std::string_view{"value"});
guard.setAttribute("c_string_key", "value");
guard.setAttribute("null_c_string_key", noText);
guard.setAttribute("int_key", std::int64_t{42});
guard.setAttribute("double_key", 0.5);
guard.setAttribute("bool_key", true);
guard.addEvent("event");
guard.setOk();
guard.setError("error");
guard.recordException(std::runtime_error("error"));
auto const ctx = guard.spanContext();
EXPECT_FALSE(ctx.isValid());
{
auto const child = guard.childSpan("null.child");
auto const linked = guard.linkedSpan("null.linked");
auto const ctxChild = ScopedSpanGuard::childSpan("null.ctx_child", ctx);
auto const ctxLinked = ScopedSpanGuard::linkedSpan("null.ctx_linked", ctx);
EXPECT_FALSE(static_cast<bool>(child));
EXPECT_FALSE(static_cast<bool>(linked));
EXPECT_FALSE(static_cast<bool>(ctxChild));
EXPECT_FALSE(static_cast<bool>(ctxLinked));
}
guard.discard();
EXPECT_FALSE(static_cast<bool>(guard));
}
// The handoff uses up its guard, so it gets a guard of its own.
ScopedSpanGuard handoff(cat, "null", "handoff");
ASSERT_FALSE(static_cast<bool>(handoff));
SpanGuard const bare = std::move(handoff);
EXPECT_FALSE(static_cast<bool>(bare));
}
// The helper's calls on a null ScopedSpanGuard are safe when no Telemetry is
// installed. Nothing can export here, so the check is that no call crashes and
// none hands back a live guard or a valid context.
TEST_F(SpanGuardScopeTest, null_scoped_guard_methods_are_safe_when_telemetry_absent)
{
Telemetry::setInstance(nullptr);
callMethodsOnNullScopedGuard(TraceCategory::Rpc);
}
// The same calls with telemetry live and only the guard's category off. A span
// in a category that is on still exports, and it is the only span exported.
TEST_F(SpanGuardScopeTest, null_scoped_guard_methods_are_safe_when_category_disabled)
{
telemetry_->setDisabledCategories({TraceCategory::Rpc});
callMethodsOnNullScopedGuard(TraceCategory::Rpc);
{
auto const control = SpanGuard::span(TraceCategory::Ledger, "control", "span");
ASSERT_TRUE(static_cast<bool>(control));
}
auto const spans = spanData()->GetSpans();
ASSERT_EQ(spans.size(), 1u);
EXPECT_EQ(nameOf(*spans[0]), "control.span");
}
// SpanGuard::activate() makes an already-owned span the ambient context for the
// activation's lifetime WITHOUT owning or ending it: before activate() the span
// is not ambient, during it the span is the current context, and after it the
@@ -6,6 +6,7 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <array>
#include <chrono>
#include <cstdint>
#include <filesystem>
@@ -148,6 +149,77 @@ constexpr char const* maxQueueSize = "max_queue_size";
constexpr char const* consensusTraceStrategy = "consensus_trace_strategy";
} // namespace key
using Setup = telemetry::Telemetry::Setup;
/**
* An on/off key of the [telemetry] section and the Setup field it sets.
*
* @code
* name=<text> --> setup.*field holds the parsed value
* name absent --> setup.*field == absentValue
* @endcode
*
* absentValue is the default that cfg/xrpld-example.cfg documents.
*/
struct FlagKey
{
/**
* Key name, as written in the config file.
*/
char const* name;
/**
* Setup field the key sets.
*/
bool Setup::* field;
/**
* Value the field takes when the key is absent.
*/
bool absentValue;
};
/**
* The on/off keys the parser reads, with the field each one sets.
*
* The names repeat the parser's own constants, which are private to
* TelemetryConfig.cpp. With a misspelled name, at least one accepting case
* fails, and so does every rejecting case.
*
* @note The rows pair keys with fields by hand. A new on/off key needs its
* own row, or these tests do not cover it.
*/
constexpr auto kFlagKeys = std::to_array<FlagKey>({
{.name = "enabled", .field = &Setup::enabled, .absentValue = false},
{.name = "use_tls", .field = &Setup::useTls, .absentValue = false},
{.name = "trace_transactions", .field = &Setup::traceTransactions, .absentValue = true},
{.name = "trace_consensus", .field = &Setup::traceConsensus, .absentValue = true},
{.name = "trace_rpc", .field = &Setup::traceRpc, .absentValue = true},
{.name = "trace_peer", .field = &Setup::tracePeer, .absentValue = true},
{.name = "trace_ledger", .field = &Setup::traceLedger, .absentValue = true},
});
/**
* Values an on/off key accepts, each with the bool it means.
*
* TRUE and False check that case is ignored.
*/
constexpr auto kAcceptedFlagValues = std::to_array<std::pair<char const*, bool>>({
{"0", false},
{"1", true},
{"true", true},
{"false", false},
{"TRUE", true},
{"False", false},
});
/**
* Values an on/off key rejects.
*
* They are other numbers and other words for true. None may turn a key on.
*/
constexpr auto kRejectedFlagValues = std::to_array<char const*>({"yes", "2", "-1", "on"});
/**
* The upper bound quoted in the expected messages below.
*
@@ -886,3 +958,48 @@ TEST(TelemetryConfig, null_telemetry_factory)
tel->start();
tel->stop();
}
TEST(TelemetryConfig, on_off_keys_accept_0_1_true_and_false_ignoring_case)
{
for (auto const& flag : kFlagKeys)
{
for (auto const& [text, expected] : kAcceptedFlagValues)
{
SCOPED_TRACE(std::string(flag.name) + "=" + text);
auto const setup = parseBatch({{flag.name, text}});
// Only the key that was set may leave its default, so a key
// wired to the wrong field fails here.
for (auto const& other : kFlagKeys)
{
SCOPED_TRACE(other.name);
bool const want = other.field == flag.field ? expected : other.absentValue;
EXPECT_EQ(setup.*other.field, want);
}
}
}
}
TEST(TelemetryConfig, on_off_keys_reject_other_values_naming_the_key)
{
for (auto const& flag : kFlagKeys)
{
std::string const message = std::string("Invalid value '") + flag.name +
"' in [telemetry]: must be 0, 1, true or false.";
for (auto const* text : kRejectedFlagValues)
{
SCOPED_TRACE(std::string(flag.name) + "=" + text);
EXPECT_EQ(batchRejection({{flag.name, text}}), message);
}
}
}
TEST(TelemetryConfig, absent_on_off_keys_take_their_defaults)
{
auto const setup = parseBatch({});
for (auto const& flag : kFlagKeys)
{
SCOPED_TRACE(flag.name);
EXPECT_EQ(setup.*flag.field, flag.absentValue);
}
}
+9 -8
View File
@@ -529,20 +529,21 @@ std::shared_ptr<telemetry::SpanGuard>
RCLConsensus::Adaptor::makeAcceptSpan(Result const& result)
{
// The whole body is telemetry: the guard, its attributes and the captured
// context serve the accept span only. With telemetry compiled out the handle
// stays empty, so accepting a ledger does not allocate a control block for a
// span that can never record. doAccept only hands the handle to
// context serve the accept span only. The handle is allocated only for a
// live span, so accepting a ledger allocates nothing for it when telemetry
// is compiled out or disabled. doAccept only hands the handle to
// activateIfLive(), which tests it, so an empty handle is safe on both the
// sync (onForceAccept) and async (onAccept) paths.
#ifdef XRPL_ENABLE_TELEMETRY
namespace cs = telemetry::consensus::span;
auto span = std::make_shared<telemetry::SpanGuard>(
telemetry::SpanGuard::childSpan(cs::accept, roundSpanContext_));
std::shared_ptr<telemetry::SpanGuard> span;
if (auto guard = telemetry::SpanGuard::childSpan(cs::accept, roundSpanContext_))
span = std::make_shared<telemetry::SpanGuard>(std::move(guard));
// Every attribute below exists only for the span, so the whole block —
// attributes and the context capture — is guarded on the span being live.
if (*span)
if (span && *span)
{
span->setAttribute(cs::attr::proposers, static_cast<int64_t>(result.proposers));
span->setAttribute(
@@ -1427,8 +1428,8 @@ RCLConsensus::Adaptor::createValidationSpan()
// Prefer linking to the accept span (matches the design diagram and
// the "validation follows acceptance" causal model). Fall back to the
// round span only if the accept context isn't yet captured (e.g.
// tracing started after onAccept, or makeAcceptSpan returned a null
// guard).
// tracing started after onAccept, or makeAcceptSpan returned an empty
// handle).
if (acceptSpanContext_.isValid())
{
return telemetry::SpanGuard::linkedSpan(cs::validationSend, acceptSpanContext_);
+4
View File
@@ -498,6 +498,10 @@ class RCLConsensus
* Create a consensus.accept span as a child of the round span.
* Returned via shared_ptr so it can be captured into the
* jtACCEPT lambda and live until doAccept completes.
*
* @param result The consensus result the span describes.
* @return A handle to the span, or an empty handle when the span is
* not live. Nothing is allocated in that case.
*/
std::shared_ptr<telemetry::SpanGuard>
makeAcceptSpan(Result const& result);
+29 -4
View File
@@ -513,6 +513,31 @@ public:
add(ledgerCleaner_.get());
}
/**
* Stop telemetry before the members are destroyed.
*
* setup() starts telemetry and run() stops it. This covers the paths that
* never reach run(): every `return false` in setup() after the start, and
* a caller that gives up after setup() succeeds. Without it the global
* Telemetry instance would point at a destroyed object. It acts only while
* this application's instance is still published, so a normal shutdown,
* where run() already stopped it, does nothing here.
*/
~ApplicationImp() override
{
// A destructor is implicitly noexcept, so a failure to stop must not
// escape and terminate the process.
try
{
if (telemetry::Telemetry::getInstance() == telemetry_.get())
telemetry_->stop();
}
catch (std::exception const& e)
{
JLOG(journal_.error()) << "Error stopping telemetry: " << e.what();
}
}
//--------------------------------------------------------------------------
bool
@@ -1695,10 +1720,10 @@ ApplicationImp::run()
ledgerCleaner_->stop();
nodeStore_->stop();
perfLog_->stop();
// Telemetry must stop last among trace-producing components.
// serverHandler_, overlay_, and jobQueue_ are already stopped above,
// so no threads should be calling startSpan() at this point.
// See TODO in TelemetryImpl::stop() re: thread-safety of sdkProvider_.
// Telemetry must stop last among trace-producing components: a span that
// ends after stop() is dropped, not exported. serverHandler_, overlay_,
// and jobQueue_ are already stopped above, so no threads should be
// calling startSpan() at this point.
telemetry_->stop();
JLOG(journal_.info()) << "Done.";
+14
View File
@@ -28,6 +28,10 @@ namespace perf {
class PerfLog;
} // namespace perf
namespace telemetry {
class Telemetry;
} // namespace telemetry
// VFALCO TODO Fix forward declares required for header dependency loops
class AmendmentTable;
@@ -166,6 +170,16 @@ public:
*/
[[nodiscard]] virtual size_t
getNumberOfThreads() const = 0;
/**
* Returns the telemetry service built from the [telemetry] config section.
*
* @return The application's telemetry instance. The reference stays valid
* for the life of the application, also when telemetry is disabled or
* compiled out.
*/
virtual telemetry::Telemetry&
getTelemetry() = 0;
};
std::unique_ptr<Application>
+35 -21
View File
@@ -188,16 +188,29 @@ class NetworkOPsImp final : public NetworkOPs
FailHard const failType;
bool applied = false;
TER result;
std::shared_ptr<telemetry::SpanGuard> span; ///< Keeps the tx.process
///< span alive until the
///< batch processes this.
/**
* The tx.process span, kept alive until the batch applies this
* transaction. Held by value, so the holder costs no heap allocation.
* Empty for an entry queued without one, such as a held transaction.
*/
std::optional<telemetry::SpanGuard> span;
/**
* Record one transaction for the next batch.
*
* @param t Transaction to apply.
* @param a Whether it came from a privileged source (TapUnlimited).
* @param l Whether a local client submitted it.
* @param f fail_hard setting from submission. Must be FailHard::No
* unless l is true.
* @param s tx.process span to keep alive until the batch applies it.
*/
TransactionStatus(
std::shared_ptr<Transaction> t,
bool a,
bool l,
FailHard f,
std::shared_ptr<telemetry::SpanGuard> s = nullptr)
std::optional<telemetry::SpanGuard> s = std::nullopt)
: transaction(std::move(t)), admin(a), local(l), failType(f), span(std::move(s))
{
XRPL_ASSERT(
@@ -414,7 +427,7 @@ public:
std::shared_ptr<Transaction> transaction,
bool bUnlimited,
FailHard failType,
std::shared_ptr<telemetry::SpanGuard> span = nullptr);
std::optional<telemetry::SpanGuard> span = std::nullopt);
/**
* For transactions not submitted by a locally connected client, fire and
@@ -424,13 +437,15 @@ public:
* @param transaction Transaction object
* @param bUnlimited Whether a privileged client connection submitted it.
* @param failType fail_hard setting from transaction submission.
* @param span Optional tx.process span to keep alive across the
* batch boundary so its context propagates to peers.
*/
void
doTransactionAsync(
std::shared_ptr<Transaction> transaction,
bool bUnlimited,
FailHard failtype,
std::shared_ptr<telemetry::SpanGuard> span = nullptr);
std::optional<telemetry::SpanGuard> span = std::nullopt);
private:
bool
@@ -1532,20 +1547,17 @@ NetworkOPsImp::processTransaction(
using namespace telemetry;
// SpanGuard is thread-free (holds no Scope), so it is safe to store here
// and end on the batch worker thread that later applies this transaction —
// no detach step is needed.
// Left null when telemetry is compiled out: there is no span to own, so
// nothing is allocated for one. The transaction pipeline already accepts a
// null span -- both doTransaction* overloads default it to nullptr -- and
// every use tests it. Without this the make_shared allocated once per
// submitted and relayed transaction to hold an empty object.
std::shared_ptr<SpanGuard> span;
// no detach step is needed. It is held by value, so the holder costs no
// heap allocation. Empty when telemetry is compiled out; both
// doTransaction* overloads default it to empty, and every use tests it.
std::optional<SpanGuard> span;
#ifdef XRPL_ENABLE_TELEMETRY
span = std::make_shared<SpanGuard>(txProcessSpan(transaction->getID()));
span.emplace(txProcessSpan(transaction->getID()));
#endif
// Guarded on the span being live because these values are not free and this
// runs for every submitted and relayed transaction: the hash string
// allocates, and the open-ledger index takes the ledger master's lock. With
// telemetry compiled out the span is null; with it compiled in the block is
// allocates, and the open-ledger index takes the open ledger's lock. With
// telemetry compiled out the span is empty; with it compiled in the block is
// skipped when telemetry is disabled at runtime or the transaction category
// is off.
if (span && *span)
@@ -1612,7 +1624,7 @@ NetworkOPsImp::doTransactionAsync(
std::shared_ptr<Transaction> transaction,
bool bUnlimited,
FailHard failType,
std::shared_ptr<telemetry::SpanGuard> span)
std::optional<telemetry::SpanGuard> span)
{
std::scoped_lock const lock(mutex_);
@@ -1636,7 +1648,7 @@ NetworkOPsImp::doTransactionSync(
std::shared_ptr<Transaction> transaction,
bool bUnlimited,
FailHard failType,
std::shared_ptr<telemetry::SpanGuard> span)
std::optional<telemetry::SpanGuard> span)
{
std::unique_lock<std::mutex> lock(mutex_);
@@ -1791,8 +1803,10 @@ NetworkOPsImp::apply(std::unique_lock<std::mutex>& batchLock)
// Parent the txq.enqueue span to this tx's tx.process span
// via an explicit captured context (the parent is explicit,
// not ambient-inherited). Null on the open-ledger rebuild
// path, where no tx.process span exists.
// not ambient-inherited). No context is passed when the
// entry has no live tx.process span: held transactions,
// queued by processTransactionSet and by the submitHeld
// push below, carry none.
auto const txProcessCtx =
(e.span && *e.span) ? e.span->spanContext() : telemetry::SpanContext{};
auto const result = registry_.get().getTxQ().apply(
@@ -1817,7 +1831,7 @@ NetworkOPsImp::apply(std::unique_lock<std::mutex>& batchLock)
validatedLedgerIndex = l->header().seq;
auto newOL = registry_.get().getOpenLedger().current();
for (TransactionStatus const& e : transactions)
for (TransactionStatus& e : transactions)
{
// Make this transaction's span ambient for the duration of its
// apply so the per-tx log lines below carry its trace_id.
+9
View File
@@ -325,6 +325,15 @@ public:
* Add a new transaction to the open ledger, hold it in the queue,
* or reject it.
*
* @param app Application the transaction is applied for.
* @param view View to apply to: the open ledger, or a copy for a dry run.
* @param tx Transaction to apply.
* @param flags Apply flags, such as TapUnlimited for an admin submit.
* @param j Journal for this call's log lines.
* @param parentCtx Context of the caller's tx.process span, used as the
* parent of the txq.enqueue span. Null, or an invalid context,
* makes that span inherit the span active on this thread, or be a
* root if there is none.
* @return A pair with the `TER` and a `bool` indicating
* whether or not the transaction was applied to
* the open ledger. If the transaction is queued,
+34 -27
View File
@@ -8,6 +8,7 @@
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/contract.h>
#include <xrpl/basics/mulDiv.h>
#include <xrpl/basics/scope.h>
#include <xrpl/beast/utility/Zero.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/config/BasicConfig.h>
@@ -50,6 +51,7 @@
#include <optional>
#include <stdexcept>
#include <string>
#include <string_view>
#include <tuple>
#include <utility>
#include <vector>
@@ -781,22 +783,29 @@ TxQ::apply(
// is ambient, and this ScopedSpanGuard's scope is RAII-bounded to this fully
// synchronous call (no coroutine yield), so no unrelated parent leaks in and
// its scope cannot leak out onto a reused worker.
// On the open-ledger rebuild path parentCtx is null, so the span nests under
// the ambient span: consensus.accept.apply inside doAccept, a root on the
// switchLastClosedLedger jump. current_ledger_seq below ties it to a ledger.
// A lambda (not a ternary) picks the factory: ScopedSpanGuard's move ctor is
// deleted, so guaranteed copy elision on each return is the only way to
// construct it conditionally.
// With no context it inherits the span active on the calling thread, or is
// a root if there is none. doAccept re-applies local transactions under
// consensus.accept.apply; an RPC thread (simulate, a local submit that runs
// the batch, standalone ledger_accept) lends its rpc.command span; the batch
// job, the consensus timer and the accept job after doAccept have none.
// current_ledger_seq below ties a root to its ledger.
// A lambda picks the factory. ScopedSpanGuard cannot move, so each branch
// returns a prvalue that initializes span in place.
auto span = [&]() -> ScopedSpanGuard {
if (parentCtx && parentCtx->isValid())
return ScopedSpanGuard::childSpan(txq_span::enqueue, *parentCtx);
return ScopedSpanGuard(
TraceCategory::Transactions, txq_span::prefix::txq, txq_span::op::enqueue);
}();
// txq_status is written once, when this function exits, not as a default
// that a later exit overwrites. Returns that set no outcome report rejected.
std::string_view outcome = txq_span::val::rejected;
ScopeExit const writeOutcome(
[&span, &outcome]() noexcept { span.setAttribute(txq_span::attr::txqStatus, outcome); });
// Guarded on the span being recorded: this runs for every transaction and
// again for each one replayed on an open-ledger rebuild, and the two hash
// strings each allocate. The compiled-out guard's operator bool() is a
// literal false, so the block disappears in that build.
// again for each local transaction doAccept replays into the new open
// ledger, and the two hash strings each allocate. The compiled-out guard's
// operator bool() is a literal false, so the block disappears in that build.
if (span)
{
span.setAttribute(txq_span::attr::txHash, to_string(tx->getTransactionID()).c_str());
@@ -808,9 +817,6 @@ TxQ::apply(
span.setAttribute(txq_span::attr::currentLedgerSeq, static_cast<std::int64_t>(view.seq()));
span.setAttribute(
txq_span::attr::currentLedgerHash, to_string(view.header().parentHash).c_str());
// Default outcome; overridden below on the direct-apply and queued
// paths. Every other early return leaves the tx rejected from the queue.
span.setAttribute(txq_span::attr::txqStatus, txq_span::val::rejected);
}
// See if the transaction is valid, properly formed,
@@ -825,19 +831,17 @@ TxQ::apply(
if (auto directApplied = tryDirectApply(app, view, tx, flags, j))
{
// A result comes back even when the apply failed, so branch on the outcome.
// transToken() builds a string, so the whole block is guarded.
if (span)
if (directApplied->applied)
{
span.setAttribute(txq_span::attr::terCode, transToken(directApplied->ter).c_str());
if (directApplied->applied)
{
span.setAttribute(txq_span::attr::txqStatus, txq_span::val::appliedDirect);
}
else
{
span.setAttribute(txq_span::attr::txqStatus, txq_span::val::failed);
}
outcome = txq_span::val::appliedDirect;
}
else
{
outcome = txq_span::val::failed;
}
// transToken() builds a string, so it runs only for a live span.
if (span)
span.setAttribute(txq_span::attr::terCode, transToken(directApplied->ter).c_str());
return *directApplied;
}
@@ -1310,7 +1314,7 @@ TxQ::apply(
/* Can't erase (*replacedTxIter) here because success
implies that it has already been deleted.
*/
span.setAttribute(txq_span::attr::txqStatus, txq_span::val::applied);
outcome = txq_span::val::applied;
return result;
}
}
@@ -1426,7 +1430,7 @@ TxQ::apply(
<< " to queue."
<< " Flags: " << flags;
span.setAttribute(txq_span::attr::txqStatus, txq_span::val::queued);
outcome = txq_span::val::queued;
return {terQUEUED, false};
}
@@ -1802,8 +1806,6 @@ TxQ::tryDirectApply(
beast::Journal j)
{
using namespace telemetry;
[[maybe_unused]] ScopedSpanGuard const span(
TraceCategory::Transactions, txq_span::prefix::txq, txq_span::op::applyDirect);
auto const account = (*tx)[sfAccount];
auto const sleAccount = view.read(keylet::account(account));
@@ -1838,6 +1840,11 @@ TxQ::tryDirectApply(
if (feeLevelPaid >= requiredFeeLevel)
{
// Opened here so the span covers a real direct apply, not the checks
// above that return early.
[[maybe_unused]] ScopedSpanGuard const applyDirectSpan(
TraceCategory::Transactions, txq_span::prefix::txq, txq_span::op::applyDirect);
// Attempt to apply the transaction directly.
auto const transactionID = tx->getTransactionID();
JLOG(j_.trace()) << "Applying transaction " << transactionID << " to open ledger.";
+29 -27
View File
@@ -1394,20 +1394,19 @@ PeerImp::handleTransaction(
//
// SpanGuard is thread-free (holds no Scope), so it is safe to hand to
// a job-queue worker and end on that thread — no detach step is needed.
// Left null when telemetry is compiled out: there is no span to own, so
// nothing is allocated for one. Every use below tests it, the job
// capture and activateIfLive() accept a null handle, and the transaction
// pipeline already takes a null span by default.
// The job closure must be copyable, because JobQueue stores it in a
// std::function, so the span sits behind a shared_ptr, not in an
// optional. The shared_ptr is made only for a live span, so a node with
// tracing off allocates nothing for it. Every use below accepts an empty
// one.
std::shared_ptr<SpanGuard> span;
#ifdef XRPL_ENABLE_TELEMETRY
span = std::make_shared<SpanGuard>(txReceiveSpan(txID, *m));
if (auto guard = txReceiveSpan(txID, *m))
span = std::make_shared<SpanGuard>(std::move(guard));
#endif
// Guarded on the span being live because these values are not free: the
// hash string allocates, and the open-ledger index takes the ledger
// master's lock. With telemetry compiled out the span is null; with it
// compiled in the block is skipped when telemetry is disabled at runtime
// or the transaction category is off.
if (span && *span)
// Guarded because these values are not free: the hash string allocates,
// and the open-ledger index takes the open ledger's lock.
if (span)
{
span->setAttribute(tx_span::attr::txHash, to_string(txID).c_str());
span->setAttribute(tx_span::attr::peerId, static_cast<int64_t>(id_));
@@ -1422,8 +1421,8 @@ PeerImp::handleTransaction(
if (auto const version = getVersion(); !version.empty())
span->setAttribute(tx_span::attr::peerVersion, version.c_str());
}
// Note: txStatus is set once at each exit path below (not as a default
// here) to avoid OTel SDK attribute duplication.
// tx_status is set once, in whichever of the three branches below runs.
// It has no default here, so each span writes the key once.
JLOG(pJournal_.debug()) << "Got tx " << txID;
@@ -1462,6 +1461,8 @@ PeerImp::handleTransaction(
}
else
{
if (span)
span->setAttribute(tx_span::attr::txStatus, tx_span::val::queuedForCheck);
app_.getJobQueue().addJob(
JtTransaction,
"RcvCheckTx",
@@ -2073,13 +2074,14 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
// Create a receive span that links to the sender's trace context
// (if propagated). shared_ptr keeps it alive across the job boundary.
// The receive span is a thread-free SpanGuard handed to the job worker;
// no scope to strip. The handle stays empty when telemetry is compiled
// out, so nothing is allocated on a path every inbound proposal takes.
// The job body only carries the handle to hold the span alive, so an
// empty handle is safe there.
// no scope to strip. The handle is allocated only for a live span, so an
// inbound proposal allocates nothing for it when telemetry is compiled out
// or disabled. The job body only carries the handle to hold the span
// alive, so an empty handle is safe there.
std::shared_ptr<telemetry::SpanGuard> proposalSpan;
#ifdef XRPL_ENABLE_TELEMETRY
proposalSpan = std::make_shared<telemetry::SpanGuard>(telemetry::proposalReceiveSpan(set));
if (auto guard = telemetry::proposalReceiveSpan(set))
proposalSpan = std::make_shared<telemetry::SpanGuard>(std::move(guard));
#endif
// Every attribute below exists only for the proposalSpan, so the block is guarded
// on the proposalSpan being live. Unguarded, each inbound proposal — trusted or
@@ -2684,13 +2686,14 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
// Create a receive span that links to the sender's trace context
// (if propagated). shared_ptr keeps it alive across the job boundary.
// The receive span is a thread-free SpanGuard handed to the job worker;
// no scope to strip. The handle stays empty when telemetry is compiled
// out, so nothing is allocated on a path every inbound validation
// takes. The job body only carries the handle to hold the span alive,
// so an empty handle is safe there.
// no scope to strip. The handle is allocated only for a live span, so
// an inbound validation allocates nothing for it when telemetry is
// compiled out or disabled. The job body only carries the handle to hold
// the span alive, so an empty handle is safe there.
std::shared_ptr<telemetry::SpanGuard> span;
#ifdef XRPL_ENABLE_TELEMETRY
span = std::make_shared<telemetry::SpanGuard>(telemetry::validationReceiveSpan(*m));
if (auto guard = telemetry::validationReceiveSpan(*m))
span = std::make_shared<telemetry::SpanGuard>(std::move(guard));
#endif
// Every attribute below exists only for the span, so the block is
// guarded on the span being live. Unguarded, each inbound validation
@@ -2712,10 +2715,9 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
static_cast<int64_t>(val->getSignTime().time_since_epoch().count()));
}
// validation_receive_status is set once on each exit below, not as a default
// here, to avoid OTel SDK attribute duplication. It is what separates
// the microsecond drop paths from the queued path, which also covers
// job wait and checkValidation.
// Each branch below sets validation_receive_status once. It separates
// the microsecond drop paths from the queued path, whose span also
// covers the job wait and checkValidation.
if (!isTrusted && (tracking_.load() == Tracking::Diverged))
{
if (span && *span)
+12 -3
View File
@@ -603,9 +603,14 @@ PathRequest::findPaths(
// 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.
//
// Internal, not Server: this runs under pathfind.compute.
using namespace telemetry;
auto span = SpanGuard::span(
TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::discover);
TraceCategory::Rpc,
pathfind_span::prefix::pathfind,
pathfind_span::op::discover,
SpanRole::Internal);
span.setAttribute(pathfind_span::attr::searchLevel, static_cast<int64_t>(level));
span.setAttribute(
pathfind_span::attr::numSourceAssets, static_cast<int64_t>(sourceAssets.size()));
@@ -764,9 +769,13 @@ PathRequest::doUpdate(
using namespace std::chrono;
using namespace telemetry;
// Scoped so pathfind.discover (created synchronously below via findPaths)
// nests under it. doUpdate does not yield, so scoping is safe.
// nests under it. doUpdate does not yield, so scoping is safe. Internal, not
// Server: this runs under pathfind.request or pathfind.update_all.
auto span = ScopedSpanGuard(
TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::compute);
TraceCategory::Rpc,
pathfind_span::prefix::pathfind,
pathfind_span::op::compute,
SpanRole::Internal);
// Guarded on the span being live because setAttribute's arguments are
// evaluated whatever the build, and doUpdate is hot: PathRequestManager
// calls it once per active path_find subscription on every ledger close, so
+5 -1
View File
@@ -100,8 +100,12 @@ PathRequestManager::updateAll(std::shared_ptr<ReadView const> const& inLedger)
std::optional<ScopedSpanGuard> span;
if (!requests.empty())
{
// Internal, not Server: a background job runs this, not a client request.
span.emplace(
TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::updateAll);
TraceCategory::Rpc,
pathfind_span::prefix::pathfind,
pathfind_span::op::updateAll,
SpanRole::Internal);
span->setAttribute(pathfind_span::attr::ledgerIndex, static_cast<int64_t>(inLedger->seq()));
span->setAttribute(pathfind_span::attr::numRequests, static_cast<int64_t>(requests.size()));
}
+5 -3
View File
@@ -106,8 +106,9 @@
* - Per-command attributes: name, API version, rpc_role, rpc_status
*
* Known gaps (not yet instrumented):
* - Early validation errors in processRequest() before rpc.process
* span (malformed JSON, auth failures, oversized requests)
* - The port, authorization and shutdown checks in onRequest(), the
* shutdown close in onWSMessage() and the non-WS upgrade refusal in
* onHandoff(), which reply before any span exists
* - Subscription push notifications (server-initiated, not RPC)
*/
@@ -188,7 +189,8 @@ inline constexpr auto admin = makeStr("admin");
inline constexpr auto user = makeStr("user");
inline constexpr auto unknownCommand = makeStr("unknown");
/**
* "invalid_json" — WS message parse failure or oversize.
* "invalid_json" — request parse failure or oversize. Span error text on
* rpc.ws_message and rpc.process.
*/
inline constexpr auto invalidJson = makeStr("invalid_json");
} // namespace val
+13 -3
View File
@@ -70,6 +70,7 @@
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <ostream>
#include <string>
#include <string_view>
@@ -727,10 +728,14 @@ ServerHandler::processRequest(
// Marks the span as failed before sending an error reply, so the
// early-return validation paths below are not later seen as successful
// (the span would otherwise end UNSET, invisible to {status.code=error}).
auto httpReplyError = [&](int status, std::string const& message) {
// The span's error description is the reply text, unless spanDescription
// is given.
auto httpReplyError = [&](int status,
std::string const& message,
std::optional<std::string_view> spanDescription = std::nullopt) {
spanHadError = true;
span.setAttribute(rpc_span::attr::rpcStatus, rpc_span::val::error);
span.setError(message);
span.setError(spanDescription.value_or(message));
httpReply(status, message, output, rpcJ);
};
@@ -740,7 +745,12 @@ ServerHandler::processRequest(
if ((request.size() > rpc::tuning::kMaxRequestSize) || !reader.parse(request, jsonOrig) ||
!jsonOrig || !jsonOrig.isObject())
{
httpReplyError(400, "Unable to parse request: " + reader.getFormattedErrorMessages());
// The parser's messages can quote parts of the request, so the
// span gets fixed text. The reply keeps the details.
httpReplyError(
400,
"Unable to parse request: " + reader.getFormattedErrorMessages(),
rpc_span::val::invalidJson);
return;
}
}
@@ -23,8 +23,12 @@ doPathFind(rpc::JsonContext& context)
using namespace telemetry;
// Scoped so pathfind.compute/discover (created synchronously below on this
// thread) nest under it. doPathFind does not yield, so scoping is safe.
// Internal, not Server: the inbound boundary is above rpc.command.
auto span = ScopedSpanGuard(
TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::request);
TraceCategory::Rpc,
pathfind_span::prefix::pathfind,
pathfind_span::op::request,
SpanRole::Internal);
// Guarded on the span being live because the account parse below is not
// free and runs on every path_find call otherwise. The compiled-out
// guard's operator bool() is a literal false, so the block disappears
@@ -31,9 +31,13 @@ doRipplePathFind(rpc::JsonContext& context)
// context.coro->yield() below; the coro-aware OTel context storage moves it
// with the coroutine on resume (it is never stranded on a worker's
// thread-local stack), so the scope pops on the correct store and the
// span's log lines stay trace-correlated.
// span's log lines stay trace-correlated. Internal, not Server: the inbound
// boundary is above rpc.command.
auto span = ScopedSpanGuard(
TraceCategory::Rpc, pathfind_span::prefix::pathfind, pathfind_span::op::request);
TraceCategory::Rpc,
pathfind_span::prefix::pathfind,
pathfind_span::op::request,
SpanRole::Internal);
// Guarded on the span being live because the account parse below is not
// free and runs on every ripple_path_find call otherwise. The compiled-out
// guard's operator bool() is a literal false, so the block disappears
+4
View File
@@ -114,6 +114,10 @@ inline constexpr auto droppedNoSync = makeStr("dropped_no_sync");
* is at MAX_TRANSACTIONS — backpressure on the receive side.
*/
inline constexpr auto droppedQueueFull = makeStr("dropped_queue_full");
/**
* Transaction was handed to the job queue to be checked.
*/
inline constexpr auto queuedForCheck = makeStr("queued_for_check");
} // namespace val
} // namespace xrpl::telemetry::tx_span