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 diff --git a/Cargo.lock b/Cargo.lock index 65ac3a4fc4..1ad96674b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3189,6 +3189,19 @@ dependencies = [ "libdd-log", ] +[[package]] +name = "libdd-otel-telemetry" +version = "0.1.0" +dependencies = [ + "libdd-shared-runtime", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "rustls", + "tokio", + "tracing", +] + [[package]] name = "libdd-otel-thread-ctx" version = "1.0.0" @@ -4158,6 +4171,81 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.17", + "tracing", +] + +[[package]] +name = "opentelemetry-http" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest", + "thiserror 2.0.17", + "tokio", + "tonic", + "tonic-types", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.0", + "thiserror 2.0.17", +] + [[package]] name = "os_info" version = "3.14.0" @@ -6162,6 +6250,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/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..e901f480d0 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" @@ -453,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 new file mode 100644 index 0000000000..891e489c52 --- /dev/null +++ b/libdd-otel-telemetry/Cargo.toml @@ -0,0 +1,35 @@ +# 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.32" +opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = [ + "metrics", +] } +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", "dep:rustls"] +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..81dd12158d --- /dev/null +++ b/libdd-otel-telemetry/src/aggregator.rs @@ -0,0 +1,379 @@ +// 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, OtelMetricsError}; +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. +/// +/// 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, + pub metrics_export_successes: u64, + pub metrics_export_failures: u64, +} + +#[derive(Debug, Default)] +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 { + Counter(Counter), + UpDownCounter(UpDownCounter), + Histogram(Histogram), + Gauge(Gauge), +} + +/// 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 OtelMetricsAggregatorBuilder { + resource: Resource, + metrics_exporter: Option, + temporality: Temporality, + export_interval: Duration, +} + +impl Default for OtelMetricsAggregatorBuilder { + fn default() -> Self { + Self { + resource: Resource::builder().build(), + metrics_exporter: None, + temporality: Temporality::default(), + export_interval: Duration::from_secs(60), + } + } +} + +impl OtelMetricsAggregatorBuilder { + 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, + ) -> (OtelMetricsAggregator, 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 aggregator = OtelMetricsAggregator { + provider, + meters: Mutex::new(HashMap::new()), + instruments: Mutex::new(HashMap::new()), + next_id: AtomicU64::new(1), + counters: Arc::new(Counters::default()), + }; + (aggregator, warnings) + } +} + +pub(crate) 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 => { + // 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())) +} + +/// 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 +/// 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 OtelMetricsAggregator { + provider: SdkMeterProvider, + meters: Mutex>, + instruments: Mutex>, + next_id: AtomicU64, + counters: Arc, +} + +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); + self.instruments + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(id, handle); + 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 = 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 = 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 = 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 = 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 { + self.counters.snapshot() + } + + pub fn force_flush(&self) -> Result<(), OtelMetricsError> { + self.provider + .force_flush() + .map_err(|e| OtelMetricsError(e.to_string())) + } + + pub fn shutdown(self) -> Result<(), OtelMetricsError> { + self.provider + .shutdown() + .map_err(|e| OtelMetricsError(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..df8c151d7a --- /dev/null +++ b/libdd-otel-telemetry/src/config.rs @@ -0,0 +1,115 @@ +// 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, + /// 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. +/// +/// 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 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 { + 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..840646386f --- /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::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 +/// 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::OtelMetricsAggregator`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OtelMetricsError(pub(crate) String); + +impl fmt::Display for OtelMetricsError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl std::error::Error for 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/instrument.rs b/libdd-otel-telemetry/src/instrument.rs new file mode 100644 index 0000000000..372507aaec --- /dev/null +++ b/libdd-otel-telemetry/src/instrument.rs @@ -0,0 +1,79 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +/// 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. +#[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. +/// +/// 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 { + pub fn new(name: impl Into, kind: InstrumentKind) -> Self { + Self { + name: name.into(), + kind, + unit: None, + description: None, + meter_name: "libdd-otel-telemetry".to_string(), + meter_version: None, + meter_schema_url: 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 + } + + /// 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 + } +} diff --git a/libdd-otel-telemetry/src/lib.rs b/libdd-otel-telemetry/src/lib.rs new file mode 100644 index 0000000000..15f0197786 --- /dev/null +++ b/libdd-otel-telemetry/src/lib.rs @@ -0,0 +1,51 @@ +// 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 +//! +//! [`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`): +//! 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; +#[cfg(any(feature = "grpc", feature = "http"))] +mod exporter; +mod instrument; +mod resource; + +pub use aggregator::{ExportCounters, OtelMetricsAggregator, OtelMetricsAggregatorBuilder}; +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}; +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..dfd76a9553 --- /dev/null +++ b/libdd-otel-telemetry/tests/basic.rs @@ -0,0 +1,90 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +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) = OtelMetricsAggregatorBuilder::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"); +} + +// 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() { + 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_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()); +}