1
+//! Chart management for Netdata metrics.
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
+use std::collections::HashMap;
7
+use std::time::{Duration, Instant};
8
+
9
+use opentelemetry_proto::tonic::metrics::v1::AggregationTemporality;
10
+
11
+use crate::aggregation::{
12
+ Aggregator, CumulativeSumAggregator, DeltaSumAggregator, GaugeAggregator,
13
+};
14
+use crate::iter::MetricDataKind;
15
+use crate::output::{ChartDefinition, ChartType, DimensionValue, write_data_slot};
16
+
17
+/// A dimension with its name, aggregator, and slot state.
18
+struct Dimension<A: Aggregator> {
19
+ // The name of the dimension.
20
+ 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,
25
+}
26
+
27
+impl<A: Aggregator + Default> Dimension<A> {
28
+ fn new(name: String) -> Self {
29
+ Self {
30
+ name,
31
+ aggregator: A::default(),
32
+ has_data_in_slot: false,
33
+ }
34
+ }
35
+}
36
+
37
+/// Configuration for chart timing.
38
+#[derive(Debug, Clone, Copy)]
39
+pub struct ChartConfig {
40
+ /// Collection interval in seconds.
41
+ pub collection_interval: u64,
42
+ /// How long to wait for data before gap-filling on a tick with no data.
43
+ pub grace_period: Duration,
44
+ /// Duration after which a chart with no new data stops emitting.
45
+ pub expiry_duration: Duration,
46
+}
47
+
48
+impl Default for ChartConfig {
49
+ fn default() -> Self {
50
+ Self {
51
+ collection_interval: 10,
52
+ grace_period: Duration::from_secs(60),
53
+ expiry_duration: Duration::from_secs(900),
54
+ }
55
+ }
56
+}
57
+
58
+/// The type of aggregation used by a chart.
59
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60
+pub enum ChartAggregationType {
61
+ Gauge,
62
+ DeltaSum,
63
+ CumulativeSum,
64
+}
65
+
66
+impl ChartAggregationType {
67
+ /// Determine the aggregation type from metric metadata.
68
+ pub fn from_metric(
69
+ data_kind: MetricDataKind,
70
+ temporality: Option<AggregationTemporality>,
71
+ is_monotonic: Option<bool>,
72
+ ) -> Option<Self> {
73
+ 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) => {
78
+ if is_monotonic == Some(false) {
79
+ // Non-monotonic cumulative sum: treat as gauge (absolute value)
80
+ Some(ChartAggregationType::Gauge)
81
+ } else {
82
+ // Monotonic (or unspecified) cumulative sum: compute deltas
83
+ Some(ChartAggregationType::CumulativeSum)
84
+ }
85
+ }
86
+ _ => None, // Unspecified temporality
87
+ },
88
+ // Histograms, ExponentialHistograms, and Summaries not supported yet
89
+ _ => None,
90
+ }
91
+ }
92
+}
93
+
94
+/// Tracks whether a chart's definition has been emitted to Netdata.
95
+enum DefinitionState {
96
+ /// No definition yet.
97
+ Unset,
98
+ /// Definition needs to be emitted (new chart or new dimensions added).
99
+ Pending(ChartDefinition),
100
+ /// Definition has been emitted and is up to date.
101
+ Emitted(ChartDefinition),
102
+}
103
+
104
+impl DefinitionState {
105
+ fn as_ref(&self) -> Option<&ChartDefinition> {
106
+ match self {
107
+ Self::Unset => None,
108
+ Self::Pending(def) | Self::Emitted(def) => Some(def),
109
+ }
110
+ }
111
+
112
+ fn as_mut(&mut self) -> Option<&mut ChartDefinition> {
113
+ match self {
114
+ Self::Unset => None,
115
+ Self::Pending(def) | Self::Emitted(def) => Some(def),
116
+ }
117
+ }
118
+
119
+ /// Transition `Emitted` → `Pending` (no-op for other states).
120
+ fn mark_pending(&mut self) {
121
+ let prev = std::mem::replace(self, Self::Unset);
122
+ *self = match prev {
123
+ Self::Emitted(def) => Self::Pending(def),
124
+ other => other,
125
+ };
126
+ }
127
+
128
+ /// Transition `Pending` → `Emitted` (no-op for other states).
129
+ fn mark_emitted(&mut self) {
130
+ let prev = std::mem::replace(self, Self::Unset);
131
+ *self = match prev {
132
+ Self::Pending(def) => Self::Emitted(def),
133
+ other => other,
134
+ };
135
+ }
136
+}
137
+
138
+/// A Netdata chart that manages dimensions and tick-driven aggregation.
139
+pub struct Chart {
140
+ /// The chart name used in Netdata protocol commands.
141
+ chart_name: String,
142
+ /// The Netdata chart type (line, heatmap, etc.).
143
+ chart_type: ChartType,
144
+ /// Collection interval in seconds.
145
+ update_every: u64,
146
+ /// Duration after which a chart with no new data stops emitting.
147
+ expiry_duration: Duration,
148
+ /// How long to wait for data before gap-filling on a tick with no data.
149
+ grace_period: Duration,
150
+ /// The currently active slot timestamp (if any).
151
+ active_slot: Option<u64>,
152
+ /// The quantized slot of the last successful emission.
153
+ last_emission_slot: Option<u64>,
154
+ /// When the chart last received data (for expiry).
155
+ last_ingest_instant: Option<Instant>,
156
+ /// Per-dimension aggregator storage.
157
+ dimensions: DimensionStore,
158
+ /// The chart definition and its emission state.
159
+ definition: DefinitionState,
160
+ /// Scratch buffer for finalized dimension values.
161
+ dim_values: Vec<DimensionValue>,
162
+}
163
+
164
+/// Type-erased dimension storage for different aggregator types.
165
+enum DimensionStore {
166
+ Gauge(HashMap<String, Dimension<GaugeAggregator>>),
167
+ DeltaSum(HashMap<String, Dimension<DeltaSumAggregator>>),
168
+ CumulativeSum(HashMap<String, Dimension<CumulativeSumAggregator>>),
169
+}
170
+
171
+impl DimensionStore {
172
+ fn len(&self) -> usize {
173
+ match self {
174
+ Self::Gauge(dims) => dims.len(),
175
+ Self::DeltaSum(dims) => dims.len(),
176
+ Self::CumulativeSum(dims) => dims.len(),
177
+ }
178
+ }
179
+}
180
+
181
+impl Chart {
182
+ /// Create a new chart with the given name, aggregation type, and chart type.
183
+ pub fn new(
184
+ name: &str,
185
+ aggregation_type: ChartAggregationType,
186
+ chart_type: ChartType,
187
+ config: ChartConfig,
188
+ ) -> Self {
189
+ let dimensions = match aggregation_type {
190
+ ChartAggregationType::Gauge => DimensionStore::Gauge(HashMap::new()),
191
+ ChartAggregationType::DeltaSum => DimensionStore::DeltaSum(HashMap::new()),
192
+ ChartAggregationType::CumulativeSum => DimensionStore::CumulativeSum(HashMap::new()),
193
+ };
194
+
195
+ Self {
196
+ chart_name: name.to_string(),
197
+ chart_type,
198
+ update_every: config.collection_interval,
199
+ expiry_duration: config.expiry_duration,
200
+ grace_period: config.grace_period,
201
+ active_slot: None,
202
+ last_emission_slot: None,
203
+ last_ingest_instant: None,
204
+ dimensions,
205
+ definition: DefinitionState::Unset,
206
+ dim_values: Vec::new(),
207
+ }
208
+ }
209
+
210
+ /// Create a chart from metric metadata.
211
+ ///
212
+ /// Returns `None` if the metric type is not supported.
213
+ pub fn from_metric(
214
+ name: &str,
215
+ data_kind: MetricDataKind,
216
+ temporality: Option<AggregationTemporality>,
217
+ is_monotonic: Option<bool>,
218
+ config: ChartConfig,
219
+ ) -> Option<Self> {
220
+ let aggregation_type =
221
+ ChartAggregationType::from_metric(data_kind, temporality, is_monotonic)?;
222
+ Some(Self::new(name, aggregation_type, ChartType::Line, config))
223
+ }
224
+
225
+ /// Compute the slot timestamp for a given nanosecond timestamp.
226
+ fn slot_for_timestamp(&self, timestamp_ns: u64) -> u64 {
227
+ let timestamp_secs = timestamp_ns / 1_000_000_000;
228
+ (timestamp_secs / self.update_every) * self.update_every
229
+ }
230
+
231
+ /// Ingest a data point into a dimension's aggregator.
232
+ pub fn ingest(
233
+ &mut self,
234
+ dimension_name: &str,
235
+ value: f64,
236
+ timestamp_ns: u64,
237
+ start_time_ns: u64,
238
+ ) {
239
+ // Update last data time.
240
+ self.last_ingest_instant = Some(Instant::now());
241
+
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
+ }
268
+
269
+ // Ingest into the dimension's aggregator.
270
+ let new_dimension =
271
+ self.dimensions
272
+ .ingest(dimension_name, value, timestamp_ns, start_time_ns);
273
+
274
+ // If a new dimension was added, update the definition and mark it
275
+ // for re-emission.
276
+ if new_dimension {
277
+ if let Some(def) = self.definition.as_mut() {
278
+ def.dimensions.push(dimension_name.to_string());
279
+ }
280
+ self.definition.mark_pending();
281
+ }
282
+ }
283
+
284
+ pub fn len(&self) -> usize {
285
+ self.dimensions.len()
286
+ }
287
+
288
+ /// Finalize the current slot and write output into `buf`.
289
+ ///
290
+ /// 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) {
295
+ // Chart must have received data at some point.
296
+ let Some(last_ingest_instant) = self.last_ingest_instant else {
297
+ return;
298
+ };
299
+
300
+ // Check if the chart has expired (no data for too long).
301
+ if last_ingest_instant.elapsed() >= self.expiry_duration {
302
+ return;
303
+ }
304
+
305
+ // Slot boundary self-regulation: only emit once per interval boundary.
306
+ let current_slot = (slot_timestamp / self.update_every) * self.update_every;
307
+ if let Some(last_emission_slot) = self.last_emission_slot {
308
+ if current_slot <= last_emission_slot {
309
+ return;
310
+ }
311
+ }
312
+
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);
316
+
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;
321
+
322
+ while catchup_slot < current_slot {
323
+ self.dimensions.gap_fill_into(&mut self.dim_values);
324
+
325
+ write_data_slot(
326
+ buf,
327
+ &self.chart_name,
328
+ self.update_every,
329
+ catchup_slot,
330
+ &self.dim_values,
331
+ )
332
+ .expect("infallible string write");
333
+
334
+ catchup_slot += self.update_every;
335
+ }
336
+ }
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);
351
+ } 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
+ };
358
+
359
+ self.emit_definition_if_needed(buf);
360
+
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");
371
+
372
+ self.last_emission_slot = Some(fill_slot);
373
+ }
374
+ }
375
+
376
+ /// Write the chart definition into `buf` if pending, then mark as emitted.
377
+ fn emit_definition_if_needed(&mut self, buf: &mut String) {
378
+ if !self.needs_definition() {
379
+ return;
380
+ }
381
+
382
+ // Sort dimensions numerically for heatmap charts so Netdata
383
+ // renders buckets in ascending order.
384
+ if matches!(self.chart_type, ChartType::Heatmap) {
385
+ if let Some(def) = self.definition.as_mut() {
386
+ def.sort_dimensions_numerically();
387
+ }
388
+ }
389
+
390
+ use std::fmt::Write;
391
+ write!(
392
+ buf,
393
+ "{}",
394
+ self.definition.as_ref().expect("definition must be set")
395
+ )
396
+ .expect("infallible string write");
397
+
398
+ self.definition.mark_emitted();
399
+ }
400
+
401
+ /// Initialize the chart definition from metric metadata.
402
+ ///
403
+ /// Caller should check [`has_definition()`](Self::has_definition) first
404
+ /// to avoid unnecessary allocations.
405
+ pub fn init_definition(
406
+ &mut self,
407
+ metric_name: &str,
408
+ title: &str,
409
+ units: &str,
410
+ labels: Vec<(String, String)>,
411
+ ) {
412
+ debug_assert!(matches!(self.definition, DefinitionState::Unset));
413
+
414
+ self.definition = DefinitionState::Pending(ChartDefinition {
415
+ chart_name: self.chart_name.clone(),
416
+ title: title.to_string(),
417
+ units: units.to_string(),
418
+ family: metric_name.replace('.', "/"),
419
+ context: format!("otel.{}", metric_name),
420
+ chart_type: self.chart_type,
421
+ update_every: self.update_every,
422
+ labels,
423
+ dimensions: Vec::new(),
424
+ });
425
+ }
426
+
427
+ /// Whether a definition has been set.
428
+ pub fn has_definition(&self) -> bool {
429
+ !matches!(self.definition, DefinitionState::Unset)
430
+ }
431
+
432
+ /// Returns `true` if the chart needs its definition (re-)emitted.
433
+ fn needs_definition(&self) -> bool {
434
+ matches!(self.definition, DefinitionState::Pending(_))
435
+ }
436
+
437
+ /// Whether the chart has expired (no data for longer than the expiry duration).
438
+ pub fn is_expired(&self) -> bool {
439
+ match self.last_ingest_instant {
440
+ Some(instant) => instant.elapsed() >= self.expiry_duration,
441
+ None => false,
442
+ }
443
+ }
444
+
445
+ /// Get a reference to the chart definition (test only).
446
+ #[cfg(test)]
447
+ fn definition(&self) -> Option<&ChartDefinition> {
448
+ self.definition.as_ref()
449
+ }
450
+
451
+ /// Access finalized dimension values (for testing).
452
+ #[cfg(test)]
453
+ pub(crate) fn dim_values(&self) -> &[DimensionValue] {
454
+ &self.dim_values
455
+ }
456
+}
457
+
458
+impl DimensionStore {
459
+ /// Check whether any dimension has pending data in the current slot.
460
+ fn has_data(&self) -> bool {
461
+ 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),
465
+ }
466
+ }
467
+
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.
473
+ /// 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 {
475
+ match self {
476
+ Self::Gauge(dims) => Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns),
477
+ Self::DeltaSum(dims) => {
478
+ Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns)
479
+ }
480
+ Self::CumulativeSum(dims) => {
481
+ Self::ingest_into(dims, name, value, timestamp_ns, start_time_ns)
482
+ }
483
+ }
484
+ }
485
+
486
+ fn ingest_into<A: Aggregator + Default>(
487
+ dims: &mut HashMap<String, Dimension<A>>,
488
+ name: &str,
489
+ value: f64,
490
+ timestamp_ns: u64,
491
+ start_time_ns: u64,
492
+ ) -> bool {
493
+ let new_dimension = !dims.contains_key(name);
494
+ let dim = dims
495
+ .entry(name.to_string())
496
+ .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;
499
+ new_dimension
500
+ }
501
+
502
+ /// Finalize all dimensions into the provided buffer.
503
+ fn finalize_into(&mut self, out: &mut Vec<DimensionValue>) {
504
+ out.clear();
505
+
506
+ 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),
510
+ }
511
+ }
512
+
513
+ /// Gap-fill all dimensions into the provided buffer.
514
+ fn gap_fill_into(&self, out: &mut Vec<DimensionValue>) {
515
+ out.clear();
516
+
517
+ 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),
521
+ }
522
+ }
523
+
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
+
530
+ 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
+ });
541
+
542
+ dim.has_data_in_slot = false;
543
+ }
544
+ }
545
+
546
+ fn gap_fill_dims<A: Aggregator>(
547
+ dims: &HashMap<String, Dimension<A>>,
548
+ out: &mut Vec<DimensionValue>,
549
+ ) {
550
+ out.reserve(dims.len());
551
+
552
+ for dim in dims.values() {
553
+ out.push(DimensionValue {
554
+ name: dim.name.clone(),
555
+ value: Some(dim.aggregator.gap_fill()),
556
+ });
557
+ }
558
+ }
559
+}
560
+
561
+#[cfg(test)]
562
+mod tests {
563
+ use super::*;
564
+
565
+ fn ns(secs: u64) -> u64 {
566
+ secs * 1_000_000_000
567
+ }
568
+
569
+ fn ms(millis: u64) -> u64 {
570
+ millis * 1_000_000
571
+ }
572
+
573
+ fn test_config() -> ChartConfig {
574
+ ChartConfig {
575
+ collection_interval: 1,
576
+ expiry_duration: Duration::from_secs(300),
577
+ grace_period: Duration::ZERO,
578
+ }
579
+ }
580
+
581
+ fn gauge_chart() -> Chart {
582
+ Chart::new(
583
+ "test",
584
+ ChartAggregationType::Gauge,
585
+ ChartType::Line,
586
+ test_config(),
587
+ )
588
+ }
589
+
590
+ fn delta_sum_chart() -> Chart {
591
+ Chart::new(
592
+ "test",
593
+ ChartAggregationType::DeltaSum,
594
+ ChartType::Line,
595
+ test_config(),
596
+ )
597
+ }
598
+
599
+ fn cumulative_sum_chart() -> Chart {
600
+ Chart::new(
601
+ "test",
602
+ ChartAggregationType::CumulativeSum,
603
+ ChartType::Line,
604
+ test_config(),
605
+ )
606
+ }
607
+
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()
611
+ }
612
+
613
+ /// Count how many SET lines are in the buf.
614
+ fn count_sets(buf: &str) -> usize {
615
+ buf.lines().filter(|line| line.starts_with("SET ")).count()
616
+ }
617
+
618
+ /// Count how many BEGIN lines are in the buf.
619
+ fn count_begins(buf: &str) -> usize {
620
+ buf.lines()
621
+ .filter(|line| line.starts_with("BEGIN "))
622
+ .count()
623
+ }
624
+
625
+ mod chart_creation {
626
+ use super::*;
627
+
628
+ #[test]
629
+ fn creates_gauge_chart() {
630
+ let chart = Chart::from_metric(
631
+ "test",
632
+ MetricDataKind::Gauge,
633
+ None,
634
+ None,
635
+ ChartConfig::default(),
636
+ );
637
+ assert!(chart.is_some());
638
+ }
639
+
640
+ #[test]
641
+ fn creates_delta_sum_chart() {
642
+ let chart = Chart::from_metric(
643
+ "test",
644
+ MetricDataKind::Sum,
645
+ Some(AggregationTemporality::Delta),
646
+ None,
647
+ ChartConfig::default(),
648
+ );
649
+ assert!(chart.is_some());
650
+ }
651
+
652
+ #[test]
653
+ fn creates_cumulative_sum_chart() {
654
+ let chart = Chart::from_metric(
655
+ "test",
656
+ MetricDataKind::Sum,
657
+ Some(AggregationTemporality::Cumulative),
658
+ None,
659
+ ChartConfig::default(),
660
+ );
661
+ assert!(chart.is_some());
662
+ }
663
+
664
+ #[test]
665
+ fn rejects_unsupported_types() {
666
+ let chart = Chart::from_metric(
667
+ "test",
668
+ MetricDataKind::Histogram,
669
+ None,
670
+ None,
671
+ ChartConfig::default(),
672
+ );
673
+ assert!(chart.is_none());
674
+ }
675
+
676
+ #[test]
677
+ fn non_monotonic_cumulative_sum_uses_gauge_aggregation() {
678
+ let chart = Chart::from_metric(
679
+ "test",
680
+ MetricDataKind::Sum,
681
+ Some(AggregationTemporality::Cumulative),
682
+ Some(false),
683
+ ChartConfig::default(),
684
+ );
685
+ let chart = chart.expect("should create chart for non-monotonic cumulative sum");
686
+ assert!(matches!(chart.dimensions, DimensionStore::Gauge(_)));
687
+ }
688
+
689
+ #[test]
690
+ fn non_monotonic_cumulative_sum_behaves_as_gauge() {
691
+ let mut chart = Chart::from_metric(
692
+ "test",
693
+ MetricDataKind::Sum,
694
+ Some(AggregationTemporality::Cumulative),
695
+ Some(false),
696
+ test_config(),
697
+ )
698
+ .unwrap();
699
+
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);
704
+
705
+ let mut buf = String::new();
706
+ chart.emit(1, &mut buf);
707
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
708
+
709
+ // Gap fill: repeats last value (gauge behavior, not 0).
710
+ buf.clear();
711
+ chart.emit(2, &mut buf);
712
+ assert!(!buf.is_empty());
713
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
714
+ }
715
+
716
+ #[test]
717
+ fn monotonic_cumulative_sum_still_computes_deltas() {
718
+ let chart = Chart::from_metric(
719
+ "test",
720
+ MetricDataKind::Sum,
721
+ Some(AggregationTemporality::Cumulative),
722
+ Some(true),
723
+ ChartConfig::default(),
724
+ );
725
+ let chart = chart.expect("should create chart for monotonic cumulative sum");
726
+ assert!(matches!(chart.dimensions, DimensionStore::CumulativeSum(_)));
727
+ }
728
+ }
729
+
730
+ mod tick_driven_emission {
731
+ use super::*;
732
+
733
+ #[test]
734
+ fn tick_without_data_emits_nothing() {
735
+ let mut chart = gauge_chart();
736
+ let mut buf = String::new();
737
+ chart.emit(1, &mut buf);
738
+ assert!(buf.is_empty());
739
+ }
740
+
741
+ #[test]
742
+ fn ingest_then_tick_produces_update() {
743
+ let mut chart = gauge_chart();
744
+
745
+ chart.ingest("dim1", 42.0, ns(5), 0);
746
+ let mut buf = String::new();
747
+ chart.emit(1, &mut buf);
748
+ assert!(!buf.is_empty());
749
+
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
754
+ }
755
+
756
+ #[test]
757
+ fn tick_after_expiry_emits_nothing() {
758
+ let mut chart = Chart::new(
759
+ "test",
760
+ ChartAggregationType::Gauge,
761
+ ChartType::Line,
762
+ ChartConfig {
763
+ collection_interval: 1,
764
+ expiry_duration: Duration::ZERO,
765
+ grace_period: Duration::ZERO,
766
+ },
767
+ );
768
+
769
+ chart.ingest("dim1", 42.0, ns(5), 0);
770
+
771
+ // With zero expiry, the chart is immediately expired.
772
+ let mut buf = String::new();
773
+ chart.emit(1, &mut buf);
774
+ assert!(buf.is_empty());
775
+ }
776
+
777
+ #[test]
778
+ fn consecutive_ticks_gap_fill() {
779
+ let mut chart = gauge_chart();
780
+
781
+ // Ingest data, then tick to finalize.
782
+ chart.ingest("dim1", 42.0, ns(5), 0);
783
+ let mut buf = String::new();
784
+ chart.emit(1, &mut buf);
785
+ assert!(!buf.is_empty());
786
+ assert_eq!(chart.dim_values()[0].value, Some(42.0));
787
+
788
+ // Second tick with no new data and zero grace: gap-fills by
789
+ // repeating the last gauge value.
790
+ buf.clear();
791
+ chart.emit(2, &mut buf);
792
+ 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));
795
+ }
796
+
797
+ #[test]
798
+ fn tick_sets_slot_timestamp_from_caller() {
799
+ let mut chart = gauge_chart();
800
+
801
+ chart.ingest("dim1", 1.0, ns(1), 0);
802
+ let mut buf = String::new();
803
+ chart.emit(1000, &mut buf);
804
+ assert!(buf.contains("END 1001\n")); // slot 1000 + interval 1
805
+
806
+ // Second tick with no data → gap slot at 1001.
807
+ buf.clear();
808
+ chart.emit(1001, &mut buf);
809
+ assert!(buf.contains("END 1002\n")); // slot 1001 + interval 1
810
+ }
811
+
812
+ #[test]
813
+ fn tick_skips_when_no_data_within_grace() {
814
+ let mut chart = Chart::new(
815
+ "test",
816
+ ChartAggregationType::Gauge,
817
+ ChartType::Line,
818
+ ChartConfig {
819
+ collection_interval: 1,
820
+ expiry_duration: Duration::from_secs(300),
821
+ grace_period: Duration::from_secs(5),
822
+ },
823
+ );
824
+
825
+ // Ingest data and tick to emit.
826
+ chart.ingest("dim1", 42.0, ns(5), 0);
827
+ let mut buf = String::new();
828
+ chart.emit(1, &mut buf);
829
+ assert!(!buf.is_empty());
830
+
831
+ // Tick again with no new data — grace period is still active,
832
+ // so the tick should skip.
833
+ buf.clear();
834
+ chart.emit(2, &mut buf);
835
+ assert!(buf.is_empty());
836
+ }
837
+
838
+ #[test]
839
+ fn tick_gap_fills_after_grace_expires() {
840
+ let mut chart = Chart::new(
841
+ "test",
842
+ ChartAggregationType::Gauge,
843
+ ChartType::Line,
844
+ ChartConfig {
845
+ collection_interval: 1,
846
+ expiry_duration: Duration::from_secs(300),
847
+ grace_period: Duration::ZERO,
848
+ },
849
+ );
850
+
851
+ // Ingest data and tick to emit.
852
+ chart.ingest("dim1", 42.0, ns(5), 0);
853
+ let mut buf = String::new();
854
+ chart.emit(1, &mut buf);
855
+ assert!(!buf.is_empty());
856
+ assert_eq!(chart.dim_values()[0].value, Some(42.0));
857
+
858
+ // Tick with no new data and zero grace period — gap-fill repeats
859
+ // the last gauge value.
860
+ buf.clear();
861
+ chart.emit(2, &mut buf);
862
+ 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));
865
+ }
866
+
867
+ #[test]
868
+ fn tick_respects_interval_boundary() {
869
+ let mut chart = Chart::new(
870
+ "test",
871
+ ChartAggregationType::Gauge,
872
+ ChartType::Line,
873
+ ChartConfig {
874
+ collection_interval: 10,
875
+ expiry_duration: Duration::from_secs(300),
876
+ grace_period: Duration::ZERO,
877
+ },
878
+ );
879
+
880
+ // Ingest data so the chart is active.
881
+ chart.ingest("dim1", 42.0, ns(5), 0);
882
+
883
+ // Tick at t=5: slot boundary = 0, first emission.
884
+ let mut buf = String::new();
885
+ chart.emit(5, &mut buf);
886
+ assert!(!buf.is_empty());
887
+
888
+ // Tick at t=9: same slot boundary (0), should not emit.
889
+ chart.ingest("dim1", 43.0, ns(9), 0);
890
+ buf.clear();
891
+ chart.emit(9, &mut buf);
892
+ assert!(buf.is_empty());
893
+
894
+ // Tick at t=10: new slot boundary (10), should emit.
895
+ chart.ingest("dim1", 44.0, ns(10), 0);
896
+ buf.clear();
897
+ chart.emit(10, &mut buf);
898
+ assert!(!buf.is_empty());
899
+ assert_eq!(chart.dim_values()[0].value, Some(44.0));
900
+
901
+ // Tick at t=11: same slot boundary (10), should not emit.
902
+ chart.ingest("dim1", 45.0, ns(11), 0);
903
+ buf.clear();
904
+ chart.emit(11, &mut buf);
905
+ assert!(buf.is_empty());
906
+
907
+ // Tick at t=20: new slot boundary (20), should emit.
908
+ chart.ingest("dim1", 46.0, ns(20), 0);
909
+ buf.clear();
910
+ chart.emit(20, &mut buf);
911
+ assert!(!buf.is_empty());
912
+ assert_eq!(chart.dim_values()[0].value, Some(46.0));
913
+ }
914
+
915
+ #[test]
916
+ fn delta_sum_tick_after_expiry_emits_nothing() {
917
+ let mut chart = Chart::new(
918
+ "test",
919
+ ChartAggregationType::DeltaSum,
920
+ ChartType::Line,
921
+ ChartConfig {
922
+ collection_interval: 1,
923
+ expiry_duration: Duration::ZERO,
924
+ grace_period: Duration::ZERO,
925
+ },
926
+ );
927
+
928
+ chart.ingest("dim1", 10.0, ns(5), 0);
929
+
930
+ let mut buf = String::new();
931
+ chart.emit(1, &mut buf);
932
+ assert!(buf.is_empty());
933
+ }
934
+
935
+ #[test]
936
+ fn delta_sum_tick_skips_when_no_data_within_grace() {
937
+ let mut chart = Chart::new(
938
+ "test",
939
+ ChartAggregationType::DeltaSum,
940
+ ChartType::Line,
941
+ ChartConfig {
942
+ collection_interval: 1,
943
+ expiry_duration: Duration::from_secs(300),
944
+ grace_period: Duration::from_secs(5),
945
+ },
946
+ );
947
+
948
+ chart.ingest("dim1", 10.0, ns(5), 0);
949
+ let mut buf = String::new();
950
+ chart.emit(1, &mut buf);
951
+ assert!(!buf.is_empty());
952
+
953
+ // Tick again with no new data — grace period is still active.
954
+ buf.clear();
955
+ chart.emit(2, &mut buf);
956
+ assert!(buf.is_empty());
957
+ }
958
+
959
+ #[test]
960
+ fn cumulative_sum_tick_after_expiry_emits_nothing() {
961
+ let mut chart = Chart::new(
962
+ "test",
963
+ ChartAggregationType::CumulativeSum,
964
+ ChartType::Line,
965
+ ChartConfig {
966
+ collection_interval: 1,
967
+ expiry_duration: Duration::ZERO,
968
+ grace_period: Duration::ZERO,
969
+ },
970
+ );
971
+
972
+ chart.ingest("dim1", 100.0, ns(5), 1_000_000_000);
973
+
974
+ let mut buf = String::new();
975
+ chart.emit(1, &mut buf);
976
+ assert!(buf.is_empty());
977
+ }
978
+
979
+ #[test]
980
+ fn cumulative_sum_tick_skips_when_no_data_within_grace() {
981
+ let mut chart = Chart::new(
982
+ "test",
983
+ ChartAggregationType::CumulativeSum,
984
+ ChartType::Line,
985
+ ChartConfig {
986
+ collection_interval: 1,
987
+ expiry_duration: Duration::from_secs(300),
988
+ grace_period: Duration::from_secs(5),
989
+ },
990
+ );
991
+
992
+ chart.ingest("dim1", 100.0, ns(5), 1_000_000_000);
993
+ let mut buf = String::new();
994
+ chart.emit(1, &mut buf);
995
+ assert!(!buf.is_empty());
996
+
997
+ // Tick again with no new data — grace period is still active.
998
+ buf.clear();
999
+ chart.emit(2, &mut buf);
1000
+ assert!(buf.is_empty());
1001
+ }
1002
+ }
1003
+
1004
+ mod gap_fill_emission {
1005
+ use super::*;
1006
+
1007
+ #[test]
1008
+ fn tick_emits_catchup_slots_before_data() {
1009
+ let mut chart = Chart::new(
1010
+ "test",
1011
+ ChartAggregationType::Gauge,
1012
+ ChartType::Line,
1013
+ ChartConfig {
1014
+ collection_interval: 10,
1015
+ expiry_duration: Duration::from_secs(300),
1016
+ grace_period: Duration::ZERO,
1017
+ },
1018
+ );
1019
+
1020
+ // Data at slot 0.
1021
+ chart.ingest("dim1", 1.0, ns(5), 0);
1022
+ let mut buf = String::new();
1023
+ chart.emit(0, &mut buf);
1024
+ assert!(!buf.is_empty());
1025
+
1026
+ // Data at slot 30 — should produce gap-filled catchup slots at
1027
+ // 10 and 20 (repeating gauge value 1.0), then data at 30.
1028
+ chart.ingest("dim1", 2.0, ns(30), 0);
1029
+ buf.clear();
1030
+ chart.emit(30, &mut buf);
1031
+ assert!(!buf.is_empty());
1032
+
1033
+ // 2 catchup slots + 1 data slot = 3 BEGIN lines.
1034
+ assert_eq!(count_begins(&buf), 3);
1035
+
1036
+ // All 3 slots have SET lines (gap-filled catchup + real data).
1037
+ assert_eq!(count_sets(&buf), 3);
1038
+
1039
+ // Verify END timestamps (slot + interval 10).
1040
+ assert!(buf.contains("END 20\n"));
1041
+ assert!(buf.contains("END 30\n"));
1042
+ assert!(buf.contains("END 40\n"));
1043
+
1044
+ // The data slot at 30 must have the NEW value (2.0), not a
1045
+ // gap-fill. This verifies that catchup slots don't consume
1046
+ // the pending data.
1047
+ assert_eq!(chart.dim_values()[0].value, Some(2.0));
1048
+ }
1049
+
1050
+ #[test]
1051
+ fn tick_drains_one_fill_per_tick() {
1052
+ let mut chart = gauge_chart();
1053
+
1054
+ // Ingest and emit at slot 1.
1055
+ chart.ingest("dim1", 1.0, ns(5), 0);
1056
+ let mut buf = String::new();
1057
+ chart.emit(1, &mut buf);
1058
+ assert!(!buf.is_empty());
1059
+
1060
+ // Grace = ZERO, so each subsequent tick with no data emits one
1061
+ // gap-filled slot (repeating the last gauge value).
1062
+ buf.clear();
1063
+ chart.emit(5, &mut buf);
1064
+ assert_eq!(count_begins(&buf), 1);
1065
+ 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));
1068
+
1069
+ buf.clear();
1070
+ chart.emit(5, &mut buf);
1071
+ 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));
1074
+
1075
+ buf.clear();
1076
+ 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));
1079
+ }
1080
+
1081
+ #[test]
1082
+ fn tick_first_emission_no_preceding_catchup() {
1083
+ let mut chart = gauge_chart();
1084
+
1085
+ // First data ever — no catchup slots should precede it.
1086
+ chart.ingest("dim1", 1.0, ns(100), 0);
1087
+ let mut buf = String::new();
1088
+ chart.emit(100, &mut buf);
1089
+
1090
+ assert_eq!(count_begins(&buf), 1);
1091
+ assert_eq!(count_sets(&buf), 1);
1092
+ assert!(buf.contains("END 101\n")); // slot 100 + interval 1
1093
+ }
1094
+
1095
+ #[test]
1096
+ fn delta_sum_gap_fills_with_zero() {
1097
+ let mut chart = delta_sum_chart();
1098
+
1099
+ chart.ingest("dim1", 10.0, ns(1), 0);
1100
+ let mut buf = String::new();
1101
+ chart.emit(1, &mut buf);
1102
+ assert_eq!(chart.dim_values()[0].value, Some(10.0));
1103
+
1104
+ // No new data — gap-fill emits 0 for delta sums.
1105
+ buf.clear();
1106
+ chart.emit(2, &mut buf);
1107
+ assert!(!buf.is_empty());
1108
+ assert_eq!(count_sets(&buf), 1);
1109
+ assert_eq!(chart.dim_values()[0].value, Some(0.0));
1110
+ }
1111
+
1112
+ #[test]
1113
+ fn gauge_catchup_repeats_last_value() {
1114
+ let mut chart = Chart::new(
1115
+ "test",
1116
+ ChartAggregationType::Gauge,
1117
+ ChartType::Line,
1118
+ ChartConfig {
1119
+ collection_interval: 10,
1120
+ expiry_duration: Duration::from_secs(300),
1121
+ grace_period: Duration::ZERO,
1122
+ },
1123
+ );
1124
+
1125
+ // Emit at slot 0 with value 42.0.
1126
+ chart.ingest("dim1", 42.0, ns(5), 0);
1127
+ let mut buf = String::new();
1128
+ chart.emit(0, &mut buf);
1129
+
1130
+ // Data at slot 20 — catchup at slot 10 should repeat 42.0.
1131
+ chart.ingest("dim1", 99.0, ns(20), 0);
1132
+ buf.clear();
1133
+ chart.emit(20, &mut buf);
1134
+
1135
+ // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1136
+ assert_eq!(count_begins(&buf), 2);
1137
+ assert_eq!(count_sets(&buf), 2);
1138
+ assert!(buf.contains("END 20\n")); // catchup slot 10 + interval 10
1139
+ assert!(buf.contains("END 30\n")); // data slot 20 + interval 10
1140
+ }
1141
+
1142
+ #[test]
1143
+ fn delta_sum_catchup_emits_zero() {
1144
+ let mut chart = Chart::new(
1145
+ "test",
1146
+ ChartAggregationType::DeltaSum,
1147
+ ChartType::Line,
1148
+ ChartConfig {
1149
+ collection_interval: 10,
1150
+ expiry_duration: Duration::from_secs(300),
1151
+ grace_period: Duration::ZERO,
1152
+ },
1153
+ );
1154
+
1155
+ // Emit at slot 0 with delta 10.
1156
+ chart.ingest("dim1", 10.0, ns(5), 0);
1157
+ let mut buf = String::new();
1158
+ chart.emit(0, &mut buf);
1159
+ assert_eq!(chart.dim_values()[0].value, Some(10.0));
1160
+
1161
+ // Data at slot 20 — catchup at slot 10 should emit 0.
1162
+ chart.ingest("dim1", 5.0, ns(20), ns(10));
1163
+ buf.clear();
1164
+ chart.emit(20, &mut buf);
1165
+
1166
+ // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1167
+ assert_eq!(count_begins(&buf), 2);
1168
+ assert_eq!(count_sets(&buf), 2);
1169
+ assert!(buf.contains("END 20\n")); // catchup slot 10 + interval 10
1170
+ assert!(buf.contains("END 30\n")); // data slot 20 + interval 10
1171
+
1172
+ // Data slot has the new delta value.
1173
+ assert_eq!(chart.dim_values()[0].value, Some(5.0));
1174
+ }
1175
+
1176
+ #[test]
1177
+ fn cumulative_sum_catchup_emits_zero() {
1178
+ let mut chart = Chart::new(
1179
+ "test",
1180
+ ChartAggregationType::CumulativeSum,
1181
+ ChartType::Line,
1182
+ ChartConfig {
1183
+ collection_interval: 10,
1184
+ expiry_duration: Duration::from_secs(300),
1185
+ grace_period: Duration::ZERO,
1186
+ },
1187
+ );
1188
+
1189
+ const START_TIME: u64 = 1_000_000_000;
1190
+
1191
+ // Slot 0: baseline (first slot returns None for cumulative sum).
1192
+ chart.ingest("dim1", 100.0, ns(5), START_TIME);
1193
+ let mut buf = String::new();
1194
+ chart.emit(0, &mut buf);
1195
+ assert_eq!(chart.dim_values()[0].value, None);
1196
+
1197
+ // Data at slot 20 — catchup at slot 10 should gap-fill with 0.
1198
+ chart.ingest("dim1", 150.0, ns(20), START_TIME);
1199
+ buf.clear();
1200
+ chart.emit(20, &mut buf);
1201
+
1202
+ // 1 catchup + 1 data = 2 BEGIN/SET/END blocks.
1203
+ assert_eq!(count_begins(&buf), 2);
1204
+ assert!(buf.contains("END 20\n")); // catchup slot 10 + interval 10
1205
+ assert!(buf.contains("END 30\n")); // data slot 20 + interval 10
1206
+
1207
+ // Data slot: delta = 150 - 100 = 50.
1208
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1209
+ }
1210
+ }
1211
+
1212
+ mod slot_tracking {
1213
+ use super::*;
1214
+
1215
+ #[test]
1216
+ fn drops_data_for_previous_slot() {
1217
+ let mut chart = gauge_chart();
1218
+
1219
+ // Active slot becomes 1.
1220
+ chart.ingest("dim1", 50.0, ns(1), 0);
1221
+
1222
+ // Data for slot 0 — should be dropped.
1223
+ chart.ingest("dim1", 42.0, ns(0), 0);
1224
+
1225
+ let mut buf = String::new();
1226
+ chart.emit(1, &mut buf);
1227
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1228
+ }
1229
+
1230
+ #[test]
1231
+ fn delta_sum_slot_transition_resets_accumulator() {
1232
+ let mut chart = delta_sum_chart();
1233
+
1234
+ // Slot 0: accumulate delta=10.
1235
+ chart.ingest("dim1", 10.0, ns(0), 0);
1236
+
1237
+ // Slot 1: transition resets per-slot state; accumulate delta=5.
1238
+ chart.ingest("dim1", 5.0, ns(1), ns(0));
1239
+
1240
+ // Tick should see only the slot-1 delta (10 was finalized on transition).
1241
+ let mut buf = String::new();
1242
+ chart.emit(1, &mut buf);
1243
+ assert_eq!(chart.dim_values()[0].value, Some(5.0));
1244
+ }
1245
+
1246
+ #[test]
1247
+ fn cumulative_sum_slot_transition_advances_baseline() {
1248
+ let mut chart = cumulative_sum_chart();
1249
+
1250
+ const START_TIME: u64 = 1_000_000_000;
1251
+
1252
+ // Slot 0: baseline cumulative=100.
1253
+ chart.ingest("dim1", 100.0, ns(0), START_TIME);
1254
+
1255
+ // Slot 1: transition finalizes slot 0 (promoting 100 to previous),
1256
+ // then ingest cumulative=150.
1257
+ chart.ingest("dim1", 150.0, ns(1), START_TIME);
1258
+
1259
+ // Tick: delta should be 150 - 100 = 50.
1260
+ let mut buf = String::new();
1261
+ chart.emit(1, &mut buf);
1262
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1263
+ }
1264
+
1265
+ #[test]
1266
+ fn cumulative_sum_restart_across_slot_transition() {
1267
+ let mut chart = cumulative_sum_chart();
1268
+
1269
+ const START_TIME: u64 = 1_000_000_000;
1270
+
1271
+ // Slot 0: baseline.
1272
+ chart.ingest("dim1", 100.0, ns(0), START_TIME);
1273
+
1274
+ // Slot 1: normal delta.
1275
+ chart.ingest("dim1", 150.0, ns(1), START_TIME);
1276
+
1277
+ // Slot 2: restart (new start_time).
1278
+ let new_start = START_TIME + 1_000_000;
1279
+ chart.ingest("dim1", 20.0, ns(2), new_start);
1280
+
1281
+ // Tick: restart slot should report 0.
1282
+ let mut buf = String::new();
1283
+ chart.emit(2, &mut buf);
1284
+ assert_eq!(chart.dim_values()[0].value, Some(0.0));
1285
+ }
1286
+
1287
+ #[test]
1288
+ fn multi_slot_ingest_then_tick() {
1289
+ let mut chart = gauge_chart();
1290
+
1291
+ // Data spanning three slots arrives before tick fires.
1292
+ chart.ingest("dim1", 10.0, ns(0), 0);
1293
+ chart.ingest("dim1", 20.0, ns(1), 0);
1294
+ chart.ingest("dim1", 30.0, ns(2), 0);
1295
+
1296
+ // Tick sees only the last slot's value (slot transitions
1297
+ // finalized the earlier ones).
1298
+ let mut buf = String::new();
1299
+ chart.emit(2, &mut buf);
1300
+ assert_eq!(chart.dim_values()[0].value, Some(30.0));
1301
+ }
1302
+
1303
+ #[test]
1304
+ fn gap_fill_across_slot_transition() {
1305
+ let mut chart = gauge_chart();
1306
+
1307
+ // Slot 0: both dimensions.
1308
+ chart.ingest("dim1", 10.0, ns(0), 0);
1309
+ chart.ingest("dim2", 20.0, ns(0), 0);
1310
+
1311
+ // Slot 1: only dim1 — triggers slot transition which finalizes
1312
+ // dim2 via gap_fill, establishing its last_emitted value.
1313
+ chart.ingest("dim1", 15.0, ns(1), 0);
1314
+
1315
+ // Tick: dim1 has slot-1 data, dim2 should gap-fill.
1316
+ 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));
1320
+ }
1321
+ }
1322
+
1323
+ mod gauge_aggregation {
1324
+ use super::*;
1325
+
1326
+ #[test]
1327
+ fn keeps_last_value_by_timestamp() {
1328
+ let mut chart = gauge_chart();
1329
+
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);
1333
+
1334
+ let mut buf = String::new();
1335
+ chart.emit(1, &mut buf);
1336
+ assert_eq!(chart.dim_values()[0].value, Some(30.0));
1337
+ }
1338
+
1339
+ #[test]
1340
+ fn gap_fills_missing_dimension() {
1341
+ let mut chart = gauge_chart();
1342
+
1343
+ // Both dimensions get data.
1344
+ chart.ingest("dim1", 10.0, ns(5), 0);
1345
+ chart.ingest("dim2", 20.0, ns(5), 0);
1346
+
1347
+ // Tick finalizes both.
1348
+ 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));
1352
+
1353
+ // Only dim1 gets new data.
1354
+ chart.ingest("dim1", 15.0, ns(6), 0);
1355
+
1356
+ // Tick: dim2 should be gap-filled with previous value.
1357
+ 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));
1361
+ }
1362
+
1363
+ #[test]
1364
+ fn out_of_order_timestamps_keeps_latest() {
1365
+ let mut chart = gauge_chart();
1366
+
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);
1370
+
1371
+ let mut buf = String::new();
1372
+ chart.emit(1, &mut buf);
1373
+ assert_eq!(chart.dim_values()[0].value, Some(30.0));
1374
+ }
1375
+
1376
+ #[test]
1377
+ fn multi_dimension_gap_fill() {
1378
+ let mut chart = gauge_chart();
1379
+
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);
1384
+
1385
+ 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));
1391
+
1392
+ // Only dim1 gets new data.
1393
+ chart.ingest("dim1", 110.0, ns(6), 0);
1394
+
1395
+ 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
1400
+ }
1401
+ }
1402
+
1403
+ mod delta_sum_aggregation {
1404
+ use super::*;
1405
+
1406
+ #[test]
1407
+ fn sums_deltas() {
1408
+ let mut chart = delta_sum_chart();
1409
+
1410
+ // All within the same slot (same second).
1411
+ chart.ingest("dim1", 10.0, ms(100), 0);
1412
+ chart.ingest("dim1", 20.0, ms(200), ms(100));
1413
+ chart.ingest("dim1", 5.0, ms(300), ms(200));
1414
+
1415
+ let mut buf = String::new();
1416
+ chart.emit(1, &mut buf);
1417
+ assert_eq!(chart.dim_values()[0].value, Some(35.0));
1418
+ }
1419
+
1420
+ #[test]
1421
+ fn accumulates_correctly_with_multiple_ingests() {
1422
+ let mut chart = delta_sum_chart();
1423
+
1424
+ // All within the same slot (same second).
1425
+ chart.ingest("dim1", 5.0, ms(100), 0);
1426
+ chart.ingest("dim1", 10.0, ms(200), ms(100));
1427
+ chart.ingest("dim1", 15.0, ms(300), ms(200));
1428
+ chart.ingest("dim1", 20.0, ms(400), ms(300));
1429
+
1430
+ let mut buf = String::new();
1431
+ chart.emit(1, &mut buf);
1432
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1433
+ }
1434
+
1435
+ #[test]
1436
+ fn no_data_gap_fills_with_zero() {
1437
+ let mut chart = delta_sum_chart();
1438
+
1439
+ chart.ingest("dim1", 10.0, ns(1), 0);
1440
+ let mut buf = String::new();
1441
+ chart.emit(1, &mut buf);
1442
+ assert_eq!(chart.dim_values()[0].value, Some(10.0));
1443
+
1444
+ // No new data — gap-fill emits 0 for delta sums.
1445
+ buf.clear();
1446
+ chart.emit(2, &mut buf);
1447
+ assert!(!buf.is_empty());
1448
+ assert_eq!(count_sets(&buf), 1);
1449
+ assert_eq!(chart.dim_values()[0].value, Some(0.0));
1450
+ }
1451
+
1452
+ #[test]
1453
+ fn non_monotonic_delta_sum_accumulates() {
1454
+ // Non-monotonic delta sums behave identically to monotonic:
1455
+ // deltas are accumulated within a slot.
1456
+ let mut chart = Chart::from_metric(
1457
+ "test",
1458
+ MetricDataKind::Sum,
1459
+ Some(AggregationTemporality::Delta),
1460
+ Some(false),
1461
+ test_config(),
1462
+ )
1463
+ .unwrap();
1464
+
1465
+ // Accumulate deltas, including a negative one.
1466
+ chart.ingest("dim1", 10.0, ms(100), 0);
1467
+ chart.ingest("dim1", -3.0, ms(200), ms(100));
1468
+ chart.ingest("dim1", 5.0, ms(300), ms(200));
1469
+
1470
+ let mut buf = String::new();
1471
+ chart.emit(1, &mut buf);
1472
+ assert_eq!(chart.dim_values()[0].value, Some(12.0));
1473
+ }
1474
+ }
1475
+
1476
+ mod cumulative_sum_aggregation {
1477
+ use super::*;
1478
+
1479
+ const START_TIME: u64 = 1_000_000_000;
1480
+
1481
+ #[test]
1482
+ fn first_slot_returns_none() {
1483
+ let mut chart = cumulative_sum_chart();
1484
+
1485
+ chart.ingest("dim1", 100.0, ns(5), START_TIME);
1486
+ let mut buf = String::new();
1487
+ chart.emit(1, &mut buf);
1488
+ assert_eq!(chart.dim_values()[0].value, None);
1489
+ }
1490
+
1491
+ #[test]
1492
+ fn computes_deltas_across_ticks() {
1493
+ let mut chart = cumulative_sum_chart();
1494
+
1495
+ // First tick: baseline, no delta.
1496
+ chart.ingest("dim1", 100.0, ns(5), START_TIME);
1497
+ let mut buf = String::new();
1498
+ chart.emit(1, &mut buf);
1499
+ assert_eq!(chart.dim_values()[0].value, None);
1500
+
1501
+ // Second tick: delta = 150 - 100 = 50.
1502
+ chart.ingest("dim1", 150.0, ns(6), START_TIME);
1503
+ buf.clear();
1504
+ chart.emit(2, &mut buf);
1505
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1506
+ }
1507
+
1508
+ #[test]
1509
+ fn detects_restart() {
1510
+ let mut chart = cumulative_sum_chart();
1511
+
1512
+ // Establish baseline.
1513
+ chart.ingest("dim1", 100.0, ns(5), START_TIME);
1514
+ let mut buf = String::new();
1515
+ chart.emit(1, &mut buf);
1516
+
1517
+ // Normal delta.
1518
+ chart.ingest("dim1", 150.0, ns(6), START_TIME);
1519
+ buf.clear();
1520
+ chart.emit(2, &mut buf);
1521
+ assert_eq!(chart.dim_values()[0].value, Some(50.0));
1522
+
1523
+ // Restart: new start_time.
1524
+ let new_start = START_TIME + 1_000_000;
1525
+ chart.ingest("dim1", 20.0, ns(7), new_start);
1526
+ buf.clear();
1527
+ chart.emit(3, &mut buf);
1528
+ assert_eq!(chart.dim_values()[0].value, Some(0.0));
1529
+ }
1530
+ }
1531
+
1532
+ mod definition {
1533
+ use super::*;
1534
+
1535
+ #[test]
1536
+ fn new_dimension_invalidates_definition() {
1537
+ let mut chart = gauge_chart();
1538
+
1539
+ chart.init_definition("metric", "title", "units", vec![]);
1540
+ // Emit definition to mark as emitted.
1541
+ let mut buf = String::new();
1542
+ chart.emit_definition_if_needed(&mut buf);
1543
+ assert!(!chart.needs_definition());
1544
+
1545
+ // Ingest a new dimension.
1546
+ chart.ingest("dim1", 1.0, ns(1), 0);
1547
+ assert!(chart.needs_definition());
1548
+ }
1549
+
1550
+ #[test]
1551
+ fn definition_tracks_dimensions() {
1552
+ let mut chart = gauge_chart();
1553
+
1554
+ chart.init_definition("metric", "title", "units", vec![]);
1555
+
1556
+ chart.ingest("dim1", 1.0, ns(1), 0);
1557
+ chart.ingest("dim2", 2.0, ns(1), 0);
1558
+
1559
+ let def = chart.definition().unwrap();
1560
+ assert_eq!(def.dimensions.len(), 2);
1561
+ assert!(def.dimensions.contains(&"dim1".to_string()));
1562
+ assert!(def.dimensions.contains(&"dim2".to_string()));
1563
+ }
1564
+
1565
+ #[test]
1566
+ fn tick_definition_includes_store_first() {
1567
+ let mut chart = gauge_chart();
1568
+
1569
+ chart.init_definition("metric", "title", "units", vec![]);
1570
+ chart.ingest("dim1", 1.0, ns(1), 0);
1571
+
1572
+ let mut buf = String::new();
1573
+ chart.emit(1, &mut buf);
1574
+
1575
+ // The CHART line should include 'store_first'.
1576
+ let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
1577
+ assert!(
1578
+ chart_line.contains("'store_first'"),
1579
+ "CHART line missing store_first: {}",
1580
+ chart_line
1581
+ );
1582
+ }
1583
+
1584
+ #[test]
1585
+ fn line_chart_emits_line_type() {
1586
+ let mut chart = Chart::new(
1587
+ "test",
1588
+ ChartAggregationType::Gauge,
1589
+ ChartType::Line,
1590
+ test_config(),
1591
+ );
1592
+
1593
+ chart.init_definition("metric", "title", "units", vec![]);
1594
+ chart.ingest("dim1", 1.0, ns(1), 0);
1595
+
1596
+ let mut buf = String::new();
1597
+ chart.emit(1, &mut buf);
1598
+
1599
+ let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
1600
+ assert!(
1601
+ chart_line.contains(" line "),
1602
+ "CHART line should contain 'line': {}",
1603
+ chart_line
1604
+ );
1605
+ }
1606
+
1607
+ #[test]
1608
+ fn heatmap_chart_emits_heatmap_type() {
1609
+ let mut chart = Chart::new(
1610
+ "test",
1611
+ ChartAggregationType::DeltaSum,
1612
+ ChartType::Heatmap,
1613
+ test_config(),
1614
+ );
1615
+
1616
+ chart.init_definition("metric", "title", "units", vec![]);
1617
+ chart.ingest("dim1", 1.0, ns(1), 0);
1618
+
1619
+ let mut buf = String::new();
1620
+ chart.emit(1, &mut buf);
1621
+
1622
+ let chart_line = buf.lines().find(|l| l.starts_with("CHART ")).unwrap();
1623
+ assert!(
1624
+ chart_line.contains(" heatmap "),
1625
+ "CHART line should contain 'heatmap': {}",
1626
+ chart_line
1627
+ );
1628
+ }
1629
+ }
1630
+}