From d088eab8e726be1ccdd7407c73c3e45f06c2d310 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Leo=20Bl=C3=B6cher?= Date: Fri, 10 Jul 2026 17:15:59 +0100 Subject: [PATCH] Split tracing metrics to allow multiple pipelines In #223 we are adding a second tracing pipeline. To avoid clashing metrics between the two, we need to add a label to distinguish them. --- foundations/src/telemetry/tracing/channel.rs | 78 +++++++++++++++----- foundations/src/telemetry/tracing/init.rs | 9 ++- foundations/src/telemetry/tracing/metrics.rs | 9 ++- 3 files changed, 71 insertions(+), 25 deletions(-) diff --git a/foundations/src/telemetry/tracing/channel.rs b/foundations/src/telemetry/tracing/channel.rs index 5e1e3c04..413a9f05 100644 --- a/foundations/src/telemetry/tracing/channel.rs +++ b/foundations/src/telemetry/tracing/channel.rs @@ -47,6 +47,17 @@ impl Receiver { } } +/// Identifies the tracing pipeline that a span channel belongs to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize)] +pub(super) enum PipelineType { + /// Regular (system) tracing pipeline. + /// + /// Encodes to an empty string for backwards compatibility with existing + /// metrics. Prometheus treats this the same as an absent label. + #[serde(rename = "")] + System, +} + /// An instrumented, multi-consumer span receiver layered on top of /// [`tokio::sync::mpsc`]. /// @@ -59,18 +70,26 @@ impl Receiver { #[derive(Clone)] pub(super) struct SharedSpanReceiver { rx: Arc>>, + + #[allow(dead_code, reason = "only used if `metrics` feature is enabled")] + metrics_label: PipelineType, } impl SharedSpanReceiver { - fn new_bounded(receiver: mpsc::Receiver) -> Self { + fn new_bounded(receiver: mpsc::Receiver, metrics_label: PipelineType) -> Self { Self { rx: Arc::new(Mutex::new(Receiver::Bounded(receiver))), + metrics_label, } } - fn new_unbounded(receiver: mpsc::UnboundedReceiver) -> Self { + fn new_unbounded( + receiver: mpsc::UnboundedReceiver, + metrics_label: PipelineType, + ) -> Self { Self { rx: Arc::new(Mutex::new(Receiver::Unbounded(receiver))), + metrics_label, } } @@ -83,7 +102,7 @@ impl SharedSpanReceiver { let res = rx.try_recv(); #[cfg(feature = "metrics")] - super::metrics::tracing::queue_size().set(rx.len() as u64); + super::metrics::tracing::queue_size(self.metrics_label).set(rx.len() as u64); res.ok() } @@ -94,7 +113,7 @@ impl SharedSpanReceiver { let rx = &mut *self.rx.lock().await; #[cfg(feature = "metrics")] - let queue_size = super::metrics::tracing::queue_size(); + let queue_size = super::metrics::tracing::queue_size(self.metrics_label); // Execute the recv_many operation. This means we are the active consumer and // are woken up if the channel length changes. @@ -133,17 +152,22 @@ impl Sender for mpsc::UnboundedSender { /// An instrumented sender for [`cf_rustracing_jaeger`] spans. #[derive(Clone)] -pub(super) struct SpanSender(S); +pub(super) struct SpanSender { + inner: S, + + #[allow(dead_code, reason = "only used if `metrics` feature is enabled")] + metrics_label: PipelineType, +} impl + Send + Sync> SpanConsumer for SpanSender { fn consume_span(&self, span: FinishedSpan) { - let _res = self.0.try_send(span); + let _res = self.inner.try_send(span); #[cfg(feature = "metrics")] { - super::metrics::tracing::spans_total().inc(); + super::metrics::tracing::spans_total(self.metrics_label).inc(); if _res.is_err() { - super::metrics::tracing::spans_dropped().inc(); + super::metrics::tracing::spans_dropped(self.metrics_label).inc(); } } } @@ -153,15 +177,32 @@ pub(super) type BoundedSpanSender = SpanSender>; pub(super) type UnboundedSpanSender = SpanSender>; /// Creates a bounded MPMC channel for [`cf_rustracing_jaeger`] spans. -pub(super) fn channel(buffer: NonZeroUsize) -> (BoundedSpanSender, SharedSpanReceiver) { +pub(super) fn channel( + buffer: NonZeroUsize, + metrics_label: PipelineType, +) -> (BoundedSpanSender, SharedSpanReceiver) { let (send, recv) = mpsc::channel(buffer.get()); - (SpanSender(send), SharedSpanReceiver::new_bounded(recv)) + ( + SpanSender { + inner: send, + metrics_label, + }, + SharedSpanReceiver::new_bounded(recv, metrics_label), + ) } /// Creates an unbounded MPMC channel for [`cf_rustracing_jaeger`] spans. -pub(super) fn unbounded_channel() -> (UnboundedSpanSender, SharedSpanReceiver) { +pub(super) fn unbounded_channel( + metrics_label: PipelineType, +) -> (UnboundedSpanSender, SharedSpanReceiver) { let (send, recv) = mpsc::unbounded_channel(); - (SpanSender(send), SharedSpanReceiver::new_unbounded(recv)) + ( + SpanSender { + inner: send, + metrics_label, + }, + SharedSpanReceiver::new_unbounded(recv, metrics_label), + ) } #[cfg(test)] @@ -174,25 +215,28 @@ mod tests { #[tokio::test] async fn test_span_metrics() { - let (send, recv) = channel(NonZeroUsize::new(3).unwrap()); + let (send, recv) = channel(NonZeroUsize::new(3).unwrap(), PipelineType::System); let tracer = Tracer::with_consumer(AllSampler, send); for _ in 0..5 { let _span = tracer.span("my span").start(); } - assert_eq!(tracing_metrics::spans_total().get(), 5); - assert_eq!(tracing_metrics::spans_dropped().get(), 2); + assert_eq!(tracing_metrics::spans_total(PipelineType::System).get(), 5); + assert_eq!( + tracing_metrics::spans_dropped(PipelineType::System).get(), + 2, + ); let mut spans = Vec::new(); let got = recv.recv_many(&mut spans, 1).await; assert_eq!(got, 1); assert_eq!(spans.len(), 1); - assert_eq!(tracing_metrics::queue_size().get(), 2); + assert_eq!(tracing_metrics::queue_size(PipelineType::System).get(), 2); let got = recv.recv_many(&mut spans, 100).await; assert_eq!(got, 2); assert_eq!(spans.len(), 3); - assert_eq!(tracing_metrics::queue_size().get(), 0); + assert_eq!(tracing_metrics::queue_size(PipelineType::System).get(), 0); } } diff --git a/foundations/src/telemetry/tracing/init.rs b/foundations/src/telemetry/tracing/init.rs index f1c4c77a..f8896d8b 100644 --- a/foundations/src/telemetry/tracing/init.rs +++ b/foundations/src/telemetry/tracing/init.rs @@ -1,4 +1,4 @@ -use super::channel::SharedSpanReceiver; +use super::channel::{PipelineType, SharedSpanReceiver}; use super::internal::{SharedSpan, Tracer}; use super::live::ActiveRoots; use super::output_jaeger_thrift_udp; @@ -77,11 +77,11 @@ pub(super) fn create_tracer_and_span_rx( }; if let Some(cap) = settings.max_queue_size { - let (consumer, span_rx) = super::channel::channel(cap); + let (consumer, span_rx) = super::channel::channel(cap, PipelineType::System); let tracer = Tracer::with_consumer(sampler, consumer); Ok((tracer, span_rx)) } else { - let (consumer, span_rx) = super::channel::unbounded_channel(); + let (consumer, span_rx) = super::channel::unbounded_channel(PipelineType::System); let tracer = Tracer::with_consumer(sampler, consumer); Ok((tracer, span_rx)) } @@ -122,7 +122,8 @@ pub(crate) fn init( .max_queue_size .map(std::num::NonZeroUsize::get) .unwrap_or(usize::MAX); - super::metrics::tracing::max_queue_size().set(max_queue_size as u64); + super::metrics::tracing::max_queue_size(PipelineType::System) + .set(max_queue_size as u64); } res = Ok(futs.initializer); diff --git a/foundations/src/telemetry/tracing/metrics.rs b/foundations/src/telemetry/tracing/metrics.rs index b823b412..6c12fa23 100644 --- a/foundations/src/telemetry/tracing/metrics.rs +++ b/foundations/src/telemetry/tracing/metrics.rs @@ -1,20 +1,21 @@ //! Tracing-related metrics. +use super::channel::PipelineType; use crate::telemetry::metrics::{Counter, Gauge}; /// Tracing metrics. #[crate::telemetry::metrics::metrics(crate_path = "crate", unprefixed)] pub mod tracing { /// Current size of the span consumer queue. - pub fn queue_size() -> Gauge; + pub fn queue_size(pipeline: PipelineType) -> Gauge; /// Maximum allowed size of the span consumer queue. `usize::MAX` for /// unbounded queues. - pub fn max_queue_size() -> Gauge; + pub fn max_queue_size(pipeline: PipelineType) -> Gauge; /// Total number of spans produced. - pub fn spans_total() -> Counter; + pub fn spans_total(pipeline: PipelineType) -> Counter; /// Total number of spans dropped because the consumer queue was full. - pub fn spans_dropped() -> Counter; + pub fn spans_dropped(pipeline: PipelineType) -> Counter; }