Add bounded buffering for global operation metrics (#47619)

## What changed

Add `record_global_operation` to record an operation count and duration in milliseconds, buffering up to 256 pairs until a metrics provider is installed. Validate both metric names and tags before accepting a pair, limit each operation to 16 tags and 1024 bytes of metadata, and drop whole pairs when the buffer is full.

Drain pending operations on installation and serialize recording with provider lifecycle changes so a pair stays with one installation. Resume buffering when the active provider shuts down without disturbing a newer installation. Disabling metrics discards pending operations and suppresses recording until metrics are enabled again.

## Testing

Add unit and integration coverage for buffer limits, validation, concurrent installation and replacement, redirected global handles, shutdown buffering, and accepted or rejected opt-out configuration changes.

GitOrigin-RevId: 3b094e69bf975016ff1fb8f99b3611c2ccddccf3

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<str>,
+    duration_name: Box<str>,
+    duration: Duration,
+    tags: Vec<(Box<str>, Box<str>)>,
+}
+
+enum State {
+    Startup(Vec<PendingOperation>),
+    Ready(MetricsClient),
+    Disabled,
+}
+
+pub(crate) struct BufferedMetrics {
+    state: Mutex<State>,
+}
+
+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(&current.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;

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::<u64>();
+            }
+            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));
+}

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()
     }
 }

codex-rs/otel/src/metrics/error.rs

@@ -4,6 +4,9 @@ pub type Result<T> = std::result::Result<T, MetricsError>;
 
 #[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,

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
 }
 

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))
     }

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::<u64>();
+            }
+            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();
+    }
+}

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;