feat: metrics capabilities (#8318)

Add metrics capabilities to Codex. The `README.md` is up to date.

This will not be merged with the metrics before this PR of course:
https://github.com/openai/codex/pull/8350
This commit is contained in:
jif-oai authored and GitHub committed 2026-01-08 11:47:36 +00:00
1 parent 8a0c2e5841
commit 634650dd25
41 files changed
+1946 -228

No files matched your search

+30
View File
@@ -3,6 +3,31 @@ use std::path::PathBuf;
use codex_utils_absolute_path::AbsolutePathBuf;
pub(crate) const STATSIG_OTLP_HTTP_ENDPOINT: &str = "https://ab.chatgpt.com/otlp/v1/metrics";
pub(crate) const STATSIG_API_KEY_HEADER: &str = "statsig-api-key";
pub(crate) const STATSIG_API_KEY: &str = "client-MkRuleRQBd6qakfnDYqJVR9JuXcY57Ljly3vi5JVUIO";
pub(crate) fn resolve_exporter(exporter: &OtelExporter) -> OtelExporter {
match exporter {
OtelExporter::Statsig => {
if cfg!(test) || cfg!(feature = "disable-default-metrics-exporter") {
return OtelExporter::None;
}
OtelExporter::OtlpHttp {
endpoint: STATSIG_OTLP_HTTP_ENDPOINT.to_string(),
headers: HashMap::from([(
STATSIG_API_KEY_HEADER.to_string(),
STATSIG_API_KEY.to_string(),
)]),
protocol: OtelHttpProtocol::Json,
tls: None,
}
}
_ => exporter.clone(),
}
}
#[derive(Clone, Debug)]
pub struct OtelSettings {
pub environment: String,
@@ -11,6 +36,7 @@ pub struct OtelSettings {
pub codex_home: PathBuf,
pub exporter: OtelExporter,
pub trace_exporter: OtelExporter,
pub metrics_exporter: OtelExporter,
}
#[derive(Clone, Debug)]
@@ -31,6 +57,10 @@ pub struct OtelTlsConfig {
#[derive(Clone, Debug)]
pub enum OtelExporter {
None,
/// Statsig metrics ingestion exporter using Codex-internal defaults.
///
/// This is intended for metrics only.
Statsig,
OtlpGrpc {
endpoint: String,
headers: HashMap<String, String>,
+171 -2
View File
@@ -1,4 +1,173 @@
pub mod config;
pub mod metrics;
pub mod traces;
pub mod otel_manager;
pub mod otel_provider;
mod otlp;
use crate::metrics::MetricsClient;
use crate::metrics::MetricsConfig;
use crate::metrics::MetricsError;
use crate::metrics::Result as MetricsResult;
use crate::metrics::timer::Timer;
use crate::metrics::validation::validate_tag_key;
use crate::metrics::validation::validate_tag_value;
use crate::traces::otel_provider::OtelProvider;
use codex_protocol::ThreadId;
use serde::Serialize;
use std::time::Duration;
use strum_macros::Display;
use tracing::Span;
#[derive(Debug, Clone, Serialize, Display)]
#[serde(rename_all = "snake_case")]
pub enum ToolDecisionSource {
Config,
User,
}
#[derive(Debug, Clone)]
pub struct OtelEventMetadata {
pub(crate) conversation_id: ThreadId,
pub(crate) auth_mode: Option<String>,
pub(crate) account_id: Option<String>,
pub(crate) account_email: Option<String>,
pub(crate) model: String,
pub(crate) slug: String,
pub(crate) log_user_prompts: bool,
pub(crate) app_version: &'static str,
pub(crate) terminal_type: String,
}
#[derive(Debug, Clone)]
pub struct OtelManager {
pub(crate) metadata: OtelEventMetadata,
pub(crate) session_span: Span,
pub(crate) metrics: Option<MetricsClient>,
pub(crate) metrics_use_metadata_tags: bool,
}
impl OtelManager {
pub fn with_model(mut self, model: &str, slug: &str) -> Self {
self.metadata.model = model.to_owned();
self.metadata.slug = slug.to_owned();
self
}
pub fn with_metrics(mut self, metrics: MetricsClient) -> Self {
self.metrics = Some(metrics);
self.metrics_use_metadata_tags = true;
self
}
pub fn with_metrics_without_metadata_tags(mut self, metrics: MetricsClient) -> Self {
self.metrics = Some(metrics);
self.metrics_use_metadata_tags = false;
self
}
pub fn with_metrics_config(self, config: MetricsConfig) -> MetricsResult<Self> {
let metrics = MetricsClient::new(config)?;
Ok(self.with_metrics(metrics))
}
pub fn with_provider_metrics(self, provider: &OtelProvider) -> Self {
match provider.metrics() {
Some(metrics) => self.with_metrics(metrics.clone()),
None => self,
}
}
pub fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) {
let res: MetricsResult<()> = (|| {
let Some(metrics) = &self.metrics else {
return Ok(());
};
let tags = self.tags_with_metadata(tags)?;
metrics.counter(name, inc, &tags)
})();
if let Err(e) = res {
tracing::warn!("metrics counter [{name}] failed: {e}");
}
}
pub fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) {
let res: MetricsResult<()> = (|| {
let Some(metrics) = &self.metrics else {
return Ok(());
};
let tags = self.tags_with_metadata(tags)?;
metrics.histogram(name, value, &tags)
})();
if let Err(e) = res {
tracing::warn!("metrics histogram [{name}] failed: {e}");
}
}
pub fn record_duration(&self, name: &str, duration: Duration, tags: &[(&str, &str)]) {
let res: MetricsResult<()> = (|| {
let Some(metrics) = &self.metrics else {
return Ok(());
};
let tags = self.tags_with_metadata(tags)?;
metrics.record_duration(name, duration, &tags)
})();
if let Err(e) = res {
tracing::warn!("metrics duration [{name}] failed: {e}");
}
}
pub fn start_timer(&self, name: &str, tags: &[(&str, &str)]) -> Result<Timer, MetricsError> {
let Some(metrics) = &self.metrics else {
return Err(MetricsError::ExporterDisabled);
};
let tags = self.tags_with_metadata(tags)?;
metrics.start_timer(name, &tags)
}
pub fn shutdown_metrics(&self) -> MetricsResult<()> {
let Some(metrics) = &self.metrics else {
return Ok(());
};
metrics.shutdown()
}
fn tags_with_metadata<'a>(
&'a self,
tags: &'a [(&'a str, &'a str)],
) -> MetricsResult<Vec<(&'a str, &'a str)>> {
let mut merged = self.metadata_tag_refs()?;
merged.extend(tags.iter().copied());
Ok(merged)
}
fn metadata_tag_refs(&self) -> MetricsResult<Vec<(&str, &str)>> {
if !self.metrics_use_metadata_tags {
return Ok(Vec::new());
}
let mut tags = Vec::with_capacity(5);
Self::push_metadata_tag(&mut tags, "auth_mode", self.metadata.auth_mode.as_deref())?;
Self::push_metadata_tag(&mut tags, "model", Some(self.metadata.model.as_str()))?;
Self::push_metadata_tag(&mut tags, "app.version", Some(self.metadata.app_version))?;
Ok(tags)
}
fn push_metadata_tag<'a>(
tags: &mut Vec<(&'a str, &'a str)>,
key: &'static str,
value: Option<&'a str>,
) -> MetricsResult<()> {
let Some(value) = value else {
return Ok(());
};
validate_tag_key(key)?;
validate_tag_value(value)?;
tags.push((key, value));
Ok(())
}
}
+291
View File
@@ -0,0 +1,291 @@
use crate::config::OtelExporter;
use crate::config::OtelHttpProtocol;
use crate::metrics::MetricsError;
use crate::metrics::Result;
use crate::metrics::config::MetricsConfig;
use crate::metrics::config::MetricsExporter;
use crate::metrics::timer::Timer;
use crate::metrics::validation::validate_metric_name;
use crate::metrics::validation::validate_tag_key;
use crate::metrics::validation::validate_tag_value;
use crate::metrics::validation::validate_tags;
use opentelemetry::KeyValue;
use opentelemetry::metrics::Counter;
use opentelemetry::metrics::Histogram;
use opentelemetry::metrics::Meter;
use opentelemetry::metrics::MeterProvider as _;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_METRICS_TIMEOUT;
use opentelemetry_otlp::Protocol;
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_otlp::WithHttpConfig;
use opentelemetry_otlp::WithTonicConfig;
use opentelemetry_otlp::tonic_types::metadata::MetadataMap;
use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::PeriodicReader;
use opentelemetry_sdk::metrics::SdkMeterProvider;
use opentelemetry_sdk::metrics::Temporality;
use opentelemetry_semantic_conventions as semconv;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::sync::Mutex;
use std::time::Duration;
use tracing::debug;
const ENV_ATTRIBUTE: &str = "env";
const METER_NAME: &str = "codex";
#[derive(Debug)]
struct MetricsClientInner {
meter_provider: SdkMeterProvider,
meter: Meter,
counters: Mutex<HashMap<String, Counter<u64>>>,
histograms: Mutex<HashMap<String, Histogram<f64>>>,
default_tags: BTreeMap<String, String>,
}
impl MetricsClientInner {
fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) -> Result<()> {
validate_metric_name(name)?;
if inc < 0 {
return Err(MetricsError::NegativeCounterIncrement {
name: name.to_string(),
inc,
});
}
let attributes = self.attributes(tags)?;
let mut counters = self
.counters
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let counter = counters
.entry(name.to_string())
.or_insert_with(|| self.meter.u64_counter(name.to_string()).build());
counter.add(inc as u64, &attributes);
Ok(())
}
fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) -> Result<()> {
validate_metric_name(name)?;
let attributes = self.attributes(tags)?;
let mut histograms = self
.histograms
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let histogram = histograms
.entry(name.to_string())
.or_insert_with(|| self.meter.f64_histogram(name.to_string()).build());
histogram.record(value as f64, &attributes);
Ok(())
}
fn attributes(&self, tags: &[(&str, &str)]) -> Result<Vec<KeyValue>> {
if tags.is_empty() {
return Ok(self
.default_tags
.iter()
.map(|(key, value)| KeyValue::new(key.clone(), value.clone()))
.collect());
}
let mut merged = self.default_tags.clone();
for (key, value) in tags {
validate_tag_key(key)?;
validate_tag_value(value)?;
merged.insert((*key).to_string(), (*value).to_string());
}
Ok(merged
.into_iter()
.map(|(key, value)| KeyValue::new(key, value))
.collect())
}
fn shutdown(&self) -> Result<()> {
debug!("flushing OTEL metrics");
self.meter_provider
.force_flush()
.map_err(|source| MetricsError::ProviderShutdown { source })?;
self.meter_provider
.shutdown()
.map_err(|source| MetricsError::ProviderShutdown { source })?;
Ok(())
}
}
/// OpenTelemetry metrics client used by Codex.
#[derive(Clone, Debug)]
pub struct MetricsClient(std::sync::Arc<MetricsClientInner>);
impl MetricsClient {
/// Build a metrics client from configuration and validate defaults.
pub fn new(config: MetricsConfig) -> Result<Self> {
validate_tags(&config.default_tags)?;
let resource = Resource::builder()
.with_service_name(config.service_name.clone())
.with_attributes(vec![
KeyValue::new(
semconv::attribute::SERVICE_VERSION,
config.service_version.clone(),
),
KeyValue::new(ENV_ATTRIBUTE, config.environment.clone()),
])
.build();
let temporality = Temporality::default();
let (meter_provider, meter) = match config.exporter {
MetricsExporter::InMemory(exporter) => {
build_provider(resource, exporter, config.export_interval)
}
MetricsExporter::Otlp(exporter) => {
let exporter = build_otlp_metric_exporter(exporter, temporality)?;
build_provider(resource, exporter, config.export_interval)
}
};
Ok(Self(std::sync::Arc::new(MetricsClientInner {
meter_provider,
meter,
counters: Mutex::new(HashMap::new()),
histograms: Mutex::new(HashMap::new()),
default_tags: config.default_tags,
})))
}
/// Send a single counter increment.
pub fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) -> Result<()> {
self.0.counter(name, inc, tags)
}
/// Send a single histogram sample.
pub fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) -> Result<()> {
self.0.histogram(name, value, tags)
}
/// Record a duration in milliseconds using a histogram.
pub fn record_duration(
&self,
name: &str,
duration: Duration,
tags: &[(&str, &str)],
) -> Result<()> {
self.histogram(
name,
duration.as_millis().min(i64::MAX as u128) as i64,
tags,
)
}
pub fn start_timer(
&self,
name: &str,
tags: &[(&str, &str)],
) -> std::result::Result<Timer, MetricsError> {
Ok(Timer::new(name, tags, self))
}
/// Flush metrics and stop the underlying OTEL meter provider.
pub fn shutdown(&self) -> Result<()> {
self.0.shutdown()
}
}
fn build_provider<E>(
resource: Resource,
exporter: E,
interval: Option<Duration>,
) -> (SdkMeterProvider, Meter)
where
E: opentelemetry_sdk::metrics::exporter::PushMetricExporter + 'static,
{
let mut reader_builder = PeriodicReader::builder(exporter);
if let Some(interval) = interval {
reader_builder = reader_builder.with_interval(interval);
}
let reader = reader_builder.build();
let provider = SdkMeterProvider::builder()
.with_resource(resource)
.with_reader(reader)
.build();
let meter = provider.meter(METER_NAME);
(provider, meter)
}
fn build_otlp_metric_exporter(
exporter: OtelExporter,
temporality: Temporality,
) -> Result<opentelemetry_otlp::MetricExporter> {
match exporter {
OtelExporter::None => Err(MetricsError::ExporterDisabled),
OtelExporter::Statsig => build_otlp_metric_exporter(
crate::config::resolve_exporter(&OtelExporter::Statsig),
temporality,
),
OtelExporter::OtlpGrpc {
endpoint,
headers,
tls,
} => {
debug!("Using OTLP Grpc exporter for metrics: {endpoint}");
let header_map = crate::otlp::build_header_map(&headers);
let base_tls_config = ClientTlsConfig::new()
.with_enabled_roots()
.assume_http2(true);
let tls_config = match tls.as_ref() {
Some(tls) => crate::otlp::build_grpc_tls_config(&endpoint, base_tls_config, tls)
.map_err(|err| MetricsError::InvalidConfig {
message: err.to_string(),
})?,
None => base_tls_config,
};
opentelemetry_otlp::MetricExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.with_temporality(temporality)
.with_metadata(MetadataMap::from_headers(header_map))
.with_tls_config(tls_config)
.build()
.map_err(|source| MetricsError::ExporterBuild { source })
}
OtelExporter::OtlpHttp {
endpoint,
headers,
protocol,
tls,
} => {
debug!("Using OTLP Http exporter for metrics: {endpoint}");
let protocol = match protocol {
OtelHttpProtocol::Binary => Protocol::HttpBinary,
OtelHttpProtocol::Json => Protocol::HttpJson,
};
let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder()
.with_http()
.with_endpoint(endpoint)
.with_temporality(temporality)
.with_protocol(protocol)
.with_headers(headers);
if let Some(tls) = tls.as_ref() {
let client =
crate::otlp::build_http_client(tls, OTEL_EXPORTER_OTLP_METRICS_TIMEOUT)
.map_err(|err| MetricsError::InvalidConfig {
message: err.to_string(),
})?;
exporter_builder = exporter_builder.with_http_client(client);
}
exporter_builder
.build()
.map_err(|source| MetricsError::ExporterBuild { source })
}
}
}
+74
View File
@@ -0,0 +1,74 @@
use crate::config::OtelExporter;
use crate::metrics::Result;
use crate::metrics::validation::validate_tag_key;
use crate::metrics::validation::validate_tag_value;
use opentelemetry_sdk::metrics::InMemoryMetricExporter;
use std::collections::BTreeMap;
use std::time::Duration;
#[derive(Clone, Debug)]
pub enum MetricsExporter {
Otlp(OtelExporter),
InMemory(InMemoryMetricExporter),
}
#[derive(Clone, Debug)]
pub struct MetricsConfig {
pub(crate) environment: String,
pub(crate) service_name: String,
pub(crate) service_version: String,
pub(crate) exporter: MetricsExporter,
pub(crate) export_interval: Option<Duration>,
pub(crate) default_tags: BTreeMap<String, String>,
}
impl MetricsConfig {
pub fn otlp(
environment: impl Into<String>,
service_name: impl Into<String>,
service_version: impl Into<String>,
exporter: OtelExporter,
) -> Self {
Self {
environment: environment.into(),
service_name: service_name.into(),
service_version: service_version.into(),
exporter: MetricsExporter::Otlp(exporter),
export_interval: None,
default_tags: BTreeMap::new(),
}
}
/// Create an in-memory config (used in tests).
pub fn in_memory(
environment: impl Into<String>,
service_name: impl Into<String>,
service_version: impl Into<String>,
exporter: InMemoryMetricExporter,
) -> Self {
Self {
environment: environment.into(),
service_name: service_name.into(),
service_version: service_version.into(),
exporter: MetricsExporter::InMemory(exporter),
export_interval: None,
default_tags: BTreeMap::new(),
}
}
/// Override the interval between periodic metric exports.
pub fn with_export_interval(mut self, interval: Duration) -> Self {
self.export_interval = Some(interval);
self
}
/// Add a default tag that will be sent with every metric.
pub fn with_tag(mut self, key: impl Into<String>, value: impl Into<String>) -> Result<Self> {
let key = key.into();
let value = value.into();
validate_tag_key(&key)?;
validate_tag_value(&value)?;
self.default_tags.insert(key, value);
Ok(self)
}
}
+37
View File
@@ -0,0 +1,37 @@
use thiserror::Error;
pub type Result<T> = std::result::Result<T, MetricsError>;
#[derive(Debug, Error)]
pub enum MetricsError {
// Metrics.
#[error("metric name cannot be empty")]
EmptyMetricName,
#[error("metric name contains invalid characters: {name}")]
InvalidMetricName { name: String },
#[error("{label} cannot be empty")]
EmptyTagComponent { label: String },
#[error("{label} contains invalid characters: {value}")]
InvalidTagComponent { label: String, value: String },
#[error("metrics exporter is disabled")]
ExporterDisabled,
#[error("counter increment must be non-negative for {name}: {inc}")]
NegativeCounterIncrement { name: String, inc: i64 },
#[error("failed to build OTLP metrics exporter")]
ExporterBuild {
#[source]
source: opentelemetry_otlp::ExporterBuildError,
},
#[error("invalid OTLP metrics configuration: {message}")]
InvalidConfig { message: String },
#[error("failed to flush or shutdown metrics provider")]
ProviderShutdown {
#[source]
source: opentelemetry_sdk::error::OTelSdkError,
},
}
+22
View File
@@ -0,0 +1,22 @@
mod client;
mod config;
mod error;
pub(crate) mod timer;
pub(crate) mod validation;
pub use crate::metrics::client::MetricsClient;
pub use crate::metrics::config::MetricsConfig;
pub use crate::metrics::config::MetricsExporter;
pub use crate::metrics::error::MetricsError;
pub use crate::metrics::error::Result;
use std::sync::OnceLock;
static GLOBAL_METRICS: OnceLock<MetricsClient> = OnceLock::new();
pub(crate) fn install_global(metrics: MetricsClient) {
let _ = GLOBAL_METRICS.set(metrics);
}
pub(crate) fn global() -> Option<MetricsClient> {
GLOBAL_METRICS.get().cloned()
}
+42
View File
@@ -0,0 +1,42 @@
use crate::metrics::MetricsClient;
use crate::metrics::error::Result;
use std::time::Instant;
pub struct Timer {
name: String,
tags: Vec<(String, String)>,
client: MetricsClient,
start_time: Instant,
}
impl Drop for Timer {
fn drop(&mut self) {
if let Err(e) = self.record() {
tracing::error!("metrics client error: {}", e);
}
}
}
impl Timer {
pub(crate) fn new(name: &str, tags: &[(&str, &str)], client: &MetricsClient) -> Self {
Self {
name: name.to_string(),
tags: tags
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
client: client.clone(),
start_time: Instant::now(),
}
}
pub fn record(&self) -> Result<()> {
let tags = self
.tags
.iter()
.map(|(k, v)| (k.as_str(), v.as_str()))
.collect::<Vec<_>>();
self.client
.record_duration(&self.name, self.start_time.elapsed(), &tags)
}
}
+55
View File
@@ -0,0 +1,55 @@
use crate::metrics::error::MetricsError;
use crate::metrics::error::Result;
use std::collections::BTreeMap;
pub(crate) fn validate_tags(tags: &BTreeMap<String, String>) -> Result<()> {
for (key, value) in tags {
validate_tag_key(key)?;
validate_tag_value(value)?;
}
Ok(())
}
pub(crate) fn validate_metric_name(name: &str) -> Result<()> {
if name.is_empty() {
return Err(MetricsError::EmptyMetricName);
}
if !name.chars().all(is_metric_char) {
return Err(MetricsError::InvalidMetricName {
name: name.to_string(),
});
}
Ok(())
}
pub(crate) fn validate_tag_key(key: &str) -> Result<()> {
validate_tag_component(key, "tag key")?;
Ok(())
}
pub(crate) fn validate_tag_value(value: &str) -> Result<()> {
validate_tag_component(value, "tag value")
}
fn validate_tag_component(value: &str, label: &str) -> Result<()> {
if value.is_empty() {
return Err(MetricsError::EmptyTagComponent {
label: label.to_string(),
});
}
if !value.chars().all(is_tag_char) {
return Err(MetricsError::InvalidTagComponent {
label: label.to_string(),
value: value.to_string(),
});
}
Ok(())
}
fn is_metric_char(c: char) -> bool {
c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')
}
fn is_tag_char(c: char) -> bool {
c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-' | '/')
}
+163
View File
@@ -0,0 +1,163 @@
use crate::config::OtelTlsConfig;
use codex_utils_absolute_path::AbsolutePathBuf;
use http::Uri;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT;
use opentelemetry_otlp::tonic_types::transport::Certificate as TonicCertificate;
use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig;
use opentelemetry_otlp::tonic_types::transport::Identity as TonicIdentity;
use reqwest::Certificate as ReqwestCertificate;
use reqwest::Identity as ReqwestIdentity;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderName;
use reqwest::header::HeaderValue;
use std::env;
use std::error::Error;
use std::fs;
use std::io;
use std::io::ErrorKind;
use std::path::PathBuf;
use std::time::Duration;
pub(crate) fn build_header_map(headers: &std::collections::HashMap<String, String>) -> HeaderMap {
let mut header_map = HeaderMap::new();
for (key, value) in headers {
if let Ok(name) = HeaderName::from_bytes(key.as_bytes())
&& let Ok(val) = HeaderValue::from_str(value)
{
header_map.insert(name, val);
}
}
header_map
}
pub(crate) fn build_grpc_tls_config(
endpoint: &str,
tls_config: ClientTlsConfig,
tls: &OtelTlsConfig,
) -> Result<ClientTlsConfig, Box<dyn Error>> {
let uri: Uri = endpoint.parse()?;
let host = uri.host().ok_or_else(|| {
config_error(format!(
"OTLP gRPC endpoint {endpoint} does not include a host"
))
})?;
let mut config = tls_config.domain_name(host.to_owned());
if let Some(path) = tls.ca_certificate.as_ref() {
let (pem, _) = read_bytes(path)?;
config = config.ca_certificate(TonicCertificate::from_pem(pem));
}
match (&tls.client_certificate, &tls.client_private_key) {
(Some(cert_path), Some(key_path)) => {
let (cert_pem, _) = read_bytes(cert_path)?;
let (key_pem, _) = read_bytes(key_path)?;
config = config.identity(TonicIdentity::from_pem(cert_pem, key_pem));
}
(Some(_), None) | (None, Some(_)) => {
return Err(config_error(
"client_certificate and client_private_key must both be provided for mTLS",
));
}
(None, None) => {}
}
Ok(config)
}
/// Build a blocking HTTP client with TLS configuration for OTLP HTTP exporters.
///
/// We use `reqwest::blocking::Client` because OTEL exporters run on dedicated
/// OS threads that are not necessarily backed by tokio.
pub(crate) fn build_http_client(
tls: &OtelTlsConfig,
timeout_var: &str,
) -> Result<reqwest::blocking::Client, Box<dyn Error>> {
if tokio::runtime::Handle::try_current().is_ok() {
tokio::task::block_in_place(|| build_http_client_inner(tls, timeout_var))
} else {
build_http_client_inner(tls, timeout_var)
}
}
fn build_http_client_inner(
tls: &OtelTlsConfig,
timeout_var: &str,
) -> Result<reqwest::blocking::Client, Box<dyn Error>> {
let mut builder =
reqwest::blocking::Client::builder().timeout(resolve_otlp_timeout(timeout_var));
if let Some(path) = tls.ca_certificate.as_ref() {
let (pem, location) = read_bytes(path)?;
let certificate = ReqwestCertificate::from_pem(pem.as_slice()).map_err(|error| {
config_error(format!(
"failed to parse certificate {}: {error}",
location.display()
))
})?;
builder = builder
.tls_built_in_root_certs(false)
.add_root_certificate(certificate);
}
match (&tls.client_certificate, &tls.client_private_key) {
(Some(cert_path), Some(key_path)) => {
let (mut cert_pem, cert_location) = read_bytes(cert_path)?;
let (key_pem, key_location) = read_bytes(key_path)?;
cert_pem.extend_from_slice(key_pem.as_slice());
let identity = ReqwestIdentity::from_pem(cert_pem.as_slice()).map_err(|error| {
config_error(format!(
"failed to parse client identity using {} and {}: {error}",
cert_location.display(),
key_location.display()
))
})?;
builder = builder.identity(identity).https_only(true);
}
(Some(_), None) | (None, Some(_)) => {
return Err(config_error(
"client_certificate and client_private_key must both be provided for mTLS",
));
}
(None, None) => {}
}
builder
.build()
.map_err(|error| Box::new(error) as Box<dyn Error>)
}
pub(crate) fn resolve_otlp_timeout(signal_var: &str) -> Duration {
if let Some(timeout) = read_timeout_env(signal_var) {
return timeout;
}
if let Some(timeout) = read_timeout_env(OTEL_EXPORTER_OTLP_TIMEOUT) {
return timeout;
}
OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT
}
fn read_timeout_env(var: &str) -> Option<Duration> {
let value = env::var(var).ok()?;
let parsed = value.parse::<i64>().ok()?;
if parsed < 0 {
return None;
}
Some(Duration::from_millis(parsed as u64))
}
fn read_bytes(path: &AbsolutePathBuf) -> Result<(Vec<u8>, PathBuf), Box<dyn Error>> {
match fs::read(path) {
Ok(bytes) => Ok((bytes, path.to_path_buf())),
Err(error) => Err(Box::new(io::Error::new(
error.kind(),
format!("failed to read {}: {error}", path.display()),
))),
}
}
fn config_error(message: impl Into<String>) -> Box<dyn Error> {
Box::new(io::Error::new(ErrorKind::InvalidData, message.into()))
}
+2
View File
@@ -0,0 +1,2 @@
pub mod otel_manager;
pub mod otel_provider;
@@ -1,4 +1,4 @@
use crate::otel_provider::traceparent_context_from_env;
use crate::traces::otel_provider::traceparent_context_from_env;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_api::ResponseEvent;
@@ -16,43 +16,19 @@ use eventsource_stream::Event as StreamEvent;
use eventsource_stream::EventStreamError as StreamError;
use reqwest::Error;
use reqwest::Response;
use serde::Serialize;
use std::borrow::Cow;
use std::fmt::Display;
use std::future::Future;
use std::time::Duration;
use std::time::Instant;
use strum_macros::Display;
use tokio::time::error::Elapsed;
use tracing::Span;
use tracing::trace_span;
use tracing_opentelemetry::OpenTelemetrySpanExt;
#[derive(Debug, Clone, Serialize, Display)]
#[serde(rename_all = "snake_case")]
pub enum ToolDecisionSource {
Config,
User,
}
#[derive(Debug, Clone)]
pub struct OtelEventMetadata {
conversation_id: ThreadId,
auth_mode: Option<String>,
account_id: Option<String>,
account_email: Option<String>,
model: String,
slug: String,
log_user_prompts: bool,
app_version: &'static str,
terminal_type: String,
}
#[derive(Debug, Clone)]
pub struct OtelManager {
metadata: OtelEventMetadata,
session_span: Span,
}
pub use crate::OtelEventMetadata;
pub use crate::OtelManager;
pub use crate::ToolDecisionSource;
impl OtelManager {
#[allow(clippy::too_many_arguments)]
@@ -86,16 +62,11 @@ impl OtelManager {
terminal_type,
},
session_span,
metrics: crate::metrics::global(),
metrics_use_metadata_tags: true,
}
}
pub fn with_model(&self, model: &str, slug: &str) -> Self {
let mut manager = self.clone();
manager.metadata.model = model.to_owned();
manager.metadata.slug = slug.to_owned();
manager
}
pub fn current_span(&self) -> &Span {
&self.session_span
}
@@ -162,7 +133,7 @@ impl OtelManager {
F: FnOnce() -> Fut,
Fut: Future<Output = Result<Response, Error>>,
{
let start = std::time::Instant::now();
let start = Instant::now();
let response = f().await;
let duration = start.elapsed();
@@ -1,9 +1,8 @@
use crate::config::OtelExporter;
use crate::config::OtelHttpProtocol;
use crate::config::OtelSettings;
use crate::config::OtelTlsConfig;
use codex_utils_absolute_path::AbsolutePathBuf;
use http::Uri;
use crate::metrics::MetricsClient;
use crate::metrics::MetricsConfig;
use opentelemetry::Context;
use opentelemetry::KeyValue;
use opentelemetry::context::ContextGuard;
@@ -14,8 +13,6 @@ use opentelemetry::trace::TracerProvider as _;
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
use opentelemetry_otlp::LogExporter;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_LOGS_TIMEOUT;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT;
use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TRACES_TIMEOUT;
use opentelemetry_otlp::Protocol;
use opentelemetry_otlp::SpanExporter;
@@ -23,9 +20,7 @@ use opentelemetry_otlp::WithExportConfig;
use opentelemetry_otlp::WithHttpConfig;
use opentelemetry_otlp::WithTonicConfig;
use opentelemetry_otlp::tonic_types::metadata::MetadataMap;
use opentelemetry_otlp::tonic_types::transport::Certificate as TonicCertificate;
use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig;
use opentelemetry_otlp::tonic_types::transport::Identity as TonicIdentity;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::logs::SdkLoggerProvider;
use opentelemetry_sdk::propagation::TraceContextPropagator;
@@ -33,21 +28,11 @@ use opentelemetry_sdk::trace::BatchSpanProcessor;
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::trace::Tracer;
use opentelemetry_semantic_conventions as semconv;
use reqwest::Certificate as ReqwestCertificate;
use reqwest::Identity as ReqwestIdentity;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderName;
use reqwest::header::HeaderValue;
use std::cell::RefCell;
use std::collections::HashMap;
use std::env;
use std::error::Error;
use std::fs;
use std::io::ErrorKind;
use std::io::{self};
use std::path::PathBuf;
use std::sync::OnceLock;
use std::time::Duration;
use tracing::debug;
use tracing::level_filters::LevelFilter;
use tracing::warn;
@@ -63,10 +48,12 @@ thread_local! {
static TRACEPARENT_GUARD: RefCell<Option<ContextGuard>> = const { RefCell::new(None) };
}
// TODO(jif) move OtelProvider out of `traces/`
pub struct OtelProvider {
pub logger: Option<SdkLoggerProvider>,
pub tracer_provider: Option<SdkTracerProvider>,
pub tracer: Option<Tracer>,
pub metrics: Option<MetricsClient>,
}
impl OtelProvider {
@@ -77,14 +64,33 @@ impl OtelProvider {
if let Some(tracer_provider) = &self.tracer_provider {
let _ = tracer_provider.shutdown();
}
if let Some(metrics) = &self.metrics {
let _ = metrics.shutdown();
}
}
pub fn from(settings: &OtelSettings) -> Result<Option<Self>, Box<dyn Error>> {
let log_enabled = !matches!(settings.exporter, OtelExporter::None);
let trace_enabled = !matches!(settings.trace_exporter, OtelExporter::None);
if !log_enabled && !trace_enabled {
debug!("No exporter enabled in OTLP settings.");
let metric_exporter = crate::config::resolve_exporter(&settings.metrics_exporter);
let metrics = if matches!(metric_exporter, OtelExporter::None) {
None
} else {
Some(MetricsClient::new(MetricsConfig::otlp(
settings.environment.clone(),
settings.service_name.clone(),
settings.service_version.clone(),
metric_exporter,
))?)
};
if let Some(metrics) = metrics.as_ref() {
crate::metrics::install_global(metrics.clone());
}
if !log_enabled && !trace_enabled && metrics.is_none() {
debug!("No OTEL exporter enabled in settings.");
return Ok(None);
}
@@ -113,6 +119,7 @@ impl OtelProvider {
logger,
tracer_provider,
tracer,
metrics,
}))
}
@@ -141,6 +148,10 @@ impl OtelProvider {
pub fn codex_export_filter(meta: &tracing::Metadata<'_>) -> bool {
meta.target().starts_with("codex_otel")
}
pub fn metrics(&self) -> Option<&MetricsClient> {
self.metrics.as_ref()
}
}
impl Drop for OtelProvider {
@@ -151,6 +162,9 @@ impl Drop for OtelProvider {
if let Some(tracer_provider) = &self.tracer_provider {
let _ = tracer_provider.shutdown();
}
if let Some(metrics) = &self.metrics {
let _ = metrics.shutdown();
}
}
}
@@ -223,8 +237,9 @@ fn build_logger(
) -> Result<SdkLoggerProvider, Box<dyn Error>> {
let mut builder = SdkLoggerProvider::builder().with_resource(resource.clone());
match exporter {
match crate::config::resolve_exporter(exporter) {
OtelExporter::None => return Ok(builder.build()),
OtelExporter::Statsig => unreachable!("statsig exporter should be resolved"),
OtelExporter::OtlpGrpc {
endpoint,
headers,
@@ -232,14 +247,14 @@ fn build_logger(
} => {
debug!("Using OTLP Grpc exporter: {endpoint}");
let header_map = build_header_map(headers);
let header_map = crate::otlp::build_header_map(&headers);
let base_tls_config = ClientTlsConfig::new()
.with_enabled_roots()
.assume_http2(true);
let tls_config = match tls.as_ref() {
Some(tls) => build_grpc_tls_config(endpoint, base_tls_config, tls)?,
Some(tls) => crate::otlp::build_grpc_tls_config(&endpoint, base_tls_config, tls)?,
None => base_tls_config,
};
@@ -269,10 +284,10 @@ fn build_logger(
.with_http()
.with_endpoint(endpoint)
.with_protocol(protocol)
.with_headers(headers.clone());
.with_headers(headers);
if let Some(tls) = tls.as_ref() {
let client = build_http_client(tls, OTEL_EXPORTER_OTLP_LOGS_TIMEOUT)?;
let client = crate::otlp::build_http_client(tls, OTEL_EXPORTER_OTLP_LOGS_TIMEOUT)?;
exporter_builder = exporter_builder.with_http_client(client);
}
@@ -289,8 +304,9 @@ fn build_tracer_provider(
resource: &Resource,
exporter: &OtelExporter,
) -> Result<SdkTracerProvider, Box<dyn Error>> {
let span_exporter = match exporter {
let span_exporter = match crate::config::resolve_exporter(exporter) {
OtelExporter::None => return Ok(SdkTracerProvider::builder().build()),
OtelExporter::Statsig => unreachable!("statsig exporter should be resolved"),
OtelExporter::OtlpGrpc {
endpoint,
headers,
@@ -298,14 +314,14 @@ fn build_tracer_provider(
} => {
debug!("Using OTLP Grpc exporter for traces: {endpoint}");
let header_map = build_header_map(headers);
let header_map = crate::otlp::build_header_map(&headers);
let base_tls_config = ClientTlsConfig::new()
.with_enabled_roots()
.assume_http2(true);
let tls_config = match tls.as_ref() {
Some(tls) => build_grpc_tls_config(endpoint, base_tls_config, tls)?,
Some(tls) => crate::otlp::build_grpc_tls_config(&endpoint, base_tls_config, tls)?,
None => base_tls_config,
};
@@ -333,10 +349,11 @@ fn build_tracer_provider(
.with_http()
.with_endpoint(endpoint)
.with_protocol(protocol)
.with_headers(headers.clone());
.with_headers(headers);
if let Some(tls) = tls.as_ref() {
let client = build_http_client(tls, OTEL_EXPORTER_OTLP_TRACES_TIMEOUT)?;
let client =
crate::otlp::build_http_client(tls, OTEL_EXPORTER_OTLP_TRACES_TIMEOUT)?;
exporter_builder = exporter_builder.with_http_client(client);
}
@@ -352,150 +369,6 @@ fn build_tracer_provider(
.build())
}
fn build_header_map(headers: &HashMap<String, String>) -> HeaderMap {
let mut header_map = HeaderMap::new();
for (key, value) in headers {
if let Ok(name) = HeaderName::from_bytes(key.as_bytes())
&& let Ok(val) = HeaderValue::from_str(value)
{
header_map.insert(name, val);
}
}
header_map
}
fn build_grpc_tls_config(
endpoint: &str,
tls_config: ClientTlsConfig,
tls: &OtelTlsConfig,
) -> Result<ClientTlsConfig, Box<dyn Error>> {
let uri: Uri = endpoint.parse()?;
let host = uri.host().ok_or_else(|| {
config_error(format!(
"OTLP gRPC endpoint {endpoint} does not include a host"
))
})?;
let mut config = tls_config.domain_name(host.to_owned());
if let Some(path) = tls.ca_certificate.as_ref() {
let (pem, _) = read_bytes(path)?;
config = config.ca_certificate(TonicCertificate::from_pem(pem));
}
match (&tls.client_certificate, &tls.client_private_key) {
(Some(cert_path), Some(key_path)) => {
let (cert_pem, _) = read_bytes(cert_path)?;
let (key_pem, _) = read_bytes(key_path)?;
config = config.identity(TonicIdentity::from_pem(cert_pem, key_pem));
}
(Some(_), None) | (None, Some(_)) => {
return Err(config_error(
"client_certificate and client_private_key must both be provided for mTLS",
));
}
(None, None) => {}
}
Ok(config)
}
/// Build a blocking HTTP client with TLS configuration for the OTLP HTTP exporter.
///
/// We use `reqwest::blocking::Client` instead of the async client because the
/// `opentelemetry_sdk` `BatchLogProcessor` spawns a dedicated OS thread that uses
/// `futures_executor::block_on()` rather than tokio. When the async reqwest client's
/// timeout calls `tokio::time::sleep()`, it panics with "no reactor running".
fn build_http_client(
tls: &OtelTlsConfig,
timeout_var: &str,
) -> Result<reqwest::blocking::Client, Box<dyn Error>> {
// Wrap in block_in_place because reqwest::blocking::Client creates its own
// internal tokio runtime, which would panic if built directly from an async context.
tokio::task::block_in_place(|| build_http_client_inner(tls, timeout_var))
}
fn build_http_client_inner(
tls: &OtelTlsConfig,
timeout_var: &str,
) -> Result<reqwest::blocking::Client, Box<dyn Error>> {
let mut builder =
reqwest::blocking::Client::builder().timeout(resolve_otlp_timeout(timeout_var));
if let Some(path) = tls.ca_certificate.as_ref() {
let (pem, location) = read_bytes(path)?;
let certificate = ReqwestCertificate::from_pem(pem.as_slice()).map_err(|error| {
config_error(format!(
"failed to parse certificate {}: {error}",
location.display()
))
})?;
// Disable built-in root certificates and use only our custom CA
builder = builder
.tls_built_in_root_certs(false)
.add_root_certificate(certificate);
}
match (&tls.client_certificate, &tls.client_private_key) {
(Some(cert_path), Some(key_path)) => {
let (mut cert_pem, cert_location) = read_bytes(cert_path)?;
let (key_pem, key_location) = read_bytes(key_path)?;
cert_pem.extend_from_slice(key_pem.as_slice());
let identity = ReqwestIdentity::from_pem(cert_pem.as_slice()).map_err(|error| {
config_error(format!(
"failed to parse client identity using {} and {}: {error}",
cert_location.display(),
key_location.display()
))
})?;
builder = builder.identity(identity).https_only(true);
}
(Some(_), None) | (None, Some(_)) => {
return Err(config_error(
"client_certificate and client_private_key must both be provided for mTLS",
));
}
(None, None) => {}
}
builder
.build()
.map_err(|error| Box::new(error) as Box<dyn Error>)
}
fn resolve_otlp_timeout(signal_var: &str) -> Duration {
if let Some(timeout) = read_timeout_env(signal_var) {
return timeout;
}
if let Some(timeout) = read_timeout_env(OTEL_EXPORTER_OTLP_TIMEOUT) {
return timeout;
}
OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT
}
fn read_timeout_env(var: &str) -> Option<Duration> {
let value = env::var(var).ok()?;
let parsed = value.parse::<i64>().ok()?;
if parsed < 0 {
return None;
}
Some(Duration::from_millis(parsed as u64))
}
fn read_bytes(path: &AbsolutePathBuf) -> Result<(Vec<u8>, PathBuf), Box<dyn Error>> {
match fs::read(path) {
Ok(bytes) => Ok((bytes, path.to_path_buf())),
Err(error) => Err(Box::new(io::Error::new(
error.kind(),
format!("failed to read {}: {error}", path.display()),
))),
}
}
fn config_error(message: impl Into<String>) -> Box<dyn Error> {
Box::new(io::Error::new(ErrorKind::InvalidData, message.into()))
}
#[cfg(test)]
mod tests {
use super::*;