From eb380eedecbc5f195a5fb6459ecaed0d092e5685 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 28 Jul 2026 16:56:59 -0400 Subject: [PATCH 1/7] feat(otel-telemetry): scaffold libdd-otel-telemetry crate Adds a new crate providing shared OpenTelemetry metrics aggregation and OTLP export, built on the upstream opentelemetry_sdk/opentelemetry-otlp crates. The public API exposes only primitives (opaque instrument ids, numeric values, string attributes) so any dd-trace-xx language binding can consume it without depending on OTel SDK types itself. Logs support and the counting-exporter wrapper for export_counters() are follow-ups. --- Cargo.lock | 135 +++++++++- Cargo.toml | 1 + LICENSE-3rdparty.csv | 6 + libdd-otel-telemetry/Cargo.toml | 32 +++ libdd-otel-telemetry/src/aggregator.rs | 331 +++++++++++++++++++++++++ libdd-otel-telemetry/src/config.rs | 66 +++++ libdd-otel-telemetry/src/error.rs | 44 ++++ libdd-otel-telemetry/src/instrument.rs | 55 ++++ libdd-otel-telemetry/src/lib.rs | 47 ++++ libdd-otel-telemetry/src/resource.rs | 66 +++++ libdd-otel-telemetry/tests/basic.rs | 54 ++++ 11 files changed, 834 insertions(+), 3 deletions(-) create mode 100644 libdd-otel-telemetry/Cargo.toml create mode 100644 libdd-otel-telemetry/src/aggregator.rs create mode 100644 libdd-otel-telemetry/src/config.rs create mode 100644 libdd-otel-telemetry/src/error.rs create mode 100644 libdd-otel-telemetry/src/instrument.rs create mode 100644 libdd-otel-telemetry/src/lib.rs create mode 100644 libdd-otel-telemetry/src/resource.rs create mode 100644 libdd-otel-telemetry/tests/basic.rs diff --git a/Cargo.lock b/Cargo.lock index 65ac3a4fc4..619de73ead 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2921,7 +2921,7 @@ dependencies = [ "rand 0.8.5", "regex", "regex-lite", - "reqwest", + "reqwest 0.13.2", "rustls", "rustls-native-certs", "rustls-platform-verifier", @@ -3120,7 +3120,7 @@ dependencies = [ "hyper", "hyper-util", "libdd-common", - "reqwest", + "reqwest 0.13.2", "rustls", "tempfile", "thiserror 2.0.17", @@ -3189,6 +3189,18 @@ dependencies = [ "libdd-log", ] +[[package]] +name = "libdd-otel-telemetry" +version = "0.1.0" +dependencies = [ + "libdd-shared-runtime", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "tokio", + "tracing", +] + [[package]] name = "libdd-otel-thread-ctx" version = "1.0.0" @@ -3239,7 +3251,7 @@ dependencies = [ "proptest", "prost", "rand 0.8.5", - "reqwest", + "reqwest 0.13.2", "rustc-hash", "rustls", "rustls-platform-verifier", @@ -4158,6 +4170,79 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" +[[package]] +name = "opentelemetry" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.17", + "tracing", +] + +[[package]] +name = "opentelemetry-http" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest 0.12.28", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest 0.12.28", + "thiserror 2.0.17", + "tokio", + "tonic", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "rand 0.9.0", + "thiserror 2.0.17", +] + [[package]] name = "os_info" version = "3.14.0" @@ -4851,6 +4936,38 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "reqwest" version = "0.13.2" @@ -5347,6 +5464,18 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "serde_with" version = "3.11.0" diff --git a/Cargo.toml b/Cargo.toml index 1aed9f883c..a829c1b38b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,7 @@ members = [ "libdd-log", "libdd-log-ffi", "libdd-sampling", + "libdd-otel-telemetry", ] # https://doc.rust-lang.org/cargo/reference/resolver.html diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index ef04646f30..b8a058a986 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -275,6 +275,11 @@ object,https://github.com/gimli-rs/object,Apache-2.0 OR MIT,The object Authors once_cell,https://github.com/matklad/once_cell,MIT OR Apache-2.0,Aleksey Kladov oorandom,https://hg.sr.ht/~icefox/oorandom,MIT,Simon Heath openssl-probe,https://github.com/alexcrichton/openssl-probe,MIT OR Apache-2.0,Alex Crichton +opentelemetry,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry,Apache-2.0,The opentelemetry Authors +opentelemetry-http,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-http,Apache-2.0,The opentelemetry-http Authors +opentelemetry-otlp,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-otlp,Apache-2.0,The opentelemetry-otlp Authors +opentelemetry-proto,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-proto,Apache-2.0,The opentelemetry-proto Authors +opentelemetry_sdk,https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-sdk,Apache-2.0,The opentelemetry_sdk Authors os_info,https://github.com/stanislav-tkach/os_info,MIT,"Jan Schulte , Stanislav Tkach " page_size,https://github.com/Elzair/page_size_rs,MIT OR Apache-2.0,Philip Woods parking,https://github.com/smol-rs/parking,Apache-2.0 OR MIT,"Stjepan Glavina , The Rust Project Developers" @@ -384,6 +389,7 @@ serde_fmt,https://github.com/KodrAus/serde_fmt,Apache-2.0 OR MIT,Ashley Mannix < serde_json,https://github.com/serde-rs/json,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " serde_regex,https://github.com/tailhook/serde-regex,MIT OR Apache-2.0,paul@colomiets.name serde_spanned,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The serde_spanned Authors +serde_urlencoded,https://github.com/nox/serde_urlencoded,MIT OR Apache-2.0,Anthony Ramine serde_with,https://github.com/jonasbb/serde_with,MIT OR Apache-2.0,"Jonas Bushart, Marcin Kaźmierczak" serde_with_macros,https://github.com/jonasbb/serde_with,MIT OR Apache-2.0,Jonas Bushart serde_yaml,https://github.com/dtolnay/serde-yaml,MIT OR Apache-2.0,David Tolnay diff --git a/libdd-otel-telemetry/Cargo.toml b/libdd-otel-telemetry/Cargo.toml new file mode 100644 index 0000000000..b5d0fd832b --- /dev/null +++ b/libdd-otel-telemetry/Cargo.toml @@ -0,0 +1,32 @@ +# Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "libdd-otel-telemetry" +version = "0.1.0" +description = "Shared OpenTelemetry metrics/logs aggregation and OTLP export for Datadog tracers." +homepage = "https://github.com/DataDog/libdatadog/tree/main/libdd-otel-telemetry" +repository = "https://github.com/DataDog/libdatadog/tree/main/libdd-otel-telemetry" +rust-version.workspace = true +edition.workspace = true +license.workspace = true +autobenches = false + +[dependencies] +opentelemetry = "0.31" +opentelemetry_sdk = { version = "0.31", features = ["metrics"] } +opentelemetry-otlp = { version = "0.31", default-features = false, features = [ + "metrics", +] } +tracing.workspace = true +libdd-shared-runtime = { version = "2.0.0", path = "../libdd-shared-runtime", default-features = false } + +[features] +default = ["grpc"] +grpc = ["opentelemetry-otlp/grpc-tonic"] +http = ["opentelemetry-otlp/http-proto", "opentelemetry-otlp/reqwest-client"] +test-utils = [] + +[dev-dependencies] +libdd-shared-runtime = { version = "2.0.0", path = "../libdd-shared-runtime" } +tokio = { version = "1.23", features = ["rt-multi-thread", "macros"] } diff --git a/libdd-otel-telemetry/src/aggregator.rs b/libdd-otel-telemetry/src/aggregator.rs new file mode 100644 index 0000000000..ea0b7b817c --- /dev/null +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -0,0 +1,331 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use libdd_shared_runtime::BlockingRuntime; +use opentelemetry::metrics::{Counter, Gauge, Histogram, MeterProvider, UpDownCounter}; +use opentelemetry::KeyValue; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; +use opentelemetry_sdk::Resource; + +use crate::config::{OtlpExporterConfig, OtlpProtocol, Temporality}; +use crate::error::{BuildWarning, TelemetryAggregatorError}; +use crate::instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; + +/// Snapshot of export attempt counters, polled by the host tracer to feed its own telemetry +/// system. Deliberately a plain data struct rather than a callback: nothing that isn't a +/// primitive crosses the aggregator's public boundary in either direction. +/// +/// NOTE: the counting exporter wrapper (`PushMetricExporter` decorator incrementing these on +/// every export attempt, mirroring dd-trace-rs's `TelemetryTrackingExporter`) is not implemented +/// yet — `export_counters()` currently always returns zeros. Follow-up before Phase 3. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ExportCounters { + pub metrics_export_attempts: u64, + pub metrics_export_successes: u64, + pub metrics_export_failures: u64, +} + +#[derive(Debug, Default)] +struct Counters { + attempts: AtomicU64, + successes: AtomicU64, + failures: AtomicU64, +} + +enum InstrumentHandle { + Counter(Counter), + UpDownCounter(UpDownCounter), + Histogram(Histogram), + Gauge(Gauge), +} + +/// Builds a [`TelemetryAggregator`]. +/// +/// Never fails outright: any misconfiguration (bad endpoint, unsupported protocol, exporter init +/// failure) is captured as a [`BuildWarning`] and the resulting aggregator silently drops +/// everything it's given instead — a misconfigured OTel pipeline must never prevent the host +/// tracer from starting. +pub struct TelemetryAggregatorBuilder { + resource: Resource, + metrics_exporter: Option, + temporality: Temporality, + export_interval: Duration, +} + +impl Default for TelemetryAggregatorBuilder { + fn default() -> Self { + Self { + resource: Resource::builder().build(), + metrics_exporter: None, + temporality: Temporality::default(), + export_interval: Duration::from_secs(60), + } + } +} + +impl TelemetryAggregatorBuilder { + pub fn new() -> Self { + Self::default() + } + + pub fn with_resource(mut self, resource: Resource) -> Self { + self.resource = resource; + self + } + + pub fn with_metrics_exporter(mut self, config: OtlpExporterConfig) -> Self { + self.metrics_exporter = Some(config); + self + } + + pub fn with_metrics_temporality(mut self, temporality: Temporality) -> Self { + self.temporality = temporality; + self + } + + pub fn with_export_interval(mut self, interval: Duration) -> Self { + self.export_interval = interval; + self + } + + /// Builds the aggregator, driving exporter construction on `runtime` since the OTLP + /// exporters require an active async context to initialize their transport. + pub fn build( + self, + runtime: &R, + ) -> (TelemetryAggregator, Vec) { + let mut warnings = Vec::new(); + + let reader = match &self.metrics_exporter { + Some(cfg) => match runtime.block_on(build_metric_exporter(cfg, self.temporality)) { + Ok(Ok(exporter)) => Some( + PeriodicReader::builder(exporter) + .with_interval(self.export_interval) + .build(), + ), + Ok(Err(warning)) => { + warnings.push(warning); + None + } + Err(_) => { + warnings.push(BuildWarning::ExporterInitFailed( + "runtime unavailable while building metrics exporter".to_string(), + )); + None + } + }, + None => None, + }; + + let mut provider_builder = SdkMeterProvider::builder().with_resource(self.resource); + if let Some(reader) = reader { + provider_builder = provider_builder.with_reader(reader); + } + let provider = provider_builder.build(); + let meter = provider.meter("libdd-otel-telemetry"); + + let aggregator = TelemetryAggregator { + provider, + meter, + instruments: Mutex::new(HashMap::new()), + next_id: AtomicU64::new(1), + counters: Arc::new(Counters::default()), + }; + (aggregator, warnings) + } +} + +async fn build_metric_exporter( + config: &OtlpExporterConfig, + temporality: Temporality, +) -> Result { + use opentelemetry_otlp::WithExportConfig; + + let result = match config.protocol { + #[cfg(feature = "grpc")] + OtlpProtocol::Grpc => opentelemetry_otlp::MetricExporter::builder() + .with_tonic() + .with_endpoint(&config.endpoint) + .with_timeout(config.timeout) + .with_temporality(temporality.into()) + .build(), + #[cfg(not(feature = "grpc"))] + OtlpProtocol::Grpc => { + return Err(BuildWarning::UnsupportedProtocol( + "grpc protocol requires the 'grpc' feature".to_string(), + )) + } + #[cfg(feature = "http")] + OtlpProtocol::HttpProtobuf => opentelemetry_otlp::MetricExporter::builder() + .with_http() + .with_endpoint(&config.endpoint) + .with_timeout(config.timeout) + .with_temporality(temporality.into()) + .build(), + #[cfg(not(feature = "http"))] + OtlpProtocol::HttpProtobuf => { + return Err(BuildWarning::UnsupportedProtocol( + "http/protobuf protocol requires the 'http' feature".to_string(), + )) + } + }; + + result.map_err(|e| BuildWarning::ExporterInitFailed(e.to_string())) +} + +/// Aggregates primitive metric observations from a host tracer and exports them via OTLP. +/// +/// This is the entire public surface a host language binds to: register an instrument once, then +/// push resolved primitive values for it. The aggregator does not know or care whether a value +/// came from a synchronous instrument call or from a host-language-scheduled observable-instrument +/// callback — both are just "a value for this instrument id." +pub struct TelemetryAggregator { + provider: SdkMeterProvider, + meter: opentelemetry::metrics::Meter, + instruments: Mutex>, + next_id: AtomicU64, + counters: Arc, +} + +impl TelemetryAggregator { + pub fn register_instrument(&self, descriptor: InstrumentDescriptor) -> InstrumentId { + let id = InstrumentId(self.next_id.fetch_add(1, Ordering::Relaxed)); + let handle = self.create_instrument(&descriptor); + self.instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(id, handle); + id + } + + fn create_instrument(&self, descriptor: &InstrumentDescriptor) -> InstrumentHandle { + let name = descriptor.name.clone(); + match descriptor.kind { + InstrumentKind::Counter | InstrumentKind::ObservableCounter => { + let mut builder = self.meter.f64_counter(name); + if let Some(unit) = &descriptor.unit { + builder = builder.with_unit(unit.clone()); + } + if let Some(description) = &descriptor.description { + builder = builder.with_description(description.clone()); + } + InstrumentHandle::Counter(builder.build()) + } + InstrumentKind::UpDownCounter | InstrumentKind::ObservableUpDownCounter => { + let mut builder = self.meter.f64_up_down_counter(name); + if let Some(unit) = &descriptor.unit { + builder = builder.with_unit(unit.clone()); + } + if let Some(description) = &descriptor.description { + builder = builder.with_description(description.clone()); + } + InstrumentHandle::UpDownCounter(builder.build()) + } + InstrumentKind::Histogram => { + let mut builder = self.meter.f64_histogram(name); + if let Some(unit) = &descriptor.unit { + builder = builder.with_unit(unit.clone()); + } + if let Some(description) = &descriptor.description { + builder = builder.with_description(description.clone()); + } + InstrumentHandle::Histogram(builder.build()) + } + InstrumentKind::ObservableGauge => { + let mut builder = self.meter.f64_gauge(name); + if let Some(unit) = &descriptor.unit { + builder = builder.with_unit(unit.clone()); + } + if let Some(description) = &descriptor.description { + builder = builder.with_description(description.clone()); + } + InstrumentHandle::Gauge(builder.build()) + } + } + } + + fn attrs(pairs: &[(String, String)]) -> Vec { + pairs + .iter() + .map(|(k, v)| KeyValue::new(k.clone(), v.clone())) + .collect() + } + + pub fn record_counter(&self, id: InstrumentId, value: f64, attrs: &[(String, String)]) { + if let Some(InstrumentHandle::Counter(counter)) = self + .instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&id) + { + counter.add(value, &Self::attrs(attrs)); + } + } + + pub fn record_up_down_counter(&self, id: InstrumentId, value: f64, attrs: &[(String, String)]) { + if let Some(InstrumentHandle::UpDownCounter(counter)) = self + .instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&id) + { + counter.add(value, &Self::attrs(attrs)); + } + } + + pub fn record_histogram(&self, id: InstrumentId, value: f64, attrs: &[(String, String)]) { + if let Some(InstrumentHandle::Histogram(histogram)) = self + .instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&id) + { + histogram.record(value, &Self::attrs(attrs)); + } + } + + /// Pushes a resolved value for an observable gauge. The host language is responsible for + /// deciding when to evaluate the user's callback; this only records the result. + pub fn observe_gauge(&self, id: InstrumentId, value: f64, attrs: &[(String, String)]) { + if let Some(InstrumentHandle::Gauge(gauge)) = self + .instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&id) + { + gauge.record(value, &Self::attrs(attrs)); + } + } + + /// Pushes a resolved value for an observable counter, same caveat as [`Self::observe_gauge`]. + pub fn observe_counter(&self, id: InstrumentId, value: f64, attrs: &[(String, String)]) { + self.record_counter(id, value, attrs); + } + + /// Snapshot of export telemetry counters accumulated so far. Poll this after `force_flush` + /// or on your own interval to report into your own telemetry system. + pub fn export_counters(&self) -> ExportCounters { + ExportCounters { + metrics_export_attempts: self.counters.attempts.load(Ordering::Relaxed), + metrics_export_successes: self.counters.successes.load(Ordering::Relaxed), + metrics_export_failures: self.counters.failures.load(Ordering::Relaxed), + } + } + + pub fn force_flush(&self) -> Result<(), TelemetryAggregatorError> { + self.provider + .force_flush() + .map_err(|e| TelemetryAggregatorError(e.to_string())) + } + + pub fn shutdown(self) -> Result<(), TelemetryAggregatorError> { + self.provider + .shutdown() + .map_err(|e| TelemetryAggregatorError(e.to_string())) + } +} diff --git a/libdd-otel-telemetry/src/config.rs b/libdd-otel-telemetry/src/config.rs new file mode 100644 index 0000000000..16b85d0dbd --- /dev/null +++ b/libdd-otel-telemetry/src/config.rs @@ -0,0 +1,66 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use std::time::Duration; + +/// Wire protocol used to speak OTLP to the configured endpoint. +/// +/// Mirrors the protocol choice already exposed by dd-trace-rs's in-house OTel pipeline +/// (`datadog-opentelemetry::configuration::OtlpProtocol`) so consumers can translate their +/// existing `OTEL_EXPORTER_OTLP_*_PROTOCOL` config directly. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OtlpProtocol { + Grpc, + HttpProtobuf, +} + +/// Aggregation temporality preference for metrics export. +/// +/// Re-exported as a crate-local type (rather than requiring consumers to depend on +/// `opentelemetry_sdk` directly) so no upstream SDK type ever needs to appear in a consumer's +/// public surface. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum Temporality { + #[default] + Delta, + Cumulative, +} + +impl From for opentelemetry_sdk::metrics::Temporality { + fn from(value: Temporality) -> Self { + match value { + Temporality::Delta => opentelemetry_sdk::metrics::Temporality::Delta, + Temporality::Cumulative => opentelemetry_sdk::metrics::Temporality::Cumulative, + } + } +} + +/// Configuration for a single OTLP exporter (metrics or logs). +#[derive(Debug, Clone)] +pub struct OtlpExporterConfig { + pub endpoint: String, + pub protocol: OtlpProtocol, + pub timeout: Duration, + pub headers: Vec<(String, String)>, +} + +impl OtlpExporterConfig { + pub fn new(endpoint: impl Into, protocol: OtlpProtocol) -> Self { + Self { + endpoint: endpoint.into(), + protocol, + timeout: Duration::from_secs(10), + headers: Vec::new(), + } + } + + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + + pub fn with_header(mut self, key: impl Into, value: impl Into) -> Self { + self.headers.push((key.into(), value.into())); + self + } +} diff --git a/libdd-otel-telemetry/src/error.rs b/libdd-otel-telemetry/src/error.rs new file mode 100644 index 0000000000..881f34b91a --- /dev/null +++ b/libdd-otel-telemetry/src/error.rs @@ -0,0 +1,44 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use std::fmt; + +/// A non-fatal problem encountered while building a [`crate::TelemetryAggregator`]. +/// +/// The aggregator is always usable after `build()` — on any of these conditions it falls back to +/// a no-op internal state (recorded values are dropped) rather than failing construction, so a +/// misconfigured exporter never prevents the host tracer from starting up. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BuildWarning { + /// The configured OTLP endpoint could not be parsed as a valid URL. + InvalidEndpoint(String), + /// The requested wire protocol is not supported (e.g. `http/json`). + UnsupportedProtocol(String), + /// The underlying `opentelemetry-otlp` exporter failed to build. + ExporterInitFailed(String), +} + +impl fmt::Display for BuildWarning { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + BuildWarning::InvalidEndpoint(msg) => write!(f, "invalid OTLP endpoint: {msg}"), + BuildWarning::UnsupportedProtocol(msg) => write!(f, "unsupported OTLP protocol: {msg}"), + BuildWarning::ExporterInitFailed(msg) => { + write!(f, "failed to initialize OTLP exporter: {msg}") + } + } + } +} + +/// Error returned from lifecycle operations (`force_flush`/`shutdown`) on a +/// [`crate::TelemetryAggregator`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TelemetryAggregatorError(pub(crate) String); + +impl fmt::Display for TelemetryAggregatorError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl std::error::Error for TelemetryAggregatorError {} diff --git a/libdd-otel-telemetry/src/instrument.rs b/libdd-otel-telemetry/src/instrument.rs new file mode 100644 index 0000000000..66aa2996e8 --- /dev/null +++ b/libdd-otel-telemetry/src/instrument.rs @@ -0,0 +1,55 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +/// Opaque handle to an instrument registered on a [`crate::TelemetryAggregator`]. +/// +/// This is the only thing consumers hold onto for an instrument — never the underlying SDK +/// object — so the handle is a plain primitive and can cross an FFI boundary unchanged. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct InstrumentId(pub u64); + +/// The kind of instrument being registered. +/// +/// The aggregator does not distinguish synchronous instruments (`Counter`, `Histogram`, +/// `UpDownCounter`) from observable/async ones (`ObservableGauge`, `ObservableCounter`) once +/// registered — both just receive resolved primitive values via `record_*`/`observe_*`. The host +/// language owns deciding *when* an observable instrument's callback runs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InstrumentKind { + Counter, + UpDownCounter, + Histogram, + ObservableGauge, + ObservableCounter, + ObservableUpDownCounter, +} + +/// Metadata needed to create the underlying instrument once, at registration time. +#[derive(Debug, Clone)] +pub struct InstrumentDescriptor { + pub name: String, + pub kind: InstrumentKind, + pub unit: Option, + pub description: Option, +} + +impl InstrumentDescriptor { + pub fn new(name: impl Into, kind: InstrumentKind) -> Self { + Self { + name: name.into(), + kind, + unit: None, + description: None, + } + } + + pub fn with_unit(mut self, unit: impl Into) -> Self { + self.unit = Some(unit.into()); + self + } + + pub fn with_description(mut self, description: impl Into) -> Self { + self.description = Some(description.into()); + self + } +} diff --git a/libdd-otel-telemetry/src/lib.rs b/libdd-otel-telemetry/src/lib.rs new file mode 100644 index 0000000000..2f7d2a0b8b --- /dev/null +++ b/libdd-otel-telemetry/src/lib.rs @@ -0,0 +1,47 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +//! Shared OpenTelemetry metrics aggregation and OTLP export for Datadog tracers. +//! +//! Every `dd-trace-xx` library today depends on and configures its own copy of the community +//! OpenTelemetry **SDK** (aggregation, views, OTLP encoding) to support `ddtrace`'s OTel +//! metrics/logs bridge. This crate centralizes that implementation in one place, built on top of +//! upstream `opentelemetry_sdk`/`opentelemetry-otlp` — this is the only crate in the Datadog +//! tracer ecosystem meant to depend on those packages going forward. +//! +//! # Design principle: primitives only +//! +//! [`TelemetryAggregator`]'s public API never accepts or returns an OTel SDK object, and never +//! accepts a callback/closure. Consumers register an instrument once (getting back an opaque +//! [`InstrumentId`]) and then push primitive values (`f64` + string key/value attributes) for it. +//! This holds even for observable/async instruments (`ObservableGauge`, `ObservableCounter`): +//! the host language keeps ownership of *when* to evaluate a user-registered callback (only it +//! can execute that closure), and pushes the resolved value the same way a synchronous +//! instrument would. The aggregator does not need to know which kind produced a given value. +//! +//! Keeping the boundary primitives-only is deliberate, not incidental: it's what makes this +//! crate usable from Rust (dd-trace-rs, today) and, later, from other languages via a C-ABI +//! `-ffi` layer (dd-trace-py via PyO3 first; Node/Ruby/PHP after) without redesigning the core. +//! +//! # What this crate does not do +//! +//! - It does not decide whether to configure OTel support at all — "defer to a user's own OTel SDK +//! setup if they've already configured one" is inherently host-language/SDK-specific and stays +//! the host tracer's responsibility. +//! - It does not read any tracer's configuration type directly — callers extract primitives from +//! their own config and pass them to the builder. +#![cfg_attr(not(test), deny(clippy::unwrap_used))] +#![cfg_attr(not(test), deny(clippy::expect_used))] +#![cfg_attr(not(test), deny(clippy::panic))] + +mod aggregator; +mod config; +mod error; +mod instrument; +mod resource; + +pub use aggregator::{ExportCounters, TelemetryAggregator, TelemetryAggregatorBuilder}; +pub use config::{OtlpExporterConfig, OtlpProtocol, Temporality}; +pub use error::{BuildWarning, TelemetryAggregatorError}; +pub use instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; +pub use resource::ResourceBuilder; diff --git a/libdd-otel-telemetry/src/resource.rs b/libdd-otel-telemetry/src/resource.rs new file mode 100644 index 0000000000..24c64e551a --- /dev/null +++ b/libdd-otel-telemetry/src/resource.rs @@ -0,0 +1,66 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use opentelemetry_sdk::Resource; + +/// Builds an OTel `Resource` from primitive attributes, with Datadog's precedence rules. +/// +/// Mirrors `datadog-opentelemetry::otlp_utils::build_otel_resource`'s merge order: explicit +/// `service`/`env`/`version` win over generic attributes, which win over defaults. Kept +/// `Config`-agnostic — each consumer extracts primitives from its own configuration and passes +/// them in, rather than this crate reading any tracer-specific config type directly. +#[derive(Debug, Default)] +pub struct ResourceBuilder { + service: Option, + env: Option, + version: Option, + attributes: Vec<(String, String)>, +} + +impl ResourceBuilder { + pub fn new() -> Self { + Self::default() + } + + pub fn with_service(mut self, service: impl Into) -> Self { + self.service = Some(service.into()); + self + } + + pub fn with_env(mut self, env: impl Into) -> Self { + self.env = Some(env.into()); + self + } + + pub fn with_version(mut self, version: impl Into) -> Self { + self.version = Some(version.into()); + self + } + + /// Adds a generic resource attribute. Later calls with the same key overwrite earlier ones; + /// `service`/`env`/`version` always take precedence over attributes added this way, + /// regardless of call order. + pub fn with_attribute(mut self, key: impl Into, value: impl Into) -> Self { + self.attributes.push((key.into(), value.into())); + self + } + + pub fn build(self) -> Resource { + let mut builder = Resource::builder(); + for (key, value) in self.attributes { + builder = builder.with_attribute(opentelemetry::KeyValue::new(key, value)); + } + if let Some(service) = self.service { + builder = builder.with_service_name(service); + } + if let Some(env) = self.env { + builder = + builder.with_attribute(opentelemetry::KeyValue::new("deployment.environment", env)); + } + if let Some(version) = self.version { + builder = + builder.with_attribute(opentelemetry::KeyValue::new("service.version", version)); + } + builder.build() + } +} diff --git a/libdd-otel-telemetry/tests/basic.rs b/libdd-otel-telemetry/tests/basic.rs new file mode 100644 index 0000000000..c4d0bbe961 --- /dev/null +++ b/libdd-otel-telemetry/tests/basic.rs @@ -0,0 +1,54 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use libdd_otel_telemetry::{InstrumentDescriptor, InstrumentKind, TelemetryAggregatorBuilder}; +use libdd_shared_runtime::BasicRuntime; +use libdd_shared_runtime::SharedRuntime; + +#[test] +fn register_and_record_without_an_exporter_never_panics() { + let runtime = BasicRuntime::new().expect("runtime"); + let (aggregator, warnings) = TelemetryAggregatorBuilder::new().build(&runtime); + assert!( + warnings.is_empty(), + "no exporter configured, expect no warnings" + ); + + let counter_id = aggregator.register_instrument(InstrumentDescriptor::new( + "requests", + InstrumentKind::Counter, + )); + let gauge_id = aggregator.register_instrument(InstrumentDescriptor::new( + "queue.depth", + InstrumentKind::ObservableGauge, + )); + + aggregator.record_counter( + counter_id, + 1.0, + &[("route".to_string(), "/health".to_string())], + ); + aggregator.observe_gauge(gauge_id, 42.0, &[]); + + aggregator + .force_flush() + .expect("force_flush should succeed even with no reader"); + aggregator.shutdown().expect("shutdown should succeed"); +} + +#[test] +fn unsupported_protocol_falls_back_to_a_warning_not_a_panic() { + use libdd_otel_telemetry::{OtlpExporterConfig, OtlpProtocol}; + + let runtime = BasicRuntime::new().expect("runtime"); + let (_, _warnings) = TelemetryAggregatorBuilder::new() + .with_metrics_exporter(OtlpExporterConfig::new( + "http://localhost:4318", + OtlpProtocol::HttpProtobuf, + )) + .build(&runtime); + + // Built with default features (grpc only), so http/protobuf should warn, not panic. + #[cfg(not(feature = "http"))] + assert_eq!(_warnings.len(), 1); +} From 2992fee8cd5ae9d3cc4d199b9f2723087709ccc3 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 28 Jul 2026 17:24:00 -0400 Subject: [PATCH 2/7] ci(otel-telemetry): add CODEOWNERS entry for libdd-otel-telemetry The new crate's files had no owner, failing the "Not Owned File Checker". Co-Authored-By: Claude Sonnet 5 --- .github/CODEOWNERS | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index eb579c55b5..a9def80e3d 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -55,6 +55,7 @@ libdd-http-client @DataDog/apm-common-components-core libdd-agent-client @DataDog/apm-common-components-core libdd-library-config*/ @DataDog/apm-sdk-capabilities-rust libdd-log*/ @DataDog/apm-common-components-core +libdd-otel-telemetry/ @DataDog/apm-common-components-core libdd-otel-thread-ctx/ @DataDog/apm-common-components-core libdd-otel-thread-ctx-ffi/ @DataDog/apm-common-components-core libdd-profiling*/ @DataDog/libdatadog-profiling From 6b884ede29373cd13ac24e6afde215a17c9e5de9 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Wed, 29 Jul 2026 18:55:17 -0400 Subject: [PATCH 3/7] refactor(otel-telemetry)!: rename TelemetryAggregator to OtelMetricsAggregator The type is OTel-metrics-specific (SdkMeterProvider + PeriodicReader + OTLP metric exporter), but "TelemetryAggregator" read as generic and collided with the tracer-telemetry concept. Rename to make the scope explicit and leave room for per-signal siblings (e.g. OtelLogs*) later: TelemetryAggregator -> OtelMetricsAggregator TelemetryAggregatorBuilder -> OtelMetricsAggregatorBuilder TelemetryAggregatorError -> OtelMetricsError Co-Authored-By: Claude Sonnet 5 --- libdd-otel-telemetry/src/aggregator.rs | 26 +++++++++++++------------- libdd-otel-telemetry/src/error.rs | 10 +++++----- libdd-otel-telemetry/src/instrument.rs | 2 +- libdd-otel-telemetry/src/lib.rs | 6 +++--- libdd-otel-telemetry/tests/basic.rs | 6 +++--- 5 files changed, 25 insertions(+), 25 deletions(-) diff --git a/libdd-otel-telemetry/src/aggregator.rs b/libdd-otel-telemetry/src/aggregator.rs index ea0b7b817c..30e274060b 100644 --- a/libdd-otel-telemetry/src/aggregator.rs +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -13,7 +13,7 @@ use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use opentelemetry_sdk::Resource; use crate::config::{OtlpExporterConfig, OtlpProtocol, Temporality}; -use crate::error::{BuildWarning, TelemetryAggregatorError}; +use crate::error::{BuildWarning, OtelMetricsError}; use crate::instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; /// Snapshot of export attempt counters, polled by the host tracer to feed its own telemetry @@ -44,20 +44,20 @@ enum InstrumentHandle { Gauge(Gauge), } -/// Builds a [`TelemetryAggregator`]. +/// Builds a [`OtelMetricsAggregator`]. /// /// Never fails outright: any misconfiguration (bad endpoint, unsupported protocol, exporter init /// failure) is captured as a [`BuildWarning`] and the resulting aggregator silently drops /// everything it's given instead — a misconfigured OTel pipeline must never prevent the host /// tracer from starting. -pub struct TelemetryAggregatorBuilder { +pub struct OtelMetricsAggregatorBuilder { resource: Resource, metrics_exporter: Option, temporality: Temporality, export_interval: Duration, } -impl Default for TelemetryAggregatorBuilder { +impl Default for OtelMetricsAggregatorBuilder { fn default() -> Self { Self { resource: Resource::builder().build(), @@ -68,7 +68,7 @@ impl Default for TelemetryAggregatorBuilder { } } -impl TelemetryAggregatorBuilder { +impl OtelMetricsAggregatorBuilder { pub fn new() -> Self { Self::default() } @@ -98,7 +98,7 @@ impl TelemetryAggregatorBuilder { pub fn build( self, runtime: &R, - ) -> (TelemetryAggregator, Vec) { + ) -> (OtelMetricsAggregator, Vec) { let mut warnings = Vec::new(); let reader = match &self.metrics_exporter { @@ -129,7 +129,7 @@ impl TelemetryAggregatorBuilder { let provider = provider_builder.build(); let meter = provider.meter("libdd-otel-telemetry"); - let aggregator = TelemetryAggregator { + let aggregator = OtelMetricsAggregator { provider, meter, instruments: Mutex::new(HashMap::new()), @@ -184,7 +184,7 @@ async fn build_metric_exporter( /// push resolved primitive values for it. The aggregator does not know or care whether a value /// came from a synchronous instrument call or from a host-language-scheduled observable-instrument /// callback — both are just "a value for this instrument id." -pub struct TelemetryAggregator { +pub struct OtelMetricsAggregator { provider: SdkMeterProvider, meter: opentelemetry::metrics::Meter, instruments: Mutex>, @@ -192,7 +192,7 @@ pub struct TelemetryAggregator { counters: Arc, } -impl TelemetryAggregator { +impl OtelMetricsAggregator { pub fn register_instrument(&self, descriptor: InstrumentDescriptor) -> InstrumentId { let id = InstrumentId(self.next_id.fetch_add(1, Ordering::Relaxed)); let handle = self.create_instrument(&descriptor); @@ -317,15 +317,15 @@ impl TelemetryAggregator { } } - pub fn force_flush(&self) -> Result<(), TelemetryAggregatorError> { + pub fn force_flush(&self) -> Result<(), OtelMetricsError> { self.provider .force_flush() - .map_err(|e| TelemetryAggregatorError(e.to_string())) + .map_err(|e| OtelMetricsError(e.to_string())) } - pub fn shutdown(self) -> Result<(), TelemetryAggregatorError> { + pub fn shutdown(self) -> Result<(), OtelMetricsError> { self.provider .shutdown() - .map_err(|e| TelemetryAggregatorError(e.to_string())) + .map_err(|e| OtelMetricsError(e.to_string())) } } diff --git a/libdd-otel-telemetry/src/error.rs b/libdd-otel-telemetry/src/error.rs index 881f34b91a..840646386f 100644 --- a/libdd-otel-telemetry/src/error.rs +++ b/libdd-otel-telemetry/src/error.rs @@ -3,7 +3,7 @@ use std::fmt; -/// A non-fatal problem encountered while building a [`crate::TelemetryAggregator`]. +/// A non-fatal problem encountered while building a [`crate::OtelMetricsAggregator`]. /// /// The aggregator is always usable after `build()` — on any of these conditions it falls back to /// a no-op internal state (recorded values are dropped) rather than failing construction, so a @@ -31,14 +31,14 @@ impl fmt::Display for BuildWarning { } /// Error returned from lifecycle operations (`force_flush`/`shutdown`) on a -/// [`crate::TelemetryAggregator`]. +/// [`crate::OtelMetricsAggregator`]. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct TelemetryAggregatorError(pub(crate) String); +pub struct OtelMetricsError(pub(crate) String); -impl fmt::Display for TelemetryAggregatorError { +impl fmt::Display for OtelMetricsError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}", self.0) } } -impl std::error::Error for TelemetryAggregatorError {} +impl std::error::Error for OtelMetricsError {} diff --git a/libdd-otel-telemetry/src/instrument.rs b/libdd-otel-telemetry/src/instrument.rs index 66aa2996e8..b827244084 100644 --- a/libdd-otel-telemetry/src/instrument.rs +++ b/libdd-otel-telemetry/src/instrument.rs @@ -1,7 +1,7 @@ // Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -/// Opaque handle to an instrument registered on a [`crate::TelemetryAggregator`]. +/// Opaque handle to an instrument registered on a [`crate::OtelMetricsAggregator`]. /// /// This is the only thing consumers hold onto for an instrument — never the underlying SDK /// object — so the handle is a plain primitive and can cross an FFI boundary unchanged. diff --git a/libdd-otel-telemetry/src/lib.rs b/libdd-otel-telemetry/src/lib.rs index 2f7d2a0b8b..b21c64f8b8 100644 --- a/libdd-otel-telemetry/src/lib.rs +++ b/libdd-otel-telemetry/src/lib.rs @@ -11,7 +11,7 @@ //! //! # Design principle: primitives only //! -//! [`TelemetryAggregator`]'s public API never accepts or returns an OTel SDK object, and never +//! [`OtelMetricsAggregator`]'s public API never accepts or returns an OTel SDK object, and never //! accepts a callback/closure. Consumers register an instrument once (getting back an opaque //! [`InstrumentId`]) and then push primitive values (`f64` + string key/value attributes) for it. //! This holds even for observable/async instruments (`ObservableGauge`, `ObservableCounter`): @@ -40,8 +40,8 @@ mod error; mod instrument; mod resource; -pub use aggregator::{ExportCounters, TelemetryAggregator, TelemetryAggregatorBuilder}; +pub use aggregator::{ExportCounters, OtelMetricsAggregator, OtelMetricsAggregatorBuilder}; pub use config::{OtlpExporterConfig, OtlpProtocol, Temporality}; -pub use error::{BuildWarning, TelemetryAggregatorError}; +pub use error::{BuildWarning, OtelMetricsError}; pub use instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; pub use resource::ResourceBuilder; diff --git a/libdd-otel-telemetry/tests/basic.rs b/libdd-otel-telemetry/tests/basic.rs index c4d0bbe961..9a4da18c3e 100644 --- a/libdd-otel-telemetry/tests/basic.rs +++ b/libdd-otel-telemetry/tests/basic.rs @@ -1,14 +1,14 @@ // Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -use libdd_otel_telemetry::{InstrumentDescriptor, InstrumentKind, TelemetryAggregatorBuilder}; +use libdd_otel_telemetry::{InstrumentDescriptor, InstrumentKind, OtelMetricsAggregatorBuilder}; use libdd_shared_runtime::BasicRuntime; use libdd_shared_runtime::SharedRuntime; #[test] fn register_and_record_without_an_exporter_never_panics() { let runtime = BasicRuntime::new().expect("runtime"); - let (aggregator, warnings) = TelemetryAggregatorBuilder::new().build(&runtime); + let (aggregator, warnings) = OtelMetricsAggregatorBuilder::new().build(&runtime); assert!( warnings.is_empty(), "no exporter configured, expect no warnings" @@ -41,7 +41,7 @@ fn unsupported_protocol_falls_back_to_a_warning_not_a_panic() { use libdd_otel_telemetry::{OtlpExporterConfig, OtlpProtocol}; let runtime = BasicRuntime::new().expect("runtime"); - let (_, _warnings) = TelemetryAggregatorBuilder::new() + let (_, _warnings) = OtelMetricsAggregatorBuilder::new() .with_metrics_exporter(OtlpExporterConfig::new( "http://localhost:4318", OtlpProtocol::HttpProtobuf, From 0009cd5e39e3a35ed8c8817fa85dd613437577f3 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 30 Jul 2026 15:03:24 -0400 Subject: [PATCH 4/7] feat(otel-telemetry): add shared Datadog OTLP PushMetricExporter Bump opentelemetry/opentelemetry_sdk/opentelemetry-otlp from 0.31 to 0.32 so the exporter matches dd-trace-rs's SDK version. Add DatadogMetricExporter, a PushMetricExporter that wraps the OTLP MetricExporter and tracks export attempts/successes/failures (centralized equivalent of dd-trace-rs's old TelemetryTrackingExporter), plus build_datadog_metric_exporter to construct it. Refactor the existing build_metric_exporter and Counters to be shared. The in-process OtelMetricsAggregator is unchanged. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 87 +++++++++----------------- LICENSE-3rdparty.csv | 2 +- libdd-otel-telemetry/Cargo.toml | 6 +- libdd-otel-telemetry/src/aggregator.rs | 34 ++++++---- libdd-otel-telemetry/src/exporter.rs | 83 ++++++++++++++++++++++++ libdd-otel-telemetry/src/lib.rs | 4 ++ 6 files changed, 140 insertions(+), 76 deletions(-) create mode 100644 libdd-otel-telemetry/src/exporter.rs diff --git a/Cargo.lock b/Cargo.lock index 619de73ead..6413d39f3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2921,7 +2921,7 @@ dependencies = [ "rand 0.8.5", "regex", "regex-lite", - "reqwest 0.13.2", + "reqwest", "rustls", "rustls-native-certs", "rustls-platform-verifier", @@ -3120,7 +3120,7 @@ dependencies = [ "hyper", "hyper-util", "libdd-common", - "reqwest 0.13.2", + "reqwest", "rustls", "tempfile", "thiserror 2.0.17", @@ -3251,7 +3251,7 @@ dependencies = [ "proptest", "prost", "rand 0.8.5", - "reqwest 0.13.2", + "reqwest", "rustc-hash", "rustls", "rustls-platform-verifier", @@ -4172,9 +4172,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "opentelemetry" -version = "0.31.0" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" dependencies = [ "futures-core", "futures-sink", @@ -4186,22 +4186,22 @@ dependencies = [ [[package]] name = "opentelemetry-http" -version = "0.31.0" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" dependencies = [ "async-trait", "bytes", "http", "opentelemetry", - "reqwest 0.12.28", + "reqwest", ] [[package]] name = "opentelemetry-otlp" -version = "0.31.1" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ "http", "opentelemetry", @@ -4209,17 +4209,18 @@ dependencies = [ "opentelemetry-proto", "opentelemetry_sdk", "prost", - "reqwest 0.12.28", + "reqwest", "thiserror 2.0.17", "tokio", "tonic", + "tonic-types", ] [[package]] name = "opentelemetry-proto" -version = "0.31.0" +version = "0.32.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" dependencies = [ "opentelemetry", "opentelemetry_sdk", @@ -4230,15 +4231,16 @@ dependencies = [ [[package]] name = "opentelemetry_sdk" -version = "0.31.0" +version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" dependencies = [ "futures-channel", "futures-executor", "futures-util", "opentelemetry", "percent-encoding", + "portable-atomic", "rand 0.9.0", "thiserror 2.0.17", ] @@ -4936,38 +4938,6 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" -[[package]] -name = "reqwest" -version = "0.12.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" -dependencies = [ - "base64 0.22.1", - "bytes", - "futures-core", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "js-sys", - "log", - "percent-encoding", - "pin-project-lite", - "serde", - "serde_json", - "serde_urlencoded", - "sync_wrapper", - "tokio", - "tower", - "tower-http", - "tower-service", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "reqwest" version = "0.13.2" @@ -5464,18 +5434,6 @@ dependencies = [ "serde", ] -[[package]] -name = "serde_urlencoded" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" -dependencies = [ - "form_urlencoded", - "itoa", - "ryu", - "serde", -] - [[package]] name = "serde_with" version = "3.11.0" @@ -6291,6 +6249,17 @@ dependencies = [ "tonic", ] +[[package]] +name = "tonic-types" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a875a902255423d34c1f20838ab374126db8eb41625b7947a1d54113b0b7399" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tools" version = "37.0.0" diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index b8a058a986..e901f480d0 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -389,7 +389,6 @@ serde_fmt,https://github.com/KodrAus/serde_fmt,Apache-2.0 OR MIT,Ashley Mannix < serde_json,https://github.com/serde-rs/json,MIT OR Apache-2.0,"Erick Tryzelaar , David Tolnay " serde_regex,https://github.com/tailhook/serde-regex,MIT OR Apache-2.0,paul@colomiets.name serde_spanned,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The serde_spanned Authors -serde_urlencoded,https://github.com/nox/serde_urlencoded,MIT OR Apache-2.0,Anthony Ramine serde_with,https://github.com/jonasbb/serde_with,MIT OR Apache-2.0,"Jonas Bushart, Marcin Kaźmierczak" serde_with_macros,https://github.com/jonasbb/serde_with,MIT OR Apache-2.0,Jonas Bushart serde_yaml,https://github.com/dtolnay/serde-yaml,MIT OR Apache-2.0,David Tolnay @@ -459,6 +458,7 @@ toml_edit,https://github.com/toml-rs/toml,MIT OR Apache-2.0,"Andronik Ordian tonic-prost,https://github.com/hyperium/tonic,MIT,Lucio Franco +tonic-types,https://github.com/hyperium/tonic,MIT,"Lucio Franco , Rafael Lemos " tower,https://github.com/tower-rs/tower,MIT,Tower Maintainers tower-http,https://github.com/tower-rs/tower-http,MIT,Tower Maintainers tower-layer,https://github.com/tower-rs/tower,MIT,Tower Maintainers diff --git a/libdd-otel-telemetry/Cargo.toml b/libdd-otel-telemetry/Cargo.toml index b5d0fd832b..1ac832f7de 100644 --- a/libdd-otel-telemetry/Cargo.toml +++ b/libdd-otel-telemetry/Cargo.toml @@ -13,9 +13,9 @@ license.workspace = true autobenches = false [dependencies] -opentelemetry = "0.31" -opentelemetry_sdk = { version = "0.31", features = ["metrics"] } -opentelemetry-otlp = { version = "0.31", default-features = false, features = [ +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = [ "metrics", ] } tracing.workspace = true diff --git a/libdd-otel-telemetry/src/aggregator.rs b/libdd-otel-telemetry/src/aggregator.rs index 30e274060b..626b88b60e 100644 --- a/libdd-otel-telemetry/src/aggregator.rs +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -20,9 +20,10 @@ use crate::instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; /// system. Deliberately a plain data struct rather than a callback: nothing that isn't a /// primitive crosses the aggregator's public boundary in either direction. /// -/// NOTE: the counting exporter wrapper (`PushMetricExporter` decorator incrementing these on -/// every export attempt, mirroring dd-trace-rs's `TelemetryTrackingExporter`) is not implemented -/// yet — `export_counters()` currently always returns zeros. Follow-up before Phase 3. +/// The counting is performed by [`crate::DatadogMetricExporter`], which increments these on every +/// export attempt (mirroring dd-trace-rs's old `TelemetryTrackingExporter`). The in-process +/// [`OtelMetricsAggregator`] builds its own reader directly and does not yet route through that +/// exporter, so its [`OtelMetricsAggregator::export_counters`] still returns zeros for now. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ExportCounters { pub metrics_export_attempts: u64, @@ -31,10 +32,21 @@ pub struct ExportCounters { } #[derive(Debug, Default)] -struct Counters { - attempts: AtomicU64, - successes: AtomicU64, - failures: AtomicU64, +pub(crate) struct Counters { + pub(crate) attempts: AtomicU64, + pub(crate) successes: AtomicU64, + pub(crate) failures: AtomicU64, +} + +impl Counters { + /// Reads the current counter values into a public [`ExportCounters`] snapshot. + pub(crate) fn snapshot(&self) -> ExportCounters { + ExportCounters { + metrics_export_attempts: self.attempts.load(Ordering::Relaxed), + metrics_export_successes: self.successes.load(Ordering::Relaxed), + metrics_export_failures: self.failures.load(Ordering::Relaxed), + } + } } enum InstrumentHandle { @@ -140,7 +152,7 @@ impl OtelMetricsAggregatorBuilder { } } -async fn build_metric_exporter( +pub(crate) async fn build_metric_exporter( config: &OtlpExporterConfig, temporality: Temporality, ) -> Result { @@ -310,11 +322,7 @@ impl OtelMetricsAggregator { /// Snapshot of export telemetry counters accumulated so far. Poll this after `force_flush` /// or on your own interval to report into your own telemetry system. pub fn export_counters(&self) -> ExportCounters { - ExportCounters { - metrics_export_attempts: self.counters.attempts.load(Ordering::Relaxed), - metrics_export_successes: self.counters.successes.load(Ordering::Relaxed), - metrics_export_failures: self.counters.failures.load(Ordering::Relaxed), - } + self.counters.snapshot() } pub fn force_flush(&self) -> Result<(), OtelMetricsError> { diff --git a/libdd-otel-telemetry/src/exporter.rs b/libdd-otel-telemetry/src/exporter.rs new file mode 100644 index 0000000000..3e72ad61df --- /dev/null +++ b/libdd-otel-telemetry/src/exporter.rs @@ -0,0 +1,83 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::time::Duration; + +use opentelemetry_sdk::error::OTelSdkResult; +use opentelemetry_sdk::metrics::data::ResourceMetrics; +use opentelemetry_sdk::metrics::exporter::PushMetricExporter; +use opentelemetry_sdk::metrics::Temporality as SdkTemporality; + +use crate::aggregator::{build_metric_exporter, Counters}; +use crate::config::{OtlpExporterConfig, Temporality}; +use crate::error::BuildWarning; +use crate::ExportCounters; + +/// A Datadog-flavored OTLP [`PushMetricExporter`] that a host tracer can plug into its own +/// `SdkMeterProvider` + `PeriodicReader`. +/// +/// It wraps an upstream [`opentelemetry_otlp::MetricExporter`] and tracks export attempts, +/// successes, and failures — the centralized equivalent of dd-trace-rs's old +/// `TelemetryTrackingExporter`. Poll [`DatadogMetricExporter::counters`] to feed those counts +/// into your own telemetry system. +pub struct DatadogMetricExporter { + inner: opentelemetry_otlp::MetricExporter, + counters: Arc, +} + +impl DatadogMetricExporter { + /// Snapshot of export telemetry counters accumulated so far. + pub fn counters(&self) -> ExportCounters { + self.counters.snapshot() + } +} + +impl std::fmt::Debug for DatadogMetricExporter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DatadogMetricExporter") + .field("counters", &self.counters.snapshot()) + .finish() + } +} + +impl PushMetricExporter for DatadogMetricExporter { + async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult { + self.counters.attempts.fetch_add(1, Ordering::Relaxed); + let result = self.inner.export(metrics).await; + match &result { + Ok(()) => self.counters.successes.fetch_add(1, Ordering::Relaxed), + Err(_) => self.counters.failures.fetch_add(1, Ordering::Relaxed), + }; + result + } + + fn force_flush(&self) -> OTelSdkResult { + self.inner.force_flush() + } + + fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult { + self.inner.shutdown_with_timeout(timeout) + } + + fn temporality(&self) -> SdkTemporality { + self.inner.temporality() + } +} + +/// Builds a [`DatadogMetricExporter`] from an [`OtlpExporterConfig`]. +/// +/// Must be called from within a tokio runtime: the underlying `opentelemetry-otlp` exporter +/// initializes its transport (tonic/reqwest) and requires an active async context. dd-trace-rs +/// builds its provider inside tokio, so this is `async` rather than driving its own runtime. +pub async fn build_datadog_metric_exporter( + config: &OtlpExporterConfig, + temporality: Temporality, +) -> Result { + let inner = build_metric_exporter(config, temporality).await?; + Ok(DatadogMetricExporter { + inner, + counters: Arc::new(Counters::default()), + }) +} diff --git a/libdd-otel-telemetry/src/lib.rs b/libdd-otel-telemetry/src/lib.rs index b21c64f8b8..61560c2e48 100644 --- a/libdd-otel-telemetry/src/lib.rs +++ b/libdd-otel-telemetry/src/lib.rs @@ -37,11 +37,15 @@ mod aggregator; mod config; mod error; +#[cfg(any(feature = "grpc", feature = "http"))] +mod exporter; mod instrument; mod resource; pub use aggregator::{ExportCounters, OtelMetricsAggregator, OtelMetricsAggregatorBuilder}; pub use config::{OtlpExporterConfig, OtlpProtocol, Temporality}; pub use error::{BuildWarning, OtelMetricsError}; +#[cfg(any(feature = "grpc", feature = "http"))] +pub use exporter::{build_datadog_metric_exporter, DatadogMetricExporter}; pub use instrument::{InstrumentDescriptor, InstrumentId, InstrumentKind}; pub use resource::ResourceBuilder; From 00c5176e7f31324a944ccfcda7d3c9fa66ce6dc1 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 30 Jul 2026 15:21:13 -0400 Subject: [PATCH 5/7] test(otel-telemetry): gate unsupported-protocol test to the no-http build MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under --all-features the `http` feature is on, so http/protobuf is a *supported* protocol and the aggregator builds a real reqwest-backed exporter — which, under opentelemetry-otlp 0.32 + workspace feature unification, eagerly builds a rustls client and panics with "No provider set". The test's scenario (unsupported protocol -> warning, not panic) only exists without the http feature, so gate it there. A positive http-exporter test needs a crypto provider installed (TODO). Co-Authored-By: Claude Sonnet 5 --- libdd-otel-telemetry/tests/basic.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/libdd-otel-telemetry/tests/basic.rs b/libdd-otel-telemetry/tests/basic.rs index 9a4da18c3e..0d2b1a3f60 100644 --- a/libdd-otel-telemetry/tests/basic.rs +++ b/libdd-otel-telemetry/tests/basic.rs @@ -36,19 +36,23 @@ fn register_and_record_without_an_exporter_never_panics() { aggregator.shutdown().expect("shutdown should succeed"); } +// Only meaningful without the `http` feature: there, http/protobuf is an *unsupported* protocol +// and must fall back to a warning rather than panic. With `http` enabled the protocol is supported +// and a real exporter is built, so the "unsupported" scenario doesn't apply. +// TODO: a positive http-exporter test needs a rustls crypto provider installed for reqwest under +// feature unification (opentelemetry-otlp 0.32 builds the client eagerly). +#[cfg(not(feature = "http"))] #[test] fn unsupported_protocol_falls_back_to_a_warning_not_a_panic() { use libdd_otel_telemetry::{OtlpExporterConfig, OtlpProtocol}; let runtime = BasicRuntime::new().expect("runtime"); - let (_, _warnings) = OtelMetricsAggregatorBuilder::new() + let (_, warnings) = OtelMetricsAggregatorBuilder::new() .with_metrics_exporter(OtlpExporterConfig::new( "http://localhost:4318", OtlpProtocol::HttpProtobuf, )) .build(&runtime); - // Built with default features (grpc only), so http/protobuf should warn, not panic. - #[cfg(not(feature = "http"))] - assert_eq!(_warnings.len(), 1); + assert_eq!(warnings.len(), 1); } From e755f3eb282f1b56f7d6e653ea879e510176b2b8 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Fri, 31 Jul 2026 10:02:34 -0400 Subject: [PATCH 6/7] fix(otel-telemetry): preserve instrumentation scope per meter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The aggregator registered every instrument on one hardcoded meter ("libdd-otel-telemetry"), so exported OTLP metrics lost the host's instrumentation scope — all metrics reported scope name "libdd-otel-telemetry" and distinct meters collapsed into one scope. Carry the meter name/version/ schema_url on InstrumentDescriptor and create/cache one SDK Meter per scope so exports keep the host's get_meter identity. Co-Authored-By: Claude Sonnet 5 --- libdd-otel-telemetry/src/aggregator.rs | 41 +++++++++++++++++++++----- libdd-otel-telemetry/src/instrument.rs | 24 +++++++++++++++ 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/libdd-otel-telemetry/src/aggregator.rs b/libdd-otel-telemetry/src/aggregator.rs index 626b88b60e..978a850a01 100644 --- a/libdd-otel-telemetry/src/aggregator.rs +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -139,11 +139,10 @@ impl OtelMetricsAggregatorBuilder { provider_builder = provider_builder.with_reader(reader); } let provider = provider_builder.build(); - let meter = provider.meter("libdd-otel-telemetry"); let aggregator = OtelMetricsAggregator { provider, - meter, + meters: Mutex::new(HashMap::new()), instruments: Mutex::new(HashMap::new()), next_id: AtomicU64::new(1), counters: Arc::new(Counters::default()), @@ -190,6 +189,9 @@ pub(crate) async fn build_metric_exporter( result.map_err(|e| BuildWarning::ExporterInitFailed(e.to_string())) } +/// Identifies an OpenTelemetry instrumentation scope: (meter name, version, schema_url). +type MeterScope = (String, Option, Option); + /// Aggregates primitive metric observations from a host tracer and exports them via OTLP. /// /// This is the entire public surface a host language binds to: register an instrument once, then @@ -198,7 +200,7 @@ pub(crate) async fn build_metric_exporter( /// callback — both are just "a value for this instrument id." pub struct OtelMetricsAggregator { provider: SdkMeterProvider, - meter: opentelemetry::metrics::Meter, + meters: Mutex>, instruments: Mutex>, next_id: AtomicU64, counters: Arc, @@ -215,11 +217,36 @@ impl OtelMetricsAggregator { id } + /// Returns the SDK `Meter` for the descriptor's instrumentation scope, creating and caching it + /// on first use so every exported metric carries the host's `get_meter` identity. + fn meter_for(&self, descriptor: &InstrumentDescriptor) -> opentelemetry::metrics::Meter { + let key: MeterScope = ( + descriptor.meter_name.clone(), + descriptor.meter_version.clone(), + descriptor.meter_schema_url.clone(), + ); + let mut meters = self.meters.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(meter) = meters.get(&key) { + return meter.clone(); + } + let mut scope = opentelemetry::InstrumentationScope::builder(descriptor.meter_name.clone()); + if let Some(version) = &descriptor.meter_version { + scope = scope.with_version(version.clone()); + } + if let Some(schema_url) = &descriptor.meter_schema_url { + scope = scope.with_schema_url(schema_url.clone()); + } + let meter = self.provider.meter_with_scope(scope.build()); + meters.insert(key, meter.clone()); + meter + } + fn create_instrument(&self, descriptor: &InstrumentDescriptor) -> InstrumentHandle { + let meter = self.meter_for(descriptor); let name = descriptor.name.clone(); match descriptor.kind { InstrumentKind::Counter | InstrumentKind::ObservableCounter => { - let mut builder = self.meter.f64_counter(name); + let mut builder = meter.f64_counter(name); if let Some(unit) = &descriptor.unit { builder = builder.with_unit(unit.clone()); } @@ -229,7 +256,7 @@ impl OtelMetricsAggregator { InstrumentHandle::Counter(builder.build()) } InstrumentKind::UpDownCounter | InstrumentKind::ObservableUpDownCounter => { - let mut builder = self.meter.f64_up_down_counter(name); + let mut builder = meter.f64_up_down_counter(name); if let Some(unit) = &descriptor.unit { builder = builder.with_unit(unit.clone()); } @@ -239,7 +266,7 @@ impl OtelMetricsAggregator { InstrumentHandle::UpDownCounter(builder.build()) } InstrumentKind::Histogram => { - let mut builder = self.meter.f64_histogram(name); + let mut builder = meter.f64_histogram(name); if let Some(unit) = &descriptor.unit { builder = builder.with_unit(unit.clone()); } @@ -249,7 +276,7 @@ impl OtelMetricsAggregator { InstrumentHandle::Histogram(builder.build()) } InstrumentKind::ObservableGauge => { - let mut builder = self.meter.f64_gauge(name); + let mut builder = meter.f64_gauge(name); if let Some(unit) = &descriptor.unit { builder = builder.with_unit(unit.clone()); } diff --git a/libdd-otel-telemetry/src/instrument.rs b/libdd-otel-telemetry/src/instrument.rs index b827244084..372507aaec 100644 --- a/libdd-otel-telemetry/src/instrument.rs +++ b/libdd-otel-telemetry/src/instrument.rs @@ -25,12 +25,20 @@ pub enum InstrumentKind { } /// Metadata needed to create the underlying instrument once, at registration time. +/// +/// The `meter_*` fields carry the OpenTelemetry instrumentation scope the instrument belongs to +/// (the name/version/schema_url the host passed to `get_meter`). The aggregator creates one SDK +/// `Meter` per distinct scope so exported metrics keep the host's scope rather than a single +/// crate-internal one. #[derive(Debug, Clone)] pub struct InstrumentDescriptor { pub name: String, pub kind: InstrumentKind, pub unit: Option, pub description: Option, + pub meter_name: String, + pub meter_version: Option, + pub meter_schema_url: Option, } impl InstrumentDescriptor { @@ -40,6 +48,9 @@ impl InstrumentDescriptor { kind, unit: None, description: None, + meter_name: "libdd-otel-telemetry".to_string(), + meter_version: None, + meter_schema_url: None, } } @@ -52,4 +63,17 @@ impl InstrumentDescriptor { self.description = Some(description.into()); self } + + /// Sets the instrumentation scope (the host's `get_meter` identity) this instrument belongs to. + pub fn with_scope( + mut self, + meter_name: impl Into, + meter_version: Option, + meter_schema_url: Option, + ) -> Self { + self.meter_name = meter_name.into(); + self.meter_version = meter_version; + self.meter_schema_url = meter_schema_url; + self + } } From 9b08d7d2fa89f3237b52af754834455aaf9bbd57 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Fri, 31 Jul 2026 12:49:33 -0400 Subject: [PATCH 7/7] feat(otel-telemetry): centralize case-insensitive OTLP config parsing + fix http TLS Add canonical, pure config-parsing helpers to `config.rs` so every consumer (dd-trace-rs, dd-trace-py via PyO3) shares one implementation instead of hand-rolling case-sensitive matches: - `OtlpProtocol::from_config_str` accepts `grpc`/`http/protobuf`/`http/json` case-insensitively (adds an `HttpJson` variant for parity with dd-trace-rs; it reports as an unsupported export protocol like before). - `Temporality::from_config_str` accepts `delta`/`cumulative` case-insensitively, defaulting to `Delta` for empty/unknown. - `parse_otlp_headers` parses `k1=v1,k2=v2` header strings. Also fix the `http` exporter: reqwest+rustls 0.23 panics with no process-default crypto provider under feature unification. Best-effort install the ring provider before building the http exporter (ring only; aws-lc-rs stays FIPS-only). Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 1 + libdd-otel-telemetry/Cargo.toml | 5 +- libdd-otel-telemetry/src/aggregator.rs | 25 +++++++-- libdd-otel-telemetry/src/config.rs | 49 ++++++++++++++++ libdd-otel-telemetry/src/lib.rs | 2 +- libdd-otel-telemetry/tests/basic.rs | 42 ++++++++++++-- libdd-otel-telemetry/tests/config.rs | 77 ++++++++++++++++++++++++++ 7 files changed, 188 insertions(+), 13 deletions(-) create mode 100644 libdd-otel-telemetry/tests/config.rs diff --git a/Cargo.lock b/Cargo.lock index 6413d39f3f..1ad96674b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3197,6 +3197,7 @@ dependencies = [ "opentelemetry", "opentelemetry-otlp", "opentelemetry_sdk", + "rustls", "tokio", "tracing", ] diff --git a/libdd-otel-telemetry/Cargo.toml b/libdd-otel-telemetry/Cargo.toml index 1ac832f7de..891e489c52 100644 --- a/libdd-otel-telemetry/Cargo.toml +++ b/libdd-otel-telemetry/Cargo.toml @@ -20,11 +20,14 @@ opentelemetry-otlp = { version = "0.32", default-features = false, features = [ ] } tracing.workspace = true libdd-shared-runtime = { version = "2.0.0", path = "../libdd-shared-runtime", default-features = false } +# Only needed for the `http` exporter: reqwest+rustls needs a process-default crypto provider +# installed. libdatadog standardizes on ring (aws-lc-rs is FIPS-only), so pin the ring provider. +rustls = { version = "0.23", default-features = false, features = ["ring"], optional = true } [features] default = ["grpc"] grpc = ["opentelemetry-otlp/grpc-tonic"] -http = ["opentelemetry-otlp/http-proto", "opentelemetry-otlp/reqwest-client"] +http = ["opentelemetry-otlp/http-proto", "opentelemetry-otlp/reqwest-client", "dep:rustls"] test-utils = [] [dev-dependencies] diff --git a/libdd-otel-telemetry/src/aggregator.rs b/libdd-otel-telemetry/src/aggregator.rs index 978a850a01..81dd12158d 100644 --- a/libdd-otel-telemetry/src/aggregator.rs +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -172,18 +172,31 @@ pub(crate) async fn build_metric_exporter( )) } #[cfg(feature = "http")] - OtlpProtocol::HttpProtobuf => opentelemetry_otlp::MetricExporter::builder() - .with_http() - .with_endpoint(&config.endpoint) - .with_timeout(config.timeout) - .with_temporality(temporality.into()) - .build(), + OtlpProtocol::HttpProtobuf => { + // reqwest+rustls 0.23 has no process-default crypto provider under feature + // unification and panics ("no process-level CryptoProvider available") when it + // builds a TLS client. libdatadog standardizes on ring, so best-effort install it + // before the exporter constructs its client. Ignoring the result is intentional: + // it errors only if a provider is already set, which is fine. + let _ = rustls::crypto::ring::default_provider().install_default(); + opentelemetry_otlp::MetricExporter::builder() + .with_http() + .with_endpoint(&config.endpoint) + .with_timeout(config.timeout) + .with_temporality(temporality.into()) + .build() + } #[cfg(not(feature = "http"))] OtlpProtocol::HttpProtobuf => { return Err(BuildWarning::UnsupportedProtocol( "http/protobuf protocol requires the 'http' feature".to_string(), )) } + OtlpProtocol::HttpJson => { + return Err(BuildWarning::UnsupportedProtocol( + "http/json protocol is not supported for OTLP metrics export".to_string(), + )) + } }; result.map_err(|e| BuildWarning::ExporterInitFailed(e.to_string())) diff --git a/libdd-otel-telemetry/src/config.rs b/libdd-otel-telemetry/src/config.rs index 16b85d0dbd..df8c151d7a 100644 --- a/libdd-otel-telemetry/src/config.rs +++ b/libdd-otel-telemetry/src/config.rs @@ -12,6 +12,26 @@ use std::time::Duration; pub enum OtlpProtocol { Grpc, HttpProtobuf, + /// HTTP with JSON encoding. Recognized for config-parsing parity with the OTel spec and + /// dd-trace-rs, but not exportable today — the exporter build reports it as unsupported. + HttpJson, +} + +impl OtlpProtocol { + /// Parse an `OTEL_EXPORTER_OTLP_*_PROTOCOL` value case-insensitively. + /// + /// Accepts `grpc`, `http/protobuf`, and `http/json` (any case, surrounding whitespace + /// trimmed). Returns `None` for empty or unrecognized values so the caller decides how to + /// surface the error. + pub fn from_config_str(s: &str) -> Option { + let s = s.trim().to_ascii_lowercase(); + match s.as_str() { + "grpc" => Some(OtlpProtocol::Grpc), + "http/protobuf" => Some(OtlpProtocol::HttpProtobuf), + "http/json" => Some(OtlpProtocol::HttpJson), + _ => None, + } + } } /// Aggregation temporality preference for metrics export. @@ -26,6 +46,35 @@ pub enum Temporality { Cumulative, } +impl Temporality { + /// Parse an `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` value case-insensitively. + /// + /// Accepts `delta` and `cumulative` (any case, surrounding whitespace trimmed). Empty or + /// unrecognized values default to [`Temporality::Delta`], matching Datadog's preference. + pub fn from_config_str(s: &str) -> Temporality { + match s.trim().to_ascii_lowercase().as_str() { + "cumulative" => Temporality::Cumulative, + _ => Temporality::Delta, + } + } +} + +/// Parse an `OTEL_EXPORTER_OTLP_*_HEADERS` string (`k1=v1,k2=v2`) into key/value pairs. +/// +/// Entries without an `=` (or with an empty key) are skipped; keys and values are trimmed. +pub fn parse_otlp_headers(s: &str) -> Vec<(String, String)> { + s.split(',') + .filter_map(|item| { + let (key, value) = item.split_once('=')?; + let key = key.trim(); + if key.is_empty() { + return None; + } + Some((key.to_string(), value.trim().to_string())) + }) + .collect() +} + impl From for opentelemetry_sdk::metrics::Temporality { fn from(value: Temporality) -> Self { match value { diff --git a/libdd-otel-telemetry/src/lib.rs b/libdd-otel-telemetry/src/lib.rs index 61560c2e48..15f0197786 100644 --- a/libdd-otel-telemetry/src/lib.rs +++ b/libdd-otel-telemetry/src/lib.rs @@ -43,7 +43,7 @@ mod instrument; mod resource; pub use aggregator::{ExportCounters, OtelMetricsAggregator, OtelMetricsAggregatorBuilder}; -pub use config::{OtlpExporterConfig, OtlpProtocol, Temporality}; +pub use config::{parse_otlp_headers, OtlpExporterConfig, OtlpProtocol, Temporality}; pub use error::{BuildWarning, OtelMetricsError}; #[cfg(any(feature = "grpc", feature = "http"))] pub use exporter::{build_datadog_metric_exporter, DatadogMetricExporter}; diff --git a/libdd-otel-telemetry/tests/basic.rs b/libdd-otel-telemetry/tests/basic.rs index 0d2b1a3f60..dfd76a9553 100644 --- a/libdd-otel-telemetry/tests/basic.rs +++ b/libdd-otel-telemetry/tests/basic.rs @@ -36,11 +36,9 @@ fn register_and_record_without_an_exporter_never_panics() { aggregator.shutdown().expect("shutdown should succeed"); } -// Only meaningful without the `http` feature: there, http/protobuf is an *unsupported* protocol -// and must fall back to a warning rather than panic. With `http` enabled the protocol is supported -// and a real exporter is built, so the "unsupported" scenario doesn't apply. -// TODO: a positive http-exporter test needs a rustls crypto provider installed for reqwest under -// feature unification (opentelemetry-otlp 0.32 builds the client eagerly). +// Without the `http` feature, http/protobuf is an *unsupported* protocol and must fall back to a +// warning rather than panic. With `http` enabled the protocol is supported and a real exporter is +// built, so the "unsupported" scenario doesn't apply. #[cfg(not(feature = "http"))] #[test] fn unsupported_protocol_falls_back_to_a_warning_not_a_panic() { @@ -56,3 +54,37 @@ fn unsupported_protocol_falls_back_to_a_warning_not_a_panic() { assert_eq!(warnings.len(), 1); } + +// With the `http` feature the exporter builds a reqwest+rustls client eagerly; this exercises the +// ring crypto-provider install so a missing default provider can't panic during setup. +#[cfg(feature = "http")] +#[test] +fn http_protobuf_exporter_builds_without_panicking() { + use libdd_otel_telemetry::{OtlpExporterConfig, OtlpProtocol}; + + let runtime = BasicRuntime::new().expect("runtime"); + let (_, warnings) = OtelMetricsAggregatorBuilder::new() + .with_metrics_exporter(OtlpExporterConfig::new( + "http://localhost:4318", + OtlpProtocol::HttpProtobuf, + )) + .build(&runtime); + + assert!(warnings.is_empty(), "http exporter should build cleanly"); +} + +// http/json is recognized by config parsing but not exportable; it must warn, not panic. +#[test] +fn http_json_protocol_falls_back_to_a_warning() { + use libdd_otel_telemetry::{OtlpExporterConfig, OtlpProtocol}; + + let runtime = BasicRuntime::new().expect("runtime"); + let (_, warnings) = OtelMetricsAggregatorBuilder::new() + .with_metrics_exporter(OtlpExporterConfig::new( + "http://localhost:4318", + OtlpProtocol::HttpJson, + )) + .build(&runtime); + + assert_eq!(warnings.len(), 1); +} diff --git a/libdd-otel-telemetry/tests/config.rs b/libdd-otel-telemetry/tests/config.rs new file mode 100644 index 0000000000..682ca1161b --- /dev/null +++ b/libdd-otel-telemetry/tests/config.rs @@ -0,0 +1,77 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use libdd_otel_telemetry::{parse_otlp_headers, OtlpProtocol, Temporality}; + +#[test] +fn protocol_parses_case_insensitively() { + assert_eq!( + OtlpProtocol::from_config_str("grpc"), + Some(OtlpProtocol::Grpc) + ); + assert_eq!( + OtlpProtocol::from_config_str("GRPC"), + Some(OtlpProtocol::Grpc) + ); + assert_eq!( + OtlpProtocol::from_config_str(" http/protobuf "), + Some(OtlpProtocol::HttpProtobuf) + ); + assert_eq!( + OtlpProtocol::from_config_str("HTTP/PROTOBUF"), + Some(OtlpProtocol::HttpProtobuf) + ); + assert_eq!( + OtlpProtocol::from_config_str("http/json"), + Some(OtlpProtocol::HttpJson) + ); +} + +#[test] +fn protocol_returns_none_for_unknown_or_empty() { + assert_eq!(OtlpProtocol::from_config_str(""), None); + assert_eq!(OtlpProtocol::from_config_str("thrift"), None); +} + +#[test] +fn temporality_parses_case_insensitively_and_defaults_to_delta() { + assert_eq!(Temporality::from_config_str("delta"), Temporality::Delta); + assert_eq!(Temporality::from_config_str("DELTA"), Temporality::Delta); + assert_eq!( + Temporality::from_config_str("Cumulative"), + Temporality::Cumulative + ); + assert_eq!( + Temporality::from_config_str(" CUMULATIVE "), + Temporality::Cumulative + ); + // Empty and unknown default to Delta. + assert_eq!(Temporality::from_config_str(""), Temporality::Delta); + assert_eq!( + Temporality::from_config_str("lowmemory"), + Temporality::Delta + ); +} + +#[test] +fn headers_parse_key_value_pairs() { + assert_eq!( + parse_otlp_headers("k1=v1,k2=v2"), + vec![ + ("k1".to_string(), "v1".to_string()), + ("k2".to_string(), "v2".to_string()) + ] + ); +} + +#[test] +fn headers_trim_and_skip_malformed_entries() { + assert_eq!( + parse_otlp_headers(" api-key = secret , , novalue , =dangling , k=v "), + vec![ + ("api-key".to_string(), "secret".to_string()), + ("k".to_string(), "v".to_string()) + ] + ); + assert!(parse_otlp_headers("").is_empty()); +}