diff --git a/benches/decode.rs b/benches/decode.rs index 4a848cf..17b0372 100644 --- a/benches/decode.rs +++ b/benches/decode.rs @@ -27,7 +27,7 @@ use std::pin::{Pin, pin}; use std::task::{Context, Poll, Waker}; use benchfmt::{CountingAlloc, Suite}; -use expfmt::arrow::{ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep}; +use expfmt::arrow::{ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep, StalenessTracker}; use expfmt::format::Format; #[global_allocator] @@ -216,6 +216,32 @@ fn main() { } } }, + ) + .bench( + format!("Decode/case={case}/mode=decode_staleness"), + move |b| { + let mut decoder = ArrowDecoder::new(); + let mut tracker = StalenessTracker::new(); + black_box( + decoder + .decode_with_staleness(payload, &Format::TEXT, &opts, &mut tracker) + .unwrap(), + ); + + b.set_bytes(bytes); + b.report_allocs(); + while b.keep_running() { + let records = decoder + .decode_with_staleness( + black_box(payload), + &Format::TEXT, + &opts, + &mut tracker, + ) + .unwrap(); + black_box(records); + } + }, ); } diff --git a/specs/SPEC.md b/specs/SPEC.md index 839466d..d30bcb4 100644 --- a/specs/SPEC.md +++ b/specs/SPEC.md @@ -128,7 +128,8 @@ scrape bytes 1. **Stage 1 (`__name__` series filter pushdown)**: Leading `keep`, `drop`, `keepequal`, and `dropequal` rules that inspect only `__name__` will run once per contiguous wire series-name transition (e.g. `foo_bucket`, `foo_sum`, `foo_count`, matching Prometheus `scrape/scrape.go`) so dropped series runs skip float parsing, label unescaping, string interning, and family assembly. 2. **Stage 2 (dictionary-vectorized relabeling and `honor_labels`)**: Once kept series have been assembled into metric families (collapsing classic histogram `_bucket`, `_sum`, and `_count` lines into a single `HistogramDataPoints` row) and their label keys and values have been deduplicated in `key_interner` and `val_interner`, attribute-level `honor_labels` collision resolution, `labeldrop` / `labelkeep`, and single-label `keep` / `drop` / value transformations (`replace`, `lowercase`, `uppercase`, `hashmod`) will execute across unique dictionary entries (memoized per `source_labels[0]` value ID) instead of $O(N_{\text{samples}})$. - The metric family assembler groups classic histogram `_bucket`, `_sum`, and `_count` lines and summary `quantile`, `_sum`, and `_count` lines (plus OpenMetrics `_created` start timestamps) into complete metric families. -- Configurable translation policies control counter `_total` and unit suffix handling and untyped series handling. +- Configurable translation policies control counter `_total` and unit suffix handling, untyped series handling, and explicit-timestamp staleness tracking (`track_timestamps_staleness`, `false` by default to match Prometheus `scrape/scrape.go`). +- Optional per-target series staleness tracking via `StalenessTracker` (`ArrowDecoder::decode_with_staleness`, `decode_stream_with_staleness`, `step_decode_with_staleness`, `finish_decode_with_staleness`, and `flush_target_stale`) detects disappeared time series across scrapes as well as incoming `STALE_NAN_BITS` (`0x7ff0_0000_0000_0002`) markers. Stale points set `DATA_POINT_FLAG_NO_RECORDED_VALUE` (`1`) in a lazily materialized `flags` (`UInt32`) column and omit `double_value` and `histogram_sum` (`NULL` in Arrow, `0.0` for `summary_sum`), leaving non-stale batch schemas unchanged. `StalenessTracker` interns active series labels in contiguous arenas with zero steady-state heap allocations, rolls back uncommitted changes on parse errors or dropped streams, and compacts deactivated series under label churn so per-target live heap stays proportional to active series. Callers that do not pass a `StalenessTracker` (`decode`, `decode_stream`, `step_decode`) pay zero cross-scrape memory overhead. - `ArrowDecoder` owns its scratch buffers and Arrow builders without internal locks or global state, so a per-core `!Send` receiver can reuse allocations across scrapes with bounded memory. ### Target and metric relabeling architecture diff --git a/src/arrow.rs b/src/arrow.rs index 046f020..6503506 100644 --- a/src/arrow.rs +++ b/src/arrow.rs @@ -25,6 +25,7 @@ mod intern; mod options; mod records; mod schema; +mod staleness; #[cfg(test)] mod tests; @@ -54,6 +55,10 @@ pub use self::options::{ }; pub use self::records::{MetricPayloadType, OtapMetricRecords}; pub use self::schema::PROMETHEUS_TYPE_METADATA_KEY; +pub use self::staleness::{ + DATA_POINT_FLAG_NO_RECORDED_VALUE, STALE_NAN, STALE_NAN_BITS, StalenessTracker, is_stale_nan, +}; +use self::staleness::{DpKind, StaleDrainStep}; /// Maximum distinct `UnivariateMetrics` rows representable in a single OTAP batch (`u16` IDs). const MAX_METRICS_PER_BATCH: usize = (u16::MAX as usize) + 1; @@ -142,6 +147,7 @@ pub struct DecodeCursor { pos: usize, chunked: bool, pending_series: bool, + tracker_started: bool, done: bool, } @@ -414,7 +420,8 @@ impl ArrowDecoder { let mut pending_series = false; let mut done = false; - let res = self.ingest_entries::( + let mut tracker = None; + let res = self.ingest_entries::( &mut parser, opts, default_ts_ns, @@ -422,6 +429,7 @@ impl ArrowDecoder { None, &mut pending_series, &mut done, + &mut tracker, ); parser.take_scratch_into(self); match res? { @@ -539,14 +547,38 @@ impl ArrowDecoder { input: &[u8], budget: DecodeBudget, ) -> Result { - self.step_cursor::(cursor, input, Some(budget)) + let mut tracker = None; + self.step_cursor::(cursor, input, Some(budget), &mut tracker) + } + + /// Advances `cursor` over `input` by at most `budget` entries, bytes, or duration while + /// tracking active series in `tracker` and draining disappeared stale series cooperatively at + /// end-of-input. + /// + /// If `cursor` is abandoned early before completion, call + /// [`ArrowDecoder::finish_decode_with_staleness`] to roll back uncommitted `tracker` state and + /// reclaim internal parser scratch buffers. + /// + /// # Errors + /// + /// Returns [`DecodeError`] under the same conditions as [`ArrowDecoder::step_decode`]. + pub fn step_decode_with_staleness( + &mut self, + cursor: &mut DecodeCursor, + input: &[u8], + budget: DecodeBudget, + tracker: &mut StalenessTracker, + ) -> Result { + let mut tracker_opt = Some(tracker); + self.step_cursor::(cursor, input, Some(budget), &mut tracker_opt) } - fn step_cursor( + fn step_cursor( &mut self, cursor: &mut DecodeCursor, input: &[u8], budget: Option, + tracker: &mut Option<&mut StalenessTracker>, ) -> Result { if cursor.done { return Ok(DecodeStep::Done); @@ -560,15 +592,28 @@ impl ArrowDecoder { } let expected_len = *cursor.input_len.get_or_insert(input.len()); if input.len() != expected_len { + if TRACKED + && cursor.tracker_started + && let Some(tr) = tracker.as_deref_mut() + { + tr.abort_scrape(); + } self.finish_decode(cursor); return Err(DecodeError::InvalidCursor( "scrape input buffer length changed across decode steps", )); } + if TRACKED + && !cursor.tracker_started + && let Some(tr) = tracker.as_deref_mut() + { + tr.begin_scrape(); + cursor.tracker_started = true; + } let state = std::mem::replace(&mut cursor.parser_state, ActiveParserState::Invalid); let mut parser = state.into_parser(input); let res = if cursor.chunked { - self.ingest_entries::( + self.ingest_entries::( &mut parser, &cursor.opts, cursor.default_ts_ns, @@ -576,9 +621,10 @@ impl ArrowDecoder { budget, &mut cursor.pending_series, &mut cursor.done, + tracker, ) } else { - self.ingest_entries::( + self.ingest_entries::( &mut parser, &cursor.opts, cursor.default_ts_ns, @@ -586,6 +632,7 @@ impl ArrowDecoder { budget, &mut cursor.pending_series, &mut cursor.done, + tracker, ) }; cursor.pos = parser.pos(); @@ -600,6 +647,9 @@ impl ArrowDecoder { } Err(err) => { cursor.done = true; + if TRACKED && let Some(tr) = tracker.as_deref_mut() { + tr.abort_scrape(); + } parser.take_scratch_into(self); self.reset_scrape(); Err(err) @@ -619,6 +669,80 @@ impl ArrowDecoder { } } + /// Ends an active [`DecodeCursor`] driven with [`ArrowDecoder::step_decode_with_staleness`], + /// rolling back uncommitted state in `tracker` and reclaiming parser scratch buffers into + /// `self`. + pub fn finish_decode_with_staleness( + &mut self, + cursor: &mut DecodeCursor, + tracker: &mut StalenessTracker, + ) { + if !cursor.done && cursor.scrape_id == self.scrape_id && cursor.tracker_started { + tracker.abort_scrape(); + } + self.finish_decode(cursor); + } + + /// Decodes a single exposition scrape buffer of `format` into OTAP metric + /// [`arrow_array::RecordBatch`]es while tracking active series in `tracker` and emitting stale + /// data points (`flags = DATA_POINT_FLAG_NO_RECORDED_VALUE` with omitted values) for any + /// series that was active in the previous scrape of this target and disappeared from `input`. + /// + /// Series present in `input` whose sample value already carries [`STALE_NAN`] are emitted with + /// `flags = DATA_POINT_FLAG_NO_RECORDED_VALUE` (and omitted values) and removed from + /// `tracker`'s active set. If decoding fails with [`DecodeError`], `tracker` rolls back to the + /// active series set from the previous successful scrape. + /// + /// # Errors + /// + /// Returns [`DecodeError`] under the same conditions as [`ArrowDecoder::decode`]. + pub fn decode_with_staleness( + &mut self, + input: &[u8], + format: &Format, + opts: &DecodeOptions, + tracker: &mut StalenessTracker, + ) -> Result { + self.reset_scrape(); + tracker.begin_scrape(); + let (mut parser, default_ts_ns) = match self.init_scrape_parser(input, format, opts) { + Ok(v) => v, + Err(err) => { + tracker.abort_scrape(); + return Err(err); + } + }; + + let mut pending_series = false; + let mut done = false; + let mut tracker_opt = Some(tracker); + let res = self.ingest_entries::( + &mut parser, + opts, + default_ts_ns, + 0, + None, + &mut pending_series, + &mut done, + &mut tracker_opt, + ); + parser.take_scratch_into(self); + match res { + Ok(DecodeStep::Batch(records)) => Ok(records), + Ok(DecodeStep::Done) => Ok(OtapMetricRecords::default()), + Ok(DecodeStep::Pending) => { + unreachable!("unbudgeted decode never returns DecodeStep::Pending") + } + Err(err) => { + if let Some(tr) = tracker_opt { + tr.abort_scrape(); + } + self.reset_scrape(); + Err(err) + } + } + } + /// Streams `input` as a sequence of non-empty [`OtapMetricRecords`] chunks bounded by /// [`DecodeOptions::max_data_points_per_batch`] and OTAP's 65,536-metric per-batch `u16` ID /// limit. @@ -633,18 +757,110 @@ impl ArrowDecoder { input: &'input [u8], format: &Format, opts: &DecodeOptions, + ) -> DecodeStream<'dec, 'input> { + self.decode_stream_inner(input, format, opts, None) + } + + /// Streams `input` as a sequence of non-empty [`OtapMetricRecords`] chunks bounded by + /// [`DecodeOptions::max_data_points_per_batch`] while tracking active series in `tracker` and + /// emitting stale data points (`flags = DATA_POINT_FLAG_NO_RECORDED_VALUE` with omitted + /// values) at end-of-scrape for any series that disappeared since the previous scrape. + /// + /// If a stream chunk returns [`DecodeError`] or [`DecodeStream`] is dropped before reaching + /// end-of-input, `tracker` rolls back to the committed active series set from the previous + /// successful scrape. + pub fn decode_stream_with_staleness<'dec, 'input>( + &'dec mut self, + input: &'input [u8], + format: &Format, + opts: &DecodeOptions, + tracker: &'dec mut StalenessTracker, + ) -> DecodeStream<'dec, 'input> { + self.decode_stream_inner(input, format, opts, Some(tracker)) + } + + /// Emits stale data points (`flags = DATA_POINT_FLAG_NO_RECORDED_VALUE` with omitted values) + /// at `opts.timestamp_ms` for every currently active series in `tracker` and clears the active + /// set. + /// + /// Call this when an entire target scrape fails (for example, on an HTTP or connection error) + /// or when a scrape target is decommissioned so downstream consumers mark all previously + /// active series for that target stale without re-emitting them on subsequent scrapes. + /// + /// # Errors + /// + /// Returns [`DecodeError`] if `opts.timestamp_ms` overflows `i64` nanoseconds or an OTAP table + /// exceeds its integer ID width. + pub fn flush_target_stale( + &mut self, + tracker: &mut StalenessTracker, + opts: &DecodeOptions, + ) -> Result { + self.reset_scrape(); + if tracker.is_empty() { + return Ok(OtapMetricRecords::default()); + } + let default_ts_ns = match opts.timestamp_ms { + Some(ms) => Some(ms_to_ns(ms)?), + None => None, + }; + tracker.begin_scrape(); + let res = tracker.finish_scrape_step( + &mut self.metrics, + &mut self.num_dp, + &mut self.sum_dp, + &mut self.hist_dp, + &mut self.key_interner, + &mut self.val_interner, + opts, + default_ts_ns, + None, + MAX_METRICS_PER_BATCH, + None, + ); + if let Err(err) = res { + tracker.abort_scrape(); + self.reset_scrape(); + return Err(err); + } + let batches = match self.finish_batches(default_ts_ns.is_some()) { + Ok(b) => b, + Err(err) => { + tracker.abort_scrape(); + self.reset_scrape(); + return Err(err); + } + }; + tracker.commit_scrape(); + Ok(batches) + } + + fn decode_stream_inner<'dec, 'input>( + &'dec mut self, + input: &'input [u8], + format: &Format, + opts: &DecodeOptions, + mut tracker: Option<&'dec mut StalenessTracker>, ) -> DecodeStream<'dec, 'input> { match self.start_decode_stream(format, opts) { Ok(mut cursor) => { + if let Some(tr) = tracker.as_deref_mut() { + tr.begin_scrape(); + cursor.tracker_started = true; + } cursor.input_len = Some(input.len()); DecodeStream { decoder: self, input, cursor, + tracker, init_err: None, } } Err(err) => { + if let Some(tr) = tracker.as_deref_mut() { + tr.abort_scrape(); + } let max_data_points = opts .max_data_points_per_batch .clamp(1, (u32::MAX as usize).saturating_add(1)); @@ -662,8 +878,10 @@ impl ArrowDecoder { pos: 0, chunked: true, pending_series: false, + tracker_started: false, done: false, }, + tracker, init_err: Some(err), } } @@ -694,6 +912,7 @@ impl ArrowDecoder { pos: 0, chunked, pending_series: false, + tracker_started: false, done: false, }) } @@ -743,6 +962,17 @@ impl ArrowDecoder { self.hist_dp.clear(); } + #[inline] + fn reset_batch_for_mode( + &mut self, + tracker: &mut Option<&mut StalenessTracker>, + ) { + self.reset_batch(); + if TRACKED && let Some(tr) = tracker.as_deref_mut() { + tr.reset_batch_ids(); + } + } + #[inline] fn total_data_points(&self) -> usize { self.num_dp.base.len() + self.sum_dp.base.len() + self.hist_dp.base.len() @@ -754,26 +984,36 @@ impl ArrowDecoder { } #[inline] - fn flush_pending_family( + fn flush_pending_family( &mut self, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { + let tr = if TRACKED { + tracker.as_deref_mut() + } else { + None + }; self.assembly.flush( &mut self.current_family, &mut self.metrics, &mut self.sum_dp, &mut self.hist_dp, + &self.key_interner, + &self.val_interner, + tr, opts, default_ts_ns, ) } #[inline] - fn transition_family_for_metadata( + fn transition_family_for_metadata( &mut self, raw_name: &[u8], name_offset: usize, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -795,19 +1035,18 @@ impl ArrowDecoder { self.series_name_buf.clear(); self.series_name_buf.extend_from_slice(unescaped); - self.flush_pending_family(opts, default_ts_ns)?; + self.flush_pending_family::(tracker, opts, default_ts_ns)?; self.current_family .start_new(raw_name, &self.series_name_buf); Ok(()) } - // Monomorphizing on `CHUNKED` and `BUDGETED` shares one entry dispatch loop across - // synchronous single-batch `decode` (`false, false`), synchronous `decode_stream` - // (`true, false`), cooperative single-batch `decode_cooperative` (`false, true`), and - // cooperative streaming `DecodeStream::step` (`true, true`), compiling out unused checks on + // Monomorphizing on `CHUNKED`, `BUDGETED`, and `TRACKED` shares one entry dispatch loop across + // synchronous single-batch `decode`, streaming `decode_stream`, cooperative decoding, and + // their `_with_staleness` counterparts while compiling out unused checks and register state on // each path. #[allow(clippy::too_many_arguments)] - fn ingest_entries( + fn ingest_entries( &mut self, parser: &mut ActiveParser<'_>, opts: &DecodeOptions, @@ -816,6 +1055,7 @@ impl ArrowDecoder { budget: Option, pending_series: &mut bool, done: &mut bool, + tracker: &mut Option<&mut StalenessTracker>, ) -> Result { let start_pos = if BUDGETED { parser.pos() } else { 0 }; let max_entries = if let Some(b) = budget.filter(|_| BUDGETED) { @@ -839,14 +1079,53 @@ impl ArrowDecoder { match parser.next_entry().map_err(DecodeError::from_parse)? { Some(entry) => entry, None => { - self.flush_pending_family(opts, default_ts_ns)?; - *done = true; + self.flush_pending_family::(tracker, opts, default_ts_ns)?; + let drain_step = if TRACKED { + let step_budget = budget.filter(|_| BUDGETED).and_then(|b| { + started.map(|st| { + (max_entries.saturating_sub(entries), st, b.max_duration) + }) + }); + if let Some(tr) = tracker.as_deref_mut() { + tr.finish_scrape_step( + &mut self.metrics, + &mut self.num_dp, + &mut self.sum_dp, + &mut self.hist_dp, + &mut self.key_interner, + &mut self.val_interner, + opts, + default_ts_ns, + CHUNKED.then_some(max_data_points), + MAX_METRICS_PER_BATCH, + step_budget, + )? + } else { + StaleDrainStep::Complete + } + } else { + StaleDrainStep::Complete + }; + if TRACKED && drain_step == StaleDrainStep::YieldPending { + return Ok(DecodeStep::Pending); + } + let has_more_stale = TRACKED && drain_step == StaleDrainStep::ChunkFull; if self.metrics.is_empty() { + *done = true; + if TRACKED && let Some(tr) = tracker.as_deref_mut() { + tr.commit_scrape(); + } return Ok(DecodeStep::Done); } let batch = self.finish_batches(default_ts_ns.is_some())?; + if !has_more_stale { + *done = true; + if TRACKED && let Some(tr) = tracker.as_deref_mut() { + tr.commit_scrape(); + } + } if CHUNKED { - self.reset_batch(); + self.reset_batch_for_mode::(tracker); } return Ok(DecodeStep::Batch(batch)); } @@ -856,9 +1135,15 @@ impl ArrowDecoder { match entry { // `Entry::Histogram` arrives with #22. Entry::Comment | Entry::Histogram => {} - Entry::Help => self.ingest_help_entry(parser, opts, default_ts_ns)?, - Entry::Type => self.ingest_type_entry(parser, opts, default_ts_ns)?, - Entry::Unit => self.ingest_unit_entry(parser, opts, default_ts_ns)?, + Entry::Help => { + self.ingest_help_entry::(parser, tracker, opts, default_ts_ns)? + } + Entry::Type => { + self.ingest_type_entry::(parser, tracker, opts, default_ts_ns)? + } + Entry::Unit => { + self.ingest_unit_entry::(parser, tracker, opts, default_ts_ns)? + } Entry::Series => { let (raw_name, name_offset) = { let (rn, _, _, _, _) = parser.raw_series_parts(); @@ -887,13 +1172,13 @@ impl ArrowDecoder { } else { self.series_name_buf.clear(); self.series_name_buf.extend_from_slice(unescaped); - self.flush_pending_family(opts, default_ts_ns)?; + self.flush_pending_family::(tracker, opts, default_ts_ns)?; if CHUNKED && self.total_data_points() >= max_data_points { self.current_family .start_new(raw_name, &self.series_name_buf); *pending_series = true; let batch = self.finish_batches(default_ts_ns.is_some())?; - self.reset_batch(); + self.reset_batch_for_mode::(tracker); return Ok(DecodeStep::Batch(batch)); } self.current_family @@ -912,23 +1197,29 @@ impl ArrowDecoder { && self.metrics.len() >= MAX_METRICS_PER_BATCH)); if split_assembled || split_scalar { if split_assembled { - self.flush_pending_family(opts, default_ts_ns)?; + self.flush_pending_family::(tracker, opts, default_ts_ns)?; } *pending_series = true; let batch = self.finish_batches(default_ts_ns.is_some())?; - self.reset_batch(); + self.reset_batch_for_mode::(tracker); return Ok(DecodeStep::Batch(batch)); } } - self.ingest_series_entry(parser, role, opts, default_ts_ns)?; + self.ingest_series_entry::( + parser, + role, + tracker, + opts, + default_ts_ns, + )?; if CHUNKED && self.assembly.is_empty() && self.total_data_points() >= max_data_points { let batch = self.finish_batches(default_ts_ns.is_some())?; - self.reset_batch(); + self.reset_batch_for_mode::(tracker); return Ok(DecodeStep::Batch(batch)); } } @@ -948,9 +1239,10 @@ impl ArrowDecoder { } #[inline] - fn ingest_help_entry( + fn ingest_help_entry( &mut self, parser: &ActiveParser<'_>, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -966,7 +1258,13 @@ impl ArrowDecoder { } return Ok(()); } - self.transition_family_for_metadata(raw_name, name_offset, opts, default_ts_ns)?; + self.transition_family_for_metadata::( + raw_name, + name_offset, + tracker, + opts, + default_ts_ns, + )?; if self.current_family.has_help || self.current_family.has_samples || !self.assembly.is_empty() @@ -984,15 +1282,22 @@ impl ArrowDecoder { } #[inline] - fn ingest_type_entry( + fn ingest_type_entry( &mut self, parser: &ActiveParser<'_>, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { let (raw_name, metric_type) = parser.metric_type(); let name_offset = parser.name_offset(); - self.transition_family_for_metadata(raw_name, name_offset, opts, default_ts_ns)?; + self.transition_family_for_metadata::( + raw_name, + name_offset, + tracker, + opts, + default_ts_ns, + )?; if self.current_family.has_type || self.current_family.has_samples || !self.assembly.is_empty() @@ -1007,9 +1312,10 @@ impl ArrowDecoder { } #[inline] - fn ingest_unit_entry( + fn ingest_unit_entry( &mut self, parser: &ActiveParser<'_>, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -1018,7 +1324,13 @@ impl ArrowDecoder { if raw_unit.is_empty() { return Ok(()); } - self.transition_family_for_metadata(raw_name, name_offset, opts, default_ts_ns)?; + self.transition_family_for_metadata::( + raw_name, + name_offset, + tracker, + opts, + default_ts_ns, + )?; if self.current_family.has_unit || self.current_family.has_samples || !self.assembly.is_empty() @@ -1035,15 +1347,18 @@ impl ArrowDecoder { } #[inline] - fn ingest_series_entry( + fn ingest_series_entry( &mut self, parser: &mut ActiveParser<'_>, role: SeriesRole, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { match role { - SeriesRole::Scalar => self.append_scalar_series(parser, opts, default_ts_ns), + SeriesRole::Scalar => { + self.append_scalar_series::(parser, tracker, opts, default_ts_ns) + } SeriesRole::HistogramBucket | SeriesRole::HistogramSum | SeriesRole::HistogramCount @@ -1054,9 +1369,10 @@ impl ArrowDecoder { } #[inline] - fn append_scalar_series( + fn append_scalar_series( &mut self, parser: &mut ActiveParser<'_>, + tracker: &mut Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -1089,7 +1405,8 @@ impl ArrowDecoder { self.extract_series_labels_into_temp(label_spans, full_input, None)?; self.num_dp.attrs.extend_for_parent(dp_id, &self.temp_attrs); - if let Some((ex_val, ex_ts_ms, ex_spans)) = exemplar_parts { + let is_stale = is_stale_nan(value); + if !is_stale && let Some((ex_val, ex_ts_ms, ex_spans)) = exemplar_parts { self.append_number_exemplar( dp_id, ex_val, @@ -1108,10 +1425,34 @@ impl ArrowDecoder { Some(ms) => Some(ms_to_ns(ms)?), None => default_ts_ns, }; - self.num_dp - .base - .push(parent_id, start_ts_ns, resolved_ts_ns); - self.num_dp.values.push(value); + if !is_stale { + self.num_dp + .base + .push(parent_id, start_ts_ns, resolved_ts_ns); + self.num_dp.values.push_value(value); + } else { + self.num_dp.base.push_with_flags( + parent_id, + start_ts_ns, + resolved_ts_ns, + DATA_POINT_FLAG_NO_RECORDED_VALUE, + ); + self.num_dp.values.push_null(dp_index); + } + + if TRACKED + && (sample_ts_ms.is_none() || opts.track_timestamps_staleness) + && let Some(tr) = tracker.as_deref_mut() + { + tr.record_series( + &mut self.current_family, + DpKind::Number, + &self.temp_attrs, + &self.key_interner, + &self.val_interner, + is_stale, + )?; + } Ok(()) } @@ -1158,6 +1499,9 @@ impl ArrowDecoder { if group.start_ts_ns.is_none() && om_start_ts_ns.is_some() { group.start_ts_ns = om_start_ts_ns; } + if is_stale_nan(value) { + group.is_stale = true; + } match role { SeriesRole::HistogramBucket | SeriesRole::SummaryQuantile => { @@ -1617,6 +1961,7 @@ pub struct DecodeStream<'dec, 'input> { decoder: &'dec mut ArrowDecoder, input: &'input [u8], cursor: DecodeCursor, + tracker: Option<&'dec mut StalenessTracker>, init_err: Option, } @@ -1654,10 +1999,28 @@ impl DecodeStream<'_, '_> { } if let Some(err) = self.init_err.take() { self.cursor.done = true; + if let Some(tr) = self.tracker.as_deref_mut() { + tr.abort_scrape(); + } return Err(err); } - self.decoder - .step_decode(&mut self.cursor, self.input, budget) + if self.tracker.is_some() { + let mut tracker = self.tracker.as_deref_mut(); + self.decoder.step_cursor::( + &mut self.cursor, + self.input, + Some(budget), + &mut tracker, + ) + } else { + let mut tracker = None; + self.decoder.step_cursor::( + &mut self.cursor, + self.input, + Some(budget), + &mut tracker, + ) + } } /// Cooperatively pulls the next non-empty [`OtapMetricRecords`] chunk from the stream, @@ -1687,12 +2050,29 @@ impl Iterator for DecodeStream<'_, '_> { } if let Some(err) = self.init_err.take() { self.cursor.done = true; + if let Some(tr) = self.tracker.as_deref_mut() { + tr.abort_scrape(); + } return Some(Err(err)); } - match self - .decoder - .step_cursor::(&mut self.cursor, self.input, None) - { + let step_res = if self.tracker.is_some() { + let mut tracker = self.tracker.as_deref_mut(); + self.decoder.step_cursor::( + &mut self.cursor, + self.input, + None, + &mut tracker, + ) + } else { + let mut tracker = None; + self.decoder.step_cursor::( + &mut self.cursor, + self.input, + None, + &mut tracker, + ) + }; + match step_res { Ok(DecodeStep::Batch(batch)) => Some(Ok(batch)), Ok(DecodeStep::Done) => None, Ok(DecodeStep::Pending) => { @@ -1707,6 +2087,9 @@ impl FusedIterator for DecodeStream<'_, '_> {} impl Drop for DecodeStream<'_, '_> { fn drop(&mut self) { + if let Some(tr) = self.tracker.as_deref_mut() { + tr.abort_scrape(); + } self.decoder.finish_decode(&mut self.cursor); } } diff --git a/src/arrow/columns.rs b/src/arrow/columns.rs index b5dd5fd..24a908f 100644 --- a/src/arrow/columns.rs +++ b/src/arrow/columns.rs @@ -519,13 +519,78 @@ impl SparseTimestampCol { } } -/// Shared leading columns (`id`, `parent_id`, `start_time_unix_nano`, `time_unix_nano`) across -/// `NumberDataPoints`, `SummaryDataPoints`, and `HistogramDataPoints`. +/// Staging buffer for an optional `flags` (`UInt32`) column on data point tables that stays +/// unallocated when every row in the batch has `flags == 0`. +/// +/// OTAP marks `flags` as `default_values_optional` on `NumberDataPoints`, `SummaryDataPoints`, and +/// `HistogramDataPoints`, omitting the column whenever all rows carry `0`. Keeping `values` +/// unallocated until the first non-zero flag arrives avoids per-batch `UInt32` buffer allocations +/// and preserves identical schemas on non-stale scrapes (>99.9% of batches), while backfilling +/// leading `0` entries if a stale point appears mid-batch. +#[derive(Default)] +pub(super) struct SparseFlagsCol { + values: Vec, +} + +impl SparseFlagsCol { + pub(super) fn clear(&mut self) { + self.values.clear(); + } + + pub(super) fn retained_memory_bytes(&self) -> usize { + self.values.capacity() * size_of::() + } + + #[inline] + pub(super) fn is_empty(&self) -> bool { + self.values.is_empty() + } + + #[inline(always)] + pub(super) fn push_zero(&mut self) { + if !self.values.is_empty() { + self.push_zero_slow(); + } + } + + #[cold] + fn push_zero_slow(&mut self) { + self.values.push(0); + } + + #[cold] + pub(super) fn push_nonzero(&mut self, row_index: usize, flags: u32) { + debug_assert_ne!(flags, 0); + if row_index > 0 && self.values.is_empty() { + self.values.resize(row_index, 0); + } + self.values.push(flags); + } + + #[inline] + pub(super) fn push(&mut self, row_index: usize, flags: u32) { + if flags == 0 { + self.push_zero(); + } else { + self.push_nonzero(row_index, flags); + } + } + + #[inline] + pub(super) fn to_array(&self) -> ArrayRef { + Arc::new(UInt32Array::new(scalar_from_slice(&self.values), None)) + } +} + +/// Shared leading columns (`id`, `parent_id`, `start_time_unix_nano`, `time_unix_nano`) and +/// optional trailing `flags` column across `NumberDataPoints`, `SummaryDataPoints`, and +/// `HistogramDataPoints`. #[derive(Default)] pub(super) struct DpBaseCols { pub(super) parent_ids: Vec, pub(super) start_time: SparseTimestampCol, pub(super) time: SparseTimestampCol, + pub(super) flags: SparseFlagsCol, } impl DpBaseCols { @@ -534,6 +599,7 @@ impl DpBaseCols { parent_ids: Vec::with_capacity(cap), start_time: SparseTimestampCol::default(), time: SparseTimestampCol::default(), + flags: SparseFlagsCol::default(), } } @@ -541,12 +607,14 @@ impl DpBaseCols { self.parent_ids.clear(); self.start_time.clear(); self.time.clear(); + self.flags.clear(); } pub(super) fn retained_memory_bytes(&self) -> usize { self.parent_ids.capacity() * size_of::() + self.start_time.retained_memory_bytes() + self.time.retained_memory_bytes() + + self.flags.retained_memory_bytes() } #[inline] @@ -559,12 +627,33 @@ impl DpBaseCols { self.parent_ids.is_empty() } + #[inline] + pub(super) fn has_flags(&self) -> bool { + !self.flags.is_empty() + } + #[inline] pub(super) fn push(&mut self, parent_id: u16, start_ts_ns: Option, ts_ns: Option) { let dp_index = self.parent_ids.len(); self.parent_ids.push(parent_id); self.start_time.push(dp_index, start_ts_ns); self.time.push(dp_index, ts_ns); + self.flags.push_zero(); + } + + #[inline] + pub(super) fn push_with_flags( + &mut self, + parent_id: u16, + start_ts_ns: Option, + ts_ns: Option, + flags: u32, + ) { + let dp_index = self.parent_ids.len(); + self.parent_ids.push(parent_id); + self.start_time.push(dp_index, start_ts_ns); + self.time.push(dp_index, ts_ns); + self.flags.push(dp_index, flags); } #[inline] @@ -609,6 +698,13 @@ impl DpBaseCols { cols.push(self.time.to_array(time_nullable)); } } + + #[inline] + pub(super) fn append_flags_column(&self, cols: &mut Vec) { + if !self.flags.is_empty() { + cols.push(self.flags.to_array()); + } + } } /// Staging buffer for a dense column (`count` or `sum` on summaries and histograms) whose validity @@ -620,6 +716,13 @@ pub(super) struct OptionalValueCol { } impl OptionalValueCol { + pub(super) fn with_capacity(cap: usize) -> Self { + Self { + values: Vec::with_capacity(cap), + valid: Vec::new(), + } + } + pub(super) fn clear(&mut self) { self.values.clear(); self.valid.clear(); @@ -634,22 +737,33 @@ impl OptionalValueCol { !self.valid.is_empty() } + #[inline(always)] + pub(super) fn push_value(&mut self, val: T) { + self.values.push(val); + if !self.valid.is_empty() { + self.push_valid_slow(); + } + } + + #[cold] + fn push_valid_slow(&mut self) { + self.valid.push(true); + } + + #[cold] + pub(super) fn push_null(&mut self, row_index: usize) { + if self.valid.is_empty() { + self.valid.resize(row_index, true); + } + self.values.push(T::default()); + self.valid.push(false); + } + #[inline] pub(super) fn push(&mut self, row_index: usize, val: Option) { match val { - Some(v) => { - self.values.push(v); - if !self.valid.is_empty() { - self.valid.push(true); - } - } - None => { - if self.valid.is_empty() { - self.valid.resize(row_index, true); - } - self.values.push(T::default()); - self.valid.push(false); - } + Some(v) => self.push_value(v), + None => self.push_null(row_index), } } @@ -926,7 +1040,7 @@ impl ExemplarTable { /// `NumberDpExemplarAttrs`. pub(super) struct NumberDpTable { pub(super) base: DpBaseCols, - pub(super) values: Vec, + pub(super) values: OptionalValueCol, pub(super) attrs: AttrTable, pub(super) exemplars: ExemplarTable, } @@ -935,7 +1049,7 @@ impl NumberDpTable { pub(super) fn with_capacity(data_points: usize, attrs: usize) -> Self { Self { base: DpBaseCols::with_capacity(data_points), - values: Vec::with_capacity(data_points), + values: OptionalValueCol::with_capacity(data_points), attrs: AttrTable::with_capacity(attrs), exemplars: ExemplarTable::default(), } @@ -950,7 +1064,7 @@ impl NumberDpTable { pub(super) fn retained_memory_bytes(&self) -> usize { self.base.retained_memory_bytes() - + self.values.capacity() * size_of::() + + self.values.retained_memory_bytes() + self.attrs.retained_memory_bytes() + self.exemplars.retained_memory_bytes() } @@ -973,14 +1087,14 @@ impl NumberDpTable { start_time_nullable, has_time, time_nullable, + self.values.is_nullable(), + self.base.has_flags(), ); let mut cols: Vec = Vec::with_capacity(schema.fields().len()); self.base .append_columns(&mut cols, has_id, has_default_ts, &mut scratch.u32_buf); - cols.push(Arc::new(Float64Array::new( - scalar_from_slice(&self.values), - None, - ))); + cols.push(self.values.to_f64_array()); + self.base.append_flags_column(&mut cols); Ok(Some(RecordBatch::try_new(schema, cols)?)) } @@ -1041,6 +1155,7 @@ impl SummaryDpTable { time_nullable, count_nullable, sum_nullable, + self.base.has_flags(), ); let mut cols: Vec = Vec::with_capacity(schema.fields().len()); self.base @@ -1069,6 +1184,7 @@ impl SummaryDpTable { None, ); cols.push(Arc::new(q_list)); + self.base.append_flags_column(&mut cols); Ok(Some(RecordBatch::try_new(schema, cols)?)) } @@ -1136,6 +1252,7 @@ impl HistogramDpTable { time_nullable, count_nullable, sum_nullable, + self.base.has_flags(), ); let mut cols: Vec = Vec::with_capacity(schema.fields().len()); self.base @@ -1164,6 +1281,7 @@ impl HistogramDpTable { None, ); cols.push(Arc::new(explicit_bounds_list)); + self.base.append_flags_column(&mut cols); Ok(Some(RecordBatch::try_new(schema, cols)?)) } diff --git a/src/arrow/family.rs b/src/arrow/family.rs index 6b2799a..e43e0df 100644 --- a/src/arrow/family.rs +++ b/src/arrow/family.rs @@ -16,6 +16,7 @@ use crate::parse::MetricType; use super::columns::{HistogramDpTable, MetricTable, SummaryDpTable, f64_to_u64_count}; +use super::intern::StringInterner; use super::options::{ CounterSuffixPolicy, DecodeError, DecodeOptions, UnitSuffixPolicy, UntypedPolicy, }; @@ -23,6 +24,7 @@ use super::schema::{ OTAP_METRIC_TYPE_GAUGE, OTAP_METRIC_TYPE_HISTOGRAM, OTAP_METRIC_TYPE_SUM, OTAP_METRIC_TYPE_SUMMARY, }; +use super::staleness::{DATA_POINT_FLAG_NO_RECORDED_VALUE, DpKind, StalenessTracker}; /// Role of a series line within the active [`CurrentFamily`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -56,6 +58,7 @@ pub(super) struct CurrentFamily { pub(super) help: Vec, pub(super) unit: Vec, pub(super) batch_id: Option, + pub(super) tracker_family_idx: Option, } impl CurrentFamily { @@ -82,6 +85,7 @@ impl CurrentFamily { self.help.clear(); self.unit.clear(); self.batch_id = None; + self.tracker_family_idx = None; } pub(super) fn retained_memory_bytes(&self) -> usize { @@ -111,6 +115,7 @@ impl CurrentFamily { self.help.clear(); self.unit.clear(); self.batch_id = None; + self.tracker_family_idx = None; } #[inline] @@ -184,104 +189,124 @@ impl CurrentFamily { return Ok(id); } - let m_idx = metrics.len(); - let id = u16::try_from(m_idx).map_err(|_| DecodeError::CapacityOverflow { - table: "UnivariateMetrics", - count: m_idx + 1, - })?; - - let fam_name = self.unescaped_name.as_slice(); - let unit = self.unit.as_slice(); - let inferred_counter = self.metric_type == MetricType::Unknown - && opts.untyped_policy == UntypedPolicy::InferFromSuffix - && fam_name.len() > 6 - && fam_name.ends_with(b"_total"); - let is_counter = self.metric_type == MetricType::Counter || inferred_counter; - - let mut base_name = if is_counter && fam_name.len() > 6 && fam_name.ends_with(b"_total") { - &fam_name[..fam_name.len() - 6] - } else { - fam_name - }; - - if opts.unit_suffix_policy == UnitSuffixPolicy::Strip - && self.has_unit - && !unit.is_empty() - && base_name.len() > unit.len() + 1 - && base_name.ends_with(unit) - && base_name[base_name.len() - unit.len() - 1] == b'_' - { - base_name = &base_name[..base_name.len() - unit.len() - 1]; - } - - let append_total = is_counter - && opts.counter_suffix_policy == CounterSuffixPolicy::Keep - && self.saw_total_suffix; - let emitted_name_len = base_name.len() + if append_total { 6 } else { 0 }; - - let next_name_len = metrics.name_bytes.len().saturating_add(emitted_name_len); - let name_end_off = - i32::try_from(next_name_len).map_err(|_| DecodeError::CapacityOverflow { - table: "UnivariateMetrics.name", - count: next_name_len, - })?; - - let next_desc_len = metrics.desc.check_next_offset( + let id = emit_family_row( + &self.unescaped_name, + self.metric_type, + self.saw_total_suffix, self.has_help, &self.help, - "UnivariateMetrics.description", + self.has_unit, + &self.unit, + metrics, + opts, )?; - let next_unit_len = - metrics - .unit - .check_next_offset(self.has_unit, &self.unit, "UnivariateMetrics.unit")?; - - metrics.name_bytes.extend_from_slice(base_name); - if append_total { - metrics.name_bytes.extend_from_slice(b"_total"); - } - metrics.name_offsets.push(name_end_off); - - let (otap_type, attr_type_key, is_monotonic) = match self.metric_type { - MetricType::Counter => (OTAP_METRIC_TYPE_SUM, 1u16, true), - MetricType::Gauge => (OTAP_METRIC_TYPE_GAUGE, 2u16, false), - MetricType::Histogram => (OTAP_METRIC_TYPE_HISTOGRAM, 3u16, false), - MetricType::GaugeHistogram => (OTAP_METRIC_TYPE_HISTOGRAM, 4u16, false), - MetricType::Summary => (OTAP_METRIC_TYPE_SUMMARY, 5u16, false), - MetricType::Info => (OTAP_METRIC_TYPE_GAUGE, 6u16, false), - MetricType::Stateset => (OTAP_METRIC_TYPE_GAUGE, 7u16, false), - MetricType::Unknown => { - if inferred_counter { - (OTAP_METRIC_TYPE_SUM, 0u16, true) - } else { - (OTAP_METRIC_TYPE_GAUGE, 0u16, false) - } - } - }; - let has_temporality = - otap_type == OTAP_METRIC_TYPE_SUM || otap_type == OTAP_METRIC_TYPE_HISTOGRAM; - let has_sum = otap_type == OTAP_METRIC_TYPE_SUM; + self.has_samples = true; + self.batch_id = Some(id); + Ok(id) + } +} + +/// Appends a `UnivariateMetrics` row and its `MetricAttrs` (`prometheus.type`) entry to `metrics` +/// for the given family metadata and translation options, returning its `u16` metric ID. +#[inline] +#[allow(clippy::too_many_arguments)] +pub(super) fn emit_family_row( + fam_name: &[u8], + metric_type: MetricType, + saw_total_suffix: bool, + has_help: bool, + help: &[u8], + has_unit: bool, + unit: &[u8], + metrics: &mut MetricTable, + opts: &DecodeOptions, +) -> Result { + let m_idx = metrics.len(); + let id = u16::try_from(m_idx).map_err(|_| DecodeError::CapacityOverflow { + table: "UnivariateMetrics", + count: m_idx + 1, + })?; + + let inferred_counter = metric_type == MetricType::Unknown + && opts.untyped_policy == UntypedPolicy::InferFromSuffix + && fam_name.len() > 6 + && fam_name.ends_with(b"_total"); + let is_counter = metric_type == MetricType::Counter || inferred_counter; + + let mut base_name = if is_counter && fam_name.len() > 6 && fam_name.ends_with(b"_total") { + &fam_name[..fam_name.len() - 6] + } else { + fam_name + }; + + if opts.unit_suffix_policy == UnitSuffixPolicy::Strip + && has_unit + && !unit.is_empty() + && base_name.len() > unit.len() + 1 + && base_name.ends_with(unit) + && base_name[base_name.len() - unit.len() - 1] == b'_' + { + base_name = &base_name[..base_name.len() - unit.len() - 1]; + } - metrics.metric_types.push(otap_type); - metrics.is_monotonic.push(is_monotonic); - metrics.attr_type_keys.push(attr_type_key); - metrics.any_temporality |= has_temporality; - metrics.all_temporality &= has_temporality; - metrics.any_sum |= has_sum; - metrics.all_sum &= has_sum; + let append_total = + is_counter && opts.counter_suffix_policy == CounterSuffixPolicy::Keep && saw_total_suffix; + let emitted_name_len = base_name.len() + if append_total { 6 } else { 0 }; + let next_name_len = metrics.name_bytes.len().saturating_add(emitted_name_len); + let name_end_off = i32::try_from(next_name_len).map_err(|_| DecodeError::CapacityOverflow { + table: "UnivariateMetrics.name", + count: next_name_len, + })?; + + let next_desc_len = metrics .desc - .append_checked(m_idx, next_desc_len, &self.help); - metrics - .unit - .append_checked(m_idx, next_unit_len, &self.unit); - - self.has_samples = true; - self.batch_id = Some(id); - Ok(id) + .check_next_offset(has_help, help, "UnivariateMetrics.description")?; + let next_unit_len = metrics + .unit + .check_next_offset(has_unit, unit, "UnivariateMetrics.unit")?; + + metrics.name_bytes.extend_from_slice(base_name); + if append_total { + metrics.name_bytes.extend_from_slice(b"_total"); } + metrics.name_offsets.push(name_end_off); + + let (otap_type, attr_type_key, is_monotonic) = match metric_type { + MetricType::Counter => (OTAP_METRIC_TYPE_SUM, 1u16, true), + MetricType::Gauge => (OTAP_METRIC_TYPE_GAUGE, 2u16, false), + MetricType::Histogram => (OTAP_METRIC_TYPE_HISTOGRAM, 3u16, false), + MetricType::GaugeHistogram => (OTAP_METRIC_TYPE_HISTOGRAM, 4u16, false), + MetricType::Summary => (OTAP_METRIC_TYPE_SUMMARY, 5u16, false), + MetricType::Info => (OTAP_METRIC_TYPE_GAUGE, 6u16, false), + MetricType::Stateset => (OTAP_METRIC_TYPE_GAUGE, 7u16, false), + MetricType::Unknown => { + if inferred_counter { + (OTAP_METRIC_TYPE_SUM, 0u16, true) + } else { + (OTAP_METRIC_TYPE_GAUGE, 0u16, false) + } + } + }; + + let has_temporality = + otap_type == OTAP_METRIC_TYPE_SUM || otap_type == OTAP_METRIC_TYPE_HISTOGRAM; + let has_sum = otap_type == OTAP_METRIC_TYPE_SUM; + + metrics.metric_types.push(otap_type); + metrics.is_monotonic.push(is_monotonic); + metrics.attr_type_keys.push(attr_type_key); + metrics.any_temporality |= has_temporality; + metrics.all_temporality &= has_temporality; + metrics.any_sum |= has_sum; + metrics.all_sum &= has_sum; + + metrics.desc.append_checked(m_idx, next_desc_len, help); + metrics.unit.append_checked(m_idx, next_unit_len, unit); + + Ok(id) } /// Sentinel index (`u32::MAX`) marking an empty linked-list pointer or hash-table slot in @@ -293,7 +318,7 @@ const NONE_IDX: u32 = u32::MAX; const FX_SEED: u64 = 0x517c_c1b7_2722_0a95; #[inline] -fn hash_attrs(attrs: &[(u32, u32)]) -> u64 { +pub(super) fn hash_attrs(attrs: &[(u32, u32)]) -> u64 { let mut hash = 0xcbf2_9ce4_8422_2325_u64; for &(k, v) in attrs { let word = (u64::from(k) << 32) | u64::from(v); @@ -327,6 +352,7 @@ pub(super) struct PendingDpGroup { pub(super) ts_ns: Option, pub(super) count: Option, pub(super) sum: Option, + pub(super) is_stale: bool, pub(super) point_head: u32, pub(super) point_tail: u32, pub(super) exemplar_head: u32, @@ -343,6 +369,7 @@ impl PendingDpGroup { ts_ns: None, count: None, sum: None, + is_stale: false, point_head: NONE_IDX, point_tail: NONE_IDX, exemplar_head: NONE_IDX, @@ -592,12 +619,16 @@ impl FamilyAssembly { } } + #[allow(clippy::too_many_arguments)] pub(super) fn flush( &mut self, current_family: &mut CurrentFamily, metrics: &mut MetricTable, sum_dp: &mut SummaryDpTable, hist_dp: &mut HistogramDpTable, + key_interner: &StringInterner, + val_interner: &StringInterner, + mut tracker: Option<&mut StalenessTracker>, opts: &DecodeOptions, default_ts_ns: Option, ) -> Result<(), DecodeError> { @@ -628,16 +659,29 @@ impl FamilyAssembly { .attrs .extend_for_parent(dp_id, &self.group_attrs[attr_start..attr_end]); - sum_dp - .base - .push(parent_id, group.start_ts_ns, group.ts_ns.or(default_ts_ns)); - sum_dp.counts.push(dp_index, group.count); - sum_dp.sums.push(dp_index, group.sum); - - self.collect_sorted_group_points(group.point_head); - for &(q, val) in &self.group_points { - sum_dp.quantile_bounds.push(q); - sum_dp.quantile_vals.push(val); + let flags = if group.is_stale { + DATA_POINT_FLAG_NO_RECORDED_VALUE + } else { + 0 + }; + sum_dp.base.push_with_flags( + parent_id, + group.start_ts_ns, + group.ts_ns.or(default_ts_ns), + flags, + ); + if group.is_stale { + sum_dp.counts.push(dp_index, Some(0)); + sum_dp.sums.push(dp_index, Some(0.0)); + } else { + sum_dp.counts.push(dp_index, group.count); + sum_dp.sums.push(dp_index, group.sum); + + self.collect_sorted_group_points(group.point_head); + for &(q, val) in &self.group_points { + sum_dp.quantile_bounds.push(q); + sum_dp.quantile_vals.push(val); + } } let q_off = i32::try_from(sum_dp.quantile_bounds.len()).map_err(|_| { DecodeError::CapacityOverflow { @@ -646,6 +690,19 @@ impl FamilyAssembly { } })?; sum_dp.quantile_offsets.push(q_off); + + if (group.ts_ns.is_none() || opts.track_timestamps_staleness) + && let Some(tr) = tracker.as_deref_mut() + { + tr.record_series( + current_family, + DpKind::Summary, + &self.group_attrs[attr_start..attr_end], + key_interner, + val_interner, + group.is_stale, + )?; + } } } MetricType::Histogram | MetricType::GaugeHistogram => { @@ -668,30 +725,65 @@ impl FamilyAssembly { .attrs .extend_for_parent(dp_id, &self.group_attrs[attr_start..attr_end]); - hist_dp - .base - .push(parent_id, group.start_ts_ns, group.ts_ns.or(default_ts_ns)); - - self.collect_sorted_group_points(group.point_head); - - let mut prev_cum = 0u64; - let mut inf_cum: Option = None; - for &(bound, cum_f64) in &self.group_points { - let cum = f64_to_u64_count(cum_f64); - if bound == f64::INFINITY { - inf_cum = Some(cum); - } else if bound.is_finite() { - hist_dp.bounds_vals.push(bound); - hist_dp.bucket_counts.push(cum.saturating_sub(prev_cum)); - prev_cum = cum; + let flags = if group.is_stale { + DATA_POINT_FLAG_NO_RECORDED_VALUE + } else { + 0 + }; + hist_dp.base.push_with_flags( + parent_id, + group.start_ts_ns, + group.ts_ns.or(default_ts_ns), + flags, + ); + + if group.is_stale { + hist_dp.counts.push(dp_index, Some(0)); + hist_dp.sums.push(dp_index, None); + } else { + self.collect_sorted_group_points(group.point_head); + + let mut prev_cum = 0u64; + let mut inf_cum: Option = None; + for &(bound, cum_f64) in &self.group_points { + let cum = f64_to_u64_count(cum_f64); + if bound == f64::INFINITY { + inf_cum = Some(cum); + } else if bound.is_finite() { + hist_dp.bounds_vals.push(bound); + hist_dp.bucket_counts.push(cum.saturating_sub(prev_cum)); + prev_cum = cum; + } + } + let final_count = group.count.or(inf_cum); + if !self.group_points.is_empty() { + let total_for_inf = inf_cum.or(group.count).unwrap_or(prev_cum); + hist_dp + .bucket_counts + .push(total_for_inf.saturating_sub(prev_cum)); + } + hist_dp.counts.push(dp_index, final_count); + hist_dp.sums.push(dp_index, group.sum); + + let mut ex_cur = group.exemplar_head; + while ex_cur != NONE_IDX { + let ex = self.exemplars[ex_cur as usize]; + ex_cur = ex.next; + let hist_ex_id = hist_dp.exemplars.next_id("HistogramDpExemplars")?; + hist_dp.exemplars.push_row( + dp_id, + ex.ts_ns, + ex.value, + ex.span_id, + ex.trace_id, + ); + let ex_attr_start = ex.attr_start as usize; + let ex_attr_end = ex_attr_start + ex.attr_len as usize; + hist_dp.exemplars.attrs.extend_for_parent( + hist_ex_id, + &self.exemplar_attrs[ex_attr_start..ex_attr_end], + ); } - } - let final_count = group.count.or(inf_cum); - if !self.group_points.is_empty() { - let total_for_inf = inf_cum.or(group.count).unwrap_or(prev_cum); - hist_dp - .bucket_counts - .push(total_for_inf.saturating_sub(prev_cum)); } let bucket_off = i32::try_from(hist_dp.bucket_counts.len()).map_err(|_| { @@ -709,27 +801,17 @@ impl FamilyAssembly { hist_dp.bucket_offsets.push(bucket_off); hist_dp.bounds_offsets.push(bounds_off); - hist_dp.counts.push(dp_index, final_count); - hist_dp.sums.push(dp_index, group.sum); - - let mut ex_cur = group.exemplar_head; - while ex_cur != NONE_IDX { - let ex = self.exemplars[ex_cur as usize]; - ex_cur = ex.next; - let hist_ex_id = hist_dp.exemplars.next_id("HistogramDpExemplars")?; - hist_dp.exemplars.push_row( - dp_id, - ex.ts_ns, - ex.value, - ex.span_id, - ex.trace_id, - ); - let ex_attr_start = ex.attr_start as usize; - let ex_attr_end = ex_attr_start + ex.attr_len as usize; - hist_dp.exemplars.attrs.extend_for_parent( - hist_ex_id, - &self.exemplar_attrs[ex_attr_start..ex_attr_end], - ); + if (group.ts_ns.is_none() || opts.track_timestamps_staleness) + && let Some(tr) = tracker.as_deref_mut() + { + tr.record_series( + current_family, + DpKind::Histogram, + &self.group_attrs[attr_start..attr_end], + key_interner, + val_interner, + group.is_stale, + )?; } } } diff --git a/src/arrow/intern.rs b/src/arrow/intern.rs index cb5d0ed..87a2a1a 100644 --- a/src/arrow/intern.rs +++ b/src/arrow/intern.rs @@ -75,7 +75,18 @@ pub(super) struct StringInterner { } impl StringInterner { + pub(super) const fn new() -> Self { + Self { + bytes: Vec::new(), + offsets: Vec::new(), + buckets: Vec::new(), + } + } + pub(super) fn with_capacity(strings: usize, byte_cap: usize) -> Self { + if strings == 0 && byte_cap == 0 { + return Self::new(); + } let bucket_cap = (strings.saturating_mul(2)).next_power_of_two().max(16); let mut offsets = Vec::with_capacity(strings + 1); offsets.push(0); @@ -89,14 +100,16 @@ impl StringInterner { /// Resets the interner for the next scrape while preserving backing capacities. pub(super) fn clear(&mut self) { self.bytes.clear(); - self.offsets.clear(); - self.offsets.push(0); + if !self.offsets.is_empty() { + self.offsets.clear(); + self.offsets.push(0); + } self.buckets.fill(Bucket::EMPTY); } #[inline] pub(super) fn len(&self) -> usize { - self.offsets.len() - 1 + self.offsets.len().saturating_sub(1) } pub(super) fn retained_memory_bytes(&self) -> usize { @@ -125,6 +138,9 @@ impl StringInterner { /// Looks up a previously interned byte slice without inserting it. #[inline] pub(super) fn lookup(&self, raw: &[u8]) -> Option { + if self.buckets.is_empty() { + return None; + } let hash = fx_hash(raw); let mask = self.buckets.len() - 1; let mut slot = (hash as usize) & mask; @@ -145,6 +161,10 @@ impl StringInterner { /// arena exceeds `i32::MAX`. #[inline] pub(super) fn intern(&mut self, raw: &[u8]) -> Option { + if self.buckets.is_empty() { + self.offsets.push(0); + self.buckets.resize(16, Bucket::EMPTY); + } let hash = fx_hash(raw); let mask = self.buckets.len() - 1; let mut slot = (hash as usize) & mask; @@ -198,9 +218,17 @@ impl StringInterner { /// Materializes all interned strings in insertion order (`0..len()`) as an Arrow `StringArray`. pub(super) fn to_string_array(&self) -> StringArray { - let offsets_buf = Buffer::from_slice_ref(&self.offsets); - let offsets = - OffsetBuffer::new(ScalarBuffer::::new(offsets_buf, 0, self.offsets.len())); + let offsets_slice = if self.offsets.is_empty() { + &[0i32][..] + } else { + &self.offsets[..] + }; + let offsets_buf = Buffer::from_slice_ref(offsets_slice); + let offsets = OffsetBuffer::new(ScalarBuffer::::new( + offsets_buf, + 0, + offsets_slice.len(), + )); let values = Buffer::from_slice_ref(&self.bytes); // SAFETY-FREE invariant: every interned entry came from a valid `&str`, and `self.offsets` // is monotonically non-decreasing starting at `0` and ending at `self.bytes.len()`. diff --git a/src/arrow/options.rs b/src/arrow/options.rs index 2b66367..ca85e83 100644 --- a/src/arrow/options.rs +++ b/src/arrow/options.rs @@ -164,6 +164,10 @@ pub struct DecodeOptions { pub unit_suffix_policy: UnitSuffixPolicy, /// Projection policy for untyped (`unknown`) metric families. pub untyped_policy: UntypedPolicy, + /// Whether samples carrying an explicit wire timestamp participate in [`super::StalenessTracker`] + /// staleness tracking (`false` by default, matching Go Prometheus's + /// `track_timestamps_staleness` scrape option). + pub track_timestamps_staleness: bool, } impl Default for DecodeOptions { @@ -184,6 +188,7 @@ impl DecodeOptions { counter_suffix_policy: CounterSuffixPolicy::Keep, unit_suffix_policy: UnitSuffixPolicy::Keep, untyped_policy: UntypedPolicy::Gauge, + track_timestamps_staleness: false, } } @@ -225,6 +230,14 @@ impl DecodeOptions { self.untyped_policy = policy; self } + + /// Sets whether samples carrying an explicit wire timestamp are tracked in + /// [`super::StalenessTracker`] (`false` by default, matching Go Prometheus). + #[must_use] + pub const fn with_track_timestamps_staleness(mut self, track: bool) -> Self { + self.track_timestamps_staleness = track; + self + } } /// Errors returned by [`super::ArrowDecoder::decode`]. diff --git a/src/arrow/schema.rs b/src/arrow/schema.rs index 23b6804..44b68b7 100644 --- a/src/arrow/schema.rs +++ b/src/arrow/schema.rs @@ -184,8 +184,10 @@ fn build_number_dp_schema( start_time_nullable: bool, has_time: bool, time_nullable: bool, + double_value_nullable: bool, + has_flags: bool, ) -> SchemaRef { - let mut fields = Vec::with_capacity(5); + let mut fields = Vec::with_capacity(6); if has_id { fields.push(Field::new("id", DataType::UInt32, false).with_metadata(plain_id_metadata())); } @@ -205,40 +207,56 @@ fn build_number_dp_schema( time_nullable, )); } - fields.push(Field::new("double_value", DataType::Float64, false)); + fields.push(Field::new( + "double_value", + DataType::Float64, + double_value_nullable, + )); + if has_flags { + fields.push(Field::new("flags", DataType::UInt32, false)); + } Arc::new(Schema::new(fields)) } -static NUMBER_DP_SCHEMAS: LazyLock<[SchemaRef; 32]> = LazyLock::new(|| { +static NUMBER_DP_SCHEMAS: LazyLock<[SchemaRef; 128]> = LazyLock::new(|| { std::array::from_fn(|mask| { let has_id = (mask & 1) != 0; let has_start_time = (mask & 2) != 0; let start_time_nullable = (mask & 4) != 0; let has_time = (mask & 8) != 0; let time_nullable = (mask & 16) != 0; + let double_value_nullable = (mask & 32) != 0; + let has_flags = (mask & 64) != 0; build_number_dp_schema( has_id, has_start_time, start_time_nullable, has_time, time_nullable, + double_value_nullable, + has_flags, ) }) }); #[inline] +#[allow(clippy::too_many_arguments)] pub(super) fn number_dp_schema( has_id: bool, has_start_time: bool, start_time_nullable: bool, has_time: bool, time_nullable: bool, + double_value_nullable: bool, + has_flags: bool, ) -> SchemaRef { let mask = usize::from(has_id) | (usize::from(has_start_time) << 1) | (usize::from(start_time_nullable) << 2) | (usize::from(has_time) << 3) - | (usize::from(time_nullable) << 4); + | (usize::from(time_nullable) << 4) + | (usize::from(double_value_nullable) << 5) + | (usize::from(has_flags) << 6); Arc::clone(&NUMBER_DP_SCHEMAS[mask]) } @@ -267,6 +285,7 @@ pub(super) fn summary_quantile_item_field() -> FieldRef { Arc::clone(&SUMMARY_QUANTILE_ITEM_FIELD) } +#[allow(clippy::too_many_arguments)] fn build_summary_dp_schema( has_id: bool, has_start_time: bool, @@ -275,8 +294,9 @@ fn build_summary_dp_schema( time_nullable: bool, count_nullable: bool, sum_nullable: bool, + has_flags: bool, ) -> SchemaRef { - let mut fields = Vec::with_capacity(7); + let mut fields = Vec::with_capacity(8); if has_id { fields.push(Field::new("id", DataType::UInt32, false).with_metadata(plain_id_metadata())); } @@ -303,10 +323,13 @@ fn build_summary_dp_schema( DataType::List(summary_quantile_item_field()), false, )); + if has_flags { + fields.push(Field::new("flags", DataType::UInt32, false)); + } Arc::new(Schema::new(fields)) } -static SUMMARY_DP_SCHEMAS: LazyLock<[SchemaRef; 128]> = LazyLock::new(|| { +static SUMMARY_DP_SCHEMAS: LazyLock<[SchemaRef; 256]> = LazyLock::new(|| { std::array::from_fn(|mask| { build_summary_dp_schema( (mask & 1) != 0, @@ -316,11 +339,13 @@ static SUMMARY_DP_SCHEMAS: LazyLock<[SchemaRef; 128]> = LazyLock::new(|| { (mask & 16) != 0, (mask & 32) != 0, (mask & 64) != 0, + (mask & 128) != 0, ) }) }); #[inline] +#[allow(clippy::too_many_arguments)] pub(super) fn summary_dp_schema( has_id: bool, has_start_time: bool, @@ -329,6 +354,7 @@ pub(super) fn summary_dp_schema( time_nullable: bool, count_nullable: bool, sum_nullable: bool, + has_flags: bool, ) -> SchemaRef { let mask = usize::from(has_id) | (usize::from(has_start_time) << 1) @@ -336,7 +362,8 @@ pub(super) fn summary_dp_schema( | (usize::from(has_time) << 3) | (usize::from(time_nullable) << 4) | (usize::from(count_nullable) << 5) - | (usize::from(sum_nullable) << 6); + | (usize::from(sum_nullable) << 6) + | (usize::from(has_flags) << 7); Arc::clone(&SUMMARY_DP_SCHEMAS[mask]) } @@ -356,6 +383,7 @@ pub(super) fn histogram_explicit_bounds_item_field() -> FieldRef { Arc::clone(&HISTOGRAM_EXPLICIT_BOUNDS_ITEM_FIELD) } +#[allow(clippy::too_many_arguments)] fn build_histogram_dp_schema( has_id: bool, has_start_time: bool, @@ -364,8 +392,9 @@ fn build_histogram_dp_schema( time_nullable: bool, count_nullable: bool, sum_nullable: bool, + has_flags: bool, ) -> SchemaRef { - let mut fields = Vec::with_capacity(8); + let mut fields = Vec::with_capacity(9); if has_id { fields.push(Field::new("id", DataType::UInt32, false).with_metadata(plain_id_metadata())); } @@ -397,10 +426,13 @@ fn build_histogram_dp_schema( DataType::List(histogram_explicit_bounds_item_field()), false, )); + if has_flags { + fields.push(Field::new("flags", DataType::UInt32, false)); + } Arc::new(Schema::new(fields)) } -static HISTOGRAM_DP_SCHEMAS: LazyLock<[SchemaRef; 128]> = LazyLock::new(|| { +static HISTOGRAM_DP_SCHEMAS: LazyLock<[SchemaRef; 256]> = LazyLock::new(|| { std::array::from_fn(|mask| { build_histogram_dp_schema( (mask & 1) != 0, @@ -410,11 +442,13 @@ static HISTOGRAM_DP_SCHEMAS: LazyLock<[SchemaRef; 128]> = LazyLock::new(|| { (mask & 16) != 0, (mask & 32) != 0, (mask & 64) != 0, + (mask & 128) != 0, ) }) }); #[inline] +#[allow(clippy::too_many_arguments)] pub(super) fn histogram_dp_schema( has_id: bool, has_start_time: bool, @@ -423,6 +457,7 @@ pub(super) fn histogram_dp_schema( time_nullable: bool, count_nullable: bool, sum_nullable: bool, + has_flags: bool, ) -> SchemaRef { let mask = usize::from(has_id) | (usize::from(has_start_time) << 1) @@ -430,7 +465,8 @@ pub(super) fn histogram_dp_schema( | (usize::from(has_time) << 3) | (usize::from(time_nullable) << 4) | (usize::from(count_nullable) << 5) - | (usize::from(sum_nullable) << 6); + | (usize::from(sum_nullable) << 6) + | (usize::from(has_flags) << 7); Arc::clone(&HISTOGRAM_DP_SCHEMAS[mask]) } diff --git a/src/arrow/staleness.rs b/src/arrow/staleness.rs new file mode 100644 index 0000000..34913f8 --- /dev/null +++ b/src/arrow/staleness.rs @@ -0,0 +1,1245 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Prometheus `StaleNaN` detection and cross-scrape per-target series staleness tracking. +//! +//! Prometheus marks disappeared time series with a distinguished signaling IEEE 754 `NaN` bit +//! pattern ([`STALE_NAN_BITS`] / [`STALE_NAN`], matching Go's `model/value.StaleNaN`). In +//! OpenTelemetry and OTAP, stale data points carry [`DATA_POINT_FLAG_NO_RECORDED_VALUE`] (`1`, +//! `DataPointFlags::FLAG_NO_RECORDED_VALUE`) in the optional `flags` (`UInt32`) column on +//! `NumberDataPoints`, `SummaryDataPoints`, and `HistogramDataPoints`. +//! +//! [`StalenessTracker`] maintains the active series set for a single scrape target across +//! consecutive scrapes in arena-backed tables so steady-state scrapes whose series set does not +//! change perform zero heap allocations. + +use std::time::{Duration, Instant}; + +use crate::parse::MetricType; + +use super::columns::{HistogramDpTable, MetricTable, NumberDpTable, SummaryDpTable}; +use super::family::{CurrentFamily, emit_family_row, hash_attrs}; +use super::intern::StringInterner; +use super::options::{ + DECODE_BUDGET_DURATION_CHECK_INTERVAL, DecodeError, DecodeOptions, UntypedPolicy, +}; + +/// Raw IEEE 754 `u64` bit pattern of Prometheus's staleness marker (`0x7ff0_0000_0000_0002`), +/// matching `prometheus/prometheus/model/value.StaleNaN`. +/// +/// Unlike normal quiet `NaN` (`0x7ff8_0000_0000_0001`), `StaleNaN` clears the MSB of the mantissa +/// (making it a signaling `NaN` bit pattern with payload `2`) to distinguish staleness markers +/// from arithmetic `NaN` results. +pub const STALE_NAN_BITS: u64 = 0x7ff0_0000_0000_0002; + +/// IEEE 754 `f64` value with [`STALE_NAN_BITS`] bit representation. +/// +/// NOTE: Because [`STALE_NAN_BITS`] clears the MSB of the mantissa (making it a signaling `NaN`), +/// floating-point arithmetic operations may quietize bit 51 under IEEE 754 / Rust float semantics. +/// Move this value bitwise (`from_bits` / `to_bits`) and inspect it with [`is_stale_nan`] rather +/// than passing it through arithmetic expressions. +pub const STALE_NAN: f64 = f64::from_bits(STALE_NAN_BITS); + +/// OpenTelemetry `DataPointFlags::FLAG_NO_RECORDED_VALUE` bitmask (`1`) indicating a stale data +/// point with no recorded value. +pub const DATA_POINT_FLAG_NO_RECORDED_VALUE: u32 = 1; + +/// Returns `true` if `v` has the exact IEEE 754 bit pattern of Prometheus's `StaleNaN` marker +/// ([`STALE_NAN_BITS`]). +/// +/// Inspects raw bits via `f64::to_bits`; callers must avoid performing arithmetic on `v` prior to +/// calling this function because arithmetic on signaling `NaN`s may quietize the mantissa MSB. +#[inline] +#[must_use] +pub const fn is_stale_nan(v: f64) -> bool { + v.to_bits() == STALE_NAN_BITS +} + +/// Sentinel index (`u32::MAX`) marking an unoccupied hash-table slot. +const NONE_IDX: u32 = u32::MAX; + +/// Multiplicative constant from `rustc-hash` (`FxHasher`). +const FX_SEED: u64 = 0x517c_c1b7_2722_0a95; + +#[inline] +const fn metric_type_tag(mt: MetricType) -> u8 { + match mt { + MetricType::Unknown => 0, + MetricType::Counter => 1, + MetricType::Gauge => 2, + MetricType::Histogram => 3, + MetricType::GaugeHistogram => 4, + MetricType::Summary => 5, + MetricType::Info => 6, + MetricType::Stateset => 7, + } +} + +#[inline] +fn canonical_family_key(current_family: &CurrentFamily) -> (&[u8], MetricType, bool) { + let raw = current_family.unescaped_name.as_slice(); + if current_family.metric_type == MetricType::Counter + && raw.len() > 6 + && raw.ends_with(b"_total") + { + (&raw[..raw.len() - 6], MetricType::Counter, true) + } else { + ( + raw, + current_family.metric_type, + current_family.saw_total_suffix, + ) + } +} + +#[inline] +fn hash_family_key(name: &[u8], metric_type: MetricType, saw_total_suffix: bool) -> u64 { + let mut hash = 0xcbf2_9ce4_8422_2325_u64; + let (chunks, remainder) = name.as_chunks::<8>(); + for chunk in chunks { + let word = u64::from_le_bytes(*chunk); + hash = (hash.rotate_left(5) ^ word).wrapping_mul(FX_SEED); + } + for &b in remainder { + hash = (hash.rotate_left(5) ^ u64::from(b)).wrapping_mul(FX_SEED); + } + let tag = (u64::from(metric_type_tag(metric_type)) << 1) | u64::from(saw_total_suffix); + hash = (hash.rotate_left(5) ^ tag).wrapping_mul(FX_SEED); + hash ^= hash >> 32; + hash = hash.wrapping_mul(FX_SEED); + hash ^ (hash >> 32) +} + +#[inline] +fn hash_series(family_idx: u32, kind: DpKind, attrs: &[(u32, u32)]) -> u64 { + let mut hash = hash_attrs(attrs); + let word = (u64::from(family_idx) << 8) | u64::from(kind as u8); + hash = (hash.rotate_left(5) ^ word).wrapping_mul(FX_SEED); + hash ^= hash >> 32; + hash = hash.wrapping_mul(FX_SEED); + hash ^ (hash >> 32) +} + +/// Outcome of a bounded EOF stale-series drain step in [`StalenessTracker::finish_scrape_step`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum StaleDrainStep { + /// The cooperative step budget expired while scanning/emitting disappeared series; yield + /// [`super::DecodeStep::Pending`] and resume from `stale_cursor` on the next step. + YieldPending, + /// The active streaming chunk reached `max_data_points` or `max_metrics`; finalize this chunk + /// and continue draining remaining stale series in the next chunk. + ChunkFull, + /// All disappeared series have been staged into the current batch. + Complete, +} + +/// OTAP data-point table kind associated with a tracked time series. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub(super) enum DpKind { + Number = 0, + Summary = 1, + Histogram = 2, +} + +/// Per-scrape observation status of a [`TrackedSeries`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +enum SeenState { + Unseen = 0, + SeenLive = 1, + SeenStale = 2, +} + +#[derive(Clone, Copy)] +struct FamilyBucket { + hash: u64, + family_idx: u32, +} + +impl FamilyBucket { + const EMPTY: Self = Self { + hash: 0, + family_idx: NONE_IDX, + }; +} + +#[derive(Clone, Copy)] +struct SeriesBucket { + hash: u64, + series_idx: u32, +} + +impl SeriesBucket { + const EMPTY: Self = Self { + hash: 0, + series_idx: NONE_IDX, + }; +} + +#[derive(Clone, Copy)] +struct TrackedFamily { + name_start: u32, + name_end: u32, + help_start: u32, + help_end: u32, + unit_start: u32, + unit_end: u32, + pending_help_start: u32, + pending_help_end: u32, + pending_unit_start: u32, + pending_unit_end: u32, + metric_type: MetricType, + saw_total_suffix: bool, + has_help: bool, + has_unit: bool, + pending_has_help: bool, + pending_has_unit: bool, + has_pending_meta: bool, + current_batch_id: Option, +} + +#[derive(Clone, Copy)] +struct TrackedSeries { + family_idx: u32, + attr_start: u32, + attr_len: u32, + kind: DpKind, + /// Committed active status from the end of the previous successful scrape. + active: bool, + /// Observation state during the current in-progress scrape. + seen: SeenState, +} + +/// Reusable per-target series staleness tracker for [`super::ArrowDecoder`]. +/// +/// `StalenessTracker` records the active time series observed for a single scrape target across +/// consecutive scrapes. When passed to [`super::ArrowDecoder::decode_with_staleness`] or +/// [`super::ArrowDecoder::decode_stream_with_staleness`], any series that was active at the end of +/// the previous scrape and is absent from the current scrape is emitted as a stale data point with +/// `flags = DATA_POINT_FLAG_NO_RECORDED_VALUE` (`1`), and then removed from the active set so +/// subsequent scrapes do not re-emit it. +/// +/// # Memory model and optional usage +/// +/// Emitting a stale OTAP data point when a series disappears from a target's exposition requires +/// retaining that target's active metric names and interned label sets across scrapes (mirroring +/// Go Prometheus's per-target `scrapeCache`). Because retaining per-target active series consumes +/// live heap proportional to the target's active series count, staleness tracking is strictly +/// opt-in: callers that do not need scrape-time staleness markers can call +/// [`super::ArrowDecoder::decode`] or [`super::ArrowDecoder::decode_stream`] with zero cross-scrape +/// series retention. +/// +/// All family metadata, label keys, label values, and sorted attribute slices are interned into +/// contiguous arena buffers so steady-state scrapes with a stable series set perform zero heap +/// allocations. Under label churn, [`StalenessTracker`] automatically compacts deactivated series, +/// unreferenced families, and unreferenced interned strings (and clears arenas when active series +/// drop to zero) so retained live heap stays bounded by the currently active series set rather than +/// cumulative historical cardinality. +pub struct StalenessTracker { + families: Vec, + family_bytes: Vec, + family_buckets: Vec, + + key_interner: StringInterner, + val_interner: StringInterner, + + series: Vec, + series_attrs: Vec<(u32, u32)>, + series_buckets: Vec, + + batch_key_to_tracker: Vec, + batch_val_to_tracker: Vec, + temp_tracker_attrs: Vec<(u32, u32)>, + + active_count: usize, + stale_cursor: usize, + scrape_in_progress: bool, +} + +impl Default for StalenessTracker { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Debug for StalenessTracker { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StalenessTracker") + .field("active_series_len", &self.active_series_len()) + .field("retained_memory_bytes", &self.retained_memory_bytes()) + .finish_non_exhaustive() + } +} + +impl StalenessTracker { + /// Creates an empty [`StalenessTracker`] that performs zero heap allocations until its first + /// non-empty scrape. + #[must_use] + pub const fn new() -> Self { + Self { + families: Vec::new(), + family_bytes: Vec::new(), + family_buckets: Vec::new(), + + key_interner: StringInterner::new(), + val_interner: StringInterner::new(), + + series: Vec::new(), + series_attrs: Vec::new(), + series_buckets: Vec::new(), + + batch_key_to_tracker: Vec::new(), + batch_val_to_tracker: Vec::new(), + temp_tracker_attrs: Vec::new(), + + active_count: 0, + stale_cursor: 0, + scrape_in_progress: false, + } + } + + /// Creates a new [`StalenessTracker`] pre-sized for `families` metric families, `series` + /// distinct time series, and `attrs` total label pairs. + #[must_use] + pub fn with_capacity(families: usize, series: usize, attrs: usize) -> Self { + if families == 0 && series == 0 && attrs == 0 { + return Self::new(); + } + let fam_bucket_cap = (families.saturating_mul(2)).next_power_of_two().max(16); + let ser_bucket_cap = (series.saturating_mul(2)).next_power_of_two().max(16); + Self { + families: Vec::with_capacity(families), + family_bytes: Vec::with_capacity(families.saturating_mul(32)), + family_buckets: vec![FamilyBucket::EMPTY; fam_bucket_cap], + + key_interner: StringInterner::with_capacity(families.min(64), families * 8), + val_interner: StringInterner::with_capacity(series.min(256), attrs * 8), + + series: Vec::with_capacity(series), + series_attrs: Vec::with_capacity(attrs), + series_buckets: vec![SeriesBucket::EMPTY; ser_bucket_cap], + + batch_key_to_tracker: Vec::with_capacity(families.min(64)), + batch_val_to_tracker: Vec::with_capacity(series.min(256)), + temp_tracker_attrs: Vec::with_capacity(16), + + active_count: 0, + stale_cursor: 0, + scrape_in_progress: false, + } + } + + /// Returns the number of currently active (non-stale) time series tracked for the target. + #[inline] + #[must_use] + pub fn active_series_len(&self) -> usize { + self.active_count + } + + /// Returns `true` if no active time series are currently tracked for the target. + #[inline] + #[must_use] + pub fn is_empty(&self) -> bool { + self.active_count == 0 + } + + /// Resets the tracker, clearing all tracked series and family metadata while retaining + /// allocated buffer capacities. + pub fn clear(&mut self) { + self.families.clear(); + self.family_bytes.clear(); + self.family_buckets.fill(FamilyBucket::EMPTY); + self.key_interner.clear(); + self.val_interner.clear(); + self.series.clear(); + self.series_attrs.clear(); + self.series_buckets.fill(SeriesBucket::EMPTY); + self.batch_key_to_tracker.clear(); + self.batch_val_to_tracker.clear(); + self.temp_tracker_attrs.clear(); + self.active_count = 0; + self.stale_cursor = 0; + self.scrape_in_progress = false; + } + + /// Returns the total heap buffer capacity in bytes currently retained by this tracker. + #[must_use] + pub fn retained_memory_bytes(&self) -> usize { + self.families.capacity() * size_of::() + + self.family_bytes.capacity() + + self.family_buckets.capacity() * size_of::() + + self.key_interner.retained_memory_bytes() + + self.val_interner.retained_memory_bytes() + + self.series.capacity() * size_of::() + + self.series_attrs.capacity() * size_of::<(u32, u32)>() + + self.series_buckets.capacity() * size_of::() + + self.batch_key_to_tracker.capacity() * size_of::() + + self.batch_val_to_tracker.capacity() * size_of::() + + self.temp_tracker_attrs.capacity() * size_of::<(u32, u32)>() + } + + /// Prepares the tracker for a new scrape. + /// + /// If a previous streaming scrape was dropped before completion, any uncommitted per-scrape + /// observation flags are rolled back first. + pub(super) fn begin_scrape(&mut self) { + if self.scrape_in_progress { + self.abort_scrape(); + } + for fam in &mut self.families { + fam.current_batch_id = None; + fam.has_pending_meta = false; + } + for s in &mut self.series { + s.seen = SeenState::Unseen; + } + self.batch_key_to_tracker.clear(); + self.batch_val_to_tracker.clear(); + self.stale_cursor = 0; + self.scrape_in_progress = true; + } + + /// Clears per-batch metric IDs and interner translation caches across streaming chunk flushes. + pub(super) fn reset_batch_ids(&mut self) { + for fam in &mut self.families { + fam.current_batch_id = None; + } + self.batch_key_to_tracker.clear(); + self.batch_val_to_tracker.clear(); + } + + /// Rolls back uncommitted observation flags and staged family metadata from a failed or + /// early-dropped scrape so the committed active series set from the previous successful scrape + /// remains intact. + pub(super) fn abort_scrape(&mut self) { + if !self.scrape_in_progress { + return; + } + for fam in &mut self.families { + fam.current_batch_id = None; + fam.has_pending_meta = false; + } + for s in &mut self.series { + s.seen = SeenState::Unseen; + } + self.batch_key_to_tracker.clear(); + self.batch_val_to_tracker.clear(); + self.stale_cursor = 0; + self.scrape_in_progress = false; + } + + /// Commits staged family metadata and updates the active series set once all chunks and + /// `RecordBatch` finalizations for the scrape have succeeded. + pub(super) fn commit_scrape(&mut self) { + let mut active_count = 0usize; + for s in &mut self.series { + let live = s.seen == SeenState::SeenLive; + s.active = live; + if live { + active_count += 1; + } + s.seen = SeenState::Unseen; + } + for fam in &mut self.families { + if fam.has_pending_meta { + fam.help_start = fam.pending_help_start; + fam.help_end = fam.pending_help_end; + fam.unit_start = fam.pending_unit_start; + fam.unit_end = fam.pending_unit_end; + fam.has_help = fam.pending_has_help; + fam.has_unit = fam.pending_has_unit; + fam.has_pending_meta = false; + } + fam.current_batch_id = None; + } + self.batch_key_to_tracker.clear(); + self.batch_val_to_tracker.clear(); + self.active_count = active_count; + self.stale_cursor = 0; + self.scrape_in_progress = false; + + if active_count == 0 { + self.clear(); + } else if self.series.len() > active_count.saturating_mul(2).max(64) + || self.families.len() > active_count.saturating_mul(2).max(64) + { + self.compact(); + } + } + + /// Rebuilds the tracker's family, interner, and series arenas containing only currently + /// `active` series so deactivated series under label churn do not accumulate unbounded memory + /// or linear-scan overhead. + #[cold] + fn compact(&mut self) { + let active_count = self.active_count; + let mut fam_remap = vec![NONE_IDX; self.families.len()]; + let mut key_remap = vec![NONE_IDX; self.key_interner.len()]; + let mut val_remap = vec![NONE_IDX; self.val_interner.len()]; + + let mut next_families: Vec = + Vec::with_capacity(self.families.len().min(active_count)); + let mut next_family_bytes: Vec = Vec::new(); + let mut next_keys = StringInterner::with_capacity(self.key_interner.len(), 0); + let mut next_vals = + StringInterner::with_capacity(active_count.min(self.val_interner.len()), 0); + + let ser_bucket_cap = (active_count.saturating_mul(2)).next_power_of_two().max(16); + let mut next_series = Vec::with_capacity(active_count); + let mut next_series_attrs = Vec::with_capacity(active_count.saturating_mul(4)); + let mut next_series_buckets = vec![SeriesBucket::EMPTY; ser_bucket_cap]; + let ser_mask = ser_bucket_cap - 1; + + for idx in 0..self.series.len() { + let s = self.series[idx]; + if !s.active { + continue; + } + + let old_fam_idx = s.family_idx as usize; + let new_fam_idx = if fam_remap[old_fam_idx] != NONE_IDX { + fam_remap[old_fam_idx] + } else { + let fam = self.families[old_fam_idx]; + let name = &self.family_bytes[fam.name_start as usize..fam.name_end as usize]; + let name_start = next_family_bytes.len() as u32; + next_family_bytes.extend_from_slice(name); + let name_end = next_family_bytes.len() as u32; + + let (help_start, help_end) = if fam.has_help { + let h = &self.family_bytes[fam.help_start as usize..fam.help_end as usize]; + let hs = next_family_bytes.len() as u32; + next_family_bytes.extend_from_slice(h); + (hs, next_family_bytes.len() as u32) + } else { + (name_end, name_end) + }; + + let (unit_start, unit_end) = if fam.has_unit { + let u = &self.family_bytes[fam.unit_start as usize..fam.unit_end as usize]; + let us = next_family_bytes.len() as u32; + next_family_bytes.extend_from_slice(u); + (us, next_family_bytes.len() as u32) + } else { + (help_end, help_end) + }; + + let new_idx = next_families.len() as u32; + next_families.push(TrackedFamily { + name_start, + name_end, + help_start, + help_end, + unit_start, + unit_end, + pending_help_start: help_start, + pending_help_end: help_end, + pending_unit_start: unit_start, + pending_unit_end: unit_end, + metric_type: fam.metric_type, + saw_total_suffix: fam.saw_total_suffix, + has_help: fam.has_help, + has_unit: fam.has_unit, + pending_has_help: fam.has_help, + pending_has_unit: fam.has_unit, + has_pending_meta: false, + current_batch_id: None, + }); + fam_remap[old_fam_idx] = new_idx; + new_idx + }; + + let attr_start = s.attr_start as usize; + let attr_end = attr_start + s.attr_len as usize; + self.temp_tracker_attrs.clear(); + let mut prev_nk: Option = None; + let mut sorted = true; + + for &(tk, tv) in &self.series_attrs[attr_start..attr_end] { + let tk_idx = tk as usize; + let nk = if key_remap[tk_idx] != NONE_IDX { + key_remap[tk_idx] + } else { + let id = next_keys + .intern(self.key_interner.get_bytes(tk)) + .expect("compacted key bytes <= existing interner bytes"); + key_remap[tk_idx] = id; + id + }; + + let tv_idx = tv as usize; + let nv = if val_remap[tv_idx] != NONE_IDX { + val_remap[tv_idx] + } else { + let id = next_vals + .intern(self.val_interner.get_bytes(tv)) + .expect("compacted val bytes <= existing interner bytes"); + val_remap[tv_idx] = id; + id + }; + + if let Some(prev) = prev_nk + && nk < prev + { + sorted = false; + } + prev_nk = Some(nk); + self.temp_tracker_attrs.push((nk, nv)); + } + + if !sorted { + self.temp_tracker_attrs.sort_unstable_by_key(|&(k, _)| k); + } + + let new_attr_start = next_series_attrs.len() as u32; + let new_attr_len = self.temp_tracker_attrs.len() as u32; + next_series_attrs.extend_from_slice(&self.temp_tracker_attrs); + + let new_series_idx = next_series.len() as u32; + next_series.push(TrackedSeries { + family_idx: new_fam_idx, + attr_start: new_attr_start, + attr_len: new_attr_len, + kind: s.kind, + active: true, + seen: SeenState::Unseen, + }); + + let hash = hash_series(new_fam_idx, s.kind, &self.temp_tracker_attrs); + let mut slot = (hash as usize) & ser_mask; + while next_series_buckets[slot].series_idx != NONE_IDX { + slot = (slot + 1) & ser_mask; + } + next_series_buckets[slot] = SeriesBucket { + hash, + series_idx: new_series_idx, + }; + } + + let fam_bucket_cap = (next_families.len().saturating_mul(2)) + .next_power_of_two() + .max(16); + let fam_mask = fam_bucket_cap - 1; + let mut next_family_buckets = vec![FamilyBucket::EMPTY; fam_bucket_cap]; + for (idx, fam) in next_families.iter().enumerate() { + let name = &next_family_bytes[fam.name_start as usize..fam.name_end as usize]; + let hash = hash_family_key(name, fam.metric_type, fam.saw_total_suffix); + let mut slot = (hash as usize) & fam_mask; + while next_family_buckets[slot].family_idx != NONE_IDX { + slot = (slot + 1) & fam_mask; + } + next_family_buckets[slot] = FamilyBucket { + hash, + family_idx: idx as u32, + }; + } + + self.families = next_families; + self.family_bytes = next_family_bytes; + self.family_buckets = next_family_buckets; + self.key_interner = next_keys; + self.val_interner = next_vals; + self.series = next_series; + self.series_attrs = next_series_attrs; + self.series_buckets = next_series_buckets; + if self.batch_key_to_tracker.capacity() > self.key_interner.len().saturating_mul(4).max(64) + { + self.batch_key_to_tracker.shrink_to(64); + } + if self.batch_val_to_tracker.capacity() > self.val_interner.len().saturating_mul(4).max(256) + { + self.batch_val_to_tracker + .shrink_to(self.val_interner.len().saturating_mul(2).max(64)); + } + } + + /// Records a data point observed during the current scrape. + pub(super) fn record_series( + &mut self, + current_family: &mut CurrentFamily, + kind: DpKind, + batch_attrs: &[(u32, u32)], + batch_keys: &StringInterner, + batch_vals: &StringInterner, + is_stale: bool, + ) -> Result<(), DecodeError> { + let fam_idx = self.find_or_insert_family(current_family)?; + + self.temp_tracker_attrs.clear(); + let mut prev_tk: Option = None; + let mut sorted = true; + + for &(bk, bv) in batch_attrs { + // Map only the batch IDs actually referenced by series attributes (`NONE_IDX` marks + // unmapped slots) so ephemeral exemplar labels in `batch_keys`/`batch_vals` are never + // copied into the tracker's persistent interners. + let bk_idx = bk as usize; + if bk_idx >= self.batch_key_to_tracker.len() { + self.batch_key_to_tracker.resize(bk_idx + 1, NONE_IDX); + } + let mut tk = self.batch_key_to_tracker[bk_idx]; + if tk == NONE_IDX { + tk = self + .key_interner + .intern(batch_keys.get_bytes(bk)) + .ok_or_else(|| DecodeError::CapacityOverflow { + table: "DpAttrs.key", + count: self.key_interner.len(), + })?; + self.batch_key_to_tracker[bk_idx] = tk; + } + + let bv_idx = bv as usize; + if bv_idx >= self.batch_val_to_tracker.len() { + self.batch_val_to_tracker.resize(bv_idx + 1, NONE_IDX); + } + let mut tv = self.batch_val_to_tracker[bv_idx]; + if tv == NONE_IDX { + tv = self + .val_interner + .intern(batch_vals.get_bytes(bv)) + .ok_or_else(|| DecodeError::CapacityOverflow { + table: "DpAttrs.str", + count: self.val_interner.len(), + })?; + self.batch_val_to_tracker[bv_idx] = tv; + } + + if let Some(prev) = prev_tk + && tk < prev + { + sorted = false; + } + prev_tk = Some(tk); + self.temp_tracker_attrs.push((tk, tv)); + } + + if !sorted { + self.temp_tracker_attrs.sort_unstable_by_key(|&(k, _)| k); + } + + if self.series_buckets.is_empty() { + self.series_buckets.resize(16, SeriesBucket::EMPTY); + } + let hash = hash_series(fam_idx, kind, &self.temp_tracker_attrs); + let mask = self.series_buckets.len() - 1; + let mut slot = (hash as usize) & mask; + + let seen_state = if is_stale { + SeenState::SeenStale + } else { + SeenState::SeenLive + }; + + loop { + let bucket = self.series_buckets[slot]; + if bucket.series_idx == NONE_IDX { + break; + } + if bucket.hash == hash { + let idx = bucket.series_idx as usize; + let s = &self.series[idx]; + if s.family_idx == fam_idx && s.kind == kind { + let start = s.attr_start as usize; + let end = start + s.attr_len as usize; + if self.series_attrs[start..end] == self.temp_tracker_attrs[..] { + self.series[idx].seen = seen_state; + return Ok(()); + } + } + } + slot = (slot + 1) & mask; + } + + let attr_start = + u32::try_from(self.series_attrs.len()).map_err(|_| DecodeError::CapacityOverflow { + table: "StalenessTracker.series_attrs", + count: self.series_attrs.len(), + })?; + let attr_len = u32::try_from(self.temp_tracker_attrs.len()).map_err(|_| { + DecodeError::CapacityOverflow { + table: "StalenessTracker.series_attrs", + count: self.temp_tracker_attrs.len(), + } + })?; + let series_idx = + u32::try_from(self.series.len()).map_err(|_| DecodeError::CapacityOverflow { + table: "StalenessTracker.series", + count: self.series.len() + 1, + })?; + if series_idx == NONE_IDX { + return Err(DecodeError::CapacityOverflow { + table: "StalenessTracker.series", + count: self.series.len() + 1, + }); + } + + self.series_attrs + .extend_from_slice(&self.temp_tracker_attrs); + self.series.push(TrackedSeries { + family_idx: fam_idx, + kind, + attr_start, + attr_len, + active: false, + seen: seen_state, + }); + self.series_buckets[slot] = SeriesBucket { hash, series_idx }; + + if self.series.len() * 10 >= self.series_buckets.len() * 7 { + self.rehash_series(); + } + Ok(()) + } + + #[inline] + fn find_or_insert_family( + &mut self, + current_family: &mut CurrentFamily, + ) -> Result { + let (name, metric_type, saw_total_suffix) = canonical_family_key(current_family); + if let Some(idx) = current_family.tracker_family_idx { + let fam = &mut self.families[idx as usize]; + if fam.metric_type == metric_type && fam.saw_total_suffix == saw_total_suffix { + fam.current_batch_id = current_family.batch_id; + return Ok(idx); + } + } + + if self.family_buckets.is_empty() { + self.family_buckets.resize(16, FamilyBucket::EMPTY); + } + let hash = hash_family_key(name, metric_type, saw_total_suffix); + let mask = self.family_buckets.len() - 1; + let mut slot = (hash as usize) & mask; + + loop { + let bucket = self.family_buckets[slot]; + if bucket.family_idx == NONE_IDX { + break; + } + if bucket.hash == hash { + let idx = bucket.family_idx as usize; + let fam = self.families[idx]; + if fam.metric_type == metric_type + && fam.saw_total_suffix == saw_total_suffix + && &self.family_bytes[fam.name_start as usize..fam.name_end as usize] == name + { + self.update_existing_family(idx, current_family)?; + current_family.tracker_family_idx = Some(bucket.family_idx); + return Ok(bucket.family_idx); + } + } + slot = (slot + 1) & mask; + } + + let (name_start, name_end) = self.append_family_bytes(name)?; + let (help_start, help_end) = if current_family.has_help { + self.append_family_bytes(¤t_family.help)? + } else { + (name_end, name_end) + }; + let (unit_start, unit_end) = if current_family.has_unit { + self.append_family_bytes(¤t_family.unit)? + } else { + (help_end, help_end) + }; + + let family_idx = + u32::try_from(self.families.len()).map_err(|_| DecodeError::CapacityOverflow { + table: "StalenessTracker.families", + count: self.families.len() + 1, + })?; + if family_idx == NONE_IDX { + return Err(DecodeError::CapacityOverflow { + table: "StalenessTracker.families", + count: self.families.len() + 1, + }); + } + + self.families.push(TrackedFamily { + name_start, + name_end, + help_start, + help_end, + unit_start, + unit_end, + pending_help_start: help_start, + pending_help_end: help_end, + pending_unit_start: unit_start, + pending_unit_end: unit_end, + metric_type, + saw_total_suffix, + has_help: false, + has_unit: false, + pending_has_help: current_family.has_help, + pending_has_unit: current_family.has_unit, + has_pending_meta: true, + current_batch_id: current_family.batch_id, + }); + self.family_buckets[slot] = FamilyBucket { hash, family_idx }; + + if self.families.len() * 10 >= self.family_buckets.len() * 7 { + self.rehash_families(); + } + current_family.tracker_family_idx = Some(family_idx); + Ok(family_idx) + } + + #[inline] + fn update_existing_family( + &mut self, + idx: usize, + current_family: &CurrentFamily, + ) -> Result<(), DecodeError> { + let fam = self.families[idx]; + let (help_start, help_end) = if current_family.has_help { + if &self.family_bytes[fam.help_start as usize..fam.help_end as usize] + == current_family.help.as_slice() + { + (fam.help_start, fam.help_end) + } else { + self.append_family_bytes(¤t_family.help)? + } + } else { + (fam.help_start, fam.help_end) + }; + + let (unit_start, unit_end) = if current_family.has_unit { + if &self.family_bytes[fam.unit_start as usize..fam.unit_end as usize] + == current_family.unit.as_slice() + { + (fam.unit_start, fam.unit_end) + } else { + self.append_family_bytes(¤t_family.unit)? + } + } else { + (fam.unit_start, fam.unit_end) + }; + + let entry = &mut self.families[idx]; + entry.pending_help_start = help_start; + entry.pending_help_end = help_end; + entry.pending_unit_start = unit_start; + entry.pending_unit_end = unit_end; + entry.pending_has_help = current_family.has_help; + entry.pending_has_unit = current_family.has_unit; + entry.has_pending_meta = true; + entry.current_batch_id = current_family.batch_id; + Ok(()) + } + + #[inline] + fn append_family_bytes(&mut self, slice: &[u8]) -> Result<(u32, u32), DecodeError> { + let start = + u32::try_from(self.family_bytes.len()).map_err(|_| DecodeError::CapacityOverflow { + table: "StalenessTracker.family_bytes", + count: self.family_bytes.len(), + })?; + let next_len = self.family_bytes.len().saturating_add(slice.len()); + let end = u32::try_from(next_len).map_err(|_| DecodeError::CapacityOverflow { + table: "StalenessTracker.family_bytes", + count: next_len, + })?; + self.family_bytes.extend_from_slice(slice); + Ok((start, end)) + } + + #[cold] + fn rehash_families(&mut self) { + let new_cap = (self.family_buckets.len() * 2).max(16); + let mask = new_cap - 1; + let mut next_buckets = vec![FamilyBucket::EMPTY; new_cap]; + for &bucket in &self.family_buckets { + if bucket.family_idx == NONE_IDX { + continue; + } + let mut slot = (bucket.hash as usize) & mask; + while next_buckets[slot].family_idx != NONE_IDX { + slot = (slot + 1) & mask; + } + next_buckets[slot] = bucket; + } + self.family_buckets = next_buckets; + } + + #[cold] + fn rehash_series(&mut self) { + let new_cap = (self.series_buckets.len() * 2).max(16); + let mask = new_cap - 1; + let mut next_buckets = vec![SeriesBucket::EMPTY; new_cap]; + for &bucket in &self.series_buckets { + if bucket.series_idx == NONE_IDX { + continue; + } + let mut slot = (bucket.hash as usize) & mask; + while next_buckets[slot].series_idx != NONE_IDX { + slot = (slot + 1) & mask; + } + next_buckets[slot] = bucket; + } + self.series_buckets = next_buckets; + } + + /// Emits stale data points for all previously active series that were not observed during the + /// current scrape, bounded by `max_data_points` when chunked streaming is active and by + /// `step_budget` (`(remaining_entries, started, max_duration)`) when cooperative budgeting is + /// active. + #[allow(clippy::too_many_arguments)] + pub(super) fn finish_scrape_step( + &mut self, + metrics: &mut MetricTable, + num_dp: &mut NumberDpTable, + sum_dp: &mut SummaryDpTable, + hist_dp: &mut HistogramDpTable, + batch_keys: &mut StringInterner, + batch_vals: &mut StringInterner, + opts: &DecodeOptions, + default_ts_ns: Option, + max_data_points: Option, + max_metrics: usize, + step_budget: Option<(usize, Instant, Duration)>, + ) -> Result { + let mut steps = 0usize; + while self.stale_cursor < self.series.len() { + if let Some((rem_entries, _, _)) = step_budget + && rem_entries == 0 + { + return Ok(StaleDrainStep::YieldPending); + } + + let idx = self.stale_cursor; + let s = self.series[idx]; + if !s.active || s.seen != SeenState::Unseen { + self.stale_cursor += 1; + } else { + let fam = self.families[s.family_idx as usize]; + if fam.metric_type == MetricType::Unknown + && opts.untyped_policy == UntypedPolicy::Drop + { + self.stale_cursor += 1; + } else { + if let Some(max_dps) = max_data_points { + let current_dps = + num_dp.base.len() + sum_dp.base.len() + hist_dp.base.len(); + let need_new_metric = fam.current_batch_id.is_none(); + if current_dps >= max_dps + || (need_new_metric && metrics.len() >= max_metrics) + { + return Ok(StaleDrainStep::ChunkFull); + } + } + + self.emit_single_stale_series( + s, + metrics, + num_dp, + sum_dp, + hist_dp, + batch_keys, + batch_vals, + opts, + default_ts_ns, + )?; + self.stale_cursor += 1; + } + } + + if let Some((rem_entries, started, max_duration)) = step_budget { + steps += 1; + if self.stale_cursor < self.series.len() { + let duration_exceeded = (max_duration.is_zero() + || steps.is_multiple_of(DECODE_BUDGET_DURATION_CHECK_INTERVAL)) + && started.elapsed() >= max_duration; + if steps >= rem_entries || duration_exceeded { + return Ok(StaleDrainStep::YieldPending); + } + } + } + } + + Ok(StaleDrainStep::Complete) + } + + #[allow(clippy::too_many_arguments)] + fn emit_single_stale_series( + &mut self, + s: TrackedSeries, + metrics: &mut MetricTable, + num_dp: &mut NumberDpTable, + sum_dp: &mut SummaryDpTable, + hist_dp: &mut HistogramDpTable, + batch_keys: &mut StringInterner, + batch_vals: &mut StringInterner, + opts: &DecodeOptions, + default_ts_ns: Option, + ) -> Result<(), DecodeError> { + let fam_idx = s.family_idx as usize; + let parent_id = if let Some(id) = self.families[fam_idx].current_batch_id { + id + } else { + let fam = self.families[fam_idx]; + let (has_help, help_start, help_end, has_unit, unit_start, unit_end) = + if fam.has_pending_meta { + ( + fam.pending_has_help, + fam.pending_help_start, + fam.pending_help_end, + fam.pending_has_unit, + fam.pending_unit_start, + fam.pending_unit_end, + ) + } else { + ( + fam.has_help, + fam.help_start, + fam.help_end, + fam.has_unit, + fam.unit_start, + fam.unit_end, + ) + }; + let id = emit_family_row( + &self.family_bytes[fam.name_start as usize..fam.name_end as usize], + fam.metric_type, + fam.saw_total_suffix, + has_help, + &self.family_bytes[help_start as usize..help_end as usize], + has_unit, + &self.family_bytes[unit_start as usize..unit_end as usize], + metrics, + opts, + )?; + self.families[fam_idx].current_batch_id = Some(id); + id + }; + + let attr_start = s.attr_start as usize; + let attr_end = attr_start + s.attr_len as usize; + self.temp_tracker_attrs.clear(); + let mut prev_bk: Option = None; + let mut sorted = true; + + for &(tk, tv) in &self.series_attrs[attr_start..attr_end] { + let bk = batch_keys + .intern(self.key_interner.get_bytes(tk)) + .ok_or_else(|| DecodeError::CapacityOverflow { + table: "DpAttrs.key", + count: batch_keys.len(), + })?; + let bv = batch_vals + .intern(self.val_interner.get_bytes(tv)) + .ok_or_else(|| DecodeError::CapacityOverflow { + table: "DpAttrs.str", + count: batch_vals.len(), + })?; + if let Some(prev) = prev_bk + && bk < prev + { + sorted = false; + } + prev_bk = Some(bk); + self.temp_tracker_attrs.push((bk, bv)); + } + + if !sorted { + self.temp_tracker_attrs.sort_unstable_by_key(|&(k, _)| k); + } + + match s.kind { + DpKind::Number => { + let dp_index = num_dp.base.len(); + let dp_id = u32::try_from(dp_index).map_err(|_| DecodeError::CapacityOverflow { + table: "NumberDataPoints", + count: dp_index + 1, + })?; + num_dp + .attrs + .extend_for_parent(dp_id, &self.temp_tracker_attrs); + num_dp.base.push_with_flags( + parent_id, + None, + default_ts_ns, + DATA_POINT_FLAG_NO_RECORDED_VALUE, + ); + num_dp.values.push(dp_index, None); + } + DpKind::Summary => { + if sum_dp.quantile_offsets.is_empty() { + sum_dp.quantile_offsets.push(0); + } + let dp_index = sum_dp.base.len(); + let dp_id = u32::try_from(dp_index).map_err(|_| DecodeError::CapacityOverflow { + table: "SummaryDataPoints", + count: dp_index + 1, + })?; + sum_dp + .attrs + .extend_for_parent(dp_id, &self.temp_tracker_attrs); + sum_dp.base.push_with_flags( + parent_id, + None, + default_ts_ns, + DATA_POINT_FLAG_NO_RECORDED_VALUE, + ); + sum_dp.counts.push(dp_index, Some(0)); + sum_dp.sums.push(dp_index, Some(0.0)); + let q_off = i32::try_from(sum_dp.quantile_bounds.len()).map_err(|_| { + DecodeError::CapacityOverflow { + table: "SummaryDataPoints.quantile", + count: sum_dp.quantile_bounds.len(), + } + })?; + sum_dp.quantile_offsets.push(q_off); + } + DpKind::Histogram => { + if hist_dp.bucket_offsets.is_empty() { + hist_dp.bucket_offsets.push(0); + hist_dp.bounds_offsets.push(0); + } + let dp_index = hist_dp.base.len(); + let dp_id = u32::try_from(dp_index).map_err(|_| DecodeError::CapacityOverflow { + table: "HistogramDataPoints", + count: dp_index + 1, + })?; + hist_dp + .attrs + .extend_for_parent(dp_id, &self.temp_tracker_attrs); + hist_dp.base.push_with_flags( + parent_id, + None, + default_ts_ns, + DATA_POINT_FLAG_NO_RECORDED_VALUE, + ); + hist_dp.counts.push(dp_index, Some(0)); + hist_dp.sums.push(dp_index, None); + let bucket_off = i32::try_from(hist_dp.bucket_counts.len()).map_err(|_| { + DecodeError::CapacityOverflow { + table: "HistogramDataPoints.histogram_bucket_counts", + count: hist_dp.bucket_counts.len(), + } + })?; + let bounds_off = i32::try_from(hist_dp.bounds_vals.len()).map_err(|_| { + DecodeError::CapacityOverflow { + table: "HistogramDataPoints.histogram_explicit_bounds", + count: hist_dp.bounds_vals.len(), + } + })?; + hist_dp.bucket_offsets.push(bucket_off); + hist_dp.bounds_offsets.push(bounds_off); + } + } + Ok(()) + } +} diff --git a/src/arrow/tests.rs b/src/arrow/tests.rs index 7b0b074..4480df2 100644 --- a/src/arrow/tests.rs +++ b/src/arrow/tests.rs @@ -1194,3 +1194,697 @@ fn cooperative_async_helpers_respect_fast_path_and_yield_budgets() { assert!(stream_err_res.1.is_none()); assert_eq!(stream_err_polls, 1); } + +#[test] +fn stale_nan_bit_pattern_and_multi_scrape_staleness_lifecycle() { + assert!(is_stale_nan(STALE_NAN)); + assert_eq!(STALE_NAN.to_bits(), STALE_NAN_BITS); + assert!(!is_stale_nan(f64::NAN)); + assert!(!is_stale_nan(f64::from_bits(0x7ff8_0000_0000_0001))); + assert!(!is_stale_nan(0.0)); + + let mut dec = ArrowDecoder::new(); + let mut tracker = StalenessTracker::new(); + assert!(tracker.is_empty()); + assert_eq!(tracker.active_series_len(), 0); + + let scrape1 = concat!( + "# HELP req_total Total requests.\n", + "# TYPE req_total counter\n", + "req_total{id=\"1\"} 10\n", + "req_total{id=\"2\"} 20\n", + "# HELP mem_bytes Memory usage.\n", + "# TYPE mem_bytes gauge\n", + "mem_bytes 512\n", + "# HELP rpc_seconds RPC latency.\n", + "# TYPE rpc_seconds summary\n", + "rpc_seconds{svc=\"auth\",quantile=\"0.5\"} 0.05\n", + "rpc_seconds_sum{svc=\"auth\"} 1.5\n", + "rpc_seconds_count{svc=\"auth\"} 10\n", + "# HELP lat_seconds Request latency.\n", + "# TYPE lat_seconds histogram\n", + "lat_seconds_bucket{route=\"/a\",le=\"0.1\"} 3\n", + "lat_seconds_bucket{route=\"/a\",le=\"+Inf\"} 5\n", + "lat_seconds_sum{route=\"/a\"} 0.8\n", + "lat_seconds_count{route=\"/a\"} 5\n", + ); + + let opts1 = DecodeOptions::new().with_timestamp_ms(1_000); + let rec1 = dec + .decode_with_staleness(scrape1.as_bytes(), &Format::TEXT, &opts1, &mut tracker) + .unwrap(); + assert_eq!(tracker.active_series_len(), 5); + assert!(!tracker.is_empty()); + assert_eq!(rec1.num_metrics(), 4); + assert_eq!(rec1.num_data_points(), 5); + // Non-stale batches must omit the optional `flags` column completely. + assert!( + rec1.number_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + assert!( + rec1.summary_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + assert!( + rec1.histogram_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + + // Scrape 2: steady-state scrape with identical series set allocates zero new tracker capacity + // and emits no `flags` column. + let warmup_tracker_mem = tracker.retained_memory_bytes(); + let opts2 = DecodeOptions::new().with_timestamp_ms(2_000); + let rec2 = dec + .decode_with_staleness(scrape1.as_bytes(), &Format::TEXT, &opts2, &mut tracker) + .unwrap(); + assert_eq!(tracker.active_series_len(), 5); + assert_eq!(tracker.retained_memory_bytes(), warmup_tracker_mem); + assert!( + rec2.number_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + + // Scrape 3: `req_total{id="2"}`, `mem_bytes`, `rpc_seconds{svc="auth"}`, and + // `lat_seconds{route="/a"}` disappear; only `req_total{id="1"}` remains live. + let scrape3 = concat!( + "# HELP req_total Total requests.\n", + "# TYPE req_total counter\n", + "req_total{id=\"1\"} 15\n", + ); + let opts3 = DecodeOptions::new().with_timestamp_ms(3_000); + let rec3 = dec + .decode_with_staleness(scrape3.as_bytes(), &Format::TEXT, &opts3, &mut tracker) + .unwrap(); + assert_eq!(tracker.active_series_len(), 1); + assert_eq!(rec3.num_metrics(), 4); + assert_eq!(rec3.num_data_points(), 5); + + let ndp3 = rec3.number_data_points().unwrap(); + assert_eq!(ndp3.num_rows(), 3); + let ndp3_flags = ndp3 + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let ndp3_vals = ndp3 + .column_by_name("double_value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let ndp3_parents = ndp3 + .column_by_name("parent_id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + // Row 0 is live `req_total{id="1"}`; Row 1 is stale `req_total{id="2"}` sharing parent_id 0; + // Row 2 is stale `mem_bytes` with parent_id 1. + assert_eq!(ndp3_flags.value(0), 0); + assert!(ndp3_vals.is_valid(0)); + assert_eq!(ndp3_vals.value(0), 15.0); + assert_eq!(ndp3_parents.value(0), 0); + + assert_eq!(ndp3_flags.value(1), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert!(ndp3_vals.is_null(1)); + assert_eq!(ndp3_parents.value(1), 0); + + assert_eq!(ndp3_flags.value(2), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert!(ndp3_vals.is_null(2)); + assert_eq!(ndp3_parents.value(2), 1); + + let sdp3 = rec3.summary_data_points().unwrap(); + assert_eq!(sdp3.num_rows(), 1); + let sdp3_flags = sdp3 + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let sdp3_sums = sdp3 + .column_by_name("sum") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(sdp3_flags.value(0), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert_eq!(sdp3_sums.value(0), 0.0); + + let hdp3 = rec3.histogram_data_points().unwrap(); + assert_eq!(hdp3.num_rows(), 1); + let hdp3_flags = hdp3 + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let hdp3_sums = hdp3 + .column_by_name("sum") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(hdp3_flags.value(0), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert!(hdp3_sums.is_null(0)); + + // Scrape 4: disappeared series are not re-emitted, and `req_total{id="2"}` can reappear. + let scrape4 = concat!( + "# HELP req_total Total requests.\n", + "# TYPE req_total counter\n", + "req_total{id=\"1\"} 18\n", + "req_total{id=\"2\"} 25\n", + ); + let opts4 = DecodeOptions::new().with_timestamp_ms(4_000); + let rec4 = dec + .decode_with_staleness(scrape4.as_bytes(), &Format::TEXT, &opts4, &mut tracker) + .unwrap(); + assert_eq!(tracker.active_series_len(), 2); + assert_eq!(rec4.num_metrics(), 1); + assert_eq!(rec4.num_data_points(), 2); + assert!( + rec4.number_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + assert!(rec4.summary_data_points().is_none()); + assert!(rec4.histogram_data_points().is_none()); + + // Scrape 5: a parse error mid-scrape rolls back the tracker so `active_series_len()` stays 2. + let bad_scrape = + b"# TYPE req_total counter\nreq_total{id=\"1\"} 20\nreq_total{id=\"1\",id=\"1\"} 21\n"; + assert!( + dec.decode_with_staleness(bad_scrape, &Format::TEXT, &opts4, &mut tracker) + .is_err() + ); + assert_eq!(tracker.active_series_len(), 2); + + // `flush_target_stale` marks all remaining active series stale and empties the tracker. + assert!(format!("{tracker:?}").contains("StalenessTracker")); + let opts5 = DecodeOptions::new().with_timestamp_ms(5_000); + let flushed = dec.flush_target_stale(&mut tracker, &opts5).unwrap(); + assert!(tracker.is_empty()); + assert_eq!(tracker.active_series_len(), 0); + assert_eq!(flushed.num_metrics(), 1); + assert_eq!(flushed.num_data_points(), 2); + let flushed_um = flushed.univariate_metrics().unwrap(); + let flushed_desc = flushed_um + .column_by_name("description") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(flushed_desc.value(0), "Total requests."); + let flushed_ndp = flushed.number_data_points().unwrap(); + let flushed_flags = flushed_ndp + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let flushed_vals = flushed_ndp + .column_by_name("double_value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let flushed_ts = flushed_ndp + .column_by_name("time_unix_nano") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(flushed_flags.value(0), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert_eq!(flushed_flags.value(1), DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert!(flushed_vals.is_null(0)); + assert!(flushed_vals.is_null(1)); + assert_eq!(flushed_ts.value(0), 5_000_000_000); + assert_eq!(flushed_ts.value(1), 5_000_000_000); + assert_eq!(flushed.number_dp_attrs().unwrap().num_rows(), 2); + + // A second flush on an already-empty tracker produces an empty record set, and consecutive + // empty scrapes return empty records cleanly. + let empty_flush = dec.flush_target_stale(&mut tracker, &opts5).unwrap(); + assert!(empty_flush.is_empty()); + assert!( + dec.decode_with_staleness(b"", &Format::TEXT, &opts5, &mut tracker) + .unwrap() + .is_empty() + ); +} + +#[test] +fn decode_stream_with_staleness_chunks_disappeared_series_and_rolls_back_on_drop() { + let mut dec = ArrowDecoder::new(); + let mut tracker = StalenessTracker::new(); + let opts = DecodeOptions::new() + .with_timestamp_ms(1_000) + .with_max_data_points_per_batch(2); + + let scrape1 = concat!( + "# TYPE g gauge\n", + "g{k=\"1\"} 1\n", + "g{k=\"2\"} 2\n", + "g{k=\"3\"} 3\n", + "g{k=\"4\"} 4\n", + ); + let chunks1: Vec<_> = dec + .decode_stream_with_staleness(scrape1.as_bytes(), &Format::TEXT, &opts, &mut tracker) + .collect::, _>>() + .unwrap(); + assert_eq!(chunks1.len(), 2); + assert_eq!(tracker.active_series_len(), 4); + + // Early drop mid-scrape (including a newly observed series `k="new"`) and mid-stale-drain + // (dropping after the first chunk of disappeared series) both roll back cleanly. + { + let mut partial = dec.decode_stream_with_staleness( + b"# TYPE g gauge\ng{k=\"new\"} 99\ng{k=\"1\"} 10\ng{k=\"2\"} 20\n", + &Format::TEXT, + &opts, + &mut tracker, + ); + assert!(partial.next().unwrap().is_ok()); + } + assert_eq!(tracker.active_series_len(), 4); + { + let mut partial_drain = dec.decode_stream_with_staleness( + b"# TYPE g gauge\ng{k=\"1\"} 10\n", + &Format::TEXT, + &opts, + &mut tracker, + ); + // Chunk 1 has 1 live (`k="1"`) + 1 stale (`k="2"`), leaving `k="3"` and `k="4"` pending. + assert_eq!(partial_drain.next().unwrap().unwrap().num_data_points(), 2); + } + assert_eq!(tracker.active_series_len(), 4); + + // Next scrape has only `g{k="1"}`, so 3 series (`k="2"`, `k="3"`, `k="4"`) disappear and are + // emitted across chunks of at most 2 data points each (1 live + 1 stale, then 2 stale). + let opts2 = DecodeOptions::new() + .with_timestamp_ms(2_000) + .with_max_data_points_per_batch(2); + let chunks2: Vec<_> = dec + .decode_stream_with_staleness( + b"# TYPE g gauge\ng{k=\"1\"} 10\n", + &Format::TEXT, + &opts2, + &mut tracker, + ) + .collect::, _>>() + .unwrap(); + assert_eq!(chunks2.len(), 2); + assert_eq!(chunks2[0].num_data_points(), 2); + assert_eq!(chunks2[1].num_data_points(), 2); + assert_eq!(chunks2[0].num_metrics(), 1); + assert_eq!(chunks2[1].num_metrics(), 1); + let c0_flags = chunks2[0] + .number_data_points() + .unwrap() + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + c0_flags.values().as_ref(), + &[0, DATA_POINT_FLAG_NO_RECORDED_VALUE] + ); + let c1_flags = chunks2[1] + .number_data_points() + .unwrap() + .column_by_name("flags") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + c1_flags.values().as_ref(), + &[ + DATA_POINT_FLAG_NO_RECORDED_VALUE, + DATA_POINT_FLAG_NO_RECORDED_VALUE + ] + ); + assert_eq!(chunks2[0].number_dp_attrs().unwrap().num_rows(), 2); + assert_eq!(chunks2[1].number_dp_attrs().unwrap().num_rows(), 2); + assert_eq!(tracker.active_series_len(), 1); +} + +#[test] +fn staleness_rehashing_compaction_exemplar_isolation_and_key_reordering() { + let mut dec = ArrowDecoder::new(); + let mut tracker = StalenessTracker::with_capacity(0, 0, 0); + assert_eq!(tracker.retained_memory_bytes(), 0); + assert_eq!(StalenessTracker::default().retained_memory_bytes(), 0); + let opts = DecodeOptions::new().with_timestamp_ms(1_000); + + // 1. Trigger multiple `rehash_families` and `rehash_series` passes starting from zero capacity. + let mut initial = String::new(); + for f in 0..40 { + initial.push_str(&format!("# HELP fam_{f} Help {f}.\n# TYPE fam_{f} gauge\n")); + initial.push_str(&format!("fam_{f}{{a=\"1\",z=\"v{f}_0\"}} {f}\n")); + initial.push_str(&format!("fam_{f}{{a=\"2\",z=\"v{f}_1\"}} {f}\n")); + } + let rec1 = dec + .decode_with_staleness(initial.as_bytes(), &Format::TEXT, &opts, &mut tracker) + .unwrap(); + assert_eq!(rec1.num_data_points(), 80); + assert_eq!(tracker.active_series_len(), 80); + + // 2. Reverse label key order (`z` interned before `a` on both batch and tracker) to exercise + // `!sorted` attribute re-sorting in `record_series` and `emit_single_stale_series`. + let reversed_live = "# TYPE fam_0 gauge\nfam_0{z=\"v0_0\",a=\"1\"} 99\n"; + let rec2 = dec + .decode_with_staleness( + reversed_live.as_bytes(), + &Format::TEXT, + &opts.with_timestamp_ms(2_000), + &mut tracker, + ) + .unwrap(); + assert_eq!(rec2.num_data_points(), 80); // 1 live + 79 stale + assert_eq!(tracker.active_series_len(), 1); + + // 3. `clear()` resets active tracking while preserving capacity. + let cap_before_clear = tracker.retained_memory_bytes(); + assert!(cap_before_clear > 0); + tracker.clear(); + assert!(tracker.is_empty()); + assert_eq!(tracker.retained_memory_bytes(), cap_before_clear); + + // 4. Ephemeral OpenMetrics exemplar attributes must never enter `StalenessTracker` interners. + let mut ex_tracker = StalenessTracker::new(); + let mut baseline_ex_mem = 0usize; + for i in 0..16 { + let om = format!( + "# TYPE rpc_total counter\n\ + rpc_total{{env=\"prod\"}} {i} # {{req_id=\"ephemeral-{i:08}\",trace_id=\"nonhex-{i:08}\"}} 1.0\n\ + rpc_total{{env=\"staging\"}} {i}\n\ + # EOF\n" + ); + dec.decode_with_staleness( + om.as_bytes(), + &Format::OPEN_METRICS_1_0_0, + &opts.with_timestamp_ms(3_000 + i * 1_000), + &mut ex_tracker, + ) + .unwrap(); + if i == 0 { + baseline_ex_mem = ex_tracker.retained_memory_bytes(); + assert!(baseline_ex_mem > 0); + } else { + assert_eq!( + ex_tracker.retained_memory_bytes(), + baseline_ex_mem, + "ephemeral exemplar strings leaked into StalenessTracker at scrape {i}" + ); + } + } + + // 5. Series churn across rollouts triggers `compact()` and bounds `retained_memory_bytes()`. + let mut churn_tracker = StalenessTracker::new(); + let mut post_compaction_mem = 0usize; + for gen_idx in 0..12 { + let mut scrape = String::from("# TYPE pod_cpu gauge\n"); + for pod in 0..80 { + scrape.push_str(&format!("pod_cpu{{pod=\"pod-{gen_idx}-{pod}\"}} 1\n")); + } + dec.decode_with_staleness( + scrape.as_bytes(), + &Format::TEXT, + &opts.with_timestamp_ms(20_000 + gen_idx * 1_000), + &mut churn_tracker, + ) + .unwrap(); + assert_eq!(churn_tracker.active_series_len(), 80); + if gen_idx == 2 { + post_compaction_mem = churn_tracker.retained_memory_bytes(); + } else if gen_idx > 2 { + assert!( + churn_tracker.retained_memory_bytes() <= post_compaction_mem * 2, + "StalenessTracker memory grew unbounded under churn at generation {gen_idx}" + ); + } + } +} + +#[test] +fn staleness_wire_nans_type_transitions_policies_and_cooperative_cursor() { + use std::pin::pin; + use std::task::{Context, Poll, Waker}; + + let mut dec = ArrowDecoder::new(); + let mut tracker = StalenessTracker::new(); + let opts = DecodeOptions::new().with_timestamp_ms(1_000); + + // 1. Wire `"NaN"` (`NORMAL_NAN`) is treated as a live sample (no `flags` column) and emits + // `FLAG_NO_RECORDED_VALUE` when it disappears on the next scrape. + let nan_scrape = concat!( + "# TYPE g_nan gauge\ng_nan NaN\n", + "# TYPE s_nan summary\ns_nan{quantile=\"0.5\"} NaN\ns_nan_sum NaN\ns_nan_count 1\n", + "# TYPE h_nan histogram\nh_nan_bucket{le=\"+Inf\"} 1\nh_nan_sum NaN\nh_nan_count 1\n", + ); + let nan_rec1 = dec + .decode_with_staleness(nan_scrape.as_bytes(), &Format::TEXT, &opts, &mut tracker) + .unwrap(); + assert_eq!(tracker.active_series_len(), 3); + assert!( + nan_rec1 + .number_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + let nan_rec2 = dec + .decode_with_staleness( + b"", + &Format::TEXT, + &opts.with_timestamp_ms(2_000), + &mut tracker, + ) + .unwrap(); + assert!(tracker.is_empty()); + assert_eq!(nan_rec2.num_data_points(), 3); + + // 2. Direct check of `SparseFlagsCol` trailing-zero backfill (`flags != 0` at row 0 followed by + // `flags == 0` at row 1) and `SeenState::SeenStale` on incoming `STALE_NAN` values. + let mut sparse_flags = super::columns::SparseFlagsCol::default(); + sparse_flags.push(0, DATA_POINT_FLAG_NO_RECORDED_VALUE); + sparse_flags.push(1, 0); + let flags_arr = sparse_flags.to_array(); + let flags_u32 = flags_arr + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + flags_u32.values().as_ref(), + &[DATA_POINT_FLAG_NO_RECORDED_VALUE, 0] + ); + + // 3. `# TYPE` transitions (`histogram` -> `gauge`) and counter `_total` naming (`# TYPE req_total counter` + // -> `# TYPE req counter` with `req_total` samples). + let mut type_tracker = StalenessTracker::new(); + dec.decode_with_staleness( + b"# TYPE m histogram\nm_bucket{id=\"1\",le=\"+Inf\"} 2\nm_sum{id=\"1\"} 1.0\nm_count{id=\"1\"} 2\n\ + # TYPE req_total counter\nreq_total{id=\"1\"} 10\n", + &Format::TEXT, + &opts, + &mut type_tracker, + ) + .unwrap(); + assert_eq!(type_tracker.active_series_len(), 2); + + let trans_rec = dec + .decode_with_staleness( + b"# TYPE m gauge\nm{id=\"1\"} 5\n# TYPE req counter\nreq_total{id=\"1\"} 15\n", + &Format::TEXT, + &opts.with_timestamp_ms(2_000), + &mut type_tracker, + ) + .unwrap(); + // `m` (gauge) + `req_total` (counter, matched without false stale!) + `m` (stale histogram) = 3 metrics, 3 dps. + assert_eq!(trans_rec.num_metrics(), 3); + assert_eq!(trans_rec.num_data_points(), 3); + assert!( + trans_rec + .number_data_points() + .unwrap() + .column_by_name("flags") + .is_none() + ); + assert_eq!(trans_rec.histogram_data_points().unwrap().num_rows(), 1); + + // 4. `UntypedPolicy::Drop` suppresses stale emission for previously tracked untyped series, and + // `track_timestamps_staleness` skips explicit wire timestamps unless enabled. + let mut pol_tracker = StalenessTracker::new(); + dec.decode_with_staleness( + b"untyped_m 1\nts_m 2 1700000000000\n", + &Format::TEXT, + &opts, + &mut pol_tracker, + ) + .unwrap(); + // `ts_m` has an explicit timestamp and `track_timestamps_staleness` defaults to `false`. + assert_eq!(pol_tracker.active_series_len(), 1); + let dropped_stale = dec + .decode_with_staleness( + b"", + &Format::TEXT, + &opts + .with_timestamp_ms(2_000) + .with_untyped_policy(UntypedPolicy::Drop), + &mut pol_tracker, + ) + .unwrap(); + assert!(dropped_stale.is_empty()); + assert!(pol_tracker.is_empty()); + + let track_ts_opts = opts.with_track_timestamps_staleness(true); + dec.decode_with_staleness( + b"ts_m 2 1700000000000\n", + &Format::TEXT, + &track_ts_opts, + &mut pol_tracker, + ) + .unwrap(); + assert_eq!(pol_tracker.active_series_len(), 1); + let ts_stale = dec + .decode_with_staleness( + b"", + &Format::TEXT, + &track_ts_opts.with_timestamp_ms(3_000), + &mut pol_tracker, + ) + .unwrap(); + assert_eq!(ts_stale.num_data_points(), 1); + + // 5. Cooperative `step_decode_with_staleness`, `finish_decode_with_staleness`, and budgeted + // `DecodeStream::next_cooperative` yielding `Pending` during EOF stale draining. + let mut coop_tracker = StalenessTracker::new(); + dec.decode_with_staleness( + b"# TYPE g gauge\ng{k=\"1\"} 1\ng{k=\"2\"} 2\ng{k=\"3\"} 3\ng{k=\"4\"} 4\n", + &Format::TEXT, + &opts, + &mut coop_tracker, + ) + .unwrap(); + assert_eq!(coop_tracker.active_series_len(), 4); + + // Init errors leave `coop_tracker` intact. + assert!( + dec.decode_with_staleness(b"g 1\n", &Format::UNKNOWN, &opts, &mut coop_tracker) + .is_err() + ); + assert!( + dec.flush_target_stale(&mut coop_tracker, &opts.with_timestamp_ms(i64::MAX)) + .is_err() + ); + let mut bad_init_stream = + dec.decode_stream_with_staleness(b"g 1\n", &Format::UNKNOWN, &opts, &mut coop_tracker); + assert!(bad_init_stream.step(DecodeBudget::default()).is_err()); + drop(bad_init_stream); + assert_eq!(coop_tracker.active_series_len(), 4); + + // Aborting `step_decode_with_staleness` via `finish_decode_with_staleness` rolls back. + let budget_1 = DecodeBudget::default().with_max_entries(1); + let mut aborted_cur = dec.start_decode(&Format::TEXT, &opts).unwrap(); + assert_eq!( + dec.step_decode_with_staleness( + &mut aborted_cur, + b"# TYPE g gauge\ng{k=\"new\"} 9\n", + budget_1, + &mut coop_tracker + ) + .unwrap(), + DecodeStep::Pending + ); + dec.finish_decode_with_staleness(&mut aborted_cur, &mut coop_tracker); + assert_eq!(coop_tracker.active_series_len(), 4); + + // Drive `step_decode_with_staleness` on `b""` with `max_entries = 1` so EOF stale draining + // yields `DecodeStep::Pending` between disappeared series. + let mut drain_cur = dec + .start_decode(&Format::TEXT, &opts.with_timestamp_ms(2_000)) + .unwrap(); + let mut pending_drains = 0usize; + let mut drained_batch = None; + loop { + match dec + .step_decode_with_staleness(&mut drain_cur, b"", budget_1, &mut coop_tracker) + .unwrap() + { + DecodeStep::Pending => pending_drains += 1, + DecodeStep::Batch(b) => drained_batch = Some(b), + DecodeStep::Done => break, + } + } + assert!(pending_drains >= 3); + assert_eq!(drained_batch.unwrap().num_data_points(), 4); + assert!(coop_tracker.is_empty()); + + // Also verify `DecodeStream::next_cooperative` with `decode_stream_with_staleness`. + dec.decode_with_staleness( + b"# TYPE g gauge\ng{k=\"1\"} 1\ng{k=\"2\"} 2\ng{k=\"3\"} 3\n", + &Format::TEXT, + &opts, + &mut coop_tracker, + ) + .unwrap(); + struct YieldOnce(bool); + impl Future for YieldOnce { + type Output = (); + fn poll(mut self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + if self.0 { + Poll::Ready(()) + } else { + self.0 = true; + Poll::Pending + } + } + } + let stream_opts = opts + .with_timestamp_ms(3_000) + .with_max_data_points_per_batch(2); + let mut yields = 0usize; + let chunks = { + let fut = async { + let mut stream = dec.decode_stream_with_staleness( + b"", + &Format::TEXT, + &stream_opts, + &mut coop_tracker, + ); + let mut chunks = Vec::new(); + while let Some(res) = stream + .next_cooperative(budget_1, || { + yields += 1; + YieldOnce(false) + }) + .await + { + chunks.push(res.unwrap()); + } + chunks + }; + let mut pinned = pin!(fut); + let mut cx = Context::from_waker(Waker::noop()); + loop { + if let Poll::Ready(out) = pinned.as_mut().poll(&mut cx) { + break out; + } + } + }; + assert!(yields >= 2); + assert_eq!(chunks.len(), 2); + assert!(coop_tracker.is_empty()); +} diff --git a/tests/otap-compat/tests/arrow_otap.rs b/tests/otap-compat/tests/arrow_otap.rs index 441a770..9a02e46 100644 --- a/tests/otap-compat/tests/arrow_otap.rs +++ b/tests/otap-compat/tests/arrow_otap.rs @@ -20,9 +20,9 @@ use std::rc::Rc; use std::time::Duration; use expfmt::arrow::{ - ArrowDecoder, CounterSuffixPolicy, DecodeBudget, DecodeCursor, DecodeError, DecodeOptions, - DecodeStep, MetricPayloadType, OtapMetricRecords, PROMETHEUS_TYPE_METADATA_KEY, - UnitSuffixPolicy, UntypedPolicy, + ArrowDecoder, CounterSuffixPolicy, DATA_POINT_FLAG_NO_RECORDED_VALUE, DecodeBudget, + DecodeCursor, DecodeError, DecodeOptions, DecodeStep, MetricPayloadType, OtapMetricRecords, + PROMETHEUS_TYPE_METADATA_KEY, StalenessTracker, UnitSuffixPolicy, UntypedPolicy, }; use expfmt::format::Format; use otel_arrow_dfe_pdata::otap::{Metrics, OtapArrowRecords, OtapBatchStore}; @@ -1203,3 +1203,241 @@ fn cooperative_decode_bounds_local_set_timer_scheduling_lateness_and_validates_i assert_equivalent(&[expected_otlp], &streamed_otlp); }); } + +#[test] +fn staleness_tracker_and_flags_round_trip_through_otap_and_otlp() { + let mut decoder = ArrowDecoder::new(); + let mut tracker = StalenessTracker::new(); + + let scrape1 = concat!( + "# HELP req_total Total requests.\n", + "# TYPE req_total counter\n", + "req_total{instance=\"a\",job=\"api\"} 10\n", + "req_total{instance=\"b\",job=\"api\"} 20\n", + "# HELP rpc_seconds RPC duration.\n", + "# TYPE rpc_seconds summary\n", + "rpc_seconds{service=\"auth\",quantile=\"0.5\"} 0.02\n", + "rpc_seconds_sum{service=\"auth\"} 4.5\n", + "rpc_seconds_count{service=\"auth\"} 100\n", + "rpc_seconds{service=\"billing\",quantile=\"0.5\"} 0.01\n", + "rpc_seconds_sum{service=\"billing\"} 1.0\n", + "rpc_seconds_count{service=\"billing\"} 20\n", + "# HELP lat_seconds Latency histogram.\n", + "# TYPE lat_seconds histogram\n", + "lat_seconds_bucket{route=\"/v1\",le=\"0.1\"} 2\n", + "lat_seconds_bucket{route=\"/v1\",le=\"+Inf\"} 5\n", + "lat_seconds_sum{route=\"/v1\"} 1.25\n", + "lat_seconds_count{route=\"/v1\"} 5\n", + "lat_seconds_bucket{route=\"/v2\",le=\"0.1\"} 1\n", + "lat_seconds_bucket{route=\"/v2\",le=\"+Inf\"} 2\n", + "lat_seconds_sum{route=\"/v2\"} 0.3\n", + "lat_seconds_count{route=\"/v2\"} 2\n", + ); + + let opts1 = DecodeOptions::new().with_timestamp_ms(1_700_000_000_000); + let rec1 = decoder + .decode_with_staleness(scrape1.as_bytes(), &Format::TEXT, &opts1, &mut tracker) + .expect("scrape 1 decodes"); + assert_matches_otap_schema(&rec1); + assert_eq!(tracker.active_series_len(), 6); + + // Scrape 2 omits `req_total{instance="b",job="api"}`, `rpc_seconds{service="auth"}`, and + // `lat_seconds{route="/v1"}`, mixing live and stale rows in `NumberDataPoints`, + // `SummaryDataPoints`, and `HistogramDataPoints`. + let scrape2 = concat!( + "# HELP req_total Total requests.\n", + "# TYPE req_total counter\n", + "req_total{instance=\"a\",job=\"api\"} 15\n", + "# HELP rpc_seconds RPC duration.\n", + "# TYPE rpc_seconds summary\n", + "rpc_seconds{service=\"billing\",quantile=\"0.5\"} 0.01\n", + "rpc_seconds_sum{service=\"billing\"} 1.5\n", + "rpc_seconds_count{service=\"billing\"} 25\n", + "# HELP lat_seconds Latency histogram.\n", + "# TYPE lat_seconds histogram\n", + "lat_seconds_bucket{route=\"/v2\",le=\"0.1\"} 2\n", + "lat_seconds_bucket{route=\"/v2\",le=\"+Inf\"} 4\n", + "lat_seconds_sum{route=\"/v2\"} 0.6\n", + "lat_seconds_count{route=\"/v2\"} 4\n", + ); + let opts2 = DecodeOptions::new().with_timestamp_ms(1_700_000_015_000); + let rec2 = decoder + .decode_with_staleness(scrape2.as_bytes(), &Format::TEXT, &opts2, &mut tracker) + .expect("scrape 2 decodes with staleness"); + assert_eq!(tracker.active_series_len(), 3); + + // Verify chunked `decode_stream_with_staleness` (`max_data_points_per_batch = 2`) produces + // OTLP-equivalent metrics across chunks. + let mut stream_dec = ArrowDecoder::new(); + let mut stream_tracker = StalenessTracker::new(); + stream_dec + .decode_with_staleness( + scrape1.as_bytes(), + &Format::TEXT, + &opts1, + &mut stream_tracker, + ) + .unwrap(); + let streamed_otlp2: Vec<_> = stream_dec + .decode_stream_with_staleness( + scrape2.as_bytes(), + &Format::TEXT, + &opts2.with_max_data_points_per_batch(2), + &mut stream_tracker, + ) + .map(|res| { + let chunk = res.expect("streamed staleness chunk succeeds"); + let mut otap = to_otap_arrow_records(&chunk); + otap.encode_transport_optimized().unwrap(); + otap.decode_transport_optimized_ids().unwrap(); + otap_to_otlp(&otap) + }) + .collect(); + assert_eq!(stream_tracker.active_series_len(), 3); + + let mut otap2 = to_otap_arrow_records(&rec2); + let plain_otlp2 = otap_to_otlp(&otap2); + + otap2 + .encode_transport_optimized() + .expect("transport optimize succeeds on batches with flags"); + otap2 + .decode_transport_optimized_ids() + .expect("transport decode IDs succeeds on batches with flags"); + let round_tripped_otlp2 = otap_to_otlp(&otap2); + + let ts2_ns = 1_700_000_015_000_000_000_u64; + let expected2 = OtlpProtoMessage::Metrics(to_metrics_data(vec![ + Metric { + name: "req_total".to_owned(), + description: "Total requests.".to_owned(), + unit: String::new(), + metadata: vec![kv_str(PROMETHEUS_TYPE_METADATA_KEY, "counter")], + data: Some(metric::Data::Sum(Sum { + aggregation_temporality: AggregationTemporality::Cumulative as i32, + is_monotonic: true, + data_points: vec![ + number_dp(&[("instance", "a"), ("job", "api")], ts2_ns, 15.0), + NumberDataPoint { + attributes: vec![kv_str("instance", "b"), kv_str("job", "api")], + start_time_unix_nano: 0, + time_unix_nano: ts2_ns, + exemplars: Vec::new(), + flags: DATA_POINT_FLAG_NO_RECORDED_VALUE, + value: None, + }, + ], + })), + }, + Metric { + name: "rpc_seconds".to_owned(), + description: "RPC duration.".to_owned(), + unit: String::new(), + metadata: vec![kv_str(PROMETHEUS_TYPE_METADATA_KEY, "summary")], + data: Some(metric::Data::Summary(Summary { + data_points: vec![ + SummaryDataPoint { + attributes: vec![kv_str("service", "billing")], + start_time_unix_nano: 0, + time_unix_nano: ts2_ns, + count: 25, + sum: 1.5, + quantile_values: vec![ValueAtQuantile { + quantile: 0.5, + value: 0.01, + }], + flags: 0, + }, + SummaryDataPoint { + attributes: vec![kv_str("service", "auth")], + start_time_unix_nano: 0, + time_unix_nano: ts2_ns, + count: 0, + sum: 0.0, + quantile_values: Vec::new(), + flags: DATA_POINT_FLAG_NO_RECORDED_VALUE, + }, + ], + })), + }, + Metric { + name: "lat_seconds".to_owned(), + description: "Latency histogram.".to_owned(), + unit: String::new(), + metadata: vec![kv_str(PROMETHEUS_TYPE_METADATA_KEY, "histogram")], + data: Some(metric::Data::Histogram(Histogram { + aggregation_temporality: AggregationTemporality::Cumulative as i32, + data_points: vec![ + HistogramDataPoint { + attributes: vec![kv_str("route", "/v2")], + start_time_unix_nano: 0, + time_unix_nano: ts2_ns, + count: 4, + sum: Some(0.6), + bucket_counts: vec![2, 2], + explicit_bounds: vec![0.1], + exemplars: Vec::new(), + flags: 0, + min: None, + max: None, + }, + HistogramDataPoint { + attributes: vec![kv_str("route", "/v1")], + start_time_unix_nano: 0, + time_unix_nano: ts2_ns, + count: 0, + sum: None, + bucket_counts: Vec::new(), + explicit_bounds: Vec::new(), + exemplars: Vec::new(), + flags: DATA_POINT_FLAG_NO_RECORDED_VALUE, + min: None, + max: None, + }, + ], + })), + }, + ])); + + assert_equivalent(std::slice::from_ref(&expected2), &[plain_otlp2]); + assert_equivalent(std::slice::from_ref(&expected2), &[round_tripped_otlp2]); + assert_equivalent(&[expected2], &streamed_otlp2); + + // Target scrape failure: `flush_target_stale` marks all 3 remaining active series stale. + let opts3 = DecodeOptions::new().with_timestamp_ms(1_700_000_030_000); + let rec3 = decoder + .flush_target_stale(&mut tracker, &opts3) + .expect("flush_target_stale succeeds"); + assert!(tracker.is_empty()); + assert_eq!(rec3.num_metrics(), 3); + assert_eq!(rec3.num_data_points(), 3); + + let mut otap3 = to_otap_arrow_records(&rec3); + otap3 + .encode_transport_optimized() + .expect("transport optimize succeeds on flushed stale batch"); + otap3 + .decode_transport_optimized_ids() + .expect("transport decode IDs succeeds on flushed stale batch"); + let OtlpProtoMessage::Metrics(otlp3) = otap_to_otlp(&otap3) else { + panic!("expected OTLP MetricsData"); + }; + assert_eq!(otlp3.resource_metrics[0].scope_metrics[0].metrics.len(), 3); + for m in &otlp3.resource_metrics[0].scope_metrics[0].metrics { + match &m.data { + Some(metric::Data::Sum(s)) => { + assert_eq!(s.data_points[0].flags, DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert_eq!(s.data_points[0].value, None); + } + Some(metric::Data::Summary(s)) => { + assert_eq!(s.data_points[0].flags, DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert_eq!(s.data_points[0].sum, 0.0); + } + Some(metric::Data::Histogram(h)) => { + assert_eq!(h.data_points[0].flags, DATA_POINT_FLAG_NO_RECORDED_VALUE); + assert_eq!(h.data_points[0].sum, None); + } + other => panic!("unexpected metric data: {other:?}"), + } + } +} diff --git a/tests/otap-compat/tests/dhat_bounded_mem.rs b/tests/otap-compat/tests/dhat_bounded_mem.rs index b0f8501..c295284 100644 --- a/tests/otap-compat/tests/dhat_bounded_mem.rs +++ b/tests/otap-compat/tests/dhat_bounded_mem.rs @@ -19,7 +19,9 @@ use std::fs; use std::path::Path; use std::rc::Rc; -use expfmt::arrow::{ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep, OtapMetricRecords}; +use expfmt::arrow::{ + ArrowDecoder, DecodeBudget, DecodeOptions, DecodeStep, OtapMetricRecords, StalenessTracker, +}; use expfmt::format::Format; use otel_arrow_dfe_pdata::otap::{Metrics, OtapArrowRecords, OtapBatchStore}; use otel_arrow_dfe_pdata::proto::opentelemetry::arrow::v1::ArrowPayloadType; @@ -113,6 +115,7 @@ fn local_set_non_send_repeated_scrapes_keep_bounded_memory() { let stream_decoder = Rc::new(RefCell::new(ArrowDecoder::new())); let stream_opts = opts.with_max_data_points_per_batch(128); let full_237_coop = Rc::clone(&full_237); + let full_237_stale = Rc::clone(&full_237); let stream_handle = tokio::task::spawn_local(async move { let mut warmup_stream_dec_mem = 0usize; let mut warmup_stream_live_heap = 0usize; @@ -239,8 +242,9 @@ fn local_set_non_send_repeated_scrapes_keep_bounded_memory() { .await .expect("cooperative streaming local task succeeds"); - // Verify OpenMetrics 1.0 scrapes (with histograms, summaries, exemplars, and units) also - // keep retained scratch memory and live heap strictly constant after warmup. + // Verify stateless OpenMetrics 1.0 `decode` (without `StalenessTracker`) keeps retained + // scratch memory and live heap strictly constant after warmup. + let om_bench_stale = Rc::clone(&om_bench); let om_decoder = Rc::new(RefCell::new(ArrowDecoder::new())); let om_opts = opts; let om_handle = tokio::task::spawn_local(async move { @@ -251,7 +255,7 @@ fn local_set_non_send_repeated_scrapes_keep_bounded_memory() { let records = om_decoder .borrow_mut() .decode(&om_bench, &Format::OPEN_METRICS_1_0_0, &om_opts) - .expect("OpenMetrics scrape succeeds"); + .expect("OpenMetrics stateless decode succeeds"); let otap = to_otap_arrow_records(&records); let dec_mem = om_decoder.borrow().retained_memory_bytes(); drop(otap); @@ -261,14 +265,150 @@ fn local_set_non_send_repeated_scrapes_keep_bounded_memory() { if i == 1 { warmup_om_dec_mem = dec_mem; warmup_om_live_heap = live_heap; + assert!(warmup_om_dec_mem > 0); } else if i > 1 { assert_eq!(dec_mem, warmup_om_dec_mem); assert_eq!(live_heap, warmup_om_live_heap); } tokio::task::yield_now().await; } + warmup_om_dec_mem }); - om_handle.await.expect("OpenMetrics local task succeeds"); + let stateless_om_dec_mem = om_handle + .await + .expect("OpenMetrics stateless task succeeds"); + + // Verify optional `StalenessTracker` starts with zero heap allocation, keeps steady-state + // OpenMetrics scrapes strictly zero-allocation, and stays bounded across alternating + // full/subset scrapes (exercising `SparseFlagsCol`, stale series emission, and compaction) + // in both `decode_with_staleness` and `decode_stream_with_staleness`. + assert_eq!(StalenessTracker::new().retained_memory_bytes(), 0); + let stale_om_decoder = Rc::new(RefCell::new(ArrowDecoder::new())); + let stale_om_tracker = Rc::new(RefCell::new(StalenessTracker::new())); + let stale_om_opts = opts.with_track_timestamps_staleness(true); + let stale_om_handle = tokio::task::spawn_local(async move { + let mut warmup_dec_mem = 0usize; + let mut warmup_tracker_mem = 0usize; + let mut warmup_live_heap = 0usize; + + for i in 0..16 { + let records = stale_om_decoder + .borrow_mut() + .decode_with_staleness( + &om_bench_stale, + &Format::OPEN_METRICS_1_0_0, + &stale_om_opts, + &mut stale_om_tracker.borrow_mut(), + ) + .expect("OpenMetrics tracked decode succeeds"); + let otap = to_otap_arrow_records(&records); + let dec_mem = stale_om_decoder.borrow().retained_memory_bytes(); + let tracker_mem = stale_om_tracker.borrow().retained_memory_bytes(); + drop(otap); + drop(records); + + let live_heap = dhat::HeapStats::get().curr_bytes; + if i == 1 { + warmup_dec_mem = dec_mem; + warmup_tracker_mem = tracker_mem; + warmup_live_heap = live_heap; + assert!(warmup_tracker_mem > 0); + assert_eq!(warmup_dec_mem, stateless_om_dec_mem); + } else if i > 1 { + assert_eq!(dec_mem, warmup_dec_mem); + assert_eq!(tracker_mem, warmup_tracker_mem); + assert_eq!(live_heap, warmup_live_heap); + } + tokio::task::yield_now().await; + } + }); + stale_om_handle + .await + .expect("OpenMetrics tracked local task succeeds"); + + let churn_decoder = Rc::new(RefCell::new(ArrowDecoder::new())); + let churn_stream_decoder = Rc::new(RefCell::new(ArrowDecoder::new())); + let churn_tracker = Rc::new(RefCell::new(StalenessTracker::new())); + let churn_stream_tracker = Rc::new(RefCell::new(StalenessTracker::new())); + let subset_prom: &[u8] = b"# TYPE go_gc_gogc_percent gauge\ngo_gc_gogc_percent 75\n"; + let churn_handle = tokio::task::spawn_local(async move { + let mut max_dec_mem = 0usize; + let mut max_stream_dec_mem = 0usize; + let mut max_tracker_mem = 0usize; + let mut max_live_heap = 0usize; + + for cycle in 0..16 { + let step_opts = opts.with_timestamp_ms(1_700_000_000_000 + (cycle * 1_000)); + let step_stream_opts = step_opts.with_max_data_points_per_batch(128); + let payload: &[u8] = if cycle % 2 == 0 { + &full_237_stale + } else { + subset_prom + }; + + let records = churn_decoder + .borrow_mut() + .decode_with_staleness( + payload, + &Format::TEXT, + &step_opts, + &mut churn_tracker.borrow_mut(), + ) + .expect("alternating decode_with_staleness succeeds"); + let otap = to_otap_arrow_records(&records); + drop(otap); + drop(records); + + { + let mut dec = churn_stream_decoder.borrow_mut(); + let mut tr = churn_stream_tracker.borrow_mut(); + for chunk_res in dec.decode_stream_with_staleness( + payload, + &Format::TEXT, + &step_stream_opts, + &mut tr, + ) { + let chunk = chunk_res.expect("alternating stream chunk succeeds"); + let otap = to_otap_arrow_records(&chunk); + drop(otap); + drop(chunk); + } + } + + let dec_mem = churn_decoder.borrow().retained_memory_bytes(); + let stream_dec_mem = churn_stream_decoder.borrow().retained_memory_bytes(); + let tracker_mem = churn_tracker.borrow().retained_memory_bytes(); + let live_heap = dhat::HeapStats::get().curr_bytes; + + if cycle == 2 { + max_dec_mem = dec_mem; + max_stream_dec_mem = stream_dec_mem; + max_tracker_mem = tracker_mem; + max_live_heap = live_heap; + } else if cycle > 2 && cycle % 2 == 0 { + assert_eq!( + dec_mem, max_dec_mem, + "alternating staleness decoder scratch grew at cycle {cycle}" + ); + assert_eq!( + stream_dec_mem, max_stream_dec_mem, + "alternating streaming staleness decoder scratch grew at cycle {cycle}" + ); + assert_eq!( + tracker_mem, max_tracker_mem, + "compacted StalenessTracker capacity grew across cycles at {cycle}" + ); + assert_eq!( + live_heap, max_live_heap, + "live heap bytes grew across alternating staleness cycles at {cycle}" + ); + } + tokio::task::yield_now().await; + } + }); + churn_handle + .await + .expect("alternating staleness churn task succeeds"); }); }