diff --git a/asap-query-engine/src/drivers/ingest/otel.rs b/asap-query-engine/src/drivers/ingest/otel.rs index 3677e54f..072c9efe 100644 --- a/asap-query-engine/src/drivers/ingest/otel.rs +++ b/asap-query-engine/src/drivers/ingest/otel.rs @@ -146,6 +146,7 @@ impl MetricsService for MetricsServiceImpl { process_otlp_request(&req, "gRPC"); if let Some(state) = &self.shared.ingest_state { route_otlp_to_precompute(&req, state).await; + route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { @@ -192,6 +193,7 @@ async fn handle_otlp_http( process_otlp_request(&req, "HTTP"); if let Some(state) = &shared.ingest_state { route_otlp_to_precompute(&req, state).await; + route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via HTTP"); Ok(Json(serde_json::json!({"rejected": 0}))) @@ -478,6 +480,255 @@ async fn route_otlp_to_precompute( } } +/// Walk the modified-OTLP first-class sketch metric variants +/// (`DDSketch` / `KLLSketch` / `CountSketch` / `CountMinSketch` / +/// `HLLSketch` on `Metric.data` tags 13–17) and dispatch each +/// `*SketchDataPoint` through the precompute engine via +/// `WorkerMessage::AccumulatorInput`. +/// +/// Per-variant decoding of the typed `sketch` bytes is delegated to +/// `decode_modified_otlp_sketch_bytes`, which in turn calls into the +/// matching concrete accumulator's `from_sketchlib_proto_bytes` +/// constructor when one exists. Variants without a concrete decoder +/// today (KLL / DDSketch / CountSketch / HLL) fall through to the +/// §5.2 fallback path so the user still gets a correct answer; PR C +/// (task #8) will close those decoder gaps. +async fn route_modified_otlp_sketches_to_precompute( + request: &ExportMetricsServiceRequest, + ingest_state: &Arc, +) { + use asap_otel_proto::tonic::metrics::v1::metric::Data; + + let ingest_received_at = Instant::now(); + let mut messages: Vec = Vec::new(); + let mut routed = 0usize; + let mut decoded_failed = 0usize; + let mut unconfigured = 0usize; + + for resource_metrics in &request.resource_metrics { + let resource_attrs = resource_metrics + .resource + .as_ref() + .map(|r| attributes_to_map(&r.attributes)) + .unwrap_or_default(); + + for scope_metrics in &resource_metrics.scope_metrics { + let scope_attrs = scope_metrics + .scope + .as_ref() + .map(|s| attributes_to_map(&s.attributes)) + .unwrap_or_default(); + + for metric in &scope_metrics.metrics { + if metric.name.is_empty() { + continue; + } + + let base_labels: HashMap = scope_attrs + .iter() + .chain(resource_attrs.iter()) + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + + // Each branch yields an iterator-like slice of (kind, attrs, + // time_unix_nano, sketch_bytes, encoding_i32) tuples. We + // then route each tuple through the same dispatcher. + let dps: Vec = match &metric.data { + Some(Data::Ddsketch(d)) => d + .data_points + .iter() + .map(|dp| ModifiedOtlpSketchDp { + kind: SketchKind::DdSketch, + attrs: merge_point_attributes(&base_labels, &dp.attributes), + time_unix_nano: dp.time_unix_nano, + sketch: dp.sketch.clone(), + encoding: dp.encoding, + }) + .collect(), + Some(Data::Kllsketch(k)) => k + .data_points + .iter() + .map(|dp| ModifiedOtlpSketchDp { + kind: SketchKind::Kll, + attrs: merge_point_attributes(&base_labels, &dp.attributes), + time_unix_nano: dp.time_unix_nano, + sketch: dp.sketch.clone(), + encoding: dp.encoding, + }) + .collect(), + Some(Data::Countsketch(c)) => c + .data_points + .iter() + .map(|dp| ModifiedOtlpSketchDp { + kind: SketchKind::CountSketch, + attrs: merge_point_attributes(&base_labels, &dp.attributes), + time_unix_nano: dp.time_unix_nano, + sketch: dp.sketch.clone(), + encoding: dp.encoding, + }) + .collect(), + Some(Data::Countminsketch(c)) => c + .data_points + .iter() + .map(|dp| ModifiedOtlpSketchDp { + kind: SketchKind::CountMin, + attrs: merge_point_attributes(&base_labels, &dp.attributes), + time_unix_nano: dp.time_unix_nano, + sketch: dp.sketch.clone(), + encoding: dp.encoding, + }) + .collect(), + Some(Data::Hllsketch(h)) => h + .data_points + .iter() + .map(|dp| ModifiedOtlpSketchDp { + kind: SketchKind::Hll, + attrs: merge_point_attributes(&base_labels, &dp.attributes), + time_unix_nano: dp.time_unix_nano, + sketch: dp.sketch.clone(), + encoding: dp.encoding, + }) + .collect(), + _ => continue, + }; + + for dp in dps { + let series_key = format_series_key(&metric.name, &dp.attrs); + let ts_ms = (dp.time_unix_nano / 1_000_000) as i64; + + let accumulator: Box = + match decode_modified_otlp_sketch_bytes(dp.kind, dp.encoding, &dp.sketch) { + Ok(acc) => acc, + Err(e) => { + decoded_failed += 1; + debug!( + "OTLP modified-proto sketch decode failed (metric={}, kind={:?}, encoding={}, bytes={}): {} — falling through to §5.2 fallback", + metric.name, + dp.kind, + dp.encoding, + dp.sketch.len(), + e + ); + continue; + } + }; + + let mut matched_any = false; + for config in &ingest_state.agg_configs { + if config.metric != metric.name + && config.spatial_filter_normalized != metric.name + && config.spatial_filter != metric.name + { + continue; + } + let group_key = IngestState::extract_group_key_for(&series_key, config); + messages.push(WorkerMessage::AccumulatorInput { + agg_id: config.aggregation_id, + group_key, + timestamp_ms: ts_ms, + accumulator: accumulator.clone_boxed_core(), + ingest_received_at, + }); + matched_any = true; + } + if matched_any { + routed += 1; + } else { + unconfigured += 1; + } + } + } + } + } + + if !messages.is_empty() { + if let Err(e) = ingest_state + .router + .route_group_batch(messages, ingest_received_at) + .await + { + warn!("OTLP modified-proto sketch routing error: {}", e); + } + } + + if routed + decoded_failed + unconfigured > 0 { + debug!( + "OTLP modified-proto sketch ingest: {} routed, {} decode-failed (fallback), {} unconfigured", + routed, decoded_failed, unconfigured + ); + } +} + +/// Sketch family carried by a modified-OTLP `*SketchDataPoint`. Used by +/// the encoding dispatcher in `decode_modified_otlp_sketch_bytes`. +#[derive(Debug, Clone, Copy)] +enum SketchKind { + DdSketch, + Kll, + CountSketch, + CountMin, + Hll, +} + +/// A single modified-OTLP sketch data point flattened across the five +/// per-variant data-point types so the routing loop can treat them +/// uniformly. +struct ModifiedOtlpSketchDp { + kind: SketchKind, + attrs: HashMap, + time_unix_nano: u64, + sketch: Vec, + encoding: i32, +} + +/// Decode the typed `sketch` bytes from a modified-OTLP +/// `*SketchDataPoint` into a concrete `AggregateCore`. +/// +/// Dispatches on the `(SketchKind, encoding)` pair. For each +/// `(kind, _ENCODING_PROTO)` pair we call the matching accumulator's +/// `from_sketchlib_proto_bytes` constructor. Variants without a +/// constructor today return `Err`; the caller falls through to §5.2 +/// fallback so the user still gets a correct answer. Per-variant +/// decoders are tracked in PR C (task #8) and PR I (task #14, for +/// `_ENCODING_MSGPACK` parity). +fn decode_modified_otlp_sketch_bytes( + kind: SketchKind, + encoding: i32, + bytes: &[u8], +) -> Result, Box> { + use crate::precompute_operators::CountMinSketchAccumulator; + + // The encoding value is the raw i32 from the proto enum. We only + // accept ENCODING_PROTO (= 1) for now; ENCODING_PROTO_DELTA (= 2) + // and ENCODING_MSGPACK / ENCODING_MSGPACK_DELTA (PR I) are routed + // to a "not yet implemented" Err so the caller falls through to + // the §5.2 fallback. + const ENCODING_PROTO: i32 = 1; + + if encoding != ENCODING_PROTO { + return Err(format!( + "modified-OTLP sketch encoding {encoding} not yet supported \ + (only ENCODING_PROTO = 1 is wired today; deltas tracked in PR C, \ + msgpack tracked in PR I)" + ) + .into()); + } + + match kind { + SketchKind::CountMin => { + let acc = CountMinSketchAccumulator::from_sketchlib_proto_bytes(bytes)?; + Ok(Box::new(acc)) + } + SketchKind::Kll | SketchKind::DdSketch | SketchKind::CountSketch | SketchKind::Hll => { + Err(format!( + "modified-OTLP sketch decoder for {kind:?} not yet implemented \ + (tracked in PR C, task #8)" + ) + .into()) + } + } +} + /// Identify the concrete sketch type inside a `SketchEnvelope` payload, /// returning a human-readable name for logging. Returns `"Unknown"` if the /// payload does not decode or the `sketch_state` variant is unset. diff --git a/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs b/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs index fe3ec330..a938aafd 100644 --- a/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs @@ -68,6 +68,77 @@ impl CountMinSketchAccumulator { }) } + /// Decode from the modified OTLP wire format's + /// `CountMinSketchDataPoint.sketch` bytes — i.e. the protobuf-encoded + /// `asap_sketchlib::proto::sketchlib::CountMinState` message used by + /// DataCollector's `countminsketchprocessor` when emitting via + /// `Metric.data = CountMinSketch{…}` with + /// `encoding = COUNT_MIN_SKETCH_ENCODING_PROTO`. + /// + /// The resulting accumulator is constructed via + /// `CountMinSketch::from_legacy_matrix` after reshaping the flat + /// `counts_int` / `counts_float` field into a `Vec>`. + pub fn from_sketchlib_proto_bytes(buffer: &[u8]) -> Result> { + use asap_sketchlib::proto::sketchlib::{CountMinState, CounterType}; + use prost::Message; + + let state = + CountMinState::decode(buffer).map_err(|e| format!("decode CountMinState: {e}"))?; + let rows = state.rows as usize; + let cols = state.cols as usize; + if rows == 0 || cols == 0 { + return Err(format!("CountMinState has zero dims (rows={rows}, cols={cols})").into()); + } + let expected_len = rows * cols; + let counter_type = CounterType::try_from(state.counter_type).map_err(|_| { + format!( + "CountMinState has unknown counter_type tag {}", + state.counter_type + ) + })?; + let flat: Vec = match counter_type { + CounterType::Int32 | CounterType::Int64 => { + if state.counts_int.len() != expected_len { + return Err(format!( + "CountMinState counts_int has {} entries, expected rows*cols = {}", + state.counts_int.len(), + expected_len + ) + .into()); + } + state.counts_int.iter().map(|&v| v as f64).collect() + } + CounterType::Float64 => { + if state.counts_float.len() != expected_len { + return Err(format!( + "CountMinState counts_float has {} entries, expected rows*cols = {}", + state.counts_float.len(), + expected_len + ) + .into()); + } + state.counts_float.clone() + } + // INT128 stores (hi, lo) pairs and would have 2 * rows * cols + // entries in counts_int; defer to PR C if a producer ever uses it. + other => { + return Err(format!( + "CountMinState counter_type {other:?} not yet supported \ + (PR C will extend coverage)" + ) + .into()); + } + }; + let mut matrix = Vec::with_capacity(rows); + for r in 0..rows { + let start = r * cols; + matrix.push(flat[start..start + cols].to_vec()); + } + Ok(Self { + inner: CountMinSketch::from_legacy_matrix(matrix, rows, cols), + }) + } + pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { if buffer.len() < 8 { return Err("Buffer too short for row_num and col_num".into()); @@ -466,4 +537,100 @@ mod tests { let mixed_accs: Vec> = vec![Box::new(cms), Box::new(sum)]; assert!(CountMinSketchAccumulator::merge_multiple(&mixed_accs).is_err()); } + + #[test] + fn test_from_sketchlib_proto_bytes_int64() { + // Hand-build a CountMinState proto with INT64 counters and verify + // round-tripping through from_sketchlib_proto_bytes yields the same + // matrix that the modified-OTLP wire format would carry. + use asap_sketchlib::proto::sketchlib::{CountMinState, CounterType}; + use prost::Message; + + let rows = 2u32; + let cols = 3u32; + // Row-major: row 0 = [1,2,3], row 1 = [4,5,6] + let counts_int: Vec = vec![1, 2, 3, 4, 5, 6]; + let state = CountMinState { + rows, + cols, + counter_type: CounterType::Int64 as i32, + counts_int: counts_int.clone(), + counts_float: Vec::new(), + sum_counts: Vec::new(), + sum2_counts: Vec::new(), + l1: Vec::new(), + l2: Vec::new(), + }; + let bytes = state.encode_to_vec(); + + let acc = CountMinSketchAccumulator::from_sketchlib_proto_bytes(&bytes).expect("decode ok"); + let matrix = acc.inner.sketch(); + assert_eq!(matrix.len(), rows as usize); + assert_eq!(matrix[0], vec![1.0, 2.0, 3.0]); + assert_eq!(matrix[1], vec![4.0, 5.0, 6.0]); + } + + #[test] + fn test_from_sketchlib_proto_bytes_float64() { + use asap_sketchlib::proto::sketchlib::{CountMinState, CounterType}; + use prost::Message; + + let state = CountMinState { + rows: 2, + cols: 2, + counter_type: CounterType::Float64 as i32, + counts_int: Vec::new(), + counts_float: vec![1.5, 2.5, 3.5, 4.5], + sum_counts: Vec::new(), + sum2_counts: Vec::new(), + l1: Vec::new(), + l2: Vec::new(), + }; + let bytes = state.encode_to_vec(); + + let acc = CountMinSketchAccumulator::from_sketchlib_proto_bytes(&bytes).expect("decode ok"); + let matrix = acc.inner.sketch(); + assert_eq!(matrix[0], vec![1.5, 2.5]); + assert_eq!(matrix[1], vec![3.5, 4.5]); + } + + #[test] + fn test_from_sketchlib_proto_bytes_dimension_mismatch() { + // counts_int has 5 entries but rows*cols = 6 → expect error + use asap_sketchlib::proto::sketchlib::{CountMinState, CounterType}; + use prost::Message; + + let state = CountMinState { + rows: 2, + cols: 3, + counter_type: CounterType::Int64 as i32, + counts_int: vec![1, 2, 3, 4, 5], + counts_float: Vec::new(), + sum_counts: Vec::new(), + sum2_counts: Vec::new(), + l1: Vec::new(), + l2: Vec::new(), + }; + let bytes = state.encode_to_vec(); + + let result = CountMinSketchAccumulator::from_sketchlib_proto_bytes(&bytes); + assert!(result.is_err()); + assert!( + result.unwrap_err().to_string().contains("counts_int"), + "error should mention counts_int dim mismatch" + ); + } + + #[test] + fn test_from_sketchlib_proto_bytes_zero_dims_rejected() { + use asap_sketchlib::proto::sketchlib::CountMinState; + use prost::Message; + + let state = CountMinState::default(); + let bytes = state.encode_to_vec(); + + let result = CountMinSketchAccumulator::from_sketchlib_proto_bytes(&bytes); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("zero dims")); + } }