Skip to content

otap: real plugin integration + Arrow IPC network transport - #6

Open
zzylol wants to merge 1 commit into
feat/schema-dictionary-record-codecfrom
feat/otap-plugin-and-network-transport
Open

otap: real plugin integration + Arrow IPC network transport#6
zzylol wants to merge 1 commit into
feat/schema-dictionary-record-codecfrom
feat/otap-plugin-and-network-transport

Conversation

@zzylol

@zzylol zzylol commented Aug 10, 2026

Copy link
Copy Markdown
Contributor

Stacked on #5 (feat/schema-dictionary-record-codec), which implemented the SeriesDictionary/SeriesDictionaryDecoder codec but only proved it end-to-end via direct calls in one process, in-memory — not through the actual OTAP plugin lifecycle, and not across a real serialize/transmit/deserialize hop. This PR closes both gaps, per review feedback on that PR's demo.

What changed

  • otap::wire (new): Arrow-IPC serializes a SketchStreamBatch's four RecordBatches into a length-prefixed frame, plus async send_stream_batch/recv_stream_batch over a TcpStream. Each sub-batch is its own self-contained IPC stream (schema + one record batch + EOS); recv_stream_batch distinguishes a clean EOF between frames from a truncated one mid-frame.
  • AsapSketchesPlugin::start_from_envelopes (new): the receiver-role counterpart to the existing producer-role start(). Consumes Stream<Item = SketchStreamBatch> instead of Stream<Item = OtapMetricRecords>, decodes via a persistent SeriesDictionaryDecoder, and routes reconstructed envelopes through Precompute::observe_envelope (merge, never expand to samples) — reusing the same ticker/control-task/graceful-drain machinery as the producer role via a new shared spawn_lifecycle helper. A receiver configured with transmit_sketch = false naturally re-emits query-mode (quantile) estimates instead of sketch bytes through its own emit channel, so a chain of AsapSketchesPlugins can compose without any new machinery.
  • examples/sketch_producer_node.rs + sketch_receiver_node.rs (new): two separate binaries — real AsapSketchesPlugin producer and receiver roles, connected over a real TCP socket via otap::wire, not the in-process mpsc channel sketch_pipeline_demo.rs uses. The producer feeds a real OTAP-shaped input stream (records::flatten + decode_batch, not a direct observe() call) and lets the plugin's actual Wakeup-style Tokio ticker close windows on its own wall-clock schedule.

examples/sketch_pipeline_demo.rs (from #5) is unchanged and still works — it remains the fast, single-process way to see the codec logic; these two binaries are the higher-fidelity version.

Testing

  • cargo test --features otap: 156/156 passing (7 new: otap::wire's IPC round-trip + real TCP loopback tests, AsapSketchesPlugin's receiver-role smoke test and full producer→receiver chain test).
  • cargo fmt --check / cargo clippy --features otap --tests --examples: clean.
  • Ran both new binaries together end to end: producer emits 5 windows (window 0 carries SCHEMA+DICTIONARY+LABELS, windows 1-4 carry RECORD only — dictionary economics holding over the real socket), receiver receives and Arrow-IPC-decodes all 5, merges them, and prints a correct p99 gauge.

🤖 Generated with Claude Code

Stacked on feat/schema-dictionary-record-codec, which implemented the
Schema/Dictionary/Record codec (SeriesDictionary/SeriesDictionaryDecoder)
but only proved it end-to-end via direct calls in one process, in-memory
-- not through the actual OTAP plugin lifecycle, and not across a real
serialize/transmit/deserialize hop.

This PR closes both gaps:

otap::wire (new): Arrow-IPC serializes a SketchStreamBatch's four
RecordBatches into a length-prefixed frame, plus async send_stream_batch/
recv_stream_batch over a TcpStream. Each sub-batch is its own
self-contained IPC stream (schema + one record batch + EOS); recv_stream_batch
distinguishes a clean EOF between frames from a truncated one mid-frame.

AsapSketchesPlugin::start_from_envelopes (new): the receiver-role
counterpart to the existing producer-role start(). Consumes
Stream<Item = SketchStreamBatch> instead of Stream<Item = OtapMetricRecords>,
decodes via a persistent SeriesDictionaryDecoder, and routes reconstructed
envelopes through Precompute::observe_envelope (merge, never expand to
samples) -- reusing the same ticker/control-task/graceful-drain machinery
as the producer role via a new shared spawn_lifecycle helper. A receiver
configured with transmit_sketch=false naturally re-emits query-mode
(quantile) estimates instead of sketch bytes through its own emit channel,
so a chain of AsapSketchesPlugins can compose without any new machinery.

examples/sketch_producer_node.rs + sketch_receiver_node.rs (new): two
separate binaries -- real AsapSketchesPlugin producer and receiver roles,
connected over a real TCP socket via otap::wire, not the in-process mpsc
channel sketch_pipeline_demo.rs uses. The producer feeds a real OTAP-shaped
input stream (records::flatten + decode_batch, not a direct observe()
call) and lets the plugin's actual Wakeup-style Tokio ticker close windows
on its own wall-clock schedule. Verified running both together: producer
emits 5 windows (window 0 carries SCHEMA+DICTIONARY+LABELS, windows 1-4
carry RECORD only), receiver receives and decodes all 5 over the socket,
merges them, and prints a correct p99 gauge.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant