diff --git a/.codecov.yml b/.codecov.yml index f4f812ddf5..7e81079067 100644 --- a/.codecov.yml +++ b/.codecov.yml @@ -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/" diff --git a/CMakeLists.txt b/CMakeLists.txt index 04ae173b1c..598bf179d0 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -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) diff --git a/cfg/xrpld-example.cfg b/cfg/xrpld-example.cfg index 980d35479c..8e53cb3a7f 100644 --- a/cfg/xrpld-example.cfg +++ b/cfg/xrpld-example.cfg @@ -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 # diff --git a/cmake/XrplCore.cmake b/cmake/XrplCore.cmake index 6c0ab2f2b6..1a84e2c010 100644 --- a/cmake/XrplCore.cmake +++ b/cmake/XrplCore.cmake @@ -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 diff --git a/conan/profiles/default b/conan/profiles/default index c30fd14bfa..c5baa17386 100644 --- a/conan/profiles/default +++ b/conan/profiles/default @@ -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 diff --git a/conanfile.py b/conanfile.py index 5a8c2318d0..8730ea1e99 100644 --- a/conanfile.py +++ b/conanfile.py @@ -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 = ( diff --git a/docs/telemetry-runbook.md b/docs/telemetry-runbook.md index 54d4d7d9f6..b59f56531f 100644 --- a/docs/telemetry-runbook.md +++ b/docs/telemetry-runbook.md @@ -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 diff --git a/include/xrpl/core/ServiceRegistry.h b/include/xrpl/core/ServiceRegistry.h index cb7a9d00ad..000bdaa7fa 100644 --- a/include/xrpl/core/ServiceRegistry.h +++ b/include/xrpl/core/ServiceRegistry.h @@ -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; diff --git a/include/xrpl/telemetry/DiscardFlag.h b/include/xrpl/telemetry/DiscardFlag.h index 1bd1f668bb..cbc9bb369f 100644 --- a/include/xrpl/telemetry/DiscardFlag.h +++ b/include/xrpl/telemetry/DiscardFlag.h @@ -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 { diff --git a/include/xrpl/telemetry/FilteringSpanProcessor.h b/include/xrpl/telemetry/FilteringSpanProcessor.h new file mode 100644 index 0000000000..226fe3c034 --- /dev/null +++ b/include/xrpl/telemetry/FilteringSpanProcessor.h @@ -0,0 +1,138 @@ +#pragma once + +#ifdef XRPL_ENABLE_TELEMETRY + +#include +#include +#include + +#include +#include + +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(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 delegate_; + +public: + /** + * @param delegate Processor that receives every span that is not discarded. + */ + explicit FilteringSpanProcessor( + std::unique_ptr 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 + 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&& 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 diff --git a/include/xrpl/telemetry/SpanGuard.h b/include/xrpl/telemetry/SpanGuard.h index a11a4e1408..9e95611b5f 100644 --- a/include/xrpl/telemetry/SpanGuard.h +++ b/include/xrpl/telemetry/SpanGuard.h @@ -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 (pimpl) | * +--------------------------------------------+ * | + (cat, prefix, name) [ctor] | @@ -703,14 +709,14 @@ public: * | + spanContext() / discard() | * | + operator bool() | * +--------------------------------------------+ - * | hides (pimpl) - * +------+--------------------+ - * | | + * | | hides (pimpl) + * | owns | * +----------+ +-----------------------------+ - * | SpanGuard| | optional | + * | 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 impl_; explicit ScopedSpanGuard(SpanGuard&& guard) noexcept; diff --git a/src/libxrpl/telemetry/FilteringSpanProcessor.cpp b/src/libxrpl/telemetry/FilteringSpanProcessor.cpp new file mode 100644 index 0000000000..2e3a197ee9 --- /dev/null +++ b/src/libxrpl/telemetry/FilteringSpanProcessor.cpp @@ -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 + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +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 + requires std::is_arithmetic_v + 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 + void + operator()(std::vector const& values) const noexcept + { + target.SetAttribute(key, opentelemetry::nostd::span{values.data(), values.size()}); + } + + // std::vector packs its bits and has no data(), so copy to plain bools. + void + operator()(std::vector const& values) const noexcept + { + auto const flags = std::make_unique(values.size()); + std::ranges::copy(values, flags.get()); + target.SetAttribute( + key, opentelemetry::nostd::span{flags.get(), values.size()}); + } + + // A string array is sent as views into the owned strings. + void + operator()(std::vector const& values) const noexcept + { + std::vector views; + views.reserve(values.size()); + for (auto const& value : values) + { + views.emplace_back(value.data(), value.size()); + } + target.SetAttribute( + key, + opentelemetry::nostd::span{ + 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; + + /** + * The delegate's recordable. Null only after flushAndRelease(). + */ + std::unique_ptr inner_; + + /** + * Kept attributes, in the order each key was first set. + */ + std::vector attributes_; + +public: + /** + * @param inner The delegate's recordable. Must not be null. + */ + explicit UniqueAttributeRecordable(std::unique_ptr 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 + 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 delegate) + : delegate_(std::move(delegate)) +{ +} + +std::unique_ptr +FilteringSpanProcessor::MakeRecordable() noexcept +{ + auto inner = delegate_->MakeRecordable(); + if (inner == nullptr) + { + return nullptr; + } + return std::make_unique(std::move(inner)); +} + +void +FilteringSpanProcessor::OnStart( + trace_sdk::Recordable& span, + trace_api::SpanContext const& parentContext) noexcept +{ + auto* const wrapper = dynamic_cast(&span); + delegate_->OnStart(wrapper != nullptr ? wrapper->inner() : span, parentContext); +} + +void +FilteringSpanProcessor::OnEnd(std::unique_ptr&& 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(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 diff --git a/src/libxrpl/telemetry/SpanGuard.cpp b/src/libxrpl/telemetry/SpanGuard.cpp index a7682ccfc3..e8c0fe42cf 100644 --- a/src/libxrpl/telemetry/SpanGuard.cpp +++ b/src/libxrpl/telemetry/SpanGuard.cpp @@ -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 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 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(std::move(guard))) +ScopedSpanGuard::ScopedSpanGuard(SpanGuard&& guard) noexcept : guard_(std::move(guard)) { + if (guard_) + impl_ = std::make_unique(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 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(impl_->guard); + return static_cast(guard_); } // ===== ScopedActivation ==================================================== diff --git a/src/libxrpl/telemetry/Telemetry.cpp b/src/libxrpl/telemetry/Telemetry.cpp index b4a75f0314..7ef2485e4a 100644 --- a/src/libxrpl/telemetry/Telemetry.cpp +++ b/src/libxrpl/telemetry/Telemetry.cpp @@ -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 #include #include -#include +#include #include #include @@ -32,10 +30,10 @@ #include #include #include +#include #include #include #include -#include #include #include #include @@ -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 | - * | | - * +---------------------------+ - * | 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 delegate_; - -public: - explicit FilteringSpanProcessor(std::unique_ptr delegate) - : delegate_(std::move(delegate)) - { - } - - std::unique_ptr - 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&& 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 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 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(std::move(batchProcessor)); // Configure resource attributes @@ -364,6 +303,10 @@ public: trace_api::Provider::SetTracerProvider( opentelemetry::nostd::shared_ptr(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 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( @@ -448,20 +388,17 @@ public: [[nodiscard]] opentelemetry::nostd::shared_ptr 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 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 @@ -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); } }; diff --git a/src/libxrpl/telemetry/TelemetryConfig.cpp b/src/libxrpl/telemetry/TelemetryConfig.cpp index a8bad8a97e..bdd09cab55 100644 --- a/src/libxrpl/telemetry/TelemetryConfig.cpp +++ b/src/libxrpl/telemetry/TelemetryConfig.cpp @@ -9,6 +9,7 @@ */ #include +#include #include #include @@ -138,6 +139,37 @@ readBounded( return static_cast(*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( + "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(key::enabled, 0) != 0; + setup.enabled = readFlag(section, key::enabled, false); setup.serviceName = section.valueOr(key::serviceName, dflt::serviceName); setup.serviceVersion = version; setup.serviceInstanceId = section.valueOr(key::serviceInstanceId, nodePublicKey); setup.tracesEndpoint = section.valueOr(key::tracesEndpoint, dflt::tracesEndpoint); - setup.useTls = section.valueOr(key::useTls, 0) != 0; + setup.useTls = readFlag(section, key::useTls, false); setup.tlsCertPath = section.valueOr(key::tlsCaCert, ""); setup.tlsClientCertPath = section.valueOr(key::tlsClientCert, ""); setup.tlsClientKeyPath = section.valueOr(key::tlsClientKey, ""); @@ -366,11 +398,11 @@ makeTelemetrySetup( setup.networkId = networkId; setup.networkType = networkTypeFromId(networkId); - setup.traceTransactions = section.valueOr(key::traceTransactions, 1) != 0; - setup.traceConsensus = section.valueOr(key::traceConsensus, 1) != 0; - setup.traceRpc = section.valueOr(key::traceRpc, 1) != 0; - setup.tracePeer = section.valueOr(key::tracePeer, 1) != 0; - setup.traceLedger = section.valueOr(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(key::consensusTraceStrategy, "")); diff --git a/src/tests/libxrpl/helpers/TestServiceRegistry.h b/src/tests/libxrpl/helpers/TestServiceRegistry.h index eae3180aa5..e763c8bde4 100644 --- a/src/tests/libxrpl/helpers/TestServiceRegistry.h +++ b/src/tests/libxrpl/helpers/TestServiceRegistry.h @@ -9,7 +9,6 @@ #include #include #include -#include #include @@ -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 diff --git a/src/tests/libxrpl/telemetry/FilteringSpanProcessor.cpp b/src/tests/libxrpl/telemetry/FilteringSpanProcessor.cpp new file mode 100644 index 0000000000..c828fdf7c9 --- /dev/null +++ b/src/tests/libxrpl/telemetry/FilteringSpanProcessor.cpp @@ -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 + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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 text{}; + id.ToLowerBase16(text); + return {text.data(), text.size()}; +} + +/** + * @return a span id holding these bytes. + */ +otel_trace::SpanId +spanId(std::array const& bytes) +{ + return otel_trace::SpanId{otel_nostd::span{ + bytes.data(), bytes.size()}}; +} + +/** + * @return a trace id holding these bytes. + */ +otel_trace::TraceId +traceId(std::array const& bytes) +{ + return otel_trace::TraceId{otel_nostd::span{ + 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 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 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 values) const + { + std::string text; + for (auto const& value : values) + { + text += std::format("{}\"{}\"", text.empty() ? "" : ",", toString(value)); + } + return std::format("string[]:[{}]", text); + } + + template + std::string + operator()(otel_nostd::span 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 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(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 +callsOf(otel_sdk_trace::Recordable const* recordable) +{ + auto const* recording = dynamic_cast(recordable); + if (recording == nullptr) + { + return {"not a RecordingRecordable"}; + } + return recording->calls; +} + +/** + * @return only the SetAttribute lines a recordable received. + */ +std::vector +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 made; + + /** + * Every recordable passed to OnStart(), in order. Compared by address only. + */ + std::vector started; + + /** + * Every recordable passed to OnEnd(), kept alive for inspection. + */ + std::vector> 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()>; + + /** + * @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 + 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&& 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 +makeRecording() +{ + return std::make_unique(); +} + +/** + * @return a recording delegate that writes to @p log. + */ +std::unique_ptr +recordingDelegate(DelegateLog& log, RecordingProcessor::Factory make = makeRecording) +{ + return std::make_unique(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{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{ + "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{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{"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{ + "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 flags{true, false, true}; + std::array counts{3, 4}; + std::string first{"a"}; + std::string second{"b"}; + std::array 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{flags.data(), flags.size()}); + span->SetAttribute( + "counts", otel_nostd::span{counts.data(), counts.size()}); + span->SetAttribute( + "names", otel_nostd::span{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{ + "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 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>{eventAttributes}); + span->AddLink(linked); + span->SetStatus(otel_trace::StatusCode::kError, "boom"); + span->SetDuration(std::chrono::nanoseconds{250}); + + std::vector 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(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{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(); + 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{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{}; })}; + + 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{ + std::make_unique()}; + })}; + 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(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(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{R"(SetAttribute txq_status=string:"queued")"})); +} + +} // namespace +} // namespace xrpl::telemetry + +#endif // XRPL_ENABLE_TELEMETRY diff --git a/src/tests/libxrpl/telemetry/SpanGuardFactory.cpp b/src/tests/libxrpl/telemetry/SpanGuardFactory.cpp index 88d045a37c..b2935d66ba 100644 --- a/src/tests/libxrpl/telemetry/SpanGuardFactory.cpp +++ b/src/tests/libxrpl/telemetry/SpanGuardFactory.cpp @@ -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); diff --git a/src/tests/libxrpl/telemetry/SpanGuardScope.cpp b/src/tests/libxrpl/telemetry/SpanGuardScope.cpp index c3ea539e00..5dcd395256 100644 --- a/src/tests/libxrpl/telemetry/SpanGuardScope.cpp +++ b/src/tests/libxrpl/telemetry/SpanGuardScope.cpp @@ -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 #include #include +#include +#include #include #include #include @@ -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 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 spanData_; + + /** + * Categories a test turned off. Empty by default, so every category is on. + */ + std::set 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(plain)); + EXPECT_FALSE(static_cast(root)); + EXPECT_FALSE(static_cast(scoped)); + EXPECT_FALSE(static_cast(scopedRoot)); + + auto const control = SpanGuard::span(on, "control", "span"); + EXPECT_TRUE(static_cast(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(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(child)); + EXPECT_FALSE(static_cast(linked)); + EXPECT_FALSE(static_cast(ctxChild)); + EXPECT_FALSE(static_cast(ctxLinked)); + } + + guard.discard(); + EXPECT_FALSE(static_cast(guard)); + } + + // The handoff uses up its guard, so it gets a guard of its own. + ScopedSpanGuard handoff(cat, "null", "handoff"); + ASSERT_FALSE(static_cast(handoff)); + SpanGuard const bare = std::move(handoff); + EXPECT_FALSE(static_cast(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(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 diff --git a/src/tests/libxrpl/telemetry/TelemetryConfig.cpp b/src/tests/libxrpl/telemetry/TelemetryConfig.cpp index 267fc7a24e..af8a52d630 100644 --- a/src/tests/libxrpl/telemetry/TelemetryConfig.cpp +++ b/src/tests/libxrpl/telemetry/TelemetryConfig.cpp @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -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= --> 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({ + {.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>({ + {"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({"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); + } +} diff --git a/src/xrpld/app/consensus/RCLConsensus.cpp b/src/xrpld/app/consensus/RCLConsensus.cpp index 029f911f5c..dd75bfa70a 100644 --- a/src/xrpld/app/consensus/RCLConsensus.cpp +++ b/src/xrpld/app/consensus/RCLConsensus.cpp @@ -529,20 +529,21 @@ std::shared_ptr 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::childSpan(cs::accept, roundSpanContext_)); + std::shared_ptr span; + if (auto guard = telemetry::SpanGuard::childSpan(cs::accept, roundSpanContext_)) + span = std::make_shared(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(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_); diff --git a/src/xrpld/app/consensus/RCLConsensus.h b/src/xrpld/app/consensus/RCLConsensus.h index af85f2bcc7..a4917f6108 100644 --- a/src/xrpld/app/consensus/RCLConsensus.h +++ b/src/xrpld/app/consensus/RCLConsensus.h @@ -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 makeAcceptSpan(Result const& result); diff --git a/src/xrpld/app/main/Application.cpp b/src/xrpld/app/main/Application.cpp index 01a7c0d071..c1522859ce 100644 --- a/src/xrpld/app/main/Application.cpp +++ b/src/xrpld/app/main/Application.cpp @@ -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."; diff --git a/src/xrpld/app/main/Application.h b/src/xrpld/app/main/Application.h index 225275afe4..a6b2b417c0 100644 --- a/src/xrpld/app/main/Application.h +++ b/src/xrpld/app/main/Application.h @@ -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 diff --git a/src/xrpld/app/misc/NetworkOPs.cpp b/src/xrpld/app/misc/NetworkOPs.cpp index 17fead873e..1032f0aa77 100644 --- a/src/xrpld/app/misc/NetworkOPs.cpp +++ b/src/xrpld/app/misc/NetworkOPs.cpp @@ -188,16 +188,29 @@ class NetworkOPsImp final : public NetworkOPs FailHard const failType; bool applied = false; TER result; - std::shared_ptr 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 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 t, bool a, bool l, FailHard f, - std::shared_ptr s = nullptr) + std::optional 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, bool bUnlimited, FailHard failType, - std::shared_ptr span = nullptr); + std::optional 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, bool bUnlimited, FailHard failtype, - std::shared_ptr span = nullptr); + std::optional 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 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 span; #ifdef XRPL_ENABLE_TELEMETRY - span = std::make_shared(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, bool bUnlimited, FailHard failType, - std::shared_ptr span) + std::optional span) { std::scoped_lock const lock(mutex_); @@ -1636,7 +1648,7 @@ NetworkOPsImp::doTransactionSync( std::shared_ptr transaction, bool bUnlimited, FailHard failType, - std::shared_ptr span) + std::optional span) { std::unique_lock lock(mutex_); @@ -1791,8 +1803,10 @@ NetworkOPsImp::apply(std::unique_lock& 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& 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. diff --git a/src/xrpld/app/misc/TxQ.h b/src/xrpld/app/misc/TxQ.h index 358b426332..d70a1101a1 100644 --- a/src/xrpld/app/misc/TxQ.h +++ b/src/xrpld/app/misc/TxQ.h @@ -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, diff --git a/src/xrpld/app/misc/detail/TxQ.cpp b/src/xrpld/app/misc/detail/TxQ.cpp index 2fcc4e7e55..8a51249e10 100644 --- a/src/xrpld/app/misc/detail/TxQ.cpp +++ b/src/xrpld/app/misc/detail/TxQ.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -50,6 +51,7 @@ #include #include #include +#include #include #include #include @@ -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(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."; diff --git a/src/xrpld/overlay/detail/PeerImp.cpp b/src/xrpld/overlay/detail/PeerImp.cpp index 822c2b4564..63ff7e9c36 100644 --- a/src/xrpld/overlay/detail/PeerImp.cpp +++ b/src/xrpld/overlay/detail/PeerImp.cpp @@ -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 span; #ifdef XRPL_ENABLE_TELEMETRY - span = std::make_shared(txReceiveSpan(txID, *m)); + if (auto guard = txReceiveSpan(txID, *m)) + span = std::make_shared(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(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 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 proposalSpan; #ifdef XRPL_ENABLE_TELEMETRY - proposalSpan = std::make_shared(telemetry::proposalReceiveSpan(set)); + if (auto guard = telemetry::proposalReceiveSpan(set)) + proposalSpan = std::make_shared(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 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 span; #ifdef XRPL_ENABLE_TELEMETRY - span = std::make_shared(telemetry::validationReceiveSpan(*m)); + if (auto guard = telemetry::validationReceiveSpan(*m)) + span = std::make_shared(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 const& m) static_cast(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) diff --git a/src/xrpld/rpc/detail/PathRequest.cpp b/src/xrpld/rpc/detail/PathRequest.cpp index a7fb54801c..774d696448 100644 --- a/src/xrpld/rpc/detail/PathRequest.cpp +++ b/src/xrpld/rpc/detail/PathRequest.cpp @@ -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(level)); span.setAttribute( pathfind_span::attr::numSourceAssets, static_cast(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 diff --git a/src/xrpld/rpc/detail/PathRequestManager.cpp b/src/xrpld/rpc/detail/PathRequestManager.cpp index 07345e325a..aaea064ca7 100644 --- a/src/xrpld/rpc/detail/PathRequestManager.cpp +++ b/src/xrpld/rpc/detail/PathRequestManager.cpp @@ -100,8 +100,12 @@ PathRequestManager::updateAll(std::shared_ptr const& inLedger) std::optional 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(inLedger->seq())); span->setAttribute(pathfind_span::attr::numRequests, static_cast(requests.size())); } diff --git a/src/xrpld/rpc/detail/RpcSpanNames.h b/src/xrpld/rpc/detail/RpcSpanNames.h index fd674640fc..d4b73d2dee 100644 --- a/src/xrpld/rpc/detail/RpcSpanNames.h +++ b/src/xrpld/rpc/detail/RpcSpanNames.h @@ -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 diff --git a/src/xrpld/rpc/detail/ServerHandler.cpp b/src/xrpld/rpc/detail/ServerHandler.cpp index 842193eeef..11e8322a8f 100644 --- a/src/xrpld/rpc/detail/ServerHandler.cpp +++ b/src/xrpld/rpc/detail/ServerHandler.cpp @@ -70,6 +70,7 @@ #include #include #include +#include #include #include #include @@ -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 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; } } diff --git a/src/xrpld/rpc/handlers/orderbook/PathFind.cpp b/src/xrpld/rpc/handlers/orderbook/PathFind.cpp index a4323b9e00..c9ba8e03e7 100644 --- a/src/xrpld/rpc/handlers/orderbook/PathFind.cpp +++ b/src/xrpld/rpc/handlers/orderbook/PathFind.cpp @@ -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 diff --git a/src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp b/src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp index 944b442299..ac0061be09 100644 --- a/src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp +++ b/src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp @@ -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 diff --git a/src/xrpld/telemetry/TxSpanNames.h b/src/xrpld/telemetry/TxSpanNames.h index 64049da6c9..b1585fdc61 100644 --- a/src/xrpld/telemetry/TxSpanNames.h +++ b/src/xrpld/telemetry/TxSpanNames.h @@ -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