feat: MessagePack encoding parity for modified-OTLP sketch hot path (PR I) - #9
Merged
Merged
Conversation
…PR I)
Adds MessagePack as a parallel encoding option alongside the
protobuf-encoded sketchlib `*State` path that PR B and PR C wired up
for all five modified-OTLP `Metric.data` sketch variants. The new
encoding is a cross-language contract between DataCollector (Go,
sketchlib-go) and ASAPQuery-backend (Rust, sketch-core) — both sides
serialize the same cross-language sketch-core wire struct via
MessagePack. This gives sketchlib-go a practical alternative to
prost-encoded sketchlib proto without touching existing paths.
## Proto
Adds two new enum variants to each of the five per-sketch encoding
enums in the vendored `opentelemetry.proto.metrics.v1`:
* `*_ENCODING_MSGPACK = 3`
* `*_ENCODING_MSGPACK_DELTA = 4`
Backward-compatible: existing producers emitting tags 0-2 see no
change. The vendored proto is the consumer side — DataCollector's
upstream proto can adopt matching enum names in a parallel PR without
breaking either side, since protobuf enums are just integer tags.
## Dispatcher (`drivers/ingest/otel.rs`)
`decode_modified_otlp_sketch_bytes` now matches on the full `encoding`
i32 value rather than only accepting `PROTO = 1`:
* `PROTO = 1` → per-sketch `from_sketchlib_proto_bytes` (existing)
* `MSGPACK = 3` → per-sketch `from_msgpack_bytes` (new)
* `PROTO_DELTA = 2`, `MSGPACK_DELTA = 4` → deferred, caller falls
through to §5.2 fallback
## Per-sketch decoders
Each of the five accumulators gains a `from_msgpack_bytes(buffer)`
constructor that wraps the sketch-core `deserialize_msgpack` already
present on the underlying type:
* `CountMinSketchAccumulator::from_msgpack_bytes`
* `CountSketchAccumulator::from_msgpack_bytes`
* `DatasketchesKLLAccumulator::from_msgpack_bytes` — and this is
notable: unlike the `_ENCODING_PROTO` path (which does lossy
statistical reconstruction via `update()` replay because
sketchlib's `KllState` proto elides level structure the Rust
backend types keep private), the msgpack path is a **bit-identical
round-trip** because sketch-core's `KllSketch` serializes its
full internal state to msgpack. This gives Go producers a way to
send KLL sketches losslessly to the Rust backend.
* `DDSketchAccumulator::from_msgpack_bytes`
* `HllSketchAccumulator::from_msgpack_bytes`
## Tests
Unit tests — 6 new, 2 per decoder (round-trip + garbage rejection)
on `CountSketchAccumulator`, `DDSketchAccumulator`, and
`HllSketchAccumulator`. CountMin + KLL msgpack paths already had
coverage through `deserialize_from_bytes_arroyo` tests.
E2e test — new `e2e_count_min_sketch_msgpack_modified_otlp_path` in
`tests/e2e_modified_otlp_sketch_path.rs`. Builds a real
`CountMinSketch` in sketch-core, serializes via msgpack, POSTs through
the OTLP HTTP receiver with `encoding = COUNT_MIN_SKETCH_ENCODING_MSGPACK`,
closes a window, and verifies the stored accumulator preserves per-key
counts. Covers the dispatcher path end-to-end for the msgpack branch;
the other four variants share the same branch so per-variant smoke
tests aren't needed for dispatcher coverage (per-variant msgpack
round-trips are covered by the unit tests).
## Drive-by: fix two clippy warnings inherited from main
`cargo clippy --all-targets -- -D warnings` on rust 1.91 flags two
pre-existing issues in the persistence PR #4 code path:
* `cache.rs` unnecessary `u64 as u64` casts
* `part.rs` test `&[snap.clone()]` → `std::slice::from_ref(&snap)`
Fixed in this PR since the CI clippy gate would otherwise block PR I.
## Validation
- cargo check --all-targets: clean
- cargo clippy --all-targets -- -D warnings: clean
- cargo fmt --check: clean
- cargo test -p sketch-core --lib: 54 passed
- cargo test -p query_engine_rust --lib: 480 passed (includes 6 new msgpack unit tests)
- cargo test -p query_engine_rust --test e2e_modified_otlp_sketch_path: 6 passed (5 existing PROTO variants + 1 new MSGPACK variant)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
to ProjectASAP/ASAPCollector
that referenced
this pull request
Apr 14, 2026
…sketches Adds `_ENCODING_MSGPACK = 3` and `_ENCODING_MSGPACK_DELTA = 4` to the five per-sketch encoding enums in the modified OTLP proto patch — matching the variants landed in ASAPQuery-backend PR #9 (ProjectASAP/ASAPQuery-backend#9) and implemented on the producer side in sketchlib-go PR #50 (ProjectASAP/sketchlib-go#50). Before this PR the three sides of the cross-language MessagePack contract had mismatched proto enum names: * ASAPQuery-backend (Rust consumer) — already had `_MSGPACK = 3` in its vendored metrics.proto, ready to decode msgpack-encoded sketch bytes from the hot path dispatcher * sketchlib-go (Go producer) — already has `wire/asapmsgpack.Marshal*` producing bytes the Rust consumer deserializes byte-for-byte * DataCollector — lacked the enum names entirely, so processors emitting msgpack bytes had to hard-code the integer tag `3` This patch aligns the DataCollector proto with the other two, closing the naming gap. Because protobuf enums are just integer tags, the change is forward-compatible: producers and consumers that already know about tag `3` at runtime continue to interoperate whether or not the name is declared on their side of the wire. So this commit is a pure readability / discoverability win; no generated code or runtime behavior changes until a follow-up wires a processor component to actually emit the new tag. Affected enums: * `DDSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `KLLSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 (reserved; sketchlib-go KLL doesn't share a byte-level backend with ASAPQuery-backend's KLL, so MSGPACK is not usable for KLL today — use PROTO instead) * `CountSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `CountMinSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `HLLSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 No changes to .proto message shapes, no reserved-range shifts, no impact on existing `_PROTO = 1` / `_PROTO_DELTA = 2` / `_DELTA = 2` producers and consumers. The generated `.pb.go` files and the opentelemetry-proto submodule checkout are NOT committed — they're regenerated from this patch file via `restore_otel_proto_patches.sh` at build time. Follow-ups: * `*processor` components in DataCollector that today emit only `_ENCODING_PROTO` can gain a config option to emit `_ENCODING_MSGPACK` instead, calling sketchlib-go's `wire/asapmsgpack.Marshal*` on the sketch state * `_DELTA` variants still deferred until sketch-core grows an `apply_delta` API — these enum entries exist only to reserve the tag numbers consistently with the Rust side Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
to ProjectASAP/ASAPCollector
that referenced
this pull request
Apr 14, 2026
…sketches (#154) Adds `_ENCODING_MSGPACK = 3` and `_ENCODING_MSGPACK_DELTA = 4` to the five per-sketch encoding enums in the modified OTLP proto patch — matching the variants landed in ASAPQuery-backend PR #9 (ProjectASAP/ASAPQuery-backend#9) and implemented on the producer side in sketchlib-go PR #50 (ProjectASAP/sketchlib-go#50). Before this PR the three sides of the cross-language MessagePack contract had mismatched proto enum names: * ASAPQuery-backend (Rust consumer) — already had `_MSGPACK = 3` in its vendored metrics.proto, ready to decode msgpack-encoded sketch bytes from the hot path dispatcher * sketchlib-go (Go producer) — already has `wire/asapmsgpack.Marshal*` producing bytes the Rust consumer deserializes byte-for-byte * DataCollector — lacked the enum names entirely, so processors emitting msgpack bytes had to hard-code the integer tag `3` This patch aligns the DataCollector proto with the other two, closing the naming gap. Because protobuf enums are just integer tags, the change is forward-compatible: producers and consumers that already know about tag `3` at runtime continue to interoperate whether or not the name is declared on their side of the wire. So this commit is a pure readability / discoverability win; no generated code or runtime behavior changes until a follow-up wires a processor component to actually emit the new tag. Affected enums: * `DDSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `KLLSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 (reserved; sketchlib-go KLL doesn't share a byte-level backend with ASAPQuery-backend's KLL, so MSGPACK is not usable for KLL today — use PROTO instead) * `CountSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `CountMinSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 * `HLLSketchEncoding` — adds MSGPACK=3, MSGPACK_DELTA=4 No changes to .proto message shapes, no reserved-range shifts, no impact on existing `_PROTO = 1` / `_PROTO_DELTA = 2` / `_DELTA = 2` producers and consumers. The generated `.pb.go` files and the opentelemetry-proto submodule checkout are NOT committed — they're regenerated from this patch file via `restore_otel_proto_patches.sh` at build time. Follow-ups: * `*processor` components in DataCollector that today emit only `_ENCODING_PROTO` can gain a config option to emit `_ENCODING_MSGPACK` instead, calling sketchlib-go's `wire/asapmsgpack.Marshal*` on the sketch state * `_DELTA` variants still deferred until sketch-core grows an `apply_delta` API — these enum entries exist only to reserve the tag numbers consistently with the Rust side Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The existing `hard_cap_back_pressure_blocks_inserts_until_flusher_drains` test asserts that at least one insert takes ≥3 ms wall-clock time when sealed memory hits the hard cap — the intuition being that a blocked insert must wait ≥1 flush tick for the condvar to fire. On CI (fast runners, small test data) the flush tick completes well under 1 ms, so the slow path runs and fully drains before the 3 ms threshold ever trips. Result: the assertion is flaky even though back-pressure is wired up correctly. Replace wall-clock timing with a dedicated monotonic counter. ## Changes * `FlusherShared` gains `back_pressure_wait_count: AtomicU64` * `FlusherHandle::wait_for_memory_under` increments the counter exactly once per call that observes `mem_counter >= cap` (i.e. every call that enters the blocking wait loop — the fast-path early return does not bump the counter) * `FlusherHandle::back_pressure_wait_count()` getter exposes it * `SimpleMapStorePerKey::back_pressure_wait_count()` proxies through to the flusher, returning 0 when persistence is disabled * The test now asserts `store.back_pressure_wait_count() > 0` after the 200-insert loop — a timing-independent signal ## Why a counter, not a timing threshold Tests that assert "this path is slower than X ms" assume the path is slow for a reason other than what it's actually measuring. Here, `wait_for_memory_under` blocks until the flusher drains, and the drain is fast, so the total wait can be sub-millisecond even though the mechanism is firing exactly as designed. Timing thresholds trade signal for noise on fast hardware. A counter sidesteps that tradeoff: it is 0 if the mechanism never fires and positive if it fires even once — exactly the semantic the test cares about. ## Validation Ran 5× back-to-back: all 5 passed (previously observed flake on CI). Full query_engine_rust --lib suite: 480 passed. Clippy + fmt clean. Drive-by fix: inherited from main (the persistence PR #4 merge), not introduced by this PR. Landing it here because the PR is already touching enough persistence-adjacent surface (clippy fixes) that CI needs to be unblocked on the full test gate, and a real fix is cleaner than a retry loop or `#[ignore]`. 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 `HotReloadStreamingConfig`, a thin `arc_swap::ArcSwap` wrapper,
plus `GET/POST /api/v1/streaming-config` HTTP endpoints so an external
controller (or a test harness) can push a new `StreamingConfig` at
runtime without restarting the query engine binary.
This is **phase 1** — the narrowest useful slice. The machinery exists
end-to-end (wire format, endpoint, swap), tests pin the contract, and
the controller has a working push target. Query execution and ingest
routing do NOT yet re-snapshot per request; those are phase 2 and are
documented on `HotReloadStreamingConfig` so the boundaries are clear.
## What's new
### `asap-query-engine/src/data_model/hot_reload_config.rs`
New module with `HotReloadStreamingConfig`, cloneable, built on
`Arc<ArcSwap<StreamingConfig>>`:
* `new(StreamingConfig)` / `from_arc(Arc<StreamingConfig>)` —
construct from initial config (two entry points so callers don't
have to double-allocate)
* `snapshot() -> Arc<StreamingConfig>` — cheap, lock-free read
* `swap(StreamingConfig) -> Arc<StreamingConfig>` — atomic replace,
returns the old `Arc` so callers can diff added/removed agg_ids
Four unit tests pin:
* initial snapshot reflects constructor
* swap replaces atomically (old handle unchanged, new reads fresh)
* clones share the underlying `ArcSwap` (so server-held handle and
test-held handle see the same swaps)
* concurrent reader never observes a torn state while a writer
swaps in a loop
### `GET / POST /api/v1/streaming-config` in `http.rs`
Two new routes on the existing control-plane surface, next to
`/api/v1/precompute`, `/api/v1/health`, `/api/v1/store/metrics`:
* **GET** — return the currently active config as JSON
(`{status, aggregation_count, aggregation_ids,
streaming_config}`). Used by tests and operators to verify a
push landed.
* **POST** — accept a YAML body matching the existing
`StreamingConfig::from_yaml_data` shape (the same format the
binary loads at startup), parse, validate, and atomically swap
via `HotReloadStreamingConfig::swap`. Returns `{status,
agg_ids_added, agg_ids_removed, new_aggregation_count}` so the
caller can confirm the transition without a round-trip to GET.
Logs a warning if any agg_ids were removed, since in-flight
precompute worker groups for those ids continue with their
construction-time config until they close naturally (phase 1
limitation).
Both endpoints return `503 Service Unavailable` when the server was
constructed without `HttpServer::with_hot_reload_config(...)` — a
clear failure mode for misconfigured deployments.
### `HttpServer::with_hot_reload_config(handle)` builder
New opt-in builder method on `HttpServer`. Without it, the endpoints
return 503 — so existing code paths (unit tests, legacy binaries) are
unaffected by this PR. `main.rs` calls it to attach the production
handle.
### `main.rs` wiring
Constructs a single `HotReloadStreamingConfig` from the startup
`Arc<StreamingConfig>` and passes it to `HttpServer::with_hot_reload_config`.
Other consumers (SimpleEngine, PrecomputeEngine, SimpleMapStore,
KafkaConsumer) still receive the startup `Arc<StreamingConfig>` and
will ignore swaps until phase 2.
## What's NOT hot-reloaded yet (documented on the module)
* **SimpleEngine query execution** — holds a startup snapshot,
doesn't re-snapshot per query. A query landing after a swap
sees the old config. Phase 2 will change this to re-snapshot on
query entry (one `Arc::clone` per query, negligible cost).
* **Ingest router / OTLP receiver** — routes by startup
`AggregationConfig` clones. New metric/labels mapped to new
agg_ids after a swap would not be routed until phase 2 wires
the router to re-snapshot per incoming message.
* **In-flight precompute worker `GroupState`** — each
`GroupState` caches `Arc<AggregationConfig>` at group creation.
Existing groups complete with their original config (correct
semantics, not a bug). New groups created after a swap use the
new config. This is the intended behavior for graceful add and
for param-tuning that only matters for future windows. The
only case where phase 1 is visibly incomplete is **removing**
an agg_id mid-window — the existing groups for that id keep
ingesting until their tumbling window closes. The POST handler
logs a warning when this happens.
## Drive-by fixes
Same two clippy-on-rust-1.91 issues in persistence code that
PR #9 also fixes:
* `persistence/cache.rs` — unnecessary `u64 as u64` cast
* `persistence/part.rs` — test `&[snap.clone()]` → `std::slice::from_ref(&snap)`
Inherited from main (persistence PR #4); fixed here so CI's clippy
gate passes on this branch too. These will be harmless duplicates if
PR #9 lands first.
## Tests
* **Unit tests** (4, in `hot_reload_config.rs`) — snapshot /
swap / clone-sharing / concurrent race.
* **HTTP integration tests** (3, in `http.rs::tests`):
- `test_streaming_config_hot_reload_round_trip`: starts a test
server with a hot-reload handle, POSTs a 2-agg YAML config,
asserts the POST response (agg_ids_added, new count), then
GETs and verifies the count + ids. Also asserts that the
externally-held `HotReloadStreamingConfig::snapshot()` sees
the swap, confirming the ArcSwap is shared.
- `test_streaming_config_hot_reload_missing_handle_503`: no
handle attached, both GET and POST return 503.
- `test_streaming_config_hot_reload_rejects_bad_yaml`: POST
with garbage YAML returns 400 + error message.
## Validation
* cargo check --all-targets: clean
* cargo clippy --all-targets -- -D warnings: clean
* cargo fmt --check: clean
* cargo test -p query_engine_rust --lib: 481 passed
## Follow-ups
* **Phase 2**: re-snapshot per query in `SimpleEngine`, per message
in the ingest router. Will likely change `SimpleEngine`'s field
from `Arc<StreamingConfig>` to `Arc<HotReloadStreamingConfig>`
and add a helper that snapshots at query entry.
* **Controller → backend channel**: once PR E's endpoint is merged,
the DataCollector controller can call it from its replanner
(`controller/src/replan.rs`) as a new transport alongside the
existing OpAMP push to agents. Tracked independently.
* **Graceful drain on removal**: PR E currently warns but does not
block config reloads that remove in-flight agg_ids. A stronger
guarantee would explicitly drain affected groups before
applying the swap.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
Apr 14, 2026
When SimpleEngine fails to match a query against any stored
aggregation — `StreamingConfig::find_compatible_aggregation` returns
`None` — the query plane now fires a fire-and-forget notification to
the DataCollector controller so the controller can generate a new
sketch plan and push it back via PR E's `/api/v1/streaming-config`
endpoint.
The query itself is **not** retried. It continues to fall through to
the existing §5.2 fallback (direct Prometheus read, SQL forwarding,
etc.) and returns whatever the fallback provides. Future queries
benefit once the new plan lands. Wiring the query retry into the plan
generation latency is explicitly out of scope — the fire-and-forget
model sidesteps the coordination problem and keeps the §5.2 fallback
as the correctness anchor.
## New module: `drivers/query/controller_client`
A thin transport-agnostic abstraction:
```rust
#[async_trait]
pub trait ControllerClient: Send + Sync {
async fn notify_capability_miss(
&self,
requirements: &QueryRequirements,
) -> Result<(), String>;
}
```
Plus:
* `HttpControllerClient` — HTTP impl with a 5-second timeout so a
slow or unreachable controller can't stall the spawned task.
POSTs a flat JSON payload projecting `QueryRequirements` into
serde-friendly fields (metric, statistics as `Debug` strings,
data_range_ms, grouping_labels as `Vec<String>`,
spatial_filter_normalized). Using a flat DTO avoids adding
`Serialize` derives to shared `asap_types` / `promql_utilities`
crates.
* `spawn_capability_miss_notify(&Option<Arc<dyn ControllerClient>>,
&QueryRequirements)` — fire-and-forget helper used by the query
hot path. Spawns via `tokio::spawn` so the query return path is
never blocked on network I/O. No-op when the option is `None`.
Errors are logged at WARN level — capability misses must never
fail the query.
## SimpleEngine wiring
* New field `controller_client: Option<Arc<dyn ControllerClient>>`
* New builder method `with_controller_client(client)` —
unset by default, so existing code paths (unit tests, callers
that don't configure a controller) are behaviorally unchanged.
* New helper `find_compatible_aggregation_with_miss_notify` —
wraps `streaming_config.find_compatible_aggregation` with the
fire-and-forget notification on `None`.
* Three capability-miss call sites updated to use the helper:
- SQL temporal/spatial (line ~1980)
- SQL spatio-temporal (line ~2130)
- PromQL (line ~2755)
## main.rs
New CLI flag:
```
--controller-endpoint <URL>
DataCollector controller endpoint for capability-miss
notifications. When set, SimpleEngine fires a fire-and-forget
POST on every miss. When unset (default), misses fall through
to the §5.2 fallback silently.
```
When set, main.rs constructs an `HttpControllerClient` and attaches
it via `SimpleEngine::with_controller_client(...)`.
## Tests
Five new unit tests in `controller_client.rs`, covering:
1. `payload_projects_all_requirement_fields` — DTO conversion from
`QueryRequirements` round-trips through JSON correctly
2. `spawn_helper_is_noop_when_client_is_none` — fire-and-forget
helper is safe to call with `None`
3. `spawn_helper_invokes_client_via_tokio_spawn` — with a mock
client, the spawned task actually fires and is observable via
a shared counter (exercises the fire-and-forget path end-to-end)
4. `http_client_reports_non_success_status` — real axum mock
server returning 500, asserts the client maps the status to
a formatted `Err`
5. `http_client_success_path_round_trips_payload` — real axum
mock server, client POSTs a capability-miss notification, the
server parses the JSON body and asserts the fields match the
source `QueryRequirements`
A public `MockControllerClient` is also exported from the test
module so follow-up tests in other crates (or other modules) can
compose a SimpleEngine with a fake client without needing an HTTP
mock.
## Drive-by
Same two clippy-on-rust-1.91 issues in persistence code that PR #9
and PR E also fix:
* `persistence/cache.rs` — unnecessary `u64 as u64` cast
* `persistence/part.rs` — test `&[snap.clone()]` →
`std::slice::from_ref(&snap)`
Inherited from main; fixed here so CI's clippy gate passes on this
branch too. Harmless duplicate when PR #9 lands first.
## Validation
* cargo check --all-targets: clean
* cargo clippy --all-targets -- -D warnings: clean
* cargo fmt --check: clean
* cargo test -p query_engine_rust --lib: 479 passed (up 5 from
main — the new unit tests)
## Follow-up
* **Query retry** — wire an optional retry after the controller
plan lands and the backend's StreamingConfig reflects the new
plan. Needs PR E phase 2 (hot-reload picked up by SimpleEngine
at query time) to be useful. Retry-once semantics, bounded wait,
with a clear fallback to the §5.2 path if the plan never lands.
* **Deduplication / rate-limiting** — the current fire-and-forget
spawns one notification per miss. Under heavy miss load on the
same `QueryRequirements`, the controller may be asked to plan
repeatedly. A simple LRU + "seen in last N seconds" filter in
`HttpControllerClient` would prevent this.
* **Retry telemetry** — track (miss → notify → plan landed) latency
across queries so we can actually tell if the loop closes fast
enough to matter.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
Apr 15, 2026
Adds `HotReloadStreamingConfig`, a thin `arc_swap::ArcSwap` wrapper,
plus `GET/POST /api/v1/streaming-config` HTTP endpoints so an external
controller (or a test harness) can push a new `StreamingConfig` at
runtime without restarting the query engine binary.
This is **phase 1** — the narrowest useful slice. The machinery exists
end-to-end (wire format, endpoint, swap), tests pin the contract, and
the controller has a working push target. Query execution and ingest
routing do NOT yet re-snapshot per request; those are phase 2 and are
documented on `HotReloadStreamingConfig` so the boundaries are clear.
## What's new
### `asap-query-engine/src/data_model/hot_reload_config.rs`
New module with `HotReloadStreamingConfig`, cloneable, built on
`Arc<ArcSwap<StreamingConfig>>`:
* `new(StreamingConfig)` / `from_arc(Arc<StreamingConfig>)` —
construct from initial config (two entry points so callers don't
have to double-allocate)
* `snapshot() -> Arc<StreamingConfig>` — cheap, lock-free read
* `swap(StreamingConfig) -> Arc<StreamingConfig>` — atomic replace,
returns the old `Arc` so callers can diff added/removed agg_ids
Four unit tests pin:
* initial snapshot reflects constructor
* swap replaces atomically (old handle unchanged, new reads fresh)
* clones share the underlying `ArcSwap` (so server-held handle and
test-held handle see the same swaps)
* concurrent reader never observes a torn state while a writer
swaps in a loop
### `GET / POST /api/v1/streaming-config` in `http.rs`
Two new routes on the existing control-plane surface, next to
`/api/v1/precompute`, `/api/v1/health`, `/api/v1/store/metrics`:
* **GET** — return the currently active config as JSON
(`{status, aggregation_count, aggregation_ids,
streaming_config}`). Used by tests and operators to verify a
push landed.
* **POST** — accept a YAML body matching the existing
`StreamingConfig::from_yaml_data` shape (the same format the
binary loads at startup), parse, validate, and atomically swap
via `HotReloadStreamingConfig::swap`. Returns `{status,
agg_ids_added, agg_ids_removed, new_aggregation_count}` so the
caller can confirm the transition without a round-trip to GET.
Logs a warning if any agg_ids were removed, since in-flight
precompute worker groups for those ids continue with their
construction-time config until they close naturally (phase 1
limitation).
Both endpoints return `503 Service Unavailable` when the server was
constructed without `HttpServer::with_hot_reload_config(...)` — a
clear failure mode for misconfigured deployments.
### `HttpServer::with_hot_reload_config(handle)` builder
New opt-in builder method on `HttpServer`. Without it, the endpoints
return 503 — so existing code paths (unit tests, legacy binaries) are
unaffected by this PR. `main.rs` calls it to attach the production
handle.
### `main.rs` wiring
Constructs a single `HotReloadStreamingConfig` from the startup
`Arc<StreamingConfig>` and passes it to `HttpServer::with_hot_reload_config`.
Other consumers (SimpleEngine, PrecomputeEngine, SimpleMapStore,
KafkaConsumer) still receive the startup `Arc<StreamingConfig>` and
will ignore swaps until phase 2.
## What's NOT hot-reloaded yet (documented on the module)
* **SimpleEngine query execution** — holds a startup snapshot,
doesn't re-snapshot per query. A query landing after a swap
sees the old config. Phase 2 will change this to re-snapshot on
query entry (one `Arc::clone` per query, negligible cost).
* **Ingest router / OTLP receiver** — routes by startup
`AggregationConfig` clones. New metric/labels mapped to new
agg_ids after a swap would not be routed until phase 2 wires
the router to re-snapshot per incoming message.
* **In-flight precompute worker `GroupState`** — each
`GroupState` caches `Arc<AggregationConfig>` at group creation.
Existing groups complete with their original config (correct
semantics, not a bug). New groups created after a swap use the
new config. This is the intended behavior for graceful add and
for param-tuning that only matters for future windows. The
only case where phase 1 is visibly incomplete is **removing**
an agg_id mid-window — the existing groups for that id keep
ingesting until their tumbling window closes. The POST handler
logs a warning when this happens.
## Drive-by fixes
Same two clippy-on-rust-1.91 issues in persistence code that
PR #9 also fixes:
* `persistence/cache.rs` — unnecessary `u64 as u64` cast
* `persistence/part.rs` — test `&[snap.clone()]` → `std::slice::from_ref(&snap)`
Inherited from main (persistence PR #4); fixed here so CI's clippy
gate passes on this branch too. These will be harmless duplicates if
PR #9 lands first.
## Tests
* **Unit tests** (4, in `hot_reload_config.rs`) — snapshot /
swap / clone-sharing / concurrent race.
* **HTTP integration tests** (3, in `http.rs::tests`):
- `test_streaming_config_hot_reload_round_trip`: starts a test
server with a hot-reload handle, POSTs a 2-agg YAML config,
asserts the POST response (agg_ids_added, new count), then
GETs and verifies the count + ids. Also asserts that the
externally-held `HotReloadStreamingConfig::snapshot()` sees
the swap, confirming the ArcSwap is shared.
- `test_streaming_config_hot_reload_missing_handle_503`: no
handle attached, both GET and POST return 503.
- `test_streaming_config_hot_reload_rejects_bad_yaml`: POST
with garbage YAML returns 400 + error message.
## Validation
* cargo check --all-targets: clean
* cargo clippy --all-targets -- -D warnings: clean
* cargo fmt --check: clean
* cargo test -p query_engine_rust --lib: 481 passed
## Follow-ups
* **Phase 2**: re-snapshot per query in `SimpleEngine`, per message
in the ingest router. Will likely change `SimpleEngine`'s field
from `Arc<StreamingConfig>` to `Arc<HotReloadStreamingConfig>`
and add a helper that snapshots at query entry.
* **Controller → backend channel**: once PR E's endpoint is merged,
the DataCollector controller can call it from its replanner
(`controller/src/replan.rs`) as a new transport alongside the
existing OpAMP push to agents. Tracked independently.
* **Graceful drain on removal**: PR E currently warns but does not
block config reloads that remove in-flight agg_ids. A stronger
guarantee would explicitly drain affected groups before
applying the swap.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
Apr 15, 2026
When SimpleEngine fails to match a query against any stored
aggregation — `StreamingConfig::find_compatible_aggregation` returns
`None` — the query plane now fires a fire-and-forget notification to
the DataCollector controller so the controller can generate a new
sketch plan and push it back via PR E's `/api/v1/streaming-config`
endpoint.
The query itself is **not** retried. It continues to fall through to
the existing §5.2 fallback (direct Prometheus read, SQL forwarding,
etc.) and returns whatever the fallback provides. Future queries
benefit once the new plan lands. Wiring the query retry into the plan
generation latency is explicitly out of scope — the fire-and-forget
model sidesteps the coordination problem and keeps the §5.2 fallback
as the correctness anchor.
## New module: `drivers/query/controller_client`
A thin transport-agnostic abstraction:
```rust
#[async_trait]
pub trait ControllerClient: Send + Sync {
async fn notify_capability_miss(
&self,
requirements: &QueryRequirements,
) -> Result<(), String>;
}
```
Plus:
* `HttpControllerClient` — HTTP impl with a 5-second timeout so a
slow or unreachable controller can't stall the spawned task.
POSTs a flat JSON payload projecting `QueryRequirements` into
serde-friendly fields (metric, statistics as `Debug` strings,
data_range_ms, grouping_labels as `Vec<String>`,
spatial_filter_normalized). Using a flat DTO avoids adding
`Serialize` derives to shared `asap_types` / `promql_utilities`
crates.
* `spawn_capability_miss_notify(&Option<Arc<dyn ControllerClient>>,
&QueryRequirements)` — fire-and-forget helper used by the query
hot path. Spawns via `tokio::spawn` so the query return path is
never blocked on network I/O. No-op when the option is `None`.
Errors are logged at WARN level — capability misses must never
fail the query.
## SimpleEngine wiring
* New field `controller_client: Option<Arc<dyn ControllerClient>>`
* New builder method `with_controller_client(client)` —
unset by default, so existing code paths (unit tests, callers
that don't configure a controller) are behaviorally unchanged.
* New helper `find_compatible_aggregation_with_miss_notify` —
wraps `streaming_config.find_compatible_aggregation` with the
fire-and-forget notification on `None`.
* Three capability-miss call sites updated to use the helper:
- SQL temporal/spatial (line ~1980)
- SQL spatio-temporal (line ~2130)
- PromQL (line ~2755)
## main.rs
New CLI flag:
```
--controller-endpoint <URL>
DataCollector controller endpoint for capability-miss
notifications. When set, SimpleEngine fires a fire-and-forget
POST on every miss. When unset (default), misses fall through
to the §5.2 fallback silently.
```
When set, main.rs constructs an `HttpControllerClient` and attaches
it via `SimpleEngine::with_controller_client(...)`.
## Tests
Five new unit tests in `controller_client.rs`, covering:
1. `payload_projects_all_requirement_fields` — DTO conversion from
`QueryRequirements` round-trips through JSON correctly
2. `spawn_helper_is_noop_when_client_is_none` — fire-and-forget
helper is safe to call with `None`
3. `spawn_helper_invokes_client_via_tokio_spawn` — with a mock
client, the spawned task actually fires and is observable via
a shared counter (exercises the fire-and-forget path end-to-end)
4. `http_client_reports_non_success_status` — real axum mock
server returning 500, asserts the client maps the status to
a formatted `Err`
5. `http_client_success_path_round_trips_payload` — real axum
mock server, client POSTs a capability-miss notification, the
server parses the JSON body and asserts the fields match the
source `QueryRequirements`
A public `MockControllerClient` is also exported from the test
module so follow-up tests in other crates (or other modules) can
compose a SimpleEngine with a fake client without needing an HTTP
mock.
## Drive-by
Same two clippy-on-rust-1.91 issues in persistence code that PR #9
and PR E also fix:
* `persistence/cache.rs` — unnecessary `u64 as u64` cast
* `persistence/part.rs` — test `&[snap.clone()]` →
`std::slice::from_ref(&snap)`
Inherited from main; fixed here so CI's clippy gate passes on this
branch too. Harmless duplicate when PR #9 lands first.
## Validation
* cargo check --all-targets: clean
* cargo clippy --all-targets -- -D warnings: clean
* cargo fmt --check: clean
* cargo test -p query_engine_rust --lib: 479 passed (up 5 from
main — the new unit tests)
## Follow-up
* **Query retry** — wire an optional retry after the controller
plan lands and the backend's StreamingConfig reflects the new
plan. Needs PR E phase 2 (hot-reload picked up by SimpleEngine
at query time) to be useful. Retry-once semantics, bounded wait,
with a clear fallback to the §5.2 path if the plan never lands.
* **Deduplication / rate-limiting** — the current fire-and-forget
spawns one notification per miss. Under heavy miss load on the
same `QueryRequirements`, the controller may be asked to plan
repeatedly. A simple LRU + "seen in last N seconds" filter in
`HttpControllerClient` would prevent this.
* **Retry telemetry** — track (miss → notify → plan landed) latency
across queries so we can actually tell if the loop closes fast
enough to matter.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This was referenced Apr 15, 2026
Merged
Merged
zzylol
added a commit
that referenced
this pull request
May 11, 2026
… + typed Predicate (#136) Additive lift of all 10 A-classified legacy variants into the canonical `intent_algebra::query_expr::QueryExpr`, plus the minimal typed `Predicate` IR for `Filter.pred` covering the 4 used-and-cleanly-shaped ScalarExpr variants. ## What landed ### Canonical `QueryExpr` grew from 5 → 15 variants Pre-existing (PR #130 / canonical): `Scan`, `Window`, `Aggregate`, `LetBinding`, `Ref`. New (this PR): `Filter`, `Project`, `Partition`, `Distinct`, `Merge`, `Join`, `SetOp`, `Sort`, `Limit`, `BinaryOp`. Naming convention: single-input variants use `child:` (matches existing canonical `Window`/`Aggregate`/`LetBinding`). Multi-input variants follow design.md §6 shapes exactly: `Merge { children }`, `Join { kind, pred, left, right }`, `SetOp { kind, all, left, right }`, `BinaryOp { op, lhs, rhs, vector_match }`. ### Supporting types lifted alongside `ColumnRef`, `PartitionKeys`, `BinaryOpKind`, `JoinKind`, `SetOpKind`, `SortKey`, `VectorMatch`/`VectorMatchKind`/`VectorGrouping`/`GroupSide`, `LiteralValue`, `ProjectItem`. All re-exported from `intent_algebra::mod.rs`. ### New typed `Predicate` ```rust pub enum Predicate { Column(ColumnRef), Literal(LiteralValue), BinaryOp { op: BinaryOpKind, lhs: Box<Predicate>, rhs: Box<Predicate> }, IsNull { expr: Box<Predicate>, negated: bool }, } ``` Plus `Predicate::from_legacy_scalar(&legacy::ScalarExpr) -> Result<Predicate, QueryExprError>`. Translates the 4 supported variants (Column / Literal / BinaryOp / IsNull); returns `QueryExprError::UnsupportedLegacyScalar(name)` for the 4 E-deferred (FunctionCall / ScalarSubquery / InList / Between). Per user decision, those stay in legacy until a real consumer demands the typed shape. Translation handles `LiteralValue::Duration → Int(nanos)` fold (canonical LiteralValue is deliberately narrower). ### Consumer migration deferred to subsequent batches **Zero consumer-site redirects in this PR** — and that's the right call. Every legacy-A-variant consumer (`query_parser/`, `physical/ {stage_split,planner,allocator}.rs`, `optimizer/engine.rs`, `legacy_lower.rs`) simultaneously matches A-variants AND C-variants (`Source`, `Aggregate`, `Window`, `SketchAgg`, `WindowedAgg`, `TopK`, `HistogramQuantile`, `PromQLSubquery`) AND constructs `ScalarExpr` predicates with E-deferred variants. Migrating them mid-pipeline would require lifting C-variants and E-variants in the same PR — explicit out-of-scope per the batch plan. The consumers migrate one C-batch at a time: - Batch 3a (PR #11): `AggFunc → AggIntent` - Batch 3b (PR #10): `TopK` split - Batch 3c (PR #8): `WindowedAgg` un-fusion - Batch 3d (PR #9): `SketchAgg → Aggregate@L3` - Batch 3e (PR #12): PromQL parser `histogram_quantile → Quantile` - Batch 13: retire `legacy_expr.rs` + `legacy_lower.rs` ### Catch-all arms added in 6 canonical-side consumers Files: `optimizer/cost/mod.rs`, `physical/colored_dag/{allocator,emitter}.rs`, `sketch_algebra/lower.rs`, `warm_tier_analysis.rs`. These previously matched canonical `QueryExpr` exhaustively over the 5 pre-existing variants; now they need conservative handling for the 10 new ones (Edge stage / Logical wrap / child-walk / zero-cost) with TODO markers for the follow-up reshape batches. ### Serde tag note Initially gave `ColumnRef`/`LiteralValue`/`PartitionKeys`/`Predicate` internally-tagged `#[serde(tag = "kind", rename_all = "snake_case")]` attrs, but serde rejects internally-tagged newtype variants containing primitives (e.g. `Named(String)`). Switched to externally-tagged `#[serde(rename_all = "snake_case")]`. Caught by the round-trip test. ## Build + test - `cargo build --release -p controller` — clean - `cargo build --release -p query_engine_rust` — clean - `cargo test -p controller --lib` — **666 passed** (was 655 after Batch 1; +11 from new typed-Predicate + schema tests) - `cargo test -p query_engine_rust --lib -- engines::warm_tier` — 13/13 ## Diff: 8 files, +794 / -8 Co-authored-by: zz_y <zz_y@node0.zz-y-304941.softmeasure-pg0.clemson.cloudlab.us> 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 I — adds MessagePack as a parallel encoding option alongside the protobuf-encoded sketchlib
*Statepath that PR B and PR C wired up for all five modified-OTLPMetric.datasketch variants. Gives sketchlib-go (Go producer) a practical alternative to prost-encoded sketchlib proto without touching the existing PROTO path.Cross-language producer ships in sketchlib-go #50 (merged) —
wire/asapmsgpack.Marshal*emits bytes that this PR'sfrom_msgpack_bytesdecoders deserialize byte-for-byte. Golden fixtures over there were generated from this Rust side'sserialize_msgpackoutput, so the cross-language contract is pinned on both sides.What's new
Proto
Adds
*_ENCODING_MSGPACK = 3and*_ENCODING_MSGPACK_DELTA = 4to each of the five per-sketch encoding enums in the vendoredmetrics.proto. Backward-compatible: producers emitting tags 0-2 see no change.Dispatcher (
drivers/ingest/otel.rs)decode_modified_otlp_sketch_bytesnow matches on the fullencodingvalue:PROTO = 1→ per-sketchfrom_sketchlib_proto_bytes(existing)MSGPACK = 3→ per-sketchfrom_msgpack_bytes(new)PROTO_DELTA = 2,MSGPACK_DELTA = 4→ deferred, falls through to §5.2 fallbackPer-sketch decoders
Each of the five accumulators gains a
from_msgpack_bytes(buffer)constructor wrapping the sketch-coredeserialize_msgpackalready present on the underlying type:CountMinSketchAccumulator::from_msgpack_bytesCountSketchAccumulator::from_msgpack_bytesDatasketchesKLLAccumulator::from_msgpack_bytes— notable: unlike the_ENCODING_PROTOpath (which does lossy statistical reconstruction because sketchlib'sKllStateproto elides level structure the Rust backend keeps private), the msgpack path is a bit-identical round-trip because sketch-core'sKllSketchserializes its full internal state. This gives Go producers a way to send KLL sketches losslessly to the Rust backend. (Note: sketchlib-go test(engine): end-to-end tests for schema-timeline query dispatch #50 does not yet implement the KLL wire format — see note below.)DDSketchAccumulator::from_msgpack_bytesHllSketchAccumulator::from_msgpack_bytesTests
CountSketchAccumulator,DDSketchAccumulator, andHllSketchAccumulator. CountMin + KLL already covered by existingdeserialize_from_bytes_arroyotests.e2e_count_min_sketch_msgpack_modified_otlp_path: builds a realCountMinSketchin sketch-core, serializes via msgpack, POSTs through OTLP HTTP withencoding = MSGPACK, closes a window, verifies stored counts preserved. Covers the dispatcher path end-to-end. Other four variants share the same branch; per-variant msgpack round-trips are covered by unit tests.sketch_core::*::serialize_msgpackoutput (CountSketch, DDSketch, HLL Regular, HLL Hip, CountMinSketch). If either side drifts, both CI gates fail.Drive-by: two clippy fixes inherited from main
cargo clippy -- -D warningson rust 1.91 flags two pre-existing issues in persistence PR #4 code:cache.rs— unnecessaryu64 as u64castspart.rs— test&[snap.clone()]→std::slice::from_ref(&snap)Fixed here since CI's clippy gate would otherwise block PR I.
Validation
cargo check --all-targets: cleancargo clippy --all-targets -- -D warnings: cleancargo fmt --check: cleancargo test -p sketch-core --lib: 54 passedcargo test -p query_engine_rust --lib: 480 passed (up 6 from main — the msgpack unit tests)cargo test -p query_engine_rust --test e2e_modified_otlp_sketch_path: 6 passed (5 existing PROTO variants + 1 new MSGPACK variant)Follow-ups outside this PR scope
*processorwire-up — a processor-side config option to emit_ENCODING_MSGPACKinstead of_ENCODING_PROTO, calling sketchlib-go'swire/asapmsgpack.Marshal*on the sketch state. Next in the producer stack.KllSketchdo not share a byte-level backend, so Go-side KLL msgpack is not reachable today. Producers emitting KLL should use_ENCODING_PROTO(the Rust side does lossy statistical reconstruction within KLL's rank-error bound). Tracked upstream._DELTAvariants) — needs a sketch-coreapply_deltaAPI design.🤖 Generated with Claude Code