tracing_opentelemetry/
metrics.rs

1use std::{collections::HashMap, fmt, sync::RwLock};
2use tracing::{field::Visit, Subscriber};
3use tracing_core::{Field, Interest, Metadata};
4
5use opentelemetry::{
6    metrics::{Counter, Gauge, Histogram, Meter, MeterProvider, UpDownCounter},
7    InstrumentationScope, KeyValue, Value,
8};
9use tracing_subscriber::{
10    filter::Filtered,
11    layer::{Context, Filter},
12    registry::LookupSpan,
13    Layer,
14};
15
16use smallvec::SmallVec;
17
18const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION");
19const INSTRUMENTATION_LIBRARY_NAME: &str = "tracing/tracing-opentelemetry";
20
21const METRIC_PREFIX_MONOTONIC_COUNTER: &str = "monotonic_counter.";
22const METRIC_PREFIX_COUNTER: &str = "counter.";
23const METRIC_PREFIX_HISTOGRAM: &str = "histogram.";
24const METRIC_PREFIX_GAUGE: &str = "gauge.";
25
26const I64_MAX: u64 = i64::MAX as u64;
27
28#[derive(Default)]
29pub(crate) struct Instruments {
30    u64_counter: MetricsMap<Counter<u64>>,
31    f64_counter: MetricsMap<Counter<f64>>,
32    i64_up_down_counter: MetricsMap<UpDownCounter<i64>>,
33    f64_up_down_counter: MetricsMap<UpDownCounter<f64>>,
34    u64_histogram: MetricsMap<Histogram<u64>>,
35    f64_histogram: MetricsMap<Histogram<f64>>,
36    u64_gauge: MetricsMap<Gauge<u64>>,
37    i64_gauge: MetricsMap<Gauge<i64>>,
38    f64_gauge: MetricsMap<Gauge<f64>>,
39}
40
41type MetricsMap<T> = RwLock<HashMap<&'static str, T>>;
42
43#[derive(Copy, Clone, Debug)]
44pub(crate) enum InstrumentType {
45    CounterU64(u64),
46    CounterF64(f64),
47    UpDownCounterI64(i64),
48    UpDownCounterF64(f64),
49    HistogramU64(u64),
50    HistogramF64(f64),
51    GaugeU64(u64),
52    GaugeI64(i64),
53    GaugeF64(f64),
54}
55
56impl Instruments {
57    pub(crate) fn update_metric(
58        &self,
59        meter: &Meter,
60        instrument_type: InstrumentType,
61        metric_name: &'static str,
62        attributes: &[KeyValue],
63    ) {
64        fn update_or_insert<T>(
65            map: &MetricsMap<T>,
66            name: &'static str,
67            insert: impl FnOnce() -> T,
68            update: impl FnOnce(&T),
69        ) {
70            {
71                let lock = map.read().unwrap();
72                if let Some(metric) = lock.get(name) {
73                    update(metric);
74                    return;
75                }
76            }
77
78            // that metric did not already exist, so we have to acquire a write lock to
79            // create it.
80            let mut lock = map.write().unwrap();
81            // handle the case where the entry was created while we were waiting to
82            // acquire the write lock
83            let metric = lock.entry(name).or_insert_with(insert);
84            update(metric)
85        }
86
87        match instrument_type {
88            InstrumentType::CounterU64(value) => {
89                update_or_insert(
90                    &self.u64_counter,
91                    metric_name,
92                    || meter.u64_counter(metric_name).build(),
93                    |ctr| ctr.add(value, attributes),
94                );
95            }
96            InstrumentType::CounterF64(value) => {
97                update_or_insert(
98                    &self.f64_counter,
99                    metric_name,
100                    || meter.f64_counter(metric_name).build(),
101                    |ctr| ctr.add(value, attributes),
102                );
103            }
104            InstrumentType::UpDownCounterI64(value) => {
105                update_or_insert(
106                    &self.i64_up_down_counter,
107                    metric_name,
108                    || meter.i64_up_down_counter(metric_name).build(),
109                    |ctr| ctr.add(value, attributes),
110                );
111            }
112            InstrumentType::UpDownCounterF64(value) => {
113                update_or_insert(
114                    &self.f64_up_down_counter,
115                    metric_name,
116                    || meter.f64_up_down_counter(metric_name).build(),
117                    |ctr| ctr.add(value, attributes),
118                );
119            }
120            InstrumentType::HistogramU64(value) => {
121                update_or_insert(
122                    &self.u64_histogram,
123                    metric_name,
124                    || meter.u64_histogram(metric_name).build(),
125                    |rec| rec.record(value, attributes),
126                );
127            }
128            InstrumentType::HistogramF64(value) => {
129                update_or_insert(
130                    &self.f64_histogram,
131                    metric_name,
132                    || meter.f64_histogram(metric_name).build(),
133                    |rec| rec.record(value, attributes),
134                );
135            }
136            InstrumentType::GaugeU64(value) => {
137                update_or_insert(
138                    &self.u64_gauge,
139                    metric_name,
140                    || meter.u64_gauge(metric_name).build(),
141                    |rec| rec.record(value, attributes),
142                );
143            }
144            InstrumentType::GaugeI64(value) => {
145                update_or_insert(
146                    &self.i64_gauge,
147                    metric_name,
148                    || meter.i64_gauge(metric_name).build(),
149                    |rec| rec.record(value, attributes),
150                );
151            }
152            InstrumentType::GaugeF64(value) => {
153                update_or_insert(
154                    &self.f64_gauge,
155                    metric_name,
156                    || meter.f64_gauge(metric_name).build(),
157                    |rec| rec.record(value, attributes),
158                );
159            }
160        };
161    }
162}
163
164pub(crate) struct MetricVisitor<'a> {
165    attributes: &'a mut SmallVec<[KeyValue; 8]>,
166    visited_metrics: &'a mut SmallVec<[(&'static str, InstrumentType); 2]>,
167}
168
169impl Visit for MetricVisitor<'_> {
170    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
171        self.attributes
172            .push(KeyValue::new(field.name(), format!("{value:?}")));
173    }
174
175    fn record_u64(&mut self, field: &Field, value: u64) {
176        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_GAUGE) {
177            self.visited_metrics
178                .push((metric_name, InstrumentType::GaugeU64(value)));
179            return;
180        }
181        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_MONOTONIC_COUNTER) {
182            self.visited_metrics
183                .push((metric_name, InstrumentType::CounterU64(value)));
184        } else if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_COUNTER) {
185            if value <= I64_MAX {
186                self.visited_metrics
187                    .push((metric_name, InstrumentType::UpDownCounterI64(value as i64)));
188            } else {
189                eprintln!(
190                    "[tracing-opentelemetry]: Received Counter metric, but \
191                    provided u64: {value} is greater than i64::MAX. Ignoring \
192                    this metric."
193                );
194            }
195        } else if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_HISTOGRAM) {
196            self.visited_metrics
197                .push((metric_name, InstrumentType::HistogramU64(value)));
198        } else if value <= I64_MAX {
199            self.attributes
200                .push(KeyValue::new(field.name(), Value::I64(value as i64)));
201        }
202    }
203
204    fn record_f64(&mut self, field: &Field, value: f64) {
205        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_GAUGE) {
206            self.visited_metrics
207                .push((metric_name, InstrumentType::GaugeF64(value)));
208            return;
209        }
210        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_MONOTONIC_COUNTER) {
211            self.visited_metrics
212                .push((metric_name, InstrumentType::CounterF64(value)));
213        } else if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_COUNTER) {
214            self.visited_metrics
215                .push((metric_name, InstrumentType::UpDownCounterF64(value)));
216        } else if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_HISTOGRAM) {
217            self.visited_metrics
218                .push((metric_name, InstrumentType::HistogramF64(value)));
219        } else {
220            self.attributes
221                .push(KeyValue::new(field.name(), Value::F64(value)));
222        }
223    }
224
225    fn record_i64(&mut self, field: &Field, value: i64) {
226        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_GAUGE) {
227            self.visited_metrics
228                .push((metric_name, InstrumentType::GaugeI64(value)));
229            return;
230        }
231        if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_MONOTONIC_COUNTER) {
232            self.visited_metrics
233                .push((metric_name, InstrumentType::CounterU64(value as u64)));
234        } else if let Some(metric_name) = field.name().strip_prefix(METRIC_PREFIX_COUNTER) {
235            self.visited_metrics
236                .push((metric_name, InstrumentType::UpDownCounterI64(value)));
237        } else {
238            self.attributes.push(KeyValue::new(field.name(), value));
239        }
240    }
241
242    fn record_str(&mut self, field: &Field, value: &str) {
243        self.attributes
244            .push(KeyValue::new(field.name(), value.to_owned()));
245    }
246
247    fn record_bool(&mut self, field: &Field, value: bool) {
248        self.attributes.push(KeyValue::new(field.name(), value));
249    }
250}
251
252/// A layer that publishes metrics via the OpenTelemetry SDK.
253///
254/// # Usage
255///
256/// No configuration is needed for this Layer, as it's only responsible for
257/// pushing data out to the `opentelemetry` family of crates. For example, when
258/// using `opentelemetry-otlp`, that crate will provide its own set of
259/// configuration options for setting up the duration metrics will be collected
260/// before exporting to the OpenTelemetry Collector, aggregation of data points,
261/// etc.
262///
263/// ```no_run
264/// use tracing_opentelemetry::MetricsLayer;
265/// use tracing_subscriber::layer::SubscriberExt;
266/// use tracing_subscriber::Registry;
267/// # use opentelemetry_sdk::metrics::SdkMeterProvider;
268///
269/// // Constructing a MeterProvider is out-of-scope for the docs here, but there
270/// // are examples in the opentelemetry repository. See:
271/// // https://github.com/open-telemetry/opentelemetry-rust/blob/dfeac078ff7853e7dc814778524b93470dfa5c9c/examples/metrics-basic/src/main.rs#L7
272/// # let meter_provider: SdkMeterProvider = unimplemented!();
273///
274/// let opentelemetry_metrics =  MetricsLayer::new(meter_provider);
275/// let subscriber = Registry::default().with(opentelemetry_metrics);
276/// tracing::subscriber::set_global_default(subscriber).unwrap();
277/// ```
278///
279/// To publish a new metric, add a key-value pair to your `tracing::Event` that
280/// contains following prefixes:
281/// - `monotonic_counter.` (non-negative numbers): Used when the counter should
282///   only ever increase
283/// - `counter.`: Used when the counter can go up or down
284/// - `histogram.`: Used to report arbitrary values that are likely to be statistically meaningful
285/// - `gauge.`: Used to report instantaneous values that can go up or down
286///
287/// Examples:
288/// ```
289/// # use tracing::info;
290/// info!(monotonic_counter.foo = 1);
291/// info!(monotonic_counter.bar = 1.1);
292///
293/// info!(counter.baz = 1);
294/// info!(counter.baz = -1);
295/// info!(counter.xyz = 1.1);
296///
297/// info!(histogram.qux = 1);
298/// info!(histogram.abc = -1);
299/// info!(histogram.def = 1.1);
300///
301/// info!(gauge.foo = 1);
302/// info!(gauge.bar = 1.1);
303/// ```
304///
305/// # Mixing data types
306///
307/// ## Floating-point numbers
308///
309/// Do not mix floating point and non-floating point numbers for the same
310/// metric. If a floating point number will be used for a given metric, be sure
311/// to cast any other usages of that metric to a floating point number.
312///
313/// Do this:
314/// ```
315/// # use tracing::info;
316/// info!(monotonic_counter.foo = 1_f64);
317/// info!(monotonic_counter.foo = 1.1);
318/// ```
319///
320/// This is because all data published for a given metric name must be the same
321/// numeric type.
322///
323/// ## Integers
324///
325/// Positive and negative integers can be mixed freely. The instrumentation
326/// provided by `tracing` assumes that all integers are `i64` unless explicitly
327/// cast to something else. In the case that an integer *is* cast to `u64`, this
328/// subscriber will handle the conversion internally.
329///
330/// For example:
331/// ```
332/// # use tracing::info;
333/// // The subscriber receives an i64
334/// info!(counter.baz = 1);
335///
336/// // The subscriber receives an i64
337/// info!(counter.baz = -1);
338///
339/// // The subscriber receives a u64, but casts it to i64 internally
340/// info!(counter.baz = 1_u64);
341///
342/// // The subscriber receives a u64, but cannot cast it to i64 because of
343/// // overflow. An error is printed to stderr, and the metric is dropped.
344/// info!(counter.baz = (i64::MAX as u64) + 1)
345/// ```
346///
347/// # Attributes
348///
349/// When `MetricsLayer` outputs metrics, it converts key-value pairs into [Attributes] and associates them with metrics.
350///
351/// [Attributes]: https://opentelemetry.io/docs/specs/otel/common/#attribute
352///
353/// For example:
354/// ```
355/// # use tracing::info;
356/// // adds attributes bar="baz" and qux=2 to the `foo` counter.
357/// info!(monotonic_counter.foo = 1, bar = "baz", qux = 2);
358/// ```
359///
360/// # Implementation Details
361///
362/// `MetricsLayer` holds a set of maps, with each map corresponding to a
363/// type of metric supported by OpenTelemetry. These maps are populated lazily.
364/// The first time that a metric is emitted by the instrumentation, a `Metric`
365/// instance will be created and added to the corresponding map. This means that
366/// any time a metric is emitted by the instrumentation, one map lookup has to
367/// be performed.
368///
369/// In the future, this can be improved by associating each `Metric` instance to
370/// its callsite, eliminating the need for any maps.
371#[cfg_attr(docsrs, doc(cfg(feature = "metrics")))]
372pub struct MetricsLayer<S, M> {
373    inner: Filtered<InstrumentLayer, MetricsFilter, S>,
374    // We need to hold onto this so that the `InstrumentLayer` can use the created `Meter`.
375    _meter_provider: M,
376}
377
378impl<S, M> MetricsLayer<S, M>
379where
380    S: Subscriber + for<'span> LookupSpan<'span>,
381{
382    /// Create a new instance of MetricsLayer.
383    pub fn new(meter_provider: M) -> MetricsLayer<S, M>
384    where
385        M: MeterProvider,
386    {
387        let meter = meter_provider.meter_with_scope(
388            InstrumentationScope::builder(INSTRUMENTATION_LIBRARY_NAME)
389                .with_version(CARGO_PKG_VERSION)
390                .build(),
391        );
392
393        let layer = InstrumentLayer {
394            meter,
395            instruments: Default::default(),
396        };
397
398        MetricsLayer {
399            inner: layer.with_filter(MetricsFilter),
400            _meter_provider: meter_provider,
401        }
402    }
403}
404
405struct MetricsFilter;
406
407impl MetricsFilter {
408    fn is_metrics_event(&self, meta: &Metadata<'_>) -> bool {
409        meta.is_event()
410            && meta.fields().iter().any(|field| {
411                let name = field.name();
412
413                if name.starts_with(METRIC_PREFIX_COUNTER)
414                    || name.starts_with(METRIC_PREFIX_MONOTONIC_COUNTER)
415                    || name.starts_with(METRIC_PREFIX_HISTOGRAM)
416                    || name.starts_with(METRIC_PREFIX_GAUGE)
417                {
418                    return true;
419                }
420
421                false
422            })
423    }
424}
425
426impl<S> Filter<S> for MetricsFilter {
427    fn enabled(&self, meta: &Metadata<'_>, _cx: &Context<'_, S>) -> bool {
428        self.is_metrics_event(meta)
429    }
430
431    fn callsite_enabled(&self, meta: &'static Metadata<'static>) -> Interest {
432        if self.is_metrics_event(meta) {
433            Interest::always()
434        } else {
435            Interest::never()
436        }
437    }
438}
439
440struct InstrumentLayer {
441    meter: Meter,
442    instruments: Instruments,
443}
444
445impl<S> Layer<S> for InstrumentLayer
446where
447    S: Subscriber + for<'span> LookupSpan<'span>,
448{
449    fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
450        let mut attributes = SmallVec::new();
451        let mut visited_metrics = SmallVec::new();
452        let mut metric_visitor = MetricVisitor {
453            attributes: &mut attributes,
454            visited_metrics: &mut visited_metrics,
455        };
456        event.record(&mut metric_visitor);
457
458        // associate attrivutes with visited metrics
459        visited_metrics
460            .into_iter()
461            .for_each(|(metric_name, value)| {
462                self.instruments.update_metric(
463                    &self.meter,
464                    value,
465                    metric_name,
466                    attributes.as_slice(),
467                );
468            })
469    }
470}
471
472impl<S, M: 'static> Layer<S> for MetricsLayer<S, M>
473where
474    S: Subscriber + for<'span> LookupSpan<'span>,
475{
476    fn on_layer(&mut self, subscriber: &mut S) {
477        self.inner.on_layer(subscriber)
478    }
479
480    fn register_callsite(&self, metadata: &'static Metadata<'static>) -> Interest {
481        self.inner.register_callsite(metadata)
482    }
483
484    fn enabled(&self, metadata: &Metadata<'_>, ctx: Context<'_, S>) -> bool {
485        self.inner.enabled(metadata, ctx)
486    }
487
488    fn on_new_span(
489        &self,
490        attrs: &tracing_core::span::Attributes<'_>,
491        id: &tracing_core::span::Id,
492        ctx: Context<'_, S>,
493    ) {
494        self.inner.on_new_span(attrs, id, ctx)
495    }
496
497    fn max_level_hint(&self) -> Option<tracing_core::LevelFilter> {
498        self.inner.max_level_hint()
499    }
500
501    fn on_record(
502        &self,
503        span: &tracing_core::span::Id,
504        values: &tracing_core::span::Record<'_>,
505        ctx: Context<'_, S>,
506    ) {
507        self.inner.on_record(span, values, ctx)
508    }
509
510    fn on_follows_from(
511        &self,
512        span: &tracing_core::span::Id,
513        follows: &tracing_core::span::Id,
514        ctx: Context<'_, S>,
515    ) {
516        self.inner.on_follows_from(span, follows, ctx)
517    }
518
519    fn on_event(&self, event: &tracing_core::Event<'_>, ctx: Context<'_, S>) {
520        self.inner.on_event(event, ctx)
521    }
522
523    fn on_enter(&self, id: &tracing_core::span::Id, ctx: Context<'_, S>) {
524        self.inner.on_enter(id, ctx)
525    }
526
527    fn on_exit(&self, id: &tracing_core::span::Id, ctx: Context<'_, S>) {
528        self.inner.on_exit(id, ctx)
529    }
530
531    fn on_close(&self, id: tracing_core::span::Id, ctx: Context<'_, S>) {
532        self.inner.on_close(id, ctx)
533    }
534
535    fn on_id_change(
536        &self,
537        old: &tracing_core::span::Id,
538        new: &tracing_core::span::Id,
539        ctx: Context<'_, S>,
540    ) {
541        self.inner.on_id_change(old, new, ctx)
542    }
543}
544
545#[cfg(test)]
546mod tests {
547    use super::*;
548    use tracing_subscriber::layer::SubscriberExt;
549
550    struct PanicLayer;
551    impl<S> Layer<S> for PanicLayer
552    where
553        S: Subscriber + for<'span> LookupSpan<'span>,
554    {
555        fn on_event(&self, _event: &tracing_core::Event<'_>, _ctx: Context<'_, S>) {
556            panic!("panic");
557        }
558    }
559
560    #[test]
561    fn filter_layer_should_filter_non_metrics_event() {
562        let layer = PanicLayer.with_filter(MetricsFilter);
563        let subscriber = tracing_subscriber::registry().with(layer);
564
565        tracing::subscriber::with_default(subscriber, || {
566            tracing::info!(key = "val", "foo");
567        });
568    }
569}
tracing_opentelemetry/metrics.rs

tracing_opentelemetry/
metrics.rs
