[zzylol]

Why

The Rust collector (asap-precompute-rs, asap_sketches on the OTAP dataflow engine) emits one ASAPv1 sketch envelope per series per tumbling window. The real backend cannot ingest these yet, so the outputs need a fake backend. It should count what arrives and, when asked, keep it on disk so an offline script can compare sketch answers with exact answers. The Go sink in otel-app/sink/ does this for the Go collector's modified-OTLP sketch types. It does not read ASAPv1 kind ids, and it runs as a separate gRPC server.

What

• New terminal exporter urn:asap:exporter:sink (src/otap/sink.rs). It is registered in the same way as the other ASAP nodes, so asap-collector can run it.
• It counts:
  • sketch envelopes per ASAPv1 kind id (messages and payload bytes), with the kind marked full or delta;
  • plain (non-sketch) data points;
  • envelopes whose payload is not ASAPv1 (for example exact windows);
  • batches, and batches that failed.
• With dump_dir set, it writes each received batch to its own file before decoding it: <receive-unix-ns>-<seq>.pb, an OTLP ExportMetricsServiceRequest protobuf. Without dump_dir, it writes nothing.
• It prints the totals to stdout on shutdown, and also every stats_every when that is set.
• configs/sink.yaml: OTLP/HTTP receiver on :4319 → sink.
• configs/collector.yaml: OTLP/HTTP receiver on :4318 → asap_sketches (one DDSketch per route per 10 s window) → OTLP/HTTP exporter to the sink.
• README module table and the asap-collector doc comment now list the sink.

The sink adds no gRPC or HTTP server code: the upstream OTLP or OTAP receiver feeds it inside its own engine process. Raw XOR chunks are not handled yet; that is a later step. The Go sink in otel-app/sink/ now does the same job and is kept for now. It will be deleted once the Rust path is verified end to end.

raw_sink was checked first and not extended. It is a processor that ends the raw baseline pipeline: it writes raw points to a log store and reports acknowledgement latency gauges. It does not count sketch envelopes, and sharing that node would mix two unrelated jobs.

How

• Decoding reuses the existing codec::decode_each. Each observation that carries an envelope is counted by asap_sketchlib::peek_kind_id on its payload. That function checks the magic, version and framing, and parses neither the metadata nor the payload.
• Delta comes from the top bit of the kind id's variant byte (ASAPv1 spec §2).
• Each batch is counted into local totals first. These are added to the running totals only if the whole batch dumps and decodes, so a failed batch adds nothing except failed.
• For the dump, the batch payload is converted to OtlpProtoBytes and those bytes are written. The file therefore holds the whole batch: labels, window bounds, resource attributes and sketch bytes.
• Every batch is acked, failed ones included. A failure is logged through the engine's info.

Key code interfaces

pub const SINK_URN: &str = "urn:asap:exporter:sink";

pub struct SinkConfig {
    pub dump_dir: Option<PathBuf>,     // unset: nothing written
    pub stats_every: Option<Duration>, // unset: totals on shutdown only
}

pub struct Sink { pub batches, pub failed_batches, pub plain_points, pub non_asapv1,
                  pub kinds: BTreeMap<Vec<u8>, KindTotals> /* {msgs, bytes} */, .. }
impl Sink {
    pub fn receive(&mut self, pdata: OtapPdata) -> Result<(), String>;
    pub fn report(&self) -> String;
}

Usage

cargo build --features collector --bin asap-collector
asap-collector --config configs/sink.yaml        # fake backend on :4319
asap-collector --config configs/collector.yaml   # OTLP :4318 -> asap_sketches -> sink
# send OTLP metrics to :4318 (e.g. a scenario_generator pipeline)

sink:
  type: urn:asap:exporter:sink
  config: {dump_dir: /data/sink-dump, stats_every: 10s}

Major algorithm changes

None.

Before

Nothing in asap-precompute-rs could receive the collector's sketch output. The only fake backend was the Go sink, which does not read ASAPv1 kind ids.

After

Two asap-collector processes, collector and sink, run the full path. The sink prints, for example:

SINK_TOTAL batches=5 failed=0 plain_points=0 non_asapv1=0
SINK_SKETCH kind=05-01 mode=full msgs=100 bytes=39280

Verification

• cargo fmt --check: clean.

• cargo clippy --all-targets --all-features: no warnings.

• cargo test --all-features: all pass (lib 100 passed and 1 ignored, stream_bench 6, kll_windows 11, otap_pipeline_e2e 1, runtime 26).

• New test otap::sink::tests::counts_kinds_and_plain_points_and_dumps_each_batch. It sends two batches, each holding a full DDSketch (05 01), a Count-Min (02 00) and one plain gauge, to a sink with dump_dir set to a temp dir. It then checks:

  • the per-kind messages and bytes, the plain-point count and the report lines;
  • that there are two files named <ns>-1.pb and <ns>-2.pb;
  • that each file decodes back through the codec to the same sketch payloads and labels.

  The collector produces no delta kind, so the test uses two full kinds.

• Local run: scenario_generator (2000 points/s, 20 routes) → configs/collector.yaml → sink (configs/sink.yaml plus dump_dir). Over about 50 s, the sink received 5 batches with 20 DDSketch envelopes each (one per route per 10 s window), and no batch failed. It wrote 5 dump files and printed the totals every 10 s and on shutdown.