diff --git a/codex-rs/app-server/tests/suite/logging.rs b/codex-rs/app-server/tests/suite/logging.rs index 31cf07d3df..f1a6dc8c92 100644 --- a/codex-rs/app-server/tests/suite/logging.rs +++ b/codex-rs/app-server/tests/suite/logging.rs @@ -30,7 +30,14 @@ use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; +use std::collections::BTreeMap; use tempfile::TempDir; +use test_case::test_case; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::sync::watch; use tokio::time::Duration; use tokio::time::timeout; use wiremock::Mock; @@ -263,6 +270,308 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> { Ok(()) } +#[tokio::test] +async fn sqlite_log_metrics_exports_do_not_create_log_cycles() -> Result<()> { + let quiet_period = codex_state::log_db::LogSinkQueueConfig::default().flush_interval + + Duration::from_millis(600); + let export_timeout = quiet_period * 3; + + for status in [200, 500] { + let collector = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/metrics")) + .respond_with(ResponseTemplate::new(status)) + .mount(&collector) + .await; + + let codex_home = TempDir::new()?; + let endpoint = format!("{}/v1/metrics", collector.uri()); + std::fs::write( + codex_home.path().join("config.toml"), + format!( + "[analytics]\nenabled = true\n\n[otel.metrics_exporter.otlp-http]\nendpoint = {endpoint:?}\nprotocol = \"json\"\n" + ), + )?; + + let _app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_env_overrides(&[ + ("OTEL_METRIC_EXPORT_INTERVAL", Some("200")), + ( + codex_app_server_transport::REMOTE_CONTROL_DISABLED_ENV_VAR, + Some("1"), + ), + ]) + .build_initialized_with_timeout(export_timeout) + .await?; + + let exports = timeout(export_timeout, async { + loop { + let requests = collector.received_requests().await.unwrap_or_default(); + if requests.iter().any(|request| { + String::from_utf8_lossy(&request.body).contains(codex_state::LOG_WRITE_METRIC) + }) { + break requests.len(); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await?; + + let requests = timeout(export_timeout, async { + let mut observed_exports = exports; + loop { + tokio::time::sleep(quiet_period).await; + let requests = collector.received_requests().await.unwrap_or_default(); + if requests.len() == observed_exports { + break requests; + } + observed_exports = requests.len(); + } + }) + .await + .with_context(|| { + format!("OTLP HTTP {status} exports did not stop after the application became idle") + })?; + + let expected_metrics = [ + (codex_state::LOG_WRITE_METRIC, "sum"), + (codex_state::LOG_WRITE_DURATION_METRIC, "histogram"), + (codex_state::LOG_WRITE_BYTES_METRIC, "histogram"), + (codex_state::LOG_WRITE_ENTRIES_METRIC, "histogram"), + (codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC, "histogram"), + ]; + let payloads = requests + .iter() + .map(|request| serde_json::from_slice::(&request.body)) + .collect::>>()?; + let mut observed_metrics = BTreeMap::<&str, (u64, f64)>::new(); + for (name, kind) in expected_metrics { + let points = payloads + .iter() + .flat_map(|payload| payload["resourceMetrics"].as_array().into_iter().flatten()) + .flat_map(|resource| resource["scopeMetrics"].as_array().into_iter().flatten()) + .flat_map(|scope| scope["metrics"].as_array().into_iter().flatten()) + .filter(|metric| metric["name"].as_str() == Some(name)) + .flat_map(|metric| metric[kind]["dataPoints"].as_array().into_iter().flatten()) + .collect::>(); + assert!( + !points.is_empty(), + "metric {name} must reach the collector as a {kind}" + ); + + for point in points { + let attributes: BTreeMap<&str, &str> = point["attributes"] + .as_array() + .context("metric data points must include attributes")? + .iter() + .filter_map(|attribute| { + Some(( + attribute["key"].as_str()?, + attribute["value"]["stringValue"].as_str()?, + )) + }) + .collect(); + assert_eq!( + attributes, + BTreeMap::from([ + ("error", "none"), + ("originator", "codex-app-server"), + ("status", "success"), + ]), + "metric {name} must preserve its production dimensions" + ); + + let count_field = if kind == "sum" { "asInt" } else { "count" }; + let count = point[count_field] + .as_u64() + .or_else(|| point[count_field].as_str()?.parse().ok()) + .with_context(|| format!("metric {name} must include a valid {count_field}"))?; + let value = if kind == "sum" { + count as f64 + } else { + point["sum"] + .as_f64() + .with_context(|| format!("metric {name} must include its sample sum"))? + }; + let observed = observed_metrics.entry(name).or_default(); + observed.0 += count; + observed.1 += value; + + if matches!( + name, + codex_state::LOG_WRITE_BYTES_METRIC + | codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC + ) { + let bounds = point["explicitBounds"] + .as_array() + .context("byte histograms must export explicit bucket bounds")?; + assert_eq!(bounds.first().and_then(Value::as_f64), Some(128.0)); + assert_eq!(bounds.last().and_then(Value::as_f64), Some(16_777_216.0)); + } + } + } + + let write_count = observed_metrics[codex_state::LOG_WRITE_METRIC].0; + assert!(write_count > 0, "at least one SQLite batch must be written"); + for (name, _) in expected_metrics { + assert_eq!( + observed_metrics[name].0, write_count, + "metric {name} must record exactly one sample per SQLite write" + ); + } + + let state = codex_state::StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + "test-provider".to_string(), + ) + .await?; + let persisted_logs = state.query_logs(&codex_state::LogQuery::default()).await?; + assert_eq!( + observed_metrics[codex_state::LOG_WRITE_ENTRIES_METRIC].1, + persisted_logs.len() as f64, + "exported batch sizes must equal the number of persisted SQLite log rows" + ); + assert!( + observed_metrics[codex_state::LOG_WRITE_BYTES_METRIC].1 + >= observed_metrics[codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC].1, + "each batch's largest entry cannot exceed the total batch size" + ); + } + + Ok(()) +} + +/// Exporting SQLite metrics must eventually stop producing new SQLite metrics, +/// including when the collector uses HTTP/2 or rejects the gRPC request. +#[test_case("0"; "success")] +#[test_case("14"; "unavailable")] +#[tokio::test] +async fn sqlite_log_metrics_grpc_exports_do_not_create_log_cycles(grpc_status: &str) -> Result<()> { + let quiet_period = codex_state::log_db::LogSinkQueueConfig::default().flush_interval + + Duration::from_millis(600); + let export_timeout = quiet_period * 3; + let (export_count, mut exports) = watch::channel(/*init*/ 0_usize); + let (ping_ack_count, mut ping_acknowledgements) = watch::channel(/*init*/ 0_usize); + let collector = MockServer::builder() + .disable_request_recording() + .start() + .await; + // An uncompressed gRPC frame containing an empty ExportMetricsServiceResponse. + // Tonic also accepts grpc-status in the initial response headers. + let response = ResponseTemplate::new(/*s*/ 200) + .set_body_raw(vec![0; 5], "application/grpc") + .insert_header("grpc-status", grpc_status); + Mock::given(method("POST")) + .and(path( + "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export", + )) + .respond_with(move |request: &wiremock::Request| { + let metric_name = codex_state::LOG_WRITE_METRIC.as_bytes(); + if request + .body + .windows(metric_name.len()) + .any(|bytes| bytes == metric_name) + { + export_count.send_modify(|count| *count += 1); + } + response.clone() + }) + .mount(&collector) + .await; + + let proxy = TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}", proxy.local_addr()?); + let collector_address = *collector.address(); + let _proxy = tokio::spawn(async move { + let (downstream, _) = proxy.accept().await?; + let upstream = TcpStream::connect(collector_address).await?; + let (mut downstream_reader, mut downstream_writer) = downstream.into_split(); + let (mut upstream_reader, mut upstream_writer) = upstream.into_split(); + + let requests = tokio::io::copy(&mut downstream_reader, &mut upstream_writer); + let responses = async move { + loop { + let mut header = [0; 9]; + if upstream_reader.read_exact(&mut header).await.is_err() { + break; + } + let length = (usize::from(header[0]) << 16) + | (usize::from(header[1]) << 8) + | usize::from(header[2]); + let mut payload = vec![0; length]; + upstream_reader.read_exact(&mut payload).await?; + downstream_writer.write_all(&header).await?; + downstream_writer.write_all(&payload).await?; + + if header[3] == 0 { + // A valid but unsolicited HTTP/2 PING ACK makes h2 emit a + // warning from its separate connection-driving task. + downstream_writer + .write_all(&[0, 0, 8, 6, 1, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8]) + .await?; + ping_ack_count.send_modify(|count| *count += 1); + } + } + Ok::<(), std::io::Error>(()) + }; + + tokio::try_join!(requests, responses)?; + Ok::<(), std::io::Error>(()) + }); + + let codex_home = TempDir::new()?; + std::fs::write( + codex_home.path().join("config.toml"), + format!( + "[analytics]\nenabled = true\n\n[otel.metrics_exporter.otlp-grpc]\nendpoint = {endpoint:?}\n" + ), + )?; + let _app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_env_overrides(&[ + ("OTEL_METRIC_EXPORT_INTERVAL", Some("200")), + ( + codex_app_server_transport::REMOTE_CONTROL_DISABLED_ENV_VAR, + Some("1"), + ), + ]) + .build_initialized_with_timeout(export_timeout) + .await?; + + timeout(export_timeout, exports.wait_for(|count| *count > 0)) + .await + .context("collector never received a SQLite log-write metric")??; + timeout( + export_timeout, + ping_acknowledgements.wait_for(|count| *count > 0), + ) + .await + .context("collector response never produced an unsolicited HTTP/2 PING acknowledgment")??; + let first_count = *exports.borrow_and_update(); + + // Allow delayed startup batches, but require silence longer than one SQLite + // flush interval plus one metric-export interval. Do not retain request bodies + // or keep resetting the overall deadline when a broken exporter feeds itself. + let quiescence = timeout(export_timeout, async { + loop { + match timeout(quiet_period, exports.changed()).await { + Err(_) => return Ok::<(), watch::error::RecvError>(()), + Ok(result) => result?, + } + } + }) + .await; + let final_count = *exports.borrow(); + quiescence.with_context(|| { + format!( + "OTLP gRPC status {grpc_status} did not become idle: SQLite log-write exports increased from {first_count} to {final_count}" + ) + })??; + + Ok(()) +} + #[tokio::test] async fn app_server_emits_structured_tool_call_timing_event() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/otel/src/metrics/client.rs b/codex-rs/otel/src/metrics/client.rs index 057d69680f..8304ab7d01 100644 --- a/codex-rs/otel/src/metrics/client.rs +++ b/codex-rs/otel/src/metrics/client.rs @@ -152,7 +152,13 @@ impl MetricsClientInner { Ok(()) } - fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) -> Result<()> { + fn histogram( + &self, + name: &str, + value: i64, + boundaries: Option<&[f64]>, + tags: &[(&str, &str)], + ) -> Result<()> { validate_metric_name(name)?; let attributes = self.attributes(tags)?; @@ -164,9 +170,13 @@ impl MetricsClientInner { .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()); + let histogram = histograms.entry(name.to_string()).or_insert_with(|| { + let builder = self.meter.f64_histogram(name.to_string()); + match boundaries { + Some(boundaries) => builder.with_boundaries(boundaries.to_vec()).build(), + None => builder.build(), + } + }); histogram.record(value as f64, &attributes); Ok(()) } @@ -401,7 +411,20 @@ impl MetricsClient { /// Send a single histogram sample. pub fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) -> Result<()> { - self.active_inner().histogram(name, value, tags) + self.active_inner() + .histogram(name, value, /*boundaries*/ None, tags) + } + + /// Send a single histogram sample using explicit bucket boundaries. + pub fn histogram_with_boundaries( + &self, + name: &str, + value: i64, + boundaries: &[f64], + tags: &[(&str, &str)], + ) -> Result<()> { + self.active_inner() + .histogram(name, value, Some(boundaries), tags) } /// Send a single gauge measurement. diff --git a/codex-rs/otel/tests/suite/send.rs b/codex-rs/otel/tests/suite/send.rs index c3e2027a63..e4d956a5fb 100644 --- a/codex-rs/otel/tests/suite/send.rs +++ b/codex-rs/otel/tests/suite/send.rs @@ -107,6 +107,26 @@ fn send_builds_payload_with_tags_and_histograms() -> Result<()> { Ok(()) } +#[test] +fn histogram_uses_explicit_bucket_boundaries() -> Result<()> { + let (metrics, exporter) = build_metrics_with_defaults(&[])?; + + metrics.histogram_with_boundaries( + "codex.payload_bytes", + /*value*/ 1024, + &[256.0, 1024.0, 4096.0], + &[], + )?; + metrics.shutdown()?; + + assert_eq!( + histogram_data(&latest_metrics(&exporter), "codex.payload_bytes"), + (vec![256.0, 1024.0, 4096.0], vec![0, 1, 0, 0], 1024.0, 1) + ); + + Ok(()) +} + // Ensures defaults merge per line and overrides take precedence. #[test] fn send_merges_default_tags_per_line() -> Result<()> { diff --git a/codex-rs/rollout/src/sqlite_metrics.rs b/codex-rs/rollout/src/sqlite_metrics.rs index 7601e79729..d0c3cdd75c 100644 --- a/codex-rs/rollout/src/sqlite_metrics.rs +++ b/codex-rs/rollout/src/sqlite_metrics.rs @@ -5,6 +5,29 @@ use codex_otel::ORIGINATOR_TAG; use codex_otel::bounded_originator_tag_value; use codex_state::DbTelemetry; use codex_state::DbTelemetryHandle; +use codex_state::LOG_WRITE_BYTES_METRIC; +use codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC; + +const LOG_WRITE_BYTES_BOUNDARIES: &[f64] = &[ + 128.0, + 256.0, + 512.0, + 1_024.0, + 2_048.0, + 4_096.0, + 8_192.0, + 16_384.0, + 32_768.0, + 65_536.0, + 131_072.0, + 262_144.0, + 524_288.0, + 1_048_576.0, + 2_097_152.0, + 4_194_304.0, + 8_388_608.0, + 16_777_216.0, +]; struct OtelDbTelemetry { metrics: codex_otel::MetricsClient, @@ -17,6 +40,16 @@ impl DbTelemetry for OtelDbTelemetry { let _ = self.metrics.counter(name, inc, &tags); } + fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) { + let tags = with_originator(tags, self.originator); + let _ = match name { + LOG_WRITE_BYTES_METRIC | LOG_WRITE_MAX_ENTRY_BYTES_METRIC => self + .metrics + .histogram_with_boundaries(name, value, LOG_WRITE_BYTES_BOUNDARIES, &tags), + _ => self.metrics.histogram(name, value, &tags), + }; + } + fn record_duration(&self, name: &str, duration: Duration, tags: &[(&str, &str)]) { let tags = with_originator(tags, self.originator); let _ = self.metrics.record_duration(name, duration, &tags); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index b2c61b4860..1638e95caa 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -119,3 +119,15 @@ pub const DB_INIT_METRIC: &str = "codex.sqlite.init.count"; pub const DB_INIT_DURATION_METRIC: &str = "codex.sqlite.init.duration_ms"; /// Rollout fallback attempts. Tags: [caller, reason] pub const DB_FALLBACK_METRIC: &str = "codex.sqlite.fallback.count"; +/// SQLite log batch write attempts. Tags: [status, error] +pub const LOG_WRITE_METRIC: &str = "codex.sqlite.logs.write.count"; +/// SQLite log batch write latency. Tags: [status, error] +pub const LOG_WRITE_DURATION_METRIC: &str = "codex.sqlite.logs.write.duration_ms"; +/// Estimated bytes in each SQLite log batch. Tags: [status, error] +pub const LOG_WRITE_BYTES_METRIC: &str = "codex.sqlite.logs.write.bytes"; +/// Number of entries in each SQLite log batch. Tags: [status, error] +pub const LOG_WRITE_ENTRIES_METRIC: &str = "codex.sqlite.logs.write.entries"; +/// Largest estimated entry size in each SQLite log batch. Tags: [status, error] +pub const LOG_WRITE_MAX_ENTRY_BYTES_METRIC: &str = "codex.sqlite.logs.write.max_entry_bytes"; +/// SQLite log entries discarded before they can be queued. Tags: [reason] +pub const LOG_QUEUE_DROPPED_METRIC: &str = "codex.sqlite.logs.queue.dropped"; diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 367d2e3376..06387efc4c 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -24,6 +24,7 @@ use std::future::Future; use std::sync::OnceLock; use std::time::Duration; +use std::time::Instant; use std::time::SystemTime; use std::time::UNIX_EPOCH; @@ -56,11 +57,16 @@ const SQLX_LOG_TARGETS: [&str; 3] = ["sqlx", "sqlx_core", "sqlx_sqlite"]; pub fn default_filter() -> Targets { Targets::new() .with_default(LevelFilter::TRACE) + .with_target("h2", LevelFilter::WARN) .with_target("hyper_util", LevelFilter::WARN) .with_target("log", LevelFilter::OFF) // Avoid constructing backend diagnostics when no other layer needs them. .with_targets(SQLX_LOG_TARGETS.map(|target| (format!("{target}::"), LevelFilter::OFF))) .with_target("codex_rmcp_client", LevelFilter::INFO) + .with_target("opentelemetry-otlp", LevelFilter::OFF) + .with_target("opentelemetry-http", LevelFilter::OFF) + .with_target("tonic::transport", LevelFilter::WARN) + .with_target("tower::buffer", LevelFilter::WARN) .with_target("codex_otel.log_only", LevelFilter::OFF) .with_target("codex_otel.trace_safe", LevelFilter::OFF) .with_target("rmcp", LevelFilter::INFO) @@ -162,7 +168,13 @@ impl LogDbLayer { } fn try_send(&self, entry: LogEntry) { - let _ = self.sender.try_send(LogDbCommand::Entry(Box::new(entry))); + if let Err(error) = self.sender.try_send(LogDbCommand::Entry(Box::new(entry))) { + let reason = match error { + mpsc::error::TrySendError::Full(_) => "full", + mpsc::error::TrySendError::Closed(_) => "closed", + }; + crate::telemetry::record_log_queue_drop(reason, /*telemetry*/ None); + } } } @@ -230,7 +242,10 @@ where // `tracing-log` checks filters with the original log target before // dispatching an event whose tracing target is `log`, so the outer // target filter cannot reliably reject these bridged events. - if target == "log" { + // OTLP exporters and their HTTP adapters log every export. Recording + // those events would create another SQLite-write metric and keep the + // exporter active indefinitely, including when its queue is full. + if matches!(target, "log" | "opentelemetry-otlp" | "opentelemetry-http") { return; } @@ -247,6 +262,18 @@ where let mut visitor = MessageVisitor::default(); event.record(&mut visitor); + // OTLP gRPC connections run on their own task, so an unsolicited PING + // ACK can emit this benign warning after every otherwise successful + // export. Drop only that warning before queue/drop accounting; other + // HTTP/2 diagnostics must remain available to unrelated clients. + if target == "h2::proto::ping_pong" + && visitor + .message + .as_deref() + .is_some_and(|message| message.starts_with("recv PING ack that we never sent: ")) + { + return; + } let thread_id = visitor .thread_id .clone() @@ -457,7 +484,14 @@ async fn flush(state_db: &StateRuntime, buffer: &mut Vec) { return; } let entries = buffer.split_off(0); - let _ = state_db.insert_logs(entries.as_slice()).await; + let started = Instant::now(); + let result = state_db.insert_logs(entries.as_slice()).await; + crate::telemetry::record_log_write( + /*telemetry*/ None, + started.elapsed(), + entries.as_slice(), + &result, + ); } #[derive(Default)] diff --git a/codex-rs/state/src/log_db_filter_tests.rs b/codex-rs/state/src/log_db_filter_tests.rs index 7c9ca17d3f..28a36c706f 100644 --- a/codex-rs/state/src/log_db_filter_tests.rs +++ b/codex-rs/state/src/log_db_filter_tests.rs @@ -32,6 +32,18 @@ async fn sqlite_sink_filters_noisy_targets_without_dropping_useful_diagnostics() tracing::warn!(target: "sqlx::pool::acquire", "dropped-slow-acquire-warning"); tracing::warn!(target: "sqlx::other", "dropped-sqlx-warning"); tracing::info!(target: "sqlx_application", "retained-unrelated-target"); + tracing::debug!(target: "opentelemetry-otlp", "dropped-otlp-export"); + tracing::error!(target: "opentelemetry-http", "dropped-http-export"); + tracing::trace!(target: "h2::proto::streams", "dropped-http2-trace"); + tracing::debug!(target: "tonic::transport::channel", "dropped-grpc-transport-debug"); + tracing::debug!(target: "tower::buffer::worker", "dropped-grpc-buffer-debug"); + tracing::warn!( + target: "h2::proto::ping_pong", + "recv PING ack that we never sent: {:?}", + "synthetic-ack" + ); + tracing::warn!(target: "h2::proto::ping_pong", "retained-http2-ping-warning"); + tracing::warn!(target: "h2::proto::streams", "retained-http2-warning"); tracing::debug!(target: "rmcp::transport", "dropped-rmcp-debug"); tracing::info!(target: "rmcp::transport", "retained-rmcp-info"); tracing::debug!( @@ -96,6 +108,12 @@ async fn sqlite_sink_filters_noisy_targets_without_dropping_useful_diagnostics() "sqlx_application", Some("retained-unrelated-target") ), + ( + "WARN", + "h2::proto::ping_pong", + Some("retained-http2-ping-warning") + ), + ("WARN", "h2::proto::streams", Some("retained-http2-warning")), ("INFO", "rmcp::transport", Some("retained-rmcp-info")), ( "INFO", diff --git a/codex-rs/state/src/log_write_telemetry_tests.rs b/codex-rs/state/src/log_write_telemetry_tests.rs new file mode 100644 index 0000000000..90e76afb77 --- /dev/null +++ b/codex-rs/state/src/log_write_telemetry_tests.rs @@ -0,0 +1,147 @@ +use std::sync::Mutex; +use std::time::Duration; + +use pretty_assertions::assert_eq; + +use super::DbTelemetry; +use super::record_log_queue_drop; +use super::record_log_write; +use crate::LOG_QUEUE_DROPPED_METRIC; +use crate::LOG_WRITE_BYTES_METRIC; +use crate::LOG_WRITE_DURATION_METRIC; +use crate::LOG_WRITE_ENTRIES_METRIC; +use crate::LOG_WRITE_MAX_ENTRY_BYTES_METRIC; +use crate::LOG_WRITE_METRIC; +use crate::LogEntry; + +type MetricEvent = (String, i64, Vec<(String, String)>); + +#[derive(Default)] +struct TestTelemetry(Mutex>); + +impl TestTelemetry { + fn record(&self, name: &str, value: i64, tags: &[(&str, &str)]) { + self.0.lock().expect("telemetry lock").push(( + name.to_string(), + value, + tags.iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect(), + )); + } +} + +impl DbTelemetry for TestTelemetry { + fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) { + self.record(name, inc, tags); + } + + fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) { + self.record(name, value, tags); + } + + fn record_duration(&self, name: &str, duration: Duration, tags: &[(&str, &str)]) { + self.record(name, duration.as_millis() as i64, tags); + } +} + +fn log_entry(message: &str, feedback_log_body: Option<&str>) -> LogEntry { + LogEntry { + ts: 1, + ts_nanos: 2, + level: "INFO".to_string(), + target: "t".to_string(), + message: Some(message.to_string()), + feedback_log_body: feedback_log_body.map(str::to_string), + thread_id: None, + process_uuid: None, + module_path: Some("m".to_string()), + file: Some("f".to_string()), + line: None, + } +} + +fn expected_events(metrics: &[(&str, i64)], tags: &[(&str, &str)]) -> Vec { + metrics + .iter() + .map(|(name, value)| { + ( + (*name).to_string(), + *value, + tags.iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect(), + ) + }) + .collect() +} + +#[test] +fn records_successful_log_write_duration_batch_size_and_largest_entry() { + let telemetry = TestTelemetry::default(); + let entries = [ + log_entry("abc", /*feedback_log_body*/ None), + log_entry("ignored", Some("longer")), + ]; + + record_log_write( + Some(&telemetry), + Duration::from_millis(17), + &entries, + &Ok(()), + ); + + assert_eq!( + *telemetry.0.lock().expect("telemetry lock"), + expected_events( + &[ + (LOG_WRITE_METRIC, 1), + (LOG_WRITE_DURATION_METRIC, 17), + (LOG_WRITE_BYTES_METRIC, 23), + (LOG_WRITE_ENTRIES_METRIC, 2), + (LOG_WRITE_MAX_ENTRY_BYTES_METRIC, 13), + ], + &[("status", "success"), ("error", "none")], + ), + ); +} + +#[test] +fn records_failed_log_write_with_bounded_sqlite_error() { + let telemetry = TestTelemetry::default(); + let entries = [log_entry("abc", /*feedback_log_body*/ None)]; + let result = Err(anyhow::Error::from(sqlx::Error::PoolTimedOut)); + + record_log_write( + Some(&telemetry), + Duration::from_millis(35), + &entries, + &result, + ); + + assert_eq!( + *telemetry.0.lock().expect("telemetry lock"), + expected_events( + &[ + (LOG_WRITE_METRIC, 1), + (LOG_WRITE_DURATION_METRIC, 35), + (LOG_WRITE_BYTES_METRIC, 10), + (LOG_WRITE_ENTRIES_METRIC, 1), + (LOG_WRITE_MAX_ENTRY_BYTES_METRIC, 10), + ], + &[("status", "failed"), ("error", "pool_timeout")], + ), + ); +} + +#[test] +fn records_queue_drop_reason() { + let telemetry = TestTelemetry::default(); + + record_log_queue_drop("full", Some(&telemetry)); + + assert_eq!( + *telemetry.0.lock().expect("telemetry lock"), + expected_events(&[(LOG_QUEUE_DROPPED_METRIC, 1)], &[("reason", "full")]), + ); +} diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs index 176ab6db2e..43685dfdd8 100644 --- a/codex-rs/state/src/model/log.rs +++ b/codex-rs/state/src/model/log.rs @@ -16,6 +16,17 @@ pub struct LogEntry { pub line: Option, } +impl LogEntry { + pub(crate) fn estimated_bytes(&self) -> i64 { + let feedback_log_body = self.feedback_log_body.as_ref().or(self.message.as_ref()); + feedback_log_body.map_or(0, String::len) as i64 + + self.level.len() as i64 + + self.target.len() as i64 + + self.module_path.as_ref().map_or(0, String::len) as i64 + + self.file.as_ref().map_or(0, String::len) as i64 + } +} + #[derive(Clone, Debug, FromRow)] pub struct LogRow { pub id: i64, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 50ccf8e43e..e7fa3cffef 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -485,6 +485,8 @@ mod tests { _tags: &[(&str, &str)], ) { } + + fn histogram(&self, _name: &str, _value: i64, _tags: &[(&str, &str)]) {} } fn tags_to_map(tags: &[(&str, &str)]) -> BTreeMap { diff --git a/codex-rs/state/src/runtime/logs.rs b/codex-rs/state/src/runtime/logs.rs index d905d9cfdb..d2525480c7 100644 --- a/codex-rs/state/src/runtime/logs.rs +++ b/codex-rs/state/src/runtime/logs.rs @@ -23,11 +23,6 @@ impl StateRuntime { // Both `query_logs` and `/feedback` read the persisted // `feedback_log_body`, while `LogEntry.message` is only a write-time // fallback for callers that still populate the old field. - let estimated_bytes = feedback_log_body.map_or(0, String::len) as i64 - + entry.level.len() as i64 - + entry.target.len() as i64 - + entry.module_path.as_ref().map_or(0, String::len) as i64 - + entry.file.as_ref().map_or(0, String::len) as i64; row.push_bind(entry.ts) .push_bind(entry.ts_nanos) .push_bind(&entry.level) @@ -38,7 +33,7 @@ impl StateRuntime { .push_bind(&entry.module_path) .push_bind(&entry.file) .push_bind(entry.line) - .push_bind(estimated_bytes); + .push_bind(entry.estimated_bytes()); }); builder.build().execute(&mut *tx).await?; self.prune_logs_after_insert(entries, &mut tx).await?; diff --git a/codex-rs/state/src/telemetry.rs b/codex-rs/state/src/telemetry.rs index 1b14fc7d44..ac32d90672 100644 --- a/codex-rs/state/src/telemetry.rs +++ b/codex-rs/state/src/telemetry.rs @@ -6,14 +6,22 @@ use std::time::Duration; use crate::DB_FALLBACK_METRIC; use crate::DB_INIT_DURATION_METRIC; use crate::DB_INIT_METRIC; +use crate::LOG_QUEUE_DROPPED_METRIC; +use crate::LOG_WRITE_BYTES_METRIC; +use crate::LOG_WRITE_DURATION_METRIC; +use crate::LOG_WRITE_ENTRIES_METRIC; +use crate::LOG_WRITE_MAX_ENTRY_BYTES_METRIC; +use crate::LOG_WRITE_METRIC; +use crate::LogEntry; use tracing::debug; -/// Low-cardinality sink for SQLite startup and fallback telemetry. +/// Low-cardinality sink for SQLite startup, fallback, and log-write telemetry. /// /// Implementations should absorb delivery failures locally. Database behavior /// must not depend on whether telemetry export succeeds. pub trait DbTelemetry: Send + Sync + 'static { fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]); + fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]); fn record_duration(&self, name: &str, duration: Duration, tags: &[(&str, &str)]); } @@ -96,6 +104,38 @@ pub fn record_fallback( ); } +pub(crate) fn record_log_write( + telemetry: Option<&dyn DbTelemetry>, + duration: Duration, + entries: &[LogEntry], + result: &anyhow::Result<()>, +) { + let Some(telemetry) = resolve_telemetry(telemetry) else { + return; + }; + + let outcome = DbOutcomeTags::from_result(result); + let tags = [("status", outcome.status), ("error", outcome.error)]; + let mut batch_bytes = 0_i64; + let mut max_entry_bytes = 0_i64; + for entry in entries { + let entry_bytes = entry.estimated_bytes(); + batch_bytes = batch_bytes.saturating_add(entry_bytes); + max_entry_bytes = max_entry_bytes.max(entry_bytes); + } + let entry_count = i64::try_from(entries.len()).unwrap_or(i64::MAX); + + telemetry.counter(LOG_WRITE_METRIC, /*inc*/ 1, &tags); + telemetry.record_duration(LOG_WRITE_DURATION_METRIC, duration, &tags); + telemetry.histogram(LOG_WRITE_BYTES_METRIC, batch_bytes, &tags); + telemetry.histogram(LOG_WRITE_ENTRIES_METRIC, entry_count, &tags); + telemetry.histogram(LOG_WRITE_MAX_ENTRY_BYTES_METRIC, max_entry_bytes, &tags); +} + +pub(crate) fn record_log_queue_drop(reason: &'static str, telemetry: Option<&dyn DbTelemetry>) { + record_counter(telemetry, LOG_QUEUE_DROPPED_METRIC, &[("reason", reason)]); +} + fn record_counter(telemetry: Option<&dyn DbTelemetry>, name: &str, tags: &[(&str, &str)]) { if let Some(telemetry) = resolve_telemetry(telemetry) { telemetry.counter(name, /*inc*/ 1, tags); @@ -205,3 +245,7 @@ mod tests { assert_eq!(classify_sqlite_code("2067"), "constraint"); } } + +#[cfg(test)] +#[path = "log_write_telemetry_tests.rs"] +mod log_write_telemetry_tests;