diff --git a/README.md b/README.md index d87fc0b..b66512d 100644 --- a/README.md +++ b/README.md @@ -22,9 +22,11 @@ OTAP Dataflow's Arrow-native streaming model as a single unified The piece OTAP's runtime actually sees: - `AsapSketchesProcessor` implements - `otap_df_engine::local::processor::Processor` — an + `otel_arrow_dfe_engine::local::processor::Processor` — an `async fn process(msg: Message, effect_handler)` handling - `Message::PData` and `NodeControlMsg::{Wakeup, Config, Shutdown, …}`. + `Message::PData` and `NodeControlMsg::{TimerTick, Config, Shutdown, …}`. + (Crate prefix `otel-arrow-dfe-*`, not the older `otap-df-*` — upstream + renamed these very recently, otel-arrow issue #1848.) - Registered via `#[distributed_slice(OTAP_PROCESSOR_FACTORIES)]` under the URN `urn:asap:processor:asap_sketches`, with a `ProcessorFactory` (`create` + `validate_config` + `WiringContract::UNRESTRICTED`). @@ -32,10 +34,20 @@ The piece OTAP's runtime actually sees: `--validate-and-exit` time (see [`otap-patch/plugins/asap_sketches/sample.toml`](./otap-patch/plugins/asap_sketches/sample.toml)). -> Note: this adapter is currently a deliberate **pass-through** — the -> `OtapPdata ↔ OtapMetricRecords` `From`/`Into` binding is the one seam -> left to wire (marked Phase D/E in the code). Everything it drives -> (Layer B) is complete and tested. +> Note: the `OtapPdata ↔ OtapMetricRecords` binding is now implemented +> (`otap-patch/all/otap_bridge.rs`) — real OTLP metrics in, +> sketch/estimate output back out as real OTLP metrics, for the +> **producer role** (`AsapSketchesPlugin::start_from_envelopes`'s +> **receiver role** — ingesting another `asap_sketches` node's +> `SketchStreamBatch` output — isn't covered; that format doesn't fit +> OTAP's metrics shape). The adapter drives a bare `Precompute` +> instance directly rather than `AsapSketchesPlugin`'s own Tokio-task +> lifecycle, whose emit shape (`SketchStreamBatch`, PR #5/#6's +> dictionary economics) diverged from what this binding needs. +> **Unverified**: `otap-patch/` has no standalone build in this repo +> (see Layer A/B split below), so this binding has been checked +> against upstream source reads, not a compiler — see +> `otap_bridge.rs`'s module doc for exactly what's confirmed. **Layer B — the runtime lifecycle + Arrow codec** (`asap-precompute-rs/src/otap/`) @@ -55,7 +67,7 @@ host-neutral edge precompute runtime. ```sh cd asap-precompute-rs cargo build --features otap -cargo test --features otap # 135 tests: runtime + OTAP codec + lifecycle +cargo test --features otap # 156 tests: runtime + OTAP codec + lifecycle cargo clippy --all-targets --features otap -- -D warnings cargo fmt --check ``` @@ -73,9 +85,12 @@ public dependency. Phases **B** (Arrow codec) and **C** (full 5-sketch plugin lifecycle) are complete and tested here. Phase **D** (the `linkme` registration + -OTAP submodule build wiring) is present as the `otap-patch/` overlay -with the processor adapter as a pass-through; the `OtapPdata` binding -and cross-host byte-parity (Phase **E**) are the remaining work. +OTAP submodule build wiring, plus the `OtapPdata` binding) is present +as the `otap-patch/` overlay — the producer-role binding is now +implemented (`otap_bridge.rs`) but **unverified** (no standalone build +of `otap-patch/` in this repo — see the Layer A note above). The +receiver-role `OtapPdata` binding and cross-host byte-parity +(Phase **E**) are the remaining work. --- diff --git a/asap-precompute-rs/Cargo.lock b/asap-precompute-rs/Cargo.lock index f579db4..e6e40ef 100644 --- a/asap-precompute-rs/Cargo.lock +++ b/asap-precompute-rs/Cargo.lock @@ -16,12 +16,6 @@ dependencies = [ "zerocopy", ] -[[package]] -name = "android-tzdata" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" - [[package]] name = "android_system_properties" version = "0.1.5" @@ -39,9 +33,9 @@ checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arrow-array" -version = "53.4.1" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7845c32b41f7053e37a075b3c2f29c6f5ea1b3ca6e5df7a2d325ee6e1b4a63cf" +checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" dependencies = [ "ahash", "arrow-buffer", @@ -50,43 +44,76 @@ dependencies = [ "chrono", "half", "hashbrown", - "num", + "num-complex", + "num-integer", + "num-traits", ] [[package]] name = "arrow-buffer" -version = "53.4.1" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b5c681a99606f3316f2a99d9c8b6fa3aad0b1d34d8f6d7a1b471893940219d8" +checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" dependencies = [ "bytes", "half", - "num", + "num-bigint", + "num-traits", ] [[package]] name = "arrow-data" -version = "53.4.1" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd962fc3bf7f60705b25bcaa8eb3318b2545aa1d528656525ebdd6a17a6cd6fb" +checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" dependencies = [ "arrow-buffer", "arrow-schema", "half", - "num", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-ipc" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", + "flatbuffers", ] [[package]] name = "arrow-schema" -version = "53.4.1" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" + +[[package]] +name = "arrow-select" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35b0f9c0c3582dd55db0f136d3b44bfa0189df07adcf7dc7f2f2e74db0f52eb8" +checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +dependencies = [ + "ahash", + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "num-traits", +] [[package]] name = "asap-precompute-rs" version = "0.1.0" dependencies = [ "arrow-array", + "arrow-ipc", "arrow-schema", "asap_sketchlib", "futures", @@ -96,12 +123,13 @@ dependencies = [ "serde_json", "thiserror", "tokio", + "tokio-stream", ] [[package]] name = "asap_sketchlib" version = "0.2.2" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=0a2ac3725f9b6d562f3ed7d9a48c6a1ae0c285e6#0a2ac3725f9b6d562f3ed7d9a48c6a1ae0c285e6" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=010457de46ba9ec5574a53476bbe8b4bdb02b965#010457de46ba9ec5574a53476bbe8b4bdb02b965" dependencies = [ "bytes", "prost", @@ -109,6 +137,7 @@ dependencies = [ "rmp-serde", "serde", "serde-big-array", + "serde_bytes", "smallvec", "twox-hash", "xxhash-rust", @@ -171,14 +200,13 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "chrono" -version = "0.4.39" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ - "android-tzdata", "iana-time-zone", "num-traits", - "windows-targets", + "windows-link", ] [[package]] @@ -241,6 +269,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flatbuffers" +version = "25.12.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3" +dependencies = [ + "bitflags", + "rustc_version", +] + [[package]] name = "fnv" version = "1.0.7" @@ -383,9 +421,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.15.5" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" [[package]] name = "iana-time-zone" @@ -487,20 +525,6 @@ dependencies = [ "windows-sys", ] -[[package]] -name = "num" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" -dependencies = [ - "num-bigint", - "num-complex", - "num-integer", - "num-iter", - "num-rational", - "num-traits", -] - [[package]] name = "num-bigint" version = "0.4.8" @@ -529,28 +553,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "num-iter" -version = "0.1.45" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1429034a0490724d0075ebb2bc9e875d6503c3cf69e235a8941aa757d83ef5bf" -dependencies = [ - "autocfg", - "num-integer", - "num-traits", -] - -[[package]] -name = "num-rational" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" -dependencies = [ - "num-bigint", - "num-integer", - "num-traits", -] - [[package]] name = "num-traits" version = "0.2.19" @@ -755,6 +757,15 @@ dependencies = [ "serde", ] +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.4" @@ -792,6 +803,12 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.228" @@ -811,6 +828,16 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.228" @@ -963,6 +990,17 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "twox-hash" version = "2.1.2" @@ -1127,70 +1165,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "windows-targets" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" -dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", -] - -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" - -[[package]] -name = "windows_aarch64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" - -[[package]] -name = "windows_i686_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" - -[[package]] -name = "windows_i686_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" - -[[package]] -name = "windows_i686_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" - -[[package]] -name = "windows_x86_64_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" - -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" - -[[package]] -name = "windows_x86_64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" - [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/asap-precompute-rs/Cargo.toml b/asap-precompute-rs/Cargo.toml index f93de8a..5695684 100644 --- a/asap-precompute-rs/Cargo.toml +++ b/asap-precompute-rs/Cargo.toml @@ -7,7 +7,7 @@ license = "MIT" # asap_sketchlib is a public dependency. [dependencies] -asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "0a2ac3725f9b6d562f3ed7d9a48c6a1ae0c285e6" } +asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "010457de46ba9ec5574a53476bbe8b4bdb02b965" } serde = { version = "1", features = ["derive"] } prost = "0.13" thiserror = "1" @@ -15,15 +15,27 @@ thiserror = "1" # Apache Arrow is only pulled in by the OTAP codec. Consumers on the # row-oriented data model never touch Arrow, so the `otap` feature is # non-default to keep their build cheap. -arrow-array = { version = "53", optional = true } -arrow-schema = { version = "53", optional = true } +# +# Pinned to 58.3 to match the OTAP Dataflow workspace's own pin +# (rust/otap-dataflow/Cargo.toml as of otel-arrow@3e85c34, 2026-08-24) +# rather than an independently-chosen version: `RecordBatch` built here +# must be the literal same type `otap-patch/`'s upstream-facing binding +# code hands to OTAP's engine, which only holds if both sides resolve +# to one canonical `arrow-array` version. An older/newer pin here would +# make that binding a type mismatch even though both are "Arrow". +arrow-array = { version = "58.3", optional = true } +arrow-schema = { version = "58.3", optional = true } +# Only pulled in for otap::wire's Arrow IPC serialization of +# SketchStreamBatch across a real transport (see docs/data_model.md's +# "crosses a node or network boundary" framing). +arrow-ipc = { version = "58.3", optional = true } # Tokio drives the plugin lifecycle (Stream consumer # task + Wakeup-driven flush ticker + control-channel poll task + -# graceful drain). Only the `rt`, `sync`, `time`, and `macros` -# sub-features are needed; the full set is pulled in via the `otap` -# feature gate so default-feature consumers don't pay for it. -tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "time", "macros"], optional = true } +# graceful drain), plus `otap::wire`'s TCP transport (`net`, +# `io-util`). The full set is pulled in via the `otap` feature gate so +# default-feature consumers don't pay for it. +tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "net", "io-util"], optional = true } futures = { version = "0.3", optional = true } [dev-dependencies] @@ -32,14 +44,27 @@ serde_json = "1" # The full tokio runtime is required for #[tokio::test]-driven # lifecycle tests under `tests/otap_*.rs`. tokio = { version = "1", features = ["full"] } +# Only used to adapt an `mpsc::UnboundedReceiver` into a `Stream` for +# chaining two `AsapSketchesPlugin`s (producer's EmitReceiver -> +# receiver's start_from_envelopes) in tests and the network-transport +# examples. The library itself is Stream-source-agnostic. +tokio-stream = "0.1" [features] default = [] # Enables the OTAP codec + plugin lifecycle at `crate::otap` # (decode_batch / encode_batch over `arrow::RecordBatch`, plus the # `AsapSketchesPlugin` Tokio runtime). -otap = ["dep:arrow-array", "dep:arrow-schema", "dep:tokio", "dep:futures"] +otap = ["dep:arrow-array", "dep:arrow-schema", "dep:arrow-ipc", "dep:tokio", "dep:futures"] [[example]] name = "sketch_pipeline_demo" required-features = ["otap"] + +[[example]] +name = "sketch_producer_node" +required-features = ["otap"] + +[[example]] +name = "sketch_receiver_node" +required-features = ["otap"] diff --git a/asap-precompute-rs/examples/sketch_producer_node.rs b/asap-precompute-rs/examples/sketch_producer_node.rs new file mode 100644 index 0000000..a00b6ad --- /dev/null +++ b/asap-precompute-rs/examples/sketch_producer_node.rs @@ -0,0 +1,214 @@ +//! Network-transport half of the `docs/data_model.md` demo: a +//! **sketch creation processor** running as a real `AsapSketchesPlugin` +//! (not called directly like `sketch_pipeline_demo.rs` does), pushing +//! each emitted [`SketchStreamBatch`] over a real TCP socket — Arrow +//! IPC-serialized via `otap::wire` — to `sketch_receiver_node` +//! (`examples/sketch_receiver_node.rs`), which must already be +//! listening. +//! +//! Run (in one terminal, first): +//! ```text +//! cargo run --example sketch_receiver_node --features otap +//! ``` +//! Then (in a second terminal): +//! ```text +//! cargo run --example sketch_producer_node --features otap +//! ``` +//! +//! Unlike `sketch_pipeline_demo.rs`'s in-process `mpsc` channel, what +//! crosses the wire here is genuinely serialized bytes over a socket +//! — the actual "crosses a node or network boundary" hop +//! `docs/data_model.md` opens with. See `otap::wire`'s module doc for +//! the exact frame layout. +//! +//! Unlike `sketch_pipeline_demo.rs` (which force-closes windows with +//! explicit `drain()` calls), this binary feeds observations through +//! a real OTAP-shaped input stream (`records::flatten` + +//! `decode_batch`, `AsapSketchesPlugin::start`'s input task) and lets +//! the plugin's real `Wakeup`-style Tokio ticker close windows on its +//! own wall-clock schedule — a `window_size` short enough (300ms) to +//! see several windows roll in one run. + +use std::time::Duration; + +use arrow_array::{BinaryArray, Float64Array, RecordBatch, StringArray, UInt32Array, UInt64Array}; +use arrow_schema::{DataType, Field, Schema}; +use asap_precompute_rs::otap::config::PluginConfig; +use asap_precompute_rs::otap::records::{ + OtapMetricRecords, ATTR_BATCH_BYTES, ATTR_BATCH_INT, ATTR_BATCH_KEY, ATTR_BATCH_PARENT_ID, + ATTR_BATCH_STR, +}; +use asap_precompute_rs::otap::wire::send_stream_batch; +use asap_precompute_rs::otap::{ + AsapSketchesPlugin, StartOptions, COLUMN_METRIC, COLUMN_TIME_UNIX_NANO, COLUMN_VALUE, +}; +use tokio::net::TcpStream; +use tokio::sync::mpsc; +use tokio_stream::wrappers::UnboundedReceiverStream; + +/// Must match `sketch_receiver_node`'s listen address. +const RECEIVER_ADDR: &str = "127.0.0.1:47821"; +const AGG_ID: u64 = 1; +const NUM_WINDOWS: usize = 4; +const WINDOW_SIZE: Duration = Duration::from_millis(300); +/// Longer than `WINDOW_SIZE` so each window's data has already landed +/// before the real ticker rotates it out — real wall-clock pacing, not +/// a guaranteed lockstep boundary, so this is a margin, not a promise. +const PACING: Duration = Duration::from_millis(400); + +#[tokio::main] +async fn main() { + let mut socket = connect_with_retry(RECEIVER_ADDR).await; + println!("[producer] connected to {RECEIVER_ADDR}"); + + let plugin_cfg = PluginConfig { + sketch_type: "ddsketch".into(), + window_size: WINDOW_SIZE, + output_metric_name: "http_request_duration_ms".into(), + agg_id: AGG_ID, + sketch_params: [("relative_accuracy".to_string(), 0.01)] + .into_iter() + .collect(), + ..Default::default() + }; + let plugin = AsapSketchesPlugin::from_plugin_config(&plugin_cfg).expect("producer config"); + + // Bridge a paced feed of synthetic OTAP-shaped input records into + // the Stream AsapSketchesPlugin::start + // wants -- this is the plugin's real ingest path (records::flatten + // + decode_batch), not a direct Precompute::observe() call. + let (input_tx, input_rx) = mpsc::unbounded_channel::(); + let feed_task = tokio::spawn(async move { + for window_idx in 0..NUM_WINDOWS { + // One series (path=/api), latency drifting upward window + // to window so the receiver's printed p99 visibly moves. + for i in 0..200u64 { + let base = 10.0 + (window_idx as f64) * 8.0; + let latency = base + (i % 25) as f64; + let records = + build_scalar_records("http_request_duration_ms", latency, now_ms(), "/api"); + if input_tx.send(records).is_err() { + return; // plugin shut down early. + } + } + // Let the ticker rotate this window out before the next + // window's data starts arriving. + tokio::time::sleep(PACING).await; + } + // Dropping input_tx here ends the plugin's input stream. + }); + + let (handle, mut emit_rx) = plugin.start( + UnboundedReceiverStream::new(input_rx), + None, + StartOptions::default(), + ); + + // Forward every emitted SketchStreamBatch over the socket as it + // arrives, concurrently with the feed task still running. + let forward_task = tokio::spawn(async move { + let mut window_idx = 0; + while let Some(batch) = emit_rx.recv().await { + println!( + "[producer] window {window_idx}: schema={} dictionary={} labels={} record={} row(s) -- sending over the wire", + batch.schema.num_rows(), + batch.dictionary.num_rows(), + batch.labels.num_rows(), + batch.record.num_rows(), + ); + send_stream_batch(&mut socket, &batch) + .await + .expect("send over socket"); + window_idx += 1; + } + socket + }); + + feed_task.await.expect("feed task"); + // Give the last window's data one more full period to roll + // naturally, then shut down -- the final drain flushes any + // residue that hadn't hit a tick boundary yet. + tokio::time::sleep(PACING).await; + handle.shutdown().await.expect("producer shutdown"); + + let socket = forward_task.await.expect("forward task"); + drop(socket); // close the connection -> receiver sees a clean EOF. + println!("[producer] done, connection closed"); +} + +/// Builds a one-row [`OtapMetricRecords`] for a raw scalar +/// observation with a single `path` label -- the OTAP-Metrics-shaped +/// input `AsapSketchesPlugin::start`'s input task consumes from a +/// real upstream OTAP source (Telegraf / Vector / another OTAP +/// collector). +fn build_scalar_records( + metric: &str, + value: f64, + timestamp_ms: u64, + path: &str, +) -> OtapMetricRecords { + let metrics_schema = std::sync::Arc::new(Schema::new(vec![ + Field::new(COLUMN_TIME_UNIX_NANO, DataType::UInt64, false), + Field::new(COLUMN_METRIC, DataType::Utf8, false), + Field::new(COLUMN_VALUE, DataType::Float64, false), + Field::new(ATTR_BATCH_PARENT_ID, DataType::UInt32, false), + ])); + let metrics = RecordBatch::try_new( + metrics_schema, + vec![ + std::sync::Arc::new(UInt64Array::from(vec![timestamp_ms * 1_000_000])), + std::sync::Arc::new(StringArray::from(vec![metric])), + std::sync::Arc::new(Float64Array::from(vec![value])), + std::sync::Arc::new(UInt32Array::from(vec![0_u32])), + ], + ) + .expect("metrics batch"); + let attributes_schema = std::sync::Arc::new(Schema::new(vec![ + Field::new(ATTR_BATCH_PARENT_ID, DataType::UInt32, false), + Field::new(ATTR_BATCH_KEY, DataType::Utf8, false), + Field::new(ATTR_BATCH_BYTES, DataType::Binary, true), + Field::new(ATTR_BATCH_STR, DataType::Utf8, true), + Field::new(ATTR_BATCH_INT, DataType::UInt64, true), + ])); + let attributes = RecordBatch::try_new( + attributes_schema, + vec![ + std::sync::Arc::new(UInt32Array::from(vec![0_u32])), + std::sync::Arc::new(StringArray::from(vec!["path"])), + std::sync::Arc::new(BinaryArray::from_opt_vec(vec![None as Option<&[u8]>])), + std::sync::Arc::new(StringArray::from(vec![Some(path)])), + std::sync::Arc::new(UInt64Array::from(vec![None as Option])), + ], + ) + .expect("attributes batch"); + OtapMetricRecords { + metrics, + attributes, + } +} + +/// Retries the connection a few times -- `sketch_receiver_node` may +/// not have bound its listener yet if both binaries are started at +/// nearly the same moment. +async fn connect_with_retry(addr: &str) -> TcpStream { + for attempt in 0..20 { + match TcpStream::connect(addr).await { + Ok(s) => return s, + Err(e) => { + if attempt == 0 { + println!("[producer] waiting for {addr} to accept connections ({e})..."); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + } + panic!("could not connect to {addr} after retries -- is sketch_receiver_node running?"); +} + +fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock before Unix epoch") + .as_millis() as u64 +} diff --git a/asap-precompute-rs/examples/sketch_receiver_node.rs b/asap-precompute-rs/examples/sketch_receiver_node.rs new file mode 100644 index 0000000..a92713f --- /dev/null +++ b/asap-precompute-rs/examples/sketch_receiver_node.rs @@ -0,0 +1,133 @@ +//! Network-transport half of the `docs/data_model.md` demo: a +//! **receive processor** running as a real `AsapSketchesPlugin` in +//! its receiver role (`start_from_envelopes` — see +//! `src/otap/lifecycle.rs`'s module doc, "The other role"), reading +//! [`SketchStreamBatch`]es off a real TCP socket (Arrow IPC-decoded +//! via `otap::wire`) sent by `sketch_producer_node` +//! (`examples/sketch_producer_node.rs`). +//! +//! Merges every reconstructed envelope via `Precompute::observe_envelope`, +//! then — because this plugin's own config sets `transmit_sketch = +//! false`, `quantiles = [0.99]` — its own ticker/drain naturally +//! produces p99 *estimate* envelopes instead of re-emitting sketch +//! bytes. Those come back out through this plugin's own emit channel +//! as another `SketchStreamBatch`, which this binary decodes and +//! prints as Prometheus text (the "Prometheus backend" stage — see +//! `sketch_pipeline_demo.rs`'s module doc for why printing stands in +//! for a real `/metrics` HTTP handler). +//! +//! Run this first (see `sketch_producer_node.rs`'s module doc): +//! ```text +//! cargo run --example sketch_receiver_node --features otap +//! ``` + +use std::time::Duration; + +use asap_precompute_rs::envelope::SketchEnvelope; +use asap_precompute_rs::otap::config::PluginConfig; +use asap_precompute_rs::otap::wire::recv_stream_batch; +use asap_precompute_rs::otap::{ + AsapSketchesPlugin, SeriesDictionaryDecoder, SketchStreamBatch, StartOptions, +}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio_stream::wrappers::UnboundedReceiverStream; + +/// Must match `sketch_producer_node`'s connect address. +const LISTEN_ADDR: &str = "127.0.0.1:47821"; +const AGG_ID: u64 = 1; + +#[tokio::main] +async fn main() { + let listener = TcpListener::bind(LISTEN_ADDR).await.expect("bind"); + println!("[receiver] listening on {LISTEN_ADDR}, waiting for sketch_producer_node..."); + let (mut socket, peer) = listener.accept().await.expect("accept"); + println!("[receiver] accepted connection from {peer}"); + + // Bridge "read framed batches off the socket" into the + // Stream start_from_envelopes wants. + let (batch_tx, batch_rx) = mpsc::unbounded_channel::(); + let socket_task = tokio::spawn(async move { + loop { + match recv_stream_batch(&mut socket).await { + Ok(Some(batch)) => { + println!( + "[receiver] received over the wire: schema={} dictionary={} labels={} record={} row(s)", + batch.schema.num_rows(), + batch.dictionary.num_rows(), + batch.labels.num_rows(), + batch.record.num_rows(), + ); + if batch_tx.send(batch).is_err() { + return; // plugin already gone. + } + } + Ok(None) => { + println!("[receiver] producer closed the connection"); + return; + } + Err(e) => { + eprintln!("[receiver] wire error: {e}"); + return; + } + } + } + }); + + let plugin_cfg = PluginConfig { + sketch_type: "ddsketch".into(), + window_size: Duration::from_secs(60), // driven by drain() on shutdown, same as the producer. + output_metric_name: "http_request_duration_ms_p99".into(), + agg_id: AGG_ID, + transmit_sketch: false, // query mode: emits quantile estimates, not sketch bytes. + quantiles: vec![0.99], + ..Default::default() + }; + let plugin = AsapSketchesPlugin::from_plugin_config(&plugin_cfg).expect("receiver config"); + let (handle, mut emit_rx) = plugin.start_from_envelopes( + UnboundedReceiverStream::new(batch_rx), + None, + StartOptions::default(), + ); + + // Wait for the producer to finish sending (socket EOF) before + // asking for the final drain -- otherwise shutdown could race a + // still-in-flight batch. + socket_task.await.expect("socket task"); + handle.shutdown().await.expect("receiver shutdown"); + + let mut decoder = SeriesDictionaryDecoder::new(); + while let Ok(Some(batch)) = + tokio::time::timeout(Duration::from_millis(500), emit_rx.recv()).await + { + for estimate in decoder.decode(&batch).expect("decode estimate batch") { + print!("{}", format_prometheus_gauge(&estimate)); + } + } +} + +/// Formats one estimate-mode [`SketchEnvelope`] (`payload` empty, +/// `value` set — see `docs/data_model.md`'s `RECORD.value`) as a +/// Prometheus text-exposition gauge sample. Stands in for a real +/// `/metrics` HTTP handler. +fn format_prometheus_gauge(env: &SketchEnvelope) -> String { + let mut labels: Vec = env + .labels + .iter() + .map(|kv| format!("{}=\"{}\"", kv.key, kv.value)) + .collect(); + labels.sort(); + let label_str = if labels.is_empty() { + String::new() + } else { + format!("{{{}}}", labels.join(",")) + }; + format!( + "# HELP {name} sketch-derived quantile estimate\n\ + # TYPE {name} gauge\n\ + {name}{label_str} {value} {ts}\n", + name = env.metric_name, + value = env.value, + ts = env.window_end_ms, + ) +} diff --git a/asap-precompute-rs/src/config.rs b/asap-precompute-rs/src/config.rs index 501cf85..3041825 100644 --- a/asap-precompute-rs/src/config.rs +++ b/asap-precompute-rs/src/config.rs @@ -152,23 +152,50 @@ pub fn sketch_param_get(params: &SketchParams, key: &str, default: f64) -> f64 { pub fn sketch_size_string(sketch_type: SketchType, params: &SketchParams) -> Option { match sketch_type { SketchType::DDSketch => params.get("relative_accuracy").map(f64::to_string), - SketchType::KLLSketch => params.get("k").map(|v| (*v as u64).to_string()), - SketchType::HLLSketch => params.get("precision").map(|v| (*v as u64).to_string()), + SketchType::KLLSketch => params + .get("k") + .and_then(|v| as_wire_u64(*v)) + .map(|v| v.to_string()), + SketchType::HLLSketch => params + .get("precision") + .and_then(|v| as_wire_u64(*v)) + .map(|v| v.to_string()), SketchType::CountSketch => match (params.get("width"), params.get("depth")) { - (Some(w), Some(d)) => Some(format!("{} x {}", *w as u64, *d as u64)), + (Some(w), Some(d)) => match (as_wire_u64(*w), as_wire_u64(*d)) { + (Some(w), Some(d)) => Some(format!("{w} x {d}")), + _ => None, + }, _ => match (params.get("epsilon"), params.get("delta")) { (Some(e), Some(d)) => Some(format!("epsilon={e}, delta={d}")), _ => None, }, }, SketchType::CountMinSketch => match (params.get("rows"), params.get("columns")) { - (Some(r), Some(c)) => Some(format!("{} x {}", *r as u64, *c as u64)), + (Some(r), Some(c)) => match (as_wire_u64(*r), as_wire_u64(*c)) { + (Some(r), Some(c)) => Some(format!("{r} x {c}")), + _ => None, + }, _ => None, }, SketchType::Unspecified => None, } } +/// Converts a sketch-param value to its `u64` wire form, or `None` if +/// it isn't cleanly representable as one — negative, non-finite, or +/// carrying a fractional part (e.g. a unit mixup like `k = 0.5`) all +/// indicate a misconfigured param. Silently truncating (`as u64`) +/// would report a fabricated, wrong `sketch_size` on the wire instead +/// (`k = 0.5` truncating to `"0"`); better to omit the field, mirroring +/// `resolve_hash_seed`'s "omit rather than fabricate" stance for a +/// malformed input. +fn as_wire_u64(v: f64) -> Option { + if !v.is_finite() || v < 0.0 || v.fract() != 0.0 { + return None; + } + Some(v as u64) +} + /// Host-neutral form of today's per-OTel-processor `Config` struct, /// plus `max_series` / `on_overflow`. #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] diff --git a/asap-precompute-rs/src/otap/dictionary.rs b/asap-precompute-rs/src/otap/dictionary.rs index ecd06fc..8648414 100644 --- a/asap-precompute-rs/src/otap/dictionary.rs +++ b/asap-precompute-rs/src/otap/dictionary.rs @@ -89,6 +89,16 @@ impl SketchStreamBatch { } } +/// Appends `s` to `buf` as a length-prefixed segment (`":"`) +/// — used by [`SeriesDictionary::identity_key`] so joining several +/// caller-controlled strings can never let one segment's content be +/// mistaken for a delimiter or for the start of the next segment. +fn push_len_prefixed(buf: &mut String, s: &str) { + buf.push_str(&s.len().to_string()); + buf.push(':'); + buf.push_str(s); +} + /// Sender-side dictionary state: assigns stable `series_id`s and /// tracks which `agg_id`s / series have already had a `SCHEMA` / /// `DICTIONARY` row emitted, so repeat windows for the same series @@ -130,19 +140,24 @@ impl SeriesDictionary { /// canonicalization is private to this dictionary and never /// crosses a process boundary itself, only the `series_id` it /// produces does. + /// + /// Each segment is length-prefixed (`":"`) rather than + /// joined with a bare `|`/`=`/`;` delimiter — those separator + /// characters are otherwise legal inside a label key/value (e.g. + /// an HTTP path label containing `;`), and an unescaped join lets + /// two genuinely different label sets collide onto the same + /// string (`{"a": "1;b=2"}` vs. `{"a": "1", "b": "2"}` both used + /// to produce `"...|a=1;b=2;"`). A length prefix makes each + /// segment's boundary unambiguous regardless of its contents. fn identity_key(env: &SketchEnvelope) -> String { let mut labels: Vec<&KeyValue> = env.labels.iter().collect(); labels.sort_by(|a, b| a.key.cmp(&b.key)); let mut buf = String::new(); - buf.push_str(&env.agg_id.to_string()); - buf.push('|'); - buf.push_str(&env.metric_name); - buf.push('|'); + push_len_prefixed(&mut buf, &env.agg_id.to_string()); + push_len_prefixed(&mut buf, &env.metric_name); for kv in labels { - buf.push_str(&kv.key); - buf.push('='); - buf.push_str(&kv.value); - buf.push(';'); + push_len_prefixed(&mut buf, &kv.key); + push_len_prefixed(&mut buf, &kv.value); } buf } @@ -212,7 +227,7 @@ impl SeriesDictionary { cfg.filter(|c| c.agg_id == env.agg_id) .and_then(|c| sketch_size_string(env.sketch_type, &c.sketch_params)), ); - let (seed, function) = resolve_hash_seed(env.hash_spec.as_ref()); + let (seed, function) = resolve_hash_seed(env.sketch_type, env.hash_spec.as_ref()); schema_hash_seed.push(seed); schema_hash_function.push(function); schema_encoding.push(env.encoding.name()); @@ -341,25 +356,39 @@ impl SeriesDictionary { /// sketch families at once. `SCHEMA_COLUMN_HASH_SEED` doesn't need /// that generality: one `SCHEMA` row already describes exactly one /// `agg_id`'s one `sketch_type`, so there's exactly one seed position -/// that matters — `canonical_seed_index` is the one both libraries use -/// by default, and the field `HashSpec` actually exposes on this proto -/// message (`docs/data_model.md`'s `hash_seed` field is deliberately -/// this single resolved value, not the whole table). +/// that matters — but *which* position depends on `sketch_type`. +/// `asap_sketchlib`'s matrix-family sketches (`CountSketch` / +/// `CountMinSketch`) always hash on the packed hot path via +/// `HashSpec::matrix_seed()`, i.e. `seed_list[0]`, regardless of +/// `canonical_seed_index` — `canonical_seed_index` only governs +/// `CanonicalHash`/`hh_keys` lookups, a different code path. Every +/// other sketch type here uses the canonical position, +/// `seed_list[canonical_seed_index]`. Reporting the wrong one would +/// silently mismatch a receiver's determinism/compatibility check +/// against the seed the bytes were actually hashed with. /// /// Returns `(None, None)` when `spec` is absent (nothing upstream -/// populates [`SketchEnvelope::hash_spec`] yet) or when -/// `canonical_seed_index` is out of bounds for `seed_list` (a -/// malformed spec — better to omit the seed than fabricate one). +/// populates [`SketchEnvelope::hash_spec`] yet) or when the resolved +/// index is out of bounds for `seed_list` (a malformed spec — better +/// to omit the seed than fabricate one). fn resolve_hash_seed( + sketch_type: SketchType, spec: Option<&asap_sketchlib::proto::sketchlib::HashSpec>, ) -> (Option, Option) { let Some(spec) = spec else { return (None, None); }; - let seed = spec - .seed_list - .get(spec.canonical_seed_index as usize) - .copied(); + let is_matrix_family = matches!( + sketch_type, + SketchType::CountSketch | SketchType::CountMinSketch + ); + let seed = if is_matrix_family { + spec.seed_list.first().copied() + } else { + spec.seed_list + .get(spec.canonical_seed_index as usize) + .copied() + }; let function = asap_sketchlib::proto::sketchlib::HashAlgorithm::try_from(spec.algorithm) .ok() .map(|a| a.as_str_name().to_string()); @@ -504,14 +533,28 @@ impl SeriesDictionaryDecoder { let metric = col_str(batch, "dictionary", DICT_COLUMN_METRIC)?; for row in 0..batch.num_rows() { let sid = series_id.value(row); - self.series.insert( - sid, - SeriesFacts { + // A DICTIONARY row for a series_id is only ever supposed + // to arrive once, paired with LABELS rows in that same + // batch (`SeriesDictionary::encode` never re-emits either + // for a series it already considers known). If one + // arrives for a series_id this decoder already has facts + // for — a duplicate/replayed batch, or a sender that lost + // its own dictionary state and resent it — preserve the + // labels already learned instead of resetting them to + // empty and hoping a fresh LABELS batch refills them + // (which `SeriesDictionary` won't send, since it still + // considers this series known). + self.series + .entry(sid) + .and_modify(|facts| { + facts.agg_id = agg_id.value(row); + facts.metric = metric.value(row).to_string(); + }) + .or_insert_with(|| SeriesFacts { agg_id: agg_id.value(row), metric: metric.value(row).to_string(), labels: Vec::new(), - }, - ); + }); } Ok(()) } diff --git a/asap-precompute-rs/src/otap/lifecycle.rs b/asap-precompute-rs/src/otap/lifecycle.rs index 85165ca..0c33d64 100644 --- a/asap-precompute-rs/src/otap/lifecycle.rs +++ b/asap-precompute-rs/src/otap/lifecycle.rs @@ -45,7 +45,25 @@ //! boundary silently drops in-flight observations. The drain runs on //! the same emit channel as the flush ticker, so the consumer sees //! exactly one final batch carrying the residue. - +//! +//! # The other role: [`AsapSketchesPlugin::start_from_envelopes`] +//! +//! Everything above is the *producer* role — raw observations in, +//! `SketchStreamBatch`es out. A node receiving from another +//! `asap_sketches` node instead needs the *receiver* role: +//! [`AsapSketchesPlugin::start_from_envelopes`] swaps the input task +//! for one that consumes `Stream`, decodes +//! each via a [`super::dictionary::SeriesDictionaryDecoder`], and +//! routes the reconstructed envelopes through +//! `Precompute::observe_envelope` (merge, never expand to samples). +//! It reuses the exact same flush ticker / control-channel task / +//! graceful-drain machinery as the producer role — a receiver is +//! still free to have its own `Precompute` config (e.g. +//! `transmit_sketch = false` to *query* the merged sketch every window +//! instead of re-emitting it), so its own emit channel carries +//! whatever that config produces. + +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -58,7 +76,7 @@ use crate::envelope::SketchEnvelope; use crate::precompute::{Precompute, PrecomputeError, PrecomputeImpl, StatsSnapshot}; use super::config::{resolve, ConfigError, PluginConfig}; -use super::dictionary::{SeriesDictionary, SketchStreamBatch}; +use super::dictionary::{SeriesDictionary, SeriesDictionaryDecoder, SketchStreamBatch}; use super::records::{flatten, OtapMetricRecords, OtapRecordsError}; use super::{decode_batch, OtapDecodeError, OtapEncodeError}; @@ -131,6 +149,17 @@ pub struct AsapSketchesPlugin { /// it's shared between the ticker task and the supervisor's final /// drain. dictionary: Arc>, + /// Count of lifecycle-task batches/windows dropped due to a + /// decode/encode/precompute error since this plugin started. The + /// input, ticker, and drain tasks all apply a "drop the bad batch, + /// keep the plugin alive" resilience policy (a single malformed + /// batch shouldn't take down the whole plugin) — this counter is + /// what keeps that policy from being completely silent. Phase D + /// routes these errors onto OTAP's real effect-handler error + /// channel instead; until then, the `dropped_batches()` accessor + /// is the only signal a caller has that batches are being + /// dropped. + dropped_batches: Arc, } impl AsapSketchesPlugin { @@ -149,6 +178,7 @@ impl AsapSketchesPlugin { inner: Arc::new(pc), window_size: pcfg.window.size, dictionary: Arc::new(Mutex::new(SeriesDictionary::new())), + dropped_batches: Arc::new(AtomicU64::new(0)), }) } @@ -164,6 +194,7 @@ impl AsapSketchesPlugin { inner: precompute, window_size, dictionary: Arc::new(Mutex::new(SeriesDictionary::new())), + dropped_batches: Arc::new(AtomicU64::new(0)), } } @@ -180,8 +211,23 @@ impl AsapSketchesPlugin { self.inner.stats() } + /// Borrow the shared dropped-batch counter — count of + /// batches/windows a lifecycle task dropped after a + /// decode/encode/precompute error, the observable counterpart to + /// the input/ticker/drain tasks' "drop the bad batch, keep the + /// plugin alive" policy. Like [`Self::precompute`], call this + /// *before* [`Self::start`] / [`Self::start_from_envelopes`] + /// (which consume `self`) and clone the returned `Arc` to retain + /// a handle — `.load(Ordering::Relaxed)` on the clone reports the + /// live count for the plugin's whole lifetime, including after + /// shutdown. + pub fn dropped_batches(&self) -> &Arc { + &self.dropped_batches + } + /// Launch the plugin's three lifecycle tasks against an OTAP - /// input stream. + /// input stream — the **producer** role (raw observations in, + /// `SketchStreamBatch`es out). /// /// `input` is the host-supplied stream of `OtapMetricRecords` /// — the OTAP shell wraps the runtime's `Stream` to @@ -201,6 +247,68 @@ impl AsapSketchesPlugin { ) -> (PluginHandle, EmitReceiver) where S: futures::Stream + Send + Unpin + 'static, + { + let precompute = self.inner.clone(); + let dropped_batches = self.dropped_batches.clone(); + self.spawn_lifecycle( + move |cancel| spawn_input_task(precompute, input, dropped_batches, cancel), + control, + opts, + ) + } + + /// Launch the plugin's three lifecycle tasks against a stream of + /// pre-aggregated envelopes — the **receiver** role: another + /// `asap_sketches` node's `SketchStreamBatch` output in, this + /// plugin's own `SketchStreamBatch` output out (which, depending + /// on this plugin's own `Precompute` config, might carry + /// re-emitted sketch state, or — with `transmit_sketch = false` — + /// query-mode estimates of the merged sketch). + /// + /// Decodes each batch through a fresh + /// [`SeriesDictionaryDecoder`] retained for the life of this + /// plugin instance, and routes every reconstructed envelope + /// through `Precompute::observe_envelope` (merge, never expand to + /// samples — see the module doc's "The other role" section). + /// Otherwise identical to [`Self::start`]: same ticker / control / + /// graceful-drain machinery, same [`PluginHandle`] / + /// [`EmitReceiver`] return shape. + pub fn start_from_envelopes( + self, + input: S, + control: Option>, + opts: StartOptions, + ) -> (PluginHandle, EmitReceiver) + where + S: futures::Stream + Send + Unpin + 'static, + { + let precompute = self.inner.clone(); + let decoder = Arc::new(Mutex::new(SeriesDictionaryDecoder::new())); + let dropped_batches = self.dropped_batches.clone(); + self.spawn_lifecycle( + move |cancel| { + spawn_envelope_input_task(precompute, decoder, input, dropped_batches, cancel) + }, + control, + opts, + ) + } + + /// Shared tail of [`Self::start`] / [`Self::start_from_envelopes`]: + /// wires up the emit channel, shutdown signal, ticker task, + /// optional control task, and the graceful-drain supervisor — + /// everything except *which* input task to spawn, which the two + /// public entry points supply as `spawn_input` (given the shared + /// [`Cancellation`] token so the input task honors shutdown the + /// same way the others do). + fn spawn_lifecycle( + self, + spawn_input: F, + control: Option>, + opts: StartOptions, + ) -> (PluginHandle, EmitReceiver) + where + F: FnOnce(Cancellation) -> JoinHandle<()>, { let (emit_tx, emit_rx) = mpsc::unbounded_channel(); let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); @@ -210,14 +318,16 @@ impl AsapSketchesPlugin { let precompute = self.inner.clone(); let window_size = self.window_size; let dictionary = self.dictionary.clone(); + let dropped_batches = self.dropped_batches.clone(); let opts = Arc::new(opts); - let input_task = spawn_input_task(precompute.clone(), input, cancellation.clone()); + let input_task = spawn_input(cancellation.clone()); let ticker_task = spawn_ticker_task( precompute.clone(), dictionary.clone(), window_size, emit_tx.clone(), + dropped_batches.clone(), cancellation.clone(), ); let control_task = control.map(|cc| { @@ -245,7 +355,9 @@ impl AsapSketchesPlugin { if !envs.is_empty() { let cfg = precompute.active_config(); let mut dict = dictionary.lock().await; - let _ = emit_drain(&emit_tx, &envs, &mut dict, cfg.as_ref()); + if emit_drain(&emit_tx, &envs, &mut dict, cfg.as_ref()).is_err() { + dropped_batches.fetch_add(1, Ordering::Relaxed); + } } }); @@ -316,6 +428,7 @@ impl Drop for PluginHandle { fn spawn_input_task( precompute: Arc, mut input: S, + dropped_batches: Arc, cancel: Cancellation, ) -> JoinHandle<()> where @@ -325,7 +438,16 @@ where tokio::spawn(async move { loop { tokio::select! { - _ = cancel.cancelled() => return, + // Biased: check `input.next()` before `cancel.cancelled()`. + // If a batch is already ready in the same poll that + // cancellation fires (e.g. a producer enqueues its + // final batch then immediately signals shutdown), + // unbiased selection could pick the cancel branch and + // drop that ready-but-unprocessed batch. Polling + // `input` first means a ready batch is always + // consumed before the next loop iteration observes + // cancellation. + biased; next = input.next() => match next { None => return, Some(records) => { @@ -334,10 +456,13 @@ where // effect-handler error channel; for Phase C // we drop the batch and continue so a // single bad batch can't take down the - // whole plugin. + // whole plugin. `dropped_batches` keeps + // this from being completely silent. + dropped_batches.fetch_add(1, Ordering::Relaxed); } } }, + _ = cancel.cancelled() => return, } } }) @@ -363,11 +488,77 @@ fn ingest_one_batch( Ok(()) } +fn spawn_envelope_input_task( + precompute: Arc, + decoder: Arc>, + mut input: S, + dropped_batches: Arc, + cancel: Cancellation, +) -> JoinHandle<()> +where + S: futures::Stream + Send + Unpin + 'static, +{ + use futures::StreamExt; + tokio::spawn(async move { + loop { + tokio::select! { + // Biased — see spawn_input_task's comment: without + // this, a batch that's already ready in the same poll + // as a shutdown signal could be dropped by an + // unbiased tie-break instead of processed. + biased; + next = input.next() => match next { + None => return, + Some(batch) => { + if let Err(_e) = ingest_one_stream_batch(&*precompute, &decoder, &batch).await { + // Same "drop the bad batch, keep the + // plugin alive" policy as ingest_one_batch. + // A decode error here means this stream's + // continuity contract was violated (see + // `OtapDecodeError::UnknownSeriesId` / + // `UnknownAggId`) — Phase D routes this + // onto OTAP's effect-handler error channel. + // `dropped_batches` keeps this from being + // completely silent in the meantime. + dropped_batches.fetch_add(1, Ordering::Relaxed); + } + } + }, + _ = cancel.cancelled() => return, + } + } + }) +} + +async fn ingest_one_stream_batch( + precompute: &dyn Precompute, + decoder: &Mutex, + batch: &SketchStreamBatch, +) -> Result<(), PluginError> { + let envelopes = { + let mut d = decoder.lock().await; + d.decode(batch)? + }; + for env in &envelopes { + // Merge only — the runtime never expands envelope bytes back + // into scalar samples (the bandwidth invariant). + match precompute.observe_envelope(env) { + Ok(()) => {} + Err(PrecomputeError::LateData) | Err(PrecomputeError::SeriesCapExceeded) => { + continue; + } + Err(e) => return Err(PluginError::Precompute(e)), + } + } + Ok(()) +} + fn spawn_ticker_task( precompute: Arc, dictionary: Arc>, window_size: Duration, emit_tx: EmitSender, + dropped_batches: Arc, cancel: Cancellation, ) -> JoinHandle<()> { tokio::spawn(async move { @@ -388,7 +579,9 @@ fn spawn_ticker_task( } let cfg = precompute.active_config(); let mut dict = dictionary.lock().await; - let _ = emit_envelopes(&emit_tx, &envs, &mut dict, cfg.as_ref()); + if emit_envelopes(&emit_tx, &envs, &mut dict, cfg.as_ref()).is_err() { + dropped_batches.fetch_add(1, Ordering::Relaxed); + } } } } @@ -532,4 +725,98 @@ mod tests { // And that the SketchType enum landed correctly via update_config. let _ = SketchType::KLLSketch; } + + #[tokio::test] + async fn receiver_role_smoke_test_drop_aborts_supervisor() { + // Mirrors handle_drop_aborts_supervisor for the receiver role: + // start_from_envelopes must compile and run against an empty + // SketchStreamBatch stream without deadlocking or panicking. + let cfg = PluginConfig { + sketch_type: "ddsketch".into(), + window_size: Duration::from_millis(50), + ..Default::default() + }; + let plugin = AsapSketchesPlugin::from_plugin_config(&cfg).expect("config"); + let input = futures::stream::empty::(); + let (handle, _rx) = plugin.start_from_envelopes(input, None, StartOptions::default()); + drop(handle); + } + + #[tokio::test] + async fn receiver_role_merges_producer_role_output_end_to_end() { + // Full producer -> receiver chain, both AsapSketchesPlugin, + // connected by an in-process channel (the network-transport + // version of this same chain lives in + // examples/sketch_producer_node.rs / + // examples/sketch_receiver_node.rs). + use crate::observation::{KeyValue, Observation, ObservationValue}; + use tokio_stream::wrappers::UnboundedReceiverStream; + + let producer_cfg = PluginConfig { + sketch_type: "ddsketch".into(), + window_size: Duration::from_secs(60), // drained explicitly below. + output_metric_name: "latency_ms".into(), + agg_id: 1, + ..Default::default() + }; + let producer = + AsapSketchesPlugin::from_plugin_config(&producer_cfg).expect("producer config"); + for v in [1.0_f64, 2.0, 3.0, 4.0, 5.0] { + let obs = Observation::new( + 1_000, + "latency_ms", + vec![], + vec![KeyValue::new("host", "h1")], + ObservationValue::float(v), + ); + producer.precompute().observe(&obs).expect("observe"); + } + let (producer_handle, producer_rx) = producer.start( + futures::stream::pending::(), + None, + StartOptions::default(), + ); + + let receiver_cfg = PluginConfig { + sketch_type: "ddsketch".into(), + window_size: Duration::from_secs(60), + output_metric_name: "latency_ms_p99".into(), + agg_id: 1, + transmit_sketch: false, + quantiles: vec![0.99], + ..Default::default() + }; + let receiver = + AsapSketchesPlugin::from_plugin_config(&receiver_cfg).expect("receiver config"); + let (receiver_handle, mut receiver_rx) = receiver.start_from_envelopes( + UnboundedReceiverStream::new(producer_rx), + None, + StartOptions::default(), + ); + + // Shut the producer down first: its final drain pushes the + // one window it accumulated onto producer_rx, which the + // receiver's envelope-input task picks up and merges. + producer_handle.shutdown().await.expect("producer shutdown"); + // Now shut the receiver down: its final drain (transmit_sketch + // = false) turns the merged sketch into a p99 estimate batch. + receiver_handle.shutdown().await.expect("receiver shutdown"); + + let mut decoder = SeriesDictionaryDecoder::new(); + let mut saw_estimate = false; + while let Ok(Some(batch)) = + tokio::time::timeout(Duration::from_secs(2), receiver_rx.recv()).await + { + for env in decoder.decode(&batch).expect("decode") { + assert_eq!(env.metric_name, "latency_ms_p99"); + assert!( + env.payload.is_empty(), + "estimate mode carries no sketch bytes" + ); + assert!(env.value > 0.0, "p99 of {{1..5}} must be positive"); + saw_estimate = true; + } + } + assert!(saw_estimate, "receiver never emitted a p99 estimate"); + } } diff --git a/asap-precompute-rs/src/otap/mod.rs b/asap-precompute-rs/src/otap/mod.rs index fb96837..31e7f88 100644 --- a/asap-precompute-rs/src/otap/mod.rs +++ b/asap-precompute-rs/src/otap/mod.rs @@ -131,6 +131,7 @@ mod schema; pub mod config; pub mod lifecycle; pub mod records; +pub mod wire; pub use decode::{decode_batch, OtapDecodeError}; pub use dictionary::{ diff --git a/asap-precompute-rs/src/otap/wire.rs b/asap-precompute-rs/src/otap/wire.rs new file mode 100644 index 0000000..f5ac6cf --- /dev/null +++ b/asap-precompute-rs/src/otap/wire.rs @@ -0,0 +1,427 @@ +//! Arrow-IPC serialization + a minimal length-prefixed framing for +//! carrying a [`SketchStreamBatch`] across a real transport (a TCP +//! socket here) — the actual "crosses a node or network boundary" +//! hop `docs/data_model.md` opens with, rather than the in-process +//! `mpsc` channel [`crate::otap::lifecycle`]'s tests and +//! `examples/sketch_pipeline_demo.rs` use. +//! +//! # Wire shape +//! +//! One [`SketchStreamBatch`] is framed as: +//! +//! ```text +//! [u32 total_len] +//! [u32 schema_len] [schema_len bytes: Arrow IPC stream] +//! [u32 dictionary_len] [dictionary_len bytes: Arrow IPC stream] +//! [u32 labels_len] [labels_len bytes: Arrow IPC stream] +//! [u32 record_len] [record_len bytes: Arrow IPC stream] +//! ``` +//! +//! Each sub-batch is its own self-contained Arrow IPC *stream* +//! (schema message + one record-batch message + EOS) via +//! [`arrow_ipc::writer::StreamWriter`] — not a shared/continuous +//! Arrow IPC stream across the whole session. That's a deliberate +//! simplification: a real continuous-stream transport would let the +//! four sub-streams themselves carry the Schema/Dictionary economics +//! at the Arrow IPC layer too (per `docs/data_model.md`'s closing +//! "Open design question"), but framing each `SketchStreamBatch` as +//! four independent one-shot streams keeps this module's job to +//! exactly "get the same four `RecordBatch`es to the other side +//! intact," leaving `SeriesDictionary`/`SeriesDictionaryDecoder` (not +//! this module) responsible for the actual dedup. +//! +//! All four sub-batch lengths (and the leading `total_len`) are +//! big-endian `u32`s. A zero-row batch still serializes to a valid +//! (small) Arrow IPC stream — a schema message plus an empty record +//! batch — so a wire-level frame always carries exactly four +//! sub-streams even when, say, `schema`/`dictionary`/`labels` are +//! empty because the series was already known. + +use std::io; + +use arrow_array::RecordBatch; +use thiserror::Error; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +use super::dictionary::SketchStreamBatch; + +/// Hard cap [`recv_stream_batch`] applies to the untrusted `total_len` +/// prefix **before** allocating a buffer for it. Not a wire-format +/// constraint (the framing itself allows up to `u32::MAX`, ~4 GiB) — +/// without this cap, a malformed or hostile peer's 4-byte length +/// prefix alone could force an allocation of up to ~4 GiB before a +/// single content byte is validated, a trivial single-connection +/// memory-exhaustion vector against any node acting as a receiver. +/// Chosen well above any `SketchStreamBatch` this crate emits today. +pub const MAX_FRAME_LEN: usize = 256 * 1024 * 1024; // 256 MiB + +/// Failure modes for [`encode_stream_batch`] / [`decode_stream_batch`] +/// / [`send_stream_batch`] / [`recv_stream_batch`]. +#[derive(Debug, Error)] +pub enum WireError { + /// Arrow IPC encode/decode failed (malformed batch, schema + /// mismatch inside a sub-stream, etc.). + #[error("otap wire: arrow ipc error: {0}")] + Arrow(#[from] arrow_schema::ArrowError), + + /// A sub-stream decoded to zero record batches (`StreamReader` + /// yielded nothing) — a well-formed IPC stream always carries + /// exactly one, even for zero rows. + #[error("otap wire: sub-batch {which:?} decoded no record batches")] + EmptyRecordBatch { + /// Which of the four sub-batches was empty. + which: &'static str, + }, + + /// The frame's length prefix didn't match the bytes actually + /// available — a truncated or corrupt frame. + #[error("otap wire: frame truncated: expected {expected} bytes, got {actual}")] + Truncated { + /// Bytes the length prefix promised. + expected: usize, + /// Bytes actually present. + actual: usize, + }, + + /// Network I/O failed while sending/receiving a frame. + #[error("otap wire: io error: {0}")] + Io(#[from] io::Error), + + /// A length prefix (the frame's `total_len`, or a sub-batch's own + /// serialized length on encode) exceeded what this module will + /// represent/accept — either a hostile/corrupt peer's inflated + /// `total_len` ([`recv_stream_batch`]'s [`MAX_FRAME_LEN`] cap), or + /// (on encode) a sub-batch too large for the wire format's `u32` + /// length-prefix field to represent at all without truncating. + #[error("otap wire: frame length {len} exceeds cap of {max} bytes")] + FrameTooLarge { + /// The length that was rejected. + len: usize, + /// The cap it exceeded. + max: usize, + }, +} + +/// Serializes one [`RecordBatch`] as a self-contained Arrow IPC +/// stream (schema + one record batch + EOS). +fn write_ipc_bytes(batch: &RecordBatch) -> Result, WireError> { + let mut buf: Vec = Vec::new(); + { + let mut writer = arrow_ipc::writer::StreamWriter::try_new(&mut buf, &batch.schema())?; + writer.write(batch)?; + writer.finish()?; + } + Ok(buf) +} + +/// Deserializes one [`RecordBatch`] from bytes written by +/// [`write_ipc_bytes`]. `which` names the sub-batch for error +/// messages only. +fn read_ipc_bytes(bytes: &[u8], which: &'static str) -> Result { + let mut reader = arrow_ipc::reader::StreamReader::try_new(bytes, None)?; + match reader.next() { + Some(batch) => Ok(batch?), + None => Err(WireError::EmptyRecordBatch { which }), + } +} + +/// Serializes a whole [`SketchStreamBatch`] into one length-prefixed +/// frame — see the module doc for the exact byte layout. Does *not* +/// include the leading `total_len` prefix; that's added by +/// [`send_stream_batch`] (or by a caller framing its own transport, +/// e.g. writing this to a file). +pub fn encode_stream_batch(batch: &SketchStreamBatch) -> Result, WireError> { + let parts = [ + write_ipc_bytes(&batch.schema)?, + write_ipc_bytes(&batch.dictionary)?, + write_ipc_bytes(&batch.labels)?, + write_ipc_bytes(&batch.record)?, + ]; + let mut out = Vec::with_capacity(parts.iter().map(|p| p.len() + 4).sum()); + for part in &parts { + // Checked, not `as u32`: a silent truncation here would write + // a length prefix smaller than the bytes that actually + // follow, desynchronizing every subsequent frame the decoder + // reads on this stream. + let len = u32::try_from(part.len()).map_err(|_| WireError::FrameTooLarge { + len: part.len(), + max: u32::MAX as usize, + })?; + out.extend_from_slice(&len.to_be_bytes()); + out.extend_from_slice(part); + } + Ok(out) +} + +/// Inverse of [`encode_stream_batch`]: reconstructs a +/// [`SketchStreamBatch`] from a frame's body bytes (i.e. everything +/// after the leading `total_len` prefix, if any). +pub fn decode_stream_batch(bytes: &[u8]) -> Result { + let mut cursor = bytes; + let names = ["schema", "dictionary", "labels", "record"]; + let mut parts: Vec = Vec::with_capacity(4); + for which in names { + let mut len_buf = [0u8; 4]; + std::io::Read::read_exact(&mut cursor, &mut len_buf).map_err(|_| WireError::Truncated { + expected: 4, + actual: cursor.len(), + })?; + let len = u32::from_be_bytes(len_buf) as usize; + if cursor.len() < len { + return Err(WireError::Truncated { + expected: len, + actual: cursor.len(), + }); + } + let (part, rest) = cursor.split_at(len); + parts.push(read_ipc_bytes(part, which)?); + cursor = rest; + } + let mut parts = parts.into_iter(); + Ok(SketchStreamBatch { + schema: parts.next().expect("schema part"), + dictionary: parts.next().expect("dictionary part"), + labels: parts.next().expect("labels part"), + record: parts.next().expect("record part"), + }) +} + +/// Sends one [`SketchStreamBatch`] over `stream`, framed with a +/// leading big-endian `u32` total length. The receiving side reads it +/// back with [`recv_stream_batch`]. +pub async fn send_stream_batch( + stream: &mut TcpStream, + batch: &SketchStreamBatch, +) -> Result<(), WireError> { + let body = encode_stream_batch(batch)?; + let len = u32::try_from(body.len()).map_err(|_| WireError::FrameTooLarge { + len: body.len(), + max: u32::MAX as usize, + })?; + stream.write_all(&len.to_be_bytes()).await?; + stream.write_all(&body).await?; + Ok(()) +} + +/// Reads one [`SketchStreamBatch`] previously written by +/// [`send_stream_batch`]. Returns `Ok(None)` on a clean EOF at a +/// frame boundary (the sender closed the connection after its last +/// batch) rather than an error. +pub async fn recv_stream_batch( + stream: &mut TcpStream, +) -> Result, WireError> { + let mut len_buf = [0u8; 4]; + if !read_exact_or_eof(stream, &mut len_buf).await? { + return Ok(None); + } + let len = u32::from_be_bytes(len_buf) as usize; + if len > MAX_FRAME_LEN { + // Reject before allocating — see MAX_FRAME_LEN's doc. A + // peer's untrusted length prefix must never itself dictate an + // allocation this large. + return Err(WireError::FrameTooLarge { + len, + max: MAX_FRAME_LEN, + }); + } + let mut body = vec![0u8; len]; + stream.read_exact(&mut body).await?; + Ok(Some(decode_stream_batch(&body)?)) +} + +/// Like [`tokio::io::AsyncReadExt::read_exact`], but distinguishes "EOF +/// before any byte of this frame" (returns `Ok(false)` — a clean +/// stream close between frames) from "EOF partway through a frame's +/// length prefix" (still surfaced as an error by the subsequent +/// `read_exact` inside [`recv_stream_batch`], since that's a +/// truncated frame, not a clean close). +async fn read_exact_or_eof(stream: &mut TcpStream, buf: &mut [u8]) -> Result { + let mut filled = 0; + while filled < buf.len() { + let n = stream.read(&mut buf[filled..]).await?; + if n == 0 { + return if filled == 0 { + Ok(false) + } else { + Err(WireError::Truncated { + expected: buf.len(), + actual: filled, + }) + }; + } + filled += n; + } + Ok(true) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::envelope::{Encoding, SketchEnvelope, SketchType}; + use crate::observation::KeyValue; + use crate::otap::dictionary::{SeriesDictionary, SeriesDictionaryDecoder}; + + fn envelope() -> SketchEnvelope { + SketchEnvelope { + schema_version: 1, + sketch_type: SketchType::DDSketch, + agg_id: 7, + resource_labels: Vec::new(), + labels: vec![KeyValue::new("path", "/api")], + window_start_ms: 1_000, + window_end_ms: 11_000, + encoding: Encoding::ProtoFull, + payload: vec![1, 2, 3, 4, 5], + hash_spec: None, + metric_name: "http_request_duration_ms".into(), + count: 42, + aggregation_temporality: 1, + value: 0.0, + } + } + + #[test] + fn encode_decode_round_trips_a_populated_batch() { + let mut dict = SeriesDictionary::new(); + let env = envelope(); + let batch = dict + .encode(std::slice::from_ref(&env), None) + .expect("encode"); + + let bytes = encode_stream_batch(&batch).expect("wire encode"); + let decoded_batch = decode_stream_batch(&bytes).expect("wire decode"); + + assert_eq!(decoded_batch.schema.num_rows(), batch.schema.num_rows()); + assert_eq!( + decoded_batch.dictionary.num_rows(), + batch.dictionary.num_rows() + ); + assert_eq!(decoded_batch.labels.num_rows(), batch.labels.num_rows()); + assert_eq!(decoded_batch.record.num_rows(), batch.record.num_rows()); + + // The whole point: joining the IPC-round-tripped batch back + // through SeriesDictionaryDecoder reconstructs the original + // envelope exactly. + let mut decoder = SeriesDictionaryDecoder::new(); + let out = decoder.decode(&decoded_batch).expect("dictionary decode"); + assert_eq!(out.len(), 1); + assert_eq!(out[0].payload, env.payload); + assert_eq!(out[0].labels, env.labels); + assert_eq!(out[0].metric_name, env.metric_name); + assert_eq!(out[0].sketch_type, env.sketch_type); + } + + #[test] + fn encode_decode_round_trips_an_empty_batch() { + let mut dict = SeriesDictionary::new(); + let batch = dict.encode(&[], None).expect("encode empty"); + let bytes = encode_stream_batch(&batch).expect("wire encode"); + let decoded = decode_stream_batch(&bytes).expect("wire decode"); + assert!(decoded.schema.num_rows() == 0); + assert!(decoded.dictionary.num_rows() == 0); + assert!(decoded.labels.num_rows() == 0); + assert!(decoded.record.num_rows() == 0); + } + + #[test] + fn encode_decode_round_trips_repeat_window_dictionary_free_batch() { + // The batch that matters most: window 2+ for an + // already-known series, where schema/dictionary/labels are + // genuinely empty and only `record` carries a row. + let mut dict = SeriesDictionary::new(); + let env1 = envelope(); + let _ = dict + .encode(std::slice::from_ref(&env1), None) + .expect("window 1"); + let env2 = SketchEnvelope { + window_start_ms: 11_000, + window_end_ms: 21_000, + payload: vec![9, 9, 9], + ..envelope() + }; + let batch2 = dict + .encode(std::slice::from_ref(&env2), None) + .expect("window 2"); + assert_eq!(batch2.schema.num_rows(), 0); + assert_eq!(batch2.dictionary.num_rows(), 0); + assert_eq!(batch2.labels.num_rows(), 0); + assert_eq!(batch2.record.num_rows(), 1); + + let bytes = encode_stream_batch(&batch2).expect("wire encode"); + let decoded_batch = decode_stream_batch(&bytes).expect("wire decode"); + + let mut decoder = SeriesDictionaryDecoder::new(); + // Must ingest window 1 first — this decoded batch alone has + // no DICTIONARY entry to resolve series_id 0 against. + let bytes1 = encode_stream_batch(&dict_only_window1(&env1)).expect("encode w1"); + let decoded1 = decode_stream_batch(&bytes1).expect("decode w1"); + decoder.decode(&decoded1).expect("decode w1 into decoder"); + + let out = decoder.decode(&decoded_batch).expect("decode w2"); + assert_eq!(out.len(), 1); + assert_eq!(out[0].payload, env2.payload); + assert_eq!(out[0].window_start_ms, 11_000); + assert_eq!(out[0].window_end_ms, 21_000); + } + + /// Helper for the test above: re-derives window 1's batch from a + /// fresh dictionary so it can be fed to a fresh decoder + /// independently of the outer test's `dict` state. + fn dict_only_window1(env: &SketchEnvelope) -> SketchStreamBatch { + let mut dict = SeriesDictionary::new(); + dict.encode(std::slice::from_ref(env), None) + .expect("encode") + } + + #[tokio::test] + async fn send_recv_round_trips_over_a_real_tcp_loopback_socket() { + use tokio::net::{TcpListener, TcpStream}; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + + let mut dict = SeriesDictionary::new(); + let env = envelope(); + let batch = dict + .encode(std::slice::from_ref(&env), None) + .expect("encode"); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept"); + recv_stream_batch(&mut socket) + .await + .expect("recv") + .expect("Some(batch)") + }); + + let mut client = TcpStream::connect(addr).await.expect("connect"); + send_stream_batch(&mut client, &batch).await.expect("send"); + drop(client); // signal EOF after the one frame. + + let received = server.await.expect("server task"); + let mut decoder = SeriesDictionaryDecoder::new(); + let out = decoder.decode(&received).expect("decode"); + assert_eq!(out.len(), 1); + assert_eq!(out[0].payload, env.payload); + } + + #[tokio::test] + async fn recv_returns_none_on_clean_eof_between_frames() { + use tokio::net::{TcpListener, TcpStream}; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept"); + recv_stream_batch(&mut socket).await + }); + + let client = TcpStream::connect(addr).await.expect("connect"); + drop(client); // close immediately, no frames sent. + + let result = server.await.expect("server task").expect("no error"); + assert!(result.is_none()); + } +} diff --git a/otap-patch/all/Cargo.toml b/otap-patch/all/Cargo.toml index d007841..57578d2 100644 --- a/otap-patch/all/Cargo.toml +++ b/otap-patch/all/Cargo.toml @@ -18,10 +18,23 @@ # `/otel-arrow/rust/otap-dataflow/crates/asap-sketches-registry/Cargo.toml` # at build time — five levels deep from the repo root, so the # relative path back to `asap-precompute-rs/` is five `../` hops. -# - `linkme` / `otap-df-engine` / `otap-df-otap` / `otap-df-config` -# are taken from the OTAP workspace via `workspace = true` so the -# versions track the upstream submodule's pin (avoids the -# "two distinct versions of linkme in the binary" failure mode). +# - `linkme` / `otel-arrow-dfe-engine` / `otel-arrow-dfe-otap` / +# `otel-arrow-dfe-config` / `otel-arrow-dfe-pdata` / +# `otel-arrow-dfe-pdata-views` / `arrow-*` are all taken from the OTAP +# workspace via `workspace = true` so the versions track the upstream +# submodule's pin (avoids the "two distinct versions of linkme [or +# arrow-array] in the binary" failure mode — see otap_bridge.rs's +# module doc for why the arrow-array version specifically has to be +# the literal same one `asap-precompute-rs` resolves to). +# +# Package names here use the `otel-arrow-dfe-*` prefix (not the older +# `otap-df-*` one asap-precompute-rs's own docs sometimes reference) — +# upstream renamed these very recently (otel-arrow issue #1848, +# `.chloggen/otel-arrow-dfe-crate-prefix.yaml`, unreleased as of +# otel-arrow@3e85c34). If the actual pinned commit `build_asap_otap.sh` +# checks out predates that rename, these names (and `otap_bridge.rs`'s +# `otel_arrow_dfe_*` imports) need reverting to `otap-df-*` / +# `otap_df_*`. [package] name = "asap_sketches_registry" @@ -54,6 +67,19 @@ serde = { workspace = true } serde_json = { workspace = true } humantime-serde = { workspace = true } -otap-df-engine = { workspace = true } -otap-df-otap = { workspace = true } -otap-df-config = { workspace = true } +otel-arrow-dfe-engine = { workspace = true } +otel-arrow-dfe-otap = { workspace = true } +otel-arrow-dfe-config = { workspace = true } + +# For otap_bridge.rs's OtapMetricRecords <-> real OtapArrowRecords +# conversion: reading real metric batches (otel-arrow-dfe-pdata-views' +# MetricsView trait family) and building them +# (otel-arrow-dfe-pdata::encode::encode_metrics_otap_batch). +otel-arrow-dfe-pdata = { workspace = true } +otel-arrow-dfe-pdata-views = { workspace = true } +thiserror = { workspace = true } + +# Must resolve to the same version asap-precompute-rs's own `arrow-*` +# deps do (both pinned "58.3") — see otap_bridge.rs's module doc. +arrow-array = { workspace = true } +arrow-schema = { workspace = true } diff --git a/otap-patch/all/mod.rs b/otap-patch/all/mod.rs index 2a2f848..797901a 100644 --- a/otap-patch/all/mod.rs +++ b/otap-patch/all/mod.rs @@ -6,8 +6,6 @@ //! //! ## What this file does //! -//! Three things, all small. -//! //! **(1)** Declares an [`OTAP_PROCESSOR_FACTORIES`] entry for //! `urn:asap:processor:asap_sketches`. This is the //! `#[distributed_slice]` static that puts the plugin into the @@ -15,20 +13,19 @@ //! discovers it via `system_info()` at startup (the function that //! produces the binary's "Available Component URNs:" banner). //! -//! **(2)** Implements a minimal [`local::Processor`] -//! adapter — [`AsapSketchesProcessor`] — that bridges OTAP's -//! `OtapPdata` message shape onto Phase C's -//! [`AsapSketchesPlugin`] runtime. Phase D's mandate is to ship the -//! build pipeline + plugin registry entry; the adapter is -//! intentionally a pass-through forward right now — the codec ↔ -//! runtime wiring (Phase C's `OtapMetricRecords::flatten()` / -//! `lift()`) lands as a follow-up because the `OtapPdata` ↔ -//! `OtapMetricRecords` `From` / `Into` adapter was an open question -//! Phase C deferred (a thin `From`/`Into` adapter is added without -//! changing the `flatten()`/`lift()` API). The URN entry is what the -//! registry inspection sees; the adapter only needs to be wireable, -//! not yet semantically complete. Functional end-to-end binding is -//! Phase E (cross-host parity) territory. +//! **(2)** Implements a real [`local::Processor`] adapter — +//! [`AsapSketchesProcessor`] — that ingests real OTAP metric traffic, +//! aggregates it, and emits sketch results back out as real OTAP +//! metric traffic. Per-`Message::PData` and per-timer-tick, it drives +//! a bare `Precompute` instance directly rather than going through +//! [`AsapSketchesPlugin`]'s own Tokio-task/`Stream` lifecycle — see +//! [`create_asap_sketches_processor`]'s doc for why (that lifecycle's +//! current emit shape, `SketchStreamBatch`, diverged from what this +//! adapter needs after PR #5/#6's dictionary-economics work). The +//! actual `OtapPdata` <-> `OtapMetricRecords` conversion lives in +//! `otap_bridge` — **that module has not been build-verified against +//! a real OTAP Dataflow workspace**; see its own module doc for +//! exactly what's confirmed vs. assumed. //! //! **(3)** Validates the user-facing TOML config against //! [`asap_precompute_rs::otap::PluginConfig`]'s shape — the @@ -48,9 +45,23 @@ //! across hosts. A controller plan rendered for one host renders //! identically for the others — no per-platform translation in the //! controller. +//! +//! ## Scope not covered here +//! +//! This adapter only handles the **producer role** — real OTLP +//! metrics in, sketch envelopes (or, in `transmit_sketch = false` +//! estimate mode, quantile/cardinality gauges) out, both as ordinary +//! OTAP metric traffic. The **receiver role** — ingesting another +//! `asap_sketches` node's `SketchStreamBatch` output +//! (`AsapSketchesPlugin::start_from_envelopes`) as `OtapPdata` — needs +//! a different binding (that format doesn't fit OTAP's metrics shape +//! at all) and isn't addressed by this file. #![deny(unsafe_op_in_unsafe_fn)] +mod otap_bridge; + +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -58,21 +69,30 @@ use async_trait::async_trait; use linkme::distributed_slice; use serde::{Deserialize, Serialize}; -use otap_df_config::error::Error as OtapConfigError; -use otap_df_config::node::NodeUserConfig; -use otap_df_engine::config::ProcessorConfig; -use otap_df_engine::context::PipelineContext; -use otap_df_engine::control::NodeControlMsg; -use otap_df_engine::error::Error; -use otap_df_engine::local::processor as local; -use otap_df_engine::message::Message; -use otap_df_engine::node::NodeId; -use otap_df_engine::processor::ProcessorWrapper; -use otap_df_engine::ProcessorFactory; -use otap_df_otap::pdata::OtapPdata; -use otap_df_otap::OTAP_PROCESSOR_FACTORIES; - -use asap_precompute_rs::otap::{AsapSketchesPlugin, PluginConfig}; +// NOTE: `otel_arrow_dfe_*` is the current (2026-08-24) upstream naming +// — see otap_bridge.rs's module doc for the very recent `otap_df_*` +// rename this assumes. Revert to `otap_df_*` throughout this file (and +// otap_bridge.rs) if the actual pinned commit predates it. +use otel_arrow_dfe_config::error::Error as OtapConfigError; +use otel_arrow_dfe_config::node::NodeUserConfig; +use otel_arrow_dfe_engine::config::ProcessorConfig; +use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::NodeControlMsg; +use otel_arrow_dfe_engine::error::Error; +use otel_arrow_dfe_engine::local::processor as local; +use otel_arrow_dfe_engine::message::Message; +use otel_arrow_dfe_engine::node::NodeId; +use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::ProcessorFactory; +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; + +use asap_precompute_rs::config::PrecomputeConfigSet; +use asap_precompute_rs::otap::config::resolve as resolve_plugin_config; +use asap_precompute_rs::otap::{decode_batch, encode_batch, flatten, lift, AsapSketchesPlugin, PluginConfig}; +use asap_precompute_rs::precompute::Precompute; + +use otap_bridge::{otap_metric_records_to_pdata, pdata_to_otap_metric_records}; /// Public URN for the unified ASAP `asap_sketches` processor. Survives /// across hosts unchanged so a controller plan addressed at this URN @@ -161,7 +181,7 @@ impl AsapSketchesUserConfig { pub static ASAP_SKETCHES_PROCESSOR_FACTORY: ProcessorFactory = ProcessorFactory { name: ASAP_SKETCHES_PROCESSOR_URN, create: create_asap_sketches_processor, - wiring_contract: otap_df_engine::wiring_contract::WiringContract::UNRESTRICTED, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, validate_config: validate_asap_sketches_config, }; @@ -179,8 +199,11 @@ fn validate_asap_sketches_config(config: &serde_json::Value) -> Result<(), OtapC /// Factory function — invoked once per pipeline instance at startup. /// Translates the user-supplied TOML into Phase C's [`PluginConfig`], -/// constructs an [`AsapSketchesPlugin`], and wraps it in OTAP's -/// `local::Processor` adapter. +/// resolves it to a `Precompute` instance via +/// [`AsapSketchesPlugin::from_plugin_config`] (reusing its validated +/// construction path), and wraps the bare `Precompute` in OTAP's +/// `local::Processor` adapter — **not** the plugin's own Tokio-task +/// lifecycle (see [`AsapSketchesProcessor`]'s doc for why). pub fn create_asap_sketches_processor( _pipeline_ctx: PipelineContext, node: NodeId, @@ -194,50 +217,113 @@ pub fn create_asap_sketches_processor( } })?; let plugin_config = user.into_plugin_config()?; + let window_size = plugin_config.window_size; - // Construct the plugin synchronously — `from_plugin_config` is - // pure (no Tokio); the plugin's `start()` runs at message time. + // `from_plugin_config` is pure (no Tokio) — it just validates and + // resolves. Grab the `Arc` it constructs and + // discard the plugin wrapper itself: `AsapSketchesPlugin::start()` + // spawns Tokio tasks around a `Stream` + // and currently emits `SketchStreamBatch` (the asap_sketches -> + // asap_sketches wire-transport shape from PR #6, dictionary + // economics) — not the OTAP-Metrics-shaped `OtapMetricRecords` + // this adapter needs for `effect_handler.send_message`. OTAP's + // own per-message `process()` + `effect_handler.start_periodic_timer` + // is a better-fitting host for a callback-driven `Precompute` + // than bridging that Stream-based lifecycle would be. let plugin = AsapSketchesPlugin::from_plugin_config(&plugin_config).map_err(|e| { OtapConfigError::InvalidUserConfig { error: format!("asap_sketches: plugin construction: {e}"), } })?; + let precompute = plugin.precompute().clone(); + drop(plugin); Ok(ProcessorWrapper::local( - AsapSketchesProcessor::new(plugin), + AsapSketchesProcessor::new(precompute, window_size), node, node_config, processor_config, )) } -/// OTAP `local::Processor` adapter for Phase C's -/// [`AsapSketchesPlugin`]. +/// OTAP `local::Processor` adapter — the real +/// `OtapPdata` <-> `OtapMetricRecords` binding (`otap_bridge`) driving +/// a bare [`Precompute`] instance directly, rather than through +/// [`AsapSketchesPlugin`]'s own Tokio-task lifecycle (see +/// [`create_asap_sketches_processor`]'s doc for why: that lifecycle's +/// emit shape and this adapter's needed shape have diverged since +/// PR #5/#6). `Precompute::observe`/`tick`/`drain` are themselves +/// callback-style, not stream-based, so driving them directly from +/// OTAP's own per-message/per-timer `process()` calls needs no +/// bridging machinery at all. /// -/// **Phase D scope deliberate:** this adapter forwards `OtapPdata` -/// messages downstream unchanged. The `OtapPdata` ↔ -/// `OtapMetricRecords` `From` / `Into` binding is an open question -/// Phase C deferred (a thin `From`/`Into` adapter that leaves the -/// `flatten()`/`lift()` API unchanged) and will land in a follow-up -/// alongside the cross-host parity test (Phase E). What Phase D -/// delivers here is the registration that brings `asap_sketches` -/// into the binary's plugin registry. +/// # Verification status /// -/// The adapter holds the plugin instance so the `OtapPdata` -/// translation can be hung off `process()` in the follow-up without -/// ABI churn. +/// See `otap_bridge.rs`'s module doc — the `OtapPdata` conversions +/// this adapter calls have not been build-verified against a real +/// OTAP Dataflow workspace. pub struct AsapSketchesProcessor { - /// Constructed plugin. Wrapped in `Option` so a future graceful - /// shutdown path can take ownership of it for the final drain. - _plugin: Option, + precompute: Arc, + window_size: Duration, + /// `effect_handler.start_periodic_timer` is `async` and OTAP has + /// no dedicated "processor started" hook — armed on the first + /// `process()` call instead (`Message::PData` or any control + /// message) rather than at construction time. + timer_started: bool, + /// `PrecomputeConfigSet::version` for the next `NodeControlMsg::Config` + /// this processor applies — monotonically increasing, independent + /// of any external controller's own versioning since this + /// adapter's `Precompute` never talks to a `ControlChannel`. + next_config_version: Arc, } impl AsapSketchesProcessor { - fn new(plugin: AsapSketchesPlugin) -> Self { + fn new(precompute: Arc, window_size: Duration) -> Self { Self { - _plugin: Some(plugin), + precompute, + window_size, + timer_started: false, + next_config_version: Arc::new(AtomicU64::new(1)), } } + + /// Encodes and emits one window's worth of envelopes, if any — + /// shared by the `TimerTick` (regular flush) and `Shutdown` + /// (final drain) paths. `envs` empty is a no-op, matching + /// `Precompute::tick`/`drain`'s own "nothing to flush" contract. + async fn emit_envelopes( + &self, + envs: &[asap_precompute_rs::envelope::SketchEnvelope], + effect_handler: &mut local::EffectHandler, + ) -> Result<(), Error> { + use otel_arrow_dfe_engine::MessageSourceLocalEffectHandlerExtension as _; + + if envs.is_empty() { + return Ok(()); + } + // The documented Phase-B path: SketchEnvelopes -> flat + // RecordBatch (encode_batch) -> OTAP-validator-safe two-batch + // family (lift) -> real OtapPdata (otap_bridge). Deliberately + // NOT `SeriesDictionary::encode` (the SCHEMA/DICTIONARY/RECORD + // wire economics from PR #5/#6) — that format is for the + // asap_sketches -> asap_sketches transport hop + // (`otap::wire`), not for riding inside an arbitrary OTAP + // pipeline as a generic metric. + let flat = match encode_batch(envs) { + Ok(flat) => flat, + Err(_e) => return Ok(()), // drop the bad window, keep the processor alive + }; + let records = match lift(&flat) { + Ok(records) => records, + Err(_e) => return Ok(()), + }; + let pdata = match otap_metric_records_to_pdata(&records) { + Ok(pdata) => pdata, + Err(_e) => return Ok(()), + }; + effect_handler.send_message_with_source_node(pdata).await?; + Ok(()) + } } #[async_trait(?Send)] @@ -247,24 +333,125 @@ impl local::Processor for AsapSketchesProcessor { msg: Message, effect_handler: &mut local::EffectHandler, ) -> Result<(), Error> { - use otap_df_engine::MessageSourceLocalEffectHandlerExtension as _; + if !self.timer_started { + self.timer_started = true; + // Best-effort: if this fails, the processor still ingests + // and observes correctly, it just never flushes a window + // on its own — degraded, not broken. `TimerCancelHandle` + // is dropped immediately: the timer keeps firing for this + // processor's lifetime rather than being cancellable + // (there's currently no shutdown-adjacent place to hold + // the handle across `process()` calls without adding + // another `Option<...>` field for a cancellation this + // adapter never actually exercises). + let _ = effect_handler.start_periodic_timer(self.window_size).await; + } + match msg { Message::PData(pdata) => { - // Phase D pass-through; codec wiring lands as a - // follow-up via the From/Into adapter open question. - // The runtime is constructed and ready to observe — - // see `_plugin` field. - effect_handler.send_message_with_source_node(pdata).await?; + let outcome = match pdata_to_otap_metric_records(pdata) { + Ok(outcome) => outcome, + Err(_e) => return Ok(()), // drop the bad batch, keep the processor alive + }; + if outcome.skipped_non_scalar > 0 { + effect_handler + .info(&format!( + "asap_sketches: skipped {} non-scalar (histogram/exponential-histogram/summary) data point(s) — only Gauge/Sum are aggregated", + outcome.skipped_non_scalar + )) + .await; + } + let Some(records) = outcome.records else { + return Ok(()); + }; + let flat = match flatten(&records) { + Ok(flat) => flat, + Err(_e) => return Ok(()), + }; + let observations = match decode_batch(&flat) { + Ok(observations) => observations, + Err(_e) => return Ok(()), + }; + for obs in &observations { + // Note: `observe` here handles both "genuine OTLP + // metric" and "sketch shipped as binary inside an + // OTAP metric" (an upstream asap_sketches node's + // `_asap_envelope`-tagged output) — no branching + // needed at this call site. `decode_batch` already + // tags the latter as `ObservationValueKind::Envelope`, + // and `Precompute::observe` already routes those + // internally to `observe_envelope` (merge as a + // pre-aggregated sketch) instead of expanding them + // to scalar samples. See otap_bridge.rs's module + // doc ("Scope") for the full picture. + // + // LateData / SeriesCapExceeded are expected, + // already-tallied-in-stats outcomes (mirrors + // `AsapSketchesPlugin`'s own ingest policy, + // lifecycle.rs's `ingest_one_batch`) — silent by + // design. Anything else (e.g. NoConfig, + // AggIdMismatch) indicates a real misconfiguration + // and is at least surfaced via `effect_handler.info` + // — a real error channel is follow-up work, but + // this keeps it from being completely invisible. + use asap_precompute_rs::precompute::PrecomputeError; + match self.precompute.observe(obs) { + Ok(()) + | Err(PrecomputeError::LateData) + | Err(PrecomputeError::SeriesCapExceeded) => {} + Err(e) => { + effect_handler + .info(&format!("asap_sketches: observe failed: {e}")) + .await; + } + } + } + Ok(()) + } + Message::Control(NodeControlMsg::TimerTick { .. }) => { + let now_ms = asap_wall_clock_ms(); + let envs = self.precompute.tick(now_ms); + self.emit_envelopes(&envs, effect_handler).await + } + Message::Control(NodeControlMsg::Shutdown { .. }) => { + let envs = self.precompute.drain(); + self.emit_envelopes(&envs, effect_handler).await + } + Message::Control(NodeControlMsg::Config { config }) => { + let user: AsapSketchesUserConfig = match serde_json::from_value(config) { + Ok(user) => user, + Err(_e) => return Ok(()), // malformed plan push — keep running on the old config + }; + let Ok(plugin_config) = user.into_plugin_config() else { + return Ok(()); + }; + let Ok((pcfg, _dispatch)) = resolve_plugin_config(&plugin_config) else { + return Ok(()); + }; + let version = self.next_config_version.fetch_add(1, Ordering::Relaxed); + self.precompute.update_config(&PrecomputeConfigSet { + version, + configs: vec![pcfg], + }); Ok(()) } - Message::Control(NodeControlMsg::Shutdown { .. }) => Ok(()), - Message::Control(NodeControlMsg::Config { .. }) => Ok(()), Message::Control(NodeControlMsg::CollectTelemetry { .. }) => Ok(()), _ => Ok(()), } } } +/// Wall-clock millisecond timestamp — mirrors +/// `asap_precompute_rs::otap::lifecycle`'s private `wall_clock_ms` +/// (not exported for reuse across the crate boundary). +fn asap_wall_clock_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + #[cfg(test)] mod tests { //! Unit tests for the registration crate. Confined to config-shape diff --git a/otap-patch/all/otap_bridge.rs b/otap-patch/all/otap_bridge.rs new file mode 100644 index 0000000..1c0b2c6 --- /dev/null +++ b/otap-patch/all/otap_bridge.rs @@ -0,0 +1,1280 @@ +// Copyright The ASAP Authors +// SPDX-License-Identifier: MIT + +//! Bridge between ASAP's flat, host-neutral [`OtapMetricRecords`] (the +//! shape `asap_precompute_rs::otap::records::flatten`/`decode_batch` +//! already know how to walk) and OTAP's real `OtapPdata` / +//! `OtapArrowRecords::Metrics` shape — the "one seam left" the crate +//! root README calls out. +//! +//! This is the piece that actually puts a self-describing sketch's +//! serialized bytes onto the wire *as an OTAP metric*: `encode_batch` +//! (already implemented, unchanged by this file) turns a +//! `SketchEnvelope` into an `OtapMetricRecords` whose per-row +//! attribute batch carries the envelope bytes as a `_asap_envelope` +//! `Bytes`-typed attribute (Strategy B, `otap/schema.rs`); this module +//! is the layer above that turns *that* into something OTAP's real +//! engine can carry as `Message::PData(OtapPdata)`. +//! +//! # Provenance / verification status +//! +//! Written against upstream `open-telemetry/otel-arrow` commit +//! `3e85c3460361446ebfce99e9f35fffd2dd5ab740` (2026-08-24), reading +//! `rust/otap-dataflow/crates/pdata/src/{payload.rs,otap.rs,lib.rs, +//! encode/mod.rs}` and `rust/otap-dataflow/crates/pdata-views/src/ +//! views/{metrics.rs,common.rs,resource.rs}` directly for every type +//! and function signature referenced below. The OTAP Dataflow crates +//! were renamed `otap_df_*` -> `otel_arrow_dfe_*` very recently and +//! unreleased as of that commit (`.chloggen/otel-arrow-dfe-crate- +//! prefix.yaml`, issue #1848); this file targets the new names. +//! +//! **This file has not been build-verified against a real OTAP +//! Dataflow workspace checkout** — `otap-patch/` isn't standalone- +//! buildable in the environment this was written in (see the repo +//! README's "No — depends on the OTAP workspace crates" note on this +//! directory). Treat every signature here as "read from source, not +//! compiled" and expect a build pass to shake out mismatches against +//! whatever commit this repo's own build script actually pins. +//! +//! # Scope +//! +//! What "handling a metric" means on decode splits into two cases, +//! decided by content, not by which OTLP metric type carried it: +//! +//! - **A data point carrying `_asap_envelope`** (this module's own +//! encode output, or any other `asap_sketches` node's) — this +//! module doesn't special-case it at all. It round-trips into +//! `OtapMetricRecords` like any other attribute, and +//! `decode_batch` (unchanged, downstream of this module) already +//! recognizes `_asap_envelope` and produces an +//! `ObservationValueKind::Envelope`-kind `Observation`; +//! `Precompute::observe` already dispatches those internally to +//! `observe_envelope` (merge as a pre-aggregated sketch), never +//! expanding them to scalar samples. So "sketch as a binary inside +//! an OTAP metric" already gets sketch-side handling for free, by +//! construction, with no branching needed here. +//! - **A genuine (non-envelope) OTLP metric** — real telemetry. +//! Gauge/Sum (`NumberDataPoints`) are handled normally: each data +//! point becomes a scalar `Observation`, the shape both +//! `Observation` and `OtapMetricRecords` (well-known +//! `time_unix_nano`/`metric`/`value` columns) already assume. +//! Histogram / ExponentialHistogram / Summary data points are +//! skipped (counted, not silently dropped — see +//! [`DecodeOutcome::skipped_non_scalar`]) rather than expanded, +//! since there's no single well-defined scalar to extract from a +//! bucket/quantile set without picking a lossy expansion strategy +//! this module doesn't want to own. +//! +//! On encode, every row is assumed to share one metric name — an +//! `AsapSketchesProcessor` instance has exactly one +//! `PluginConfig::output_metric_name`, so this holds by construction +//! for anything `encode_batch` itself produces; [`otap_metric_records_to_pdata`] +//! still checks it explicitly and errors loudly on a real mismatch +//! rather than silently dropping rows for the "wrong" metric. +//! +//! Resource and scope are not modelled — [`OtapMetricRecords`] itself +//! deliberately omits them (see that type's own doc); every emitted +//! metric attaches to an empty `Resource` / unnamed `Scope`. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use arrow_array::{Array, BinaryArray, Float64Array, RecordBatch, StringArray, UInt32Array, UInt64Array}; +use arrow_schema::{DataType, Field, Schema}; +use thiserror::Error; + +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_pdata::encode::encode_metrics_otap_batch; +use otel_arrow_dfe_pdata::otap::OtapArrowRecords; +use otel_arrow_dfe_pdata::views::otap::OtapMetricsView; +use otel_arrow_dfe_pdata::{OtapPayload, TryFromWithOptions, TryIntoWithOptions}; +use otel_arrow_dfe_pdata_views::views::common::{ + AnyValueView, AttributeView, InstrumentationScopeView, Str, ValueType, +}; +use otel_arrow_dfe_pdata_views::views::metrics::{ + AggregationTemporality, BucketsView, DataPointFlags, DataType as MetricKind, DataView, + ExemplarView, ExponentialHistogramDataPointView, ExponentialHistogramView, GaugeView, + HistogramDataPointView, HistogramView, MetricView, MetricsView, NumberDataPointView, + ResourceMetricsView, ScopeMetricsView, SumView, SummaryDataPointView, SummaryView, + Value as DpValue, ValueAtQuantileView, +}; +use otel_arrow_dfe_pdata_views::views::resource::ResourceView; + +use asap_precompute_rs::otap::records::{ + ATTR_BATCH_BYTES, ATTR_BATCH_INT, ATTR_BATCH_KEY, ATTR_BATCH_PARENT_ID, ATTR_BATCH_STR, +}; +use asap_precompute_rs::otap::{ + OtapMetricRecords, COLUMN_METRIC, COLUMN_TIME_UNIX_NANO, COLUMN_VALUE, +}; + +/// Failure modes for [`otap_metric_records_to_pdata`] / [`pdata_to_otap_metric_records`]. +#[derive(Debug, Error)] +pub enum BridgeError { + /// A required column was missing or had the wrong Arrow type. + #[error("otap bridge: column {column:?} on batch {batch:?} missing or wrong type")] + BadColumn { + /// Which sibling batch. + batch: &'static str, + /// Column name. + column: &'static str, + }, + /// More than one distinct `metric` name appeared in one + /// `OtapMetricRecords.metrics` batch — an + /// `AsapSketchesProcessor` instance has exactly one + /// `output_metric_name`, so this indicates the caller fed rows + /// from more than one processor instance into a single call. + #[error( + "otap bridge: one OtapMetricRecords batch carries more than one metric name ({first:?} and {second:?} seen)" + )] + MixedMetricNames { + /// First metric name seen. + first: String, + /// A second, different metric name seen in the same batch. + second: String, + }, + /// Building the real `OtapArrowRecords::Metrics` batch failed. + #[error("otap bridge: encoding real OTAP metrics batch failed: {0}")] + Encode(String), + /// Reading the real `OtapArrowRecords::Metrics` batch failed. + #[error("otap bridge: reading real OTAP metrics batch failed: {0}")] + Decode(String), + /// Constructing the flat output `RecordBatch` failed. + #[error("otap bridge: arrow record-batch construction failed: {0}")] + Arrow(#[from] arrow_schema::ArrowError), +} + +// ============================================================================ +// Encode direction: OtapMetricRecords -> OtapPdata +// ============================================================================ + +/// Converts an [`OtapMetricRecords`] (as `encode_batch` produces it) +/// into a real `OtapPdata` carrying an `OtapArrowRecords::Metrics` +/// payload — the actual "put the sketch envelope bytes onto an OTAP +/// metric" step. +/// +/// Builds a fresh, contextless `OtapPdata` (`OtapPdata::new_todo_context`) +/// rather than propagating any single input message's context: the +/// plugin's flush ticker emits one window's worth of envelopes on its +/// own wall-clock schedule, decoupled from any specific triggering +/// input message, so there is no single Ack/Nack chain to attach the +/// output to (the same reasoning any periodic/windowed aggregator's +/// output would follow). +pub fn otap_metric_records_to_pdata(records: &OtapMetricRecords) -> Result { + let view = AsapMetricsView::try_new(records)?; + let arrow_records = + encode_metrics_otap_batch(&view).map_err(|e| BridgeError::Encode(e.to_string()))?; + let payload: OtapPayload = arrow_records.into(); + Ok(OtapPdata::new_todo_context(payload)) +} + +/// Zero-copy-ish adapter presenting an [`OtapMetricRecords`]'s two +/// flat batches as a `MetricsView` — one Resource (empty), one Scope +/// (unnamed), one Metric (`records.metrics`'s single distinct `metric` +/// value), one Gauge, N `NumberDataPoint`s (one per row). +struct AsapMetricsView { + metric_name: String, + n_rows: usize, + time_col: UInt64Array, + value_col: Float64Array, + /// `attr_index[parent_id]` = row indices into `records.attributes` + /// carrying that parent's attributes. Precomputed once so per-data-point + /// `attributes()` calls don't rescan the whole attribute batch. + attr_index: BTreeMap>, + attr_key_col: StringArray, + attr_bytes_col: Option, + attr_str_col: Option, + attr_int_col: Option, +} + +impl<'a> AsapMetricsView { + fn try_new(records: &'a OtapMetricRecords) -> Result { + let n_rows = records.metrics.num_rows(); + + let metric_col = require_string(&records.metrics, "metrics", COLUMN_METRIC)?; + let mut metric_name: Option = None; + for row in 0..n_rows { + let name = metric_col.value(row); + match &metric_name { + None => metric_name = Some(name.to_string()), + Some(seen) if seen == name => {} + Some(seen) => { + return Err(BridgeError::MixedMetricNames { + first: seen.clone(), + second: name.to_string(), + }); + } + } + } + + let time_col = require_uint64(&records.metrics, "metrics", COLUMN_TIME_UNIX_NANO)?.clone(); + let value_col = require_float64(&records.metrics, "metrics", COLUMN_VALUE)?.clone(); + let parent_col = require_uint32(&records.metrics, "metrics", ATTR_BATCH_PARENT_ID)?; + + let attr_parent_col = require_uint32(&records.attributes, "attributes", ATTR_BATCH_PARENT_ID)?; + let attr_key_col = require_string(&records.attributes, "attributes", ATTR_BATCH_KEY)?.clone(); + let attr_bytes_col = optional_binary(&records.attributes, ATTR_BATCH_BYTES)?; + let attr_str_col = optional_string(&records.attributes, ATTR_BATCH_STR)?; + let attr_int_col = optional_uint64(&records.attributes, ATTR_BATCH_INT)?; + + let mut attr_index: BTreeMap> = BTreeMap::new(); + for row in 0..records.attributes.num_rows() { + if attr_parent_col.is_null(row) { + continue; + } + attr_index + .entry(attr_parent_col.value(row)) + .or_default() + .push(row); + } + // Not currently used beyond validating the column exists — + // parent_id on `metrics` isn't needed for row->attribute + // linking here (each metrics row's own index doubles as its + // parent_id by `encode_batch`'s own convention), but keep the + // column checked so a future divergence surfaces as a + // decode-time BadColumn error rather than silent misjoin. + let _ = parent_col; + + Ok(Self { + metric_name: metric_name.unwrap_or_default(), + n_rows, + time_col, + value_col, + attr_index, + attr_key_col, + attr_bytes_col, + attr_str_col, + attr_int_col, + }) + } + + fn attributes_for_row(&self, parent_id: u32) -> Vec> { + let Some(rows) = self.attr_index.get(&parent_id) else { + return Vec::new(); + }; + rows.iter() + .filter_map(|&row| { + let key = self.attr_key_col.value(row); + let value = attr_row_value( + self.attr_bytes_col.as_ref(), + self.attr_str_col.as_ref(), + self.attr_int_col.as_ref(), + row, + )?; + Some(AsapAttribute { key, value }) + }) + .collect() + } +} + +impl MetricsView for AsapMetricsView { + type ResourceMetrics<'res> + = AsapResourceMetricsView<'res> + where + Self: 'res; + type ResourceMetricsIter<'res> + = std::vec::IntoIter> + where + Self: 'res; + + fn resources(&self) -> Self::ResourceMetricsIter<'_> { + if self.n_rows == 0 { + Vec::new().into_iter() + } else { + vec![AsapResourceMetricsView { view: self }].into_iter() + } + } +} + +struct AsapResourceMetricsView<'a> { + view: &'a AsapMetricsView, +} + +impl<'a> ResourceMetricsView for AsapResourceMetricsView<'a> { + type Resource<'res> + = AsapNoResource + where + Self: 'res; + type ScopeMetrics<'scp> + = AsapScopeMetricsView<'scp> + where + Self: 'scp; + type ScopesIter<'scp> + = std::vec::IntoIter> + where + Self: 'scp; + + fn resource(&self) -> Option> { + None + } + + fn scopes(&self) -> Self::ScopesIter<'_> { + vec![AsapScopeMetricsView { view: self.view }].into_iter() + } + + fn schema_url(&self) -> Option> { + None + } +} + +struct AsapScopeMetricsView<'a> { + view: &'a AsapMetricsView, +} + +impl<'a> ScopeMetricsView for AsapScopeMetricsView<'a> { + type Scope<'scp> + = AsapNoScope + where + Self: 'scp; + type Metric<'met> + = AsapMetricView<'met> + where + Self: 'met; + type MetricIter<'met> + = std::vec::IntoIter> + where + Self: 'met; + + fn scope(&self) -> Option> { + None + } + + fn metrics(&self) -> Self::MetricIter<'_> { + vec![AsapMetricView { view: self.view }].into_iter() + } + + fn schema_url(&self) -> Str<'_> { + b"" + } +} + +struct AsapMetricView<'a> { + view: &'a AsapMetricsView, +} + +impl<'a> MetricView for AsapMetricView<'a> { + type Data<'dat> + = AsapDataView<'dat> + where + Self: 'dat; + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + + fn name(&self) -> Str<'_> { + self.view.metric_name.as_bytes() + } + + fn description(&self) -> Str<'_> { + b"" + } + + fn unit(&self) -> Str<'_> { + b"" + } + + fn data(&self) -> Option> { + Some(AsapDataView { view: self.view }) + } + + fn metadata(&self) -> Self::AttributeIter<'_> { + Vec::new().into_iter() + } +} + +struct AsapDataView<'a> { + view: &'a AsapMetricsView, +} + +impl<'a> DataView<'a> for AsapDataView<'a> { + type Gauge<'gauge> + = AsapGaugeView<'gauge> + where + Self: 'gauge; + type Sum<'sum> + = AsapNoSum + where + Self: 'sum; + type Histogram<'histogram> + = AsapNoHistogram + where + Self: 'histogram; + type ExponentialHistogram<'exp> + = AsapNoExpHistogram + where + Self: 'exp; + type Summary<'summary> + = AsapNoSummary + where + Self: 'summary; + + fn value_type(&self) -> MetricKind { + MetricKind::Gauge + } + + fn as_gauge(&self) -> Option> { + Some(AsapGaugeView { view: self.view }) + } + + fn as_sum(&self) -> Option> { + None + } + + fn as_histogram(&self) -> Option> { + None + } + + fn as_exponential_histogram(&self) -> Option> { + None + } + + fn as_summary(&self) -> Option> { + None + } +} + +struct AsapGaugeView<'a> { + view: &'a AsapMetricsView, +} + +impl<'a> GaugeView for AsapGaugeView<'a> { + type NumberDataPoint<'dp> + = AsapNumberDataPointView<'dp> + where + Self: 'dp; + type NumberDataPointIter<'dp> + = std::vec::IntoIter> + where + Self: 'dp; + + fn data_points(&self) -> Self::NumberDataPointIter<'_> { + (0..self.view.n_rows) + .map(|row| AsapNumberDataPointView { + view: self.view, + row, + }) + .collect::>() + .into_iter() + } +} + +struct AsapNumberDataPointView<'a> { + view: &'a AsapMetricsView, + row: usize, +} + +impl<'a> NumberDataPointView for AsapNumberDataPointView<'a> { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + type Exemplar<'ex> + = AsapNoExemplar + where + Self: 'ex; + type ExemplarIter<'ex> + = std::vec::IntoIter + where + Self: 'ex; + + fn start_time_unix_nano(&self) -> u64 { + 0 + } + + fn time_unix_nano(&self) -> u64 { + self.view.time_col.value(self.row) + } + + fn value(&self) -> Option { + if self.view.value_col.is_null(self.row) { + None + } else { + Some(DpValue::Double(self.view.value_col.value(self.row))) + } + } + + fn attributes(&self) -> Self::AttributeIter<'_> { + // `encode_batch`'s own convention: the metrics row's ordinal + // index doubles as the parent_id its attribute rows join + // against (see records.rs's `ATTR_BATCH_PARENT_ID` doc). + self.view + .attributes_for_row(self.row as u32) + .into_iter() + } + + fn exemplars(&self) -> Self::ExemplarIter<'_> { + Vec::new().into_iter() + } + + fn flags(&self) -> DataPointFlags { + DataPointFlags::new(0) + } +} + +/// One typed attribute value ASAP's flat attribute batch can carry — +/// mirrors `records.rs`'s internal `AttrValue` three-way union +/// (`bytes` / `str` / `int` columns). +enum AsapAnyValue<'a> { + Str(&'a str), + Int(u64), + Bytes(&'a [u8]), +} + +struct AsapAttribute<'a> { + key: &'a str, + value: AsapAnyValue<'a>, +} + +impl<'a> AttributeView for AsapAttribute<'a> { + type Val<'val> + = AsapAnyValueView<'val> + where + Self: 'val; + + fn key(&self) -> Str<'_> { + self.key.as_bytes() + } + + fn value(&self) -> Option> { + Some(AsapAnyValueView(&self.value)) + } +} + +struct AsapAnyValueView<'a>(&'a AsapAnyValue<'a>); + +impl<'a> AnyValueView<'a> for AsapAnyValueView<'a> { + type KeyValue = AsapAttribute<'a>; + type ArrayIter<'arr> + = std::iter::Empty + where + Self: 'arr; + type KeyValueIter<'kv> + = std::iter::Empty + where + Self: 'kv; + + fn value_type(&self) -> ValueType { + match self.0 { + AsapAnyValue::Str(_) => ValueType::String, + AsapAnyValue::Int(_) => ValueType::Int64, + AsapAnyValue::Bytes(_) => ValueType::Bytes, + } + } + + fn as_string(&self) -> Option> { + match self.0 { + AsapAnyValue::Str(s) => Some(s.as_bytes()), + _ => None, + } + } + + fn as_bool(&self) -> Option { + None + } + + fn as_int64(&self) -> Option { + match self.0 { + // Lossy above i64::MAX, which ASAP never actually + // produces here (the only `_asap_*` int-typed attributes + // are small counters/timestamps well under that bound). + AsapAnyValue::Int(v) => Some(*v as i64), + _ => None, + } + } + + fn as_double(&self) -> Option { + None + } + + fn as_bytes(&self) -> Option<&[u8]> { + match self.0 { + AsapAnyValue::Bytes(b) => Some(b), + _ => None, + } + } + + fn as_array(&self) -> Option> { + None + } + + fn as_kvlist(&self) -> Option> { + None + } +} + +/// Reads the value at `row` from whichever of the three typed-value +/// columns actually has a non-null entry there — mirrors +/// `records.rs`'s private `build_attr_index` join logic (bytes, then +/// str, then int; exactly one is non-null per row by construction). +fn attr_row_value<'a>( + bytes: Option<&'a BinaryArray>, + strs: Option<&'a StringArray>, + ints: Option<&'a UInt64Array>, + row: usize, +) -> Option> { + if let Some(arr) = bytes { + if !arr.is_null(row) { + return Some(AsapAnyValue::Bytes(arr.value(row))); + } + } + if let Some(arr) = strs { + if !arr.is_null(row) { + return Some(AsapAnyValue::Str(arr.value(row))); + } + } + if let Some(arr) = ints { + if !arr.is_null(row) { + return Some(AsapAnyValue::Int(arr.value(row))); + } + } + None +} + +// -- Uninhabited placeholder types ------------------------------------------- +// +// `resource()`/`scope()` always return `None`, and only Gauge is ever +// produced (`as_sum`/`as_histogram`/`as_exponential_histogram`/ +// `as_summary` always return `None`, and there are no exemplars) — but +// the view traits still require *some* concrete, well-formed type for +// each associated type regardless of whether an instance is ever +// constructed. An uninhabited enum (`enum X {}`) lets every trait +// method be `match *self {}` — valid because no value of an +// uninhabited type can ever exist to call it on, so the body coerces +// to any return type without ever actually running. + +enum AsapNoResource {} +impl ResourceView for AsapNoResource { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributesIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + fn attributes(&self) -> Self::AttributesIter<'_> { + match *self {} + } + fn dropped_attributes_count(&self) -> u32 { + match *self {} + } +} + +enum AsapNoScope {} +impl InstrumentationScopeView for AsapNoScope { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + fn name(&self) -> Option> { + match *self {} + } + fn version(&self) -> Option> { + match *self {} + } + fn attributes(&self) -> Self::AttributeIter<'_> { + match *self {} + } + fn dropped_attributes_count(&self) -> u32 { + match *self {} + } +} + +enum AsapNoExemplar {} +impl ExemplarView for AsapNoExemplar { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + fn filtered_attributes(&self) -> Self::AttributeIter<'_> { + match *self {} + } + fn time_unix_nano(&self) -> u64 { + match *self {} + } + fn value(&self) -> Option { + match *self {} + } + fn span_id(&self) -> Option<&otel_arrow_dfe_pdata_views::SpanId> { + match *self {} + } + fn trace_id(&self) -> Option<&otel_arrow_dfe_pdata_views::TraceId> { + match *self {} + } +} + +enum AsapNoSum {} +impl SumView for AsapNoSum { + type NumberDataPoint<'dp> + = AsapNumberDataPointView<'dp> + where + Self: 'dp; + type NumberDataPointIter<'dp> + = std::vec::IntoIter> + where + Self: 'dp; + fn data_points(&self) -> Self::NumberDataPointIter<'_> { + match *self {} + } + fn aggregation_temporality(&self) -> AggregationTemporality { + match *self {} + } + fn is_monotonic(&self) -> bool { + match *self {} + } +} + +enum AsapNoHistogram {} +impl HistogramView for AsapNoHistogram { + type HistogramDataPoint<'dp> + = AsapNoHistogramDataPoint + where + Self: 'dp; + type HistogramDataPointIter<'dp> + = std::vec::IntoIter + where + Self: 'dp; + fn data_points(&self) -> Self::HistogramDataPointIter<'_> { + match *self {} + } + fn aggregation_temporality(&self) -> AggregationTemporality { + match *self {} + } +} + +enum AsapNoHistogramDataPoint {} +impl HistogramDataPointView for AsapNoHistogramDataPoint { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + type BucketCountIter<'bc> + = std::iter::Empty + where + Self: 'bc; + type ExplicitBoundsIter<'eb> + = std::iter::Empty + where + Self: 'eb; + type Exemplar<'ex> + = AsapNoExemplar + where + Self: 'ex; + type ExemplarIter<'ex> + = std::vec::IntoIter + where + Self: 'ex; + fn attributes(&self) -> Self::AttributeIter<'_> { + match *self {} + } + fn start_time_unix_nano(&self) -> u64 { + match *self {} + } + fn time_unix_nano(&self) -> u64 { + match *self {} + } + fn count(&self) -> u64 { + match *self {} + } + fn sum(&self) -> Option { + match *self {} + } + fn bucket_counts(&self) -> Self::BucketCountIter<'_> { + match *self {} + } + fn explicit_bounds(&self) -> Self::ExplicitBoundsIter<'_> { + match *self {} + } + fn exemplars(&self) -> Self::ExemplarIter<'_> { + match *self {} + } + fn flags(&self) -> DataPointFlags { + match *self {} + } + fn min(&self) -> Option { + match *self {} + } + fn max(&self) -> Option { + match *self {} + } +} + +enum AsapNoExpHistogram {} +impl ExponentialHistogramView for AsapNoExpHistogram { + type ExponentialHistogramDataPoint<'edp> + = AsapNoExpHistogramDataPoint + where + Self: 'edp; + type ExponentialHistogramDataPointIter<'edp> + = std::vec::IntoIter + where + Self: 'edp; + fn data_points(&self) -> Self::ExponentialHistogramDataPointIter<'_> { + match *self {} + } + fn aggregation_temporality(&self) -> AggregationTemporality { + match *self {} + } +} + +enum AsapNoExpHistogramDataPoint {} +impl ExponentialHistogramDataPointView for AsapNoExpHistogramDataPoint { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + type Buckets<'b> + = AsapNoBuckets + where + Self: 'b; + type Exemplar<'ex> + = AsapNoExemplar + where + Self: 'ex; + type ExemplarIter<'ex> + = std::vec::IntoIter + where + Self: 'ex; + fn attributes(&self) -> Self::AttributeIter<'_> { + match *self {} + } + fn start_time_unix_nano(&self) -> u64 { + match *self {} + } + fn time_unix_nano(&self) -> u64 { + match *self {} + } + fn count(&self) -> u64 { + match *self {} + } + fn sum(&self) -> Option { + match *self {} + } + fn scale(&self) -> i32 { + match *self {} + } + fn zero_count(&self) -> u64 { + match *self {} + } + fn positive(&self) -> Option> { + match *self {} + } + fn negative(&self) -> Option> { + match *self {} + } + fn flags(&self) -> DataPointFlags { + match *self {} + } + fn exemplars(&self) -> Self::ExemplarIter<'_> { + match *self {} + } + fn min(&self) -> Option { + match *self {} + } + fn max(&self) -> Option { + match *self {} + } + fn zero_threshold(&self) -> f64 { + match *self {} + } +} + +enum AsapNoBuckets {} +impl BucketsView for AsapNoBuckets { + type BucketCountIter<'bc> + = std::iter::Empty + where + Self: 'bc; + fn offset(&self) -> i32 { + match *self {} + } + fn bucket_counts(&self) -> Self::BucketCountIter<'_> { + match *self {} + } +} + +enum AsapNoSummary {} +impl SummaryView for AsapNoSummary { + type SummaryDataPoint<'dp> + = AsapNoSummaryDataPoint + where + Self: 'dp; + type SummaryDataPointIter<'dp> + = std::vec::IntoIter + where + Self: 'dp; + fn data_points(&self) -> Self::SummaryDataPointIter<'_> { + match *self {} + } +} + +enum AsapNoSummaryDataPoint {} +impl SummaryDataPointView for AsapNoSummaryDataPoint { + type Attribute<'att> + = AsapAttribute<'att> + where + Self: 'att; + type AttributeIter<'att> + = std::vec::IntoIter> + where + Self: 'att; + type ValueAtQuantile<'vaq> + = AsapNoValueAtQuantile + where + Self: 'vaq; + type ValueAtQuantileIter<'vaq> + = std::vec::IntoIter + where + Self: 'vaq; + fn attributes(&self) -> Self::AttributeIter<'_> { + match *self {} + } + fn start_time_unix_nano(&self) -> u64 { + match *self {} + } + fn time_unix_nano(&self) -> u64 { + match *self {} + } + fn count(&self) -> u64 { + match *self {} + } + fn sum(&self) -> f64 { + match *self {} + } + fn quantile_values(&self) -> Self::ValueAtQuantileIter<'_> { + match *self {} + } + fn flags(&self) -> DataPointFlags { + match *self {} + } +} + +enum AsapNoValueAtQuantile {} +impl ValueAtQuantileView for AsapNoValueAtQuantile { + fn quantile(&self) -> f64 { + match *self {} + } + fn value(&self) -> f64 { + match *self {} + } +} + +// ============================================================================ +// Decode direction: OtapPdata -> OtapMetricRecords +// ============================================================================ + +/// Outcome of [`pdata_to_otap_metric_records`]: the reconstructed flat +/// batch pair (`None` if `pdata` carried a non-Metrics signal or truly +/// zero metric rows), plus a count of rows this call had to skip. +pub struct DecodeOutcome { + /// `None` when `pdata` wasn't Metrics, or decoded to zero rows. + pub records: Option, + /// Histogram / ExponentialHistogram / Summary data points seen and + /// skipped — not silently dropped, see this module's doc "Scope" + /// section for why they aren't expanded. + pub skipped_non_scalar: usize, +} + +/// Converts a real `OtapPdata` into ASAP's flat [`OtapMetricRecords`] +/// shape (feeding `decode_batch` on the producer role's ingest path). +/// +/// Accepts `pdata` by value (consumed) — the caller's `Message::PData` +/// match arm already owns it and has no further use for the original +/// context once this conversion runs. +pub fn pdata_to_otap_metric_records(pdata: OtapPdata) -> Result { + let (_context, payload) = pdata.into_parts(); + let arrow_records: OtapArrowRecords = payload + .try_into_with_default() + .map_err(|e| BridgeError::Decode(format!("{e}")))?; + + let OtapArrowRecords::Metrics(_) = &arrow_records else { + return Ok(DecodeOutcome { + records: None, + skipped_non_scalar: 0, + }); + }; + + let view = + OtapMetricsView::try_from(&arrow_records).map_err(|e| BridgeError::Decode(format!("{e}")))?; + + let mut time_unix_nano: Vec = Vec::new(); + let mut metric_names: Vec = Vec::new(); + let mut values: Vec = Vec::new(); + let mut attr_parent_ids: Vec = Vec::new(); + let mut attr_keys: Vec = Vec::new(); + let mut attr_strs: Vec> = Vec::new(); + let mut attr_ints: Vec> = Vec::new(); + let mut attr_bytes: Vec>> = Vec::new(); + let mut skipped_non_scalar = 0usize; + let mut next_parent_id: u32 = 0; + + for resource in view.resources() { + for scope in resource.scopes() { + for metric in scope.metrics() { + let name = String::from_utf8_lossy(metric.name()).into_owned(); + let Some(data) = metric.data() else { + continue; + }; + let number_dps: Vec<_> = if let Some(gauge) = data.as_gauge() { + gauge.data_points().collect() + } else if let Some(sum) = data.as_sum() { + sum.data_points().collect() + } else { + // Histogram / ExponentialHistogram / Summary: no + // single well-defined scalar — count, don't expand. + skipped_non_scalar += match data.value_type() { + MetricKind::Histogram => data + .as_histogram() + .map(|h| h.data_points().count()) + .unwrap_or(0), + MetricKind::ExponentialHistogram => data + .as_exponential_histogram() + .map(|h| h.data_points().count()) + .unwrap_or(0), + MetricKind::Summary => { + data.as_summary().map(|s| s.data_points().count()).unwrap_or(0) + } + _ => 0, + }; + continue; + }; + + for dp in number_dps { + let Some(value) = dp.value() else { continue }; + let value = match value { + DpValue::Double(v) => v, + DpValue::Integer(v) => v as f64, + }; + + let parent_id = next_parent_id; + next_parent_id += 1; + + time_unix_nano.push(dp.time_unix_nano()); + metric_names.push(name.clone()); + values.push(value); + + for attr in dp.attributes() { + let Some(val) = attr.value() else { continue }; + attr_parent_ids.push(parent_id); + attr_keys.push(String::from_utf8_lossy(attr.key()).into_owned()); + let (mut s, mut i, mut b) = (None, None, None); + match val.value_type() { + ValueType::String => { + s = val.as_string().map(|v| String::from_utf8_lossy(v).into_owned()); + } + ValueType::Bytes => { + b = val.as_bytes().map(|v| v.to_vec()); + } + ValueType::Int64 => { + // ASAP's int attribute column is + // unsigned; a genuinely negative int + // attribute (rare for label-shaped + // data) isn't representable, so it's + // stringified instead of silently + // reinterpreted as a huge positive + // value. + match val.as_int64() { + Some(v) if v >= 0 => i = Some(v as u64), + Some(v) => s = Some(v.to_string()), + None => {} + } + } + ValueType::Double => { + s = val.as_double().map(|v| v.to_string()); + } + ValueType::Bool => { + s = val.as_bool().map(|v| v.to_string()); + } + ValueType::Empty | ValueType::Array | ValueType::KeyValueList => { + // Not representable in ASAP's flat + // label model; drop this one + // attribute (the data point itself is + // still kept). + attr_parent_ids.pop(); + attr_keys.pop(); + continue; + } + } + attr_strs.push(s); + attr_ints.push(i); + attr_bytes.push(b); + } + } + } + } + } + + if time_unix_nano.is_empty() { + return Ok(DecodeOutcome { + records: None, + skipped_non_scalar, + }); + } + + let parent_ids: Vec = (0..time_unix_nano.len() as u32).collect(); + let metrics = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new(COLUMN_TIME_UNIX_NANO, DataType::UInt64, false), + Field::new(COLUMN_METRIC, DataType::Utf8, false), + Field::new(COLUMN_VALUE, DataType::Float64, false), + Field::new(ATTR_BATCH_PARENT_ID, DataType::UInt32, false), + ])), + vec![ + Arc::new(UInt64Array::from(time_unix_nano)), + Arc::new(StringArray::from(metric_names)), + Arc::new(Float64Array::from(values)), + Arc::new(UInt32Array::from(parent_ids)), + ], + )?; + + let attributes = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new(ATTR_BATCH_PARENT_ID, DataType::UInt32, false), + Field::new(ATTR_BATCH_KEY, DataType::Utf8, false), + Field::new(ATTR_BATCH_STR, DataType::Utf8, true), + Field::new(ATTR_BATCH_INT, DataType::UInt64, true), + Field::new(ATTR_BATCH_BYTES, DataType::Binary, true), + ])), + vec![ + Arc::new(UInt32Array::from(attr_parent_ids)), + Arc::new(StringArray::from(attr_keys)), + Arc::new(StringArray::from(attr_strs)), + Arc::new(UInt64Array::from(attr_ints)), + Arc::new(BinaryArray::from_opt_vec( + attr_bytes.iter().map(|b| b.as_deref()).collect(), + )), + ], + )?; + + Ok(DecodeOutcome { + records: Some(OtapMetricRecords { metrics, attributes }), + skipped_non_scalar, + }) +} + +// -- Small typed-column helpers (mirrors records.rs's own private ones) ----- + +fn require_string<'a>( + batch: &'a RecordBatch, + which: &'static str, + column: &'static str, +) -> Result<&'a StringArray, BridgeError> { + batch + .column_by_name(column) + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or(BridgeError::BadColumn { + batch: which, + column, + }) +} + +fn require_uint64<'a>( + batch: &'a RecordBatch, + which: &'static str, + column: &'static str, +) -> Result<&'a UInt64Array, BridgeError> { + batch + .column_by_name(column) + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or(BridgeError::BadColumn { + batch: which, + column, + }) +} + +fn require_uint32<'a>( + batch: &'a RecordBatch, + which: &'static str, + column: &'static str, +) -> Result<&'a UInt32Array, BridgeError> { + batch + .column_by_name(column) + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or(BridgeError::BadColumn { + batch: which, + column, + }) +} + +fn require_float64<'a>( + batch: &'a RecordBatch, + which: &'static str, + column: &'static str, +) -> Result<&'a Float64Array, BridgeError> { + batch + .column_by_name(column) + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or(BridgeError::BadColumn { + batch: which, + column, + }) +} + +fn optional_string(batch: &RecordBatch, column: &'static str) -> Result, BridgeError> { + match batch.column_by_name(column) { + None => Ok(None), + Some(c) => c + .as_any() + .downcast_ref::() + .cloned() + .map(Some) + .ok_or(BridgeError::BadColumn { + batch: "attributes", + column, + }), + } +} + +fn optional_uint64(batch: &RecordBatch, column: &'static str) -> Result, BridgeError> { + match batch.column_by_name(column) { + None => Ok(None), + Some(c) => c + .as_any() + .downcast_ref::() + .cloned() + .map(Some) + .ok_or(BridgeError::BadColumn { + batch: "attributes", + column, + }), + } +} + +fn optional_binary(batch: &RecordBatch, column: &'static str) -> Result, BridgeError> { + match batch.column_by_name(column) { + None => Ok(None), + Some(c) => c + .as_any() + .downcast_ref::() + .cloned() + .map(Some) + .ok_or(BridgeError::BadColumn { + batch: "attributes", + column, + }), + } +}