Add support for timestamp encoding in exemplars

While timestamps are optional, they are required for native histograms:

* https://github.com/prometheus/prometheus/blame/4aee718013/scrape/scrape.go#L1833

Old histograms can be converted to native histograms at the ingestion time,
which in turn makes timestamps required for old histograms too.

Signed-off-by: Ivan Babrou <[email protected]>

Author: bobrik

src/encoding.rs

@@ -11,6 +11,7 @@ use std::fmt::Write;
 use std::ops::Deref;
 use std::rc::Rc;
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 #[cfg(feature = "protobuf")]
 #[cfg_attr(docsrs, doc(cfg(feature = "protobuf")))]
@@ -760,6 +761,18 @@ impl EncodeExemplarValue for u32 {
     }
 }
 
+/// An encodable exemplar time.
+pub trait EncodeExemplarTime {
+    /// Encode the time in the OpenMetrics text encoding.
+    fn encode(&self, encoder: ExemplarValueEncoder) -> Result<(), std::fmt::Error>;
+}
+
+impl EncodeExemplarTime for SystemTime {
+    fn encode(&self, mut encoder: ExemplarValueEncoder) -> Result<(), std::fmt::Error> {
+        encoder.encode(self.duration_since(UNIX_EPOCH).unwrap().as_secs_f64())
+    }
+}
+
 /// Encoder for an exemplar value.
 #[derive(Debug)]
 pub struct ExemplarValueEncoder<'a>(ExemplarValueEncoderInner<'a>);

src/encoding/protobuf.rs

@@ -311,7 +311,7 @@ impl<S: EncodeLabelSet, V: EncodeExemplarValue> TryFrom<&Exemplar<S, V>>
 
         Ok(openmetrics_data_model::Exemplar {
             value,
-            timestamp: Default::default(),
+            timestamp: Some(exemplar.time.into()),
             label: labels,
         })
     }
@@ -442,6 +442,8 @@ impl std::fmt::Write for LabelValueEncoder<'_> {
 
 #[cfg(test)]
 mod tests {
+    use prost_types::Timestamp;
+
     use super::*;
     use crate::metrics::counter::Counter;
     use crate::metrics::exemplar::{CounterWithExemplar, HistogramWithExemplars};
@@ -454,6 +456,7 @@ mod tests {
     use std::collections::HashSet;
     use std::sync::atomic::AtomicI64;
     use std::sync::atomic::AtomicU64;
+    use std::time::SystemTime;
 
     #[test]
     fn encode_counter_int() {
@@ -531,6 +534,9 @@ mod tests {
 
     #[test]
     fn encode_counter_with_exemplar() {
+        let now = SystemTime::now();
+        let now_ts: Timestamp = now.into();
+
         let mut registry = Registry::default();
 
         let counter_with_exemplar: CounterWithExemplar<Vec<(String, f64)>, f64> =
@@ -543,6 +549,14 @@ mod tests {
 
         counter_with_exemplar.inc_by(1.0, Some(vec![("user_id".to_string(), 42.0)]));
 
+        counter_with_exemplar
+            .inner
+            .write()
+            .exemplar
+            .as_mut()
+            .unwrap()
+            .time = now;
+
         let metric_set = encode(&registry).unwrap();
 
         let family = metric_set.metric_families.first().unwrap();
@@ -563,6 +577,8 @@ mod tests {
                 let exemplar = value.exemplar.as_ref().unwrap();
                 assert_eq!(1.0, exemplar.value);
 
+                assert_eq!(&now_ts, exemplar.timestamp.as_ref().unwrap());
+
                 let expected_label = {
                     openmetrics_data_model::Label {
                         name: "user_id".to_string(),
@@ -784,11 +800,23 @@ mod tests {
 
     #[test]
     fn encode_histogram_with_exemplars() {
+        let now = SystemTime::now();
+        let now_ts: Timestamp = now.into();
+
         let mut registry = Registry::default();
         let histogram = HistogramWithExemplars::new(exponential_buckets(1.0, 2.0, 10));
         registry.register("my_histogram", "My histogram", histogram.clone());
         histogram.observe(1.0, Some(vec![("user_id".to_string(), 42u64)]));
 
+        histogram
+            .inner
+            .write()
+            .exemplars
+            .get_mut(&0)
+            .as_mut()
+            .unwrap()
+            .time = now;
+
         let metric_set = encode(&registry).unwrap();
 
         let family = metric_set.metric_families.first().unwrap();
@@ -805,6 +833,8 @@ mod tests {
                 let exemplar = value.buckets.first().unwrap().exemplar.as_ref().unwrap();
                 assert_eq!(1.0, exemplar.value);
 
+                assert_eq!(&now_ts, exemplar.timestamp.as_ref().unwrap());
+
                 let expected_label = {
                     openmetrics_data_model::Label {
                         name: "user_id".to_string(),

src/encoding/text.rs

@@ -37,7 +37,7 @@
 //! assert_eq!(expected_msg, buffer);
 //! ```
 
-use crate::encoding::{EncodeExemplarValue, EncodeLabelSet, NoLabelSet};
+use crate::encoding::{EncodeExemplarTime, EncodeExemplarValue, EncodeLabelSet, NoLabelSet};
 use crate::metrics::exemplar::Exemplar;
 use crate::metrics::MetricType;
 use crate::registry::{Prefix, Registry, Unit};
@@ -460,6 +460,13 @@ impl MetricEncoder<'_> {
             }
             .into(),
         )?;
+        self.writer.write_char(' ')?;
+        exemplar.time.encode(
+            ExemplarValueEncoder {
+                writer: self.writer,
+            }
+            .into(),
+        )?;
         Ok(())
     }
 
@@ -737,6 +744,7 @@ mod tests {
     use std::borrow::Cow;
     use std::fmt::Error;
     use std::sync::atomic::{AtomicI32, AtomicU32};
+    use std::time::{SystemTime, UNIX_EPOCH};
 
     #[test]
     fn encode_counter() {
@@ -776,6 +784,8 @@ mod tests {
 
     #[test]
     fn encode_counter_with_exemplar() {
+        let now = SystemTime::now();
+
         let mut registry = Registry::default();
 
         let counter_with_exemplar: CounterWithExemplar<Vec<(String, u64)>> =
@@ -789,14 +799,24 @@ mod tests {
 
         counter_with_exemplar.inc_by(1, Some(vec![("user_id".to_string(), 42)]));
 
+        counter_with_exemplar
+            .inner
+            .write()
+            .exemplar
+            .as_mut()
+            .unwrap()
+            .time = now;
+
         let mut encoded = String::new();
         encode(&mut encoded, &registry).unwrap();
 
         let expected = "# HELP my_counter_with_exemplar_seconds My counter with exemplar.\n"
             .to_owned()
             + "# TYPE my_counter_with_exemplar_seconds counter\n"
             + "# UNIT my_counter_with_exemplar_seconds seconds\n"
-            + "my_counter_with_exemplar_seconds_total 1 # {user_id=\"42\"} 1.0\n"
+            + "my_counter_with_exemplar_seconds_total 1 # {user_id=\"42\"} 1.0 "
+            + &dtoa::Buffer::new().format(now.duration_since(UNIX_EPOCH).unwrap().as_secs_f64())
+            + "\n"
             + "# EOF\n";
         assert_eq!(expected, encoded);
 
@@ -953,19 +973,32 @@ mod tests {
 
     #[test]
     fn encode_histogram_with_exemplars() {
+        let now = SystemTime::now();
+
         let mut registry = Registry::default();
         let histogram = HistogramWithExemplars::new(exponential_buckets(1.0, 2.0, 10));
         registry.register("my_histogram", "My histogram", histogram.clone());
         histogram.observe(1.0, Some([("user_id".to_string(), 42u64)]));
 
+        histogram
+            .inner
+            .write()
+            .exemplars
+            .get_mut(&0)
+            .as_mut()
+            .unwrap()
+            .time = now;
+
         let mut encoded = String::new();
         encode(&mut encoded, &registry).unwrap();
 
         let expected = "# HELP my_histogram My histogram.\n".to_owned()
             + "# TYPE my_histogram histogram\n"
             + "my_histogram_sum 1.0\n"
             + "my_histogram_count 1\n"
-            + "my_histogram_bucket{le=\"1.0\"} 1 # {user_id=\"42\"} 1.0\n"
+            + "my_histogram_bucket{le=\"1.0\"} 1 # {user_id=\"42\"} 1.0 "
+            + &dtoa::Buffer::new().format(now.duration_since(UNIX_EPOCH).unwrap().as_secs_f64())
+            + "\n"
             + "my_histogram_bucket{le=\"2.0\"} 1\n"
             + "my_histogram_bucket{le=\"4.0\"} 1\n"
             + "my_histogram_bucket{le=\"8.0\"} 1\n"

src/metrics/exemplar.rs

@@ -16,12 +16,14 @@ use std::sync::atomic::AtomicU32;
 #[cfg(target_has_atomic = "64")]
 use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
+use std::time::SystemTime;
 
 /// An OpenMetrics exemplar.
 #[derive(Debug)]
 pub struct Exemplar<S, V> {
     pub(crate) label_set: S,
     pub(crate) value: V,
+    pub(crate) time: SystemTime,
 }
 
 /////////////////////////////////////////////////////////////////////////////////
@@ -121,6 +123,7 @@ impl<S, N: Clone, A: counter::Atomic<N>> CounterWithExemplar<S, N, A> {
         inner.exemplar = label_set.map(|label_set| Exemplar {
             label_set,
             value: v.clone(),
+            time: SystemTime::now(),
         });
 
         inner.counter.inc_by(v)
@@ -256,6 +259,7 @@ impl<S> HistogramWithExemplars<S> {
                 Exemplar {
                     label_set,
                     value: v,
+                    time: SystemTime::now(),
                 },
             );
         }