some more tinkering with metrics

core/src/extent.rs

@@ -141,3 +141,14 @@ impl ToExtent for Range<Timestamp> {
         Extent::new(self.clone())
     }
 }
+
+impl ToExtent for Range<Option<Timestamp>> {
+    fn to_extent(&self) -> Extent {
+        match (self.start, self.end) {
+            (Some(start), Some(end)) => (start..end).to_extent(),
+            (Some(start), None) => start.to_extent(),
+            (None, Some(end)) => end.to_extent(),
+            (None, None) => None::<Timestamp>.to_extent(),
+        }
+    }
+}

core/src/metrics.rs

@@ -19,10 +19,6 @@ impl<'m, T> Metric<'m, T> {
         Metric { name, kind, value }
     }
 
-    pub const fn counter(name: Key<'m>, value: T) -> Self {
-        Metric::new(name, MetricKind::Counter, value)
-    }
-
     pub const fn name(&self) -> &Key<'m> {
         &self.name
     }
@@ -35,11 +31,33 @@ impl<'m, T> Metric<'m, T> {
         &self.value
     }
 
-    pub fn read<'a, U: ToValue>(&'a self, read: impl FnOnce(&'a T) -> U) -> Metric<'a, U> {
+    pub fn value_mut(&mut self) -> &mut T {
+        &mut self.value
+    }
+
+    pub fn sample<'a, U: ToValue>(
+        &'a self,
+        sample: impl FnOnce(MetricKind, &'a T) -> (MetricKind, U),
+    ) -> Metric<'a, U> {
+        let (kind, value) = sample(self.kind, &self.value);
+
+        Metric {
+            name: self.name.by_ref(),
+            kind,
+            value,
+        }
+    }
+
+    pub fn sample_mut<'a, U: ToValue>(
+        &'a mut self,
+        sample: impl FnOnce(MetricKind, &'a mut T) -> (MetricKind, U),
+    ) -> Metric<'a, U> {
+        let (kind, value) = sample(self.kind, &mut self.value);
+
         Metric {
             name: self.name.by_ref(),
-            kind: self.kind,
-            value: read(&self.value),
+            kind,
+            value,
         }
     }
 }

tests/smoke-test/main.rs

@@ -11,23 +11,48 @@ use emit::Props;
 #[macro_use]
 extern crate serde_derive;
 
-static COUNT: emit::Metric<'static, AtomicUsize> =
-    emit::Metric::counter(emit::Key::new("smoke_test::count"), AtomicUsize::new(0));
+struct DeltaCount {
+    last: std::sync::Mutex<(Option<emit::timestamp::Timestamp>, usize)>,
+    value: AtomicUsize,
+}
 
-fn increment(metric: &emit::Metric<AtomicUsize>) {
-    metric.value().fetch_add(1, Ordering::Relaxed);
+static COUNT: emit::Metric<'static, DeltaCount> = emit::Metric::new(
+    emit::Key::new("smoke_test::count"),
+    emit::metrics::MetricKind::Counter,
+    DeltaCount {
+        last: std::sync::Mutex::new((None, 0)),
+        value: AtomicUsize::new(0),
+    },
+);
+
+fn increment(metric: &emit::Metric<DeltaCount>) {
+    metric.value().value.fetch_add(1, Ordering::Relaxed);
 }
 
-fn flush_metrics<'a>(metrics: impl IntoIterator<Item = &'a emit::Metric<'a, AtomicUsize>> + 'a) {
+fn flush_metrics<'a>(metrics: impl IntoIterator<Item = &'a emit::Metric<'a, DeltaCount>> + 'a) {
+    let now = emit::now();
+
     for metric in metrics {
+        let mut start = None;
+        let delta = metric.sample(|_, value| {
+            let mut previous = value.last.lock().unwrap();
+            let current = value.value.load(Ordering::Relaxed);
+
+            start = previous.0;
+            let delta = current.saturating_sub(previous.1);
+
+            previous.0 = now;
+            previous.1 = current;
+
+            (emit::metrics::MetricKind::Counter, delta)
+        });
+
         emit::emit(&emit::Event::new(
-            emit::now(),
+            start..now,
             emit::tpl!("{metric_name} read {metric_value} with {ordering}"),
-            metric
-                .read(|value| value.load(Ordering::Relaxed))
-                .chain(emit::props! {
-                    ordering: "relaxed"
-                }),
+            delta.chain(emit::props! {
+                ordering: "relaxed"
+            }),
         ));
     }
 }
@@ -53,7 +78,13 @@ async fn main() {
         .and_to(emit_term::stdout())
         .init();
 
-    for i in 0..10 {
+    for i in 0..9 {
+        let _ = in_ctxt(i).await;
+    }
+
+    flush_metrics([&COUNT]);
+
+    for i in 0..7 {
         let _ = in_ctxt(i).await;
     }