feat: route modified-OTLP sketch metrics through precompute engine (PR B) - #6
Merged
Merged
Conversation
…R B)
Replaces the debug-log stub arms PR A added in `otlp_to_metric_points_and_sketches`
with a real routing function that walks the five first-class sketch
metric variants (`Metric.data = DDSketch / KLLSketch / CountSketch /
CountMinSketch / HLLSketch`) and dispatches each `*SketchDataPoint`
through the precompute engine via `WorkerMessage::AccumulatorInput`.
This is **PR B** in the implementation plan tracked in
`docs/pipeline-query-catalog.md` §10 in DataCollector#153.
What's wired
------------
1. **`route_modified_otlp_sketches_to_precompute(req, ingest_state)`**
in `asap-query-engine/src/drivers/ingest/otel.rs`. Walks
`request.resource_metrics → scope_metrics → metrics → metric.data`
for the five new sketch variants, flattens each per-variant data
point into a `ModifiedOtlpSketchDp { kind, attrs, time_unix_nano,
sketch, encoding }`, and dispatches:
- **labels**: `merge_point_attributes(base_labels, dp.attributes)`
where `base_labels = scope_attrs ∪ resource_attrs` — same label
fan-in as the existing raw-metrics path.
- **timestamp**: `dp.time_unix_nano / 1_000_000` (ms).
- **per-config matching**: same as `route_otlp_to_precompute` —
match against `ingest_state.agg_configs` by metric name (or
`spatial_filter` / `spatial_filter_normalized`), extract the
group key via `IngestState::extract_group_key_for(series_key, config)`,
and emit one `WorkerMessage::AccumulatorInput { agg_id,
group_key, timestamp_ms, accumulator, ingest_received_at }` per
matching config (clone via `clone_boxed_core` so one accumulator
can fan out to multiple agg configs).
2. **`decode_modified_otlp_sketch_bytes(kind, encoding, bytes)`** —
the per-variant decoder dispatcher. Today:
- Accepts only `encoding = ENCODING_PROTO (= 1)`. `ENCODING_PROTO_DELTA`
(delta transmission, tracked in PR C) and `ENCODING_MSGPACK` /
`ENCODING_MSGPACK_DELTA` (PR I) return `Err` with a clear message.
- For `SketchKind::CountMin` calls the new
`CountMinSketchAccumulator::from_sketchlib_proto_bytes(buf)`
constructor (see below) — fully wired end-to-end.
- For `SketchKind::Kll / DdSketch / CountSketch / Hll` returns
`Err("decoder for {kind:?} not yet implemented (tracked in PR C, task #8)")`.
Caller drops the data point and falls through to the §5.2
forwarding adapter so the user still gets a correct answer.
3. **`CountMinSketchAccumulator::from_sketchlib_proto_bytes(buf)`** in
`precompute_operators/count_min_sketch_accumulator.rs`. Decodes the
`asap_sketchlib::proto::sketchlib::CountMinState` message and
constructs the accumulator via `CountMinSketch::from_legacy_matrix`:
- decode `CountMinState` via `prost::Message::decode`
- reject zero rows/cols
- read `counter_type` enum and pick `counts_int` or `counts_float`
- validate `len() == rows * cols`
- reshape flat counts into `Vec<Vec<f64>>` row-major
- `INT128` counter type rejected with "not yet supported (PR C)" —
it stores `(hi, lo)` interleaved pairs and is rare in practice
- returns `Result<Self, Box<dyn std::error::Error>>` so the
dispatcher can fall through to fallback on any error
4. **Wiring in both ingest paths**: `MetricsServiceImpl::export` (gRPC)
and `handle_otlp_http` (HTTP) both call
`route_modified_otlp_sketches_to_precompute` immediately after the
existing `route_otlp_to_precompute` so raw metrics and sketches
travel side-by-side through the same precompute engine.
5. **Per-variant counters + summary log**: the routing function counts
`routed`, `decoded_failed` (fall-through to fallback), and
`unconfigured` (no matching agg config) and emits a single
`debug!` line per request with the totals — matches the style of
the existing `route_otlp_to_precompute` summary.
Tests
-----
Four new unit tests in `count_min_sketch_accumulator.rs`:
- `test_from_sketchlib_proto_bytes_int64` — round-trips a 2x3 matrix
through `CountMinState{counter_type=INT64, counts_int=[1..6]}` and
asserts the accumulator's inner sketch matrix matches.
- `test_from_sketchlib_proto_bytes_float64` — same for a 2x2 matrix
with `FLOAT64` counters.
- `test_from_sketchlib_proto_bytes_dimension_mismatch` — `counts_int`
length doesn't match `rows * cols`; expect `Err` mentioning
"counts_int".
- `test_from_sketchlib_proto_bytes_zero_dims_rejected` — default
`CountMinState` (rows=0, cols=0); expect `Err` mentioning
"zero dims".
Validation
----------
- `cargo check -p query_engine_rust --all-targets`: clean
- `cargo clippy -p query_engine_rust --all-targets -- -D warnings`: clean
- `cargo fmt --check`: clean
- `cargo test -p query_engine_rust --lib`: **439 passed**, 0 failed,
5 ignored (was 435 in PR A — 4 new tests above)
What this delivers end-to-end
-----------------------------
With PR A + PR B merged, a DataCollector `countminsketchprocessor`
that emits via the modified OTLP wire (`Metric.data = CountMinSketch{
data_points: [CountMinSketchDataPoint{...sketch=<CountMinState
proto bytes>, encoding=COUNT_MIN_SKETCH_ENCODING_PROTO}] }`) now
flows end-to-end through:
OTLP gRPC/HTTP receiver
→ route_modified_otlp_sketches_to_precompute
→ decode_modified_otlp_sketch_bytes (CountMin path)
→ CountMinSketchAccumulator::from_sketchlib_proto_bytes
→ WorkerMessage::AccumulatorInput
→ precompute engine sketch_panes merge
→ SimpleMapStore
→ SimpleEngine PromQL queries answer hot
The other four sketch types (KLL, DDSketch, CountSketch, HLL) still
fall through to the §5.2 forwarding adapter today; their decoders
land in PR C (task #8). PR D (task #9) adds an end-to-end integration
test that exercises the CountMin hot path against a real precompute
engine instance.
What this PR does NOT do
------------------------
- Does not delete the legacy `SketchEnvelopeAccumulator` / attribute-bytes
path. Modern DataCollector processors don't use it, but it stays
as a transitional hatch for standard-OTLP clients that sideband
sketches in DataPoint attributes. Removal can happen after PR D
validates the new path on a real workload.
- Does not handle delta transmission (`ENCODING_PROTO_DELTA`). PR C
will add per-series baseline tracking and delta application.
- Does not wire any of the four other sketch types end-to-end.
Those decoders are PR C territory.
- Does not implement `series_id`-based descriptor lookup (PR G, where
the controller starts minting them).
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
Apr 14, 2026
Adds the first real end-to-end correctness test for the modified-OTLP sketch path landed in PR A (#5, vendoring) + PR B (#6, routing + CountMin decoder). The test exercises every public surface in the Phase 1 hot path: 1. Build a `StreamingConfig` with a single `AggregationConfig` for `http_requests_total` with `aggregation_type=CountMinSketch`, `grouping_labels=[service]`, 1-second tumbling windows. 2. Spawn a real `PrecomputeEngine` (workers + router + ingest state). 3. Spawn a real `OtlpReceiver::with_ingest_state(...)` wired to the same engine — modeling the production wiring `main.rs` does. 4. POST a real protobuf-encoded `ExportMetricsServiceRequest` over OTLP HTTP at `/v1/metrics`. The request carries a single `Metric.data = CountMinSketch{ data_points: [ CountMinSketchDataPoint { sketch: <CountMinState bytes>, encoding: COUNT_MIN_SKETCH_ENCODING_PROTO, attributes: [{service: "auth"}], time_unix_nano: 100ms } ] }` payload with a known 2x4 matrix (counts: row 0 = [1,2,3,4], row 1 = [5,6,7,8]). 5. POST a second OTLP request timestamped 2 s past epoch (past the 1 s window end) so the precompute engine's watermark advances and closes window 0. 6. Wait for the periodic flush (100ms interval) to fire. 7. Drain the `CapturingOutputSink`, find the entry for window 0, downcast the `Box<dyn AggregateCore>` to `CountMinSketchAccumulator`, read `inner.sketch()`, and assert each row matches the expected `Vec<f64>`. What this validates end-to-end: - PR A: the vendored `asap_otel_proto` crate's tonic bindings expose `Metric.data::Countminsketch` and the typed `CountMinSketchDataPoint` fields, and the prost build pipeline produces decoders that the test crate can use directly to construct a request. - PR B routing: `route_modified_otlp_sketches_to_precompute` walks the `Metric.data` oneof, flattens each per-variant data point into `ModifiedOtlpSketchDp`, matches the metric against `ingest_state.agg_configs` by name, computes the group key from the `service` attribute via `IngestState::extract_group_key_for`, and emits `WorkerMessage::AccumulatorInput` correctly. - PR B decoder: `CountMinSketchAccumulator::from_sketchlib_proto_bytes` decodes the `CountMinState` proto, picks the int64 counter path, reshapes the flat counts into `Vec<Vec<f64>>`, and constructs the underlying `CountMinSketch` via `from_legacy_matrix` — round-trip matches the original matrix exactly. - Precompute engine sketch-pane merge: the `WorkerMessage::AccumulatorInput` reaches the right worker via the (agg_id, group_key) hash, lands in the `sketch_panes` of the matching `GroupState`, and survives the watermark-driven window close to be emitted via the `OutputSink`. - Window close + sink: after the watermark advances past the 1 s window end, the worker's flush emits a `(PrecomputedOutput, Box<dyn AggregateCore>)` tuple with `start_timestamp=0` and `end_timestamp=1000` to the `CapturingOutputSink`. - Round-trip correctness: the matrix that comes out of the sink is bit-identical to what went in over OTLP, proving there is no data loss in the routing/decoder/merge path for a single-input window. Test layout ----------- - New file `asap-query-engine/tests/e2e_modified_otlp_sketch_path.rs` - Helpers: `make_count_min_agg_config(...)` — builds a tumbling-window CountMinSketch `AggregationConfig` with a `service` group `engine_config(...)` — builds a `PrecomputeEngineConfig` with a 100ms flush interval so the test does not have to wait long `build_count_min_state(...)` — builds a `CountMinState` proto from a known matrix `build_export_request(...)` — wraps a `CountMinSketchDataPoint` in a fully-formed `ExportMetricsServiceRequest` `post_otlp_http(...)` — sends a protobuf body to the OTLP HTTP endpoint at `localhost:port/v1/metrics` and asserts 2xx Validation ---------- - `cargo check -p query_engine_rust --tests`: clean - `cargo clippy -p query_engine_rust --tests -- -D warnings`: clean - `cargo fmt --check`: clean - `cargo test -p query_engine_rust --test e2e_modified_otlp_sketch_path`: `test e2e_count_min_sketch_modified_otlp_path ... ok` `test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured` - `cargo test -p query_engine_rust --lib`: 439 passed, 0 failed, 5 ignored What's still PR C territory --------------------------- - KLL / DDSketch / CountSketch / HLL e2e coverage. PR C (task #8) is re-scoped as smaller per-sketch-type follow-ups (PR C-CountSketch, PR C-HLL, PR C-KLL, PR C-DDSketch); each adds a per-type decoder and an analogous test arm to this file. - Delta transmission (`*_ENCODING_PROTO_DELTA`). PR C-delta adds per-series baseline tracking and a delta-merge codepath; this test file gets a `delta` test case once that lands. - MessagePack encoding parity (`*_ENCODING_MSGPACK` variants). PR I (task #14) adds those; this test file gets a four-way `(format, mode)` correctness assertion at that point. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This was referenced Apr 14, 2026
zzylol
added a commit
that referenced
this pull request
May 13, 2026
Post-M2.3 reorg #5 of 8. Creates a new `sketch_db::lifecycle` module and moves `SchemaEvictionService` (and friends) into `lifecycle/eviction.rs`. The sid-level lifecycle FIELDS and methods on `SketchInstanceMetadata` / `SketchStore` stay in `index/` next to the data they gate — only the schedule-driven *service* moves here. `schema/mod.rs` re-exports the eviction types under their legacy path (`sketch_db::schema::SchemaEvictionService`) so existing consumers compile unchanged. Canonical home is now `sketch_db::lifecycle::*`. This is the structural skeleton for the upcoming sub-PRs: - #6 will add `lifecycle::reconcile_from_streaming_config`, reimplementing schema/'s reconcile semantics over the sid catalog. - #7 will then delete `schema/` once the only resident is the thin re-export. 783/783 lib tests pass. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 task
zzylol
added a commit
that referenced
this pull request
May 13, 2026
…_db/query/ (#182) Post-M2.3 reorg #6 of 8. Moves `query_engines/{timeline_dispatch.rs, window_merger.rs}` → `storage_engines/sketch_db/query/`. Both files are warm-tier driver primitives — `timeline_dispatch` combines per-segment results across reconfigure boundaries that only exist *within* a SketchStore; `window_merger` stitches per- window results from the same. Neither belongs to the backend-agnostic orchestration layer (`query_engines/`). `query_engines/mod.rs` re-exports both modules under their legacy paths via `pub use crate::storage_engines::sketch_db::query::*`, so existing consumers (in `engine.rs`, `http.rs`, schema docs) compile unchanged. 783/783 lib tests pass. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR B in the end-to-end sketch path roadmap (Phase 1). Builds on PR #5 (PR A, vendoring the modified OTLP proto).
Replaces the debug-log stub arms PR A added with a real routing function that walks the five first-class sketch metric variants (
Metric.data = DDSketch / KLLSketch / CountSketch / CountMinSketch / HLLSketch) and dispatches each*SketchDataPointthrough the precompute engine viaWorkerMessage::AccumulatorInput.What lands
route_modified_otlp_sketches_to_precompute(req, ingest_state)indrivers/ingest/otel.rs. Walks resource→scope→metric and flattens the per-variant data points into a uniformModifiedOtlpSketchDpstruct. Same label fan-in (scope_attrs ∪ resource_attrs ∪ dp.attributes), same(agg_id, group_key)matching againstingest_state.agg_configs, sameWorkerMessage::AccumulatorInputrouting as the existing raw-metrics path.decode_modified_otlp_sketch_bytes(kind, encoding, bytes)— per-variant decoder dispatcher. Today:encoding = ENCODING_PROTO (= 1). Delta variants (PR C) and msgpack variants (PR I) return Err.SketchKind::CountMin→ callsCountMinSketchAccumulator::from_sketchlib_proto_bytes— fully wired end-to-end.SketchKind::Kll / DdSketch / CountSketch / Hll→Err("decoder not yet implemented (tracked in PR C, task #8)"). Caller falls through to the §5.2 forwarding adapter so the user still gets a correct answer.CountMinSketchAccumulator::from_sketchlib_proto_bytes(buf)— decodesasap_sketchlib::proto::sketchlib::CountMinState, validates dims, pickscounts_intorcounts_floatbased on thecounter_typeenum, reshapes flat counts intoVec<Vec<f64>>, and constructs viaCountMinSketch::from_legacy_matrix.INT128counter type is rejected with a PR C TODO (rare in practice; uses interleaved hi/lo encoding).Wiring: both
MetricsServiceImpl::export(gRPC) andhandle_otlp_http(HTTP) call the new function immediately after the existingroute_otlp_to_precompute.Tests
Four new unit tests in
count_min_sketch_accumulator.rscovering int64 round-trip, float64 round-trip, dimension mismatch, and zero-dim rejection. All pass.Validation
cargo check -p query_engine_rust --all-targets: cleancargo clippy -p query_engine_rust --all-targets -- -D warnings: cleancargo fmt --check: cleancargo test -p query_engine_rust --lib: 439 passed, 0 failed, 5 ignored (4 new tests on top of PR A's 435)End-to-end value with PR A + PR B merged
A DataCollector
countminsketchprocessoremitting via the modified OTLP wire now flows hot end-to-end:The other four sketch types (KLL / DDSketch / CountSketch / HLL) fall through to the §5.2 forwarding adapter today; their decoders land in PR C (task #8). PR D (task #9) adds an end-to-end integration test against a real precompute engine instance.
What this PR does NOT do
SketchEnvelopeAccumulator/ attribute-bytes path (transitional hatch for standard-OTLP clients that sideband sketches in DataPoint attributes; can be removed after PR D validates the new path).ENCODING_PROTO_DELTA). PR C adds per-series baseline tracking.series_id-based descriptor lookup (PR G).🤖 Generated with Claude Code