diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index ed2d4df2..c69419ed 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -953,7 +953,9 @@ async fn process_via_simple_engine( // before falling through to the unsupported-query branch // — `execute` uses // `idx.sids_for_policy(fp)` + `SketchReducer::evaluate` - // and handles sketches natively. + // and handles sketches natively, AND (since #273) unions + // `instances_matching` for sketches with `policy_fp = + // UNSET`. // // Trait dispatch loses `KeyByLabelNames` (the trait // returns just `QueryResult`); we surface an empty @@ -961,6 +963,18 @@ async fn process_via_simple_engine( // handles the same trait surface — the Prometheus // adapter renders an empty `metric: {}` object, a valid // shape that PromQL clients accept. + // + // Schema-retirement #5 status: an earlier draft of this + // PR reordered to "modern first, legacy as fallback" so + // the legacy path could be retired entirely. That broke + // `http_capability_miss_feedback_loop_closes_over_http` + // — the capability-miss notify side-effect happens + // inside legacy `find_compatible_aggregation_with_miss_notify` + // (engine.rs:~1772), and a pre-existing time=0 underflow + // bug at engine.rs:792 surfaces when legacy is reached + // via the modern-Err fallback because of subtle test + // setup state. Modern needs to spawn its own + // capability-miss notify before we can reorder cleanly. use crate::query_engines::routing::query_engine_routing::QueryEngine; let modern_result = state .query_engine diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 178f28bb..60094e14 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -3210,6 +3210,16 @@ impl ASAPQueryEngine { )); }; + // Schema-retirement #5 step 2: apply the same metric-rename + // rewrite the modern execute() instant path does, so range + // queries like `quantile_over_time(0.99, http_latency[5m])` + // bind to the suffixed series the agent's DDSketch processor + // emits. Mirrors the legacy `handle_query_promql` entry. + let query_owned = self + .resolve_sketch_metric_alias(query) + .unwrap_or_else(|| query.to_string()); + let query = query_owned.as_str(); + let analysis = control_plane::asap_tier_analysis::analyze_promql_for_asap_tier(query); @@ -3240,14 +3250,38 @@ impl ASAPQueryEngine { > = None; for candidate in &analysis.candidates { + // Resolve candidate → {sids} via the sid catalog. Schema- + // retirement #5: prefer `instances_matching` over the + // policy-fp reverse index — it's the more general + // primitive and works whether or not the ingest path was + // able to bind the sid back to a streaming-config policy. + // + // History: an earlier PR removed an `instances_matching` + // fallback under the assumption every production sid + // registration would populate `policy_fp`. The MVP smoke + // test (issue #271 / tracking #272) showed that + // assumption is wrong — sketches arriving from the agent + // carry the full wire-attr set rather than the streaming- + // config's `grouping_labels` subset, so + // `derive_sketch_policy_fp` returns `UNSET` and + // `sids_for_policy(fp)` returns empty. The agg_id-aware + // path is preserved for ExactAgg sids minted via + // `ingest_precompute_for_agg_config` (those carry a + // populated `policy_fp`) but its result is unioned with + // the catalog-walk result so we don't miss the sketches. let policy_fps = control_plane::asap_tier_analysis::find_matching_policies( &policy_registry, candidate, ); - let mut sids: Vec = Vec::new(); + let mut sids: std::collections::BTreeSet = + std::collections::BTreeSet::new(); for fp in &policy_fps { sids.extend(idx.sids_for_policy(*fp)); } + sids.extend(idx.instances_matching( + &candidate.metric_name, + &candidate.group_by_keys, + )); if sids.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( asap_types::StorageBackend::SketchStore.data_source_id(), @@ -3673,6 +3707,20 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // over to archive (no per-candidate hybrid stitch yet — // that's the documented follow-up). if let Some(idx) = self.sketch_index.as_ref() { + // Schema-retirement #5 step 2: apply the agent-side + // INGEST-time metric-rename rewrite (DDSketch/KLL + // `_quantile`, HLL `_hll`) here at the top of modern + // execute() so bare-metric PromQL still hits the + // suffixed series the ASAP tier actually holds. The + // legacy `handle_query_promql` did this rewrite at + // its own entry; with the legacy path slated for + // retirement, the modern path needs the same + // capability so it can fully supersede. + let query_owned = self + .resolve_sketch_metric_alias(query) + .unwrap_or_else(|| query.to_string()); + let query = query_owned.as_str(); + let analysis = control_plane::asap_tier_analysis::analyze_promql_for_asap_tier(query); // Branch 1 — the control plane analyzer rejects the shape. @@ -3746,22 +3794,30 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu if candidate.range_seconds > 0 { any_range_candidate = true; } - // Content-addressed sid lookup: find matching policies - // in the registry → resolve each policy_fp → {sids} - // via the reverse index. Both hops are O(1)-amortized. - // The legacy `instances_matching(metric, gbk)` - // fallback was retired in this PR — every production - // sid registration path now populates `policy_fp`, - // and sids that don't are intentionally unreachable - // (raw mode, etc. — they map to capability misses). + // Schema-retirement #5: resolve candidate → {sids} by + // unioning the policy-fp reverse index (fast path for + // ExactAgg sids minted via `ingest_precompute_for_agg_config` + // where `policy_fp` is set) with `instances_matching` + // (catalog walk that subset-matches on + // `group_by_keys`, covering raw sketches whose + // `derive_sketch_policy_fp` returned `UNSET` because + // the wire-attr set didn't match any streaming-config + // policy). The earlier policy-fp-only path returned + // empty for the MVP demo workload — see issue #271 / + // tracking #272. let policy_fps = control_plane::asap_tier_analysis::find_matching_policies( &policy_registry, candidate, ); - let mut sids: Vec = Vec::new(); + let mut sids: std::collections::BTreeSet = + std::collections::BTreeSet::new(); for fp in &policy_fps { sids.extend(idx.sids_for_policy(*fp)); } + sids.extend(idx.instances_matching( + &candidate.metric_name, + &candidate.group_by_keys, + )); if sids.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( asap_types::StorageBackend::SketchStore.data_source_id(), @@ -6266,6 +6322,62 @@ mod asap_tier_classify_tests { } other => panic!("expected CapabilityMiss fall-over to archive, got {other:?}")} } + + /// Schema-retirement #5 regression: a sketch sid registered with + /// a wider-than-requested `group_by_keys` and `policy_fp=UNSET` + /// must still be findable by the query path. Mirrors the MVP + /// smoke-test failure (issue #271 / tracking #272): the agent + /// emits DDSketch DPs carrying every wire attribute, so the sid + /// catalog ends up with `group_by_keys=[zone,rack,node,pod,...]` + /// and `derive_sketch_policy_fp` returns `UNSET` because no + /// streaming-config policy has that exact key set. The query + /// asks for `grouping=[zone]` — a subset. With the policy-fp-only + /// lookup the query returned `CapabilityMiss → archive`; with the + /// `instances_matching` fallback restored it resolves to the sid + /// (and bottoms out at the reducer's sample-state check rather + /// than at sid resolution). + #[tokio::test] + async fn full_attr_sketch_sid_findable_via_subset_grouping() { + let idx = Arc::new(SketchStore::new()); + // Register with the SUPERSET of attrs the agent would emit: + // zone, rack, node, pod — none of which the streaming-config + // would list directly in `grouping_labels=[zone]`. + idx.register(dd_meta(42, "http_latency_ms", &["node", "pod", "rack", "zone"])); + idx.append_sample( + 42, + BTreeMap::from([ + ("zone".to_string(), "z0".to_string()), + ("rack".to_string(), "r0".to_string()), + ("node".to_string(), "n0".to_string()), + ("pod".to_string(), "p0".to_string()), + ]), + (1_000, 1_010), + SketchSampleState { + bytes: vec![0], + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull}, + ); + + let engine = build_engine_with_index(idx); + // The query asks for grouping=[zone] (subset of registered + // group_by_keys). Pre-fix this returned CapabilityMiss because + // `sids_for_policy(UNSET)` is empty; post-fix the fallback + // finds sid 42 via `instances_matching` and the request + // proceeds to the reducer. + let result = engine + .execute("quantile_over_time(0.99, http_latency_ms{zone=\"z0\"}[5m])") + .await; + // The reducer can't produce a real quantile from the canned + // payload (just `vec![0]`), but it MUST reach the reducer — + // the sid-resolution-step CapabilityMiss with "no policy for + // metric" detail is the regression we're guarding against. + if let Err(EngineError::CapabilityMiss { detail, .. }) = &result { + assert!( + !detail.contains("has no policy for metric"), + "regression: sid was lost at policy-resolution step \ + instead of being found via instances_matching: {detail}" + ); + } + } } // ===========================================================================