Repository navigation
Conversation
Buildable snapshot of ASAP's OTAP Dataflow processor implementation, extracted from the private ASAPCollector monorepo for OTAP review. - asap-precompute-rs/: precompute + sketching runtime plus the OTAP integration layer (src/otap/) under the `otap` feature — Arrow codec (decode/encode/records/schema), 5-sketch sketch_type dispatch, and the AsapSketchesPlugin Tokio lifecycle (input / Wakeup-driven flush ticker / control-channel poll + graceful drain). `cargo build --features otap`; 135 tests pass; clippy + fmt clean. - otap-patch/: overlay for the upstream OTAP workspace — the local::Processor<OtapPdata> node + linkme OTAP_PROCESSOR_FACTORIES registration under urn:asap:processor:asap_sketches. asap_sketchlib dependency pinned to its public GitHub commit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Remove references to internal design docs, ADRs, PR/issue numbers, the private Go mirror and sibling integrations from source doc-comments and the READMEs; simplify the dependency note. Comments/prose only — no code changes; build, tests (135), clippy, and fmt all still pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Wire msgpack full + delta into the asap_sketches processor
Adds msgpack as a selectable wire format alongside proto, driven by
`cfg.encoding`:
- New `Encoding::MsgpackDelta` + `Encoding::is_msgpack()`; OTAP codec
string `MSGPACK_DELTA` (decode.rs).
- Each of the four delta-capable wrappers (DDSketch/HLL/CMS/CountSketch)
carries a `wire_encoding` (baked by the OTAP factory from
`cfg.encoding`); `snapshot` / `compute_delta_against` /
`delta_against_empty_base` / `merge` dispatch proto vs msgpack off it.
- New `Sketch::apply_delta_encoded(payload, encoding)` handles the
inbound tag (which may differ from the configured outbound format);
`window.rs::observe_envelope` routes the two new encodings through it.
- `precompute::serialize_series` stops hard-coding proto and maps
`cfg.encoding` to the emitted full/delta tag.
- `PluginConfig` gains `encoding` + `delta_transmission`; KLL + msgpack
is rejected at config time (`ConfigError::MsgpackUnsupported`) since
KLL is history-backed / full-only.
Depends on asap_sketchlib PR #67 (pinned via rev in Cargo.toml).
Tests: end-to-end two-window DDSketch (Msgpack full -> MsgpackDelta,
reconstructed) and a CMS full re-ingest across two nodes, plus config
guards. 140 tests pass; clippy + fmt clean.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Bump asap_sketchlib pin to cba7711 (CI-clean tip of #67)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Add KLL msgpack full support; bump sketchlib pin to ca7062f
KLL now emits/ingests the portable KLL msgpack full form (KLL is
full-only, no delta):
- KLLWrapper carries `wire_encoding`; msgpack snapshot builds
`KllSketchData { k, sketch_bytes }` (byte-identical to
`KllSketch::to_msgpack`) via the new sketchlib `MessagePackCodec for
KllSketchData`.
- Msgpack ingest replays the peer's retained `wire_items` (same approach
as the proto path) rather than `KLL::merge`, which loses weight on
empty-target merges. Cross-host KLL merge accuracy remains a known,
pre-existing limitation shared by both wire formats.
- Removes the KLL+msgpack config gate (`MsgpackUnsupported`); every
encoding now resolves for KLL.
Pins asap_sketchlib to ca7062f (adds MessagePackCodec for KllSketchData).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Add estimate output mode (transmit_sketch=false): quantile + cardinality gauges
Wires the previously-inert `transmit_sketch` / `quantiles` config into the
emit path (OTel/Prometheus convention):
- New `Sketch::estimate(quantiles, top_k) -> Vec<EstimatePoint>` (default
empty). DDSketch/KLL emit one point per quantile (with a `quantile`
label); HLL emits a single cardinality point. Frequency sketches
(CMS/CountSketch) stay on the empty default — enumerating top-k keys
needs a heavy-hitter tracker the wire-format wrappers don't carry.
- `SketchEnvelope` gains a `value: f64` field; `serialize_series` returns
`Vec<SketchEnvelope>` and, when `transmit_sketch = false`, emits one
typed scalar row per estimate point (encoding=Unspecified, empty
payload, value=estimate). `encode_batch` writes `value`; `decode` reads
it back.
- `PrecomputeConfig` gets a manual `Default` so `transmit_sketch` defaults
to `true` (historical behavior); estimate mode is opt-in. `PluginConfig`
gains `transmit_sketch` + `quantiles`, threaded through `resolve`.
Tests: DDSketch emits p50/p99 gauges within the alpha bound; HLL emits a
cardinality gauge within ~10%.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* docs: df-processor topology figures (edge-terminal vs edge→central)
Mermaid diagrams of the two asap_sketches deployment topologies and the
four cases (estimate vs sketch-on-wire; msgpack full/delta), plus the
zero-copy/serialization ledger and the one-processor-vs-two rationale.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Re-pin asap_sketchlib to merged main (0a2ac37, was #67 branch ca7062f)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Strips the NitroSketch-style geometric admission sampling that was wired into the count/quantile wrappers but never exposed via PluginConfig: - delete `src/sampling.rs` (GeometricSampler, wire_sample_p) - remove the `SampleSetter` trait (defined but never driven by the runtime) - DDSketch/CMS/CountSketch: drop `sample_p`/`sampler` fields, the `with_sample_p`/`set_sample_p`/`sample_p()` methods, the `update()` admission gate, the `impl SampleSetter`, and the `*_SAMPLE_SEED` consts - emitted envelopes now stamp `sample_p = 0.0` (the unsampled proto value) - drop the sampling-specific tests Behavior for all existing callers is unchanged (sampling was exact/off by default). The proto `SketchEnvelope.sample_p` wire field is kept (always 0.0). cargo test (137) + clippy + fmt clean. Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Design doc for the Schema / Dictionary / Record split in asap_sketches' sketch envelopes, drawn as an OTAP-style ER diagram with worked examples for agg_id vs series_id. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Remove the phantom batch-lifetime tier (window bounds belong to each sketch instance, not a shared batch), fix the hash_seed/hash_function table (KLL isn't hash-based; HLL genuinely needs one to merge correctly), and disambiguate "sketch instance" (per-window) from "series" (per-DICTIONARY-entry). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Contrasts OTAP's (currently unimplemented, closed-not-planned) multivariate metrics redundancy axis — cross-sectional, same instant/different metrics — against Schema/Dictionary/Record's temporal axis, at the two rates already established earlier in the doc: config-time (SCHEMA) and series-lifetime (DICTIONARY). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
The OTAP Strategy-B wire codec (encode_batch/decode_batch) emitted one
fully self-describing row per envelope: metric name, labels, sketch_type,
agg_id, schema_version, and encoding all repeated on every RECORD row,
every window, forever -- exactly the anti-pattern docs/data_model.md's
Schema/Dictionary/Record split exists to avoid. Nothing in the codec
implemented the doc's "sent once, referenced by index thereafter"
economics; the design existed only as a field-categorization scheme, not
an implemented protocol behavior.
Add otap::dictionary with SeriesDictionary (sender) / SeriesDictionaryDecoder
(receiver):
- SeriesDictionary::encode assigns stable series_ids and emits SCHEMA /
DICTIONARY / LABELS rows only the first time an agg_id / series is seen;
RECORD always carries just series_id + window bounds + envelope/value.
- SeriesDictionaryDecoder retains SCHEMA/DICTIONARY/LABELS state across a
continuous stream and reconstructs full SketchEnvelopes by joining RECORD
rows back against it, per the doc's own statefulness caveat -- an unknown
series_id/agg_id is a hard decode error, not a silent partial result.
- sketch_size is resolved from PrecomputeConfig::sketch_params via the new
Precompute::active_config() trait method.
- hash_seed/hash_function resolve HashSpec down to the one canonical seed
position (seed_list[canonical_seed_index]) rather than carrying
asap_sketchlib's full 20-entry seed table, per its own self-describing
wire-format doc; SeriesDictionaryDecoder::schema_for() exposes the
resolved value without fabricating a lossy reconstructed HashSpec.
StubPlugin and AsapSketchesPlugin now use this codec for tick/drain
(encode) and inbound-envelope (decode), each with persistent dictionary
state across calls. encode_batch/decode_batch/records::{flatten,lift}
are kept as-is -- they solve a different problem (disguising a payload as
OTAP-Metrics-shaped to transit a generic OTAP pipeline hop) that
docs/data_model.md was never about (its own first line scopes it to the
asap_sketches-to-asap_sketches hop).
Also adds examples/sketch_pipeline_demo.rs: a runnable three-stage demo
(sketch creation processor -> receive processor that merges + queries via
estimate mode -> Prometheus text exposition) showing the dictionary
economics directly in the output -- window 0 sends all four batches,
every later window sends RECORD only.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Stacked on feat/schema-dictionary-record-codec, which implemented the Schema/Dictionary/Record codec (SeriesDictionary/SeriesDictionaryDecoder) but only proved it end-to-end via direct calls in one process, in-memory -- not through the actual OTAP plugin lifecycle, and not across a real serialize/transmit/deserialize hop. This PR closes both gaps: otap::wire (new): Arrow-IPC serializes a SketchStreamBatch's four RecordBatches into a length-prefixed frame, plus async send_stream_batch/ recv_stream_batch over a TcpStream. Each sub-batch is its own self-contained IPC stream (schema + one record batch + EOS); recv_stream_batch distinguishes a clean EOF between frames from a truncated one mid-frame. AsapSketchesPlugin::start_from_envelopes (new): the receiver-role counterpart to the existing producer-role start(). Consumes Stream<Item = SketchStreamBatch> instead of Stream<Item = OtapMetricRecords>, decodes via a persistent SeriesDictionaryDecoder, and routes reconstructed envelopes through Precompute::observe_envelope (merge, never expand to samples) -- reusing the same ticker/control-task/graceful-drain machinery as the producer role via a new shared spawn_lifecycle helper. A receiver configured with transmit_sketch=false naturally re-emits query-mode (quantile) estimates instead of sketch bytes through its own emit channel, so a chain of AsapSketchesPlugins can compose without any new machinery. examples/sketch_producer_node.rs + sketch_receiver_node.rs (new): two separate binaries -- real AsapSketchesPlugin producer and receiver roles, connected over a real TCP socket via otap::wire, not the in-process mpsc channel sketch_pipeline_demo.rs uses. The producer feeds a real OTAP-shaped input stream (records::flatten + decode_batch, not a direct observe() call) and lets the plugin's actual Wakeup-style Tokio ticker close windows on its own wall-clock schedule. Verified running both together: producer emits 5 windows (window 0 carries SCHEMA+DICTIONARY+LABELS, windows 1-4 carry RECORD only), receiver receives and decodes all 5 over the socket, merges them, and prints a correct p99 gauge. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Bumps the pinned asap_sketchlib git rev from 0a2ac37 (2026-07-13) to 010457d (2026-08-22) — 48 commits, including ASAPv1 wire-format work (KLL payload + compaction seed, CMS/HLL hardening, Count-Min metadata layout), the CMS RegularPath i32::MAX clamp fix, and UnivMon-Q additions. Verified: `cargo build`/`cargo build --features otap`, `cargo test --features otap` (156/156 passing), `cargo clippy --all-targets --features otap -- -D warnings`, and `cargo fmt --check` are all clean against the new rev — no API breakage surfaced.
Fixes surfaced by code review of #5 (feat/schema-dictionary-record-codec). Files here originate on that branch — cherry-pick onto it if the fix should land in #5 itself rather than only on this stacked branch. - dictionary.rs: SeriesDictionary::identity_key joined agg_id/metric/ labels with unescaped '|'/'='/';' delimiters, so two genuinely different label sets could collide onto the same series_id (e.g. {"a": "1;b=2"} vs. {"a": "1", "b": "2"}). Now length-prefixes each segment, which makes every segment boundary unambiguous regardless of its contents. - dictionary.rs: SeriesDictionaryDecoder::ingest_dictionary unconditionally reset a series' labels to empty on every DICTIONARY row, so a duplicate/replayed row (or a sender that lost its own dictionary state and resent it) would silently wipe out previously- learned labels. Now preserves existing labels via the entry API. - dictionary.rs: resolve_hash_seed always read seed_list[canonical_seed_index], but asap_sketchlib's matrix-family sketches (CountSketch/CountMinSketch) always hash via HashSpec::matrix_seed() (seed_list[0]) on the packed hot path, regardless of canonical_seed_index. Now dispatches on sketch_type. - config.rs: sketch_size_string cast sketch params from f64 to u64 with `as`, silently truncating a misconfigured value (e.g. k = 0.5 truncating to "0") into a fabricated, wrong SCHEMA.sketch_size on the wire. Now validates (finite, non-negative, integral) and omits the field instead of fabricating a value, mirroring resolve_hash_seed's own "omit rather than fabricate" stance. Verified: cargo build/test (otap + default), cargo clippy -D warnings, cargo fmt --check all clean.
Fixes surfaced by code review of #6 (feat/otap-plugin-and-network-transport). - lifecycle.rs: spawn_input_task/spawn_envelope_input_task raced cancellation against reading the next stream item with an unbiased tokio::select!, so graceful shutdown could drop an already-ready batch instead of processing it (e.g. a producer enqueues its final batch then immediately signals shutdown). Now biased toward the input branch, so a ready batch is always consumed before the next loop iteration observes cancellation. - lifecycle.rs: decode/encode/precompute errors in the input, ticker, and drain tasks were caught and completely discarded (`if let Err(_e) = ...` / `let _ = ...`) with zero signal, even for errors the codec's own design says must be loud (OtapDecodeError::UnknownSeriesId/UnknownAggId). Added a dropped_batches counter (Arc<AtomicU64>, exposed via AsapSketchesPlugin::dropped_batches()) so this "drop the bad batch, keep the plugin alive" resilience policy is at least observable instead of fully silent, pending Phase D's real OTAP effect-handler error channel. - wire.rs: encode_stream_batch/send_stream_batch cast serialized lengths to u32 with `as`, so a sub-batch or frame body at/above 4 GiB would silently wrap to a too-small length prefix, desyncing every subsequent frame the decoder reads on that connection. Now a checked u32::try_from that errors instead of truncating. - wire.rs: recv_stream_batch allocated a buffer sized directly from an untrusted 4-byte length prefix with no cap, letting a hostile or corrupt peer force an ~4 GiB allocation attempt before a single content byte was validated. Added a 256 MiB MAX_FRAME_LEN sanity cap, checked before allocating. Verified: cargo build/test (otap + default), cargo clippy -D warnings, cargo fmt --check all clean.
The OTAP Dataflow workspace (rust/otap-dataflow/Cargo.toml, checked against otel-arrow@3e85c34) pins arrow-array/-schema/-ipc at "58.3". This crate was still on "53" — a 5-major-version skew. Since Phase D's OtapPdata <-> OtapMetricRecords binding lives in otap-patch/ and gets staged as a path-dependency member of the OTAP workspace itself (per otap-patch/all/Cargo.toml's own doc), the whole build resolves one shared Cargo.lock: RecordBatch built here and RecordBatch expected by OTAP's engine must be the literal same arrow-array version, or the binding is a type mismatch despite both nominally being "Arrow". Matching the same "58.3" requirement lets Cargo's resolver unify to one shared version automatically once staged, rather than requiring an exact patch pin. Verified: cargo build/test (otap + default, 156/156 passing), cargo clippy -D warnings, cargo fmt --check all clean against arrow 58.4.0 (the version "58.3" currently resolves to).
…er role)
Implements the "one seam left" the README calls out: putting a
self-describing sketch's serialized bytes onto the wire as a real OTAP
metric, and the reverse (ingesting real OTAP metrics into the
precompute runtime). This is the producer-role half of Phase D/E;
receiver-role (ingesting another asap_sketches node's SketchStreamBatch
as OtapPdata) is a separate, larger piece not covered here.
## What changed
- otap-patch/all/otap_bridge.rs (new): the actual binding.
- Encode (OtapMetricRecords -> OtapPdata): implements the upstream
`MetricsView` trait family (MetricsView/ResourceMetricsView/
ScopeMetricsView/MetricView/DataView/GaugeView/NumberDataPointView/
AttributeView/AnyValueView, plus uninhabited-enum placeholders for
Sum/Histogram/ExponentialHistogram/Summary/Exemplar, which this
binding never produces) as a thin adapter over OtapMetricRecords's
existing flat Arrow schema, then calls upstream's own
`encode_metrics_otap_batch` to build a real `OtapArrowRecords::Metrics`
— pushes all the low-level builder/dictionary-encoding correctness
onto already-tested upstream code instead of hand-rolling it.
- Decode (OtapPdata -> OtapMetricRecords): converts payload to
`OtapArrowRecords` (transparently handles OTLP-proto-bytes or
Arrow-record input via upstream's `TryIntoWithOptions`), walks it
via upstream's own `OtapMetricsView` reader, and rebuilds
OtapMetricRecords's flat 2-batch shape with plain arrow-array
builders (the same pattern records.rs's own `flatten`/`lift` use).
- Scope: only Gauge/Sum (NumberDataPoints) — the scalar shape
Observation/OtapMetricRecords already assume. Histogram/
ExponentialHistogram/Summary rows are skipped and counted
(DecodeOutcome::skipped_non_scalar), not silently dropped.
- otap-patch/all/mod.rs: AsapSketchesProcessor rewritten from a
pass-through into a real processor. Drives a bare `Precompute`
instance directly (obtained via
`AsapSketchesPlugin::from_plugin_config(...).precompute().clone()`,
discarding the plugin wrapper) rather than through
`AsapSketchesPlugin::start()`'s own Tokio-task/Stream lifecycle —
that lifecycle's emit channel now carries `SketchStreamBatch` (PR
#5/#6's dictionary-economics wire format for the asap_sketches ->
asap_sketches transport hop), not the OTAP-Metrics-shaped
`OtapMetricRecords` this adapter needs, so bridging it would need
reconciling formats rather than genuinely fitting. Precompute's own
observe/tick/drain are callback-style already, so OTAP's per-message
`process()` + `effect_handler.start_periodic_timer` (emitting
NodeControlMsg::TimerTick) hosts them directly with no bridging
machinery needed. Also wires NodeControlMsg::Config ->
Precompute::update_config (previously a no-op) — live
reconfiguration for this one processor instance.
- otap-patch/all/Cargo.toml, otap-patch/all/mod.rs,
otap-patch/all/otap_bridge.rs: renamed all OTAP Dataflow crate
imports `otap_df_*` -> `otel_arrow_dfe_*`, matching upstream's very
recent (unreleased as of the pinned commit) package rename,
otel-arrow issue #1848 / .chloggen/otel-arrow-dfe-crate-prefix.yaml.
Revert to `otap_df_*` throughout if the actual internal build pin
predates that rename.
- asap-precompute-rs/{Cargo.toml,Cargo.lock} (prior commit, prerequisite
for this one): Arrow 53 -> 58.3, matching the OTAP workspace's own
pin so RecordBatch is the literal same type across this binding.
## Verification status — read carefully
`otap-patch/` has no standalone build in this repo (confirmed: no
OTAP Dataflow workspace checkout, no lockfile pinning one — see the
repo README's existing "No — depends on the OTAP workspace crates"
note on this directory, which predates this change). Every upstream
type/function signature referenced in otap_bridge.rs and mod.rs was
read directly from a fresh clone of open-telemetry/otel-arrow at
commit 3e85c3460361446ebfce99e9f35fffd2dd5ab740 (2026-08-24) — not
compiled against. The Arrow-only portions (RecordBatch construction,
typed-column accessors) were extracted and compile+test verified in
isolation against the real arrow-array 58.4.0 crate (round-trip test
passing) since that part doesn't depend on the OTAP-specific crates.
The OTAP-specific view-trait implementation (the biggest, riskiest
part) could not be compiled here at all.
Expect a real build pass against the actual pinned OTAP Dataflow
commit to surface mismatches — this is a first cut at the binding,
not a verified-working one. asap-precompute-rs itself (the standalone,
buildable part) is unaffected except for the Arrow version bump, and
remains fully verified: cargo build/test (156/156 passing), clippy
-D warnings, fmt --check all clean.
Per review feedback: the previous "Scope" doc framed decode handling as type-based (Gauge/Sum handled, Histogram/ExpHistogram/Summary skipped), which obscured the actual, more important distinction — content-based, not type-based: - A data point carrying `_asap_envelope` (sketch shipped as binary inside an OTAP metric, from this module's own encode output or any other asap_sketches node) needs no special casing in this module at all: it round-trips into OtapMetricRecords like any other attribute, and downstream `decode_batch` already tags it ObservationValueKind::Envelope; `Precompute::observe` already routes that internally to `observe_envelope` (merge as pre-aggregated sketch, never expanded to samples) — confirmed by reading PrecomputeImpl::observe's body (precompute.rs). - A genuine (non-envelope) OTLP metric gets scalar-sample handling for Gauge/Sum; Histogram/ExponentialHistogram/Summary are still skipped and counted (DecodeOutcome::skipped_non_scalar) — confirmed keeping this as-is rather than inventing a lossy bucket/quantile expansion policy. Rewrote otap_bridge.rs's module doc "Scope" section and added a comment at the actual `precompute.observe(obs)` call site in mod.rs (previously the dual-path behavior was implicit / only discoverable by reading precompute.rs directly). No functional change — the routing described was already correct; only the documentation was misleading. Also re-confirmed against upstream: otel-arrow's `main` is still at commit 3e85c3460361446ebfce99e9f35fffd2dd5ab740 (re-fetched, no new commits since the prior otap_bridge.rs commit), so the otel_arrow_dfe_* naming and every signature referenced there remains current as of this commit.
…codec otap: implement Schema/Dictionary/Record codec from data_model.md
…ransport otap: real plugin integration + Arrow IPC network transport
…pipeline E2E (#9) * fix(otap): build/lint/test-verify otap_bridge.rs + mod.rs against a real OTAP workspace Per request: actually get an OTAP Dataflow checkout and compile/test against it, rather than leaving otap_bridge.rs's binding as "read from source, never compiled." Staged both files into a real clone of open-telemetry/otel-arrow @ 3e85c3460361446ebfce99e9f35fffd2dd5ab740 (2026-08-24) as a `crates/asap-sketches-registry` workspace member (path-depping back to asap-precompute-rs exactly as Cargo.toml's own doc describes) and ran the real cargo build/clippy/fmt/test there. ## Bugs the real compiler caught (none of these were guessable from ## source reading alone) - `ProcessorFactory::create`'s function-pointer type gained a fifth parameter, `capabilities: &capability::registry::Capabilities` ("per-node, one-shot view of extension capabilities... factories that don't depend on any extension can ignore the parameter") — not present in whatever version this adapter was originally read against. Added as an unused parameter to create_asap_sketches_processor. - Two borrow-checker errors in the OtapPdata decode path: `if let Some(gauge) = data.as_gauge() { gauge.data_points().collect() }` doesn't compile because the collected Vec's items borrow from `gauge`, which is dropped at the end of the `if let` arm. Fixed by factoring the per-data-point body into a `DecodeAccumulator` struct with a generic `push_data_point<D: NumberDataPointView>` method, called inline from each of the Gauge/Sum branches instead of trying to unify their different concrete NumberDataPointView types into one Vec first. - One unused import (`TryFromWithOptions`) clippy caught immediately. ## What's now genuinely verified (not just "compiles") Added 3 new tests to otap_bridge.rs exercising the real OtapArrowRecords::Metrics / OtapPdata / encode_metrics_otap_batch / OtapMetricsView machinery end to end: - encode_then_decode_round_trips_a_scalar_metric - encode_then_decode_round_trips_a_sketch_envelope_carried_as_a_metric_attribute (the "self-describing sketch binary inside an OTAP metric" case -- an `_asap_envelope` Bytes attribute survives the round trip byte-for-byte) - decode_returns_none_records_for_zero_rows All 10 tests in the staged crate pass (7 pre-existing config-shape tests + 3 new), `cargo clippy -D warnings` and `cargo fmt --check` both clean, against the real workspace. ## Not changed Applied `cargo fmt`'s reformatting (this workspace's rustfmt config differs slightly from what the file was originally written against -- import grouping mainly) back onto the canonical otap-patch/ copies. Updated mod.rs's, otap_bridge.rs's, and README.md's doc comments from "unverified" to reflect what's actually confirmed now, and precisely what "verified" means here: staged into a *separate*, temporary checkout of the real workspace, not something this repo's own build wires up -- otap-patch/ itself still has no standalone build in this repo. Added .gitignore entry for /otel-arrow/ (the temporary checkout used for this verification, not committed). * feat(otap): unify wire lane and metric lane onto one real OtapPdata encoding Per request: merge the two paths into one, with the wire lane genuinely supporting the same dictionary/reuse economics. New otap_wire.rs carries a real OtapArrowRecords::Metrics (built via otap_bridge::otap_metric_records_to_pdata -- the exact same encoding the generic-pipeline path already uses) directly over a persistent TCP connection, instead of ASAP's own SCHEMA/DICTIONARY/RECORD SketchStreamBatch protocol (asap_precompute_rs::otap::{wire,dictionary}). One encoding, used identically whether a producer/receiver pair is directly connected or routed through other OTAP pipeline components. ## What changed - otap-patch/all/otap_wire.rs (new): OtapWireWriter/OtapWireReader, reusing the persistent-per-connection Arrow IPC design asap_precompute_rs::otap::wire::{WireWriter,WireReader} already validates (each payload type's Schema message sent once per connection, not once per window) -- generalized from SketchStreamBatch's fixed 4 roles to however many ArrowPayloadTypes a given OtapArrowRecords::Metrics actually populates (a real OtapPdata can carry up to 19 different payload types; ASAP's encode path only ever populates a handful, so roles are tracked in a BTreeMap keyed by ArrowPayloadType rather than 4 named fields). - Duplicated (not shared) with asap-precompute-rs's design on purpose: asap-precompute-rs deliberately has no dependency on the OTAP Dataflow crates this needs, and pulling one in would break that crate's "builds standalone" property -- see the module's own doc. - otap-patch/all/Cargo.toml: added arrow-ipc, tokio deps (both already pinned at the OTAP workspace level, so no new version to reconcile). ## Verification Same workflow as the previous commit on this PR: staged into a real open-telemetry/otel-arrow checkout (3e85c3460361446ebfce99e9f35fffd2dd5ab740, 2026-08-24) as the asap-sketches-registry workspace member. Two new tests genuinely round-trip a real OtapPdata over an actual TCP loopback socket: - round_trips_a_real_otap_pdata_over_a_tcp_loopback_socket - round_trips_multiple_windows_over_one_persistent_connection (two different metric values sent over one persistent connection, both decoded correctly on the far end -- the scenario that actually proves "one path" rather than just "compiles") All 12 tests in the staged crate pass (10 from the previous commit + these 2), cargo clippy -D warnings and cargo fmt --check both clean. ## Not done here otap_wire.rs has no consumer inside this crate yet -- it's a complete, tested transport module, not baked into AsapSketchesProcessor's runtime behavior (documented honestly via #![allow(dead_code)] with an explanation, not silently hidden). Wiring a config-driven choice of transport into AsapSketchesProcessor itself (a peer_addr option, connection lifecycle, reconnect behavior) is real follow-up work -- see the module's own "Not wired into AsapSketchesProcessor yet" doc section for the two shapes that follow-up could take. * feat(otap): wire the wire-lane transport into real running processor/receiver nodes Both otap_wire.rs's transport (OtapWireWriter/OtapWireReader) and the wire lane's receive side had no consumer inside the crate — this closes that: real OTAP nodes now actually drive them at runtime, not just tests. - New otap_receiver.rs: AsapSketchesReceiver, a real local::Receiver<OtapPdata> (not local::Processor — that trait is purely reactive and can't independently accept a TCP connection). Listens on a configured address, decodes each connection with OtapWireReader, pushes into the pipeline via effect_handler.send_message. Registered under urn:asap:receiver:asap_sketches_wire via OTAP_RECEIVER_FACTORIES. - AsapSketchesProcessor gains an optional peer_addr config field: emit_envelopes now forwards over a lazily-connected, persistent OtapWireWriter connection to that peer when set, falling back to the existing generic pipeline hop otherwise. A send failure drops the window and resets the connection so the next window reconnects. - otap_wire.rs's #![allow(dead_code)] is gone — both OtapWireWriter and OtapWireReader now have real, non-test callers. - New end-to-end test: the real AsapSketchesReceiver::start() bound to a real loopback TCP socket, fed by a real OtapWireWriter::send from a connected client, decoded and pushed through the actual pipeline machinery (OTAP's own TestRuntime harness) — not a mock. Verified against the same staged open-telemetry/otel-arrow checkout (3e85c3460361446ebfce99e9f35fffd2dd5ab740, 2026-08-24): build, clippy -D warnings, fmt --check, and test all clean; 16/16 tests passing (up from 12/12), including the new end-to-end receiver test. asap-precompute-rs's own suite: 156/156, unaffected. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * revert(otap): remove custom TCP wire lane — OTAP's own Arrow encoding already gives us dictionary economics The direct-TCP wire lane added in the previous commit (otap_wire.rs + otap_receiver.rs's AsapSketchesReceiver, peer_addr config) is gone. There is now exactly one transport: effect_handler.send_message_with_source_node, i.e. whatever pipeline this node is already wired into. Why: the wire lane existed to give sketch traffic real dictionary/ schema-reuse economics (asap_precompute_rs::otap::dictionary's SCHEMA/ DICTIONARY/RECORD tiering, from PR #5/#6) that riding a generic OTLP metric didn't have. But that's solving a problem OTAP's real Arrow encoding already solves — otap_bridge's encode_metrics_otap_batch dictionary-encodes the metric name and every string-valued attribute key/value by construction. Verified against the real staged workspace: === payload_type UnivariateMetrics === name : Dictionary(UInt8, Utf8) === payload_type NumberDpAttrs === parent_id : Dictionary(UInt8, UInt32) key : Dictionary(UInt8, Utf8) str : Dictionary(UInt16, Utf8) That's the same "send the dictionary once, reference it after that" shape SeriesDictionary was reinventing at the application layer, done instead at the columnar/Arrow-IPC level — with no dictionary state for this adapter to track, no second wire protocol, and none of the "must be one ordered, single-consumer stream or the series_id reference dangles" correctness constraint a hand-rolled scheme has. Added a permanent regression test guarding this fact (otap_bridge::tests::real_otap_encoding_dictionary_encodes_metric_name_and_string_attributes) so a silent upstream schema change surfaces loudly rather than being rediscovered from scratch. asap_precompute_rs::otap::dictionary (SeriesDictionary / SketchStreamBatch) stays in the tree, tested, and still backs the legacy asap_precompute_rs::otap::wire example binaries — it's just not part of this adapter's path. Re-verified against the same staged open-telemetry/otel-arrow checkout (3e85c3460361446ebfce99e9f35fffd2dd5ab740, 2026-08-24): build, clippy -D warnings, fmt --check, and test all clean; 11/11 tests passing. asap-precompute-rs's own suite: 156/156, unaffected. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * refactor(otap)!: merge otap-patch into asap-precompute-rs as one crate, real OtapPdata everywhere Full merge, per explicit direction: no more otap-patch/ overlay requiring manual staging into a checked-out OTAP Dataflow workspace to compile. asap-precompute-rs is now the ONE crate, and depends directly on the real otel-arrow-dfe-* crates (plain git dependency, pinned to 3e85c3460361446ebfce99e9f35fffd2dd5ab740, same pattern already used for asap_sketchlib) behind a new `otap-engine` Cargo feature. `cargo build --features otap-engine` just works — no staging script, no temporary checkout, no copying files by hand. New modules (moved + rewritten from otap-patch/all/{mod.rs,otap_bridge.rs}): - otap::processor — AsapSketchesProcessor, the real local::Processor<OtapPdata> node, linkme-registered under urn:asap:processor:asap_sketches. - otap::codec (renamed from the working name `bridge` — it's not bridging two representations anymore, see below) — the real SketchEnvelope <-> OtapPdata binding. The bigger change is what codec.rs does differently from the old otap_bridge.rs it replaces: it builds/reads a real OtapPdata *directly* from/to &[SketchEnvelope], skipping the intermediate flat RecordBatch (encode_batch) and OtapMetricRecords two-batch family (lift) hops entirely. Those three representations for one job only existed because of speculative multi-adapter generality (Telegraf/Vector) that never materialized in this repo, which only ever ships to OTAP -- so codec.rs implements OTAP's own MetricsView trait family straight over envelope slices instead. encode_batch/decode_batch/records::{flatten,lift} still exist, still tested, and still back the legacy SeriesDictionary/otap::wire transport and its standalone example binaries -- they're just not part of the otap-engine path anymore. A real bug found and fixed along the way: encode_batch unconditionally sets `_asap_envelope` even for estimate-mode envelopes (empty payload), and arrow_array::BinaryArray treats `Some(&[])` as present- not-null -- confirmed empirically -- so an estimate-mode gauge would misroute through the envelope decode path with an empty payload instead of the scalar path. codec.rs's direct encoder only attaches _asap_envelope (and its sibling attributes) when the payload is actually non-empty, with a defensive filter on decode too. New test: estimate_mode_envelope_round_trips_as_a_scalar_not_an_empty_envelope. The legacy encode_batch/decode_batch pair keeps the original behavior untouched -- fixing it there was out of scope for this rewrite. Also moved plugins/asap_sketches/{README.md,sample.toml} to asap-precompute-rs/plugins/asap_sketches/ (content updated to match: no more Phase D/"deliberately deferred" language, since it's done), dropped the empty src/mod.rs placeholder it predated, and rewrote the crate-root and repo-root README/module docs throughout to describe the one-crate reality instead of the old Layer A/otap-patch split. Verified via this repo's own `cargo build/test/clippy/fmt` at all three feature levels (default, otap, otap-engine) -- no staging into any external checkout: 100/113/169 tests pass respectively (default/ otap/otap-engine), clippy -D warnings clean, fmt clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(otap): apply runtime processor config safely * test(otap): run sketch processor in real YAML pipeline --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Extend the real-RuntimePipeline E2E to three independently configured
`urn:asap:processor:asap_sketches` nodes sharing one stable URN:
- create (transmit_sketch=true) scalar gauges -> self-describing DDSketch
- merge (transmit_sketch=true) merges the envelope as sketch state,
never re-expanding it to scalar samples
- estimate (transmit_sketch=false, quantiles=[0.5,0.99]) -> scalar p50/p99
Assertions are made at the exporter boundary: p50/p99 relative accuracy,
genuinely-scalar output shape, output metric name, and `route` label
preserved across all three processor hops plus three timer flushes.
Thread the estimate- and wire-control knobs through
`AsapSketchesUserConfig` -> `PluginConfig` (transmit_sketch, quantiles,
encoding, delta_transmission, default_key, omit_resource_attrs,
global_aggregation, emit_window_stats) and account for the
series-identity / factory-affecting ones in `requires_precompute_rebuild`
so a runtime `NodeControlMsg::Config` push rebuilds the Precompute when
it must. Document them in sample.toml and the plugin README.
Add docs/demo-guide.md: a presentation-ready walkthrough of the demo.
Verified: cargo test --features otap-engine (all pass), clippy
--all-targets --features otap-engine -D warnings, cargo fmt --check.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JTn742jM8FKwqdTDXGSz3p
Replace the plain-text topology sketch in demo-guide.md with a mermaid flowchart matching df-processor-topology.md's style: the five nodes in one RuntimePipeline subgraph, each processor labelled with its config, edges labelled with the payload that crosses them. Drop the now-duplicated "test nodes vs. real implementations" sentence. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JTn742jM8FKwqdTDXGSz3p
1/6 feat(otap): add create-merge-estimate pipeline E2E
2/6 feat(otap): map sketches onto native metrics joins
3/6 feat(demo): add runnable two-source sketch pipeline
4/6 feat(otap): carry sketches as ASAPv1 OTLP summaries
5/6 refactor(otap): remove obsolete private sketch transport
6/6 refactor(otap): isolate processors and use official debugging
perf(otap): add nightly ASAP sketch benchmark
* bench: scale traffic and report per-component resources * Fix benchmark timing and stage CPU accounting * bench: measure persistent OTLP streaming over real network links * Pin each streaming component to one CPU * Use OTAP traffic generator for scaling benchmark * Define sustainable capacity speedup evaluation * Remove redundant pdata round trip from branch creation * Scale observation time with aggregation window * Allow bounded high-rate generator skew * Optimize fixed-k KLL ingestion path * Document generator provisioning for capacity tests * Allow per-scenario generator provisioning * Aggregate per-scenario maximum downstream capacity * Automate downstream maximum capacity sweeps * Scale KLL workload generation through 32 cores * Refine high-load capacity boundary detection * Report downstream plateau throughput speedup * Remove validating backend from capacity budget * Update CAPACITY_DESIGN.md * Remove benchmark container memory caps * Support benchmark smoke runs on small CI hosts * Measure processor output backpressure
The demo now uses the streaming benchmark topology: branch A/B, merge and estimate are separate processes, each pinned to one CPU and connected by OTLP/HTTP over TCP. `run-streaming.py --demo` runs the exact and KLL pipelines side by side until Ctrl+C (or SIGTERM) and serves per-process throughput, CPU, egress, RSS and validated answers to Prometheus; a new Grafana dashboard shows them. Estimate can also merge the last N windows for rolling p50/p99 (`--rolling-windows`, default off for capacity runs). Replaces the single-process asap-kll-stream-demo. The Grafana stack uses host networking bound to 127.0.0.1. Docs include sweep results from a 56-core x86 server.
The pinned rev predates the CMSHeap ASAPv1 envelope (asap_sketchlib#115). On main, DdSketchState also carries negative and zero stores, so the DDSketch wrapper now uses sketchlib's own to_proto/from_proto mapping.
Count-Min sketch with a top-k heap: full ASAPv1 snapshots (kind 0x03 0x00), merge that re-ranks both heaps on merged counters, and FrequencySketch top_k. Envelopes name it CountMinSketchWithHeap.
The carrier validated every sketch.envelope as KLL. CountMinSketchWithHeap envelopes are now validated as CMSHeap; other types keep the KLL check.
topk-exact ships per-key count maps and sums them; cmsheap ships one ASAPv1 CMSHeap per source and merges sketches. Keys follow Zipf(1.1) over 2^20 ranks. The backend validates every window: exact must match, the sketch must reach recall@10 >= 0.9. The demo exports top-k, recall and count error next to per-process throughput, CPU and egress.
zzylol
force-pushed
the
feat/cmsheap-topk-demo
branch
from
October 13, 2026 00:20
d856571 to
fb6fa0f
Compare
zzylol
force-pushed
the
feat/kll-prometheus-grafana
branch
2 times, most recently
from
October 19, 2026 17:33
a0466df to
e2c3e32
Compare
zzylol
force-pushed
the
feat/cmsheap-topk-demo
branch
from
October 19, 2026 17:33
fb6fa0f to
57720fc
Compare
zzylol
force-pushed
the
feat/kll-prometheus-grafana
branch
from
October 19, 2026 18:04
e2c3e32 to
a0466df
Compare
Contributor
Author
|
Replaced by #39. The history rewrite closed this PR and GitHub will not reopen it; same branch, same commits. |
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
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.
Stacked on #38. Adds a Count-Min sketch with heavy-hitter heap (
asap_sketchlib::CMSHeap) and a CMS + heap vs exact top-k demo and benchmark, using the same multi-process pinned topology as the KLL demo in #20.Before → after
CountSketch::top_kreturns nothing, and the OTAP carrier accepted only KLL ASAPv1 payloads.CMSHeapWrapperwrapsasap_sketchlib::CMSHeap. It sends ASAPv1 envelopes (kind0x03 0x00,CountMinSketchWithHeap) and implementsFrequencySketch::top_k. Merging re-ranks both heaps on the merged counters. The OTAP carrier now accepts and validates these payloads.topk-exactandcmsheap, run on keyed traffic: Zipf(1.1) over 2^20 keys.topk-exact: each branch counts keys in a hash map and ships every (key, count) entry; merge sums them.cmsheap: each branch ships one sketch per window (4 x 2048 counters, 10-slot heap); merge merges sketches.run-demo.sh --scenarios topk-exact cmsheapserves a new Grafana dashboard: both pipelines' current top 10 side by side, recall@10, count error, throughput, and per-process CPU, egress and memory.Run
demos/kll-grafana/run-demo.sh --scenarios topk-exact cmsheap # Ctrl+C stopsOpen http://localhost:3000/d/asap-topk-stream.
56-core x86 server, 50,000 signals/s per source. Both pipelines keep up. CMS + heap uses 1.7 processor CPU-seconds per million signals; exact uses 13.0. CMS + heap returns the same ten keys in the same order, with counts within 0.6%.
Throughput: CMS + heap vs exact top-k (vs raw forwarding)
65,536-key windows, 3 runs per point, one dedicated CPU per processor, a 56-core x86 server. Median backend-validated signals/s, both sources:
Largest sustainable offered rate per source (CAPACITY_DESIGN criteria): 50,000 for exact top-k, 100,000 for raw, and at least 400,000 (the highest tested) for CMS + heap. Above that, exact top-k plateaus near 130,000 signals/s.
Processor CPU and egress summed over branches, merge and estimate, at the same load:
Accuracy
The backend checked every window of every run:
Library pin
This PR moves the
asap_sketchlibpin from010457deto3fc3442c(current main), because the CMSHeap ASAPv1 payload exists only there (asap_sketchlib#115 and follow-ups). On main,DdSketchStatealso gained negative and zero stores. The DDSketch wrapper now uses sketchlib's ownto_proto/from_protoinstead of a local field mapping. All existing sketch, codec and runtime tests pass on the new pin.Validation
All tests pass, along with fmt and Clippy. The demo ran end to end on a 56-core x86 server.