diff --git a/asap-query-engine/src/drivers/ingest/otel.rs b/asap-query-engine/src/drivers/ingest/otel.rs index ba99e8099..e1e8f706e 100644 --- a/asap-query-engine/src/drivers/ingest/otel.rs +++ b/asap-query-engine/src/drivers/ingest/otel.rs @@ -185,9 +185,11 @@ impl MetricsService for MetricsServiceImpl { if let Some(cache) = &self.shared.probe_cache { capture_freshness_probe_samples(&req, cache); } + let mut unknown_series_ids: Vec = Vec::new(); if let Some(state) = &self.shared.ingest_state { route_otlp_to_precompute(&req, state).await; - route_modified_otlp_sketches_to_precompute(&req, state).await; + unknown_series_ids = + route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { @@ -196,13 +198,48 @@ impl MetricsService for MetricsServiceImpl { // this field; not yet wired (PR B will populate it when the // backend learns to mint series_ids). series_assignments: Vec::new(), - // Refactor-2026-05: backend signals senders to evict cached sids - // here. Empty for now — populated by Phase 4 (centralized - // ResolveSeriesIDs resolver) when sid-cache divergence is - // detected (e.g., backend restart without persistence). - unknown_series_ids: Vec::new(), + // Phase 4 — backend signals senders to evict cached sids here + // when this Export carried a sid the resolver does not + // recognize (sid-cache divergence — e.g. after a backend + // restart without persistence, or when the sender's sid + // disagrees with the resolved sid for the same attrs). + unknown_series_ids, })) } + + async fn resolve_series_i_ds( + &self, + request: Request< + asap_otel_proto::tonic::collector::metrics::v1::ResolveSeriesIDsRequest, + >, + ) -> Result< + Response, + Status, + > { + use asap_otel_proto::tonic::collector::metrics::v1::{ + ResolveSeriesIDsResponse, SeriesAssignment, + }; + let req = request.into_inner(); + let mut assignments = Vec::with_capacity(req.queries.len()); + if let Some(state) = &self.shared.ingest_state { + for q in req.queries { + // The fingerprint travels as opaque bytes on the wire, but + // the resolver hashes it as a string (the sender's + // canonical fingerprint algorithm matches our + // `canonical_attrs_fingerprint`). UTF-8 is lossy here only + // for malformed inputs — those produce a degraded but + // deterministic key, never a panic. + let fp = String::from_utf8_lossy(&q.attributes_fingerprint).into_owned(); + let sid = state.series_resolver.resolve(&q.metric_name, &fp); + assignments.push(SeriesAssignment { + attributes_fingerprint: q.attributes_fingerprint, + series_id: sid, + ..Default::default() + }); + } + } + Ok(Response::new(ResolveSeriesIDsResponse { assignments })) + } } async fn handle_otlp_http( @@ -240,12 +277,16 @@ async fn handle_otlp_http( if let Some(cache) = &shared.probe_cache { capture_freshness_probe_samples(&req, cache); } + let mut unknown_series_ids: Vec = Vec::new(); if let Some(state) = &shared.ingest_state { route_otlp_to_precompute(&req, state).await; - route_modified_otlp_sketches_to_precompute(&req, state).await; + unknown_series_ids = route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via HTTP"); - Ok(Json(serde_json::json!({"rejected": 0}))) + Ok(Json(serde_json::json!({ + "rejected": 0, + "unknown_series_ids": unknown_series_ids, + }))) } /// A parsed metric data point: name, labels, timestamp (nanos), and numeric value. @@ -629,7 +670,7 @@ async fn route_otlp_to_precompute( async fn route_modified_otlp_sketches_to_precompute( request: &ExportMetricsServiceRequest, ingest_state: &Arc, -) { +) -> Vec { use asap_otel_proto::tonic::metrics::v1::metric::Data; let ingest_received_at = Instant::now(); @@ -643,6 +684,12 @@ async fn route_modified_otlp_sketches_to_precompute( let mut decoded_failed = 0usize; let mut unconfigured = 0usize; let mut barrier_drops: HashMap = HashMap::new(); + // Phase 4 — sids the receiver did not recognize this Export. Returned + // to the caller so the gRPC / HTTP handler can stamp them into + // `ExportMetricsServiceResponse.unknown_series_ids`. Senders evict + // these sids and re-emit with attributes; backend re-resolves and + // returns fresh `series_assignments`. + let mut unknown_sids: Vec = Vec::new(); for resource_metrics in &request.resource_metrics { let resource_attrs = resource_metrics @@ -673,61 +720,96 @@ async fn route_modified_otlp_sketches_to_precompute( // 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(), + Some(Data::Ddsketch(d)) => { + let cfg = crate::stores::sketch_index::SketchConfig::DDSketch { + relative_accuracy: d.relative_accuracy, + }; + 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, + series_id: dp.series_id, + start_time_unix_nano: dp.start_time_unix_nano, + container_config: cfg.clone(), + }) + .collect() + } + Some(Data::Kllsketch(k)) => { + let cfg = crate::stores::sketch_index::SketchConfig::Kll { k: k.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, + series_id: dp.series_id, + start_time_unix_nano: dp.start_time_unix_nano, + container_config: cfg.clone(), + }) + .collect() + } + Some(Data::Countsketch(c)) => { + let cfg = crate::stores::sketch_index::SketchConfig::CountSketch { + rows: c.rows, + cols: c.cols, + }; + 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, + series_id: dp.series_id, + start_time_unix_nano: dp.start_time_unix_nano, + container_config: cfg.clone(), + }) + .collect() + } + Some(Data::Countminsketch(c)) => { + let cfg = crate::stores::sketch_index::SketchConfig::CountMin { + rows: c.rows, + cols: c.cols, + }; + 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, + series_id: dp.series_id, + start_time_unix_nano: dp.start_time_unix_nano, + container_config: cfg.clone(), + }) + .collect() + } + Some(Data::Hllsketch(h)) => { + let cfg = crate::stores::sketch_index::SketchConfig::Hll { + precision: h.precision, + }; + 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, + series_id: dp.series_id, + start_time_unix_nano: dp.start_time_unix_nano, + container_config: cfg.clone(), + }) + .collect() + } _ => continue, }; @@ -735,6 +817,123 @@ async fn route_modified_otlp_sketches_to_precompute( let series_key = format_series_key(&metric.name, &dp.attrs); let ts_ms = (dp.time_unix_nano / 1_000_000) as i64; + // Phase 4 — sid resolution gate. The four cases mirror + // the design doc §5.4 invariant: + // (sid=0, attrs) → mint a fresh sid and use it + // (sid!=0, attrs) → trust attrs; if cached value + // disagrees, the sender's sid + // is stale → push to + // `unknown_sids` so the + // response evicts it + // (sid!=0, no attrs) → reverse-lookup; if unknown, + // push to `unknown_sids` and + // drop this DP (sender will + // re-emit with attrs next pass) + // (sid=0, no attrs) → invalid wire shape, drop + let attrs_pairs: Vec<(&str, &str)> = + dp.attrs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&attrs_pairs); + let resolved_sid: Option = if attrs_pairs.is_empty() { + // No attrs on the wire — sid alone must be + // recognized, otherwise signal stale. + match dp.series_id { + 0 => None, + sid => { + if ingest_state.series_resolver.is_known(sid) { + Some(sid) + } else { + unknown_sids.push(sid); + None + } + } + } + } else if dp.series_id == 0 { + // Attrs present, no sid yet → mint or fetch. + Some( + ingest_state + .series_resolver + .resolve(&metric.name, &fp), + ) + } else { + // Both populated: attrs are the source of truth. + // Backend-resolved value wins; mismatched sender + // sid is signalled stale. + let cached = ingest_state + .series_resolver + .resolve(&metric.name, &fp); + if cached != dp.series_id { + unknown_sids.push(dp.series_id); + } + Some(cached) + }; + let Some(sid) = resolved_sid else { + continue; + }; + + // Phase 5 — register a `SketchInstanceMetadata` on + // first sight of `sid` and append this DP's sketch + // state to the per-sid columnar storage. The instance + // is keyed by sid, so subsequent DPs on the same sid + // skip the register step. `group_by_keys` is + // `dp.attrs.keys()` — after the agent's `AggregateBy` + // rollup, `attributes` is the group-by VALUES vector, + // and its key set IS the group-by KEY set. + { + use crate::stores::sketch_index::{ + AccuracyBound, Capability, SketchEncoding, SketchInstanceMetadata, + SketchKindHandle, SketchSampleState, + }; + use std::collections::{BTreeMap, BTreeSet}; + + if ingest_state.sketch_index.instance(sid).is_none() { + let kind = sketch_kind_handle_for(&dp); + let cap = match kind { + SketchKindHandle::DDSketch | SketchKindHandle::Kll => { + Capability::QuantileApprox(kind) + } + SketchKindHandle::Hll => Capability::CardinalityApprox, + SketchKindHandle::CountSketch + | SketchKindHandle::CountMin => { + Capability::FrequencyTopk(kind) + } + }; + let group_by_keys: BTreeSet = + dp.attrs.keys().cloned().collect(); + let cfg = dp.container_config.clone(); + ingest_state.sketch_index.register(SketchInstanceMetadata { + sid, + metric_name: metric.name.clone(), + group_by_keys, + capability: cap, + sketch_kind: kind, + sketch_config: cfg.clone(), + accuracy: AccuracyBound::from_config(&cfg), + first_seen_unix_ms: ts_ms, + }); + } + + let label_values: BTreeMap = dp + .attrs + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + let window: crate::stores::epoch_columnar::TimestampRange = ( + dp.start_time_unix_nano / 1_000_000, + dp.time_unix_nano / 1_000_000, + ); + let encoding = encoding_to_handle(dp.encoding) + .unwrap_or(SketchEncoding::ProtoFull); + ingest_state.sketch_index.append_sample( + sid, + label_values, + window, + SketchSampleState { + bytes: dp.sketch.clone(), + encoding, + }, + ); + } + // Encoding dispatch: full frames (PROTO / // MSGPACK) decode standalone and refresh the // per-series snapshot cache; delta frames @@ -827,6 +1026,13 @@ async fn route_modified_otlp_sketches_to_precompute( continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); + // DEPRECATED: aggregation_id-keyed write — remove + // after warm-tier validation. The Phase 5 + // SketchIndex above is the new write path; this + // legacy router push stays in tandem until the + // query path's warm-tier reducer is wired + // end-to-end and the streaming-config / + // SimpleMapStore call sites can be deleted. messages.push(WorkerMessage::AccumulatorInput { agg_id: config.aggregation_id, group_key, @@ -864,6 +1070,39 @@ async fn route_modified_otlp_sketches_to_precompute( routed, decoded_failed, unconfigured ); } + + unknown_sids +} + +/// Phase 5 helper — map a `ModifiedOtlpSketchDp` to the matching +/// `SketchKindHandle` so registration and capability classification +/// share one source of truth. +fn sketch_kind_handle_for( + dp: &ModifiedOtlpSketchDp, +) -> crate::stores::sketch_index::SketchKindHandle { + use crate::stores::sketch_index::SketchKindHandle; + match dp.kind { + SketchKind::DdSketch => SketchKindHandle::DDSketch, + SketchKind::Kll => SketchKindHandle::Kll, + SketchKind::Hll => SketchKindHandle::Hll, + SketchKind::CountSketch => SketchKindHandle::CountSketch, + SketchKind::CountMin => SketchKindHandle::CountMin, + } +} + +/// Phase 5 helper — translate the wire-format `encoding` integer to the +/// SketchIndex's `SketchEncoding` enum. Returns `None` for the unset +/// (0) encoding so callers can default to `ProtoFull` (the dominant +/// case for full-state frames). +fn encoding_to_handle(encoding: i32) -> Option { + use crate::stores::sketch_index::SketchEncoding; + match encoding { + ENCODING_PROTO => Some(SketchEncoding::ProtoFull), + ENCODING_PROTO_DELTA => Some(SketchEncoding::ProtoDelta), + ENCODING_MSGPACK => Some(SketchEncoding::MsgpackFull), + ENCODING_MSGPACK_DELTA => Some(SketchEncoding::MsgpackDelta), + _ => None, + } } /// Sketch family carried by a modified-OTLP `*SketchDataPoint`. Used by @@ -886,6 +1125,19 @@ struct ModifiedOtlpSketchDp { time_unix_nano: u64, sketch: Vec, encoding: i32, + /// Phase 4 — sender-supplied series_id, 0 when unset / first emit. + /// Backend's resolver mints a fresh sid when this is 0 with attrs + /// populated; pushes the sid into `unknown_series_ids` when this is + /// non-zero with empty attrs and the resolver doesn't recognize it. + series_id: u64, + /// Phase 5 — DataPoint-level start of the sketch window. Combined + /// with `time_unix_nano` to form the `(start_ms, end_ms)` window + /// the SketchIndex's columnar storage keys on. + start_time_unix_nano: u64, + /// Phase 5 — sketch-instance configuration lifted off the parent + /// container. Drives `SketchInstanceMetadata.sketch_config` and the + /// derived `AccuracyBound`. + container_config: crate::stores::sketch_index::SketchConfig, } /// Decode the typed `sketch` bytes from a modified-OTLP @@ -1575,3 +1827,192 @@ mod dispatcher_tests { assert!(err.contains("apply_modified_otlp_delta_bytes")); } } + +/// Phase 4 — sid-resolution gate tests. Construct an OTLP DDSketch +/// Export with one DataPoint per scenario, run it through +/// `route_modified_otlp_sketches_to_precompute`, and assert on the +/// returned `unknown_series_ids` plus the SeriesIdResolver / SketchIndex +/// state on the shared IngestState. +#[cfg(test)] +mod sid_resolution_tests { + use super::*; + use crate::data_model::{HotReloadStreamingConfig, StreamingConfig}; + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use crate::precompute_engine::series_router::SeriesRouter; + use crate::stores::sketch_db::SchemaRegistry; + use crate::stores::sketch_index::SketchIndex; + use asap_otel_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest; + use asap_otel_proto::tonic::common::v1::{any_value::Value as AnyVal, AnyValue, KeyValue}; + use asap_otel_proto::tonic::metrics::v1::{ + metric::Data, DdSketch as PbDDSketch, DdSketchDataPoint, Metric as PbMetric, + ResourceMetrics, ScopeMetrics, + }; + use std::sync::Arc; + use tokio::sync::mpsc; + + async fn make_state() -> (Arc, tokio::task::JoinHandle<()>) { + let (tx, mut rx) = mpsc::channel(1024); + let router = SeriesRouter::new(vec![tx]); + let streaming = StreamingConfig::new(std::collections::HashMap::new()); + let hot_reload = HotReloadStreamingConfig::new(streaming.clone()); + let schemas = Arc::new(SchemaRegistry::from_streaming_config(&streaming)); + let state = Arc::new(IngestState { + router, + samples_ingested: std::sync::atomic::AtomicU64::new(0), + samples_blocked_by_schema_barrier: std::sync::atomic::AtomicU64::new(0), + hot_reload_config: hot_reload, + schemas, + pass_raw_samples: false, + sketch_snapshots: dashmap::DashMap::new(), + series_resolver: Arc::new(SeriesIdResolver::new()), + sketch_index: Arc::new(SketchIndex::new()), + }); + let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); + (state, drain) + } + + fn kv(k: &str, v: &str) -> KeyValue { + KeyValue { + key: k.to_string(), + value: Some(AnyValue { + value: Some(AnyVal::StringValue(v.to_string())), + }), + } + } + + fn build_request(metric_name: &str, dp: DdSketchDataPoint) -> ExportMetricsServiceRequest { + ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + resource: None, + scope_metrics: vec![ScopeMetrics { + scope: None, + metrics: vec![PbMetric { + name: metric_name.to_string(), + description: String::new(), + unit: String::new(), + metadata: Vec::new(), + data: Some(Data::Ddsketch(PbDDSketch { + data_points: vec![dp], + aggregation_temporality: 0, + relative_accuracy: 0.01, + })), + }], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } + } + + #[tokio::test] + async fn fresh_sid_minted_when_sender_supplies_zero_with_attrs() { + let (state, drain) = make_state().await; + let dp = DdSketchDataPoint { + attributes: vec![kv("zone", "z0")], + start_time_unix_nano: 1_000_000, + time_unix_nano: 11_000_000, + sketch: vec![1, 2, 3], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: 0, + }; + let req = build_request("http_latency_ms", dp); + + let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; + assert!(unknown.is_empty(), "no unknown sids on a fresh-attrs DP"); + assert_eq!(state.series_resolver.len(), 1, "resolver minted one sid"); + assert_eq!( + state.sketch_index.instance_count(), + 1, + "SketchIndex registered one instance" + ); + + drop(state); + let _ = drain.await; + } + + #[tokio::test] + async fn unknown_sid_with_empty_attrs_is_returned_in_response() { + let (state, drain) = make_state().await; + // sid != 0, no attrs — resolver doesn't know it; should land in + // unknown_sids and the DP must be dropped (no instance registered). + let dp = DdSketchDataPoint { + attributes: Vec::new(), + start_time_unix_nano: 0, + time_unix_nano: 5_000_000, + sketch: vec![9], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: 7777, + }; + let req = build_request("http_latency_ms", dp); + + let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; + assert_eq!(unknown, vec![7777]); + assert_eq!(state.series_resolver.len(), 0); + assert_eq!(state.sketch_index.instance_count(), 0); + + drop(state); + let _ = drain.await; + } + + #[tokio::test] + async fn sid_attrs_disagreement_signals_stale_sid_but_uses_resolved_value() { + let (state, drain) = make_state().await; + // First, mint the resolver's view by sending sid=0 with attrs. + let dp_seed = DdSketchDataPoint { + attributes: vec![kv("zone", "z0")], + start_time_unix_nano: 1_000_000, + time_unix_nano: 11_000_000, + sketch: vec![1], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: 0, + }; + let _ = route_modified_otlp_sketches_to_precompute( + &build_request("http_latency_ms", dp_seed), + &state, + ) + .await; + let resolved_sid = state.series_resolver.lookup( + "http_latency_ms", + &crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]), + ); + let resolved_sid = resolved_sid.expect("seed mints"); + + // Now arrive with the same attrs but a STALE sid. + let stale = resolved_sid.wrapping_add(123); + let dp_disagree = DdSketchDataPoint { + attributes: vec![kv("zone", "z0")], + start_time_unix_nano: 1_000_000, + time_unix_nano: 12_000_000, + sketch: vec![2], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: stale, + }; + let unknown = route_modified_otlp_sketches_to_precompute( + &build_request("http_latency_ms", dp_disagree), + &state, + ) + .await; + assert_eq!(unknown, vec![stale], "stale sid should be signalled"); + // Cache stays at the originally-resolved value — the second call + // returns the same sid via the canonical fingerprint. + let still_resolved = state + .series_resolver + .lookup( + "http_latency_ms", + &crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]), + ) + .expect("still cached"); + assert_eq!(still_resolved, resolved_sid); + + drop(state); + let _ = drain.await; + } +} diff --git a/asap-query-engine/src/engines/simple/engine.rs b/asap-query-engine/src/engines/simple/engine.rs index 51de3caea..01dbd2d21 100644 --- a/asap-query-engine/src/engines/simple/engine.rs +++ b/asap-query-engine/src/engines/simple/engine.rs @@ -79,6 +79,64 @@ fn replace_metric_token(haystack: &str, needle: &str, replacement: &str) -> Stri out } +/// Phase 5 helper — extract `(metric_name, label_matcher_key_set)` from a +/// PromQL query for warm-tier candidate selection. Walks the AST to find +/// the first `VectorSelector` / `MatrixSelector`, returns its metric name +/// (drawn either from `vs.name` or from a `__name__=...` matcher) and +/// the user-specified label-matcher KEYS (excluding the synthetic +/// `__name__`). Returns `None` for queries that don't reference a +/// concrete metric. +/// +/// Intentionally lightweight: callers use the result to filter warm-tier +/// candidates via `SketchIndex::instances_matching`. Any over-approximation +/// is tolerable — the candidates are subsequently classified, and on +/// `Ghost` / `Unknown` outcomes the query falls through to the archive +/// engine via the EngineRouter's `CapabilityMiss` failover. +fn extract_metric_and_label_keys( + query: &str, +) -> Option<(String, std::collections::BTreeSet)> { + use promql_parser::parser::Expr; + let ast = promql_parser::parser::parse(query).ok()?; + + fn walk( + expr: &Expr, + ) -> Option<( + String, + std::collections::BTreeSet, + )> { + match expr { + Expr::VectorSelector(vs) => { + let mut keys = std::collections::BTreeSet::new(); + let mut metric = vs.name.clone().unwrap_or_default(); + for m in &vs.matchers.matchers { + if m.name == "__name__" { + if metric.is_empty() { + metric = m.value.clone(); + } + continue; + } + keys.insert(m.name.clone()); + } + if metric.is_empty() { + None + } else { + Some((metric, keys)) + } + } + Expr::MatrixSelector(ms) => walk(&Expr::VectorSelector(ms.vs.clone())), + Expr::Call(call) => call.args.args.iter().find_map(|a| walk(a)), + Expr::Aggregate(agg) => walk(&agg.expr), + Expr::Binary(bin) => walk(&bin.lhs).or_else(|| walk(&bin.rhs)), + Expr::Subquery(sq) => walk(&sq.expr), + Expr::Paren(p) => walk(&p.expr), + Expr::Unary(u) => walk(&u.expr), + _ => None, + } + } + + walk(&ast) +} + /// Length of the UTF-8 character starting at `b` (the first byte). /// Returns 1 for invalid leading bytes, never panics. fn utf8_char_len(b: u8) -> usize { @@ -224,6 +282,14 @@ pub struct SimpleEngine { /// [`Self::with_schema_registry`] to share the same registry the /// ingest path is reconciling. schema_registry: Arc, + /// Phase 5 — warm-tier sketch index. When `Some`, the trait's + /// `execute` adapter classifies the query's metric/group-by against + /// the index and short-circuits to `EngineError::CapabilityMiss` when + /// no warm-tier identity covers the request — driving the + /// EngineRouter's archive failover (Phase 6). When `None`, the + /// engine behaves as it did before Phase 5 wire-in (every query + /// goes through `handle_query`'s legacy path). + sketch_index: Option>, } impl SimpleEngine { @@ -405,9 +471,22 @@ impl SimpleEngine { query_language, controller_client: None, schema_registry: Arc::new(crate::stores::sketch_db::SchemaRegistry::empty()), + sketch_index: None, } } + /// Phase 5 — attach the shared `SketchIndex` so the `QueryEngine` + /// trait adapter's classify+failover logic is active. Without this + /// call, the engine keeps the pre-Phase-5 behavior (route every + /// query through `handle_query`). + pub fn with_sketch_index( + mut self, + index: Arc, + ) -> Self { + self.sketch_index = Some(index); + self + } + /// Take a fresh snapshot of the current `StreamingConfig`. Each /// call observes whatever was most recently pushed through PR #10's /// `POST /api/v1/streaming-config` endpoint. The returned `Arc` @@ -3672,6 +3751,65 @@ impl crate::routing::engine_router::QueryEngine for SimpleEngine { &self, query: &str, ) -> Result { + // Phase 5 wire-in (refactor 2026-05) — classify against the + // sketch-warm-tier index BEFORE handing the query to + // `handle_query`. The classification is intentionally crude + // because the warm-tier sketch reducer is still a stub: as soon + // as ANY sid is `Ghost` / `Unknown`, OR no instance even matches + // the metric / group-by KEY set, we surface + // `EngineError::CapabilityMiss(SketchWarmTier, ...)` so the + // EngineRouter (Phase 6) fails over to the archive engine. + // + // When `instances_matching` returns sids that all classify as + // `Hit`, we still fall through to `handle_query`'s legacy code + // path — wiring per-Capability sketch reducers on top of + // `SketchIndex.query_range` is out of scope for this PR and + // tracked as a follow-up. The "hybrid stitch" covering + // `[t0..t1']` from warm + `[t1'..t1]` from archive is also + // deferred (`QueryResult` would need timestamp coverage + // metadata to express it). + if let Some(idx) = self.sketch_index.as_ref() { + if let Some((metric_name, required_keys)) = + extract_metric_and_label_keys(query) + { + let candidates = idx.instances_matching(&metric_name, &required_keys); + if candidates.is_empty() { + return Err(crate::engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchWarmTier.data_source_id(), + format!( + "SketchWarmTier has no instance for metric `{metric_name}` \ + with group_by_keys ⊇ {:?}", + required_keys + ), + )); + } + let mut all_hit = true; + for sid in &candidates { + match idx.classify(*sid) { + crate::stores::sketch_index::SidLookup::Hit => {} + crate::stores::sketch_index::SidLookup::Ghost + | crate::stores::sketch_index::SidLookup::Unknown => { + all_hit = false; + break; + } + } + } + if !all_hit { + return Err(crate::engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchWarmTier.data_source_id(), + format!( + "SketchWarmTier ghost/unknown sid for metric `{metric_name}` \ + — failing over to archive" + ), + )); + } + // All sids `Hit` → fall through to the legacy path. + // Per-Capability reducer over `query_range` is a + // follow-up; for now `handle_query` answers from the + // legacy `SimpleMapStore`. See block comment above. + } + } + // `handle_query` is sync + needs a `time: f64` (epoch millis as float). // The router doesn't pass a query time, so we use wall-clock now — // matches `GorillaQueryEngine::execute`'s convention. @@ -5933,3 +6071,142 @@ mod cms_rate_capability_tests { ); } } + +/// Phase 5 — `QueryEngine::execute` warm-tier classification tests. +/// Pre-Phase-5 the trait adapter unconditionally delegated to +/// `handle_query`. After Phase 5 wire-in, when a `SketchIndex` is +/// attached, the adapter classifies first and surfaces +/// `EngineError::CapabilityMiss(SketchWarmTier, ...)` on Ghost / Unknown +/// / no-instance outcomes so the EngineRouter (Phase 6) can fall +/// through to the archive engine. +#[cfg(test)] +mod warm_tier_classify_tests { + use super::*; + use crate::data_model::{CleanupPolicy, HotReloadStreamingConfig, InferenceConfig}; + use crate::engines::EngineError; + use crate::routing::engine_router::QueryEngine as _; + use crate::stores::sketch_db::simple_map_store::SimpleMapStore; + use crate::stores::sketch_index::{ + AccuracyBound, Capability, SketchConfig, SketchIndex, SketchInstanceMetadata, + SketchKindHandle, SketchSampleState, + }; + use std::collections::{BTreeMap, BTreeSet}; + + fn build_engine_with_index(idx: Arc) -> SimpleEngine { + let streaming_config = Arc::new(crate::data_model::StreamingConfig::default()); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config); + let inference_config = InferenceConfig::new( + crate::data_model::QueryLanguage::promql, + CleanupPolicy::NoCleanup, + ); + SimpleEngine::new_with_hot_reload( + store, + inference_config, + hot_reload, + 15000, + crate::data_model::QueryLanguage::promql, + ) + .with_sketch_index(idx) + } + + fn dd_meta(sid: u64, metric: &str, group_by: &[&str]) -> SketchInstanceMetadata { + let cfg = SketchConfig::DDSketch { relative_accuracy: 0.01 }; + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys: group_by.iter().map(|s| s.to_string()).collect::>(), + capability: Capability::QuantileApprox(SketchKindHandle::DDSketch), + sketch_kind: SketchKindHandle::DDSketch, + sketch_config: cfg.clone(), + accuracy: AccuracyBound::from_config(&cfg), + first_seen_unix_ms: 0, + } + } + + #[tokio::test] + async fn execute_returns_capability_miss_when_no_instance_matches() { + // No instance for `unknown_metric` is registered → adapter must + // capability-miss rather than burn a `handle_query` round-trip. + let idx = Arc::new(SketchIndex::new()); + let engine = build_engine_with_index(idx); + let err = engine.execute("unknown_metric{zone=\"z0\"}").await.expect_err( + "warm-tier with no matching instance must yield CapabilityMiss", + ); + match err { + EngineError::CapabilityMiss { engine_id, .. } => { + assert_eq!( + engine_id, + asap_types::StorageBackend::SketchWarmTier.data_source_id() + ); + } + other => panic!("expected CapabilityMiss, got {other:?}"), + } + } + + #[tokio::test] + async fn execute_returns_capability_miss_when_classify_is_ghost() { + // Register instance metadata but never call append_sample → the + // sid classifies as Ghost. Adapter must short-circuit to + // CapabilityMiss so the EngineRouter (Phase 6) fails over. + let idx = Arc::new(SketchIndex::new()); + idx.register(dd_meta(1, "http_latency_ms", &["zone"])); + let engine = build_engine_with_index(idx); + let err = engine + .execute("http_latency_ms{zone=\"z0\"}") + .await + .expect_err("ghost classification must yield CapabilityMiss"); + match err { + EngineError::CapabilityMiss { engine_id, detail } => { + assert_eq!( + engine_id, + asap_types::StorageBackend::SketchWarmTier.data_source_id() + ); + assert!( + detail.contains("ghost") || detail.contains("Ghost") || detail.contains("unknown"), + "detail mentions ghost/unknown: {detail}" + ); + } + other => panic!("expected CapabilityMiss, got {other:?}"), + } + } + + #[tokio::test] + async fn execute_proceeds_to_handle_query_when_all_sids_hit() { + // Register an instance AND append a sample so the sid Hits. The + // adapter then falls through to `handle_query`; with an empty + // store + no inference-config patterns, that path returns its + // own CapabilityMiss — but the failure mode is the legacy "no + // compatible aggregation" detail, distinct from the warm-tier + // ghost/unknown detail. The contract verified here is "Hit + // does NOT short-circuit to the warm-tier-specific miss". + let idx = Arc::new(SketchIndex::new()); + idx.register(dd_meta(2, "http_latency_ms", &["zone"])); + idx.append_sample( + 2, + BTreeMap::from([("zone".to_string(), "z0".to_string())]), + (1_000, 1_010), + SketchSampleState { + bytes: vec![0], + encoding: crate::stores::sketch_index::SketchEncoding::ProtoFull, + }, + ); + + let engine = build_engine_with_index(idx); + let result = engine.execute("http_latency_ms{zone=\"z0\"}").await; + match result { + Err(EngineError::CapabilityMiss { detail, .. }) => { + assert!( + detail.contains("no compatible aggregation"), + "Hit path delegated to handle_query, which produced legacy miss: {detail}" + ); + } + other => panic!( + "expected handle_query's legacy CapabilityMiss after Hit, got {other:?}" + ), + } + } +} diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index b6519f0f6..c7715130d 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -428,6 +428,21 @@ async fn main() -> Result<()> { )) }; + // Phase 4 + 5 wire-in (refactor 2026-05): allocate the shared + // SeriesIdResolver + SketchIndex once. The OTLP receive path + // (sid resolution + unknown_series_ids stamping; SketchIndex + // .append_sample on every modified-OTLP sketch DP) AND the + // SimpleEngine query path (SketchIndex.classify / query_range + // for warm-tier reads) hold clones of these Arcs. Allocated + // here before BOTH the SimpleEngine and the precompute engine + // are constructed so both can be wired with a single canonical + // instance — even when precompute is disabled, the engine still + // needs the index for the Phase 6 archive failover trigger. + let series_resolver = Arc::new( + query_engine_rust::drivers::ingest::series_resolver::SeriesIdResolver::new(), + ); + let sketch_index = Arc::new(query_engine_rust::stores::sketch_index::SketchIndex::new()); + // Setup query engine. SimpleEngine shares the same // HotReloadStreamingConfig handle as the HTTP server, so a POST // to /api/v1/streaming-config is observable by the next query @@ -441,7 +456,13 @@ async fn main() -> Result<()> { hot_reload_config.clone(), args.prometheus_scrape_interval, args.query_language, - ); + ) + // Phase 5 wire-in (refactor 2026-05): hand the warm-tier + // SketchIndex to the query engine so SidLookup classification + // drives the Phase 6 archive failover via + // EngineError::CapabilityMiss when the warm tier is empty + // / ghost / unknown. + .with_sketch_index(sketch_index.clone()); if let Some(controller_endpoint) = args.controller_endpoint.as_ref() { info!( "Capability-miss notifications enabled → {}", @@ -542,8 +563,13 @@ async fn main() -> Result<()> { schema_persist_path: args.schema_persist_path.clone(), }; let output_sink = Arc::new(StoreOutputSink::new(store.clone())); - let engine = - PrecomputeEngine::new(precompute_config, hot_reload_config.clone(), output_sink); + let engine = PrecomputeEngine::new( + precompute_config, + hot_reload_config.clone(), + output_sink, + series_resolver.clone(), + sketch_index.clone(), + ); let worker_diagnostics = engine.diagnostics(); let ingest_state = engine.ingest_state(); info!("Starting precompute engine (OTLP-fed; no HTTP ingest port)"); diff --git a/asap-query-engine/src/precompute_engine/engine.rs b/asap-query-engine/src/precompute_engine/engine.rs index e3cdffefa..90d7ceb17 100644 --- a/asap-query-engine/src/precompute_engine/engine.rs +++ b/asap-query-engine/src/precompute_engine/engine.rs @@ -38,6 +38,8 @@ impl PrecomputeEngine { config: PrecomputeEngineConfig, hot_reload_config: HotReloadStreamingConfig, output_sink: Arc, + series_resolver: Arc, + sketch_index: Arc, ) -> Self { let worker_group_counts = (0..config.num_workers) .map(|_| Arc::new(AtomicUsize::new(0))) @@ -92,6 +94,8 @@ impl PrecomputeEngine { schemas, pass_raw_samples: config.pass_raw_samples, sketch_snapshots: dashmap::DashMap::new(), + series_resolver, + sketch_index, }); Self { diff --git a/asap-query-engine/src/precompute_engine/ingest_handler.rs b/asap-query-engine/src/precompute_engine/ingest_handler.rs index b480b3373..4c8c9d81d 100644 --- a/asap-query-engine/src/precompute_engine/ingest_handler.rs +++ b/asap-query-engine/src/precompute_engine/ingest_handler.rs @@ -53,6 +53,18 @@ pub struct IngestState { /// timestamp so long-running deployments don't leak memory /// on retired series. pub sketch_snapshots: dashmap::DashMap>, + /// Phase 4 — centralized series_id resolver. Shared across the OTLP + /// receive path (sid resolution + `unknown_series_ids` population) and + /// the `ResolveSeriesIDs` RPC (eager batch resolution from the agent's + /// exporter). Holding it on `IngestState` lets every ingest source + /// reach the same idempotent compute-or-mint cache. + pub series_resolver: Arc, + /// Phase 5 — two-level sketch warm tier (instance metadata + + /// per-sid columnar state). Populated by the OTLP ingest path on + /// every modified-OTLP first-class sketch DataPoint; queried by + /// the `SimpleEngine` query path (warm-tier hit / ghost / unknown + /// classification drives the Phase 6 archive failover). + pub sketch_index: Arc, } impl IngestState { @@ -168,6 +180,10 @@ mod tests { schemas, pass_raw_samples: false, sketch_snapshots: dashmap::DashMap::new(), + series_resolver: Arc::new( + crate::drivers::ingest::series_resolver::SeriesIdResolver::new(), + ), + sketch_index: Arc::new(crate::stores::sketch_index::SketchIndex::new()), }); let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); diff --git a/asap-query-engine/src/stores/sketch_index.rs b/asap-query-engine/src/stores/sketch_index.rs index cd67d471d..6d5eac1a6 100644 --- a/asap-query-engine/src/stores/sketch_index.rs +++ b/asap-query-engine/src/stores/sketch_index.rs @@ -328,6 +328,31 @@ impl SketchIndex { .collect() } + /// Find every registered sid whose instance matches `metric_name` and + /// whose `group_by_keys` is a superset of (or equal to) the user's + /// requested label-key set. Phase 5 query path uses this to pick + /// candidate sids for warm-tier dispatch — a sid whose group-by KEYS + /// don't cover the user's PromQL label matchers can't answer the + /// query and must fall through to archive. + /// + /// Returns `Vec` rather than an iterator so callers can release + /// the read lock immediately. The `instances` map is read-mostly + /// (one write per first-seen sid), so taking the lock per query is + /// inexpensive. + pub fn instances_matching( + &self, + metric_name: &str, + required_keys: &BTreeSet, + ) -> Vec { + let g = self.instances.read().unwrap(); + g.iter() + .filter(|(_, m)| { + m.metric_name == metric_name && required_keys.is_subset(&m.group_by_keys) + }) + .map(|(sid, _)| *sid) + .collect() + } + /// Number of distinct sids carrying state (excludes ghosts). pub fn series_len(&self) -> usize { self.series.len() diff --git a/crates/asap_otel_proto/proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto b/crates/asap_otel_proto/proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto index 0c105abb4..0b4d12fe9 100644 --- a/crates/asap_otel_proto/proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto +++ b/crates/asap_otel_proto/proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto @@ -29,6 +29,34 @@ option go_package = "go.opentelemetry.io/proto/otlp/collector/metrics/v1"; // central collector. service MetricsService { rpc Export(ExportMetricsServiceRequest) returns (ExportMetricsServiceResponse) {} + + // Refactor-2026-05 (Phase 4): Eager batch resolution of series identifiers. + // Senders that have an attribute-only batch ready before the next Export + // can pre-resolve sids in one round-trip, then attach the returned sids + // to subsequent Exports without sending attributes again. The resolver + // is idempotent: same `(metric_name, attributes_fingerprint)` always + // produces the same sid for the lifetime of the cache (see design doc + // §5.4 "Idempotency invariant on `ResolveSeriesIDs`"). + rpc ResolveSeriesIDs(ResolveSeriesIDsRequest) returns (ResolveSeriesIDsResponse) {} +} + +message ResolveSeriesIDsRequest { + repeated SeriesQuery queries = 1; +} + +message SeriesQuery { + // The metric this attribute set belongs to. + string metric_name = 1; + // Opaque fingerprint of the attribute set. Senders MUST use the same + // canonical fingerprint algorithm as the receiver (sorted keys joined + // by `key=value;`); the receiver re-derives the cache key from this + // bytes blob, so any disagreement on canonicalization produces a + // cache miss and a fresh sid. + bytes attributes_fingerprint = 2; +} + +message ResolveSeriesIDsResponse { + repeated SeriesAssignment assignments = 1; } message ExportMetricsServiceRequest {