/** * OpenTelemetry SDK implementation of the Telemetry interface. * * 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, * trace-ID-ratio sampler, and resource attributes. * - NullTelemetryOtel: no-op fallback used when telemetry is compiled in * but disabled at runtime (enabled=0 in config). * - makeTelemetry(): factory that selects the appropriate implementation. */ #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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace xrpl::telemetry { // beast cannot include this header, so it duplicates the meter scope. Fail the // build if the copies drift: instruments would land off the views' scope. static_assert(kMeterName == beast::insight::kOTelMeterName); static_assert(kMeterVersion == beast::insight::kOTelMeterVersion); namespace { namespace trace_api = opentelemetry::trace; namespace trace_sdk = opentelemetry::sdk::trace; namespace metrics_api = opentelemetry::metrics; namespace metrics_sdk = opentelemetry::sdk::metrics; namespace otlp_http = opentelemetry::exporter::otlp; namespace resource = opentelemetry::sdk::resource; /** * SpanProcessor decorator that drops discarded spans. * * 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. * * 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. */ class FilteringSpanProcessor : public trace_sdk::SpanProcessor { 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); } }; /** * No-op implementation used when XRPL_ENABLE_TELEMETRY is defined but * setup.enabled is false at runtime. * * Lives in the anonymous namespace so there is no ODR conflict with the * NullTelemetry in NullTelemetry.cpp. */ class NullTelemetryOtel : public Telemetry { /** * Retained configuration (unused, kept for diagnostic access). */ Setup const setup_; public: explicit NullTelemetryOtel(Setup setup) : setup_(std::move(setup)) { } void start() override { Telemetry::setInstance(this); } void stop() override { Telemetry::setInstance(nullptr); } [[nodiscard]] bool isEnabled() const override { return false; } [[nodiscard]] bool shouldTraceTransactions() const override { return false; } [[nodiscard]] bool shouldTraceConsensus() const override { return false; } [[nodiscard]] bool shouldTraceRpc() const override { return false; } [[nodiscard]] bool shouldTracePeer() const override { return false; } [[nodiscard]] bool shouldTraceLedger() const override { return false; } [[nodiscard]] ConsensusTraceStrategy getConsensusTraceStrategy() const override { return setup_.consensusTraceStrategy; } [[nodiscard]] opentelemetry::nostd::shared_ptr getTracer(std::string_view) override { static auto noopTracer = opentelemetry::nostd::shared_ptr(new trace_api::NoopTracer()); return noopTracer; } [[nodiscard]] opentelemetry::nostd::shared_ptr getMeter(std::string_view name) override { // Mirrors the noop tracer above: instruments created from it are inert. return noopMeter(name); } [[nodiscard]] opentelemetry::nostd::shared_ptr startSpan(std::string_view, trace_api::SpanKind) override { return opentelemetry::nostd::shared_ptr(new trace_api::NoopSpan(nullptr)); } [[nodiscard]] opentelemetry::nostd::shared_ptr startSpan(std::string_view, opentelemetry::context::Context const&, trace_api::SpanKind) override { return opentelemetry::nostd::shared_ptr(new trace_api::NoopSpan(nullptr)); } }; /** * Full OTel SDK implementation that exports trace spans via OTLP/HTTP. * * Configures an OTLP/HTTP exporter, batch span processor, * TraceIdRatioBasedSampler, and resource attributes on start(). */ class TelemetryImpl : public Telemetry { /** * Configuration from the [telemetry] config section. * Non-const so setServiceInstanceId() and setNodeId() can update the * identity attributes before start() creates the OTel resource. */ Setup setup_; /** * Journal used for log output during start/stop. */ beast::Journal const journal_; /** * The SDK TracerProvider that owns the export pipeline. * * Held as std::shared_ptr so we can call ForceFlush() on shutdown. * Wrapped in a nostd::shared_ptr when registered as the global provider. */ std::shared_ptr sdkProvider_; /** * The SDK MeterProvider that owns the metric export pipeline. * * Symmetric with sdkProvider_ on the tracing side. Held as * std::shared_ptr so we can ForceFlush() on shutdown; wrapped in a * nostd::shared_ptr when registered as the global meter provider. */ std::shared_ptr meterProvider_; /** * Set by stop(), so a second call does nothing. */ bool stopped_{false}; public: TelemetryImpl(Setup setup, beast::Journal journal) : setup_(std::move(setup)), journal_(journal) { // Publish the MeterProvider before any subsystem is constructed; see // initMetrics(). setup_.serviceInstanceId is already resolved by the // caller, so the resource is complete. // // A failure must never stop the node starting: the global provider // stays noop and every instrument call remains valid. try { initMetrics(); } catch (std::exception const& e) { // Drop the half-built provider. initMetrics() publishes globally as // its last step, so a throw leaves the global noop; keeping our own // pointer would send getMeter() callers to a provider that nothing // else can reach. meterProvider_.reset(); JLOG(journal_.error()) << "Telemetry metrics pipeline failed to initialise, " "continuing without metrics: " << e.what(); } catch (...) { // initMetrics() reaches third-party SDK code, which may throw // something outside std::exception. Escaping a constructor on the // startup path would stop the node starting, so drop the // half-built provider exactly as the clause above does. meterProvider_.reset(); JLOG(journal_.error()) << "Telemetry metrics pipeline failed to initialise, " "continuing without metrics: unknown exception"; } } /** * Override the service instance id, for callers that learn it late. * * Affects only the tracer resource, which start() builds. The metrics * resource is built by the constructor and is immutable, so supply the id * through Setup to have it on both. * * @param id The instance id to report on spans. */ void setServiceInstanceId(std::string const& id) override { setup_.serviceInstanceId = id; } void setNodeId(std::string const& id) override { setup_.nodeId = id; } void start() override { JLOG(journal_.info()) << "Telemetry starting: traces_endpoint=" << setup_.tracesEndpoint << " metrics_endpoint=" << setup_.metricsEndpoint << " sampling=" << setup_.samplingRatio; auto exporter = otlp_http::OtlpHttpExporterFactory::Create(makeTraceExporterOptions(setup_)); // Configure batch processor trace_sdk::BatchSpanProcessorOptions processorOpts; processorOpts.max_queue_size = setup_.maxQueueSize; processorOpts.schedule_delay_millis = std::chrono::milliseconds(setup_.batchDelay); processorOpts.max_export_batch_size = setup_.batchSize; 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()). auto processor = std::make_unique(std::move(batchProcessor)); // Configure resource attributes auto resourceAttrs = makeTracerResource(); // Configure sampler. Head sampling is fixed at 1.0 (sample everything); // setup_.samplingRatio is not config-driven. Wrap the ratio sampler in a // ParentBasedSampler so spans with a remote parent honor the upstream // sampled flag — this keeps keep/drop decisions coherent for a single // distributed trace spanning multiple nodes. Volume reduction is left to // the collector's tail sampling. auto rootSampler = std::make_shared(setup_.samplingRatio); auto sampler = trace_sdk::ParentBasedSamplerFactory::Create(std::move(rootSampler)); // Create TracerProvider with a DeterministicIdGenerator. It returns a // deterministic trace_id when a PendingTraceId is active on the thread, // else a random one — letting hash-derived roots (introduced on a later // branch) become true trace roots. Dormant until such a caller exists. sdkProvider_ = trace_sdk::TracerProviderFactory::Create( std::move(processor), resourceAttrs, std::move(sampler), std::make_unique()); // main() installs the coroutine-aware runtime-context storage while the // process is single-threaded. It cannot be installed here: start() runs // from setup(), by which point the io threads read that global pointer // on every log line. // Set as global provider trace_api::Provider::SetTracerProvider( opentelemetry::nostd::shared_ptr(sdkProvider_)); // The metrics pipeline (meterProvider_) was already built and published // in the constructor via initMetrics(), before any subsystem could // create a beast::insight instrument. See initMetrics(). // Register as the global Telemetry instance so SpanGuard factory // methods can access it without callers passing a reference. Telemetry::setInstance(this); JLOG(journal_.info()) << "Telemetry started successfully"; } /** * Build the tracer resource: the process-identity attributes stamped on * every exported span. * * Called from start(), which runs after Application::setup() has injected * the node identity, so setup_.nodeId is populated by then. * * @return The resource attached to the TracerProvider. */ [[nodiscard]] resource::Resource makeTracerResource() const { return resource::Resource::Create({ {opentelemetry::semconv::service::kServiceName, setup_.serviceName}, {opentelemetry::semconv::service::kServiceVersion, setup_.serviceVersion}, {opentelemetry::semconv::service::kServiceInstanceId, setup_.serviceInstanceId}, {std::string(attr::networkId), static_cast(setup_.networkId)}, // LCOV_EXCL_LINE {std::string(attr::networkType), setup_.networkType}, // LCOV_EXCL_LINE {std::string(attr::nodeId), setup_.nodeId}, // LCOV_EXCL_LINE }); } /** * Build the metrics resource: the same attributes as the tracer resource * in start(), so metrics and traces share one identity. * * xrpl.node.id is added only when setup_.nodeId already holds a value. * setNodeId() runs after the constructor that calls this, so on the normal * startup path the attribute is left off rather than stamped blank. * * @return The resource attached to the MeterProvider. */ [[nodiscard]] resource::Resource makeMetricsResource() const { resource::ResourceAttributes attrs{ {opentelemetry::semconv::service::kServiceName, setup_.serviceName}, {opentelemetry::semconv::service::kServiceVersion, setup_.serviceVersion}, {opentelemetry::semconv::service::kServiceInstanceId, setup_.serviceInstanceId}, {std::string(attr::networkId), static_cast(setup_.networkId)}, {std::string(attr::networkType), setup_.networkType}, }; if (!setup_.nodeId.empty()) attrs[std::string(attr::nodeId)] = setup_.nodeId; return resource::Resource::Create(attrs); } /** * Build and publish the metrics pipeline (MeterProvider + periodic * reader + OTLP exporter + histogram view). * * Called from the constructor, NOT start(), so the global MeterProvider * exists before subsystems construct their beast::insight instruments * during ApplicationImp's member-init list. The metrics resource uses * setup_.serviceInstanceId from config; it is immutable once the provider * is built, so a later node-key setServiceInstanceId() does not affect it. * The same applies to setNodeId(): xrpl.node.id reaches this resource only * if setup_.nodeId is already populated when the constructor runs. */ void initMetrics() { // Configure OTLP HTTP metric exporter, honoring the same TLS // options as the trace exporter. The URL is used verbatim: metrics // have their own config key and are not derived from traces. otlp_http::OtlpHttpMetricExporterOptions metricExporterOpts; metricExporterOpts.url = setup_.metricsEndpoint; if (setup_.useTls) { metricExporterOpts.ssl_ca_cert_path = setup_.tlsCertPath; metricExporterOpts.ssl_client_cert_path = setup_.tlsClientCertPath; metricExporterOpts.ssl_client_key_path = setup_.tlsClientKeyPath; } auto metricExporter = otlp_http::OtlpHttpMetricExporterFactory::Create(metricExporterOpts); // Both come from [telemetry], which has already checked that they are // positive and that the timeout is below the interval. metrics_sdk::PeriodicExportingMetricReaderOptions readerOpts; readerOpts.export_interval_millis = setup_.metricExportInterval; readerOpts.export_timeout_millis = setup_.metricExportTimeout; auto reader = metrics_sdk::PeriodicExportingMetricReaderFactory::Create( std::move(metricExporter), readerOpts); auto resourceAttrs = makeMetricsResource(); // Create MeterProvider with the shared resource, then attach reader. meterProvider_ = metrics_sdk::MeterProviderFactory::Create( std::make_unique(), resourceAttrs); meterProvider_->AddMetricReader(std::move(reader)); // One histogram view per unit. The unit is the selector, so an // instrument gets the ladder that fits what it measures -- a byte // count no longer inherits a latency ladder. Edges come from // HistogramBuckets.h, which owns every ladder. // // Each view keeps the "*" name pattern and an EMPTY view name: a // non-empty view name would rename every matching histogram to it and // collapse them (ios_latency, rpc_size, rpc_time, pathfind_*, and all // the jobq_* pairs) into a single series. // // The meter selector MUST match the meter name used by getMeter() and // the beast OTelCollector (kMeterName = "xrpld"); otherwise a view // never applies and instruments fall back to the SDK default ladder, // whose ceiling is 10,000. auto const addUnitView = [this]( std::string const& unitCode, std::vector boundaries, std::string const& description) { auto selector = metrics_sdk::InstrumentSelectorFactory::Create( metrics_sdk::InstrumentType::kHistogram, "*", unitCode); auto meterSelector = metrics_sdk::MeterSelectorFactory::Create(std::string(kMeterName), "", ""); auto config = std::make_shared(); config->boundaries_ = std::move(boundaries); auto view = metrics_sdk::ViewFactory::Create( "", // empty name: keep each instrument's own name, only set buckets description, metrics_sdk::AggregationType::kHistogram, std::move(config)); meterProvider_->AddView(std::move(selector), std::move(meterSelector), std::move(view)); }; addUnitView( beast::insight::otelUnitCode(beast::insight::Unit::Millis), buckets::toVector(buckets::kMillisecondBuckets), "Duration buckets, 1 ms to 120 s"); addUnitView( beast::insight::otelUnitCode(beast::insight::Unit::Bytes), buckets::toVector(buckets::kByteBuckets), "Size buckets, 512 B to 1 MiB"); // Publish as the global meter provider so developers (and the beast // OTelCollector shim) reach the same pipeline. metrics_api::Provider::SetMeterProvider( opentelemetry::nostd::shared_ptr(meterProvider_)); } void stop() override { if (stopped_) return; stopped_ = true; JLOG(journal_.info()) << "Telemetry stopping"; // Unregister global instance before tearing down the pipeline, but only // if this object is the one that published it. if (Telemetry::getInstance() == this) Telemetry::setInstance(nullptr); if (sdkProvider_) { // 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. sdkProvider_.reset(); trace_api::Provider::SetTracerProvider( opentelemetry::nostd::shared_ptr( new trace_api::NoopTracerProvider())); } if (meterProvider_) { // Mirror the tracer teardown: bounded flush, drop our provider, // then restore a noop global provider. Same shutdown-order // caveat as sdkProvider_ above applies to getMeter() callers. meterProvider_->ForceFlush(std::chrono::milliseconds(5000)); meterProvider_.reset(); metrics_api::Provider::SetMeterProvider( opentelemetry::nostd::shared_ptr( new metrics_api::NoopMeterProvider())); } JLOG(journal_.info()) << "Telemetry stopped"; } [[nodiscard]] bool isEnabled() const override { return true; } [[nodiscard]] bool shouldTraceTransactions() const override { return setup_.traceTransactions; } [[nodiscard]] bool shouldTraceConsensus() const override { return setup_.traceConsensus; } [[nodiscard]] bool shouldTraceRpc() const override { return setup_.traceRpc; } [[nodiscard]] bool shouldTracePeer() const override { return setup_.tracePeer; } [[nodiscard]] bool shouldTraceLedger() const override { return setup_.traceLedger; } [[nodiscard]] ConsensusTraceStrategy getConsensusTraceStrategy() const override { return setup_.consensusTraceStrategy; } [[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)); } [[nodiscard]] opentelemetry::nostd::shared_ptr getMeter(std::string_view name = kMeterName) override { if (!meterProvider_) { return metrics_api::Provider::GetMeterProvider()->GetMeter( std::string(name), std::string(kMeterVersion)); } return meterProvider_->GetMeter(std::string(name), std::string(kMeterVersion)); } [[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); } [[nodiscard]] opentelemetry::nostd::shared_ptr startSpan( std::string_view name, 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); } }; } // namespace opentelemetry::nostd::shared_ptr noopMeter(std::string_view name) { // One provider for the process: it holds a single inert meter, so nothing // is gained by building another. static auto const kProvider = opentelemetry::nostd::shared_ptr( new metrics_api::NoopMeterProvider()); return kProvider->GetMeter(std::string(name), std::string(kMeterVersion)); } opentelemetry::exporter::otlp::OtlpHttpExporterOptions makeTraceExporterOptions(Telemetry::Setup const& setup) { otlp_http::OtlpHttpExporterOptions opts; opts.url = setup.tracesEndpoint; if (setup.useTls) { opts.ssl_ca_cert_path = setup.tlsCertPath; // Present a client cert for mutual TLS. When both paths are // empty the connection falls back to one-way (server) TLS. opts.ssl_client_cert_path = setup.tlsClientCertPath; opts.ssl_client_key_path = setup.tlsClientKeyPath; } return opts; } std::unique_ptr makeTelemetry(Telemetry::Setup const& setup, beast::Journal journal) { if (setup.enabled) { return std::make_unique(setup, journal); } return std::make_unique(setup); } } // namespace xrpl::telemetry #endif // XRPL_ENABLE_TELEMETRY