From c377a8c2eb188300fbdcce5706ad5d9890e16656 Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:29:59 +0000 Subject: [PATCH 1/2] Observe points without per-point allocation The sketch branch spent 4.1 us per point before the sketch saw it: decoding allocated every label as new strings, the per-attribute split copied each point again, and observe cloned the whole config, took three locks and built a fresh series-key string. - codec: decode_each walks the OTAP view and rewrites one observation per point; decode_pdata_to_observations is built on it. - matchers: per_attribute writes each split into a reused observation; write_series_key fills a reused buffer and skips sorting labels that are already sorted. - precompute: the config is shared through an Arc; window and stats sit under one lock. - The sketch and exact branches both use decode_each. Per point on a pinned core (top-k, 100k Zipf routes, CMS 4x512): decode 2.02 -> 1.09 us, split 1.14 -> 0.33 us, observe 0.98 -> 0.29 us. Live top-k stack: sketch branch 0.158 -> 0.075 cores, exact branch 0.095 -> 0.064 cores. --- asap-precompute-rs/src/config.rs | 29 ++- asap-precompute-rs/src/matchers.rs | 112 +++++++-- asap-precompute-rs/src/otap/baseline_nodes.rs | 24 +- asap-precompute-rs/src/otap/codec.rs | 92 +++++--- asap-precompute-rs/src/otap/processor.rs | 214 ++++++++++++------ asap-precompute-rs/src/precompute.rs | 88 ++++--- asap-precompute-rs/src/window.rs | 104 +++++---- 7 files changed, 444 insertions(+), 219 deletions(-) diff --git a/asap-precompute-rs/src/config.rs b/asap-precompute-rs/src/config.rs index dc45427..e6d695f 100644 --- a/asap-precompute-rs/src/config.rs +++ b/asap-precompute-rs/src/config.rs @@ -347,18 +347,27 @@ impl PrecomputeConfig { /// Thin wrapper over [`crate::matchers::series_key`], which owns /// the full canonical-byte formatting. pub fn series_key_for(&self, obs: &Observation) -> String { + let mut buf = String::new(); + self.write_series_key_for(&mut buf, obs); + buf + } + + /// Writes [`Self::series_key_for`] into `buf`, replacing its contents. + pub fn write_series_key_for(&self, buf: &mut String, obs: &Observation) { + use crate::matchers::write_series_key; if self.global_aggregation { - return crate::matchers::series_key(self.agg_id, &[], &[], &[]); - } - if self.omit_resource_attrs { - return crate::matchers::series_key(self.agg_id, &[], &obs.labels, &self.aggregate_by); + write_series_key(buf, self.agg_id, &[], &[], &[]); + } else if self.omit_resource_attrs { + write_series_key(buf, self.agg_id, &[], &obs.labels, &self.aggregate_by); + } else { + write_series_key( + buf, + self.agg_id, + &obs.resource_labels, + &obs.labels, + &self.aggregate_by, + ); } - crate::matchers::series_key( - self.agg_id, - &obs.resource_labels, - &obs.labels, - &self.aggregate_by, - ) } /// Rebuilds the same key from a series entry's stored labels. diff --git a/asap-precompute-rs/src/matchers.rs b/asap-precompute-rs/src/matchers.rs index bc04473..23b3378 100644 --- a/asap-precompute-rs/src/matchers.rs +++ b/asap-precompute-rs/src/matchers.rs @@ -137,12 +137,26 @@ pub fn series_key( aggregate_by: &[String], ) -> String { let mut buf = String::new(); - buf.push_str(&agg_id.to_string()); + write_series_key(&mut buf, agg_id, resource_labels, labels, aggregate_by); + buf +} + +/// Writes [`series_key`] into `buf`, replacing its contents, so a hot path +/// can reuse one buffer. +pub fn write_series_key( + buf: &mut String, + agg_id: AggId, + resource_labels: &[KeyValue], + labels: &[KeyValue], + aggregate_by: &[String], +) { + use std::fmt::Write; + buf.clear(); + let _ = write!(buf, "{agg_id}"); buf.push('|'); - write_attributes_key(&mut buf, resource_labels, &[]); + write_attributes_key(buf, resource_labels, &[]); buf.push('|'); - write_attributes_key(&mut buf, labels, aggregate_by); - buf + write_attributes_key(buf, labels, aggregate_by); } /// Returns just the per-label-set portion of [`series_key`]. @@ -160,6 +174,17 @@ pub fn attributes_key(labels: &[KeyValue], aggregate_by: &[String]) -> String { fn write_attributes_key(buf: &mut String, labels: &[KeyValue], aggregate_by: &[String]) { if aggregate_by.is_empty() { + // Already sorted with unique keys (e.g. one label): same output as + // the sorted walk below, without collecting the keys. + if labels.windows(2).all(|w| w[0].key < w[1].key) { + for kv in labels { + buf.push_str(&kv.key); + buf.push('='); + buf.push_str(&kv.value); + buf.push(';'); + } + return; + } // Sort by key for stable output. let mut keys: Vec<&str> = labels.iter().map(|kv| kv.key.as_str()).collect(); keys.sort(); @@ -213,19 +238,35 @@ pub const ATTRIBUTE_KEY_LABEL: &str = "attribute.key"; /// the attribute's value as bytes. A cardinality or frequency sketch over /// these series answers "how many distinct values" and "which values /// dominate" per attribute key. -pub fn per_attribute(obs: &Observation, keys: &[String]) -> Vec { +/// +/// Each split is written into `scratch` and passed to `f`, so a stream of +/// points reuses one observation's buffers instead of allocating per point. +pub fn per_attribute( + obs: &Observation, + keys: &[String], + scratch: &mut Observation, + mut f: impl FnMut(&Observation) -> Result<(), E>, +) -> Result<(), E> { let all = keys.iter().any(|k| k == "*"); - obs.labels - .iter() - .filter(|kv| all || keys.contains(&kv.key)) - .map(|kv| Observation { - timestamp_ms: obs.timestamp_ms, - metric: obs.metric.clone(), - resource_labels: obs.resource_labels.clone(), - labels: vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, kv.key.clone())], - value: crate::observation::ObservationValue::bytes(kv.value.as_bytes().to_vec()), - }) - .collect() + for kv in &obs.labels { + if !all && !keys.contains(&kv.key) { + continue; + } + scratch.timestamp_ms = obs.timestamp_ms; + scratch.metric.clone_from(&obs.metric); + scratch.resource_labels.clone_from(&obs.resource_labels); + scratch.labels.resize_with(1, KeyValue::default); + scratch.labels[0].key.clear(); + scratch.labels[0].key.push_str(ATTRIBUTE_KEY_LABEL); + scratch.labels[0].value.clone_from(&kv.key); + let value = &mut scratch.value; + value.kind = crate::observation::ObservationValueKind::Bytes; + value.envelope = None; + value.bytes.clear(); + value.bytes.extend_from_slice(kv.value.as_bytes()); + f(scratch)?; + } + Ok(()) } #[cfg(test)] @@ -264,4 +305,43 @@ mod tests { let matchers = vec![LabelMatcher::equal("region", "us-east")]; assert!(LabelMatcher::matches_all(&matchers, &obs)); } + + #[test] + fn sorted_fast_path_writes_the_same_key_as_the_sorted_walk() { + let sorted = vec![KeyValue::new("a", "1"), KeyValue::new("b", "2")]; + let unsorted = vec![KeyValue::new("b", "2"), KeyValue::new("a", "1")]; + assert_eq!(attributes_key(&sorted, &[]), "a=1;b=2;"); + assert_eq!(attributes_key(&unsorted, &[]), "a=1;b=2;"); + let mut buf = "stale".to_string(); + write_series_key(&mut buf, 7, &sorted, &unsorted, &[]); + assert_eq!(buf, series_key(7, &sorted, &unsorted, &[])); + } + + #[test] + fn per_attribute_reuses_the_scratch_for_each_selected_key() { + let obs = Observation::new( + 5, + "m", + vec![KeyValue::new("service", "s")], + vec![KeyValue::new("route", "/a"), KeyValue::new("user", "u1")], + ObservationValue::float(1.0), + ); + let mut scratch = Observation::default(); + let mut seen = Vec::new(); + per_attribute::<()>(&obs, &["*".into()], &mut scratch, |o| { + seen.push(o.clone()); + Ok(()) + }) + .unwrap(); + let attr = |key: &str, value: &str| { + Observation::new( + 5, + "m", + vec![KeyValue::new("service", "s")], + vec![KeyValue::new(ATTRIBUTE_KEY_LABEL, key)], + ObservationValue::bytes(value.as_bytes().to_vec()), + ) + }; + assert_eq!(seen, vec![attr("route", "/a"), attr("user", "u1")]); + } } diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs index 3013f92..a6f1352 100644 --- a/asap-precompute-rs/src/otap/baseline_nodes.rs +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -68,7 +68,7 @@ use super::answer_nodes::{ default_grace, default_k, default_novelty_windows, default_quantiles, default_window, emit, gauges_pdata, invalid, now_ms, observations, series_key, }; -use super::codec::OtapSketchEncoder; +use super::codec::{decode_each, OtapSketchEncoder}; use crate::answers::{ exact_answers, gauge, ks_distance, topk_of, CardinalityClassifier, CardinalityRules, DriftRules, DriftTracker, Event, ExactWindow, NoveltyTracker, @@ -273,14 +273,20 @@ impl local::Processor for ExactBranch { Message::PData(pdata) => { // Consumed into state here: ack so upstream can move on. effects.notify_ack(AckMsg::new(pdata.clone())).await?; - let points = observations(pdata, effects).await; - // A window closes as soon as a point's event time passes its end. - let out = match points.first() { - Some(first) => self.close(first.timestamp_ms), - None => Vec::new(), - }; - points.iter().for_each(|obs| self.observe(obs)); - out + let mut out = None; + let decoded = decode_each(pdata, |obs| { + // A window closes as soon as a point's event time passes its end. + if out.is_none() { + out = Some(self.close(obs.timestamp_ms)); + } + self.observe(obs); + }); + if let Err(e) = decoded { + effects + .info(&format!("exact: undecodable input: {e}")) + .await; + } + out.unwrap_or_default() } Message::Control(NodeControlMsg::TimerTick { .. }) => { self.close(now_ms().saturating_sub(IDLE_CLOSE_GRACE.as_millis() as u64)) diff --git a/asap-precompute-rs/src/otap/codec.rs b/asap-precompute-rs/src/otap/codec.rs index 61dbbaf..8b35aee 100644 --- a/asap-precompute-rs/src/otap/codec.rs +++ b/asap-precompute-rs/src/otap/codec.rs @@ -1310,19 +1310,35 @@ fn pretty_arrow_type(data_type: &arrow_schema::DataType) -> String { /// match arm already owns it and has no further use for the original /// context once this conversion runs. pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result { + let mut observations = Vec::new(); + let skipped_non_scalar = decode_each(pdata, |obs| observations.push(obs.clone()))?; + Ok(DecodeOutcome { + observations, + skipped_non_scalar, + }) +} + +/// Walks `pdata` like [`decode_pdata_to_observations`], passing each data +/// point to `f` instead of collecting them. The observation is one buffer +/// rewritten per point, so a hot path decodes without allocating per point. +/// Returns the number of non-scalar data points skipped. +pub fn decode_each(pdata: OtapPdata, f: impl FnMut(&Observation)) -> Result { let (_context, payload) = pdata.into_parts(); let arrow_records: OtapArrowRecords = payload .try_into_with_default() .map_err(|e| CodecError::Decode(format!("{e}")))?; let OtapArrowRecords::Metrics(_) = &arrow_records else { - return Ok(DecodeOutcome::default()); + return Ok(0); }; let view = OtapMetricsView::try_from(&arrow_records) .map_err(|e| CodecError::Decode(format!("{e}")))?; - let mut acc = DecodeAccumulator::default(); + let mut acc = DecodeAccumulator { + scratch: Observation::default(), + sink: f, + }; let mut skipped_non_scalar = 0usize; for resource in view.resources() { @@ -1418,13 +1434,23 @@ pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result, index: usize, key: &[u8], value: &[u8]) { + if index == labels.len() { + labels.push(KeyValue::default()); + } + let label = &mut labels[index]; + label.key.clear(); + label.key.push_str(&String::from_utf8_lossy(key)); + label.value.clear(); + label.value.push_str(&String::from_utf8_lossy(value)); +} + +/// Passes one `Observation` per data point to `sink` while walking the real /// `OtapMetricsView` — factored out of [`decode_pdata_to_observations`] /// so the Gauge and Sum branches (whose `NumberDataPointView` /// implementations are different concrete types, both borrowing from @@ -1433,9 +1459,10 @@ pub fn decode_pdata_to_observations(pdata: OtapPdata) -> Result, +struct DecodeAccumulator { + /// Rewritten for each data point. + scratch: Observation, + sink: F, } /// Parses an optional `sketch.type` value; absent means `Unspecified`. @@ -1454,7 +1481,7 @@ struct DecodeSchema { schema_version: Option, } -impl DecodeAccumulator { +impl DecodeAccumulator { fn push_summary_data_point( &mut self, dp: D, @@ -1512,7 +1539,7 @@ impl DecodeAccumulator { aggregation_temporality: 0, value: dp.sum(), }; - self.observations.push(Observation::new( + (self.sink)(&Observation::new( timestamp_ms, metric_name.to_string(), resource_labels.to_vec(), @@ -1549,12 +1576,21 @@ impl DecodeAccumulator { let mut schema_version = inherited_schema.schema_version; let mut window_start_ms: Option = None; let mut window_end_ms: Option = None; - let mut labels: Vec = series_labels.to_vec(); + let obs = &mut self.scratch; + for (index, label) in series_labels.iter().enumerate() { + set_label( + &mut obs.labels, + index, + label.key.as_bytes(), + label.value.as_bytes(), + ); + } + let mut label_count = series_labels.len(); for attr in dp.attributes() { let Some(val) = attr.value() else { continue }; - let key = String::from_utf8_lossy(attr.key()).into_owned(); - match key.as_str() { + let key = attr.key(); + match std::str::from_utf8(key).unwrap_or_default() { ATTR_ENVELOPE => envelope_bytes = val.as_bytes().map(<[u8]>::to_vec), ATTR_SKETCH_TYPE => { sketch_type_raw = val @@ -1576,15 +1612,17 @@ impl DecodeAccumulator { // string-typed values are representable as a label; // a non-string, non-reserved attribute is dropped. if let Some(s) = val.as_string() { - labels.push(KeyValue::new(key, String::from_utf8_lossy(s).into_owned())); + set_label(&mut obs.labels, label_count, key, s); + label_count += 1; } } } } + obs.labels.truncate(label_count); // An empty `sketch.envelope` from some other producer routes to // the scalar path rather than an empty-payload envelope. - let observation_value = match envelope_bytes.filter(|b| !b.is_empty()) { + obs.value = match envelope_bytes.filter(|b| !b.is_empty()) { Some(payload) => { let sketch_type = parse_sketch_type(sketch_type_raw.as_deref())?; validate_self_describing_payload(sketch_type, &payload) @@ -1594,7 +1632,7 @@ impl DecodeAccumulator { sketch_type, agg_id: agg_id.unwrap_or(0), resource_labels: Vec::new(), - labels: labels.clone(), + labels: obs.labels.clone(), window_start_ms: window_start_ms.unwrap_or(0), window_end_ms: window_end_ms.unwrap_or(0), payload, @@ -1606,14 +1644,16 @@ impl DecodeAccumulator { } None => ObservationValue::float(value), }; - - self.observations.push(Observation::new( - timestamp_ms, - metric_name.to_string(), - resource_labels.to_vec(), - labels, - observation_value, - )); + obs.timestamp_ms = timestamp_ms; + if obs.metric != metric_name { + obs.metric.clear(); + obs.metric.push_str(metric_name); + } + if obs.resource_labels.as_slice() != resource_labels { + obs.resource_labels.clear(); + obs.resource_labels.extend_from_slice(resource_labels); + } + (self.sink)(obs); Ok(()) } } diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index e9dd3b2..d022039 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -96,12 +96,12 @@ use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; use crate::config::PrecomputeConfigSet; use crate::envelope::SketchEnvelope; use crate::matchers::per_attribute; -use crate::observation::ObservationValueKind; +use crate::observation::{Observation, ObservationValueKind}; use crate::otap::config::resolve as resolve_plugin_config; use crate::otap::PluginConfig; use crate::precompute::{Precompute, PrecomputeError, PrecomputeImpl}; -use super::codec::{decode_pdata_to_observations, OtapSketchEncoder}; +use super::codec::{decode_each, OtapSketchEncoder}; /// Public URN for the unified ASAP `asap_sketches` processor. Survives /// across hosts unchanged so a controller plan addressed at this URN @@ -350,6 +350,8 @@ pub struct AsapSketchesProcessor { next_config_version: Arc, /// Stable SCHEMA/DICTIONARY identity for this processor's output stream. sketch_encoder: OtapSketchEncoder, + /// Reused buffer for per-attribute splits of the point being observed. + split: Observation, } impl AsapSketchesProcessor { @@ -364,6 +366,7 @@ impl AsapSketchesProcessor { timer: None, next_config_version: Arc::new(AtomicU64::new(1)), sketch_encoder, + split: Observation::default(), } } @@ -426,8 +429,62 @@ impl local::Processor for AsapSketchesProcessor { effect_handler .notify_ack(AckMsg::new(pdata.clone())) .await?; - let outcome = match decode_pdata_to_observations(pdata) { - Ok(outcome) => outcome, + let tumbling = self.plugin_config.window_slide.is_zero(); + let precompute = &self.precompute; + let keys = &self.plugin_config.per_attribute; + let split = &mut self.split; + let mut closed = None; + let mut failure = None; + // Note: `observe` handles both "genuine OTLP metric" and + // "sketch shipped as binary inside an OTAP metric" (an + // upstream asap_sketches node's `sketch.envelope` output): + // the decoder tags the latter `ObservationValueKind::Envelope` + // and `Precompute::observe` merges those as pre-aggregated + // sketches instead of expanding them to scalar samples. See + // `codec.rs`'s module doc ("Scope"). + // + // LateData / SeriesCapExceeded are expected outcomes already + // tallied in stats, so silent by design. Anything else (e.g. + // NoConfig, AggIdMismatch) is a misconfiguration, reported + // once per batch below. + let mut observe = |obs: &Observation| { + match precompute.observe(obs) { + Ok(()) + | Err(PrecomputeError::LateData) + | Err(PrecomputeError::SeriesCapExceeded) => {} + Err(e) => { + let _ = failure.get_or_insert(e); + } + } + Ok::<(), std::convert::Infallible>(()) + }; + let decoded = decode_each(pdata, |obs| { + // A tumbling window closes as soon as a point's event time + // passes its end, so no point lands in an earlier window. + if tumbling + && closed.is_none() + && obs.value.kind != ObservationValueKind::Envelope + { + closed = Some(precompute.tick(obs.timestamp_ms)); + } + let _ = if keys.is_empty() { + observe(obs) + } else { + per_attribute(obs, keys, split, &mut observe) + }; + }); + if let Some(envs) = closed { + self.emit_envelopes(&envs, effect_handler).await?; + } + match decoded { + Ok(0) => {} + Ok(skipped) => { + effect_handler + .info(&format!( + "asap_sketches: skipped {skipped} unsupported histogram/exponential-histogram data point(s)" + )) + .await; + } Err(error) => { effect_handler .info(&format!( @@ -436,69 +493,12 @@ impl local::Processor for AsapSketchesProcessor { .await; return Ok(()); } - }; - if outcome.skipped_non_scalar > 0 { + } + if let Some(e) = failure { effect_handler - .info(&format!( - "asap_sketches: skipped {} unsupported histogram/exponential-histogram data point(s)", - outcome.skipped_non_scalar - )) + .info(&format!("asap_sketches: observe failed: {e}")) .await; } - // A tumbling window closes as soon as a point's event time - // passes its end, so no point lands in an earlier window. - if self.plugin_config.window_slide.is_zero() { - if let Some(first) = outcome - .observations - .iter() - .find(|o| o.value.kind != ObservationValueKind::Envelope) - { - let envs = self.precompute.tick(first.timestamp_ms); - self.emit_envelopes(&envs, effect_handler).await?; - } - } - let observations = if self.plugin_config.per_attribute.is_empty() { - outcome.observations - } else { - outcome - .observations - .iter() - .flat_map(|o| per_attribute(o, &self.plugin_config.per_attribute)) - .collect() - }; - 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 - // `sketch.envelope` Summary output) — no branching - // needed at this call site. `decode_pdata_to_observations` - // 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 `codec.rs`'s module doc - // ("Scope") for the full picture. - // - // LateData / SeriesCapExceeded are expected, - // already-tallied-in-stats outcomes (mirrors - // the processor's ingest policy) — 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. - match self.precompute.observe(obs) { - Ok(()) - | Err(PrecomputeError::LateData) - | Err(PrecomputeError::SeriesCapExceeded) => {} - Err(e) => { - effect_handler - .info(&format!("asap_sketches: observe failed: {e}")) - .await; - } - } - } if !self.plugin_config.window_slide.is_zero() { // Completed upstream panes already have a boundary. Emit // promptly instead of waiting another whole slide interval. @@ -590,6 +590,92 @@ fn asap_wall_clock_ms() -> u64 { #[cfg(test)] mod tests { + /// Per-point cost of each step of the top-k branch on generated batches + /// (100,000 Zipf routes, CMS 4 x 512 + top-10 heap). Run with + /// `cargo test --release --features otap-engine branch_phase_costs -- --ignored --nocapture`. + #[test] + #[ignore] + fn branch_phase_costs() { + use crate::otap::scenario_generator::{Scenario, ScenarioConfig}; + use otel_arrow_dfe_otap::pdata::OtapPdata; + use otel_arrow_dfe_pdata::proto::OtlpProtoMessage; + use otel_arrow_dfe_pdata::OtapPayload; + use std::time::Instant; + + let scenario: ScenarioConfig = serde_json::from_value(serde_json::json!({ + "seed": 1, "signals_per_second": 20000, "max_batch_size": 1024, + "metric_name": "http.server.request.duration", + "resource_attributes": {"service.name": "checkout", "source": "a"}, + "attributes": [{"key": "http.route", "cardinality": 100000, "zipf": 1.1}], + "value": {"lognormal": {"median": 50.0, "sigma": 0.6}} + })) + .unwrap(); + let mut scenario = Scenario::new(scenario).unwrap(); + let mut pdata = |second: u64, index: usize| { + let data = scenario.batch(second, index, 1_000_000_000 * (60 + second)); + let payload: OtapPayload = OtlpProtoMessage::Metrics(data).try_into().unwrap(); + OtapPdata::new_todo_context(payload) + }; + let batches: Vec = (0..20u64) + .flat_map(|s| (0..20).map(move |i| (s, i))) + .map(|(s, i)| pdata(s, i)) + .collect(); + let user: super::AsapSketchesUserConfig = serde_json::from_value(serde_json::json!({ + "sketch_type": "countminsketch_heap", "window_size": "60s", + "output_metric_name": "http.server.request.duration", + "per_attribute": ["http.route"], "omit_resource_attrs": true, + "transmit_sketch": true, "sketch_params": {"rows": 4, "columns": 512, "k": 10} + })) + .unwrap(); + let plugin = user.into_plugin_config().unwrap(); + let (cfg, dispatch) = super::resolve_plugin_config(&plugin).unwrap(); + let precompute = super::build_precompute(cfg, dispatch).unwrap(); + + let timed = |f: &mut dyn FnMut(&crate::observation::Observation)| { + let t = Instant::now(); + for b in &batches { + crate::otap::codec::decode_each(b.clone(), &mut *f).unwrap(); + } + t.elapsed() + }; + let mut points = 0usize; + let decode = timed(&mut |_| points += 1); + let mut split = crate::observation::Observation::default(); + let mut keys: Vec> = Vec::new(); + let decode_split = timed(&mut |obs| { + crate::matchers::per_attribute::<()>(obs, &plugin.per_attribute, &mut split, |o| { + keys.push(o.value.bytes.clone()); + Ok(()) + }) + .unwrap() + }); + let all = timed(&mut |obs| { + crate::matchers::per_attribute(obs, &plugin.per_attribute, &mut split, |o| { + precompute.observe(o) + }) + .unwrap() + }); + let mut cms = crate::sketches::CMSHeapWrapper::new(4, 512, 10); + let t = Instant::now(); + for k in &keys { + cms.update(k, 1); + } + let cms_only = t.elapsed(); + let us = |d: std::time::Duration| d.as_secs_f64() * 1e6 / points as f64; + println!("{points} points"); + println!("decode OTAP -> observations {:6.2} us/point", us(decode)); + println!( + "per_attribute split {:6.2} us/point", + us(decode_split) - us(decode) + ); + println!( + "precompute.observe (all) {:6.2} us/point", + us(all) - us(decode_split) + ); + println!("total {:6.2} us/point", us(all)); + println!(" of which CMS+heap update {:6.2} us/point", us(cms_only)); + } + #[test] fn sliding_config_selects_pane_timer_and_validates_alignment() { let parse = |size: &str, slide: &str, transmit: bool| { diff --git a/asap-precompute-rs/src/precompute.rs b/asap-precompute-rs/src/precompute.rs index b306a7c..0ed793f 100644 --- a/asap-precompute-rs/src/precompute.rs +++ b/asap-precompute-rs/src/precompute.rs @@ -2,7 +2,7 @@ use std::any::Any; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use std::time::Duration; use thiserror::Error; @@ -316,11 +316,12 @@ pub type SketchFactory = Box Box + Send + Sync>; /// can construct one directly. Fields are private; construction /// goes through [`PrecomputeImpl::new`]. pub struct PrecomputeImpl { - cfg: Mutex>, + /// Shared so `observe` takes a reference count, not a deep copy. + cfg: Mutex>>, sketch_factory: Option, observer: Option, - window: Mutex, - stats: Mutex, + /// Window and counters under one lock: every observation touches both. + state: Mutex, sketch_type: SketchType, closed: AtomicBool, } @@ -345,11 +346,10 @@ impl PrecomputeImpl { .map(|c| c.sketch_type) .unwrap_or(SketchType::Unspecified); Self { - cfg: Mutex::new(initial_cfg), + cfg: Mutex::new(initial_cfg.map(Arc::new)), sketch_factory, observer, - window: Mutex::new(WindowState::new()), - stats: Mutex::new(StatsSnapshot::default()), + state: Mutex::new(State::default()), sketch_type, closed: AtomicBool::new(false), } @@ -366,12 +366,23 @@ impl PrecomputeImpl { } } +/// The mutable state of a [`PrecomputeImpl`]. +#[derive(Default)] +struct State { + window: WindowState, + stats: StatsSnapshot, +} + impl PrecomputeImpl { - /// Returns a clone of the active config or `None`. - fn active_config(&self) -> Option { + /// Returns the active config or `None`. + fn shared_config(&self) -> Option> { self.cfg.lock().expect("config lock poisoned").clone() } + fn lock_state(&self) -> std::sync::MutexGuard<'_, State> { + self.state.lock().expect("state lock poisoned") + } + /// Walks the closed series, serializes each into a /// [`SketchEnvelope`], and /// updates the rolling stats counters. @@ -384,7 +395,7 @@ impl PrecomputeImpl { if closed.is_empty() { return Vec::new(); } - let cfg = match self.active_config() { + let cfg = match self.shared_config() { Some(c) => c, None => return Vec::new(), }; @@ -398,7 +409,7 @@ impl PrecomputeImpl { envelopes.extend(envs); } } - let mut stats = self.stats.lock().expect("stats lock poisoned"); + let stats = &mut self.lock_state().stats; stats.output_envelopes = stats .output_envelopes .saturating_add(envelopes.len() as u64); @@ -493,7 +504,7 @@ impl Precompute for PrecomputeImpl { if self.closed.load(Ordering::Acquire) { return Err(PrecomputeError::Other("instance is closed".into())); } - let cfg = match self.active_config() { + let cfg = match self.shared_config() { Some(c) => c, None => return Err(PrecomputeError::NoConfig), }; @@ -502,17 +513,20 @@ impl Precompute for PrecomputeImpl { // pre-aggregated path so we never explode them to scalars. // The input_observations counter is bumped before // routing so envelope-valued observations also count. - { - let mut stats = self.stats.lock().expect("stats lock poisoned"); - stats.input_observations = stats.input_observations.saturating_add(1); - } - if obs.value.kind == ObservationValueKind::Envelope { if let Some(env) = obs.value.envelope.as_ref() { + { + let stats = &mut self.lock_state().stats; + stats.input_observations = stats.input_observations.saturating_add(1); + } return self.observe_envelope(env); } } + let mut state = self.lock_state(); + let State { window, stats } = &mut *state; + stats.input_observations = stats.input_observations.saturating_add(1); + if !cfg.matches(obs) { return Ok(()); } @@ -526,9 +540,7 @@ impl Precompute for PrecomputeImpl { .as_ref() .ok_or_else(|| PrecomputeError::Other("sketch observer not configured".into()))?; - let mut window = self.window.lock().expect("window lock poisoned"); - let mut stats = self.stats.lock().expect("stats lock poisoned"); - let result = window.observe(obs, &cfg, sketch_factory, observer, &mut stats); + let result = window.observe(obs, &cfg, sketch_factory, observer, stats); if let Err(err) = &result { match err { PrecomputeError::SeriesCapExceeded => { @@ -547,7 +559,7 @@ impl Precompute for PrecomputeImpl { if self.closed.load(Ordering::Acquire) { return Err(PrecomputeError::Other("instance is closed".into())); } - let cfg = match self.active_config() { + let cfg = match self.shared_config() { Some(c) => c, None => return Err(PrecomputeError::NoConfig), }; @@ -574,10 +586,10 @@ impl Precompute for PrecomputeImpl { .as_ref() .ok_or_else(|| PrecomputeError::Other("sketch factory not configured".into()))?; - let mut window = self.window.lock().expect("window lock poisoned"); - let mut stats = self.stats.lock().expect("stats lock poisoned"); + let mut state = self.lock_state(); + let State { window, stats } = &mut *state; stats.input_envelopes = stats.input_envelopes.saturating_add(1); - let result = window.observe_envelope(env, &cfg, sketch_factory, &mut stats); + let result = window.observe_envelope(env, &cfg, sketch_factory, stats); if let Err(err) = &result { if matches!(err, PrecomputeError::SeriesCapExceeded) { stats.dropped_overflow = stats.dropped_overflow.saturating_add(1); @@ -587,39 +599,25 @@ impl Precompute for PrecomputeImpl { } fn tick(&self, now_ms: u64) -> Vec { - let cfg = match self.active_config() { + let cfg = match self.shared_config() { Some(c) => c, None => return Vec::new(), }; - let (closed, rng) = { - let mut window = self.window.lock().expect("window lock poisoned"); - window.rotate(now_ms, &cfg) - }; + let (closed, rng) = self.lock_state().window.rotate(now_ms, &cfg); self.finish_rotate(closed, rng, now_ms) } fn drain(&self) -> Vec { - let cfg = match self.active_config() { + let cfg = match self.shared_config() { Some(c) => c, None => return Vec::new(), }; - let (closed, rng) = { - let mut window = self.window.lock().expect("window lock poisoned"); - window.drain(&cfg) - }; + let (closed, rng) = self.lock_state().window.drain(&cfg); self.finish_rotate(closed, rng, rng[1]) } fn active_config(&self) -> Option { - // Duplicated (rather than delegating to the inherent - // `PrecomputeImpl::active_config` above) so the trait method's - // body doesn't read as a same-name recursive call — inherent - // methods win method resolution over trait methods for the - // same receiver, so `self.active_config()` here would in fact - // call the inherent one and not recurse, but spelling out the - // two-line lock+clone directly is clearer than relying on that - // priority rule. - self.cfg.lock().expect("config lock poisoned").clone() + self.shared_config().map(|c| (*c).clone()) } fn update_config(&self, cs: &PrecomputeConfigSet) { @@ -637,12 +635,12 @@ impl Precompute for PrecomputeImpl { None => cs.configs.first(), }; if let Some(c) = chosen { - *guard = Some(c.clone()); + *guard = Some(Arc::new(c.clone())); } } fn stats(&self) -> StatsSnapshot { - *self.stats.lock().expect("stats lock poisoned") + self.lock_state().stats } fn shutdown(&self) -> Result<(), PrecomputeError> { diff --git a/asap-precompute-rs/src/window.rs b/asap-precompute-rs/src/window.rs index 4fc05f8..047c7d1 100644 --- a/asap-precompute-rs/src/window.rs +++ b/asap-precompute-rs/src/window.rs @@ -60,6 +60,8 @@ pub struct WindowState { /// initialized for the active config. Lazy-init avoids needing /// the constructor to know the config up front. pub(crate) initialized: bool, + /// Reused buffer for the series key of the observation being routed. + key: String, } impl Default for WindowState { @@ -85,6 +87,7 @@ impl WindowState { active_start_ms: 0, active_end_ms: 0, initialized: false, + key: String::new(), } } @@ -139,61 +142,64 @@ impl WindowState { } } - let key = cfg.series_key_for(obs); + cfg.write_series_key_for(&mut self.key, obs); + if let Some(entry) = self.series.get_mut(self.key.as_str()) { + if obs.timestamp_ms > entry.last_seen_ms { + entry.last_seen_ms = obs.timestamp_ms; + } + observer.observe(entry.sketch.as_mut(), obs)?; + entry.count += 1; + return Ok(()); + } + let key = self.key.clone(); - if !self.series.contains_key(&key) { - // New series — check cap. - if cfg.max_series > 0 && self.series.len() as u64 >= cfg.max_series { - match cfg.on_overflow { - OnOverflow::Drop | OnOverflow::Block => { - // Block degrades to Drop: latency-hostile - // semantics belong to integration tests, not - // the runtime hot path. - return Err(PrecomputeError::SeriesCapExceeded); - } - OnOverflow::EvictOldest => { - // Find and evict the oldest series. - let mut oldest_key: Option = None; - let mut oldest_ms: u64 = u64::MAX; - for (k, e) in self.series.iter() { - if e.last_seen_ms < oldest_ms { - oldest_ms = e.last_seen_ms; - oldest_key = Some(k.clone()); - } - } - if let Some(k) = oldest_key { - self.series.remove(&k); - stats.active_series -= 1; + // New series — check cap. + if cfg.max_series > 0 && self.series.len() as u64 >= cfg.max_series { + match cfg.on_overflow { + OnOverflow::Drop | OnOverflow::Block => { + // Block degrades to Drop: latency-hostile + // semantics belong to integration tests, not + // the runtime hot path. + return Err(PrecomputeError::SeriesCapExceeded); + } + OnOverflow::EvictOldest => { + // Find and evict the oldest series. + let mut oldest_key: Option = None; + let mut oldest_ms: u64 = u64::MAX; + for (k, e) in self.series.iter() { + if e.last_seen_ms < oldest_ms { + oldest_ms = e.last_seen_ms; + oldest_key = Some(k.clone()); } } + if let Some(k) = oldest_key { + self.series.remove(&k); + stats.active_series -= 1; + } } } - let sketch = sketch_factory(); - // Honor parity-mode flags by stripping the labels we - // promised not to surface. GlobalAggregation collapses - // everything; OmitResourceAttrs zeroes only the resource - // segment. - let (resource_copy, labels_copy) = if cfg.global_aggregation { - (Vec::new(), Vec::new()) - } else if cfg.omit_resource_attrs { - (Vec::new(), obs.labels.clone()) - } else { - (obs.resource_labels.clone(), obs.labels.clone()) - }; - let entry = SeriesEntry { - sketch, - resource_labels: resource_copy, - labels: labels_copy, - last_seen_ms: obs.timestamp_ms, - count: 0, - }; - self.series.insert(key.clone(), entry); - stats.active_series += 1; - } else if let Some(entry) = self.series.get_mut(&key) { - if obs.timestamp_ms > entry.last_seen_ms { - entry.last_seen_ms = obs.timestamp_ms; - } } + let sketch = sketch_factory(); + // Honor parity-mode flags by stripping the labels we + // promised not to surface. GlobalAggregation collapses + // everything; OmitResourceAttrs zeroes only the resource + // segment. + let (resource_copy, labels_copy) = if cfg.global_aggregation { + (Vec::new(), Vec::new()) + } else if cfg.omit_resource_attrs { + (Vec::new(), obs.labels.clone()) + } else { + (obs.resource_labels.clone(), obs.labels.clone()) + }; + let entry = SeriesEntry { + sketch, + resource_labels: resource_copy, + labels: labels_copy, + last_seen_ms: obs.timestamp_ms, + count: 0, + }; + self.series.insert(key.clone(), entry); + stats.active_series += 1; let entry = self .series From 30cc1fb1c7226eaae20b9a9309c3802db06b029a Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Tue, 20 Oct 2026 23:19:45 +0000 Subject: [PATCH 2/2] Time the per-attribute split without collecting keys; insert new series once The split timing no longer clones every key into a vector inside the timed run; the keys for the CMS-only timing are collected in a separate untimed pass. A new series is inserted with the entry API instead of insert then get_mut, and the series-key writers are crate-private. --- asap-precompute-rs/src/config.rs | 2 +- asap-precompute-rs/src/matchers.rs | 2 +- asap-precompute-rs/src/otap/processor.rs | 20 ++++++++++++++------ asap-precompute-rs/src/window.rs | 7 +------ 4 files changed, 17 insertions(+), 14 deletions(-) diff --git a/asap-precompute-rs/src/config.rs b/asap-precompute-rs/src/config.rs index e6d695f..64658e9 100644 --- a/asap-precompute-rs/src/config.rs +++ b/asap-precompute-rs/src/config.rs @@ -353,7 +353,7 @@ impl PrecomputeConfig { } /// Writes [`Self::series_key_for`] into `buf`, replacing its contents. - pub fn write_series_key_for(&self, buf: &mut String, obs: &Observation) { + pub(crate) fn write_series_key_for(&self, buf: &mut String, obs: &Observation) { use crate::matchers::write_series_key; if self.global_aggregation { write_series_key(buf, self.agg_id, &[], &[], &[]); diff --git a/asap-precompute-rs/src/matchers.rs b/asap-precompute-rs/src/matchers.rs index 23b3378..8a8c5fe 100644 --- a/asap-precompute-rs/src/matchers.rs +++ b/asap-precompute-rs/src/matchers.rs @@ -143,7 +143,7 @@ pub fn series_key( /// Writes [`series_key`] into `buf`, replacing its contents, so a hot path /// can reuse one buffer. -pub fn write_series_key( +pub(crate) fn write_series_key( buf: &mut String, agg_id: AggId, resource_labels: &[KeyValue], diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index d022039..ed4cffb 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -641,14 +641,22 @@ mod tests { let mut points = 0usize; let decode = timed(&mut |_| points += 1); let mut split = crate::observation::Observation::default(); - let mut keys: Vec> = Vec::new(); let decode_split = timed(&mut |obs| { - crate::matchers::per_attribute::<()>(obs, &plugin.per_attribute, &mut split, |o| { - keys.push(o.value.bytes.clone()); - Ok(()) - }) - .unwrap() + crate::matchers::per_attribute::<()>(obs, &plugin.per_attribute, &mut split, |_| Ok(())) + .unwrap() }); + // The keys for the CMS-only timing, collected outside any timed run. + let mut keys: Vec> = Vec::new(); + for b in &batches { + crate::otap::codec::decode_each(b.clone(), |obs| { + crate::matchers::per_attribute::<()>(obs, &plugin.per_attribute, &mut split, |o| { + keys.push(o.value.bytes.clone()); + Ok(()) + }) + .unwrap() + }) + .unwrap(); + } let all = timed(&mut |obs| { crate::matchers::per_attribute(obs, &plugin.per_attribute, &mut split, |o| { precompute.observe(o) diff --git a/asap-precompute-rs/src/window.rs b/asap-precompute-rs/src/window.rs index 047c7d1..01dd7c4 100644 --- a/asap-precompute-rs/src/window.rs +++ b/asap-precompute-rs/src/window.rs @@ -198,13 +198,8 @@ impl WindowState { last_seen_ms: obs.timestamp_ms, count: 0, }; - self.series.insert(key.clone(), entry); stats.active_series += 1; - - let entry = self - .series - .get_mut(&key) - .expect("series entry must exist after insert"); + let entry = self.series.entry(key).or_insert(entry); observer.observe(entry.sketch.as_mut(), obs)?; entry.count += 1; Ok(())