From 9067949941da401109ada59b3a2546d58735f061 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Wed, 19 Aug 2026 16:27:45 -0400 Subject: [PATCH 1/4] feat(otlp): persist trace batches idempotently --- src/db/otlp_traces.rs | 235 ++++++++++++++++++++++++++++++++++++ src/db/otlp_traces_tests.rs | 217 +++++++++++++++++++++++++++++++++ 2 files changed, 452 insertions(+) create mode 100644 src/db/otlp_traces_tests.rs diff --git a/src/db/otlp_traces.rs b/src/db/otlp_traces.rs index 75f179d9..2ce4d0e4 100644 --- a/src/db/otlp_traces.rs +++ b/src/db/otlp_traces.rs @@ -1,5 +1,22 @@ +//! OTLP trace persistence models and idempotent batch writes. + +use anyhow::Result; +use rusqlite::{Transaction, params}; use serde::{Deserialize, Serialize}; +use super::{DbPool, TRANSIENT_SQLITE_RETRY_DELAYS_MS, is_transient_sqlite_lock, write_lock}; + +const MAX_METADATA_JSON_BYTES: usize = 256 * 1024; +const MAX_SPAN_NAME_CHARS: usize = 1024; +const MAX_TRACE_STATE_CHARS: usize = 512; +const MAX_STATUS_MESSAGE_CHARS: usize = 4096; +const MAX_HOSTNAME_CHARS: usize = 255; +const MAX_SERVICE_CHARS: usize = 512; +const MAX_SCOPE_CHARS: usize = 512; +const MAX_TOOL_BYTES: usize = 64; +const MAX_PROJECT_BYTES: usize = 512; +const MAX_SESSION_BYTES: usize = 128; + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct OtelSpanInput { pub trace_id: String, @@ -30,3 +47,221 @@ pub struct OtelSpanInput { pub received_at: String, pub content_scrubbed: bool, } + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OtelTraceBatchResult { + pub accepted: usize, + pub duplicates: usize, + pub rejected: usize, +} + +impl OtelTraceBatchResult { + pub const fn total(self) -> usize { + self.accepted + self.duplicates + self.rejected + } +} + +pub fn insert_otel_spans_batch( + pool: &DbPool, + entries: &[OtelSpanInput], +) -> Result { + let mut attempt = 0usize; + loop { + match insert_otel_spans_batch_once(pool, entries) { + Ok(result) => return Ok(result), + Err(error) + if is_transient_sqlite_lock(&error) + && attempt < TRANSIENT_SQLITE_RETRY_DELAYS_MS.len() => + { + let delay_ms = TRANSIENT_SQLITE_RETRY_DELAYS_MS[attempt]; + tracing::warn!( + error = %error, + attempt = attempt + 1, + retry_delay_ms = delay_ms, + entry_count = entries.len(), + "Transient SQLite lock during OTLP trace batch insert - retrying" + ); + std::thread::sleep(std::time::Duration::from_millis(delay_ms)); + attempt += 1; + } + Err(error) => return Err(error), + } + } +} + +fn insert_otel_spans_batch_once( + pool: &DbPool, + entries: &[OtelSpanInput], +) -> Result { + if entries.is_empty() { + return Ok(OtelTraceBatchResult::default()); + } + let mut conn = pool.get()?; + let _write_guard = write_lock(); + let tx = conn.transaction()?; + let result = insert_otel_spans_batch_in_tx(&tx, entries)?; + tx.commit()?; + tracing::debug!( + accepted = result.accepted, + duplicates = result.duplicates, + rejected = result.rejected, + "Committed OTLP trace batch transaction" + ); + Ok(result) +} + +fn insert_otel_spans_batch_in_tx( + tx: &Transaction<'_>, + entries: &[OtelSpanInput], +) -> Result { + let mut result = OtelTraceBatchResult::default(); + let mut stmt = tx.prepare_cached( + "INSERT INTO otel_spans + (trace_id, span_id, parent_span_id, trace_state, flags, span_name, span_kind, + start_time_unix_nano, end_time_unix_nano, duration_nano, status_code, + status_message, hostname, service_name, service_version, scope_name, + scope_version, ai_tool, ai_project, ai_session_id, run_id, resource_json, + attributes_json, events_json, links_json, received_at, content_scrubbed) + VALUES + (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, + ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27) + ON CONFLICT(trace_id, span_id) DO NOTHING", + )?; + + for entry in entries { + if !valid_span_input(tx, entry)? { + result.rejected += 1; + continue; + } + let changed = stmt.execute(params![ + entry.trace_id, + entry.span_id, + entry.parent_span_id, + entry.trace_state, + entry.flags, + entry.span_name, + entry.span_kind, + entry.start_time_unix_nano, + entry.end_time_unix_nano, + entry.duration_nano, + entry.status_code, + entry.status_message, + entry.hostname, + entry.service_name, + entry.service_version, + entry.scope_name, + entry.scope_version, + entry.ai_tool, + entry.ai_project, + entry.ai_session_id, + entry.run_id, + entry.resource_json, + entry.attributes_json, + entry.events_json, + entry.links_json, + entry.received_at, + entry.content_scrubbed, + ])?; + if changed == 1 { + result.accepted += 1; + } else { + result.duplicates += 1; + } + } + Ok(result) +} + +fn valid_span_input(tx: &Transaction<'_>, entry: &OtelSpanInput) -> Result { + if !valid_hex_id(&entry.trace_id, 32) + || !valid_hex_id(&entry.span_id, 16) + || entry + .parent_span_id + .as_deref() + .is_some_and(|value| !valid_hex_id(value, 16)) + || entry.duration_nano < 0 + || entry.end_time_unix_nano < entry.start_time_unix_nano + || entry + .end_time_unix_nano + .checked_sub(entry.start_time_unix_nano) + != Some(entry.duration_nano) + || chrono::DateTime::parse_from_rfc3339(&entry.received_at).is_err() + || !within_chars(&entry.span_name, MAX_SPAN_NAME_CHARS) + || !entry + .trace_state + .as_deref() + .is_none_or(|value| within_chars(value, MAX_TRACE_STATE_CHARS)) + || !within_chars(&entry.hostname, MAX_HOSTNAME_CHARS) + || !optional_chars(&entry.service_name, MAX_SERVICE_CHARS) + || !optional_chars(&entry.service_version, MAX_SERVICE_CHARS) + || !optional_chars(&entry.scope_name, MAX_SCOPE_CHARS) + || !optional_chars(&entry.scope_version, MAX_SCOPE_CHARS) + || !optional_chars(&entry.status_message, MAX_STATUS_MESSAGE_CHARS) + || !optional_bytes(&entry.ai_tool, MAX_TOOL_BYTES) + || !optional_bytes(&entry.ai_project, MAX_PROJECT_BYTES) + || !optional_bytes(&entry.ai_session_id, MAX_SESSION_BYTES) + || !json_shape(&entry.resource_json, JsonShape::Object) + || !json_shape(&entry.attributes_json, JsonShape::Object) + || !json_shape(&entry.events_json, JsonShape::Array) + || !json_shape(&entry.links_json, JsonShape::Array) + { + return Ok(false); + } + if let Some(run_id) = entry.run_id { + if run_id <= 0 { + return Ok(false); + } + let exists: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM agent_runs WHERE id = ?1)", + [run_id], + |row| row.get(0), + )?; + if !exists { + return Ok(false); + } + } + Ok(true) +} + +fn valid_hex_id(value: &str, expected_chars: usize) -> bool { + value.len() == expected_chars + && value.bytes().all(|byte| byte.is_ascii_hexdigit()) + && value.bytes().any(|byte| byte != b'0') +} + +fn within_chars(value: &str, maximum: usize) -> bool { + value.chars().count() <= maximum +} + +fn optional_chars(value: &Option, maximum: usize) -> bool { + value + .as_deref() + .is_none_or(|value| within_chars(value, maximum)) +} + +fn optional_bytes(value: &Option, maximum: usize) -> bool { + value.as_ref().is_none_or(|value| value.len() <= maximum) +} + +#[derive(Clone, Copy)] +enum JsonShape { + Object, + Array, +} + +fn json_shape(raw: &str, shape: JsonShape) -> bool { + if raw.len() > MAX_METADATA_JSON_BYTES { + return false; + } + let Ok(value) = serde_json::from_str::(raw) else { + return false; + }; + matches!( + (shape, value), + (JsonShape::Object, serde_json::Value::Object(_)) + | (JsonShape::Array, serde_json::Value::Array(_)) + ) +} + +#[cfg(test)] +#[path = "otlp_traces_tests.rs"] +mod tests; diff --git a/src/db/otlp_traces_tests.rs b/src/db/otlp_traces_tests.rs new file mode 100644 index 00000000..a95cd82f --- /dev/null +++ b/src/db/otlp_traces_tests.rs @@ -0,0 +1,217 @@ +use super::*; + +use crate::config::StorageConfig; +use crate::db::init_pool; + +fn pool(name: &str) -> (tempfile::TempDir, DbPool) { + let dir = tempfile::tempdir().unwrap(); + let pool = init_pool(&StorageConfig::for_test(dir.path().join(name))).unwrap(); + (dir, pool) +} + +fn span(trace: u8, span: u8) -> OtelSpanInput { + OtelSpanInput { + trace_id: format!("{trace:02x}").repeat(16), + span_id: format!("{span:02x}").repeat(8), + parent_span_id: Some("33".repeat(8)), + trace_state: Some("vendor=value".to_string()), + flags: 0x101, + span_name: "tool.call".to_string(), + span_kind: 3, + start_time_unix_nano: 1_700_000_000_000_000_000, + end_time_unix_nano: 1_700_000_000_000_025_000, + duration_nano: 25_000, + status_code: 2, + status_message: Some("boom".to_string()), + hostname: "devhost".to_string(), + service_name: Some("claude-code".to_string()), + service_version: Some("1.2.3".to_string()), + scope_name: Some("cortex.trace.tests".to_string()), + scope_version: Some("0.1.0".to_string()), + ai_tool: Some("claude".to_string()), + ai_project: Some("/workspace/cortex".to_string()), + ai_session_id: Some("session-123".to_string()), + run_id: None, + resource_json: r#"{"resource":{"attributes":{}},"scope":{"attributes":{}}}"#.to_string(), + attributes_json: r#"{"custom":"value"}"#.to_string(), + events_json: r#"[{"time_unix_nano":1,"name":"event"}]"#.to_string(), + links_json: + r#"[{"trace_id":"44444444444444444444444444444444","span_id":"5555555555555555"}]"# + .to_string(), + received_at: "2026-08-18T20:15:00.000Z".to_string(), + content_scrubbed: true, + } +} + +fn row_count(pool: &DbPool) -> i64 { + pool.get() + .unwrap() + .query_row("SELECT COUNT(*) FROM otel_spans", [], |row| row.get(0)) + .unwrap() +} + +#[test] +fn empty_batch_is_a_noop() { + let (_dir, pool) = pool("empty.db"); + let result = insert_otel_spans_batch(&pool, &[]).unwrap(); + assert_eq!(result, OtelTraceBatchResult::default()); + assert_eq!(result.total(), 0); + assert_eq!(row_count(&pool), 0); +} + +#[test] +fn duplicate_export_is_idempotent_and_reported_as_duplicate_not_rejected() { + let (_dir, pool) = pool("duplicate.db"); + let input = span(0x11, 0x22); + + let first = insert_otel_spans_batch(&pool, std::slice::from_ref(&input)).unwrap(); + assert_eq!( + first, + OtelTraceBatchResult { + accepted: 1, + duplicates: 0, + rejected: 0, + } + ); + let second = insert_otel_spans_batch(&pool, std::slice::from_ref(&input)).unwrap(); + assert_eq!( + second, + OtelTraceBatchResult { + accepted: 0, + duplicates: 1, + rejected: 0, + } + ); + assert_eq!(row_count(&pool), 1); + + let conn = pool.get().unwrap(); + let persisted: (String, String, String, i64, String, String, String, bool) = conn + .query_row( + "SELECT trace_id, span_id, span_name, duration_nano, attributes_json, + events_json, links_json, content_scrubbed + FROM otel_spans", + [], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + row.get(6)?, + row.get(7)?, + )) + }, + ) + .unwrap(); + assert_eq!(persisted.0, input.trace_id); + assert_eq!(persisted.1, input.span_id); + assert_eq!(persisted.2, input.span_name); + assert_eq!(persisted.3, input.duration_nano); + assert_eq!(persisted.4, input.attributes_json); + assert_eq!(persisted.5, input.events_json); + assert_eq!(persisted.6, input.links_json); + assert!(persisted.7); +} + +#[test] +fn duplicate_inside_one_batch_counts_one_accept_and_one_duplicate() { + let (_dir, pool) = pool("same-batch.db"); + let input = span(0x11, 0x22); + let result = insert_otel_spans_batch(&pool, &[input.clone(), input]).unwrap(); + assert_eq!( + result, + OtelTraceBatchResult { + accepted: 1, + duplicates: 1, + rejected: 0, + } + ); + assert_eq!(result.total(), 2); + assert_eq!(row_count(&pool), 1); +} + +#[test] +fn malformed_rows_are_rejected_without_poisoning_valid_neighbors() { + let (_dir, pool) = pool("mixed.db"); + let valid = span(0x11, 0x22); + let mut invalid = Vec::new(); + + let mut bad = span(0x21, 0x31); + bad.trace_id = "0".repeat(32); + invalid.push(bad); + + let mut bad = span(0x22, 0x32); + bad.span_id = "zz".repeat(8); + invalid.push(bad); + + let mut bad = span(0x23, 0x33); + bad.parent_span_id = Some("0".repeat(16)); + invalid.push(bad); + + let mut bad = span(0x24, 0x34); + bad.duration_nano += 1; + invalid.push(bad); + + let mut bad = span(0x25, 0x35); + bad.attributes_json = "not-json".to_string(); + invalid.push(bad); + + let mut bad = span(0x26, 0x36); + bad.events_json = "{}".to_string(); + invalid.push(bad); + + let mut bad = span(0x27, 0x37); + bad.received_at = "not-a-time".to_string(); + invalid.push(bad); + + let mut bad = span(0x28, 0x38); + bad.run_id = Some(9_999_999); + invalid.push(bad); + + let mut entries = vec![valid.clone()]; + entries.extend(invalid); + let result = insert_otel_spans_batch(&pool, &entries).unwrap(); + assert_eq!( + result, + OtelTraceBatchResult { + accepted: 1, + duplicates: 0, + rejected: 8, + } + ); + assert_eq!(result.total(), entries.len()); + assert_eq!(row_count(&pool), 1); + + let conn = pool.get().unwrap(); + let only: (String, String) = conn + .query_row("SELECT trace_id, span_id FROM otel_spans", [], |row| { + Ok((row.get(0)?, row.get(1)?)) + }) + .unwrap(); + assert_eq!(only, (valid.trace_id, valid.span_id)); +} + +#[test] +fn metadata_size_and_flattened_field_bounds_reject_direct_db_bypass() { + let (_dir, pool) = pool("bounds.db"); + let mut oversized_json = span(0x11, 0x22); + oversized_json.attributes_json = serde_json::json!({ + "payload": "x".repeat(MAX_METADATA_JSON_BYTES) + }) + .to_string(); + + let mut oversized_name = span(0x12, 0x23); + oversized_name.span_name = "n".repeat(MAX_SPAN_NAME_CHARS + 1); + + let mut oversized_tool = span(0x13, 0x24); + oversized_tool.ai_tool = Some("t".repeat(MAX_TOOL_BYTES + 1)); + + let result = + insert_otel_spans_batch(&pool, &[oversized_json, oversized_name, oversized_tool]).unwrap(); + assert_eq!(result.accepted, 0); + assert_eq!(result.duplicates, 0); + assert_eq!(result.rejected, 3); + assert_eq!(row_count(&pool), 0); +} From 20bd4c86331d81a9b8c9c37a431c37a0db33a41d Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Wed, 19 Aug 2026 17:15:20 -0400 Subject: [PATCH 2/4] feat(otlp): ingest trace exports --- docs/plans/agent-observatory/proof/PROOF.md | 18 ++ src/db/otlp_traces.rs | 1 + src/otlp.rs | 89 ++++--- src/otlp/trace_http.rs | 234 ++++++++++++++++ src/otlp/traces.rs | 9 +- src/otlp_tests.rs | 281 ++++++++++++++++++-- src/runtime.rs | 5 + 7 files changed, 573 insertions(+), 64 deletions(-) create mode 100644 src/otlp/trace_http.rs diff --git a/docs/plans/agent-observatory/proof/PROOF.md b/docs/plans/agent-observatory/proof/PROOF.md index a10da94b..452e1c5f 100644 --- a/docs/plans/agent-observatory/proof/PROOF.md +++ b/docs/plans/agent-observatory/proof/PROOF.md @@ -508,3 +508,21 @@ FIX: adversarial review made nested-array truncation explicit, deterministic att PROOF: 6 focused privacy tests and 12 trace tests passed; ordered multi-event/link fixtures preserved exact fields/order, producer and Cortex truncation diagnostics were exact, invalid links/caps rejected safely, path/content opt-ins behaved as configured, and structural-secret negative fixtures proved no planted bearer-token values survived REGRESSION: complete OTLP library filter ran 68 tests with 0 failures on the definitive locked harness GATE: locked workspace Clippy passed with `-D warnings`; canonical workspace rustfmt, full 500-line production Rust module-size gate, Agent Observatory golden contracts, `git diff --check`, and no Cargo.toml/Cargo.lock drift from AO-042 all passed + +## AO-044 Persist trace spans idempotently +commit/worktree SHA: dce43df4 (checkpoint committed) +RED: normalized spans had no durable write path, so repeat exports could not prove idempotency or distinguish duplicates from malformed records +GREEN: added one-transaction `otel_spans` batch persistence with `ON CONFLICT(trace_id, span_id) DO NOTHING`, shared bounded transient-lock retry, write serialization, and explicit accepted/duplicate/rejected accounting +FIX: direct DB-bypass validation rejects malformed/all-zero IDs, invalid timing, nonexistent run IDs, oversized flattened fields or metadata, and wrong JSON shapes per row without poisoning valid neighbors; the performance cleanup remains intact with no resurrected `OtelSpanRow` scaffold or blanket dead-code allowance +PROOF: focused locked `db::otlp_traces::tests` passed 5/5 covering empty no-op, repeat-export idempotency, same-batch duplicates, malformed-neighbor isolation, and metadata/flattened-field bounds +GATE: pre-commit `diff_check`, `env_guard`, 500-line production module-size, and rustfmt hooks passed + +## AO-045 Mount functional /v1/traces +commit/worktree SHA: AO-045 checkpoint (this commit) +RED: authenticated `/v1/traces` still returned the deferred 404 and had no protobuf decode, bounded request handling, persistence, or OTLP partial-success response +GREEN: mounted an authenticated protobuf trace endpoint with an 8 MiB route-specific body cap, blocking decode/persistence offloaded through `spawn_blocking`, a 5,000-span request cap, AO-043 privacy-aware normalization, AO-044 idempotent persistence, and encoded `ExportTraceServiceResponse` output +PARTIAL: malformed individual spans, over-cap spans, configured storage-budget refusal, and direct-storage validation failures are counted as rejected without poisoning valid neighbors; duplicate exports remain successful and do not inflate rejection counts +STRUCTURE: extracted trace HTTP handling to a focused sidecar so `src/otlp.rs` is 293 lines and `src/otlp/trace_http.rs` is 234 lines, leaving runway for metrics without approaching the 500-line production module gate +PROOF: definitive locked handler suite passed 13/13 covering valid 200/protobuf, missing and invalid bearer 401, malformed protobuf 400, unsupported media 415, trace 8 MiB and preserved logs 4 MiB body-limit 413 plus Retry-After, invalid-span partial success, 5,000-span cap, storage-budget partial success, and duplicate idempotency +REGRESSION: definitive locked full OTLP library sweep passed 82/82, including all 5 AO-044 persistence tests plus existing auth/log/normalization/privacy/trace/runtime coverage +GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and full 500-line production Rust module-size gate passed; pre-push Clippy remains the push-time gate diff --git a/src/db/otlp_traces.rs b/src/db/otlp_traces.rs index 2ce4d0e4..6e0ec2b1 100644 --- a/src/db/otlp_traces.rs +++ b/src/db/otlp_traces.rs @@ -56,6 +56,7 @@ pub struct OtelTraceBatchResult { } impl OtelTraceBatchResult { + #[cfg(test)] pub const fn total(self) -> usize { self.accepted + self.duplicates + self.rejected } diff --git a/src/otlp.rs b/src/otlp.rs index 0b6edec9..27d71c18 100644 --- a/src/otlp.rs +++ b/src/otlp.rs @@ -1,16 +1,18 @@ -//! OTLP/HTTP receiver — accepts OpenTelemetry log records over HTTP and feeds -//! them into the existing cortex ingest pipeline. Logs only — `/v1/traces` -//! returns 404 (deferred) and `/v1/metrics` returns 404 (deferred). +//! OTLP/HTTP receiver for OpenTelemetry logs and traces. Logs feed the existing +//! Cortex ingest pipeline; traces normalize into the Agent Observatory span store. +//! `/v1/metrics` remains deferred. //! -//! Mounted on the same axum server as MCP. Body limit: 4 MiB. Bearer auth uses -//! the same static MCP token as `/mcp` — `config.mcp.api_token`, set via +//! Mounted on the same axum server as MCP. Logs retain a 4 MiB body limit; +//! traces and the future metrics endpoint use 8 MiB. Bearer auth uses the same +//! static MCP token as `/mcp` — `config.mcp.api_token`, set via //! `CORTEX_TOKEN`. It is NOT `CORTEX_API_TOKEN` (that is `config.api.api_token`, //! the separate REST `/api/*` token) and NOT `CORTEX_API_ADMIN_TOKEN`. //! Loopback / trusted-gateway policies skip the check; OAuth-only deployments //! with no static token deny OTLP outright (no OAuth flow for exporters). //! -//! Request → response wiring lives here; `AnyValue`/`LogBatchEntry` -//! conversion is in [`entries`] and the bearer-token gate is in [`auth`]. +//! Log request wiring lives here; trace HTTP handling is in [`trace_http`], +//! `AnyValue`/`LogBatchEntry` conversion is in [`entries`], and the bearer-token +//! gate is in [`auth`]. use std::net::SocketAddr; use std::sync::Arc; @@ -27,16 +29,20 @@ use axum::{ }; use bytes::Bytes; use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use parking_lot::Mutex; use prost::Message; use serde_json::json; use tower_http::limit::RequestBodyLimitLayer; +use crate::config::AgentObservatoryPrivacyConfig; +use crate::db::{DbPool, StorageBudgetState}; use crate::ingest::IngestTx; mod auth; mod entries; mod normalization; mod privacy; +mod trace_http; mod traces; use auth::{ @@ -44,10 +50,14 @@ use auth::{ unauthorized_diagnostics, }; use entries::build_entries; +#[cfg(test)] +use trace_http::MAX_SPANS_PER_REQUEST; +use trace_http::{TraceIngestState, traces_handler}; -/// Per-request body cap. Matches the OpenTelemetry Collector default for -/// HTTP receivers. Larger payloads receive 413 + `Retry-After: 86400`. +/// Existing `/v1/logs` body cap. Larger payloads receive 413 + `Retry-After: 86400`. pub const OTLP_BODY_LIMIT_BYTES: usize = 4 * 1024 * 1024; +/// Agent Observatory trace and metric body cap. +pub const OTLP_SIGNAL_BODY_LIMIT_BYTES: usize = 8 * 1024 * 1024; /// Atomic counters for the OTLP receiver, surfaced via `/health`. #[derive(Debug, Default)] @@ -63,6 +73,7 @@ pub struct OtlpState { pub api_token: Option, pub counters: Arc, pub auth_policy: AuthPolicy, + trace_ingest: Option, } impl OtlpState { @@ -77,19 +88,37 @@ impl OtlpState { api_token, counters, auth_policy, + trace_ingest: None, } } + + pub(crate) fn with_trace_ingest( + mut self, + pool: Arc, + storage_state: Arc>>, + privacy: AgentObservatoryPrivacyConfig, + ) -> Self { + self.trace_ingest = Some(TraceIngestState::new(pool, storage_state, privacy)); + self + } } -/// Build the OTLP router. Mounts `/v1/logs` (functional ingest), -/// `/v1/metrics` (404 — deferred), `/v1/traces` (404 — deferred) on the same -/// axum server as MCP. +/// Build the OTLP router. Logs retain their existing 4 MiB body cap while +/// traces and the future metrics endpoint use the Agent Observatory 8 MiB cap. pub fn router(state: OtlpState) -> Router { Router::new() - .route("/v1/logs", post(logs_handler)) - .route("/v1/metrics", post(metrics_handler)) - .route("/v1/traces", post(traces_handler)) - .layer(RequestBodyLimitLayer::new(OTLP_BODY_LIMIT_BYTES)) + .route( + "/v1/logs", + post(logs_handler).layer(RequestBodyLimitLayer::new(OTLP_BODY_LIMIT_BYTES)), + ) + .route( + "/v1/metrics", + post(metrics_handler).layer(RequestBodyLimitLayer::new(OTLP_SIGNAL_BODY_LIMIT_BYTES)), + ) + .route( + "/v1/traces", + post(traces_handler).layer(RequestBodyLimitLayer::new(OTLP_SIGNAL_BODY_LIMIT_BYTES)), + ) .layer(from_fn(add_retry_after_on_413)) .with_state(state) } @@ -253,33 +282,7 @@ async fn metrics_handler( StatusCode::NOT_FOUND, Json(json!({ "error": "metrics_not_supported", - "message": "OTLP metrics deferred. Use /v1/logs only." - })), - ) - .into_response() -} - -async fn traces_handler( - State(state): State, - headers: HeaderMap, -) -> axum::response::Response { - if !is_authorized(&state, &headers) { - return unauthorized(); - } - let content_length = headers - .get("content-length") - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); - tracing::warn!( - content_length, - "OTLP traces received but traces ingestion is not supported" - ); - ( - StatusCode::NOT_FOUND, - Json(json!({ - "error": "traces_not_supported", - "message": "OTLP traces deferred. Use /v1/logs only." + "message": "OTLP metrics ingestion is deferred." })), ) .into_response() diff --git a/src/otlp/trace_http.rs b/src/otlp/trace_http.rs new file mode 100644 index 00000000..fea9980c --- /dev/null +++ b/src/otlp/trace_http.rs @@ -0,0 +1,234 @@ +//! OTLP/HTTP trace request handling and Agent Observatory persistence. + +use std::net::SocketAddr; +use std::sync::Arc; + +use axum::{ + extract::{ConnectInfo, State}, + http::{HeaderMap, HeaderValue, StatusCode, header::CONTENT_TYPE}, + response::{IntoResponse, Json}, +}; +use bytes::Bytes; +use opentelemetry_proto::tonic::collector::trace::v1::{ + ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse, +}; +use parking_lot::Mutex; +use prost::Message; +use serde_json::json; + +use crate::config::AgentObservatoryPrivacyConfig; +use crate::db::{DbPool, StorageBudgetState}; + +use super::OtlpState; +use super::auth::{is_authorized, unauthorized}; +use super::traces::normalize_span_with_privacy; + +/// Maximum spans accepted from one OTLP trace request. +pub(super) const MAX_SPANS_PER_REQUEST: usize = 5_000; + +#[derive(Clone)] +pub(super) struct TraceIngestState { + pool: Arc, + storage_state: Arc>>, + privacy: AgentObservatoryPrivacyConfig, +} + +impl TraceIngestState { + pub(super) fn new( + pool: Arc, + storage_state: Arc>>, + privacy: AgentObservatoryPrivacyConfig, + ) -> Self { + Self { + pool, + storage_state, + privacy, + } + } +} + +pub(super) async fn traces_handler( + State(state): State, + ConnectInfo(peer): ConnectInfo, + headers: HeaderMap, + body: Bytes, +) -> axum::response::Response { + if !is_authorized(&state, &headers) { + return unauthorized(); + } + if !is_protobuf_content_type(&headers) { + return ( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + Json(json!({"error": "unsupported_content_type"})), + ) + .into_response(); + } + + let Some(trace_ingest) = state.trace_ingest.clone() else { + tracing::error!("OTLP trace ingest state is unavailable"); + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"error": "trace_ingest_unavailable"})), + ) + .into_response(); + }; + + let decoded = + tokio::task::spawn_blocking(move || ExportTraceServiceRequest::decode(body)).await; + let req = match decoded { + Ok(Ok(req)) => req, + Ok(Err(err)) => { + state + .counters + .decode_errors + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tracing::warn!(error = %err, source_ip = %peer, "OTLP /v1/traces decode failed"); + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "decode_failed"})), + ) + .into_response(); + } + Err(err) => { + state + .counters + .decode_errors + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tracing::error!(error = %err, "OTLP trace decode task panicked"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "internal"})), + ) + .into_response(); + } + }; + + let received_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let mut normalized = Vec::new(); + let mut rejected = 0usize; + let mut seen = 0usize; + let mut rejected_invalid = false; + let mut rejected_over_cap = false; + + for resource_spans in &req.resource_spans { + for scope_spans in &resource_spans.scope_spans { + for span in &scope_spans.spans { + seen += 1; + if seen > MAX_SPANS_PER_REQUEST { + rejected += 1; + rejected_over_cap = true; + continue; + } + match normalize_span_with_privacy( + resource_spans.resource.as_ref(), + &resource_spans.schema_url, + scope_spans.scope.as_ref(), + &scope_spans.schema_url, + span, + &trace_ingest.privacy, + &received_at, + ) { + Ok(span) => normalized.push(span), + Err(error) => { + rejected += 1; + rejected_invalid = true; + tracing::debug!(error = %error, source_ip = %peer, "Rejected invalid OTLP span"); + } + } + } + } + } + + let mut messages = Vec::new(); + if rejected_invalid { + messages.push("invalid spans rejected"); + } + if rejected_over_cap { + messages.push("request exceeded 5000 span limit"); + } + + if trace_ingest + .storage_state + .lock() + .as_ref() + .is_some_and(|state| state.write_blocked) + { + rejected += normalized.len(); + if !normalized.is_empty() { + messages.push("trace storage temporarily blocked by configured storage budget"); + } + tracing::warn!( + source_ip = %peer, + rejected, + "OTLP trace persistence blocked by storage budget" + ); + return trace_success_response(rejected, &messages); + } + + let pool = Arc::clone(&trace_ingest.pool); + let persisted = tokio::task::spawn_blocking(move || { + crate::db::otlp_traces::insert_otel_spans_batch(&pool, &normalized) + }) + .await; + let result = match persisted { + Ok(Ok(result)) => result, + Ok(Err(error)) => { + tracing::error!(error = %error, source_ip = %peer, "OTLP trace persistence failed"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "trace_persistence_failed"})), + ) + .into_response(); + } + Err(error) => { + tracing::error!(error = %error, "OTLP trace persistence task panicked"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "internal"})), + ) + .into_response(); + } + }; + + rejected += result.rejected; + if result.rejected > 0 { + messages.push("spans rejected by storage validation"); + } + tracing::info!( + source_ip = %peer, + accepted = result.accepted, + duplicates = result.duplicates, + rejected, + "OTLP traces ingested" + ); + trace_success_response(rejected, &messages) +} + +fn is_protobuf_content_type(headers: &HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|media_type| { + media_type + .trim() + .eq_ignore_ascii_case("application/x-protobuf") + }) +} + +fn trace_success_response(rejected: usize, messages: &[&str]) -> axum::response::Response { + let partial_success = (rejected > 0).then(|| ExportTracePartialSuccess { + rejected_spans: i64::try_from(rejected).unwrap_or(i64::MAX), + error_message: messages.join("; "), + }); + let response = ExportTraceServiceResponse { partial_success }; + ( + StatusCode::OK, + [( + CONTENT_TYPE, + HeaderValue::from_static("application/x-protobuf"), + )], + Bytes::from(response.encode_to_vec()), + ) + .into_response() +} diff --git a/src/otlp/traces.rs b/src/otlp/traces.rs index 2f975fb9..eb5e4655 100644 --- a/src/otlp/traces.rs +++ b/src/otlp/traces.rs @@ -1,11 +1,7 @@ //! OTLP trace-span normalization into the Agent Observatory DB input contract. //! -//! HTTP decoding/persistence is intentionally deferred to later Phase 3 slices. -//! This module owns the pure, deterministic conversion of one protobuf span. - -// AO-042 intentionally lands the pure converter before AO-045 wires the HTTP -// handler. Keep the production seam compiled now without pretending it is live. -#![allow(dead_code)] +//! This module owns the pure, deterministic conversion of one protobuf span; +//! the HTTP receiver applies it before idempotent persistence. use std::fmt::Write as _; @@ -180,6 +176,7 @@ fn resource_scope_json( } /// Normalize one OTLP protobuf span with the default Agent Observatory privacy policy. +#[cfg(test)] pub(crate) fn normalize_span( resource: Option<&Resource>, resource_schema_url: &str, diff --git a/src/otlp_tests.rs b/src/otlp_tests.rs index e164a250..f3728dff 100644 --- a/src/otlp_tests.rs +++ b/src/otlp_tests.rs @@ -1,51 +1,302 @@ -//! Handler-level tests for the OTLP HTTP receiver (status-code contract for -//! the deferred `/v1/metrics` and `/v1/traces` routes, and counters). Pure -//! `AnyValue`/`build_entries` logic lives in `otlp::entries`'s sidecar tests; -//! the bearer-token gate lives in `otlp::auth`'s sidecar tests. +//! Handler-level tests for the OTLP HTTP receiver. use super::*; +use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; use std::sync::atomic::Ordering; -fn state_with_token(token: Option<&str>) -> OtlpState { +use axum::body::{Body, to_bytes}; +use axum::http::{ + Request, + header::{AUTHORIZATION, CONTENT_TYPE, RETRY_AFTER}, +}; +use opentelemetry_proto::tonic::collector::trace::v1::{ + ExportTraceServiceRequest, ExportTraceServiceResponse, +}; +use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span}; +use parking_lot::Mutex; +use tower::util::ServiceExt; + +use crate::config::StorageConfig; +use crate::db::{DbPool, StorageBudgetState, get_storage_metrics, init_pool}; + +struct TestOtlpState { + _dir: tempfile::TempDir, + state: OtlpState, + pool: Arc, + storage: StorageConfig, + storage_state: Arc>>, +} + +fn state_with_token(token: Option<&str>) -> TestOtlpState { + let dir = tempfile::tempdir().unwrap(); + let storage = StorageConfig::for_test(dir.path().join("otlp.db")); + let pool = Arc::new(init_pool(&storage).unwrap()); + let storage_state = Arc::new(Mutex::new(None)); let (tx, _rx) = tokio::sync::mpsc::channel::(10); let ingest = crate::ingest::IngestTx::from_sender_for_test(tx); - // Use Mounted when a token is configured so is_authorized enforces it. let auth_policy = if token.is_some() { crate::mcp::AuthPolicy::Mounted { auth_state: None } } else { crate::mcp::AuthPolicy::LoopbackDev }; - OtlpState::new( + let state = OtlpState::new( ingest, token.map(String::from), Arc::new(OtlpCounters::default()), auth_policy, ) + .with_trace_ingest( + Arc::clone(&pool), + Arc::clone(&storage_state), + AgentObservatoryPrivacyConfig::default(), + ); + TestOtlpState { + _dir: dir, + state, + pool, + storage, + storage_state, + } +} + +fn peer() -> ConnectInfo { + ConnectInfo(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4318)) +} + +fn protobuf_headers(token: Option<&str>) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert( + CONTENT_TYPE, + HeaderValue::from_static("application/x-protobuf"), + ); + if let Some(token) = token { + headers.insert( + AUTHORIZATION, + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + } + headers +} + +fn span(id: u64) -> Span { + Span { + trace_id: vec![0x11; 16], + span_id: id.to_be_bytes().to_vec(), + name: format!("span-{id}"), + start_time_unix_nano: 1_700_000_000_000_000_000, + end_time_unix_nano: 1_700_000_000_000_001_000, + ..Default::default() + } +} + +fn trace_request(spans: Vec) -> ExportTraceServiceRequest { + ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + resource: None, + scope_spans: vec![ScopeSpans { + scope: None, + spans, + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } +} + +async fn call_traces( + state: &OtlpState, + headers: HeaderMap, + request: ExportTraceServiceRequest, +) -> axum::response::Response { + traces_handler( + State(state.clone()), + peer(), + headers, + Bytes::from(request.encode_to_vec()), + ) + .await +} + +async fn decode_trace_response(response: axum::response::Response) -> ExportTraceServiceResponse { + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers().get(CONTENT_TYPE).unwrap(), + "application/x-protobuf" + ); + let body = to_bytes(response.into_body(), 64 * 1024).await.unwrap(); + ExportTraceServiceResponse::decode(body).unwrap() +} + +fn span_rows(pool: &DbPool) -> i64 { + pool.get() + .unwrap() + .query_row("SELECT COUNT(*) FROM otel_spans", [], |row| row.get(0)) + .unwrap() } #[tokio::test] async fn metrics_handler_returns_not_supported() { - let response = metrics_handler(State(state_with_token(None)), HeaderMap::new()).await; + let test = state_with_token(None); + let response = metrics_handler(State(test.state), HeaderMap::new()).await; assert_eq!(response.status(), StatusCode::NOT_FOUND); } #[tokio::test] async fn traces_handler_requires_bearer_when_token_configured() { - let response = traces_handler(State(state_with_token(Some("secret"))), HeaderMap::new()).await; + let test = state_with_token(Some("secret")); + let response = call_traces(&test.state, HeaderMap::new(), trace_request(vec![span(1)])).await; assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(span_rows(&test.pool), 0); } #[tokio::test] -async fn traces_handler_returns_not_supported_after_auth() { +async fn traces_handler_rejects_invalid_bearer() { + let test = state_with_token(Some("secret")); + let response = call_traces( + &test.state, + protobuf_headers(Some("wrong")), + trace_request(vec![span(1)]), + ) + .await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(span_rows(&test.pool), 0); +} + +#[tokio::test] +async fn traces_handler_rejects_unsupported_content_type() { + let test = state_with_token(None); let mut headers = HeaderMap::new(); - headers.insert( - axum::http::header::AUTHORIZATION, - HeaderValue::from_static("Bearer secret"), + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + let response = call_traces(&test.state, headers, trace_request(vec![span(1)])).await; + assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); + assert_eq!(span_rows(&test.pool), 0); +} + +#[tokio::test] +async fn traces_handler_rejects_malformed_protobuf() { + let test = state_with_token(None); + let response = traces_handler( + State(test.state.clone()), + peer(), + protobuf_headers(None), + Bytes::from_static(&[0xff]), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(span_rows(&test.pool), 0); + assert_eq!(test.state.counters.decode_errors.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn traces_handler_persists_valid_protobuf_and_returns_otlp_response() { + let test = state_with_token(Some("secret")); + let response = call_traces( + &test.state, + protobuf_headers(Some("secret")), + trace_request(vec![span(1)]), + ) + .await; + let decoded = decode_trace_response(response).await; + assert!(decoded.partial_success.is_none()); + assert_eq!(span_rows(&test.pool), 1); +} + +#[tokio::test] +async fn traces_handler_reports_invalid_span_as_partial_success() { + let test = state_with_token(None); + let mut invalid = span(2); + invalid.trace_id = vec![0; 16]; + let response = call_traces( + &test.state, + protobuf_headers(None), + trace_request(vec![span(1), invalid]), + ) + .await; + let decoded = decode_trace_response(response).await; + let partial = decoded.partial_success.unwrap(); + assert_eq!(partial.rejected_spans, 1); + assert!(partial.error_message.contains("invalid spans rejected")); + assert_eq!(span_rows(&test.pool), 1); +} + +#[tokio::test] +async fn duplicate_trace_export_is_successful_and_idempotent() { + let test = state_with_token(None); + let request = trace_request(vec![span(1)]); + let first = call_traces(&test.state, protobuf_headers(None), request.clone()).await; + assert!(decode_trace_response(first).await.partial_success.is_none()); + let second = call_traces(&test.state, protobuf_headers(None), request).await; + assert!( + decode_trace_response(second) + .await + .partial_success + .is_none() ); - let response = traces_handler(State(state_with_token(Some("secret"))), headers).await; - assert_eq!(response.status(), StatusCode::NOT_FOUND); + assert_eq!(span_rows(&test.pool), 1); +} + +#[tokio::test] +async fn trace_span_cap_rejects_only_excess_spans() { + let test = state_with_token(None); + let spans = (1..=(MAX_SPANS_PER_REQUEST as u64 + 1)).map(span).collect(); + let response = call_traces(&test.state, protobuf_headers(None), trace_request(spans)).await; + let decoded = decode_trace_response(response).await; + let partial = decoded.partial_success.unwrap(); + assert_eq!(partial.rejected_spans, 1); + assert!(partial.error_message.contains("5000 span limit")); + assert_eq!(span_rows(&test.pool), MAX_SPANS_PER_REQUEST as i64); +} + +#[tokio::test] +async fn storage_budget_block_is_otlp_partial_success_not_server_failure() { + let test = state_with_token(None); + let metrics = get_storage_metrics(&test.pool, &test.storage).unwrap(); + *test.storage_state.lock() = Some(StorageBudgetState { + metrics, + write_blocked: true, + }); + let response = call_traces( + &test.state, + protobuf_headers(None), + trace_request(vec![span(1)]), + ) + .await; + let decoded = decode_trace_response(response).await; + let partial = decoded.partial_success.unwrap(); + assert_eq!(partial.rejected_spans, 1); + assert!(partial.error_message.contains("storage budget")); + assert_eq!(span_rows(&test.pool), 0); +} + +#[tokio::test] +async fn traces_router_enforces_eight_mib_body_limit_with_retry_after() { + let test = state_with_token(None); + let request = Request::builder() + .method("POST") + .uri("/v1/traces") + .header(CONTENT_TYPE, "application/x-protobuf") + .extension(peer()) + .body(Body::from(vec![0_u8; OTLP_SIGNAL_BODY_LIMIT_BYTES + 1])) + .unwrap(); + let response = router(test.state).oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!(response.headers().get(RETRY_AFTER).unwrap(), "86400"); +} + +#[tokio::test] +async fn logs_router_preserves_four_mib_body_limit_with_retry_after() { + let test = state_with_token(None); + let request = Request::builder() + .method("POST") + .uri("/v1/logs") + .extension(peer()) + .body(Body::from(vec![0_u8; OTLP_BODY_LIMIT_BYTES + 1])) + .unwrap(); + let response = router(test.state).oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!(response.headers().get(RETRY_AFTER).unwrap(), "86400"); } #[test] diff --git a/src/runtime.rs b/src/runtime.rs index b7fcfe2f..cb29d103 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -344,6 +344,11 @@ impl RuntimeCore { self.config.mcp.api_token.0.clone(), Arc::clone(&self.otlp_counters), self.auth_policy.clone(), + ) + .with_trace_ingest( + Arc::clone(&self.pool), + Arc::clone(&self.storage_state), + self.config.agent_observatory.privacy.clone(), ); otlp::router(state) } From a5d0de6130527ae466dc25e9f645d88c20c01391 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Wed, 19 Aug 2026 19:46:21 -0400 Subject: [PATCH 3/4] feat(otlp): normalize metric number points --- Cargo.toml | 1 + docs/contracts/agent-observatory.md | 7 +- docs/plans/agent-observatory/proof/PROOF.md | 13 +- src/otlp.rs | 2 + src/otlp/metrics.rs | 424 ++++++++++++++++++++ src/otlp/metrics_payload.rs | 151 +++++++ src/otlp/metrics_tests.rs | 340 ++++++++++++++++ 7 files changed, 934 insertions(+), 4 deletions(-) create mode 100644 src/otlp/metrics.rs create mode 100644 src/otlp/metrics_payload.rs create mode 100644 src/otlp/metrics_tests.rs diff --git a/Cargo.toml b/Cargo.toml index 78bde281..3ef18eed 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -171,6 +171,7 @@ serde_path_to_error = "0.1" opentelemetry-proto = { version = "0.32", default-features = false, features = [ "gen-tonic-messages", "logs", + "metrics", "trace", ] } prost = "0.14" diff --git a/docs/contracts/agent-observatory.md b/docs/contracts/agent-observatory.md index fbd34bbd..48c52fd6 100644 --- a/docs/contracts/agent-observatory.md +++ b/docs/contracts/agent-observatory.md @@ -79,11 +79,12 @@ Source kinds and variants are ASCII lower snake case. Event keys are determinist The metric point key is SHA-256 hex over the canonical tuple: ```text -resource fingerprint, scope, metric name, instrument kind, -start timestamp, point timestamp, sorted attributes, value, exemplar IDs +resource fingerprint, scope, metric name, instrument kind, unit, +aggregation temporality, monotonicity, start timestamp, point timestamp, +sorted attributes, value, exemplar IDs ``` -This makes repeated OTLP export idempotent without assuming producer point IDs. +Unit and applicable aggregation temporality/monotonicity are stream-identifying properties and therefore participate in the key; description does not. `value` is the canonical bounded point payload, including data-point flags. This makes repeated OTLP export idempotent without assuming producer point IDs. ## 3. Enumerations diff --git a/docs/plans/agent-observatory/proof/PROOF.md b/docs/plans/agent-observatory/proof/PROOF.md index 452e1c5f..f84c1194 100644 --- a/docs/plans/agent-observatory/proof/PROOF.md +++ b/docs/plans/agent-observatory/proof/PROOF.md @@ -525,4 +525,15 @@ PARTIAL: malformed individual spans, over-cap spans, configured storage-budget r STRUCTURE: extracted trace HTTP handling to a focused sidecar so `src/otlp.rs` is 293 lines and `src/otlp/trace_http.rs` is 234 lines, leaving runway for metrics without approaching the 500-line production module gate PROOF: definitive locked handler suite passed 13/13 covering valid 200/protobuf, missing and invalid bearer 401, malformed protobuf 400, unsupported media 415, trace 8 MiB and preserved logs 4 MiB body-limit 413 plus Retry-After, invalid-span partial success, 5,000-span cap, storage-budget partial success, and duplicate idempotency REGRESSION: definitive locked full OTLP library sweep passed 82/82, including all 5 AO-044 persistence tests plus existing auth/log/normalization/privacy/trace/runtime coverage -GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and full 500-line production Rust module-size gate passed; pre-push Clippy remains the push-time gate +GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and full 500-line production Rust module-size gate passed; real pre-push `cargo clippy --all-targets --all-features --locked -- -D warnings` passed before the AO-045 push + +## AO-046 Normalize gauge and sum points +commit/worktree SHA: AO-046 checkpoint (this commit) +RED: Cortex did not enable the pinned `opentelemetry-proto` metrics message feature and had no generic normalized metric-point input, number-point converter, or deterministic point key implementation +GREEN: enabled only the existing pinned metrics feature with no lockfile churn; added privacy-aware integer/double gauge and sum conversion with exact timestamps, gauge start-time semantics, raw temporality integers, monotonicity, provider/session/project identity, canonical resource/scope metadata including entity refs, bounded/sorted point attributes, exemplars, and JSON-safe non-finite double tokens +IDEMPOTENCY: SHA-256 point keys use fixed-width component framing over canonical resource fingerprint, scope, stream identity, timestamps, sorted attributes, bounded value payload, and sorted exemplar IDs; resource/point/entity-ref input order does not affect the key +FIX: adversarial review found the original Cortex point-key contract omitted OpenTelemetry stream-identifying unit, aggregation temporality, and monotonicity, which could collapse distinct streams; corrected contract section 2.6 and implementation, preserved data-point flags inside `value_json`, and proved description remains non-identifying +STRUCTURE: split canonical exemplar/value/key encoding into `metrics_payload.rs`; production metric modules remain comfortably below the 500-line gate +PROOF: definitive locked focused metric normalization suite passed 8/8, including gauge/sum semantics, deterministic reorder-insensitive keys, stream-identity/flags collision resistance, exemplar validation, non-finite doubles, privacy policy, and fail-closed invalid fields +REGRESSION: definitive locked full OTLP library sweep passed 90/90 with all prior logs/traces/auth/privacy/runtime/database coverage intact +GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and the full 500-line production Rust module-size gate passed; exact `cargo clippy --all-targets --all-features --locked -- -D warnings` passed after structurally reducing the point-normalizer argument surface and fixing the test config construction lint diff --git a/src/otlp.rs b/src/otlp.rs index 27d71c18..ccba509c 100644 --- a/src/otlp.rs +++ b/src/otlp.rs @@ -40,6 +40,8 @@ use crate::ingest::IngestTx; mod auth; mod entries; +mod metrics; +mod metrics_payload; mod normalization; mod privacy; mod trace_http; diff --git a/src/otlp/metrics.rs b/src/otlp/metrics.rs new file mode 100644 index 00000000..84dfff61 --- /dev/null +++ b/src/otlp/metrics.rs @@ -0,0 +1,424 @@ +//! Pure OTLP gauge/sum normalization for Agent Observatory metric points. + +use super::metrics_payload::{ + PointKeyParts, exemplar_ids, number_value, point_key, serialize_exemplars, +}; +use super::normalization::{MAX_RESOURCE_ATTRIBUTES, MAX_SIGNAL_ATTRIBUTES, normalize_attributes}; +use super::privacy::{private_attributes, private_text}; +use crate::config::AgentObservatoryPrivacyConfig; +use opentelemetry_proto::tonic::{ + common::v1::{EntityRef, InstrumentationScope}, + metrics::v1::{Metric, NumberDataPoint, metric}, + resource::v1::Resource, +}; +use serde_json::{Value, json}; +use thiserror::Error; + +const MAX_METADATA_JSON_BYTES: usize = 256 * 1024; +const MAX_METRIC_NAME_CHARS: usize = 512; +const MAX_DESCRIPTION_CHARS: usize = 4096; +const MAX_UNIT_CHARS: usize = 128; +const MAX_SCOPE_NAME_CHARS: usize = 512; +const MAX_SCOPE_VERSION_CHARS: usize = 512; +const MAX_HOSTNAME_CHARS: usize = 255; +const MAX_SERVICE_NAME_CHARS: usize = 512; +const MAX_SERVICE_VERSION_CHARS: usize = 512; +const MAX_EXEMPLARS: usize = 128; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct MetricPointInput { + pub point_key: String, + pub metric_name: String, + pub description: String, + pub unit: String, + pub instrument_kind: String, + pub aggregation_temporality: Option, + pub monotonic: Option, + pub start_time_unix_nano: Option, + pub time_unix_nano: i64, + pub hostname: String, + pub service_name: Option, + pub service_version: Option, + pub scope_name: Option, + pub scope_version: Option, + pub ai_tool: Option, + pub ai_project: Option, + pub ai_session_id: Option, + pub run_id: Option, + pub resource_json: String, + pub attributes_json: String, + pub value_json: String, + pub exemplars_json: String, + pub received_at: String, + pub content_scrubbed: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub(crate) enum MetricNormalizeError { + #[error("metric kind is not a gauge or sum")] + UnsupportedInstrument, + #[error("metric name must not be empty")] + EmptyMetricName, + #[error("number data point has no value")] + MissingValue, + #[error("sum aggregation temporality must not be unspecified")] + UnspecifiedTemporality, + #[error("{field} contains {actual} attributes; maximum is {maximum}")] + AttributeLimit { + field: &'static str, + actual: usize, + maximum: usize, + }, + #[error("point contains {actual} exemplars; maximum is {maximum}")] + ExemplarLimit { actual: usize, maximum: usize }, + #[error("{field} must be empty or exactly {expected} non-zero bytes; got {actual}")] + InvalidOptionalId { + field: &'static str, + expected: usize, + actual: usize, + }, + #[error("{field} exceeds maximum length {maximum}")] + FieldTooLong { field: &'static str, maximum: usize }, + #[error("{field} does not fit SQLite INTEGER")] + IntegerOverflow { field: &'static str }, + #[error("point time must be non-zero")] + MissingPointTime, + #[error("point time precedes start time")] + TimeBeforeStart, + #[error("received_at must be RFC3339")] + InvalidReceivedAt, + #[error("{field} JSON is {actual} bytes; maximum is {maximum}")] + MetadataTooLarge { + field: &'static str, + actual: usize, + maximum: usize, + }, +} + +#[derive(Clone, Copy)] +struct NumberMetricContext<'a> { + resource: Option<&'a Resource>, + resource_schema_url: &'a str, + scope: Option<&'a InstrumentationScope>, + scope_schema_url: &'a str, + metric: &'a Metric, + privacy: &'a AgentObservatoryPrivacyConfig, + received_at: &'a str, + instrument_kind: &'static str, + aggregation_temporality: Option, + monotonic: Option, + ignore_start_time: bool, +} + +#[cfg(test)] +pub(crate) fn normalize_number_metric( + resource: Option<&Resource>, + resource_schema_url: &str, + scope: Option<&InstrumentationScope>, + scope_schema_url: &str, + metric: &Metric, + received_at: &str, +) -> Result, MetricNormalizeError> { + normalize_number_metric_with_privacy( + resource, + resource_schema_url, + scope, + scope_schema_url, + metric, + &AgentObservatoryPrivacyConfig::default(), + received_at, + ) +} + +/// AO-046 stages this converter before AO-049/050 consume it from DB/HTTP paths. +#[allow(dead_code)] +pub(crate) fn normalize_number_metric_with_privacy( + resource: Option<&Resource>, + resource_schema_url: &str, + scope: Option<&InstrumentationScope>, + scope_schema_url: &str, + metric: &Metric, + privacy: &AgentObservatoryPrivacyConfig, + received_at: &str, +) -> Result, MetricNormalizeError> { + validate_metric_envelope(resource, scope, metric, received_at)?; + let (instrument_kind, aggregation_temporality, monotonic, ignore_start_time) = + match metric.data.as_ref() { + Some(metric::Data::Gauge(_)) => ("gauge", None, None, true), + Some(metric::Data::Sum(sum)) => { + if sum.aggregation_temporality == 0 { + return Err(MetricNormalizeError::UnspecifiedTemporality); + } + ( + "sum", + Some(sum.aggregation_temporality), + Some(sum.is_monotonic), + false, + ) + } + _ => return Err(MetricNormalizeError::UnsupportedInstrument), + }; + let context = NumberMetricContext { + resource, + resource_schema_url, + scope, + scope_schema_url, + metric, + privacy, + received_at, + instrument_kind, + aggregation_temporality, + monotonic, + ignore_start_time, + }; + let points = match metric.data.as_ref() { + Some(metric::Data::Gauge(gauge)) => &gauge.data_points, + Some(metric::Data::Sum(sum)) => &sum.data_points, + _ => unreachable!("instrument kind checked above"), + }; + points + .iter() + .map(|point| normalize_number_point(context, point)) + .collect() +} + +fn validate_metric_envelope( + resource: Option<&Resource>, + scope: Option<&InstrumentationScope>, + metric: &Metric, + received_at: &str, +) -> Result<(), MetricNormalizeError> { + if metric.name.is_empty() { + return Err(MetricNormalizeError::EmptyMetricName); + } + check_chars(&metric.name, MAX_METRIC_NAME_CHARS, "metric_name")?; + check_chars(&metric.description, MAX_DESCRIPTION_CHARS, "description")?; + check_chars(&metric.unit, MAX_UNIT_CHARS, "unit")?; + let resource_count = resource.map_or(0, |value| value.attributes.len()); + if resource_count > MAX_RESOURCE_ATTRIBUTES { + return Err(MetricNormalizeError::AttributeLimit { + field: "resource", + actual: resource_count, + maximum: MAX_RESOURCE_ATTRIBUTES, + }); + } + let scope_count = scope.map_or(0, |value| value.attributes.len()); + if scope_count > MAX_RESOURCE_ATTRIBUTES { + return Err(MetricNormalizeError::AttributeLimit { + field: "scope", + actual: scope_count, + maximum: MAX_RESOURCE_ATTRIBUTES, + }); + } + if let Some(scope) = scope { + check_chars(&scope.name, MAX_SCOPE_NAME_CHARS, "scope_name")?; + check_chars(&scope.version, MAX_SCOPE_VERSION_CHARS, "scope_version")?; + } + chrono::DateTime::parse_from_rfc3339(received_at) + .map_err(|_| MetricNormalizeError::InvalidReceivedAt)?; + Ok(()) +} + +fn normalize_number_point( + context: NumberMetricContext<'_>, + point: &NumberDataPoint, +) -> Result { + if point.attributes.len() > MAX_SIGNAL_ATTRIBUTES { + return Err(MetricNormalizeError::AttributeLimit { + field: "point", + actual: point.attributes.len(), + maximum: MAX_SIGNAL_ATTRIBUTES, + }); + } + if point.exemplars.len() > MAX_EXEMPLARS { + return Err(MetricNormalizeError::ExemplarLimit { + actual: point.exemplars.len(), + maximum: MAX_EXEMPLARS, + }); + } + if point.time_unix_nano == 0 { + return Err(MetricNormalizeError::MissingPointTime); + } + let time_unix_nano = checked_i64(point.time_unix_nano, "time_unix_nano")?; + let start_time_unix_nano = if context.ignore_start_time || point.start_time_unix_nano == 0 { + None + } else { + Some(checked_i64( + point.start_time_unix_nano, + "start_time_unix_nano", + )?) + }; + if start_time_unix_nano.is_some_and(|start| time_unix_nano < start) { + return Err(MetricNormalizeError::TimeBeforeStart); + } + let normalized = normalize_attributes( + context + .resource + .map_or(&[], |value| value.attributes.as_slice()), + &point.attributes, + ); + check_chars(&normalized.host_name, MAX_HOSTNAME_CHARS, "hostname")?; + if let Some(value) = normalized.service_name.as_deref() { + check_chars(value, MAX_SERVICE_NAME_CHARS, "service_name")?; + } + if let Some(value) = normalized.service_version.as_deref() { + check_chars(value, MAX_SERVICE_VERSION_CHARS, "service_version")?; + } + + let resource_value = resource_key_value( + context.resource, + context.resource_schema_url, + context.privacy, + ); + let scope_value = scope_key_value(context.scope, context.scope_schema_url, context.privacy); + let resource_json_value = json!({ + "resource": resource_value.clone(), + "scope": scope_value.clone(), + }); + let attributes_value = + private_attributes(&point.attributes, MAX_SIGNAL_ATTRIBUTES, context.privacy); + let value_value = number_value( + point + .value + .as_ref() + .ok_or(MetricNormalizeError::MissingValue)?, + point.flags, + ); + let exemplars_value = serialize_exemplars(&point.exemplars, context.privacy)?; + let resource_json = encode_json(resource_json_value, "resource")?; + let attributes_json = encode_json(attributes_value.clone(), "attributes")?; + let value_json = encode_json(value_value.clone(), "value")?; + let exemplars_json = encode_json(exemplars_value, "exemplars")?; + let exemplar_ids = exemplar_ids(&point.exemplars)?; + let metric_name = private_text(&context.metric.name); + let unit = private_text(&context.metric.unit); + let point_key = point_key(PointKeyParts { + resource: &resource_value, + scope: &scope_value, + metric_name: &metric_name, + instrument_kind: context.instrument_kind, + unit: &unit, + aggregation_temporality: context.aggregation_temporality, + monotonic: context.monotonic, + start_time_unix_nano, + time_unix_nano, + attributes: &attributes_value, + value: &value_value, + exemplar_ids: &exemplar_ids, + }); + + Ok(MetricPointInput { + point_key, + metric_name, + description: private_text(&context.metric.description), + unit, + instrument_kind: context.instrument_kind.to_string(), + aggregation_temporality: context.aggregation_temporality, + monotonic: context.monotonic, + start_time_unix_nano, + time_unix_nano, + hostname: private_text(&normalized.host_name), + service_name: normalized.service_name.map(|value| private_text(&value)), + service_version: normalized.service_version.map(|value| private_text(&value)), + scope_name: context + .scope + .and_then(|value| nonempty_private(&value.name)), + scope_version: context + .scope + .and_then(|value| nonempty_private(&value.version)), + ai_tool: normalized.ai_tool.map(|value| private_text(&value)), + ai_project: context + .privacy + .include_paths + .then_some(normalized.ai_project) + .flatten() + .map(|value| private_text(&value)), + ai_session_id: normalized.ai_session_id.map(|value| private_text(&value)), + run_id: None, + resource_json, + attributes_json, + value_json, + exemplars_json, + received_at: context.received_at.to_string(), + content_scrubbed: true, + }) +} + +fn resource_key_value( + resource: Option<&Resource>, + resource_schema_url: &str, + privacy: &AgentObservatoryPrivacyConfig, +) -> Value { + json!({ + "schema_url": private_text(resource_schema_url), + "attributes": resource.map_or_else(|| json!({}), |value| private_attributes(&value.attributes, MAX_RESOURCE_ATTRIBUTES, privacy)), + "dropped_attributes_count": resource.map_or(0, |value| value.dropped_attributes_count), + "entity_refs": resource.map_or_else(Vec::new, |value| canonical_entity_refs(&value.entity_refs)), + }) +} + +fn canonical_entity_refs(entity_refs: &[EntityRef]) -> Vec { + let mut values = entity_refs + .iter() + .map(|entity| { + let mut id_keys = entity.id_keys.clone(); + id_keys.sort(); + let mut description_keys = entity.description_keys.clone(); + description_keys.sort(); + json!({ + "schema_url": private_text(&entity.schema_url), + "type": private_text(&entity.r#type), + "id_keys": id_keys, + "description_keys": description_keys, + }) + }) + .collect::>(); + values.sort_by_cached_key(Value::to_string); + values +} + +fn scope_key_value( + scope: Option<&InstrumentationScope>, + scope_schema_url: &str, + privacy: &AgentObservatoryPrivacyConfig, +) -> Value { + json!({ + "schema_url": private_text(scope_schema_url), + "name": scope.map(|value| private_text(&value.name)), + "version": scope.map(|value| private_text(&value.version)), + "attributes": scope.map_or_else(|| json!({}), |value| private_attributes(&value.attributes, MAX_RESOURCE_ATTRIBUTES, privacy)), + "dropped_attributes_count": scope.map_or(0, |value| value.dropped_attributes_count), + }) +} + +pub(super) fn checked_i64(value: u64, field: &'static str) -> Result { + i64::try_from(value).map_err(|_| MetricNormalizeError::IntegerOverflow { field }) +} +fn check_chars( + value: &str, + maximum: usize, + field: &'static str, +) -> Result<(), MetricNormalizeError> { + if value.chars().count() > maximum { + return Err(MetricNormalizeError::FieldTooLong { field, maximum }); + } + Ok(()) +} +fn encode_json(value: Value, field: &'static str) -> Result { + let encoded = value.to_string(); + if encoded.len() > MAX_METADATA_JSON_BYTES { + return Err(MetricNormalizeError::MetadataTooLarge { + field, + actual: encoded.len(), + maximum: MAX_METADATA_JSON_BYTES, + }); + } + Ok(encoded) +} +fn nonempty_private(value: &str) -> Option { + (!value.is_empty()).then(|| private_text(value)) +} + +#[cfg(test)] +#[path = "metrics_tests.rs"] +mod tests; diff --git a/src/otlp/metrics_payload.rs b/src/otlp/metrics_payload.rs new file mode 100644 index 00000000..743b390d --- /dev/null +++ b/src/otlp/metrics_payload.rs @@ -0,0 +1,151 @@ +//! Canonical value, exemplar, and point-key encoding for OTLP metrics. + +use std::fmt::Write as _; + +use opentelemetry_proto::tonic::metrics::v1::{Exemplar, exemplar, number_data_point}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; + +use crate::config::AgentObservatoryPrivacyConfig; + +use super::metrics::{MetricNormalizeError, checked_i64}; +use super::normalization::MAX_SIGNAL_ATTRIBUTES; +use super::privacy::private_attributes; + +pub(super) fn serialize_exemplars( + exemplars: &[Exemplar], + privacy: &AgentObservatoryPrivacyConfig, +) -> Result { + exemplars + .iter() + .map(|value| { + Ok(json!({ + "filtered_attributes": private_attributes(&value.filtered_attributes, MAX_SIGNAL_ATTRIBUTES, privacy), + "time_unix_nano": checked_i64(value.time_unix_nano, "exemplar.time_unix_nano")?, + "trace_id": optional_hex_id(&value.trace_id, 16, "exemplar.trace_id")?, + "span_id": optional_hex_id(&value.span_id, 8, "exemplar.span_id")?, + "value": value.value.as_ref().map(exemplar_value).unwrap_or(Value::Null), + })) + }) + .collect::, _>>() + .map(Value::Array) +} + +pub(super) fn exemplar_ids(exemplars: &[Exemplar]) -> Result, MetricNormalizeError> { + let mut ids = exemplars + .iter() + .map(|value| { + let trace = + optional_hex_id(&value.trace_id, 16, "exemplar.trace_id")?.unwrap_or_default(); + let span = optional_hex_id(&value.span_id, 8, "exemplar.span_id")?.unwrap_or_default(); + Ok(format!("{trace}:{span}")) + }) + .collect::, MetricNormalizeError>>()?; + ids.sort(); + Ok(ids) +} + +pub(super) fn number_value(value: &number_data_point::Value, flags: u32) -> Value { + match value { + number_data_point::Value::AsInt(value) => { + json!({"type": "int", "value": value, "flags": flags}) + } + number_data_point::Value::AsDouble(value) => { + json!({"type": "double", "value": safe_double(*value), "flags": flags}) + } + } +} + +fn exemplar_value(value: &exemplar::Value) -> Value { + match value { + exemplar::Value::AsInt(value) => json!({"type": "int", "value": value}), + exemplar::Value::AsDouble(value) => json!({"type": "double", "value": safe_double(*value)}), + } +} + +fn safe_double(value: f64) -> Value { + if value.is_nan() { + return Value::String("nan".to_string()); + } + if value == f64::INFINITY { + return Value::String("+infinity".to_string()); + } + if value == f64::NEG_INFINITY { + return Value::String("-infinity".to_string()); + } + serde_json::Number::from_f64(value) + .map(Value::Number) + .unwrap_or(Value::Null) +} + +fn optional_hex_id( + bytes: &[u8], + expected: usize, + field: &'static str, +) -> Result, MetricNormalizeError> { + if bytes.is_empty() { + return Ok(None); + } + if bytes.len() != expected || bytes.iter().all(|byte| *byte == 0) { + return Err(MetricNormalizeError::InvalidOptionalId { + field, + expected, + actual: bytes.len(), + }); + } + let mut encoded = String::with_capacity(expected * 2); + for byte in bytes { + write!(&mut encoded, "{byte:02x}").expect("writing hex to String cannot fail"); + } + Ok(Some(encoded)) +} + +pub(super) struct PointKeyParts<'a> { + pub resource: &'a Value, + pub scope: &'a Value, + pub metric_name: &'a str, + pub instrument_kind: &'a str, + pub unit: &'a str, + pub aggregation_temporality: Option, + pub monotonic: Option, + pub start_time_unix_nano: Option, + pub time_unix_nano: i64, + pub attributes: &'a Value, + pub value: &'a Value, + pub exemplar_ids: &'a [String], +} + +pub(super) fn point_key(parts: PointKeyParts<'_>) -> String { + let resource_encoded = parts.resource.to_string(); + let resource_fingerprint = format!("{:x}", Sha256::digest(resource_encoded.as_bytes())); + let mut hasher = Sha256::new(); + for component in [ + resource_fingerprint, + parts.scope.to_string(), + parts.metric_name.to_string(), + parts.instrument_kind.to_string(), + parts.unit.to_string(), + parts + .aggregation_temporality + .map_or_else(String::new, |value| value.to_string()), + parts + .monotonic + .map_or_else(String::new, |value| value.to_string()), + parts + .start_time_unix_nano + .map_or_else(String::new, |value| value.to_string()), + parts.time_unix_nano.to_string(), + parts.attributes.to_string(), + parts.value.to_string(), + parts.exemplar_ids.join(","), + ] { + hash_component(&mut hasher, &component); + } + format!("{:x}", hasher.finalize()) +} + +fn hash_component(hasher: &mut Sha256, value: &str) { + let length = u64::try_from(value.len()).expect("metric point key component length fits u64"); + hasher.update(length.to_be_bytes()); + hasher.update(value.as_bytes()); +} diff --git a/src/otlp/metrics_tests.rs b/src/otlp/metrics_tests.rs new file mode 100644 index 00000000..c824ee92 --- /dev/null +++ b/src/otlp/metrics_tests.rs @@ -0,0 +1,340 @@ +use super::*; +use opentelemetry_proto::tonic::{ + common::v1::{ + AnyValue, EntityRef, InstrumentationScope, KeyValue, any_value::Value as AnyValueKind, + }, + metrics::v1::{ + AggregationTemporality, Exemplar, Gauge, Sum, exemplar, metric, number_data_point, + }, + resource::v1::Resource, +}; + +const RECEIVED_AT: &str = "2026-08-19T18:00:00.000Z"; + +fn kv(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(AnyValueKind::StringValue(value.to_string())), + }), + key_strindex: 0, + } +} +fn resource(attrs: Vec) -> Resource { + Resource { + attributes: attrs, + ..Default::default() + } +} +fn scope() -> InstrumentationScope { + InstrumentationScope { + name: "agent.metrics".into(), + version: "1.2.3".into(), + ..Default::default() + } +} +fn number_point(value: number_data_point::Value) -> NumberDataPoint { + NumberDataPoint { + attributes: vec![kv("z", "last"), kv("a", "first")], + start_time_unix_nano: 100, + time_unix_nano: 200, + value: Some(value), + ..Default::default() + } +} +fn gauge(point: NumberDataPoint) -> Metric { + Metric { + name: "agent.queue.depth".into(), + description: "queued work".into(), + unit: "{item}".into(), + data: Some(metric::Data::Gauge(Gauge { + data_points: vec![point], + })), + ..Default::default() + } +} +fn sum(point: NumberDataPoint, temporality: AggregationTemporality, monotonic: bool) -> Metric { + Metric { + name: "agent.tokens".into(), + description: "tokens".into(), + unit: "{token}".into(), + data: Some(metric::Data::Sum(Sum { + data_points: vec![point], + aggregation_temporality: temporality as i32, + is_monotonic: monotonic, + })), + ..Default::default() + } +} +fn normalize(metric: &Metric, resource: &Resource) -> MetricPointInput { + let scope = scope(); + normalize_number_metric( + Some(resource), + "resource/v1", + Some(&scope), + "scope/v1", + metric, + RECEIVED_AT, + ) + .unwrap() + .into_iter() + .next() + .unwrap() +} + +#[test] +fn integer_gauge_ignores_start_time_and_normalizes_identity() { + let resource = resource(vec![ + kv("service.name", "claude-code"), + kv("host.name", "dookie"), + ]); + let output = normalize( + &gauge(number_point(number_data_point::Value::AsInt(7))), + &resource, + ); + assert_eq!(output.instrument_kind, "gauge"); + assert_eq!(output.start_time_unix_nano, None); + assert_eq!(output.aggregation_temporality, None); + assert_eq!(output.monotonic, None); + assert_eq!(output.time_unix_nano, 200); + assert_eq!(output.hostname, "dookie"); + assert_eq!(output.ai_tool.as_deref(), Some("claude")); + assert_eq!( + serde_json::from_str::(&output.value_json).unwrap(), + json!({"type":"int","value":7,"flags":0}) + ); +} + +#[test] +fn cumulative_and_delta_sums_preserve_temporality_monotonic_and_start() { + let resource = resource(Vec::new()); + for (temporality, monotonic, value) in [ + ( + AggregationTemporality::Cumulative, + true, + number_data_point::Value::AsInt(42), + ), + ( + AggregationTemporality::Delta, + false, + number_data_point::Value::AsDouble(2.5), + ), + ] { + let output = normalize(&sum(number_point(value), temporality, monotonic), &resource); + assert_eq!(output.instrument_kind, "sum"); + assert_eq!(output.aggregation_temporality, Some(temporality as i32)); + assert_eq!(output.monotonic, Some(monotonic)); + assert_eq!(output.start_time_unix_nano, Some(100)); + } +} + +#[test] +fn repeated_fixture_and_reordered_attributes_have_same_point_key() { + let mut resource = resource(vec![kv("service.name", "codex"), kv("host.name", "dookie")]); + resource.entity_refs = vec![ + EntityRef { + schema_url: "entity/v1".into(), + r#type: "service".into(), + id_keys: vec!["service.name".into(), "host.name".into()], + description_keys: vec!["host.name".into()], + }, + EntityRef { + schema_url: "entity/v1".into(), + r#type: "host".into(), + id_keys: vec!["host.name".into()], + description_keys: Vec::new(), + }, + ]; + let metric = gauge(number_point(number_data_point::Value::AsInt(7))); + let first = normalize(&metric, &resource); + let second = normalize(&metric, &resource); + assert_eq!(first.point_key, second.point_key); + + let mut reordered_point = number_point(number_data_point::Value::AsInt(7)); + reordered_point.attributes.reverse(); + let third = normalize(&gauge(reordered_point), &resource); + assert_eq!(first.point_key, third.point_key); + assert_eq!(first.attributes_json, third.attributes_json); + + let mut reordered_resource = resource.clone(); + reordered_resource.attributes.reverse(); + reordered_resource.entity_refs.reverse(); + reordered_resource.entity_refs[1].id_keys.reverse(); + let fourth = normalize(&metric, &reordered_resource); + assert_eq!(first.point_key, fourth.point_key); + assert_eq!(first.resource_json, fourth.resource_json); +} + +#[test] +fn point_key_separates_identifying_stream_properties_and_flags() { + let resource = resource(Vec::new()); + let point = number_point(number_data_point::Value::AsInt(7)); + let base_metric = sum(point.clone(), AggregationTemporality::Cumulative, true); + let base = normalize(&base_metric, &resource); + + let mut changed_description = base_metric.clone(); + changed_description.description = "documentation only".into(); + assert_eq!( + base.point_key, + normalize(&changed_description, &resource).point_key + ); + + let mut changed_unit = base_metric.clone(); + changed_unit.unit = "ms".into(); + assert_ne!( + base.point_key, + normalize(&changed_unit, &resource).point_key + ); + + let changed_temporality = normalize( + &sum(point.clone(), AggregationTemporality::Delta, true), + &resource, + ); + assert_ne!(base.point_key, changed_temporality.point_key); + + let changed_monotonic = normalize( + &sum(point.clone(), AggregationTemporality::Cumulative, false), + &resource, + ); + assert_ne!(base.point_key, changed_monotonic.point_key); + + let mut flagged_point = point; + flagged_point.flags = 1; + let flagged = normalize( + &sum(flagged_point, AggregationTemporality::Cumulative, true), + &resource, + ); + assert_ne!(base.point_key, flagged.point_key); + let value: Value = serde_json::from_str(&flagged.value_json).unwrap(); + assert_eq!(value["flags"], 1); +} + +#[test] +fn non_finite_double_values_are_lossless_valid_json_tokens() { + let resource = resource(Vec::new()); + for (value, expected) in [ + (f64::NAN, "nan"), + (f64::INFINITY, "+infinity"), + (f64::NEG_INFINITY, "-infinity"), + ] { + let output = normalize( + &gauge(number_point(number_data_point::Value::AsDouble(value))), + &resource, + ); + let parsed: Value = serde_json::from_str(&output.value_json).unwrap(); + assert_eq!(parsed["value"], expected); + } +} + +#[test] +fn exemplar_ids_are_validated_serialized_and_affect_point_key() { + let resource = resource(Vec::new()); + let mut point = number_point(number_data_point::Value::AsInt(7)); + point.exemplars.push(Exemplar { + filtered_attributes: vec![kv("user.email", "alice@example.invalid")], + time_unix_nano: 150, + span_id: vec![0x22; 8], + trace_id: vec![0x11; 16], + value: Some(exemplar::Value::AsDouble(1.5)), + }); + let with_exemplar = normalize(&gauge(point.clone()), &resource); + let exemplars: Value = serde_json::from_str(&with_exemplar.exemplars_json).unwrap(); + assert_eq!(exemplars[0]["trace_id"], "11111111111111111111111111111111"); + assert!( + exemplars[0]["filtered_attributes"]["user.email"] + .as_str() + .unwrap() + .starts_with("sha256:") + ); + point.exemplars[0].span_id = vec![0x33; 8]; + let changed = normalize(&gauge(point), &resource); + assert_ne!(with_exemplar.point_key, changed.point_key); +} + +#[test] +fn invalid_or_missing_number_point_fields_fail_closed() { + let resource = resource(Vec::new()); + let scope = scope(); + let mut missing = number_point(number_data_point::Value::AsInt(1)); + missing.value = None; + assert_eq!( + normalize_number_metric( + Some(&resource), + "", + Some(&scope), + "", + &gauge(missing), + RECEIVED_AT + ) + .unwrap_err(), + MetricNormalizeError::MissingValue + ); + let mut zero_time = number_point(number_data_point::Value::AsInt(1)); + zero_time.time_unix_nano = 0; + assert_eq!( + normalize_number_metric( + Some(&resource), + "", + Some(&scope), + "", + &gauge(zero_time), + RECEIVED_AT + ) + .unwrap_err(), + MetricNormalizeError::MissingPointTime + ); + let unspecified = sum( + number_point(number_data_point::Value::AsInt(1)), + AggregationTemporality::Unspecified, + true, + ); + assert_eq!( + normalize_number_metric( + Some(&resource), + "", + Some(&scope), + "", + &unspecified, + RECEIVED_AT + ) + .unwrap_err(), + MetricNormalizeError::UnspecifiedTemporality + ); +} + +#[test] +fn metric_attributes_follow_configured_privacy_policy() { + let resource = resource(vec![kv("project.path", "/secret/project")]); + let mut point = number_point(number_data_point::Value::AsInt(1)); + point.attributes.push(kv("gen_ai.prompt", "secret prompt")); + let metric = gauge(point); + + let default_output = normalize(&metric, &resource); + let attrs: Value = serde_json::from_str(&default_output.attributes_json).unwrap(); + assert_eq!(attrs["gen_ai.prompt"], "[REDACTED]"); + assert_eq!( + default_output.ai_project.as_deref(), + Some("/secret/project") + ); + + let privacy = AgentObservatoryPrivacyConfig { + include_paths: false, + ..Default::default() + }; + let scope = scope(); + let private_output = normalize_number_metric_with_privacy( + Some(&resource), + "resource/v1", + Some(&scope), + "scope/v1", + &metric, + &privacy, + RECEIVED_AT, + ) + .unwrap() + .into_iter() + .next() + .unwrap(); + assert_eq!(private_output.ai_project, None); + assert!(!private_output.resource_json.contains("/secret/project")); +} From ef1df6a67bdee1983395469ef9e0299c2e6fa9a4 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Thu, 20 Aug 2026 12:00:19 -0400 Subject: [PATCH 4/4] test(otlp): scrub private hostname fixtures --- docs/plans/agent-observatory/proof/PROOF.md | 2 +- src/otlp/metrics_tests.rs | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/docs/plans/agent-observatory/proof/PROOF.md b/docs/plans/agent-observatory/proof/PROOF.md index f84c1194..36828a0e 100644 --- a/docs/plans/agent-observatory/proof/PROOF.md +++ b/docs/plans/agent-observatory/proof/PROOF.md @@ -536,4 +536,4 @@ FIX: adversarial review found the original Cortex point-key contract omitted Ope STRUCTURE: split canonical exemplar/value/key encoding into `metrics_payload.rs`; production metric modules remain comfortably below the 500-line gate PROOF: definitive locked focused metric normalization suite passed 8/8, including gauge/sum semantics, deterministic reorder-insensitive keys, stream-identity/flags collision resistance, exemplar validation, non-finite doubles, privacy policy, and fail-closed invalid fields REGRESSION: definitive locked full OTLP library sweep passed 90/90 with all prior logs/traces/auth/privacy/runtime/database coverage intact -GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and the full 500-line production Rust module-size gate passed; exact `cargo clippy --all-targets --all-features --locked -- -D warnings` passed after structurally reducing the point-normalizer argument surface and fixing the test config construction lint +GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and the full 500-line production Rust module-size gate passed; exact `cargo clippy --all-targets --all-features --locked -- -D warnings` passed after structurally reducing the point-normalizer argument surface and fixing the test config construction lint; the exact four-script Public Identity CI bundle passed after replacing private hostname fixtures with a neutral test host diff --git a/src/otlp/metrics_tests.rs b/src/otlp/metrics_tests.rs index c824ee92..9a8db193 100644 --- a/src/otlp/metrics_tests.rs +++ b/src/otlp/metrics_tests.rs @@ -86,7 +86,7 @@ fn normalize(metric: &Metric, resource: &Resource) -> MetricPointInput { fn integer_gauge_ignores_start_time_and_normalizes_identity() { let resource = resource(vec![ kv("service.name", "claude-code"), - kv("host.name", "dookie"), + kv("host.name", "fixture-host"), ]); let output = normalize( &gauge(number_point(number_data_point::Value::AsInt(7))), @@ -97,7 +97,7 @@ fn integer_gauge_ignores_start_time_and_normalizes_identity() { assert_eq!(output.aggregation_temporality, None); assert_eq!(output.monotonic, None); assert_eq!(output.time_unix_nano, 200); - assert_eq!(output.hostname, "dookie"); + assert_eq!(output.hostname, "fixture-host"); assert_eq!(output.ai_tool.as_deref(), Some("claude")); assert_eq!( serde_json::from_str::(&output.value_json).unwrap(), @@ -130,7 +130,10 @@ fn cumulative_and_delta_sums_preserve_temporality_monotonic_and_start() { #[test] fn repeated_fixture_and_reordered_attributes_have_same_point_key() { - let mut resource = resource(vec![kv("service.name", "codex"), kv("host.name", "dookie")]); + let mut resource = resource(vec![ + kv("service.name", "codex"), + kv("host.name", "fixture-host"), + ]); resource.entity_refs = vec![ EntityRef { schema_url: "entity/v1".into(),