vey-statsd: fix diff value when aggregate in exporter

Author: zh-jq

vey-statsd/CHANGELOG

@@ -1,5 +1,6 @@
 
 v0.2.0:
+ - BUG FIX: fix diff value when aggregate in exporter
  - Compatibility: bump MSRV to 1.90.0
 
 v0.1.1:

vey-statsd/src/runtime/export/aggregate/mod.rs

@@ -137,9 +137,14 @@ impl<T: AggregateExport> AggregateExportRuntime<T> {
                     .inner
                     .entry(record.tag_map.clone())
                     .and_modify(|v| {
-                        v.time = self.store_time;
-                        v.sum += record.value;
-                        v.diff = record.value;
+                        if v.time < self.store_time {
+                            v.time = self.store_time;
+                            v.sum += record.value;
+                            v.diff = record.value;
+                        } else {
+                            v.sum += record.value;
+                            v.diff += record.value;
+                        }
                     })
                     .or_insert(CounterStoreValue {
                         time: self.store_time,
@@ -160,3 +165,100 @@ impl<T: AggregateExport> AggregateExportRuntime<T> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use tokio::sync::mpsc;
+    use vey_types::metrics::MetricTagMap;
+
+    struct TestExporter {
+        counters: AHashMap<MetricName, AHashMap<Arc<MetricTagMap>, (MetricValue, MetricValue)>>,
+    }
+
+    impl AggregateExport for TestExporter {
+        fn emit_interval(&self) -> Duration {
+            Duration::from_secs(10)
+        }
+
+        fn emit_gauge(
+            &mut self,
+            _name: &MetricName,
+            _values: &AHashMap<Arc<MetricTagMap>, GaugeStoreValue>,
+        ) {
+        }
+
+        fn emit_counter(
+            &mut self,
+            name: &MetricName,
+            values: &AHashMap<Arc<MetricTagMap>, CounterStoreValue>,
+        ) {
+            let map = self.counters.entry(name.clone()).or_default();
+            for (tags, v) in values {
+                map.insert(tags.clone(), (v.sum, v.diff));
+            }
+        }
+    }
+
+    #[test]
+    fn test_counter_diff_accumulation_and_reset() {
+        let (_tx, rx) = mpsc::unbounded_channel();
+        let exporter = TestExporter {
+            counters: AHashMap::default(),
+        };
+        let mut runtime = AggregateExportRuntime::new(exporter, rx);
+
+        let name = Arc::new(MetricName::parse("test.counter").unwrap());
+        let tag_map = Arc::new(MetricTagMap::default());
+
+        // Interval 1 - First record
+        runtime.add_record(MetricRecord {
+            name: name.clone(),
+            tag_map: tag_map.clone(),
+            r#type: MetricType::Counter,
+            value: MetricValue::Signed(10),
+        });
+
+        // Interval 1 - Second record (same interval)
+        runtime.add_record(MetricRecord {
+            name: name.clone(),
+            tag_map: tag_map.clone(),
+            r#type: MetricType::Counter,
+            value: MetricValue::Signed(5),
+        });
+
+        // Verify state before retain
+        let counter_entry = &runtime.counter.get(&name).unwrap().inner[&tag_map];
+        assert_eq!(counter_entry.sum, MetricValue::Signed(15));
+        assert_eq!(counter_entry.diff, MetricValue::Signed(15));
+
+        // Simulate tick / interval transition
+        runtime.retain();
+
+        // Interval 2 - First record in new interval
+        runtime.add_record(MetricRecord {
+            name: name.clone(),
+            tag_map: tag_map.clone(),
+            r#type: MetricType::Counter,
+            value: MetricValue::Signed(3),
+        });
+
+        // Verify diff reset for new interval, sum accumulated
+        let counter_entry = &runtime.counter.get(&name).unwrap().inner[&tag_map];
+        assert_eq!(counter_entry.sum, MetricValue::Signed(18));
+        assert_eq!(counter_entry.diff, MetricValue::Signed(3));
+
+        // Interval 2 - Second record in new interval
+        runtime.add_record(MetricRecord {
+            name: name.clone(),
+            tag_map: tag_map.clone(),
+            r#type: MetricType::Counter,
+            value: MetricValue::Signed(7),
+        });
+
+        let counter_entry = &runtime.counter.get(&name).unwrap().inner[&tag_map];
+        assert_eq!(counter_entry.sum, MetricValue::Signed(25));
+        assert_eq!(counter_entry.diff, MetricValue::Signed(10));
+    }
+}