diff --git a/crates/ourios-bench/src/comparative.rs b/crates/ourios-bench/src/comparative.rs index d1738894..9a666aa4 100644 --- a/crates/ourios-bench/src/comparative.rs +++ b/crates/ourios-bench/src/comparative.rs @@ -295,6 +295,92 @@ fn body_bytes(body: &ourios_querier::LogBody) -> Result, BenchError> { }) } +/// Parse a Loki `query_range` **streams** response into [`LineKey`]s — the +/// Loki half of the RFC0031.1 equivalence check. +/// +/// Loki returns matching lines under `data.result[].values[]`, each a +/// `["", ""]` pair. Each becomes a +/// `LineKey` keyed the same way as the Ourios side (`(timestamp, body)`), +/// so the two feed [`compare_lines`]. The timestamp is Loki's nanosecond +/// string; the body is the log line bytes. +/// +/// # Errors +/// +/// [`BenchError::Pipeline`] if the response isn't JSON; is a Loki **error** +/// response (`status == "error"` — surfaces Loki's `errorType` / `error`); +/// is missing the `data.result` array; has a stream missing its `values` +/// array; has a `values` entry that isn't a two-element `[string, string]` +/// pair; has a timestamp or log line that isn't a string; or has a +/// timestamp string that isn't a `u64`. Malformed-entry errors carry the +/// stream + value indices for debugging against real Loki responses. +pub fn parse_loki_streams(response_json: &str) -> Result, BenchError> { + let root: serde_json::Value = + serde_json::from_str(response_json).map_err(|e| BenchError::Pipeline { + detail: format!("Loki response is not JSON: {e}"), + })?; + + // A Loki error response ({"status":"error", "errorType":..., "error":...}) + // would otherwise fail below as "missing data.result" — surface Loki's + // own diagnostic instead, which is what an operator needs to see. + if root.get("status").and_then(serde_json::Value::as_str) == Some("error") { + let error_type = root + .get("errorType") + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown"); + let error = root + .get("error") + .and_then(serde_json::Value::as_str) + .unwrap_or("(no message)"); + return Err(BenchError::Pipeline { + detail: format!("Loki query error [{error_type}]: {error}"), + }); + } + + let result = root + .get("data") + .and_then(|d| d.get("result")) + .and_then(serde_json::Value::as_array) + .ok_or_else(|| BenchError::Pipeline { + detail: "Loki response missing `data.result` array".to_string(), + })?; + + let mut lines = Vec::new(); + for (si, stream) in result.iter().enumerate() { + let values = stream + .get("values") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| BenchError::Pipeline { + detail: format!("Loki stream {si} missing `values` array"), + })?; + for (vi, pair) in values.iter().enumerate() { + let entry = + pair.as_array() + .filter(|a| a.len() == 2) + .ok_or_else(|| BenchError::Pipeline { + detail: format!( + "Loki stream {si} value {vi} is not a [timestamp, line] pair" + ), + })?; + let ts_str = entry[0].as_str().ok_or_else(|| BenchError::Pipeline { + detail: format!("Loki stream {si} value {vi} timestamp is not a string"), + })?; + let timestamp_unix_nanos = ts_str.parse::().map_err(|e| BenchError::Pipeline { + detail: format!( + "Loki stream {si} value {vi} timestamp `{ts_str}` is not a u64: {e}" + ), + })?; + let body = entry[1].as_str().ok_or_else(|| BenchError::Pipeline { + detail: format!("Loki stream {si} value {vi} log line is not a string"), + })?; + lines.push(LineKey { + timestamp_unix_nanos, + body: body.as_bytes().to_vec(), + }); + } + } + Ok(lines) +} + /// A truncated, lossy preview of a body for a mismatch report — bounded /// so an arbitrarily large body can't blow up the stderr summary. fn body_preview(body: &[u8]) -> String { @@ -389,6 +475,76 @@ mod tests { assert!(m.examples[0].contains("loki=3")); } + #[test] + fn parse_loki_streams_keys_compatibly_with_the_ourios_side() { + // A synthetic Loki `query_range` streams response — three lines + // across two streams. + let response = r#"{ + "status": "success", + "data": { + "resultType": "streams", + "result": [ + { "stream": {"service_name": "a"}, "values": [ + ["1775127480000000000", "user 1 logged in"], + ["1775127480000000001", "user 2 logged in"] + ]}, + { "stream": {"service_name": "b"}, "values": [ + ["1775127480000000002", "user 3 logged in"] + ]} + ] + } + }"#; + let loki = parse_loki_streams(response).expect("parse loki streams"); + assert_eq!(loki.len(), 3); + + // The parsed keys must be byte-for-byte what the Ourios side would + // produce for the same lines — otherwise the equivalence check is + // comparing incompatibly-keyed sets. + let ourios = vec![ + LineKey { + timestamp_unix_nanos: 1_775_127_480_000_000_000, + body: b"user 1 logged in".to_vec(), + }, + LineKey { + timestamp_unix_nanos: 1_775_127_480_000_000_001, + body: b"user 2 logged in".to_vec(), + }, + LineKey { + timestamp_unix_nanos: 1_775_127_480_000_000_002, + body: b"user 3 logged in".to_vec(), + }, + ]; + assert!( + compare_lines(&ourios, &loki, 8).is_equal(), + "Loki-parsed lines must key-match the Ourios side", + ); + } + + #[test] + fn parse_loki_streams_rejects_malformed_responses() { + // Missing data.result, a non-pair value, and a non-numeric + // timestamp all error rather than silently drop rows. + assert!(parse_loki_streams(r#"{"status":"success"}"#).is_err()); + assert!( + parse_loki_streams(r#"{"data":{"result":[{"values":[["1","a","extra"]]}]}}"#).is_err() + ); + assert!( + parse_loki_streams(r#"{"data":{"result":[{"values":[["notanum","a"]]}]}}"#).is_err() + ); + + // A Loki error response surfaces Loki's own diagnostic, not the + // misleading "missing data.result". + let err = parse_loki_streams( + r#"{"status":"error","errorType":"parse error","error":"unexpected token"}"#, + ) + .expect_err("Loki error response must error"); + let BenchError::Pipeline { detail } = err else { + panic!("expected a pipeline error"); + }; + assert!(detail.contains("parse error"), "{detail}"); + assert!(detail.contains("unexpected token"), "{detail}"); + } + #[test] fn body_bytes_defers_non_string_kinds() { // An absent body (RFC 0025) must NOT be silently lowered to an diff --git a/crates/ourios-bench/src/lib.rs b/crates/ourios-bench/src/lib.rs index 5df7b691..38aac7be 100644 --- a/crates/ourios-bench/src/lib.rs +++ b/crates/ourios-bench/src/lib.rs @@ -44,7 +44,7 @@ mod store; pub use calibrate::{CALIBRATION_DIR, extract_manifest, write_manifest}; pub use comparative::{ AggKey, EquivalenceOutcome, LineKey, Mismatch, compare_aggregations, compare_lines, - ourios_query_lines, + ourios_query_lines, parse_loki_streams, }; pub use corpus::TxtSeverity; pub use reference::ReferenceCorpus;