diff --git a/codex-rs/otel/src/metrics/buffered.rs b/codex-rs/otel/src/metrics/buffered.rs new file mode 100644 index 0000000000..96d6e31bc6 --- /dev/null +++ b/codex-rs/otel/src/metrics/buffered.rs @@ -0,0 +1,177 @@ +//! Buffers bounded operation counts and durations until process metrics are configured. +//! Each pair is admitted together; lifecycle transitions serialize with local SDK recording. +//! Shutting down the active installation resumes buffering until its replacement is ready. + +use super::MetricsClient; +use super::MetricsError; +use super::Result; +use super::validation::validate_metric_name; +use super::validation::validate_tag_key; +use super::validation::validate_tag_value; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +const MAX_PENDING_OPERATIONS: usize = 256; +const MAX_METADATA_BYTES: usize = 1024; +const MAX_TAGS: usize = 16; + +struct PendingOperation { + count_name: Box, + duration_name: Box, + duration: Duration, + tags: Vec<(Box, Box)>, +} + +enum State { + Startup(Vec), + Ready(MetricsClient), + Disabled, +} + +pub(crate) struct BufferedMetrics { + state: Mutex, +} + +pub(crate) static GLOBAL: BufferedMetrics = BufferedMetrics::new(); + +/// Records one operation count and its duration in milliseconds through `MetricsClient`. +/// While awaiting an active provider, retains at most 256 pairs, each with at most 16 tags and +/// 1024 bytes of metric names, tag keys, and tag values. Overflow drops the entire pair. +/// Supply bounded classifications, never credentials or user-provided identifiers. +/// Disabling metrics discards pending pairs and suppresses subsequent recording. +pub fn record_global_operation( + count_name: &str, + duration_name: &str, + duration: Duration, + tags: &[(&str, &str)], +) -> Result<()> { + GLOBAL.record(count_name, duration_name, duration, tags) +} + +impl BufferedMetrics { + const fn new() -> Self { + Self { + state: Mutex::new(State::Startup(Vec::new())), + } + } + + fn record( + &self, + count_name: &str, + duration_name: &str, + duration: Duration, + tags: &[(&str, &str)], + ) -> Result<()> { + if tags.len() > MAX_TAGS + || tags.iter().fold( + count_name.len().saturating_add(duration_name.len()), + |bytes, (key, value)| bytes.saturating_add(key.len()).saturating_add(value.len()), + ) > MAX_METADATA_BYTES + { + return Err(MetricsError::OperationMetadataTooLarge); + } + // Validate both instruments before accepting either half of the pair. + for name in [count_name, duration_name] { + validate_metric_name(name)?; + // The SDK otherwise silently builds a no-op instrument for these names. + if name.len() > 255 || !name.as_bytes()[0].is_ascii_alphabetic() { + return Err(MetricsError::InvalidMetricName { + name: name.to_owned(), + }); + } + } + for (key, value) in tags { + validate_tag_key(key)?; + validate_tag_value(value)?; + } + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + match &mut *state { + State::Startup(pending) => { + if pending.len() < MAX_PENDING_OPERATIONS { + pending.push(PendingOperation { + count_name: count_name.into(), + duration_name: duration_name.into(), + duration, + tags: tags + .iter() + .map(|(key, value)| ((*key).into(), (*value).into())) + .collect(), + }); + } + Ok(()) + } + State::Ready(metrics) => record(metrics, count_name, duration_name, duration, tags), + State::Disabled => Ok(()), + } + } + + pub(crate) fn enable(&self, metrics: &MetricsClient) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // Bind to this installation so a provider replacement cannot split a pair + // between exporters through the redirectable global MetricsClient handle. + let metrics = MetricsClient { + inner: Arc::clone(&metrics.inner), + active: None, + }; + if let State::Startup(pending) = &mut *state { + for operation in pending.drain(..) { + let tags: Vec<_> = operation + .tags + .iter() + .map(|(key, value)| (key.as_ref(), value.as_ref())) + .collect(); + let _ = record( + &metrics, + &operation.count_name, + &operation.duration_name, + operation.duration, + &tags, + ); + } + } + *state = State::Ready(metrics); + } + + pub(crate) fn disable(&self) { + *self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = State::Disabled; + } + + pub(super) fn suspend(&self, metrics: &MetricsClient) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // A replaced provider may shut down after its successor is installed. + // It must neither detach that successor nor undo an explicit opt-out. + if let State::Ready(current) = &*state + && Arc::ptr_eq(¤t.inner, &metrics.inner) + { + *state = State::Startup(Vec::new()); + } + } +} + +fn record( + metrics: &MetricsClient, + count_name: &str, + duration_name: &str, + duration: Duration, + tags: &[(&str, &str)], +) -> Result<()> { + metrics.counter(count_name, /*inc*/ 1, tags)?; + metrics.record_duration(duration_name, duration, tags) +} + +#[cfg(test)] +#[path = "buffered_tests.rs"] +mod tests; diff --git a/codex-rs/otel/src/metrics/buffered_tests.rs b/codex-rs/otel/src/metrics/buffered_tests.rs new file mode 100644 index 0000000000..3fc1c2c630 --- /dev/null +++ b/codex-rs/otel/src/metrics/buffered_tests.rs @@ -0,0 +1,175 @@ +use super::*; +use crate::MetricsConfig; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; +use pretty_assertions::assert_eq; +use std::sync::Barrier; +use std::sync::RwLock; + +fn client() -> MetricsClient { + MetricsClient::new( + MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(), + ) + .unwrap() +} + +fn observe(buffer: &BufferedMetrics) { + buffer + .record( + "test.operations", + "test.duration", + Duration::from_millis(/*millis*/ 7), + &[("component", "test")], + ) + .unwrap(); +} + +fn totals(metrics: &MetricsClient) -> (u64, u64, f64) { + let snapshot = metrics.snapshot().unwrap(); + let mut totals = (0, 0, 0.0); + for metric in snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + { + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + totals.0 += sum + .data_points() + .map(opentelemetry_sdk::metrics::data::SumDataPoint::value) + .sum::(); + } + AggregatedMetrics::F64(MetricData::Histogram(histogram)) => { + for point in histogram.data_points() { + totals.1 += point.count(); + totals.2 += point.sum(); + } + } + _ => panic!("unexpected metric"), + } + } + totals +} + +#[test] +fn startup_limits_and_validation_preserve_whole_pairs() { + let buffer = BufferedMetrics::new(); + let oversized = "x".repeat(MAX_METADATA_BYTES + 1); + assert!(matches!( + buffer.record("count", "duration", Duration::ZERO, &[("tag", &oversized)]), + Err(MetricsError::OperationMetadataTooLarge) + )); + assert!(matches!( + buffer.record( + "count", + "duration", + Duration::ZERO, + &[("tag", "value"); MAX_TAGS + 1] + ), + Err(MetricsError::OperationMetadataTooLarge) + )); + for name in ["invalid duration", "1duration", &"d".repeat(256)] { + assert!(buffer.record("count", name, Duration::ZERO, &[]).is_err()); + assert!( + buffer + .record(name, "duration", Duration::ZERO, &[]) + .is_err() + ); + } + assert!( + buffer + .record( + "count", + "duration", + Duration::ZERO, + &[("tag", "invalid value")] + ) + .is_err() + ); + for _ in 0..MAX_PENDING_OPERATIONS + 1 { + observe(&buffer); + } + let metrics = client(); + buffer.enable(&metrics); + assert_eq!(totals(&metrics), (256, 256, 1792.0)); + buffer.enable(&metrics); + assert_eq!(totals(&metrics), (0, 0, 0.0)); + for name in ["invalid duration", "1duration", &"d".repeat(256)] { + assert!(buffer.record("count", name, Duration::ZERO, &[]).is_err()); + assert!( + buffer + .record(name, "duration", Duration::ZERO, &[]) + .is_err() + ); + } + assert_eq!(totals(&metrics), (0, 0, 0.0)); +} + +#[test] +fn opt_out_discards_pending_and_live_observations_until_reenabled() { + let buffer = BufferedMetrics::new(); + observe(&buffer); + buffer.disable(); + observe(&buffer); + let first = client(); + buffer.enable(&first); + assert_eq!(totals(&first), (0, 0, 0.0)); + observe(&buffer); + assert_eq!(totals(&first), (1, 1, 7.0)); + buffer.disable(); + observe(&buffer); + assert_eq!(totals(&first), (0, 0, 0.0)); + let second = client(); + buffer.enable(&second); + observe(&buffer); + assert_eq!(totals(&second), (1, 1, 7.0)); + assert_eq!(totals(&first), (0, 0, 0.0)); +} + +#[test] +fn concurrent_installation_and_replacement_record_each_pair_once() { + let buffer = BufferedMetrics::new(); + let first = client(); + let second = client(); + let barrier = Barrier::new(/*n*/ 5); + observe(&buffer); + std::thread::scope(|scope| { + for _ in 0..4 { + scope.spawn(|| { + barrier.wait(); + for _ in 0..16 { + observe(&buffer); + } + }); + } + barrier.wait(); + buffer.enable(&first); + buffer.enable(&second); + }); + let a = totals(&first); + let b = totals(&second); + assert_eq!((a.0 + b.0, a.1 + b.1, a.2 + b.2), (65, 65, 455.0)); + assert_eq!((a.0, b.0), (a.1, b.1)); + buffer.enable(&second); + assert_eq!(totals(&second), (0, 0, 0.0)); +} + +#[test] +fn redirecting_a_global_handle_does_not_change_the_buffers_installation() { + let buffer = BufferedMetrics::new(); + let first = client(); + let second = client(); + let active = Arc::new(RwLock::new(Arc::clone(&first.inner))); + let redirected = MetricsClient { + inner: Arc::clone(&first.inner), + active: Some(Arc::clone(&active)), + }; + buffer.enable(&redirected); + *active.write().unwrap() = Arc::clone(&second.inner); + observe(&buffer); + assert_eq!(totals(&first), (1, 1, 7.0)); + assert_eq!(totals(&second), (0, 0, 0.0)); + buffer.enable(&second); + observe(&buffer); + assert_eq!(totals(&first), (0, 0, 0.0)); + assert_eq!(totals(&second), (1, 1, 7.0)); +} diff --git a/codex-rs/otel/src/metrics/client.rs b/codex-rs/otel/src/metrics/client.rs index c8894c8b6e..ca7fa74393 100644 --- a/codex-rs/otel/src/metrics/client.rs +++ b/codex-rs/otel/src/metrics/client.rs @@ -543,6 +543,7 @@ impl MetricsClient { /// Flush metrics and stop the underlying OTEL meter provider. pub fn shutdown(&self) -> Result<()> { + super::buffered::GLOBAL.suspend(self); self.inner.shutdown() } } diff --git a/codex-rs/otel/src/metrics/error.rs b/codex-rs/otel/src/metrics/error.rs index db6aba157a..7312992b1b 100644 --- a/codex-rs/otel/src/metrics/error.rs +++ b/codex-rs/otel/src/metrics/error.rs @@ -4,6 +4,9 @@ pub type Result = std::result::Result; #[derive(Debug, Error)] pub enum MetricsError { + #[error("operation metrics exceed the metadata size or tag count limit")] + OperationMetadataTooLarge, + // Metrics. #[error("metric name cannot be empty")] EmptyMetricName, diff --git a/codex-rs/otel/src/metrics/mod.rs b/codex-rs/otel/src/metrics/mod.rs index 32660753ac..c4f9d3155d 100644 --- a/codex-rs/otel/src/metrics/mod.rs +++ b/codex-rs/otel/src/metrics/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod buffered; mod client; mod config; mod error; @@ -9,6 +10,7 @@ pub(crate) mod timer; pub(crate) mod validation; use crate::config::StatsigMetricsSettings; +pub use crate::metrics::buffered::record_global_operation; pub use crate::metrics::client::MetricsClient; pub use crate::metrics::config::MetricsConfig; pub use crate::metrics::config::MetricsExporter; @@ -36,6 +38,7 @@ pub(crate) fn install_global(mut metrics: MetricsClient) -> MetricsClient { .unwrap_or_else(std::sync::PoisonError::into_inner) = Arc::clone(&metrics.inner); metrics.active = Some(active); let _ = GLOBAL_METRICS.set(metrics.clone()); + buffered::GLOBAL.enable(&metrics); metrics } diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index 6d35b555a4..0b2aeb58cd 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -201,6 +201,7 @@ impl OtelProvider { // Tracestate propagation is process-global; clear it when these // settings do not install an active provider. crate::trace_context::set_tracestate_entries(BTreeMap::new())?; + crate::metrics::buffered::GLOBAL.disable(); debug!("No OTEL exporter enabled in settings."); return Ok(None); } @@ -279,6 +280,8 @@ impl OtelProvider { environment: settings.environment.clone(), }); } + } else { + crate::metrics::buffered::GLOBAL.disable(); } Ok(Some(provider)) } diff --git a/codex-rs/otel/tests/suite/buffered_operations_tests.rs b/codex-rs/otel/tests/suite/buffered_operations_tests.rs new file mode 100644 index 0000000000..50789799d1 --- /dev/null +++ b/codex-rs/otel/tests/suite/buffered_operations_tests.rs @@ -0,0 +1,125 @@ +//! Exercises global operation buffering across shutdown and rejected configuration changes. + +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; +use codex_otel::OtelExporter; +use codex_otel::OtelHttpProtocol; +use codex_otel::OtelProvider; +use codex_otel::OtelSettings; +use codex_otel::install_global_metrics; +use codex_otel::record_global_operation; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use std::time::Duration; + +fn install() -> MetricsClient { + install_global_metrics( + MetricsClient::new( + MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(), + ) + .expect("build in-memory metrics"), + ) +} + +fn observe() { + record_global_operation( + "test.operations", + "test.duration", + Duration::from_millis(/*millis*/ 7), + &[], + ) + .expect("record an operation"); +} + +fn totals(metrics: &MetricsClient) -> (u64, u64, f64) { + let snapshot = metrics.snapshot().expect("collect runtime metrics"); + let mut totals = (0, 0, 0.0); + for metric in snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + { + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + totals.0 += sum + .data_points() + .map(opentelemetry_sdk::metrics::data::SumDataPoint::value) + .sum::(); + } + AggregatedMetrics::F64(MetricData::Histogram(histogram)) => { + for point in histogram.data_points() { + totals.1 += point.count(); + totals.2 += point.sum(); + } + } + _ => panic!("unexpected metric"), + } + } + totals +} + +#[test] +fn shutdown_buffers_until_replacement_and_preserves_newer_installations() { + let first = install(); + first.shutdown().unwrap(); + observe(); + let second = install(); + assert_eq!(totals(&second), (1, 1, 7.0)); + + let third = install(); + second.shutdown().unwrap(); + observe(); + assert_eq!(totals(&third), (1, 1, 7.0)); + third.shutdown().unwrap(); +} + +#[test] +fn rejected_opt_out_preserves_recording_and_accepted_opt_out_survives_shutdown() { + let metrics = install(); + let mut settings = OtelSettings { + http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + environment: "test".to_string(), + service_name: "test".to_string(), + service_version: "1".to_string(), + codex_home: ".".into(), + exporter: OtelExporter::None, + trace_exporter: OtelExporter::OtlpHttp { + endpoint: "http://127.0.0.1:1/v1/traces".to_string(), + headers: Default::default(), + protocol: OtelHttpProtocol::Json, + tls: None, + }, + metrics_exporter: OtelExporter::None, + runtime_metrics: false, + span_attributes: BTreeMap::from([(String::new(), "invalid".to_string())]), + tracestate: BTreeMap::new(), + }; + assert_eq!( + OtelProvider::try_new(&settings).err().unwrap().to_string(), + "configured span attribute key must not be empty" + ); + observe(); + assert_eq!(totals(&metrics), (1, 1, 7.0)); + + metrics.shutdown().unwrap(); + observe(); + assert!(OtelProvider::try_new(&settings).is_err()); + let replacement = install(); + assert_eq!(totals(&replacement), (1, 1, 7.0)); + replacement.shutdown().unwrap(); + + settings.span_attributes.clear(); + for trace_exporter in [settings.trace_exporter.clone(), OtelExporter::None] { + let metrics = install(); + settings.trace_exporter = trace_exporter; + let _provider = OtelProvider::try_new(&settings).unwrap(); + metrics.shutdown().unwrap(); + observe(); + let replacement = install(); + assert_eq!(totals(&replacement), (0, 0, 0.0)); + replacement.shutdown().unwrap(); + } +} diff --git a/codex-rs/otel/tests/suite/mod.rs b/codex-rs/otel/tests/suite/mod.rs index 79ee1e475e..2422d083b2 100644 --- a/codex-rs/otel/tests/suite/mod.rs +++ b/codex-rs/otel/tests/suite/mod.rs @@ -1,3 +1,5 @@ +#[path = "buffered_operations_tests.rs"] +mod buffered_operations; mod manager_metrics; mod otel_export_routing_policy; mod otlp_http_loopback;