debugging: add trace logging to show exactly which buckets are being flushed during aggregation

Author: tobz

Diff for: lib/saluki-components/src/transforms/aggregate/mod.rs

@@ -533,6 +533,13 @@ impl AggregationState {
             // This means we'll always remove all-closed/empty non-counter metrics, and we _may_ remove all-closed/empty
             // counters.
             if let Some(closed_bucket_values) = am.values.split_at_timestamp(split_timestamp) {
+                trace!(
+                    metric_name = &**context.name(),
+                    metric_tags = %context.tags(),
+                    points = %closed_bucket_values,
+                    "Flushing closed buckets."
+                );
+
                 // We got some closed bucket values, so flush those out.
                 transform_and_push_metric(
                     context.clone(),

Diff for: lib/saluki-context/src/tags/mod.rs

@@ -368,6 +368,22 @@ impl Tagged for TagSet {
     }
 }
 
+impl fmt::Display for TagSet {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[")?;
+
+        for (i, tag) in self.0.iter().enumerate() {
+            if i > 0 {
+                write!(f, ",")?;
+            }
+
+            write!(f, "{}", tag.as_str())?;
+        }
+
+        write!(f, "]")
+    }
+}
+
 /// A shared, read-only set of tags.
 #[derive(Clone, Debug)]
 pub struct SharedTagSet(Arc<TagSet>);

Diff for: lib/saluki-event/src/metric/value/histogram.rs

@@ -1,4 +1,4 @@
-use std::num::NonZeroU64;
+use std::{fmt, num::NonZeroU64};
 
 use ordered_float::OrderedFloat;
 use smallvec::SmallVec;
@@ -291,6 +291,28 @@ impl<'a> IntoIterator for &'a mut HistogramPoints {
     }
 }
 
+impl fmt::Display for HistogramPoints {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[")?;
+        for (i, point) in self.0.values.iter().enumerate() {
+            if i > 0 {
+                write!(f, ",")?;
+            }
+
+            let ts = point.timestamp.map(|ts| ts.get()).unwrap_or_default();
+            write!(f, "({}, [", ts)?;
+            for (j, sample) in point.value.samples().iter().enumerate() {
+                if j > 0 {
+                    write!(f, ",")?;
+                }
+                write!(f, "{{{} * {}}}", sample.value, sample.weight)?;
+            }
+            write!(f, "])")?;
+        }
+        write!(f, "]")
+    }
+}
+
 pub struct HistogramIter {
     inner: smallvec::IntoIter<[TimestampedValue<Histogram>; 1]>,
 }

Diff for: lib/saluki-event/src/metric/value/mod.rs

@@ -1,6 +1,6 @@
 mod iter;
 
-use std::{collections::HashSet, num::NonZeroU64, time::Duration};
+use std::{collections::HashSet, fmt, num::NonZeroU64, time::Duration};
 
 use ddsketch_agent::DDSketch;
 use ordered_float::OrderedFloat;
@@ -472,6 +472,19 @@ impl MetricValues {
     }
 }
 
+impl fmt::Display for MetricValues {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Counter(points) => write!(f, "{}", points),
+            Self::Rate(points, interval) => write!(f, "{} over {:?}", points, interval),
+            Self::Gauge(points) => write!(f, "{}", points),
+            Self::Set(points) => write!(f, "{}", points),
+            Self::Histogram(points) => write!(f, "{}", points),
+            Self::Distribution(points) => write!(f, "{}", points),
+        }
+    }
+}
+
 fn collapse_scalar_merge(dest: &mut OrderedFloat<f64>, src: &mut OrderedFloat<f64>) {
     *dest += *src;
 }

Diff for: lib/saluki-event/src/metric/value/scalar.rs

@@ -1,4 +1,4 @@
-use std::num::NonZeroU64;
+use std::{fmt, num::NonZeroU64};
 
 use ordered_float::OrderedFloat;
 
@@ -145,3 +145,18 @@ impl<'a> IntoIterator for &'a ScalarPoints {
         PointsIterRef::scalar(self.0.values.iter())
     }
 }
+
+impl fmt::Display for ScalarPoints {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[")?;
+        for (i, (timestamp, value)) in self.into_iter().enumerate() {
+            if i > 0 {
+                write!(f, ",")?;
+            }
+
+            let ts = timestamp.map(|ts| ts.get()).unwrap_or_default();
+            write!(f, "({}, {})", ts, value)?;
+        }
+        write!(f, "]")
+    }
+}

Diff for: lib/saluki-event/src/metric/value/set.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, num::NonZeroU64};
+use std::{collections::HashSet, fmt, num::NonZeroU64};
 
 use super::{
     iter::{PointsIter, PointsIterRef},
@@ -97,3 +97,25 @@ impl<'a> IntoIterator for &'a SetPoints {
         PointsIterRef::set(self.0.values.iter())
     }
 }
+
+impl fmt::Display for SetPoints {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[")?;
+        for (i, point) in self.0.values.iter().enumerate() {
+            if i > 0 {
+                write!(f, ",")?;
+            }
+
+            let ts = point.timestamp.map(|ts| ts.get()).unwrap_or_default();
+            write!(f, "({}, [", ts)?;
+            for (j, value) in point.value.iter().enumerate() {
+                if j > 0 {
+                    write!(f, ",")?;
+                }
+                write!(f, "{}", value)?;
+            }
+            write!(f, "])")?;
+        }
+        write!(f, "]")
+    }
+}

Diff for: lib/saluki-event/src/metric/value/sketch.rs

@@ -1,4 +1,4 @@
-use std::num::NonZeroU64;
+use std::{fmt, num::NonZeroU64};
 
 use ddsketch_agent::DDSketch;
 
@@ -168,6 +168,32 @@ impl<'a> IntoIterator for &'a SketchPoints {
     }
 }
 
+impl fmt::Display for SketchPoints {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[")?;
+        for (i, point) in self.0.values.iter().enumerate() {
+            if i > 0 {
+                write!(f, ",")?;
+            }
+
+            let ts = point.timestamp.map(|ts| ts.get()).unwrap_or_default();
+            let sketch = &point.value;
+            write!(
+                f,
+                "({}, {{cnt={} min={} max={} avg={} sum={} bin_count={}}})",
+                ts,
+                sketch.count(),
+                sketch.min().unwrap_or(0.0),
+                sketch.max().unwrap_or(0.0),
+                sketch.avg().unwrap_or(0.0),
+                sketch.sum().unwrap_or(0.0),
+                sketch.bin_count(),
+            )?;
+        }
+        write!(f, "]")
+    }
+}
+
 pub struct SketchesIter {
     inner: smallvec::IntoIter<[TimestampedValue<DDSketch>; 1]>,
 }