From 09e03f8acd226b61e0170c0b9951093e033db69d Mon Sep 17 00:00:00 2001 From: Jeff Larson Date: Wed, 22 Jul 2026 21:25:12 -0700 Subject: [PATCH] fix(server): batch OTLP log/trace inserts to kill ~30s ingest p99 (JEF-495) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit store_logs and store_traces inserted one row at a time in a nested loop, so a large OTLP batch was N sequential awaited round-trips, each acquiring a pool connection — driving ingest.logs p99 to ~30s. Decode the whole request into rows, then write them in chunked INSERT ... SELECT * FROM unnest($1::type[], ...) statements (all values bound, no interpolation), collapsing N round-trips into O(chunks). A shared write_chunked helper drives both paths; on a chunk error it falls back to per-row inserts so a single bad row can't drop the whole batch, keeping DROP_INSERT accounting accurate. Each insert is one isolatable execute so JEF-496 can wrap it in read-only-failover retry. Spans keep ON CONFLICT (trace_id, span_id) DO NOTHING, which dedupes intra-batch duplicates too. Stored columns/values are unchanged. Tests: smoke.rs ingests a 250-record log batch and a 100-span trace batch (with an intra-batch duplicate) in one call each, asserting every row persists with identical values and DROP_INSERT is untouched. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JbrmfzHsTMMzPaSrUkZgWo --- server/src/otlp.rs | 253 ++++++++++++++++++++++++++++++------------ server/tests/smoke.rs | 129 +++++++++++++++++++++ 2 files changed, 313 insertions(+), 69 deletions(-) diff --git a/server/src/otlp.rs b/server/src/otlp.rs index fb72a55..5519c93 100644 --- a/server/src/otlp.rs +++ b/server/src/otlp.rs @@ -142,28 +142,29 @@ pub async fn ingest_metrics( // --------------------------------------------------------------------------- pub async fn store_traces(pool: &PgPool, req: ExportTraceServiceRequest) -> u64 { - let mut count = 0; + // Decode the whole request into rows first, then write them in batched + // statements (one per chunk) instead of one INSERT round-trip per span — a + // large export was N sequential awaited inserts, each taking a pool connection. + let mut rows: Vec = Vec::new(); for rs in &req.resource_spans { let rattrs = resource_attrs(rs.resource.as_ref()); let service = service_name(rattrs); for ss in &rs.scope_spans { for span in &ss.spans { - match insert_span(pool, service.as_deref(), rattrs, span).await { - Ok(()) => count += 1, - Err(e) => { - tracing::warn!("insert span failed: {e}"); - crate::selfmon::DROP_INSERT.fetch_add(1, Ordering::Relaxed); - } - } + rows.push(span_row(service.as_deref(), rattrs, span)); } } } + let count = write_chunked(pool, &rows, insert_spans).await; crate::selfmon::SPANS_INGESTED.fetch_add(count, Ordering::Relaxed); count } pub async fn store_logs(pool: &PgPool, req: ExportLogsServiceRequest) -> u64 { - let mut count = 0; + // Decode the whole request into rows first, then write them in batched + // statements (one per chunk) instead of one INSERT round-trip per record — + // the per-row loop was the source of the ~30s ingest p99 (JEF-495). + let mut rows: Vec = Vec::new(); for rl in &req.resource_logs { // Keep resource attributes (k8s.pod.name / node / container, …) so logs // can be filtered by pod/host, not just service. @@ -171,17 +172,59 @@ pub async fn store_logs(pool: &PgPool, req: ExportLogsServiceRequest) -> u64 { let service = service_name(rattrs); for sl in &rl.scope_logs { for rec in &sl.log_records { - match insert_log(pool, service.as_deref(), rattrs, rec).await { - Ok(()) => count += 1, - Err(e) => { - tracing::warn!("insert log failed: {e}"); - crate::selfmon::DROP_INSERT.fetch_add(1, Ordering::Relaxed); + rows.push(log_row(service.as_deref(), rattrs, rec)); + } + } + } + let count = write_chunked(pool, &rows, insert_logs).await; + crate::selfmon::LOGS_INGESTED.fetch_add(count, Ordering::Relaxed); + count +} + +/// Rows per batched INSERT. UNNEST binds each column as a single array parameter +/// (so the 65_535 bind-param ceiling is irrelevant), but chunking still bounds the +/// per-statement array size and memory. 5_000 collapses any realistic OTLP batch +/// into one or two round-trips. +const INSERT_CHUNK: usize = 5_000; + +/// Write `rows` in chunks, one batched `INSERT` per chunk. On a chunk error, fall +/// back to inserting each row of that chunk on its own so a single bad row can't +/// drop the whole batch — every row that still fails is logged and counted in +/// `DROP_INSERT`, preserving the per-row drop accounting the old loop had. Returns +/// the number of rows written. +/// +/// `insert` runs exactly one `execute` per call (per chunk, and per row on the +/// fallback path); JEF-496 will wrap that whole-batch write in read-only-failover +/// retry, so keep it a single isolatable statement. +async fn write_chunked( + pool: &PgPool, + rows: &[T], + insert: impl AsyncFn(&PgPool, &[T]) -> Result<(), sqlx::Error>, +) -> u64 { + let mut count = 0u64; + for chunk in rows.chunks(INSERT_CHUNK) { + match insert(pool, chunk).await { + // A successful batch counts every row it attempted — matching the old + // per-row loop, which counted each `execute` (an ON CONFLICT no-op still + // succeeded and still counted). + Ok(()) => count += chunk.len() as u64, + Err(e) => { + tracing::warn!( + "batch insert failed ({} rows), isolating per-row: {e}", + chunk.len() + ); + for row in chunk { + match insert(pool, std::slice::from_ref(row)).await { + Ok(()) => count += 1, + Err(e) => { + tracing::warn!("insert row failed: {e}"); + crate::selfmon::DROP_INSERT.fetch_add(1, Ordering::Relaxed); + } } } } } } - crate::selfmon::LOGS_INGESTED.fetch_add(count, Ordering::Relaxed); count } @@ -227,21 +270,36 @@ pub async fn store_metrics(pool: &PgPool, req: ExportMetricsServiceRequest) -> u // Inserts // --------------------------------------------------------------------------- -async fn insert_span( - pool: &PgPool, - service: Option<&str>, - resource: &[KeyValue], - span: &Span, -) -> anyhow::Result<()> { - let trace_id = hex::encode(&span.trace_id); - let span_id = hex::encode(&span.span_id); - let parent = (!span.parent_span_id.is_empty()).then(|| hex::encode(&span.parent_span_id)); - let start = ts(span.start_time_unix_nano); - let end = ts(span.end_time_unix_nano); - let duration_ms = span - .end_time_unix_nano - .saturating_sub(span.start_time_unix_nano) as f64 - / 1_000_000.0; +/// One decoded span, ready to batch-insert. Mirrors the `spans` columns. +struct SpanRow { + trace_id: String, + span_id: String, + parent_span_id: Option, + service: Option, + name: String, + kind: i32, + start_time: DateTime, + end_time: DateTime, + duration_ms: f64, + status_code: Option, + status_message: Option, + attrs: serde_json::Value, +} + +/// One decoded log record, ready to batch-insert. Mirrors the `logs` columns. +struct LogRow { + time: DateTime, + trace_id: Option, + span_id: Option, + service: Option, + severity_number: i32, + severity_text: String, + body: Option, + attrs: serde_json::Value, +} + +/// Decode one span into a `SpanRow`. No DB I/O — `store_traces` batches these. +fn span_row(service: Option<&str>, resource: &[KeyValue], span: &Span) -> SpanRow { let (status_code, status_message) = match &span.status { Some(s) => ( Some(s.code), @@ -249,60 +307,117 @@ async fn insert_span( ), None => (None, None), }; - let attrs = merged_attrs(resource, &span.attributes); + SpanRow { + trace_id: hex::encode(&span.trace_id), + span_id: hex::encode(&span.span_id), + parent_span_id: (!span.parent_span_id.is_empty()) + .then(|| hex::encode(&span.parent_span_id)), + service: service.map(str::to_string), + name: span.name.clone(), + kind: span.kind, + start_time: ts(span.start_time_unix_nano), + end_time: ts(span.end_time_unix_nano), + duration_ms: span + .end_time_unix_nano + .saturating_sub(span.start_time_unix_nano) as f64 + / 1_000_000.0, + status_code, + status_message, + attrs: merged_attrs(resource, &span.attributes), + } +} + +/// Decode one log record into a `LogRow`. No DB I/O — `store_logs` batches these. +fn log_row(service: Option<&str>, resource: &[KeyValue], rec: &LogRecord) -> LogRow { + let nanos = if rec.time_unix_nano != 0 { + rec.time_unix_nano + } else { + rec.observed_time_unix_nano + }; + LogRow { + time: ts(nanos), + trace_id: (!rec.trace_id.is_empty()).then(|| hex::encode(&rec.trace_id)), + span_id: (!rec.span_id.is_empty()).then(|| hex::encode(&rec.span_id)), + service: service.map(str::to_string), + severity_number: rec.severity_number, + severity_text: rec.severity_text.clone(), + body: rec.body.as_ref().map(any_value_to_text), + attrs: merged_attrs(resource, &rec.attributes), + } +} + +/// Batch-insert spans: one `INSERT … SELECT * FROM unnest(...)` over parallel +/// per-column arrays (all bound params — no string interpolation). `ON CONFLICT +/// (trace_id, span_id) DO NOTHING` de-dupes against existing rows and intra-batch +/// duplicates alike, matching the old single-row insert. One `execute` per call so +/// JEF-496 can wrap it in read-only-failover retry. +async fn insert_spans(pool: &PgPool, rows: &[SpanRow]) -> Result<(), sqlx::Error> { + // Borrow each column into a parallel array; the row slice outlives the query, + // so no per-row cloning is needed. + let trace_ids: Vec<&str> = rows.iter().map(|r| r.trace_id.as_str()).collect(); + let span_ids: Vec<&str> = rows.iter().map(|r| r.span_id.as_str()).collect(); + let parents: Vec> = rows.iter().map(|r| r.parent_span_id.as_deref()).collect(); + let services: Vec> = rows.iter().map(|r| r.service.as_deref()).collect(); + let names: Vec<&str> = rows.iter().map(|r| r.name.as_str()).collect(); + let kinds: Vec = rows.iter().map(|r| r.kind).collect(); + let starts: Vec> = rows.iter().map(|r| r.start_time).collect(); + let ends: Vec> = rows.iter().map(|r| r.end_time).collect(); + let durations: Vec = rows.iter().map(|r| r.duration_ms).collect(); + let status_codes: Vec> = rows.iter().map(|r| r.status_code).collect(); + let status_msgs: Vec> = rows.iter().map(|r| r.status_message.as_deref()).collect(); + let attrs: Vec<&serde_json::Value> = rows.iter().map(|r| &r.attrs).collect(); sqlx::query( "INSERT INTO spans (trace_id, span_id, parent_span_id, service, name, kind, start_time, end_time, duration_ms, status_code, status_message, attributes) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) + SELECT * FROM unnest($1::text[], $2::text[], $3::text[], $4::text[], $5::text[], + $6::int[], $7::timestamptz[], $8::timestamptz[], $9::float8[], + $10::int[], $11::text[], $12::jsonb[]) ON CONFLICT (trace_id, span_id) DO NOTHING", ) - .bind(trace_id) - .bind(span_id) - .bind(parent) - .bind(service) - .bind(&span.name) - .bind(span.kind) - .bind(start) - .bind(end) - .bind(duration_ms) - .bind(status_code) - .bind(status_message) - .bind(attrs) + .bind(&trace_ids) + .bind(&span_ids) + .bind(&parents) + .bind(&services) + .bind(&names) + .bind(&kinds) + .bind(&starts) + .bind(&ends) + .bind(&durations) + .bind(&status_codes) + .bind(&status_msgs) + .bind(&attrs) .execute(pool) .await?; Ok(()) } -async fn insert_log( - pool: &PgPool, - service: Option<&str>, - resource: &[KeyValue], - rec: &LogRecord, -) -> anyhow::Result<()> { - let nanos = if rec.time_unix_nano != 0 { - rec.time_unix_nano - } else { - rec.observed_time_unix_nano - }; - let time = ts(nanos); - let trace_id = (!rec.trace_id.is_empty()).then(|| hex::encode(&rec.trace_id)); - let span_id = (!rec.span_id.is_empty()).then(|| hex::encode(&rec.span_id)); - let body = rec.body.as_ref().map(any_value_to_text); - let attrs = merged_attrs(resource, &rec.attributes); +/// Batch-insert logs: one `INSERT … SELECT * FROM unnest(...)` over parallel +/// per-column arrays (all bound params). One `execute` per call so JEF-496 can wrap +/// it in read-only-failover retry. +async fn insert_logs(pool: &PgPool, rows: &[LogRow]) -> Result<(), sqlx::Error> { + let times: Vec> = rows.iter().map(|r| r.time).collect(); + let trace_ids: Vec> = rows.iter().map(|r| r.trace_id.as_deref()).collect(); + let span_ids: Vec> = rows.iter().map(|r| r.span_id.as_deref()).collect(); + let services: Vec> = rows.iter().map(|r| r.service.as_deref()).collect(); + let sev_nums: Vec = rows.iter().map(|r| r.severity_number).collect(); + let sev_texts: Vec<&str> = rows.iter().map(|r| r.severity_text.as_str()).collect(); + let bodies: Vec> = rows.iter().map(|r| r.body.as_deref()).collect(); + let attrs: Vec<&serde_json::Value> = rows.iter().map(|r| &r.attrs).collect(); sqlx::query( "INSERT INTO logs (time, trace_id, span_id, service, severity_number, severity_text, body, attributes) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", + SELECT * FROM unnest($1::timestamptz[], $2::text[], $3::text[], $4::text[], + $5::int[], $6::text[], $7::text[], $8::jsonb[])", ) - .bind(time) - .bind(trace_id) - .bind(span_id) - .bind(service) - .bind(rec.severity_number) - .bind(&rec.severity_text) - .bind(body) - .bind(attrs) + .bind(×) + .bind(&trace_ids) + .bind(&span_ids) + .bind(&services) + .bind(&sev_nums) + .bind(&sev_texts) + .bind(&bodies) + .bind(&attrs) .execute(pool) .await?; Ok(()) diff --git a/server/tests/smoke.rs b/server/tests/smoke.rs index b85eb4a..f8de406 100644 --- a/server/tests/smoke.rs +++ b/server/tests/smoke.rs @@ -380,6 +380,135 @@ async fn ingest_and_query_a_log() { assert_eq!(arr[0]["severity_text"], "INFO"); } +#[tokio::test] +#[serial] +async fn ingest_a_log_batch_persists_every_row_in_one_call() { + // JEF-495: store_logs batches a whole request into chunked INSERTs. A single + // multi-record request must land every row (identical column values) and not + // touch DROP_INSERT. + let Some(pool) = pool_or_skip().await else { + return; + }; + let router = app(pool.clone()); + + const N: usize = 250; + let drops_before = selfmon::DROP_INSERT.load(std::sync::atomic::Ordering::Relaxed); + let records: Vec = (0..N) + .map(|i| LogRecord { + time_unix_nano: 1_000_000_000 + i as u64, + severity_number: 9, + severity_text: "INFO".to_string(), + trace_id: vec![(i % 251) as u8 + 1; 16], + body: Some(AnyValue { + value: Some(any_value::Value::StringValue(format!("line {i}"))), + }), + ..Default::default() + }) + .collect(); + let req = ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(Resource { + attributes: vec![kv("service.name", "batcher")], + ..Default::default() + }), + scope_logs: vec![ScopeLogs { + log_records: records, + ..Default::default() + }], + ..Default::default() + }], + }; + + assert_eq!( + post_proto(&router, "/v1/logs", req.encode_to_vec()).await, + StatusCode::OK + ); + + // Every row persisted in the one call — the batched path, not O(N) inserts. + assert_eq!(count(&pool, "logs").await, N as i64); + assert_eq!( + selfmon::DROP_INSERT.load(std::sync::atomic::Ordering::Relaxed), + drops_before, + "a clean batch drops nothing" + ); + + // Column values survive the UNNEST round-trip intact. + let (status, logs) = get_json(&router, "/api/logs?service=batcher&limit=1000").await; + assert_eq!(status, StatusCode::OK); + let arr = logs.as_array().unwrap(); + assert_eq!(arr.len(), N); + let first = arr + .iter() + .find(|l| l["body"] == "line 0") + .expect("line 0 present"); + assert_eq!(first["service"], "batcher"); + assert_eq!(first["severity_text"], "INFO"); + assert!(first["trace_id"].is_string(), "trace_id preserved"); +} + +#[tokio::test] +#[serial] +async fn ingest_a_span_batch_persists_and_dedupes_in_one_call() { + // JEF-495: store_traces batches too. Distinct spans all land; an intra-batch + // duplicate (same trace_id/span_id) is collapsed by ON CONFLICT DO NOTHING, + // exactly as the old per-row insert did. + let Some(pool) = pool_or_skip().await else { + return; + }; + let router = app(pool.clone()); + + const N: usize = 100; + let start = nanos_ago(5); + let mut spans: Vec = (0..N) + .map(|i| Span { + trace_id: vec![7u8; 16], + span_id: (i as u64 + 1).to_be_bytes().to_vec(), + name: format!("op {i}"), + start_time_unix_nano: start, + end_time_unix_nano: start + 1_000_000, + ..Default::default() + }) + .collect(); + // A duplicate of span #1 in the same batch — must not create a second row. + spans.push(Span { + trace_id: vec![7u8; 16], + span_id: 1u64.to_be_bytes().to_vec(), + name: "op 0 dup".to_string(), + start_time_unix_nano: start, + end_time_unix_nano: start + 1_000_000, + ..Default::default() + }); + + let req = ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + resource: Some(Resource { + attributes: vec![kv("service.name", "spanbatch")], + ..Default::default() + }), + scope_spans: vec![ScopeSpans { + spans, + ..Default::default() + }], + ..Default::default() + }], + }; + + assert_eq!( + post_proto(&router, "/v1/traces", req.encode_to_vec()).await, + StatusCode::OK + ); + + // N distinct spans landed; the duplicate did not add a row. + let stored: i64 = sqlx::query_scalar("SELECT count(*) FROM spans WHERE service = 'spanbatch'") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + stored, N as i64, + "distinct spans persisted, duplicate deduped" + ); +} + #[tokio::test] #[serial] async fn ingest_and_query_a_metric() {