From 0b94751cc463d02dec397c4c4dbb77fd9b93d94d Mon Sep 17 00:00:00 2001 From: Alexi Christakis <167903946+alexi-openai@users.noreply.github.com> Date: Tue, 25 Aug 2026 22:36:01 +0000 Subject: [PATCH] Add telemetry for SQLite log persistence (#40726) ## Why SQLite log persistence needs visibility into batch size, write latency, failures, and entries dropped before they reach the writer. Exporter diagnostics must not feed back into the SQLite log sink and keep metric exports active indefinitely. ## What changed - Record write count, duration, batch bytes, entry count, largest-entry size, and queue drops with bounded outcome tags. - Add explicit byte histogram boundaries for log batch metrics. - Filter OTLP transport noise and the benign unsolicited HTTP/2 PING warning from SQLite log persistence to prevent export cycles while retaining other HTTP/2 warnings. ## Testing - Cover metric values, tags, queue-drop reasons, and histogram boundaries. - Verify successful and failed OTLP HTTP and gRPC exports become idle instead of generating recurring SQLite log-write metrics. GitOrigin-RevId: a63224ae15945fe84adc977a8b7d55bb47a4de53 --- codex-rs/app-server/tests/suite/logging.rs | 309 ++++++++++++++++++ codex-rs/otel/src/metrics/client.rs | 33 +- codex-rs/otel/tests/suite/send.rs | 20 ++ codex-rs/rollout/src/sqlite_metrics.rs | 33 ++ codex-rs/state/src/lib.rs | 12 + codex-rs/state/src/log_db.rs | 40 ++- codex-rs/state/src/log_db_filter_tests.rs | 18 + .../state/src/log_write_telemetry_tests.rs | 147 +++++++++ codex-rs/state/src/model/log.rs | 11 + codex-rs/state/src/runtime.rs | 2 + codex-rs/state/src/runtime/logs.rs | 7 +- codex-rs/state/src/telemetry.rs | 46 ++- 12 files changed, 663 insertions(+), 15 deletions(-) create mode 100644 codex-rs/state/src/log_write_telemetry_tests.rs 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;