@cryptotaxi247 / netdata-1 / commits / 49476ccc2

Support multi-slot ingestion and batch emission (#21893)

* Support multi-slot ingestion and batch emission - Split Aggregator trait into SlotAccumulator (per-slot) and CrossSlotContext (cross-slot), allowing dimensions to hold multiple in-flight slots via BTreeMap instead of a single active slot. This accepts out-of-order and early-arriving data points without dropping them. - Emit ready slots in chronological order using a cutoff-based readiness check (slot must have fully elapsed + 1s buffer) instead of the previous active-slot model. - Normalize DeltaSum and CumulativeSum values to per-second rates by dividing by update_every. - Detect cumulative counter wraps (negative delta with unchanged start_time) and emit 0 instead of a negative value. - Align tick loop to second boundaries and compensate for work duration/jitter. - Sanitize newlines in plugin protocol output to prevent agent parser breakage. * Drain stale slots unconditionally in emit() to fix late-arrival leak We don't want to make the ingestion path aware about the emission logic.

vkalintiris committed Mar 5, 2026 at 15:58 UTC 49476ccc2f224115fa549de8006850439bb33228
5 files changed +887 -638
src/crates/netdata-otel/otel-plugin/src/aggregation.rs
+269 -216
@@ -1,55 +1,70 @@
1 -#![allow(dead_code)]
2 -
1 //! Aggregation logic for mapping OpenTelemetry's event-based metrics to Netdata's
2 //! fixed-interval collection model.
3 //!
6 -//! Each aggregator is a state machine that:
7 -//! - Accepts data points via `ingest()`
8 -//! - Produces a value for the current slot via `finalize_slot()`
9 -//! - Provides gap-fill values when no data arrives via `gap_fill()`
10 -
11 -/// Trait for metric aggregators that map OpenTelemetry data points to Netdata values.
4 +//! The design separates per-slot accumulation from cross-slot context:
5 +//!
6 +//! - [`SlotAccumulator`]: collects data points within a single time slot.
7 +//! One instance per (dimension, slot) pair, created on demand.
8 +//! - [`CrossSlotContext`]: persistent per-dimension state that spans slot
9 +//! boundaries (e.g. the previous cumulative value for delta computation).
10 +//! Owns finalization logic that consumes a slot accumulator and produces
11 +//! the value to emit.
12 +
13 +/// Per-slot accumulation state.
14 ///
13 -/// Aggregators maintain state across collection intervals and handle the conversion
14 -/// from OpenTelemetry's event-based model to Netdata's fixed-interval model.
15 -pub trait Aggregator {
16 - /// Ingest a data point for the current pending slot.
17 - ///
18 - /// # Arguments
19 - /// * `value` - The numeric value from the data point
20 - /// * `timestamp_ns` - The `time_unix_nano` field (when the measurement became current)
21 - /// * `start_time_ns` - The `start_time_unix_nano` field (start of observation interval)
22 - fn ingest(&mut self, value: f64, timestamp_ns: u64, start_time_ns: u64);
15 +/// One instance per (dimension, slot) pair. Created on demand when the first
16 +/// data point for a slot arrives, consumed by [`CrossSlotContext::finalize`]
17 +/// at emission time.
18 +pub trait SlotAccumulator: Default + std::fmt::Debug {
19 + /// Record a data point into this slot's accumulator.
20 + fn record(&mut self, value: f64, timestamp_ns: u64, start_time_ns: u64);
21 +
22 + /// Whether any data was recorded into this accumulator.
23 + #[allow(dead_code)]
24 + fn has_data(&self) -> bool;
25 +}
26
24 - /// Finalize the current slot and return the value to emit to Netdata.
25 - ///
26 - /// This is called when the slot's grace period has expired and we need to
27 - /// produce a final value. After this call, internal per-slot accumulators
28 - /// should be reset, but cross-slot state (like previous cumulative values)
29 - /// should be preserved.
27 +/// Cross-slot context that persists across slot boundaries.
28 +///
29 +/// One instance per dimension (not per slot). Holds state needed to convert
30 +/// raw slot data into emittable values (e.g. the previous cumulative value
31 +/// for computing deltas).
32 +pub trait CrossSlotContext: Default + std::fmt::Debug {
33 + /// The slot accumulator type paired with this context.
34 + type Slot: SlotAccumulator;
35 +
36 + /// Finalize a slot accumulator into an emittable value, updating
37 + /// cross-slot state as needed.
38 ///
31 - /// Returns `None` if no value can be produced (e.g., first observation for cumulative).
32 - fn finalize_slot(&mut self) -> Option<f64>;
39 + /// Returns `None` if no value can be produced (e.g. first cumulative
40 + /// observation establishing a baseline).
41 + fn finalize(&mut self, slot: Self::Slot) -> Option<f64>;
42
34 - /// Return the value to use when no data arrived for a slot (gap filling).
35 - ///
36 - /// This is called when a slot is finalized but no data points were ingested.
43 + /// Value to emit when no data arrived for a slot (gap filling).
44 fn gap_fill(&self) -> f64;
45
39 - /// Reset all state. Called when the dimension is being re-initialized.
46 + /// Reset all cross-slot state. Called when the dimension is re-initialized.
47 + #[allow(dead_code)]
48 fn reset(&mut self);
49 }
50
43 -/// Aggregator for Gauge metrics.
51 +// ---------------------------------------------------------------------------
52 +// Gauge
53 +// ---------------------------------------------------------------------------
54 +
55 +/// Per-slot state for Gauge metrics.
56 ///
45 -/// Gauges represent instantaneous values with no defined aggregation semantics.
46 -/// When multiple values arrive within a slot, we keep the last one (by timestamp).
47 -/// Gap filling repeats the last observed value.
57 +/// Keeps the last value by timestamp within the slot.
58 #[derive(Debug, Default)]
49 -pub struct GaugeAggregator {
50 - /// The last value seen in the current slot (with its timestamp for ordering)
59 +pub struct GaugeSlot {
60 pending: Option<PendingValue>,
52 - /// The last emitted value (for gap filling)
61 +}
62 +
63 +/// Cross-slot context for Gauge metrics.
64 +///
65 +/// Tracks the last emitted value for gap-fill (repeat last known value).
66 +#[derive(Debug, Default)]
67 +pub struct GaugeContext {
68 last_emitted: Option<f64>,
69 }
70
@@ -59,19 +74,10 @@ struct PendingValue {
74 timestamp_ns: u64,
75 }
76
62 -impl GaugeAggregator {
63 - pub fn new() -> Self {
64 - Self::default()
65 - }
66 -}
67 -
68 -impl Aggregator for GaugeAggregator {
69 - fn ingest(&mut self, value: f64, timestamp_ns: u64, _start_time_ns: u64) {
70 - // Keep the value with the latest timestamp
77 +impl SlotAccumulator for GaugeSlot {
78 + fn record(&mut self, value: f64, timestamp_ns: u64, _start_time_ns: u64) {
79 match &self.pending {
72 - Some(pending) if timestamp_ns <= pending.timestamp_ns => {
73 - // Ignore older or equal timestamp
74 - }
80 + Some(pending) if timestamp_ns <= pending.timestamp_ns => {}
81 _ => {
82 self.pending = Some(PendingValue {
83 value,
@@ -81,8 +87,16 @@ impl Aggregator for GaugeAggregator {
87 }
88 }
89
84 - fn finalize_slot(&mut self) -> Option<f64> {
85 - let value = self.pending.take().map(|p| p.value);
90 + fn has_data(&self) -> bool {
91 + self.pending.is_some()
92 + }
93 +}
94 +
95 +impl CrossSlotContext for GaugeContext {
96 + type Slot = GaugeSlot;
97 +
98 + fn finalize(&mut self, slot: GaugeSlot) -> Option<f64> {
99 + let value = slot.pending.map(|p| p.value);
100 if let Some(v) = value {
101 self.last_emitted = Some(v);
102 }
@@ -94,42 +108,46 @@ impl Aggregator for GaugeAggregator {
108 }
109
110 fn reset(&mut self) {
97 - self.pending = None;
111 self.last_emitted = None;
112 }
113 }
114
102 -/// Aggregator for Sum metrics with Delta temporality.
115 +// ---------------------------------------------------------------------------
116 +// Delta Sum
117 +// ---------------------------------------------------------------------------
118 +
119 +/// Per-slot state for Delta Sum metrics.
120 ///
104 -/// Delta sums report the change since the last report. When multiple deltas
105 -/// arrive within a slot, we sum them (addition is the decomposable aggregate).
106 -/// Gap filling returns 0 (no change occurred).
121 +/// Accumulates deltas within a slot by summing them.
122 #[derive(Debug, Default)]
108 -pub struct DeltaSumAggregator {
109 - /// Accumulated delta for the current slot
123 +pub struct DeltaSumSlot {
124 accumulated: f64,
111 - /// Whether we've received any data for the current slot
125 has_data: bool,
126 }
127
115 -impl DeltaSumAggregator {
116 - pub fn new() -> Self {
117 - Self::default()
118 - }
119 -}
128 +/// Cross-slot context for Delta Sum metrics.
129 +///
130 +/// No cross-slot state needed — each slot is independent.
131 +#[derive(Debug, Default)]
132 +pub struct DeltaSumContext;
133
121 -impl Aggregator for DeltaSumAggregator {
122 - fn ingest(&mut self, value: f64, _timestamp_ns: u64, _start_time_ns: u64) {
134 +impl SlotAccumulator for DeltaSumSlot {
135 + fn record(&mut self, value: f64, _timestamp_ns: u64, _start_time_ns: u64) {
136 self.accumulated += value;
137 self.has_data = true;
138 }
139
127 - fn finalize_slot(&mut self) -> Option<f64> {
128 - if self.has_data {
129 - let value = self.accumulated;
130 - self.accumulated = 0.0;
131 - self.has_data = false;
132 - Some(value)
140 + fn has_data(&self) -> bool {
141 + self.has_data
142 + }
143 +}
144 +
145 +impl CrossSlotContext for DeltaSumContext {
146 + type Slot = DeltaSumSlot;
147 +
148 + fn finalize(&mut self, slot: DeltaSumSlot) -> Option<f64> {
149 + if slot.has_data {
150 + Some(slot.accumulated)
151 } else {
152 None
153 }
@@ -139,69 +157,58 @@ impl Aggregator for DeltaSumAggregator {
157 0.0
158 }
159
142 - fn reset(&mut self) {
143 - self.accumulated = 0.0;
144 - self.has_data = false;
145 - }
160 + #[allow(dead_code)]
161 + fn reset(&mut self) {}
162 }
163
148 -/// Aggregator for Sum metrics with Cumulative temporality.
164 +// ---------------------------------------------------------------------------
165 +// Cumulative Sum
166 +// ---------------------------------------------------------------------------
167 +
168 +/// Per-slot state for Cumulative Sum metrics.
169 ///
150 -/// Cumulative sums report the total since a fixed start time. We convert to
151 -/// deltas by tracking the previous cumulative value and computing the difference.
170 +/// Keeps the latest cumulative value by timestamp within the slot.
171 +#[derive(Debug, Default)]
172 +pub struct CumulativeSumSlot {
173 + pending: Option<CumulativePending>,
174 +}
175 +
176 +/// Cross-slot context for Cumulative Sum metrics.
177 ///
153 -/// Restart detection uses `start_time_unix_nano` - when it changes, we know
154 -/// the counter has reset and we cannot compute a meaningful delta across
155 -/// the boundary.
178 +/// Tracks the previous cumulative value across slot boundaries to compute
179 +/// deltas. Detects counter restarts via `start_time_unix_nano` changes.
180 #[derive(Debug, Default)]
157 -pub struct CumulativeSumAggregator {
158 - /// State from the previous finalized slot
181 +pub struct CumulativeSumContext {
182 + /// State from the previous finalized slot.
183 previous: Option<CumulativeState>,
160 - /// Pending data for the current slot (last value by timestamp)
161 - pending: Option<CumulativePending>,
162 - /// The last emitted delta (for gap filling)
163 - last_emitted_delta: Option<f64>,
184 }
185
186 #[derive(Debug, Clone, Copy)]
187 struct CumulativeState {
168 - /// The cumulative value at the end of the last slot
188 value: f64,
170 - /// The start_time_unix_nano from that observation
189 start_time_ns: u64,
190 }
191
192 #[derive(Debug, Clone, Copy)]
193 struct CumulativePending {
176 - /// The last cumulative value seen in this slot
194 value: f64,
178 - /// Timestamp of that observation (for keeping "last by timestamp")
195 timestamp_ns: u64,
180 - /// The start_time_unix_nano from that observation
196 start_time_ns: u64,
197 }
198
184 -impl CumulativeSumAggregator {
185 - pub fn new() -> Self {
186 - Self::default()
187 - }
188 -
189 - /// Check if a restart occurred between the previous state and the pending data.
199 +impl CumulativeSumContext {
200 fn is_restart(&self, pending: &CumulativePending) -> bool {
201 match &self.previous {
202 Some(prev) => prev.start_time_ns != pending.start_time_ns,
193 - None => false, // No previous state, so not a restart
203 + None => false,
204 }
205 }
206 }
207
198 -impl Aggregator for CumulativeSumAggregator {
199 - fn ingest(&mut self, value: f64, timestamp_ns: u64, start_time_ns: u64) {
200 - // Keep the value with the latest timestamp within the slot
208 +impl SlotAccumulator for CumulativeSumSlot {
209 + fn record(&mut self, value: f64, timestamp_ns: u64, start_time_ns: u64) {
210 match &self.pending {
202 - Some(pending) if timestamp_ns <= pending.timestamp_ns => {
203 - // Ignore older or equal timestamp
204 - }
211 + Some(pending) if timestamp_ns <= pending.timestamp_ns => {}
212 _ => {
213 self.pending = Some(CumulativePending {
214 value,
@@ -212,52 +219,48 @@ impl Aggregator for CumulativeSumAggregator {
219 }
220 }
221
215 - fn finalize_slot(&mut self) -> Option<f64> {
216 - let pending = self.pending.take()?;
222 + fn has_data(&self) -> bool {
223 + self.pending.is_some()
224 + }
225 +}
226 +
227 +impl CrossSlotContext for CumulativeSumContext {
228 + type Slot = CumulativeSumSlot;
229 +
230 + fn finalize(&mut self, slot: CumulativeSumSlot) -> Option<f64> {
231 + let pending = slot.pending?;
232
218 - let delta = if self.is_restart(&pending) {
219 - // Restart detected - we can't compute a meaningful delta
220 - // Update state to track the new sequence
233 + if self.is_restart(&pending) {
234 self.previous = Some(CumulativeState {
235 value: pending.value,
236 start_time_ns: pending.start_time_ns,
237 });
225 - // Return 0 for the restart slot (no contribution)
238 Some(0.0)
239 } else if let Some(prev) = &self.previous {
228 - // Normal case: compute delta from previous cumulative value
240 let delta = pending.value - prev.value;
241 self.previous = Some(CumulativeState {
242 value: pending.value,
243 start_time_ns: pending.start_time_ns,
244 });
234 - Some(delta)
245 + // Negative delta on a monotonic counter means a wrap or silent
246 + // restart that didn't update start_time. Treat as reset.
247 + if delta < 0.0 { Some(0.0) } else { Some(delta) }
248 } else {
236 - // First observation - establish baseline, can't compute delta yet
249 self.previous = Some(CumulativeState {
250 value: pending.value,
251 start_time_ns: pending.start_time_ns,
252 });
241 - // Return None to indicate no value for this slot
253 None
243 - };
244 -
245 - if let Some(d) = delta {
246 - self.last_emitted_delta = Some(d);
254 }
248 -
249 - delta
255 }
256
257 fn gap_fill(&self) -> f64 {
253 - // No new cumulative value means no change in delta
258 0.0
259 }
260
261 + #[allow(dead_code)]
262 fn reset(&mut self) {
263 self.previous = None;
259 - self.pending = None;
260 - self.last_emitted_delta = None;
264 }
265 }
266
@@ -270,49 +273,53 @@ mod tests {
273
274 #[test]
275 fn keeps_last_value_by_timestamp() {
273 - let mut agg = GaugeAggregator::new();
276 + let mut slot = GaugeSlot::default();
277
275 - // Ingest multiple values with different timestamps
276 - agg.ingest(10.0, 1000, 0);
277 - agg.ingest(30.0, 3000, 0); // Latest timestamp
278 - agg.ingest(20.0, 2000, 0); // Earlier timestamp, should be ignored
278 + slot.record(10.0, 1000, 0);
279 + slot.record(30.0, 3000, 0); // Latest timestamp
280 + slot.record(20.0, 2000, 0); // Earlier timestamp, should be ignored
281
280 - assert_eq!(agg.finalize_slot(), Some(30.0));
282 + let mut ctx = GaugeContext::default();
283 + assert_eq!(ctx.finalize(slot), Some(30.0));
284 }
285
286 #[test]
287 fn returns_none_when_no_data() {
285 - let mut agg = GaugeAggregator::new();
286 - assert_eq!(agg.finalize_slot(), None);
288 + let slot = GaugeSlot::default();
289 + let mut ctx = GaugeContext::default();
290 + assert_eq!(ctx.finalize(slot), None);
291 }
292
293 #[test]
294 fn gap_fill_returns_last_emitted() {
291 - let mut agg = GaugeAggregator::new();
295 + let mut slot = GaugeSlot::default();
296 + slot.record(42.0, 1000, 0);
297
293 - agg.ingest(42.0, 1000, 0);
294 - agg.finalize_slot();
298 + let mut ctx = GaugeContext::default();
299 + ctx.finalize(slot);
300
296 - // Now gap fill should return 42.0
297 - assert_eq!(agg.gap_fill(), 42.0);
301 + assert_eq!(ctx.gap_fill(), 42.0);
302 }
303
304 #[test]
305 fn gap_fill_returns_zero_when_never_emitted() {
302 - let agg = GaugeAggregator::new();
303 - assert_eq!(agg.gap_fill(), 0.0);
306 + let ctx = GaugeContext::default();
307 + assert_eq!(ctx.gap_fill(), 0.0);
308 }
309
310 #[test]
311 fn reset_clears_state() {
308 - let mut agg = GaugeAggregator::new();
309 - agg.ingest(42.0, 1000, 0);
310 - agg.finalize_slot();
312 + let mut slot = GaugeSlot::default();
313 + slot.record(42.0, 1000, 0);
314
312 - agg.reset();
315 + let mut ctx = GaugeContext::default();
316 + ctx.finalize(slot);
317
314 - assert_eq!(agg.finalize_slot(), None);
315 - assert_eq!(agg.gap_fill(), 0.0);
318 + ctx.reset();
319 +
320 + let empty = GaugeSlot::default();
321 + assert_eq!(ctx.finalize(empty), None);
322 + assert_eq!(ctx.gap_fill(), 0.0);
323 }
324 }
325
@@ -321,49 +328,56 @@ mod tests {
328
329 #[test]
330 fn sums_multiple_deltas() {
324 - let mut agg = DeltaSumAggregator::new();
331 + let mut slot = DeltaSumSlot::default();
332
326 - agg.ingest(10.0, 1000, 0);
327 - agg.ingest(20.0, 2000, 1000);
328 - agg.ingest(5.0, 3000, 2000);
333 + slot.record(10.0, 1000, 0);
334 + slot.record(20.0, 2000, 1000);
335 + slot.record(5.0, 3000, 2000);
336
330 - assert_eq!(agg.finalize_slot(), Some(35.0));
337 + let mut ctx = DeltaSumContext;
338 + assert_eq!(ctx.finalize(slot), Some(35.0));
339 }
340
341 #[test]
342 fn returns_none_when_no_data() {
335 - let mut agg = DeltaSumAggregator::new();
336 - assert_eq!(agg.finalize_slot(), None);
343 + let slot = DeltaSumSlot::default();
344 + let mut ctx = DeltaSumContext;
345 + assert_eq!(ctx.finalize(slot), None);
346 }
347
348 #[test]
349 fn gap_fill_returns_zero() {
341 - let mut agg = DeltaSumAggregator::new();
342 - agg.ingest(100.0, 1000, 0);
343 - agg.finalize_slot();
350 + let mut slot = DeltaSumSlot::default();
351 + slot.record(100.0, 1000, 0);
352 +
353 + let mut ctx = DeltaSumContext;
354 + ctx.finalize(slot);
355
345 - assert_eq!(agg.gap_fill(), 0.0);
356 + assert_eq!(ctx.gap_fill(), 0.0);
357 }
358
359 #[test]
360 fn resets_accumulator_after_finalize() {
350 - let mut agg = DeltaSumAggregator::new();
361 + let mut ctx = DeltaSumContext;
362
352 - agg.ingest(10.0, 1000, 0);
353 - assert_eq!(agg.finalize_slot(), Some(10.0));
363 + let mut slot1 = DeltaSumSlot::default();
364 + slot1.record(10.0, 1000, 0);
365 + assert_eq!(ctx.finalize(slot1), Some(10.0));
366
355 - agg.ingest(5.0, 2000, 1000);
356 - assert_eq!(agg.finalize_slot(), Some(5.0));
367 + let mut slot2 = DeltaSumSlot::default();
368 + slot2.record(5.0, 2000, 1000);
369 + assert_eq!(ctx.finalize(slot2), Some(5.0));
370 }
371
372 #[test]
373 fn handles_negative_deltas() {
361 - let mut agg = DeltaSumAggregator::new();
374 + let mut slot = DeltaSumSlot::default();
375
363 - agg.ingest(10.0, 1000, 0);
364 - agg.ingest(-3.0, 2000, 1000);
376 + slot.record(10.0, 1000, 0);
377 + slot.record(-3.0, 2000, 1000);
378
366 - assert_eq!(agg.finalize_slot(), Some(7.0));
379 + let mut ctx = DeltaSumContext;
380 + assert_eq!(ctx.finalize(slot), Some(7.0));
381 }
382 }
383
@@ -374,111 +388,150 @@ mod tests {
388
389 #[test]
390 fn first_observation_returns_none() {
377 - let mut agg = CumulativeSumAggregator::new();
378 -
379 - agg.ingest(100.0, 1000, START_TIME);
391 + let mut slot = CumulativeSumSlot::default();
392 + slot.record(100.0, 1000, START_TIME);
393
381 - // First observation establishes baseline, no delta yet
382 - assert_eq!(agg.finalize_slot(), None);
394 + let mut ctx = CumulativeSumContext::default();
395 + assert_eq!(ctx.finalize(slot), None);
396 }
397
398 #[test]
399 fn computes_delta_from_previous() {
387 - let mut agg = CumulativeSumAggregator::new();
400 + let mut ctx = CumulativeSumContext::default();
401
402 // First slot: establish baseline
390 - agg.ingest(100.0, 1000, START_TIME);
391 - agg.finalize_slot();
403 + let mut slot1 = CumulativeSumSlot::default();
404 + slot1.record(100.0, 1000, START_TIME);
405 + ctx.finalize(slot1);
406
407 // Second slot: should compute delta
394 - agg.ingest(150.0, 2000, START_TIME);
395 - assert_eq!(agg.finalize_slot(), Some(50.0));
408 + let mut slot2 = CumulativeSumSlot::default();
409 + slot2.record(150.0, 2000, START_TIME);
410 + assert_eq!(ctx.finalize(slot2), Some(50.0));
411
412 // Third slot: another delta
398 - agg.ingest(160.0, 3000, START_TIME);
399 - assert_eq!(agg.finalize_slot(), Some(10.0));
413 + let mut slot3 = CumulativeSumSlot::default();
414 + slot3.record(160.0, 3000, START_TIME);
415 + assert_eq!(ctx.finalize(slot3), Some(10.0));
416 }
417
418 #[test]
419 fn detects_restart_via_start_time_change() {
404 - let mut agg = CumulativeSumAggregator::new();
420 + let mut ctx = CumulativeSumContext::default();
421
422 // Establish baseline
407 - agg.ingest(100.0, 1000, START_TIME);
408 - agg.finalize_slot();
423 + let mut slot1 = CumulativeSumSlot::default();
424 + slot1.record(100.0, 1000, START_TIME);
425 + ctx.finalize(slot1);
426
410 - agg.ingest(150.0, 2000, START_TIME);
411 - agg.finalize_slot();
427 + let mut slot2 = CumulativeSumSlot::default();
428 + slot2.record(150.0, 2000, START_TIME);
429 + ctx.finalize(slot2);
430
431 // Restart: start_time changes, value resets
432 let new_start_time = START_TIME + 1_000_000;
415 - agg.ingest(20.0, 3000, new_start_time);
416 -
417 - // Should return 0 for restart slot
418 - assert_eq!(agg.finalize_slot(), Some(0.0));
433 + let mut slot3 = CumulativeSumSlot::default();
434 + slot3.record(20.0, 3000, new_start_time);
435 + assert_eq!(ctx.finalize(slot3), Some(0.0));
436
437 // Next slot should compute delta from new baseline
421 - agg.ingest(30.0, 4000, new_start_time);
422 - assert_eq!(agg.finalize_slot(), Some(10.0));
438 + let mut slot4 = CumulativeSumSlot::default();
439 + slot4.record(30.0, 4000, new_start_time);
440 + assert_eq!(ctx.finalize(slot4), Some(10.0));
441 }
442
443 #[test]
444 fn keeps_last_value_by_timestamp_in_slot() {
427 - let mut agg = CumulativeSumAggregator::new();
445 + let mut ctx = CumulativeSumContext::default();
446
447 // Establish baseline
430 - agg.ingest(100.0, 1000, START_TIME);
431 - agg.finalize_slot();
448 + let mut slot1 = CumulativeSumSlot::default();
449 + slot1.record(100.0, 1000, START_TIME);
450 + ctx.finalize(slot1);
451
452 // Multiple values in one slot - should use latest by timestamp
434 - agg.ingest(150.0, 2000, START_TIME);
435 - agg.ingest(200.0, 4000, START_TIME); // Latest timestamp
436 - agg.ingest(175.0, 3000, START_TIME); // Earlier, should be ignored
453 + let mut slot2 = CumulativeSumSlot::default();
454 + slot2.record(150.0, 2000, START_TIME);
455 + slot2.record(200.0, 4000, START_TIME); // Latest timestamp
456 + slot2.record(175.0, 3000, START_TIME); // Earlier, should be ignored
457
438 - assert_eq!(agg.finalize_slot(), Some(100.0)); // 200 - 100
458 + assert_eq!(ctx.finalize(slot2), Some(100.0)); // 200 - 100
459 }
460
461 #[test]
462 fn gap_fill_returns_zero() {
443 - let mut agg = CumulativeSumAggregator::new();
463 + let mut ctx = CumulativeSumContext::default();
464
445 - agg.ingest(100.0, 1000, START_TIME);
446 - agg.finalize_slot();
465 + let mut slot1 = CumulativeSumSlot::default();
466 + slot1.record(100.0, 1000, START_TIME);
467 + ctx.finalize(slot1);
468
448 - agg.ingest(150.0, 2000, START_TIME);
449 - agg.finalize_slot();
469 + let mut slot2 = CumulativeSumSlot::default();
470 + slot2.record(150.0, 2000, START_TIME);
471 + ctx.finalize(slot2);
472
451 - // No change in cumulative value = no delta
452 - assert_eq!(agg.gap_fill(), 0.0);
473 + assert_eq!(ctx.gap_fill(), 0.0);
474 }
475
476 #[test]
477 fn returns_none_when_no_data_in_slot() {
457 - let mut agg = CumulativeSumAggregator::new();
458 - assert_eq!(agg.finalize_slot(), None);
478 + let mut ctx = CumulativeSumContext::default();
479 +
480 + let empty = CumulativeSumSlot::default();
481 + assert_eq!(ctx.finalize(empty), None);
482
483 // Even after establishing baseline, empty slot returns None
461 - agg.ingest(100.0, 1000, START_TIME);
462 - agg.finalize_slot();
484 + let mut slot1 = CumulativeSumSlot::default();
485 + slot1.record(100.0, 1000, START_TIME);
486 + ctx.finalize(slot1);
487 +
488 + let empty2 = CumulativeSumSlot::default();
489 + assert_eq!(ctx.finalize(empty2), None);
490 + }
491
464 - assert_eq!(agg.finalize_slot(), None);
492 + #[test]
493 + fn counter_wrap_emits_zero_and_resets_baseline() {
494 + let mut ctx = CumulativeSumContext::default();
495 +
496 + // Establish baseline
497 + let mut slot1 = CumulativeSumSlot::default();
498 + slot1.record(100.0, 1000, START_TIME);
499 + ctx.finalize(slot1);
500 +
501 + // Normal increment
502 + let mut slot2 = CumulativeSumSlot::default();
503 + slot2.record(200.0, 2000, START_TIME);
504 + assert_eq!(ctx.finalize(slot2), Some(100.0));
505 +
506 + // Counter wraps: value drops but start_time unchanged
507 + let mut slot3 = CumulativeSumSlot::default();
508 + slot3.record(5.0, 3000, START_TIME);
509 + assert_eq!(ctx.finalize(slot3), Some(0.0));
510 +
511 + // Next slot computes delta from new baseline
512 + let mut slot4 = CumulativeSumSlot::default();
513 + slot4.record(15.0, 4000, START_TIME);
514 + assert_eq!(ctx.finalize(slot4), Some(10.0));
515 }
516
517 #[test]
518 fn reset_clears_all_state() {
469 - let mut agg = CumulativeSumAggregator::new();
519 + let mut ctx = CumulativeSumContext::default();
520
471 - agg.ingest(100.0, 1000, START_TIME);
472 - agg.finalize_slot();
521 + let mut slot1 = CumulativeSumSlot::default();
522 + slot1.record(100.0, 1000, START_TIME);
523 + ctx.finalize(slot1);
524
474 - agg.ingest(150.0, 2000, START_TIME);
475 - agg.finalize_slot();
525 + let mut slot2 = CumulativeSumSlot::default();
526 + slot2.record(150.0, 2000, START_TIME);
527 + ctx.finalize(slot2);
528
477 - agg.reset();
529 + ctx.reset();
530
531 // After reset, next observation is treated as first
480 - agg.ingest(50.0, 3000, START_TIME);
481 - assert_eq!(agg.finalize_slot(), None); // First observation again
532 + let mut slot3 = CumulativeSumSlot::default();
533 + slot3.record(50.0, 3000, START_TIME);
534 + assert_eq!(ctx.finalize(slot3), None);
535 }
536 }
537 }
src/crates/netdata-otel/otel-plugin/src/chart.rs
+583 -396
@@ -2,34 +2,50 @@
2 //!
3 //! A `Chart` manages dimensions and slot-based aggregation, mapping OpenTelemetry's
4 //! event-based metrics to Netdata's fixed-interval collection model.
5 +//!
6 +//! Ingestion is purely additive: data points are recorded into per-slot
7 +//! accumulators within a `BTreeMap`. Emission drains ready slots in order,
8 +//! handling finalization, gap-filling, and output.
9
6 -use std::collections::HashMap;
10 +use std::collections::{BTreeMap, BTreeSet, HashMap};
11 use std::time::{Duration, Instant};
12
13 use opentelemetry_proto::tonic::metrics::v1::AggregationTemporality;
14
15 use crate::aggregation::{
12 - Aggregator, CumulativeSumAggregator, DeltaSumAggregator, GaugeAggregator,
16 + CrossSlotContext, CumulativeSumContext, DeltaSumContext, GaugeContext, SlotAccumulator,
17 };
18 use crate::iter::MetricDataKind;
19 use crate::output::{ChartDefinition, ChartType, DimensionValue, write_data_slot};
20
17 -/// A dimension with its name, aggregator, and slot state.
18 -struct Dimension<A: Aggregator> {
19 - // The name of the dimension.
21 +/// A dimension with its cross-slot context and per-slot accumulators.
22 +///
23 +/// # Multi-Slot Architecture
24 +///
25 +/// Unlike the previous design which tracked only a single active slot, this
26 +/// implementation maintains a `BTreeMap` of slot accumulators. This enables:
27 +///
28 +/// - Out-of-order ingestion: Data points can arrive for any slot that
29 +/// hasn't been emitted yet, not just the "current" slot.
30 +/// - Batch emission: Multiple ready slots can be emitted in a single tick,
31 +/// in chronological order.
32 +/// - Late data acceptance: Data arriving late (but before emission) is
33 +/// properly accumulated rather than dropped.
34 +struct Dimension<Ctx: CrossSlotContext> {
35 + /// The name of the dimension.
36 name: String,
21 - // The aggregator that ingests values of the dimension.
22 - aggregator: A,
23 - /// Whether this dimension has received data in the current slot.
24 - has_data_in_slot: bool,
37 + /// Cross-slot context (persists across slot boundaries).
38 + context: Ctx,
39 + /// Per-slot accumulators, keyed by slot timestamp.
40 + slots: BTreeMap<u64, Ctx::Slot>,
41 }
42
27 -impl<A: Aggregator + Default> Dimension<A> {
43 +impl<Ctx: CrossSlotContext> Dimension<Ctx> {
44 fn new(name: String) -> Self {
45 Self {
46 name,
31 - aggregator: A::default(),
32 - has_data_in_slot: false,
47 + context: Ctx::default(),
48 + slots: BTreeMap::new(),
49 }
50 }
51 }
@@ -49,14 +65,14 @@ impl Default for ChartConfig {
65 fn default() -> Self {
66 Self {
67 collection_interval: 10,
52 - grace_period: Duration::from_secs(60),
68 expiry_duration: Duration::from_secs(900),
69 + grace_period: Duration::from_secs(60),
70 }
71 }
72 }
73
58 -/// The type of aggregation used by a chart.
59 -#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74 +/// The aggregation type for a chart.
75 +#[derive(Debug, Clone, Copy)]
76 pub enum ChartAggregationType {
77 Gauge,
78 DeltaSum,
@@ -71,52 +87,52 @@ impl ChartAggregationType {
87 is_monotonic: Option<bool>,
88 ) -> Option<Self> {
89 match data_kind {
74 - MetricDataKind::Gauge => Some(ChartAggregationType::Gauge),
75 - MetricDataKind::Sum => match temporality {
76 - Some(AggregationTemporality::Delta) => Some(ChartAggregationType::DeltaSum),
77 - Some(AggregationTemporality::Cumulative) => {
90 + MetricDataKind::Gauge => Some(Self::Gauge),
91 + MetricDataKind::Sum => match temporality? {
92 + AggregationTemporality::Delta => Some(Self::DeltaSum),
93 + AggregationTemporality::Cumulative => {
94 + // Non-monotonic cumulative sums behave as gauges: the
95 + // value can go up or down, so delta computation is
96 + // meaningless. Default (None) is treated as monotonic.
97 if is_monotonic == Some(false) {
79 - // Non-monotonic cumulative sum: treat as gauge (absolute value)
80 - Some(ChartAggregationType::Gauge)
98 + Some(Self::Gauge)
99 } else {
82 - // Monotonic (or unspecified) cumulative sum: compute deltas
83 - Some(ChartAggregationType::CumulativeSum)
100 + Some(Self::CumulativeSum)
101 }
102 }
86 - _ => None, // Unspecified temporality
103 + _ => None,
104 },
88 - // Histograms, ExponentialHistograms, and Summaries not supported yet
105 _ => None,
106 }
107 }
108 }
109
94 -/// Tracks whether a chart's definition has been emitted to Netdata.
110 +/// Tracks whether the chart definition has been emitted.
111 enum DefinitionState {
96 - /// No definition yet.
112 + /// No definition set yet.
113 Unset,
98 - /// Definition needs to be emitted (new chart or new dimensions added).
114 + /// Definition ready but not yet written to output.
115 Pending(ChartDefinition),
100 - /// Definition has been emitted and is up to date.
116 + /// Definition has been written to output.
117 Emitted(ChartDefinition),
118 }
119
120 impl DefinitionState {
121 fn as_ref(&self) -> Option<&ChartDefinition> {
122 match self {
107 - Self::Unset => None,
123 Self::Pending(def) | Self::Emitted(def) => Some(def),
124 + Self::Unset => None,
125 }
126 }
127
128 fn as_mut(&mut self) -> Option<&mut ChartDefinition> {
129 match self {
114 - Self::Unset => None,
130 Self::Pending(def) | Self::Emitted(def) => Some(def),
131 + Self::Unset => None,
132 }
133 }
134
119 - /// Transition `Emitted` → `Pending` (no-op for other states).
135 + /// Mark the definition as needing (re-)emission.
136 fn mark_pending(&mut self) {
137 let prev = std::mem::replace(self, Self::Unset);
138 *self = match prev {
@@ -125,7 +141,7 @@ impl DefinitionState {
141 };
142 }
143
128 - /// Transition `Pending` → `Emitted` (no-op for other states).
144 + /// Mark the definition as emitted.
145 fn mark_emitted(&mut self) {
146 let prev = std::mem::replace(self, Self::Unset);
147 *self = match prev {
@@ -147,28 +163,37 @@ pub struct Chart {
163 expiry_duration: Duration,
164 /// How long to wait for data before gap-filling on a tick with no data.
165 grace_period: Duration,
150 - /// The currently active slot timestamp (if any).
151 - active_slot: Option<u64>,
166 /// The quantized slot of the last successful emission.
167 last_emission_slot: Option<u64>,
168 /// When the chart last received data (for expiry).
169 last_ingest_instant: Option<Instant>,
156 - /// Per-dimension aggregator storage.
170 + /// Per-dimension aggregator storage with per-slot accumulators.
171 dimensions: DimensionStore,
172 /// The chart definition and its emission state.
173 definition: DefinitionState,
160 - /// Scratch buffer for finalized dimension values.
161 - dim_values: Vec<DimensionValue>,
174 +}
175 +
176 +/// Apply per-second rate normalization to a value.
177 +///
178 +/// DeltaSum and CumulativeSum metrics report totals over the collection
179 +/// interval, so we divide by `update_every` to produce a per-second rate.
180 +/// Gauge metrics pass `None` and are not normalized.
181 +fn normalize(value: Option<f64>, interval_divisor: Option<u64>) -> Option<f64> {
182 + match (value, interval_divisor) {
183 + (Some(v), Some(d)) => Some(v / d as f64),
184 + _ => value,
185 + }
186 }
187
188 /// Type-erased dimension storage for different aggregator types.
189 enum DimensionStore {
166 - Gauge(HashMap<String, Dimension<GaugeAggregator>>),
167 - DeltaSum(HashMap<String, Dimension<DeltaSumAggregator>>),
168 - CumulativeSum(HashMap<String, Dimension<CumulativeSumAggregator>>),
190 + Gauge(HashMap<String, Dimension<GaugeContext>>),
191 + DeltaSum(HashMap<String, Dimension<DeltaSumContext>>),
192 + CumulativeSum(HashMap<String, Dimension<CumulativeSumContext>>),
193 }
194
195 impl DimensionStore {
196 + #[allow(dead_code)]
197 fn len(&self) -> usize {
198 match self {
199 Self::Gauge(dims) => dims.len(),
@@ -198,12 +223,10 @@ impl Chart {
223 update_every: config.collection_interval,
224 expiry_duration: config.expiry_duration,
225 grace_period: config.grace_period,
201 - active_slot: None,
226 last_emission_slot: None,
227 last_ingest_instant: None,
228 dimensions,
229 definition: DefinitionState::Unset,
206 - dim_values: Vec::new(),
230 }
231 }
232
@@ -228,7 +251,7 @@ impl Chart {
251 (timestamp_secs / self.update_every) * self.update_every
252 }
253
231 - /// Ingest a data point into a dimension's aggregator.
254 + /// Ingest a data point into a dimension's per-slot accumulator.
255 pub fn ingest(
256 &mut self,
257 dimension_name: &str,
@@ -236,40 +259,14 @@ impl Chart {
259 timestamp_ns: u64,
260 start_time_ns: u64,
261 ) {
239 - // Update last data time.
262 self.last_ingest_instant = Some(Instant::now());
263
242 - let new_slot = self.slot_for_timestamp(timestamp_ns);
243 -
244 - // Figure out how to handle the data slot:
245 - // - active_slot is None: set it to data slot
246 - // - new_slot < active_slot: drop it
247 - // - new_slot = active_slot: update aggregator with value
248 - // - new_slot > active_slot: flush the aggregator and set active_slot = data_slot
249 -
250 - match self.active_slot {
251 - None => {
252 - self.active_slot = Some(new_slot);
253 - }
254 - Some(active_slot) if new_slot < active_slot => {
255 - // Data for a previous slot — drop it.
256 - return;
257 - }
258 - Some(active_slot) if new_slot > active_slot => {
259 - // Data for a newer slot — finalize aggregator per-slot state
260 - // so it resets properly, then advance the active slot.
261 - self.dimensions.finalize_into(&mut self.dim_values);
262 - self.active_slot = Some(new_slot);
263 - }
264 - Some(_) => {
265 - // Data for the current active slot.
266 - }
267 - }
264 + let slot = self.slot_for_timestamp(timestamp_ns);
265
269 - // Ingest into the dimension's aggregator.
266 + // Record into the correct slot's accumulator.
267 let new_dimension =
268 self.dimensions
272 - .ingest(dimension_name, value, timestamp_ns, start_time_ns);
269 + .ingest(dimension_name, value, timestamp_ns, start_time_ns, slot);
270
271 // If a new dimension was added, update the definition and mark it
272 // for re-emission.
@@ -281,17 +278,24 @@ impl Chart {
278 }
279 }
280
281 + #[allow(dead_code)]
282 pub fn len(&self) -> usize {
283 self.dimensions.len()
284 }
285
288 - /// Finalize the current slot and write output into `buf`.
286 + /// Finalize ready slots and write output into `buf`.
287 + ///
288 + /// A slot is considered ready when its end time (`slot + update_every`) is
289 + /// at least 1 second before `tick_timestamp`. This gives in-flight data
290 + /// points a 1-second window to land before the slot is finalized.
291 ///
292 /// Three emission scenarios:
291 - /// 1. **Data present**: emit gap-filled catchup slots for missed intervals, then the data slot.
292 - /// 2. **No data, within grace period**: emit nothing (wait for late data).
293 - /// 3. **No data, grace expired**: emit one gap-filled slot per tick (drain oldest first).
294 - pub fn emit(&mut self, slot_timestamp: u64, buf: &mut String) {
293 + /// 1. **Ready slots exist**: emit them in order with gap-fill catchup
294 + /// for any gaps.
295 + /// 2. **Data pending but not ready**: slot hasn't fully elapsed yet — skip.
296 + /// 3. **No data at all**: respect grace period, then gap-fill one slot
297 + /// per tick.
298 + pub fn emit(&mut self, tick_timestamp: u64, buf: &mut String) {
299 // Chart must have received data at some point.
300 let Some(last_ingest_instant) = self.last_ingest_instant else {
301 return;
@@ -302,74 +306,68 @@ impl Chart {
306 return;
307 }
308
305 - // Slot boundary self-regulation: only emit once per interval boundary.
306 - let current_slot = (slot_timestamp / self.update_every) * self.update_every;
309 + // A slot S is ready when tick_timestamp > S + update_every, i.e.,
310 + // at least 1 second has passed since the slot ended (integer seconds).
311 + // Using this as the exclusive upper bound for ready_slots means:
312 + // slot < cutoff ⟹ slot + update_every < tick_timestamp.
313 + let cutoff = tick_timestamp.saturating_sub(self.update_every);
314 +
315 + // Nothing can be ready if the cutoff hasn't advanced past the last emission.
316 if let Some(last_emission_slot) = self.last_emission_slot {
308 - if current_slot <= last_emission_slot {
317 + if cutoff <= last_emission_slot {
318 return;
319 }
320 }
321
313 - if self.dimensions.has_data() {
314 - // Data present — emit definition if needed, catchup slots, then data slot.
315 - self.emit_definition_if_needed(buf);
322 + // Collect slots that are ready: after last_emission_slot, before cutoff.
323 + let ready_slots = self.dimensions.ready_slots(self.last_emission_slot, cutoff);
324
317 - // Emit gap-filled catchup slots for any missed intervals between
318 - // last emission and current.
319 - if let Some(last) = self.last_emission_slot {
320 - let mut catchup_slot = last + self.update_every;
325 + self.emit_definition_if_needed(buf);
326
322 - while catchup_slot < current_slot {
323 - self.dimensions.gap_fill_into(&mut self.dim_values);
327 + if !ready_slots.is_empty() {
328 + for &slot in &ready_slots {
329 + // Gap-fill from last emission up to this slot.
330 + if let Some(last) = self.last_emission_slot {
331 + let mut catchup = last + self.update_every;
332 + while catchup < slot {
333 + let values = self.dimensions.gap_fill(self.update_every);
334 + write_data_slot(buf, &self.chart_name, self.update_every, catchup, &values)
335 + .expect("infallible string write");
336 + catchup += self.update_every;
337 + }
338 + }
339
325 - write_data_slot(
326 - buf,
327 - &self.chart_name,
328 - self.update_every,
329 - catchup_slot,
330 - &self.dim_values,
331 - )
340 + // Finalize this slot across all dimensions.
341 + let values = self.dimensions.finalize_slot(slot, self.update_every);
342 + write_data_slot(buf, &self.chart_name, self.update_every, slot, &values)
343 .expect("infallible string write");
344
334 - catchup_slot += self.update_every;
335 - }
345 + self.last_emission_slot = Some(slot);
346 }
337 -
338 - // Finalize and emit the data slot.
339 - self.dimensions.finalize_into(&mut self.dim_values);
340 -
341 - write_data_slot(
342 - buf,
343 - &self.chart_name,
344 - self.update_every,
345 - current_slot,
346 - &self.dim_values,
347 - )
348 - .expect("infallible string write");
349 -
350 - self.last_emission_slot = Some(current_slot);
347 + } else if self.dimensions.has_any_data() {
348 + // Data exists but isn't ready yet (slot hasn't fully elapsed).
349 + // Don't gap-fill — wait for the slot to become ready.
350 } else if last_ingest_instant.elapsed() < self.grace_period {
352 - // No data, within grace period — skip this tick.
353 - } else {
354 - // No data, grace period expired — gap-fill and emit one slot.
355 - let Some(last) = self.last_emission_slot else {
356 - return;
357 - };
351 + // No data anywhere, within grace period — skip this tick.
352 + } else if let Some(last) = self.last_emission_slot {
353 + // No data anywhere, grace period expired — gap-fill one slot.
354 + let fill_slot = last + self.update_every;
355
359 - self.emit_definition_if_needed(buf);
356 + // Don't gap-fill a slot whose end time hasn't passed the cutoff.
357 + if fill_slot < cutoff {
358 + let values = self.dimensions.gap_fill(self.update_every);
359 + write_data_slot(buf, &self.chart_name, self.update_every, fill_slot, &values)
360 + .expect("infallible string write");
361
361 - let fill_slot = last + self.update_every;
362 - self.dimensions.finalize_into(&mut self.dim_values);
363 - write_data_slot(
364 - buf,
365 - &self.chart_name,
366 - self.update_every,
367 - fill_slot,
368 - &self.dim_values,
369 - )
370 - .expect("infallible string write");
362 + self.last_emission_slot = Some(fill_slot);
363 + }
364 + }
365
372 - self.last_emission_slot = Some(fill_slot);
366 + // Unconditionally drain slots at or below the last emission point.
367 + // Late-arriving data for already-emitted slots is silently discarded
368 + // to prevent unbounded BTreeMap growth and stale `has_any_data()`.
369 + if let Some(last) = self.last_emission_slot {
370 + self.dimensions.drain_up_to(last);
371 }
372 }
373
@@ -411,10 +409,16 @@ impl Chart {
409 ) {
410 debug_assert!(matches!(self.definition, DefinitionState::Unset));
411
412 + let units = if self.is_rate_normalized() && !units.is_empty() {
413 + format!("{units}/s")
414 + } else {
415 + units.to_string()
416 + };
417 +
418 self.definition = DefinitionState::Pending(ChartDefinition {
419 chart_name: self.chart_name.clone(),
420 title: title.to_string(),
417 - units: units.to_string(),
421 + units,
422 family: metric_name.replace('.', "/"),
423 context: format!("otel.{}", metric_name),
424 chart_type: self.chart_type,
@@ -429,6 +433,14 @@ impl Chart {
433 !matches!(self.definition, DefinitionState::Unset)
434 }
435
436 + /// Whether this chart's values are divided by `update_every` to produce per-second rates.
437 + fn is_rate_normalized(&self) -> bool {
438 + matches!(
439 + self.dimensions,
440 + DimensionStore::DeltaSum(_) | DimensionStore::CumulativeSum(_)
441 + )
442 + }
443 +
444 /// Returns `true` if the chart needs its definition (re-)emitted.
445 fn needs_definition(&self) -> bool {
446 matches!(self.definition, DefinitionState::Pending(_))
@@ -447,115 +459,161 @@ impl Chart {
459 fn definition(&self) -> Option<&ChartDefinition> {
460 self.definition.as_ref()
461 }
450 -
451 - /// Access finalized dimension values (for testing).
452 - #[cfg(test)]
453 - pub(crate) fn dim_values(&self) -> &[DimensionValue] {
454 - &self.dim_values
455 - }
462 }
463
464 impl DimensionStore {
459 - /// Check whether any dimension has pending data in the current slot.
460 - fn has_data(&self) -> bool {
465 + /// Whether any dimension has pending data in any slot.
466 + fn has_any_data(&self) -> bool {
467 match self {
462 - Self::Gauge(dims) => Self::any_has_data(dims),
463 - Self::DeltaSum(dims) => Self::any_has_data(dims),
464 - Self::CumulativeSum(dims) => Self::any_has_data(dims),
468 + Self::Gauge(dims) => dims.values().any(|d| !d.slots.is_empty()),
469 + Self::DeltaSum(dims) => dims.values().any(|d| !d.slots.is_empty()),
470 + Self::CumulativeSum(dims) => dims.values().any(|d| !d.slots.is_empty()),
471 }
472 }
473
468 - fn any_has_data<A: Aggregator>(dims: &HashMap<String, Dimension<A>>) -> bool {
469 - dims.values().any(|dim| dim.has_data_in_slot)
470 - }
471 -
472 - /// Ingest a value into a dimension's aggregator, creating the dimension if needed.
474 + /// Ingest a value into a dimension's per-slot accumulator, creating the
475 + /// dimension and/or slot entry if needed.
476 + ///
477 /// Returns `true` if a new dimension was created.
474 - fn ingest(&mut self, name: &str, value: f64, timestamp_ns: u64, start_time_ns: u64) -> bool {
478 + fn ingest(
479 + &mut self,
480 + name: &str,
481 + value: f64,
482 + timestamp_ns: u64,
483 + start_time_ns: u64,
484 + slot: u64,
485 + ) -> bool {
486 match self {
476 - Self::Gauge(dims) => Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns),
487 + Self::Gauge(dims) => {
488 + Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns, slot)
489 + }
490 Self::DeltaSum(dims) => {
478 - Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns)
491 + Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns, slot)
492 }
493 Self::CumulativeSum(dims) => {
481 - Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns)
494 + Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns, slot)
495 }
496 }
497 }
498
486 - fn ingest_into<A: Aggregator + Default>(
487 - dims: &mut HashMap<String, Dimension<A>>,
499 + fn ingest_into<Ctx: CrossSlotContext>(
500 + dims: &mut HashMap<String, Dimension<Ctx>>,
501 name: &str,
502 value: f64,
503 timestamp_ns: u64,
504 start_time_ns: u64,
505 + slot: u64,
506 ) -> bool {
507 let new_dimension = !dims.contains_key(name);
508 let dim = dims
509 .entry(name.to_string())
510 .or_insert_with(|| Dimension::new(name.to_string()));
497 - dim.aggregator.ingest(value, timestamp_ns, start_time_ns);
498 - dim.has_data_in_slot = true;
511 +
512 + let acc = dim.slots.entry(slot).or_default();
513 + acc.record(value, timestamp_ns, start_time_ns);
514 +
515 new_dimension
516 }
517
502 - /// Finalize all dimensions into the provided buffer.
503 - fn finalize_into(&mut self, out: &mut Vec<DimensionValue>) {
504 - out.clear();
505 -
518 + /// Get all slot timestamps with data in the range `(after, cutoff)`,
519 + /// sorted ascending. Slots at or below `after` are skipped.
520 + fn ready_slots(&self, after: Option<u64>, cutoff: u64) -> Vec<u64> {
521 + let mut slots = BTreeSet::new();
522 match self {
507 - Self::Gauge(dims) => Self::finalize_dims(dims, out),
508 - Self::DeltaSum(dims) => Self::finalize_dims(dims, out),
509 - Self::CumulativeSum(dims) => Self::finalize_dims(dims, out),
523 + Self::Gauge(dims) => Self::collect_slots(dims, after, cutoff, &mut slots),
524 + Self::DeltaSum(dims) => Self::collect_slots(dims, after, cutoff, &mut slots),
525 + Self::CumulativeSum(dims) => Self::collect_slots(dims, after, cutoff, &mut slots),
526 }
527 + slots.into_iter().collect()
528 }
529
513 - /// Gap-fill all dimensions into the provided buffer.
514 - fn gap_fill_into(&self, out: &mut Vec<DimensionValue>) {
515 - out.clear();
530 + fn collect_slots<Ctx: CrossSlotContext>(
531 + dims: &HashMap<String, Dimension<Ctx>>,
532 + after: Option<u64>,
533 + cutoff: u64,
534 + slots: &mut BTreeSet<u64>,
535 + ) {
536 + let start = after.map_or(0, |a| a + 1);
537 + for dim in dims.values() {
538 + for (&slot, _) in dim.slots.range(start..cutoff) {
539 + slots.insert(slot);
540 + }
541 + }
542 + }
543
544 + /// Drop all slot entries with keys <= `cutoff` from all dimensions.
545 + fn drain_up_to(&mut self, cutoff: u64) {
546 match self {
518 - Self::Gauge(dims) => Self::gap_fill_dims(dims, out),
519 - Self::DeltaSum(dims) => Self::gap_fill_dims(dims, out),
520 - Self::CumulativeSum(dims) => Self::gap_fill_dims(dims, out),
547 + Self::Gauge(dims) => Self::drain_dims(dims, cutoff),
548 + Self::DeltaSum(dims) => Self::drain_dims(dims, cutoff),
549 + Self::CumulativeSum(dims) => Self::drain_dims(dims, cutoff),
550 }
551 }
552
524 - fn finalize_dims<A: Aggregator>(
525 - dims: &mut HashMap<String, Dimension<A>>,
526 - out: &mut Vec<DimensionValue>,
527 - ) {
528 - out.reserve(dims.len());
529 -
553 + fn drain_dims<Ctx: CrossSlotContext>(dims: &mut HashMap<String, Dimension<Ctx>>, cutoff: u64) {
554 for dim in dims.values_mut() {
531 - let value = if dim.has_data_in_slot {
532 - dim.aggregator.finalize_slot()
533 - } else {
534 - Some(dim.aggregator.gap_fill())
535 - };
536 -
537 - out.push(DimensionValue {
538 - name: dim.name.clone(),
539 - value,
540 - });
555 + // split_off(&K) returns all entries with keys >= K, leaving < K in the original map.
556 + // To drop slots <= cutoff and keep slots > cutoff, we must split at cutoff + 1.
557 + let kept = dim.slots.split_off(&(cutoff + 1));
558 + let _ = std::mem::replace(&mut dim.slots, kept);
559 + }
560 + }
561
542 - dim.has_data_in_slot = false;
562 + /// Finalize a specific slot across all dimensions.
563 + ///
564 + /// For each dimension, removes the slot's accumulator (if present) and
565 + /// calls `context.finalize()`, or calls `context.gap_fill()` if the
566 + /// dimension had no data for this slot.
567 + fn finalize_slot(&mut self, slot: u64, update_every: u64) -> Vec<DimensionValue> {
568 + match self {
569 + Self::Gauge(dims) => Self::finalize_slot_dims(dims, slot, None),
570 + Self::DeltaSum(dims) => Self::finalize_slot_dims(dims, slot, Some(update_every)),
571 + Self::CumulativeSum(dims) => Self::finalize_slot_dims(dims, slot, Some(update_every)),
572 }
573 }
574
546 - fn gap_fill_dims<A: Aggregator>(
547 - dims: &HashMap<String, Dimension<A>>,
548 - out: &mut Vec<DimensionValue>,
549 - ) {
550 - out.reserve(dims.len());
575 + fn finalize_slot_dims<Ctx: CrossSlotContext>(
576 + dims: &mut HashMap<String, Dimension<Ctx>>,
577 + slot: u64,
578 + interval_divisor: Option<u64>,
579 + ) -> Vec<DimensionValue> {
580 + dims.values_mut()
581 + .map(|dim| {
582 + let value = match dim.slots.remove(&slot) {
583 + Some(acc) => dim.context.finalize(acc),
584 + None => Some(dim.context.gap_fill()),
585 + };
586 +
587 + DimensionValue {
588 + name: dim.name.clone(),
589 + value: normalize(value, interval_divisor),
590 + }
591 + })
592 + .collect()
593 + }
594
552 - for dim in dims.values() {
553 - out.push(DimensionValue {
554 - name: dim.name.clone(),
555 - value: Some(dim.aggregator.gap_fill()),
556 - });
595 + /// Gap-fill all dimensions.
596 + ///
597 + /// Uses the same per-second normalization as [`finalize_slot`](Self::finalize_slot).
598 + fn gap_fill(&self, update_every: u64) -> Vec<DimensionValue> {
599 + match self {
600 + Self::Gauge(dims) => Self::gap_fill_dims(dims, None),
601 + Self::DeltaSum(dims) => Self::gap_fill_dims(dims, Some(update_every)),
602 + Self::CumulativeSum(dims) => Self::gap_fill_dims(dims, Some(update_every)),
603 }
604 }
605 +
606 + fn gap_fill_dims<Ctx: CrossSlotContext>(
607 + dims: &HashMap<String, Dimension<Ctx>>,
608 + interval_divisor: Option<u64>,
609 + ) -> Vec<DimensionValue> {
610 + dims.values()
611 + .map(|dim| DimensionValue {
612 + name: dim.name.clone(),
613 + value: normalize(Some(dim.context.gap_fill()), interval_divisor),
614 + })
615 + .collect()
616 + }
617 }
618
619 #[cfg(test)]
@@ -605,9 +663,46 @@ mod tests {
663 )
664 }
665
608 - /// Helper to find a dimension value by name in the chart's dim_values.
609 - fn find_dim<'a>(chart: &'a Chart, name: &str) -> &'a DimensionValue {
610 - chart.dim_values().iter().find(|d| d.name == name).unwrap()
666 + /// Parse a SET line into (dimension_name, Option<f64>).
667 + ///
668 + /// Format: "SET dim_name = 12345" → ("dim_name", Some(12.345))
669 + /// "SET dim_name =" → ("dim_name", None)
670 + fn parse_set(line: &str) -> (&str, Option<f64>) {
671 + let rest = line.strip_prefix("SET ").expect("not a SET line");
672 + let (name, rhs) = rest.split_once(" = ").unwrap_or_else(|| {
673 + let (name, _) = rest.split_once(" =").expect("malformed SET");
674 + (name, "")
675 + });
676 + let value = if rhs.is_empty() {
677 + None
678 + } else {
679 + Some(rhs.trim().parse::<i64>().expect("bad SET value") as f64 / 1000.0)
680 + };
681 + (name, value)
682 + }
683 +
684 + /// Extract the SET values from the last BEGIN/END block in `buf`.
685 + fn last_block_sets(buf: &str) -> Vec<(&str, Option<f64>)> {
686 + let lines: Vec<&str> = buf.lines().collect();
687 + // Find the last BEGIN line.
688 + let begin_idx = lines
689 + .iter()
690 + .rposition(|l| l.starts_with("BEGIN "))
691 + .expect("no BEGIN in buf");
692 + lines[begin_idx..]
693 + .iter()
694 + .filter(|l| l.starts_with("SET "))
695 + .map(|l| parse_set(l))
696 + .collect()
697 + }
698 +
699 + /// Get the value of a dimension from the last emitted block.
700 + fn last_dim_value(buf: &str, name: &str) -> Option<f64> {
701 + last_block_sets(buf)
702 + .into_iter()
703 + .find(|(n, _)| *n == name)
704 + .expect("dimension not found in last block")
705 + .1
706 }
707
708 /// Count how many SET lines are in the buf.
@@ -697,20 +792,21 @@ mod tests {
792 )
793 .unwrap();
794
700 - // Ingest values — should keep last by timestamp (gauge behavior).
701 - chart.ingest("dim1", 42.0, ns(1), 0);
702 - chart.ingest("dim1", 50.0, ns(3), 0); // Latest
703 - chart.ingest("dim1", 45.0, ns(2), 0);
795 + // Ingest values within the same slot — should keep last by timestamp
796 + // (gauge behavior). Use ms() to stay within slot 0.
797 + chart.ingest("dim1", 42.0, ms(100), 0);
798 + chart.ingest("dim1", 50.0, ms(300), 0); // Latest
799 + chart.ingest("dim1", 45.0, ms(200), 0);
800
801 let mut buf = String::new();
706 - chart.emit(1, &mut buf);
707 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
802 + chart.emit(2, &mut buf);
803 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
804
805 // Gap fill: repeats last value (gauge behavior, not 0).
806 buf.clear();
711 - chart.emit(2, &mut buf);
807 + chart.emit(3, &mut buf);
808 assert!(!buf.is_empty());
713 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
809 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
810 }
811
812 #[test]
@@ -734,7 +830,7 @@ mod tests {
830 fn tick_without_data_emits_nothing() {
831 let mut chart = gauge_chart();
832 let mut buf = String::new();
737 - chart.emit(1, &mut buf);
833 + chart.emit(2, &mut buf);
834 assert!(buf.is_empty());
835 }
836
@@ -742,15 +838,16 @@ mod tests {
838 fn ingest_then_tick_produces_update() {
839 let mut chart = gauge_chart();
840
745 - chart.ingest("dim1", 42.0, ns(5), 0);
841 + chart.ingest("dim1", 42.0, ms(500), 0);
842 let mut buf = String::new();
747 - chart.emit(1, &mut buf);
843 + chart.emit(2, &mut buf);
844 assert!(!buf.is_empty());
845
750 - assert_eq!(chart.dim_values().len(), 1);
751 - assert_eq!(chart.dim_values()[0].name, "dim1");
752 - assert_eq!(chart.dim_values()[0].value, Some(42.0));
753 - assert!(buf.contains("END 2\n")); // slot 1 + interval 1
846 + let sets = last_block_sets(&buf);
847 + assert_eq!(sets.len(), 1);
848 + assert_eq!(sets[0].0, "dim1");
849 + assert_eq!(sets[0].1, Some(42.0));
850 + assert!(buf.contains("END 1\n")); // slot 0 + interval 1
851 }
852
853 #[test]
@@ -770,7 +867,7 @@ mod tests {
867
868 // With zero expiry, the chart is immediately expired.
869 let mut buf = String::new();
773 - chart.emit(1, &mut buf);
870 + chart.emit(2, &mut buf);
871 assert!(buf.is_empty());
872 }
873
@@ -779,34 +876,35 @@ mod tests {
876 let mut chart = gauge_chart();
877
878 // Ingest data, then tick to finalize.
782 - chart.ingest("dim1", 42.0, ns(5), 0);
879 + chart.ingest("dim1", 42.0, ms(500), 0);
880 let mut buf = String::new();
784 - chart.emit(1, &mut buf);
881 + chart.emit(2, &mut buf);
882 assert!(!buf.is_empty());
786 - assert_eq!(chart.dim_values()[0].value, Some(42.0));
883 + assert_eq!(last_dim_value(&buf, "dim1"), Some(42.0));
884
885 // Second tick with no new data and zero grace: gap-fills by
886 // repeating the last gauge value.
887 buf.clear();
791 - chart.emit(2, &mut buf);
888 + chart.emit(3, &mut buf);
889 assert!(!buf.is_empty());
793 - assert!(buf.contains("END 3\n")); // slot 2 + interval 1
794 - assert_eq!(chart.dim_values()[0].value, Some(42.0));
890 + assert!(buf.contains("END 2\n")); // gap-fill slot 1 + interval 1
891 + assert_eq!(last_dim_value(&buf, "dim1"), Some(42.0));
892 }
893
894 #[test]
895 fn tick_sets_slot_timestamp_from_caller() {
896 let mut chart = gauge_chart();
897
801 - chart.ingest("dim1", 1.0, ns(1), 0);
898 + // Ingest in slot 0. Slot 0 ends at t=1; with 1s buffer, ready at t=2.
899 + chart.ingest("dim1", 1.0, ms(500), 0);
900 let mut buf = String::new();
803 - chart.emit(1000, &mut buf);
804 - assert!(buf.contains("END 1001\n")); // slot 1000 + interval 1
901 + chart.emit(2, &mut buf);
902 + assert!(buf.contains("END 1\n")); // slot 0 + interval 1
903
806 - // Second tick with no data → gap slot at 1001.
904 + // Next tick with no data and zero grace -> gap-fill slot 1.
905 buf.clear();
808 - chart.emit(1001, &mut buf);
809 - assert!(buf.contains("END 1002\n")); // slot 1001 + interval 1
906 + chart.emit(3, &mut buf);
907 + assert!(buf.contains("END 2\n")); // gap-fill slot 1 + interval 1
908 }
909
910 #[test]
@@ -823,15 +921,15 @@ mod tests {
921 );
922
923 // Ingest data and tick to emit.
826 - chart.ingest("dim1", 42.0, ns(5), 0);
924 + chart.ingest("dim1", 42.0, ms(500), 0);
925 let mut buf = String::new();
828 - chart.emit(1, &mut buf);
926 + chart.emit(2, &mut buf);
927 assert!(!buf.is_empty());
928
929 // Tick again with no new data — grace period is still active,
930 // so the tick should skip.
931 buf.clear();
834 - chart.emit(2, &mut buf);
932 + chart.emit(3, &mut buf);
933 assert!(buf.is_empty());
934 }
935
@@ -849,19 +947,19 @@ mod tests {
947 );
948
949 // Ingest data and tick to emit.
852 - chart.ingest("dim1", 42.0, ns(5), 0);
950 + chart.ingest("dim1", 42.0, ms(500), 0);
951 let mut buf = String::new();
854 - chart.emit(1, &mut buf);
952 + chart.emit(2, &mut buf);
953 assert!(!buf.is_empty());
856 - assert_eq!(chart.dim_values()[0].value, Some(42.0));
954 + assert_eq!(last_dim_value(&buf, "dim1"), Some(42.0));
955
956 // Tick with no new data and zero grace period — gap-fill repeats
957 // the last gauge value.
958 buf.clear();
861 - chart.emit(2, &mut buf);
959 + chart.emit(3, &mut buf);
960 assert!(!buf.is_empty());
863 - assert!(buf.contains("END 3\n")); // slot 2 + interval 1
864 - assert_eq!(chart.dim_values()[0].value, Some(42.0));
961 + assert!(buf.contains("END 2\n")); // gap-fill slot 1 + interval 1
962 + assert_eq!(last_dim_value(&buf, "dim1"), Some(42.0));
963 }
964
965 #[test]
@@ -877,39 +975,41 @@ mod tests {
975 },
976 );
977
880 - // Ingest data so the chart is active.
978 + // Ingest data in slot 0.
979 chart.ingest("dim1", 42.0, ns(5), 0);
980
883 - // Tick at t=5: slot boundary = 0, first emission.
981 + // Tick at t=11: slot 0 ends at t=10, +1s buffer -> ready.
982 let mut buf = String::new();
885 - chart.emit(5, &mut buf);
983 + chart.emit(11, &mut buf);
984 assert!(!buf.is_empty());
985
888 - // Tick at t=9: same slot boundary (0), should not emit.
889 - chart.ingest("dim1", 43.0, ns(9), 0);
986 + // Ingest data in slot 10 (ns(15) -> slot 10).
987 + chart.ingest("dim1", 43.0, ns(15), 0);
988 +
989 + // Tick at t=16: slot 10 ends at t=20, not ready yet.
990 buf.clear();
891 - chart.emit(9, &mut buf);
991 + chart.emit(16, &mut buf);
992 assert!(buf.is_empty());
993
894 - // Tick at t=10: new slot boundary (10), should emit.
895 - chart.ingest("dim1", 44.0, ns(10), 0);
994 + // Tick at t=21: slot 10 ends at t=20, +1s buffer -> ready.
995 buf.clear();
897 - chart.emit(10, &mut buf);
996 + chart.emit(21, &mut buf);
997 assert!(!buf.is_empty());
899 - assert_eq!(chart.dim_values()[0].value, Some(44.0));
998 + assert_eq!(last_dim_value(&buf, "dim1"), Some(43.0));
999 +
1000 + // Ingest data in slot 20 (ns(25) -> slot 20).
1001 + chart.ingest("dim1", 44.0, ns(25), 0);
1002
901 - // Tick at t=11: same slot boundary (10), should not emit.
902 - chart.ingest("dim1", 45.0, ns(11), 0);
1003 + // Tick at t=26: slot 20 ends at t=30, not ready yet.
1004 buf.clear();
904 - chart.emit(11, &mut buf);
1005 + chart.emit(26, &mut buf);
1006 assert!(buf.is_empty());
1007
907 - // Tick at t=20: new slot boundary (20), should emit.
908 - chart.ingest("dim1", 46.0, ns(20), 0);
1008 + // Tick at t=31: slot 20 ends at t=30, +1s buffer -> ready.
1009 buf.clear();
910 - chart.emit(20, &mut buf);
1010 + chart.emit(31, &mut buf);
1011 assert!(!buf.is_empty());
912 - assert_eq!(chart.dim_values()[0].value, Some(46.0));
1012 + assert_eq!(last_dim_value(&buf, "dim1"), Some(44.0));
1013 }
1014
1015 #[test]
@@ -928,7 +1028,7 @@ mod tests {
1028 chart.ingest("dim1", 10.0, ns(5), 0);
1029
1030 let mut buf = String::new();
931 - chart.emit(1, &mut buf);
1031 + chart.emit(2, &mut buf);
1032 assert!(buf.is_empty());
1033 }
1034
@@ -945,14 +1045,14 @@ mod tests {
1045 },
1046 );
1047
948 - chart.ingest("dim1", 10.0, ns(5), 0);
1048 + chart.ingest("dim1", 10.0, ms(500), 0);
1049 let mut buf = String::new();
950 - chart.emit(1, &mut buf);
1050 + chart.emit(2, &mut buf);
1051 assert!(!buf.is_empty());
1052
1053 // Tick again with no new data — grace period is still active.
1054 buf.clear();
955 - chart.emit(2, &mut buf);
1055 + chart.emit(3, &mut buf);
1056 assert!(buf.is_empty());
1057 }
1058
@@ -972,7 +1072,7 @@ mod tests {
1072 chart.ingest("dim1", 100.0, ns(5), 1_000_000_000);
1073
1074 let mut buf = String::new();
975 - chart.emit(1, &mut buf);
1075 + chart.emit(2, &mut buf);
1076 assert!(buf.is_empty());
1077 }
1078
@@ -989,14 +1089,14 @@ mod tests {
1089 },
1090 );
1091
992 - chart.ingest("dim1", 100.0, ns(5), 1_000_000_000);
1092 + chart.ingest("dim1", 100.0, ms(500), 1_000_000_000);
1093 let mut buf = String::new();
994 - chart.emit(1, &mut buf);
1094 + chart.emit(2, &mut buf);
1095 assert!(!buf.is_empty());
1096
1097 // Tick again with no new data — grace period is still active.
1098 buf.clear();
999 - chart.emit(2, &mut buf);
1099 + chart.emit(3, &mut buf);
1100 assert!(buf.is_empty());
1101 }
1102 }
@@ -1020,14 +1120,14 @@ mod tests {
1120 // Data at slot 0.
1121 chart.ingest("dim1", 1.0, ns(5), 0);
1122 let mut buf = String::new();
1023 - chart.emit(0, &mut buf);
1123 + chart.emit(11, &mut buf);
1124 assert!(!buf.is_empty());
1125
1126 // Data at slot 30 — should produce gap-filled catchup slots at
1127 // 10 and 20 (repeating gauge value 1.0), then data at 30.
1128 chart.ingest("dim1", 2.0, ns(30), 0);
1129 buf.clear();
1030 - chart.emit(30, &mut buf);
1130 + chart.emit(41, &mut buf);
1131 assert!(!buf.is_empty());
1132
1133 // 2 catchup slots + 1 data slot = 3 BEGIN lines.
@@ -1044,17 +1144,17 @@ mod tests {
1144 // The data slot at 30 must have the NEW value (2.0), not a
1145 // gap-fill. This verifies that catchup slots don't consume
1146 // the pending data.
1047 - assert_eq!(chart.dim_values()[0].value, Some(2.0));
1147 + assert_eq!(last_dim_value(&buf, "dim1"), Some(2.0));
1148 }
1149
1150 #[test]
1151 fn tick_drains_one_fill_per_tick() {
1152 let mut chart = gauge_chart();
1153
1054 - // Ingest and emit at slot 1.
1055 - chart.ingest("dim1", 1.0, ns(5), 0);
1154 + // Ingest in slot 0 and emit.
1155 + chart.ingest("dim1", 1.0, ms(500), 0);
1156 let mut buf = String::new();
1057 - chart.emit(1, &mut buf);
1157 + chart.emit(2, &mut buf);
1158 assert!(!buf.is_empty());
1159
1160 // Grace = ZERO, so each subsequent tick with no data emits one
@@ -1063,19 +1163,19 @@ mod tests {
1163 chart.emit(5, &mut buf);
1164 assert_eq!(count_begins(&buf), 1);
1165 assert_eq!(count_sets(&buf), 1);
1066 - assert!(buf.contains("END 3\n")); // slot 2 + interval 1
1067 - assert_eq!(chart.dim_values()[0].value, Some(1.0));
1166 + assert!(buf.contains("END 2\n")); // gap-fill slot 1 + interval 1
1167 + assert_eq!(last_dim_value(&buf, "dim1"), Some(1.0));
1168
1169 buf.clear();
1170 chart.emit(5, &mut buf);
1171 assert_eq!(count_begins(&buf), 1);
1072 - assert!(buf.contains("END 4\n")); // slot 3 + interval 1
1073 - assert_eq!(chart.dim_values()[0].value, Some(1.0));
1172 + assert!(buf.contains("END 3\n")); // gap-fill slot 2 + interval 1
1173 + assert_eq!(last_dim_value(&buf, "dim1"), Some(1.0));
1174
1175 buf.clear();
1176 chart.emit(5, &mut buf);
1077 - assert!(buf.contains("END 5\n")); // slot 4 + interval 1
1078 - assert_eq!(chart.dim_values()[0].value, Some(1.0));
1177 + assert!(buf.contains("END 4\n")); // gap-fill slot 3 + interval 1
1178 + assert_eq!(last_dim_value(&buf, "dim1"), Some(1.0));
1179 }
1180
1181 #[test]
@@ -1085,7 +1185,7 @@ mod tests {
1185 // First data ever — no catchup slots should precede it.
1186 chart.ingest("dim1", 1.0, ns(100), 0);
1187 let mut buf = String::new();
1088 - chart.emit(100, &mut buf);
1188 + chart.emit(102, &mut buf);
1189
1190 assert_eq!(count_begins(&buf), 1);
1191 assert_eq!(count_sets(&buf), 1);
@@ -1096,17 +1196,17 @@ mod tests {
1196 fn delta_sum_gap_fills_with_zero() {
1197 let mut chart = delta_sum_chart();
1198
1099 - chart.ingest("dim1", 10.0, ns(1), 0);
1199 + chart.ingest("dim1", 10.0, ms(500), 0);
1200 let mut buf = String::new();
1101 - chart.emit(1, &mut buf);
1102 - assert_eq!(chart.dim_values()[0].value, Some(10.0));
1201 + chart.emit(2, &mut buf);
1202 + assert_eq!(last_dim_value(&buf, "dim1"), Some(10.0));
1203
1204 // No new data — gap-fill emits 0 for delta sums.
1205 buf.clear();
1106 - chart.emit(2, &mut buf);
1206 + chart.emit(3, &mut buf);
1207 assert!(!buf.is_empty());
1208 assert_eq!(count_sets(&buf), 1);
1109 - assert_eq!(chart.dim_values()[0].value, Some(0.0));
1209 + assert_eq!(last_dim_value(&buf, "dim1"), Some(0.0));
1210 }
1211
1212 #[test]
@@ -1125,12 +1225,12 @@ mod tests {
1225 // Emit at slot 0 with value 42.0.
1226 chart.ingest("dim1", 42.0, ns(5), 0);
1227 let mut buf = String::new();
1128 - chart.emit(0, &mut buf);
1228 + chart.emit(11, &mut buf);
1229
1230 // Data at slot 20 — catchup at slot 10 should repeat 42.0.
1231 chart.ingest("dim1", 99.0, ns(20), 0);
1232 buf.clear();
1133 - chart.emit(20, &mut buf);
1233 + chart.emit(31, &mut buf);
1234
1235 // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1236 assert_eq!(count_begins(&buf), 2);
@@ -1152,16 +1252,16 @@ mod tests {
1252 },
1253 );
1254
1155 - // Emit at slot 0 with delta 10.
1255 + // Emit at slot 0 with delta 10; divided by update_every (10) = 1.0/s.
1256 chart.ingest("dim1", 10.0, ns(5), 0);
1257 let mut buf = String::new();
1158 - chart.emit(0, &mut buf);
1159 - assert_eq!(chart.dim_values()[0].value, Some(10.0));
1258 + chart.emit(11, &mut buf);
1259 + assert_eq!(last_dim_value(&buf, "dim1"), Some(1.0));
1260
1261 // Data at slot 20 — catchup at slot 10 should emit 0.
1262 chart.ingest("dim1", 5.0, ns(20), ns(10));
1263 buf.clear();
1164 - chart.emit(20, &mut buf);
1264 + chart.emit(31, &mut buf);
1265
1266 // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1267 assert_eq!(count_begins(&buf), 2);
@@ -1169,8 +1269,8 @@ mod tests {
1269 assert!(buf.contains("END 20\n")); // catchup slot 10 + interval 10
1270 assert!(buf.contains("END 30\n")); // data slot 20 + interval 10
1271
1172 - // Data slot has the new delta value.
1173 - assert_eq!(chart.dim_values()[0].value, Some(5.0));
1272 + // Data slot: delta 5 / update_every 10 = 0.5/s.
1273 + assert_eq!(last_dim_value(&buf, "dim1"), Some(0.5));
1274 }
1275
1276 #[test]
@@ -1191,21 +1291,21 @@ mod tests {
1291 // Slot 0: baseline (first slot returns None for cumulative sum).
1292 chart.ingest("dim1", 100.0, ns(5), START_TIME);
1293 let mut buf = String::new();
1194 - chart.emit(0, &mut buf);
1195 - assert_eq!(chart.dim_values()[0].value, None);
1294 + chart.emit(11, &mut buf);
1295 + assert_eq!(last_dim_value(&buf, "dim1"), None);
1296
1297 // Data at slot 20 — catchup at slot 10 should gap-fill with 0.
1298 chart.ingest("dim1", 150.0, ns(20), START_TIME);
1299 buf.clear();
1200 - chart.emit(20, &mut buf);
1300 + chart.emit(31, &mut buf);
1301
1302 // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1303 assert_eq!(count_begins(&buf), 2);
1304 assert!(buf.contains("END 20\n")); // catchup slot 10 + interval 10
1305 assert!(buf.contains("END 30\n")); // data slot 20 + interval 10
1306
1207 - // Data slot: delta = 150 - 100 = 50.
1208 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1307 + // Data slot: delta = 150 - 100 = 50, divided by update_every (10) = 5.0/s.
1308 + assert_eq!(last_dim_value(&buf, "dim1"), Some(5.0));
1309 }
1310 }
1311
@@ -1213,38 +1313,71 @@ mod tests {
1313 use super::*;
1314
1315 #[test]
1216 - fn drops_data_for_previous_slot() {
1316 + fn accepts_earlier_slot_if_not_yet_emitted() {
1317 let mut chart = gauge_chart();
1318
1219 - // Active slot becomes 1.
1319 + // Ingest into slot 1 first, then slot 0.
1320 chart.ingest("dim1", 50.0, ns(1), 0);
1221 -
1222 - // Data for slot 0 — should be dropped.
1321 chart.ingest("dim1", 42.0, ns(0), 0);
1322
1323 + // Both slots are in the BTreeMap. Emit drains them in order.
1324 + let mut buf = String::new();
1325 + chart.emit(3, &mut buf);
1326 +
1327 + // Slot 0 emitted first (42.0), then slot 1 (50.0).
1328 + // Last emitted block reflects the last finalized slot.
1329 + assert_eq!(count_begins(&buf), 2);
1330 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1331 + }
1332 +
1333 + #[test]
1334 + fn drops_data_for_already_emitted_slot() {
1335 + let mut chart = gauge_chart();
1336 +
1337 + // Ingest and emit slot 0.
1338 + chart.ingest("dim1", 10.0, ms(500), 0);
1339 let mut buf = String::new();
1226 - chart.emit(1, &mut buf);
1227 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1340 + chart.emit(2, &mut buf);
1341 +
1342 + // Late arrival for slot 0 — already emitted, should be dropped.
1343 + chart.ingest("dim1", 99.0, ms(600), 0);
1344 +
1345 + // Ingest into slot 1.
1346 + chart.ingest("dim1", 50.0, ns(1), 0);
1347 +
1348 + buf.clear();
1349 + chart.emit(3, &mut buf);
1350 +
1351 + // Only slot 1 emitted. The late 99.0 for slot 0 was dropped.
1352 + assert_eq!(count_begins(&buf), 1);
1353 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1354 }
1355
1356 #[test]
1231 - fn delta_sum_slot_transition_resets_accumulator() {
1357 + fn delta_sum_slot_transition_preserved_in_btreemap() {
1358 let mut chart = delta_sum_chart();
1359
1360 // Slot 0: accumulate delta=10.
1361 chart.ingest("dim1", 10.0, ns(0), 0);
1362
1237 - // Slot 1: transition resets per-slot state; accumulate delta=5.
1363 + // Slot 1: accumulate delta=5.
1364 chart.ingest("dim1", 5.0, ns(1), ns(0));
1365
1240 - // Tick should see only the slot-1 delta (10 was finalized on transition).
1366 + // Tick at 2: both slots drained in order.
1367 let mut buf = String::new();
1242 - chart.emit(1, &mut buf);
1243 - assert_eq!(chart.dim_values()[0].value, Some(5.0));
1368 + chart.emit(3, &mut buf);
1369 +
1370 + // Two slots emitted: slot 0 + slot 1.
1371 + assert_eq!(count_begins(&buf), 2);
1372 + assert!(buf.contains("END 1\n")); // slot 0 + interval 1
1373 + assert!(buf.contains("END 2\n")); // slot 1 + interval 1
1374 +
1375 + // Last emitted block reflects the last finalize (slot 1).
1376 + assert_eq!(last_dim_value(&buf, "dim1"), Some(5.0));
1377 }
1378
1379 #[test]
1247 - fn cumulative_sum_slot_transition_advances_baseline() {
1380 + fn cumulative_sum_slot_transition_preserved_in_btreemap() {
1381 let mut chart = cumulative_sum_chart();
1382
1383 const START_TIME: u64 = 1_000_000_000;
@@ -1252,14 +1385,13 @@ mod tests {
1385 // Slot 0: baseline cumulative=100.
1386 chart.ingest("dim1", 100.0, ns(0), START_TIME);
1387
1255 - // Slot 1: transition finalizes slot 0 (promoting 100 to previous),
1256 - // then ingest cumulative=150.
1388 + // Slot 1: cumulative=150.
1389 chart.ingest("dim1", 150.0, ns(1), START_TIME);
1390
1259 - // Tick: delta should be 150 - 100 = 50.
1391 + // Tick at 2: slot 0 (baseline, None) + slot 1 (delta=150-100=50).
1392 let mut buf = String::new();
1261 - chart.emit(1, &mut buf);
1262 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1393 + chart.emit(3, &mut buf);
1394 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1395 }
1396
1397 #[test]
@@ -1278,10 +1410,10 @@ mod tests {
1410 let new_start = START_TIME + 1_000_000;
1411 chart.ingest("dim1", 20.0, ns(2), new_start);
1412
1281 - // Tick: restart slot should report 0.
1413 + // Tick at 3: should report 0 for the restart slot.
1414 let mut buf = String::new();
1283 - chart.emit(2, &mut buf);
1284 - assert_eq!(chart.dim_values()[0].value, Some(0.0));
1415 + chart.emit(4, &mut buf);
1416 + assert_eq!(last_dim_value(&buf, "dim1"), Some(0.0));
1417 }
1418
1419 #[test]
@@ -1293,11 +1425,10 @@ mod tests {
1425 chart.ingest("dim1", 20.0, ns(1), 0);
1426 chart.ingest("dim1", 30.0, ns(2), 0);
1427
1296 - // Tick sees only the last slot's value (slot transitions
1297 - // finalized the earlier ones).
1428 + // Tick at 3: all three slots drained in order.
1429 let mut buf = String::new();
1299 - chart.emit(2, &mut buf);
1300 - assert_eq!(chart.dim_values()[0].value, Some(30.0));
1430 + chart.emit(4, &mut buf);
1431 + assert_eq!(last_dim_value(&buf, "dim1"), Some(30.0));
1432 }
1433
1434 #[test]
@@ -1308,15 +1439,69 @@ mod tests {
1439 chart.ingest("dim1", 10.0, ns(0), 0);
1440 chart.ingest("dim2", 20.0, ns(0), 0);
1441
1311 - // Slot 1: only dim1 — triggers slot transition which finalizes
1312 - // dim2 via gap_fill, establishing its last_emitted value.
1442 + // Slot 1: only dim1.
1443 chart.ingest("dim1", 15.0, ns(1), 0);
1444
1315 - // Tick: dim1 has slot-1 data, dim2 should gap-fill.
1445 + // Tick at 2: slot 0 and 1 drained. dim2 gap-fills for slot 1.
1446 + let mut buf = String::new();
1447 + chart.emit(3, &mut buf);
1448 + assert_eq!(last_dim_value(&buf, "dim1"), Some(15.0));
1449 + assert_eq!(last_dim_value(&buf, "dim2"), Some(20.0));
1450 + }
1451 +
1452 + #[test]
1453 + fn late_arrival_for_emitted_slot_is_dropped() {
1454 + let mut chart = delta_sum_chart();
1455 +
1456 + // Ingest and emit slot 0.
1457 + chart.ingest("dim1", 10.0, ns(0), 0);
1458 + let mut buf = String::new();
1459 + chart.emit(2, &mut buf);
1460 + assert_eq!(last_dim_value(&buf, "dim1"), Some(10.0));
1461 +
1462 + // Late arrival for slot 0 — dropped (already emitted).
1463 + chart.ingest("dim1", 3.0, ns(0), 0);
1464 + // Data for slot 1.
1465 + chart.ingest("dim1", 5.0, ns(1), ns(0));
1466 +
1467 + buf.clear();
1468 + chart.emit(3, &mut buf);
1469 +
1470 + // Only slot 1 emitted. The 3.0 was dropped because slot 0
1471 + // was already emitted.
1472 + assert_eq!(count_begins(&buf), 1);
1473 + assert_eq!(last_dim_value(&buf, "dim1"), Some(5.0));
1474 + }
1475 +
1476 + #[test]
1477 + fn delta_sum_no_data_loss_with_early_arrival() {
1478 + // When data for the next slot arrives before the tick fires,
1479 + // both slots are preserved in the BTreeMap and drained in order.
1480 + let mut chart = Chart::new(
1481 + "test",
1482 + ChartAggregationType::DeltaSum,
1483 + ChartType::Line,
1484 + ChartConfig {
1485 + collection_interval: 10,
1486 + expiry_duration: Duration::from_secs(300),
1487 + grace_period: Duration::ZERO,
1488 + },
1489 + );
1490 +
1491 + // 10 data points for slot 0 (ts=0..9s), then 1 for slot 10.
1492 + for t in 0..11 {
1493 + chart.ingest("dim1", 1.0, ns(t), 0);
1494 + }
1495 +
1496 + // emit() drains both: slot 0 (10 deltas) + slot 10 (1 delta).
1497 let mut buf = String::new();
1317 - chart.emit(1, &mut buf);
1318 - assert_eq!(find_dim(&chart, "dim1").value, Some(15.0));
1319 - assert_eq!(find_dim(&chart, "dim2").value, Some(20.0));
1498 + chart.emit(21, &mut buf);
1499 +
1500 + // Slot 0 + slot 10 = 2 BEGIN blocks.
1501 + assert_eq!(count_begins(&buf), 2);
1502 +
1503 + assert!(buf.contains("END 10\n")); // slot 0 + interval 10
1504 + assert!(buf.contains("END 20\n")); // slot 10 + interval 10
1505 }
1506 }
1507
@@ -1327,76 +1512,78 @@ mod tests {
1512 fn keeps_last_value_by_timestamp() {
1513 let mut chart = gauge_chart();
1514
1330 - chart.ingest("dim1", 10.0, ns(1), 0);
1331 - chart.ingest("dim1", 30.0, ns(3), 0); // Latest
1332 - chart.ingest("dim1", 20.0, ns(2), 0);
1515 + // Use ms() to keep all ingests within the same slot.
1516 + chart.ingest("dim1", 10.0, ms(100), 0);
1517 + chart.ingest("dim1", 30.0, ms(300), 0); // Latest
1518 + chart.ingest("dim1", 20.0, ms(200), 0);
1519
1520 let mut buf = String::new();
1335 - chart.emit(1, &mut buf);
1336 - assert_eq!(chart.dim_values()[0].value, Some(30.0));
1521 + chart.emit(2, &mut buf);
1522 + assert_eq!(last_dim_value(&buf, "dim1"), Some(30.0));
1523 }
1524
1525 #[test]
1526 fn gap_fills_missing_dimension() {
1527 let mut chart = gauge_chart();
1528
1343 - // Both dimensions get data.
1344 - chart.ingest("dim1", 10.0, ns(5), 0);
1345 - chart.ingest("dim2", 20.0, ns(5), 0);
1529 + // Both dimensions get data in slot 0.
1530 + chart.ingest("dim1", 10.0, ms(500), 0);
1531 + chart.ingest("dim2", 20.0, ms(500), 0);
1532
1347 - // Tick finalizes both.
1533 + // Tick at 1: finalizes slot 0.
1534 let mut buf = String::new();
1349 - chart.emit(1, &mut buf);
1350 - assert_eq!(find_dim(&chart, "dim1").value, Some(10.0));
1351 - assert_eq!(find_dim(&chart, "dim2").value, Some(20.0));
1535 + chart.emit(2, &mut buf);
1536 + assert_eq!(last_dim_value(&buf, "dim1"), Some(10.0));
1537 + assert_eq!(last_dim_value(&buf, "dim2"), Some(20.0));
1538
1353 - // Only dim1 gets new data.
1354 - chart.ingest("dim1", 15.0, ns(6), 0);
1539 + // Only dim1 gets new data in slot 1.
1540 + chart.ingest("dim1", 15.0, ms(1500), 0);
1541
1356 - // Tick: dim2 should be gap-filled with previous value.
1542 + // Tick at 2: dim2 should be gap-filled with previous value.
1543 buf.clear();
1358 - chart.emit(2, &mut buf);
1359 - assert_eq!(find_dim(&chart, "dim1").value, Some(15.0));
1360 - assert_eq!(find_dim(&chart, "dim2").value, Some(20.0));
1544 + chart.emit(3, &mut buf);
1545 + assert_eq!(last_dim_value(&buf, "dim1"), Some(15.0));
1546 + assert_eq!(last_dim_value(&buf, "dim2"), Some(20.0));
1547 }
1548
1549 #[test]
1550 fn out_of_order_timestamps_keeps_latest() {
1551 let mut chart = gauge_chart();
1552
1367 - chart.ingest("dim1", 20.0, ns(2), 0);
1368 - chart.ingest("dim1", 30.0, ns(3), 0); // Latest
1369 - chart.ingest("dim1", 10.0, ns(1), 0);
1553 + // Use ms() to keep all ingests within the same slot.
1554 + chart.ingest("dim1", 20.0, ms(200), 0);
1555 + chart.ingest("dim1", 30.0, ms(300), 0); // Latest
1556 + chart.ingest("dim1", 10.0, ms(100), 0);
1557
1558 let mut buf = String::new();
1372 - chart.emit(1, &mut buf);
1373 - assert_eq!(chart.dim_values()[0].value, Some(30.0));
1559 + chart.emit(2, &mut buf);
1560 + assert_eq!(last_dim_value(&buf, "dim1"), Some(30.0));
1561 }
1562
1563 #[test]
1564 fn multi_dimension_gap_fill() {
1565 let mut chart = gauge_chart();
1566
1380 - // All three dimensions get data.
1381 - chart.ingest("dim1", 100.0, ns(5), 0);
1382 - chart.ingest("dim2", 200.0, ns(5), 0);
1383 - chart.ingest("dim3", 300.0, ns(5), 0);
1567 + // All three dimensions get data in slot 0.
1568 + chart.ingest("dim1", 100.0, ms(500), 0);
1569 + chart.ingest("dim2", 200.0, ms(500), 0);
1570 + chart.ingest("dim3", 300.0, ms(500), 0);
1571
1572 let mut buf = String::new();
1386 - chart.emit(1, &mut buf);
1387 - assert_eq!(chart.dim_values().len(), 3);
1388 - assert_eq!(find_dim(&chart, "dim1").value, Some(100.0));
1389 - assert_eq!(find_dim(&chart, "dim2").value, Some(200.0));
1390 - assert_eq!(find_dim(&chart, "dim3").value, Some(300.0));
1573 + chart.emit(2, &mut buf);
1574 + assert_eq!(last_block_sets(&buf).len(), 3);
1575 + assert_eq!(last_dim_value(&buf, "dim1"), Some(100.0));
1576 + assert_eq!(last_dim_value(&buf, "dim2"), Some(200.0));
1577 + assert_eq!(last_dim_value(&buf, "dim3"), Some(300.0));
1578
1392 - // Only dim1 gets new data.
1393 - chart.ingest("dim1", 110.0, ns(6), 0);
1579 + // Only dim1 gets new data in slot 1.
1580 + chart.ingest("dim1", 110.0, ms(1500), 0);
1581
1582 buf.clear();
1396 - chart.emit(2, &mut buf);
1397 - assert_eq!(find_dim(&chart, "dim1").value, Some(110.0));
1398 - assert_eq!(find_dim(&chart, "dim2").value, Some(200.0)); // gap-fill
1399 - assert_eq!(find_dim(&chart, "dim3").value, Some(300.0)); // gap-fill
1583 + chart.emit(3, &mut buf);
1584 + assert_eq!(last_dim_value(&buf, "dim1"), Some(110.0));
1585 + assert_eq!(last_dim_value(&buf, "dim2"), Some(200.0)); // gap-fill
1586 + assert_eq!(last_dim_value(&buf, "dim3"), Some(300.0)); // gap-fill
1587 }
1588 }
1589
@@ -1413,8 +1600,8 @@ mod tests {
1600 chart.ingest("dim1", 5.0, ms(300), ms(200));
1601
1602 let mut buf = String::new();
1416 - chart.emit(1, &mut buf);
1417 - assert_eq!(chart.dim_values()[0].value, Some(35.0));
1603 + chart.emit(2, &mut buf);
1604 + assert_eq!(last_dim_value(&buf, "dim1"), Some(35.0));
1605 }
1606
1607 #[test]
@@ -1428,25 +1615,25 @@ mod tests {
1615 chart.ingest("dim1", 20.0, ms(400), ms(300));
1616
1617 let mut buf = String::new();
1431 - chart.emit(1, &mut buf);
1432 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1618 + chart.emit(2, &mut buf);
1619 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1620 }
1621
1622 #[test]
1623 fn no_data_gap_fills_with_zero() {
1624 let mut chart = delta_sum_chart();
1625
1439 - chart.ingest("dim1", 10.0, ns(1), 0);
1626 + chart.ingest("dim1", 10.0, ms(500), 0);
1627 let mut buf = String::new();
1441 - chart.emit(1, &mut buf);
1442 - assert_eq!(chart.dim_values()[0].value, Some(10.0));
1628 + chart.emit(2, &mut buf);
1629 + assert_eq!(last_dim_value(&buf, "dim1"), Some(10.0));
1630
1631 // No new data — gap-fill emits 0 for delta sums.
1632 buf.clear();
1446 - chart.emit(2, &mut buf);
1633 + chart.emit(3, &mut buf);
1634 assert!(!buf.is_empty());
1635 assert_eq!(count_sets(&buf), 1);
1449 - assert_eq!(chart.dim_values()[0].value, Some(0.0));
1636 + assert_eq!(last_dim_value(&buf, "dim1"), Some(0.0));
1637 }
1638
1639 #[test]
@@ -1468,8 +1655,8 @@ mod tests {
1655 chart.ingest("dim1", 5.0, ms(300), ms(200));
1656
1657 let mut buf = String::new();
1471 - chart.emit(1, &mut buf);
1472 - assert_eq!(chart.dim_values()[0].value, Some(12.0));
1658 + chart.emit(2, &mut buf);
1659 + assert_eq!(last_dim_value(&buf, "dim1"), Some(12.0));
1660 }
1661 }
1662
@@ -1482,50 +1669,50 @@ mod tests {
1669 fn first_slot_returns_none() {
1670 let mut chart = cumulative_sum_chart();
1671
1485 - chart.ingest("dim1", 100.0, ns(5), START_TIME);
1672 + chart.ingest("dim1", 100.0, ms(500), START_TIME);
1673 let mut buf = String::new();
1487 - chart.emit(1, &mut buf);
1488 - assert_eq!(chart.dim_values()[0].value, None);
1674 + chart.emit(2, &mut buf);
1675 + assert_eq!(last_dim_value(&buf, "dim1"), None);
1676 }
1677
1678 #[test]
1679 fn computes_deltas_across_ticks() {
1680 let mut chart = cumulative_sum_chart();
1681
1495 - // First tick: baseline, no delta.
1496 - chart.ingest("dim1", 100.0, ns(5), START_TIME);
1682 + // First tick: baseline in slot 0, no delta.
1683 + chart.ingest("dim1", 100.0, ms(500), START_TIME);
1684 let mut buf = String::new();
1498 - chart.emit(1, &mut buf);
1499 - assert_eq!(chart.dim_values()[0].value, None);
1685 + chart.emit(2, &mut buf);
1686 + assert_eq!(last_dim_value(&buf, "dim1"), None);
1687
1501 - // Second tick: delta = 150 - 100 = 50.
1502 - chart.ingest("dim1", 150.0, ns(6), START_TIME);
1688 + // Second tick: data in slot 1, delta = 150 - 100 = 50.
1689 + chart.ingest("dim1", 150.0, ns(1), START_TIME);
1690 buf.clear();
1504 - chart.emit(2, &mut buf);
1505 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1691 + chart.emit(3, &mut buf);
1692 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1693 }
1694
1695 #[test]
1696 fn detects_restart() {
1697 let mut chart = cumulative_sum_chart();
1698
1512 - // Establish baseline.
1513 - chart.ingest("dim1", 100.0, ns(5), START_TIME);
1699 + // Establish baseline in slot 0.
1700 + chart.ingest("dim1", 100.0, ms(500), START_TIME);
1701 let mut buf = String::new();
1515 - chart.emit(1, &mut buf);
1702 + chart.emit(2, &mut buf);
1703
1517 - // Normal delta.
1518 - chart.ingest("dim1", 150.0, ns(6), START_TIME);
1704 + // Normal delta in slot 1.
1705 + chart.ingest("dim1", 150.0, ns(1), START_TIME);
1706 buf.clear();
1520 - chart.emit(2, &mut buf);
1521 - assert_eq!(chart.dim_values()[0].value, Some(50.0));
1707 + chart.emit(3, &mut buf);
1708 + assert_eq!(last_dim_value(&buf, "dim1"), Some(50.0));
1709
1523 - // Restart: new start_time.
1710 + // Restart in slot 2: new start_time.
1711 let new_start = START_TIME + 1_000_000;
1525 - chart.ingest("dim1", 20.0, ns(7), new_start);
1712 + chart.ingest("dim1", 20.0, ns(2), new_start);
1713 buf.clear();
1527 - chart.emit(3, &mut buf);
1528 - assert_eq!(chart.dim_values()[0].value, Some(0.0));
1714 + chart.emit(4, &mut buf);
1715 + assert_eq!(last_dim_value(&buf, "dim1"), Some(0.0));
1716 }
1717 }
1718
@@ -1570,7 +1757,7 @@ mod tests {
1757 chart.ingest("dim1", 1.0, ns(1), 0);
1758
1759 let mut buf = String::new();
1573 - chart.emit(1, &mut buf);
1760 + chart.emit(2, &mut buf);
1761
1762 // The CHART line should include 'store_first'.
1763 let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
@@ -1594,7 +1781,7 @@ mod tests {
1781 chart.ingest("dim1", 1.0, ns(1), 0);
1782
1783 let mut buf = String::new();
1597 - chart.emit(1, &mut buf);
1784 + chart.emit(2, &mut buf);
1785
1786 let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
1787 assert!(
@@ -1617,7 +1804,7 @@ mod tests {
1804 chart.ingest("dim1", 1.0, ns(1), 0);
1805
1806 let mut buf = String::new();
1620 - chart.emit(1, &mut buf);
1807 + chart.emit(2, &mut buf);
1808
1809 let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
1810 assert!(
src/crates/netdata-otel/otel-plugin/src/lib.rs
+17 -6
@@ -111,16 +111,20 @@ async fn run_internal() -> Result<()> {
111 // 7. Tick loop for periodic metric emission
112 let writer_for_tick = Arc::clone(&writer);
113 let tick_handle = tokio::spawn(async move {
114 - let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
114 let mut buf = String::new();
115
117 - loop {
118 - interval.tick().await;
116 + // Wait until the next second boundary before starting the loop.
117 + let now = SystemTime::now()
118 + .duration_since(UNIX_EPOCH)
119 + .expect("system clock before UNIX epoch");
120 + let next_sec = std::time::Duration::from_secs(now.as_secs() + 1);
121 + tokio::time::sleep(next_sec.saturating_sub(now)).await;
122
120 - let slot_timestamp = SystemTime::now()
123 + loop {
124 + let now = SystemTime::now()
125 .duration_since(UNIX_EPOCH)
122 - .expect("system clock before UNIX epoch")
123 - .as_secs();
126 + .expect("system clock before UNIX epoch");
127 + let slot_timestamp = now.as_secs();
128
129 buf.clear();
130 {
@@ -135,6 +139,13 @@ async fn run_internal() -> Result<()> {
139 break;
140 }
141 }
142 +
143 + // Sleep until the next second boundary, compensating for work duration and jitter.
144 + let elapsed = SystemTime::now()
145 + .duration_since(UNIX_EPOCH)
146 + .expect("system clock before UNIX epoch");
147 + let next_sec = std::time::Duration::from_secs(elapsed.as_secs() + 1);
148 + tokio::time::sleep(next_sec.saturating_sub(elapsed)).await;
149 }
150 });
151
src/crates/netdata-otel/otel-plugin/src/metrics_service.rs
+1 -12
@@ -86,6 +86,7 @@ impl ChartManager {
86 });
87 }
88
89 + #[allow(dead_code)]
90 pub fn len(&self) -> usize {
91 self.charts.len()
92 }
@@ -548,18 +549,6 @@ impl NetdataMetricsService {
549 budget
550 );
551 }
551 -
552 - let mut stored_dimensions = 0;
553 - for (_, chart) in chart_manager.charts.iter() {
554 - stored_dimensions += chart.len();
555 - }
556 -
557 - tracing::trace!(
558 - "charts: {}, dimensions: {}, new charts in request: {}",
559 - chart_manager.len(),
560 - stored_dimensions,
561 - new_charts
562 - );
552 }
553 }
554
src/crates/netdata-otel/otel-plugin/src/output.rs
+17 -8
@@ -27,20 +27,29 @@ impl fmt::Display for ChartType {
27 /// it back out for display: `displayed = SET_value * 1 / DIVISOR`.
28 pub const PRECISION_DIVISOR: i64 = 1000;
29
30 -/// Wrapper that writes a string with single quotes replaced by double quotes.
30 +/// Wrapper that sanitizes a string for use inside single-quoted plugin protocol fields.
31 ///
32 /// The Netdata plugin protocol uses single quotes to delimit fields in CHART
33 -/// and CLABEL lines. If a value contains a literal single quote, it breaks
34 -/// the agent's parser. There is no escape mechanism, so we replace `'` with `"`.
33 +/// and CLABEL lines, and is line-based (`\n`-delimited). Two characters can
34 +/// break the agent's parser:
35 +///
36 +/// - `'` — breaks the quote-delimited field boundaries.
37 +/// - `\n` / `\r` — breaks line-based parsing; the agent's buffered reader
38 +/// splits on `\n`, so an embedded newline turns one command into two
39 +/// malformed lines, causing the agent to kill the plugin.
40 +///
41 +/// There is no escape mechanism, so we replace `'` with `"` and newlines
42 +/// with their escaped representation (`\n`, `\r`).
43 struct SanitizedQuote<'a>(&'a str);
44
45 impl fmt::Display for SanitizedQuote<'_> {
46 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47 for ch in self.0.chars() {
40 - if ch == '\'' {
41 - f.write_char('"')?;
42 - } else {
43 - f.write_char(ch)?;
48 + match ch {
49 + '\'' => f.write_char('"')?,
50 + '\n' => f.write_str("\\n")?,
51 + '\r' => f.write_str("\\r")?,
52 + _ => f.write_char(ch)?,
53 }
54 }
55 Ok(())
@@ -153,7 +162,7 @@ pub fn write_data_slot(
162 let scaled = (v * PRECISION_DIVISOR as f64) as i64;
163 writeln!(f, "SET {} = {}", dim.name, scaled)?;
164 }
156 - None => writeln!(f, "SET {} = U", dim.name)?,
165 + None => writeln!(f, "SET {} =", dim.name)?,
166 }
167 }
168