codex: address PR review feedback (#18725)

This commit is contained in:
Eric Traut
2026-04-20 15:13:51 -07:00
parent 715f78cffc
commit 81c6cde5eb
6 changed files with 609 additions and 52 deletions
@@ -2628,27 +2628,6 @@ impl CodexMessageProcessor {
};
}
if let Some(otel_reloader) = listener_task_context.otel_reloader.as_ref() {
let otel_reload_error = otel_reloader
.reload_from_config(&config)
.err()
.map(|err| err.to_string());
if let Some(err) = otel_reload_error {
listener_task_context
.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to reload otel config: {err}"),
data: None,
},
)
.await;
return;
}
}
let instruction_sources = Self::instruction_sources_from_config(&config).await;
let dynamic_tools = dynamic_tools.unwrap_or_default();
let core_dynamic_tools = if dynamic_tools.is_empty() {
@@ -2677,6 +2656,31 @@ impl CodexMessageProcessor {
.collect()
};
let core_dynamic_tool_count = core_dynamic_tools.len();
let otel_reload = if let Some(otel_reloader) = listener_task_context.otel_reloader.as_ref()
{
let otel_reload_result = otel_reloader
.reload_from_config(&config)
.map_err(|err| err.to_string());
match otel_reload_result {
Ok(otel_reload) => Some(otel_reload),
Err(err) => {
listener_task_context
.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to reload otel config: {err}"),
data: None,
},
)
.await;
return;
}
}
} else {
None
};
match listener_task_context
.thread_manager
@@ -2702,6 +2706,9 @@ impl CodexMessageProcessor {
.await
{
Ok(new_conv) => {
if let Some(otel_reload) = otel_reload {
otel_reload.commit();
}
let NewThread {
thread_id,
thread,
+2 -3
View File
@@ -517,13 +517,12 @@ pub async fn run_main_with_transport(
let log_db_layer = log_db
.clone()
.map(|layer| layer.with_filter(Targets::new().with_default(Level::TRACE)));
let otel_tracing_layer = otel.as_ref().and_then(|provider| provider.tracing_layer());
let (otel_layer, otel_reloader) =
let (otel_layer, otel_trace_layer, otel_reloader) =
otel_reload::OtelReloader::new(&config, otel, default_analytics_enabled);
let _ = tracing_subscriber::registry()
.with(stderr_fmt)
.with(otel_layer)
.with(otel_tracing_layer)
.with(otel_trace_layer)
.with(feedback_layer)
.with(feedback_metadata_layer)
.with(log_db_layer)
+243 -26
View File
@@ -1,17 +1,21 @@
//! App-server OpenTelemetry reloading for configuration resolved after startup.
//!
//! The app server installs its tracing subscriber once, but `thread/start` can
//! later load project-scoped config from the requested cwd. This module keeps
//! the installed log layer stable while swapping the underlying OTel provider
//! when that effective thread config changes.
//! The app server installs its tracing subscriber once, before `thread/start`
//! can load project-scoped config from the requested cwd. This module keeps the
//! installed OTel layers stable while initializing their provider from the first
//! effective thread config.
use codex_config::types::OtelConfig;
use codex_core::config::Config;
use codex_otel::OtelLoggerLayer;
use codex_otel::OtelProvider;
use codex_otel::OtelTraceLayer;
use codex_otel::OtelTraceLayerHandle;
use std::error::Error;
use std::sync::Arc;
use std::sync::Mutex;
use tracing::Subscriber;
use tracing_subscriber::registry::LookupSpan;
#[derive(Clone, Debug, PartialEq)]
struct OtelProviderKey {
@@ -23,45 +27,68 @@ struct OtelProviderKey {
struct OtelReloadState {
key: Option<OtelProviderKey>,
provider: Option<OtelProvider>,
retired_providers: Vec<OtelProvider>,
initialized_from_thread_config: bool,
}
#[derive(Clone)]
pub(crate) struct OtelReloader {
logger_layer: OtelLoggerLayer,
trace_layer: OtelTraceLayerHandle,
state: Arc<Mutex<OtelReloadState>>,
default_analytics_enabled: bool,
}
impl OtelReloader {
pub(crate) fn new(
pub(crate) fn new<S>(
initial_config: &Config,
provider: Option<OtelProvider>,
default_analytics_enabled: bool,
) -> (OtelLoggerLayer, OtelReloader) {
) -> (OtelLoggerLayer, OtelTraceLayer<S>, OtelReloader)
where
S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static,
{
let logger_layer = OtelLoggerLayer::from_provider(provider.as_ref());
let (trace_layer, trace_layer_handle) = OtelTraceLayer::from_provider(provider.as_ref());
(
logger_layer.clone(),
trace_layer,
OtelReloader {
logger_layer,
trace_layer: trace_layer_handle,
state: Arc::new(Mutex::new(OtelReloadState {
key: Some(provider_key(initial_config, default_analytics_enabled)),
provider,
retired_providers: Vec::new(),
initialized_from_thread_config: false,
})),
default_analytics_enabled,
},
)
}
pub(crate) fn reload_from_config(&self, config: &Config) -> Result<(), Box<dyn Error>> {
pub(crate) fn reload_from_config(
&self,
config: &Config,
) -> Result<OtelReloadCommit, Box<dyn Error>> {
let next_key = provider_key(config, self.default_analytics_enabled);
{
let state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state.key.as_ref() == Some(&next_key) {
return Ok(());
}
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state.key.as_ref() == Some(&next_key) {
return Ok(OtelReloadCommit::new(
Arc::clone(&self.state),
next_key,
state.key.clone(),
/*old_provider*/ None,
self.logger_layer.clone(),
self.trace_layer.clone(),
/*applied_reload*/ false,
));
}
if state.initialized_from_thread_config {
return Err("app-server OTel config is already initialized from a different effective thread config; restart the app server to use a different project OTel config".into());
}
let next_provider = codex_core::otel_init::build_provider(
@@ -71,23 +98,104 @@ impl OtelReloader {
self.default_analytics_enabled,
)?;
self.logger_layer.replace_provider(next_provider.as_ref());
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.key = Some(next_key);
state.provider = next_provider;
Ok(())
self.trace_layer.replace_provider(next_provider.as_ref());
let previous_key = state.key.clone();
state.key = Some(next_key.clone());
let old_provider = std::mem::replace(&mut state.provider, next_provider);
Ok(OtelReloadCommit::new(
Arc::clone(&self.state),
next_key,
previous_key,
old_provider,
self.logger_layer.clone(),
self.trace_layer.clone(),
/*applied_reload*/ true,
))
}
pub(crate) fn shutdown(&self) {
self.logger_layer.replace_provider(/*provider*/ None);
self.trace_layer.shutdown();
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.provider.take();
state.retired_providers.clear();
}
}
pub(crate) struct OtelReloadCommit {
state: Arc<Mutex<OtelReloadState>>,
key: OtelProviderKey,
previous_key: Option<OtelProviderKey>,
old_provider: Option<OtelProvider>,
logger_layer: OtelLoggerLayer,
trace_layer: OtelTraceLayerHandle,
applied_reload: bool,
committed: bool,
}
impl OtelReloadCommit {
fn new(
state: Arc<Mutex<OtelReloadState>>,
key: OtelProviderKey,
previous_key: Option<OtelProviderKey>,
old_provider: Option<OtelProvider>,
logger_layer: OtelLoggerLayer,
trace_layer: OtelTraceLayerHandle,
applied_reload: bool,
) -> Self {
Self {
state,
key,
previous_key,
old_provider,
logger_layer,
trace_layer,
applied_reload,
committed: false,
}
}
pub(crate) fn commit(mut self) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state.key.as_ref() == Some(&self.key) {
if let Some(old_provider) = self.old_provider.take() {
state.retired_providers.push(old_provider);
}
state.initialized_from_thread_config = true;
self.committed = true;
}
}
}
impl Drop for OtelReloadCommit {
fn drop(&mut self) {
if self.committed {
return;
}
if !self.applied_reload {
return;
}
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if !state.initialized_from_thread_config && state.key.as_ref() == Some(&self.key) {
let _abandoned_provider =
std::mem::replace(&mut state.provider, self.old_provider.take());
state.key = self.previous_key.clone();
self.logger_layer.replace_provider(state.provider.as_ref());
self.trace_layer.replace_provider(state.provider.as_ref());
} else if let Some(old_provider) = self.old_provider.take() {
state.retired_providers.push(old_provider);
}
}
}
@@ -130,12 +238,14 @@ mod tests {
.codex_home(codex_home.path().to_path_buf())
.build()
.await?;
let (layer, reloader) = OtelReloader::new(
let (logger_layer, trace_layer, reloader) = OtelReloader::new(
&initial_config,
/*provider*/ None,
/*default_analytics_enabled*/ false,
);
let subscriber = tracing_subscriber::registry().with(layer);
let subscriber = tracing_subscriber::registry()
.with(logger_layer)
.with(trace_layer);
let _guard = tracing::subscriber::set_default(subscriber);
let mut project_config = initial_config.clone();
@@ -146,7 +256,7 @@ mod tests {
tls: None,
};
reloader.reload_from_config(&project_config)?;
reloader.reload_from_config(&project_config)?.commit();
tracing::event!(
target: "codex_otel.log_only",
tracing::Level::INFO,
@@ -164,6 +274,113 @@ mod tests {
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn reload_from_config_rejects_different_second_config() -> Result<(), Box<dyn Error>> {
let server = MockServer::start().await;
let codex_home = TempDir::new()?;
let initial_config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.build()
.await?;
let (_logger_layer, _trace_layer, reloader): (
_,
OtelTraceLayer<tracing_subscriber::Registry>,
_,
) = OtelReloader::new(
&initial_config,
/*provider*/ None,
/*default_analytics_enabled*/ false,
);
let mut first_config = initial_config.clone();
first_config.otel.exporter = OtelExporterKind::OtlpHttp {
endpoint: format!("{}/v1/logs", server.uri()),
headers: HashMap::new(),
protocol: OtelHttpProtocol::Json,
tls: None,
};
reloader.reload_from_config(&first_config)?.commit();
let mut second_config = initial_config;
second_config.otel.exporter = OtelExporterKind::OtlpHttp {
endpoint: format!("{}/other/logs", server.uri()),
headers: HashMap::new(),
protocol: OtelHttpProtocol::Json,
tls: None,
};
let err = match reloader.reload_from_config(&second_config) {
Ok(_) => panic!("different effective OTel config should be rejected"),
Err(err) => err,
};
assert!(
err.to_string()
.contains("already initialized from a different effective thread config"),
"unexpected error: {err}"
);
reloader.shutdown();
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn reload_from_config_does_not_pin_until_committed() -> Result<(), Box<dyn Error>> {
let server = MockServer::start().await;
let codex_home = TempDir::new()?;
let initial_config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.build()
.await?;
let (_logger_layer, _trace_layer, reloader): (
_,
OtelTraceLayer<tracing_subscriber::Registry>,
_,
) = OtelReloader::new(
&initial_config,
/*provider*/ None,
/*default_analytics_enabled*/ false,
);
let mut first_config = initial_config.clone();
first_config.otel.exporter = OtelExporterKind::OtlpHttp {
endpoint: format!("{}/v1/logs", server.uri()),
headers: HashMap::new(),
protocol: OtelHttpProtocol::Json,
tls: None,
};
let first_reload = reloader.reload_from_config(&first_config)?;
drop(first_reload);
{
let state = reloader
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
assert!(state.provider.is_none());
assert!(state.retired_providers.is_empty());
}
let mut second_config = initial_config;
second_config.otel.exporter = OtelExporterKind::OtlpHttp {
endpoint: format!("{}/other/logs", server.uri()),
headers: HashMap::new(),
protocol: OtelHttpProtocol::Json,
tls: None,
};
reloader.reload_from_config(&second_config)?.commit();
let err = match reloader.reload_from_config(&first_config) {
Ok(_) => panic!("different effective OTel config should be rejected after commit"),
Err(err) => err,
};
assert!(
err.to_string()
.contains("already initialized from a different effective thread config"),
"unexpected error: {err}"
);
reloader.shutdown();
Ok(())
}
async fn wait_for_otel_logs_payload(server: &MockServer) -> Result<Vec<u8>, Box<dyn Error>> {
let body = timeout(Duration::from_secs(10), async {
loop {
+3
View File
@@ -4,6 +4,7 @@ mod logger_layer;
pub(crate) mod metrics;
pub(crate) mod provider;
pub(crate) mod trace_context;
mod trace_layer;
mod otlp;
mod targets;
@@ -32,6 +33,8 @@ pub use crate::trace_context::set_parent_from_context;
pub use crate::trace_context::set_parent_from_w3c_trace_context;
pub use crate::trace_context::span_w3c_trace_context;
pub use crate::trace_context::traceparent_context_from_env;
pub use crate::trace_layer::OtelTraceLayer;
pub use crate::trace_layer::OtelTraceLayerHandle;
pub use codex_utils_string::sanitize_metric_tag_value;
#[derive(Debug, Clone, Serialize, Display)]
+2 -2
View File
@@ -84,9 +84,8 @@ impl OtelProvider {
Some(MetricsClient::new(config)?)
};
crate::metrics::replace_global(metrics.clone());
if !log_enabled && !trace_enabled && metrics.is_none() {
crate::metrics::replace_global(/*metrics*/ None);
debug!("No OTEL exporter enabled in settings.");
return Ok(None);
}
@@ -109,6 +108,7 @@ impl OtelProvider {
global::set_tracer_provider(provider);
global::set_text_map_propagator(TraceContextPropagator::new());
}
crate::metrics::replace_global(metrics.clone());
Ok(Some(Self {
logger,
tracer_provider,
+331
View File
@@ -0,0 +1,331 @@
//! Reloadable OpenTelemetry trace layer.
//!
//! This layer stays registered with `tracing-subscriber` while allowing callers
//! to replace the OTel tracer behind it. It is used by long-lived processes that
//! discover their effective telemetry config after subscriber initialization.
use crate::OtelProvider;
use opentelemetry_sdk::trace::Tracer;
use std::any::TypeId;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;
use tracing::Event;
use tracing::Subscriber;
use tracing::span::Attributes;
use tracing::span::Id;
use tracing::span::Record;
use tracing_subscriber::Layer;
use tracing_subscriber::layer::Context;
use tracing_subscriber::registry::LookupSpan;
type TraceLayer<S> = tracing_opentelemetry::OpenTelemetryLayer<S, Tracer>;
type ReplaceProviderFn = dyn Fn(Option<&OtelProvider>) + Send + Sync;
#[derive(Clone, Default)]
pub struct OtelTraceLayer<S> {
state: Arc<RwLock<OtelTraceLayerState<S>>>,
}
struct OtelTraceLayerState<S> {
current: Option<Arc<TraceLayer<S>>>,
retired: Vec<Arc<TraceLayer<S>>>,
active_spans: HashMap<Id, Arc<TraceLayer<S>>>,
}
impl<S> Default for OtelTraceLayerState<S> {
fn default() -> Self {
Self {
current: None,
retired: Vec::new(),
active_spans: HashMap::new(),
}
}
}
#[derive(Clone)]
pub struct OtelTraceLayerHandle {
replace_provider: Arc<ReplaceProviderFn>,
shutdown: Arc<dyn Fn() + Send + Sync>,
}
impl OtelTraceLayerHandle {
pub fn replace_provider(&self, provider: Option<&OtelProvider>) {
(self.replace_provider)(provider);
}
pub fn shutdown(&self) {
(self.shutdown)();
}
}
impl<S> OtelTraceLayer<S>
where
S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static,
{
pub fn from_provider(provider: Option<&OtelProvider>) -> (Self, OtelTraceLayerHandle) {
let state = Arc::new(RwLock::new(OtelTraceLayerState {
current: trace_layer(provider),
retired: Vec::new(),
active_spans: HashMap::new(),
}));
let layer = Self {
state: Arc::clone(&state),
};
let replace_state = Arc::clone(&state);
let shutdown_state = Arc::clone(&state);
let handle = OtelTraceLayerHandle {
replace_provider: Arc::new(move |provider| {
replace_trace_layer(&replace_state, trace_layer(provider));
}),
shutdown: Arc::new(move || {
let mut state = shutdown_state
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.current = None;
state.retired.clear();
state.active_spans.clear();
}),
};
(layer, handle)
}
fn current_layer(&self) -> Option<Arc<TraceLayer<S>>> {
self.state
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.current
.clone()
}
fn record_span_layer(&self, id: &Id, layer: Arc<TraceLayer<S>>) {
self.state
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.active_spans
.insert(id.clone(), layer);
}
fn layer_for_span(&self, id: &Id) -> Option<Arc<TraceLayer<S>>> {
let state = self
.state
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state
.active_spans
.get(id)
.cloned()
.or_else(|| state.current.clone())
}
fn remove_span_layer(&self, id: &Id) -> Option<Arc<TraceLayer<S>>> {
let mut state = self
.state
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state
.active_spans
.remove(id)
.or_else(|| state.current.clone())
}
}
impl<S> Layer<S> for OtelTraceLayer<S>
where
S: Subscriber + for<'span> LookupSpan<'span> + Send + Sync + 'static,
{
fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
if !OtelProvider::trace_export_filter(attrs.metadata()) {
return;
}
if let Some(layer) = self.current_layer() {
layer.on_new_span(attrs, id, ctx);
self.record_span_layer(id, layer);
}
}
fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) {
if let Some(layer) = self.layer_for_span(id) {
layer.on_record(id, values, ctx);
}
}
fn on_follows_from(&self, id: &Id, follows: &Id, ctx: Context<'_, S>) {
if let Some(layer) = self.layer_for_span(id) {
layer.on_follows_from(id, follows, ctx);
}
}
fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
if !OtelProvider::trace_export_filter(event.metadata()) {
return;
}
if let Some(layer) = self.current_layer() {
layer.on_event(event, ctx);
}
}
fn on_enter(&self, id: &Id, ctx: Context<'_, S>) {
if let Some(layer) = self.layer_for_span(id) {
layer.on_enter(id, ctx);
}
}
fn on_exit(&self, id: &Id, ctx: Context<'_, S>) {
if let Some(layer) = self.layer_for_span(id) {
layer.on_exit(id, ctx);
}
}
fn on_close(&self, id: Id, ctx: Context<'_, S>) {
if let Some(layer) = self.remove_span_layer(&id) {
layer.on_close(id, ctx);
}
}
fn on_id_change(&self, old: &Id, new: &Id, ctx: Context<'_, S>) {
if let Some(layer) = self.layer_for_span(old) {
layer.on_id_change(old, new, ctx);
let mut state = self
.state
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(layer) = state.active_spans.remove(old) {
state.active_spans.insert(new.clone(), layer);
}
}
}
unsafe fn downcast_raw(&self, id: TypeId) -> Option<*const ()> {
let state = self
.state
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let layer = state.current.as_ref()?;
unsafe { Layer::<S>::downcast_raw(layer.as_ref(), id) }
}
}
fn replace_trace_layer<S>(
state: &RwLock<OtelTraceLayerState<S>>,
next_layer: Option<Arc<TraceLayer<S>>>,
) {
let mut state = state
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(current) = state.current.take() {
state.retired.push(current);
}
state.current = next_layer;
}
fn trace_layer<S>(provider: Option<&OtelProvider>) -> Option<Arc<TraceLayer<S>>>
where
S: Subscriber + for<'span> LookupSpan<'span>,
{
provider
.and_then(|provider| provider.tracer.clone())
.map(|tracer| Arc::new(tracing_opentelemetry::layer().with_tracer(tracer)))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::OtelProvider;
use crate::current_span_trace_id;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_sdk::error::OTelSdkResult;
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::trace::SpanData;
use opentelemetry_sdk::trace::SpanExporter;
use std::sync::Mutex;
use tracing::trace_span;
use tracing_subscriber::prelude::*;
#[derive(Clone, Debug, Default)]
struct RecordingSpanExporter {
spans: Arc<Mutex<Vec<SpanData>>>,
}
impl SpanExporter for RecordingSpanExporter {
async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
self.spans
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.extend(batch);
Ok(())
}
}
fn provider_with_tracer(scope: &'static str) -> (OtelProvider, Arc<Mutex<Vec<SpanData>>>) {
let exporter = RecordingSpanExporter::default();
let spans = Arc::clone(&exporter.spans);
let tracer_provider = SdkTracerProvider::builder()
.with_simple_exporter(exporter)
.build();
let tracer = tracer_provider.tracer(scope);
let provider = OtelProvider {
logger: None,
tracer_provider: Some(tracer_provider),
tracer: Some(tracer),
metrics: None,
};
(provider, spans)
}
#[test]
fn trace_layer_preserves_span_context_downcasting() {
let tracer_provider = SdkTracerProvider::builder().build();
let tracer = tracer_provider.tracer("codex-otel-tests");
let provider = OtelProvider {
logger: None,
tracer_provider: Some(tracer_provider),
tracer: Some(tracer),
metrics: None,
};
let (layer, _handle) = OtelTraceLayer::from_provider(Some(&provider));
let subscriber = tracing_subscriber::registry().with(layer);
let _guard = tracing::subscriber::set_default(subscriber);
let span = trace_span!("test_span");
let _entered = span.enter();
let trace_id = current_span_trace_id().expect("trace id");
assert_eq!(trace_id.len(), 32);
assert!(trace_id.chars().all(|ch| ch.is_ascii_hexdigit()));
assert_ne!(trace_id, "00000000000000000000000000000000");
}
#[test]
fn span_closes_with_original_layer_after_provider_replacement() {
let (first_provider, first_spans) = provider_with_tracer("first");
let (second_provider, second_spans) = provider_with_tracer("second");
let (layer, handle) = OtelTraceLayer::from_provider(Some(&first_provider));
let subscriber = tracing_subscriber::registry().with(layer);
let _guard = tracing::subscriber::set_default(subscriber);
let span = trace_span!("old_span");
{
let _entered = span.enter();
}
handle.replace_provider(Some(&second_provider));
drop(span);
assert_eq!(
first_spans
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.len(),
1
);
assert_eq!(
second_spans
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.len(),
0
);
}
}