feat(metrics): Add received_at timestamp in bucket metadata

relay-config/src/config.rs

@@ -524,13 +524,17 @@ struct SentryMetrics {
     ///
     /// Defaults to 5.
     pub meta_locations_max: usize,
+    /// Whether to override the [`received_at`] field in the [`BucketMetadata`] with the current
+    /// receive time of the instance.
+    pub override_received_at_metadata: bool,
 }
 
 impl Default for SentryMetrics {
     fn default() -> Self {
         Self {
             meta_locations_expiry: 15 * 24 * 60 * 60,
             meta_locations_max: 5,
+            override_received_at_metadata: false,
         }
     }
 }
@@ -1947,6 +1951,11 @@ impl Config {
         Duration::from_secs(self.values.sentry_metrics.meta_locations_expiry)
     }
 
+    /// Returns whether we want to override the [`received_at`] field of [`BucketMetadata`].
+    pub fn metrics_override_received_at_metadata(&self) -> bool {
+        self.values.sentry_metrics.override_received_at_metadata
+    }
+
     /// Returns the default timeout for all upstream HTTP requests.
     pub fn http_timeout(&self) -> Duration {
         Duration::from_secs(self.values.http.timeout.into())

relay-metrics/src/aggregator.rs

@@ -899,14 +899,15 @@ mod tests {
         }
     }
 
-    fn some_bucket() -> Bucket {
+    fn some_bucket(timestamp: Option<UnixTimestamp>) -> Bucket {
+        let timestamp = timestamp.map_or(UnixTimestamp::from_secs(999994711), |t| t);
         Bucket {
-            timestamp: UnixTimestamp::from_secs(999994711),
+            timestamp,
             width: 0,
             name: "c:transactions/foo".into(),
             value: BucketValue::counter(42.into()),
             tags: BTreeMap::new(),
-            metadata: BucketMetadata::new(),
+            metadata: BucketMetadata::new(timestamp),
         }
     }
 
@@ -916,7 +917,7 @@ mod tests {
         let project_key = ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap();
         let mut aggregator = Aggregator::new(test_config());
 
-        let bucket1 = some_bucket();
+        let bucket1 = some_bucket(None);
 
         let mut bucket2 = bucket1.clone();
         bucket2.value = BucketValue::counter(43.into());
@@ -1006,7 +1007,7 @@ mod tests {
         let project_key = ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap();
         let mut aggregator = Aggregator::new(config);
 
-        let bucket1 = some_bucket();
+        let bucket1 = some_bucket(None);
 
         let mut bucket2 = bucket1.clone();
         bucket2.timestamp = UnixTimestamp::from_secs(999994712);
@@ -1069,8 +1070,12 @@ mod tests {
         let mut aggregator = Aggregator::new(config);
 
         // It's OK to have same metric with different projects:
-        aggregator.merge(project_key1, some_bucket(), None).unwrap();
-        aggregator.merge(project_key2, some_bucket(), None).unwrap();
+        aggregator
+            .merge(project_key1, some_bucket(None), None)
+            .unwrap();
+        aggregator
+            .merge(project_key2, some_bucket(None), None)
+            .unwrap();
 
         assert_eq!(aggregator.buckets.len(), 2);
     }
@@ -1151,13 +1156,14 @@ mod tests {
         let mut aggregator = Aggregator::new(test_config());
         let project_key = ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fed").unwrap();
 
+        let timestamp = UnixTimestamp::from_secs(999994711);
         let bucket = Bucket {
-            timestamp: UnixTimestamp::from_secs(999994711),
+            timestamp,
             width: 0,
             name: "c:transactions/foo@none".into(),
             value: BucketValue::counter(42.into()),
             tags: BTreeMap::new(),
-            metadata: BucketMetadata::new(),
+            metadata: BucketMetadata::new(timestamp),
         };
         let bucket_key = BucketKey {
             project_key,
@@ -1407,13 +1413,14 @@ mod tests {
 
     #[test]
     fn test_aggregator_cost_enforcement_total() {
+        let timestamp = UnixTimestamp::from_secs(999994711);
         let bucket = Bucket {
-            timestamp: UnixTimestamp::from_secs(999994711),
+            timestamp,
             width: 0,
             name: "c:transactions/foo".into(),
             value: BucketValue::counter(42.into()),
             tags: BTreeMap::new(),
-            metadata: BucketMetadata::new(),
+            metadata: BucketMetadata::new(timestamp),
         };
 
         let mut aggregator = Aggregator::new(test_config());
@@ -1438,13 +1445,14 @@ mod tests {
         let mut config = test_config();
         config.max_project_key_bucket_bytes = Some(1);
 
+        let timestamp = UnixTimestamp::from_secs(999994711);
         let bucket = Bucket {
-            timestamp: UnixTimestamp::from_secs(999994711),
+            timestamp,
             width: 0,
             name: "c:transactions/foo".into(),
             value: BucketValue::counter(42.into()),
             tags: BTreeMap::new(),
-            metadata: BucketMetadata::new(),
+            metadata: BucketMetadata::new(timestamp),
         };
 
         let mut aggregator = Aggregator::new(config);
@@ -1475,23 +1483,36 @@ mod tests {
         let project_key = ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap();
         let mut aggregator = Aggregator::new(config);
 
-        let bucket1 = some_bucket();
-        let bucket2 = some_bucket();
+        let bucket1 = some_bucket(Some(UnixTimestamp::from_secs(999994711)));
+        let bucket2 = some_bucket(Some(UnixTimestamp::from_secs(999994711)));
 
-        // Create a bucket with already 3 merges.
-        let mut bucket3 = some_bucket();
-        bucket3.metadata.merge(BucketMetadata::new());
-        bucket3.metadata.merge(BucketMetadata::new());
+        // We create a bucket with 3 merges and monotonically increasing timestamps.
+        let mut bucket3 = some_bucket(Some(UnixTimestamp::from_secs(999994711)));
+        bucket3
+            .metadata
+            .merge(BucketMetadata::new(UnixTimestamp::from_secs(999997811)));
+        bucket3
+            .metadata
+            .merge(BucketMetadata::new(UnixTimestamp::from_secs(999999811)));
 
-        aggregator.merge(project_key, bucket1, None).unwrap();
-        aggregator.merge(project_key, bucket2, None).unwrap();
-        aggregator.merge(project_key, bucket3, None).unwrap();
+        aggregator
+            .merge(project_key, bucket1.clone(), None)
+            .unwrap();
+        aggregator
+            .merge(project_key, bucket2.clone(), None)
+            .unwrap();
+        aggregator
+            .merge(project_key, bucket3.clone(), None)
+            .unwrap();
 
-        let buckets: Vec<_> = aggregator.buckets.values().map(|v| &v.metadata).collect();
-        insta::assert_debug_snapshot!(buckets, @r###"
+        let buckets_metadata: Vec<_> = aggregator.buckets.values().map(|v| &v.metadata).collect();
+        insta::assert_debug_snapshot!(buckets_metadata, @r###"
         [
             BucketMetadata {
                 merges: 5,
+                received_at: Some(
+                    UnixTimestamp(999994711),
+                ),
             },
         ]
         "###);

relay-metrics/src/aggregatorservice.rs

@@ -471,13 +471,14 @@ mod tests {
     }
 
     fn some_bucket() -> Bucket {
+        let timestamp = UnixTimestamp::from_secs(999994711);
         Bucket {
-            timestamp: UnixTimestamp::from_secs(999994711),
+            timestamp,
             width: 0,
             name: "c:transactions/foo".into(),
             value: BucketValue::counter(42.into()),
             tags: BTreeMap::new(),
-            metadata: BucketMetadata::new(),
+            metadata: BucketMetadata::new(timestamp),
         }
     }
 

relay-metrics/src/bucket.rs

@@ -1,3 +1,4 @@
+use std::cmp::min;
 use std::collections::{BTreeMap, BTreeSet};
 use std::hash::Hash;
 use std::iter::FusedIterator;
@@ -502,7 +503,7 @@ pub struct Bucket {
     /// # Statsd Format
     ///
     /// In statsd, timestamps are part of the `|`-separated list following values. Timestamps start
-    /// with with the literal character `'T'` followed by the UNIX timestamp.
+    /// with the literal character `'T'` followed by the UNIX timestamp.
     ///
     /// The timestamp must be a positive integer in decimal notation representing the value of the
     /// UNIX timestamp.
@@ -748,33 +749,55 @@ pub struct BucketMetadata {
     /// For example: Merging two un-merged buckets will yield a total
     /// of `2` merges.
     pub merges: NonZeroU32,
+
+    /// Received timestamp of the first metric in this bucket.
+    ///
+    /// This field should be set to the time in which the first metric of a specific bucket was
+    /// received in the outermost internal Relay.
+    pub received_at: Option<UnixTimestamp>,
 }
 
 impl BucketMetadata {
     /// Creates a fresh metadata instance.
     ///
-    /// The new metadata is initialized with `1` merge.
-    pub fn new() -> Self {
+    /// The new metadata is initialized with `1` merge and a given [`received_at`] timestamp.
+    pub fn new(received_at: UnixTimestamp) -> Self {
         Self {
             merges: NonZeroU32::MIN,
+            received_at: Some(received_at),
         }
     }
 
     /// Whether the metadata does not contain more information than the default.
     pub fn is_default(&self) -> bool {
-        let Self { merges } = self;
-        *merges == NonZeroU32::MIN
+        let Self {
+            merges,
+            received_at,
+        } = self;
+
+        *merges == NonZeroU32::MIN && received_at.is_none()
     }
 
     /// Merges another metadata object into the current one.
     pub fn merge(&mut self, other: Self) {
         self.merges = self.merges.saturating_add(other.merges.get());
+        self.received_at = match (self.received_at, other.received_at) {
+            (Some(received_at), None) => Some(received_at),
+            (None, Some(received_at)) => Some(received_at),
+            (Some(left_received_at), Some(right_received_at)) => {
+                Some(min(left_received_at, right_received_at))
+            }
+            (None, None) => None,
+        };
     }
 }
 
 impl Default for BucketMetadata {
     fn default() -> Self {
-        Self::new()
+        Self {
+            merges: NonZeroU32::MIN,
+            received_at: None,
+        }
     }
 }
 
@@ -907,6 +930,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -940,6 +966,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -991,6 +1020,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -1050,6 +1082,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -1079,6 +1114,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -1102,6 +1140,9 @@ mod tests {
             tags: {},
             metadata: BucketMetadata {
                 merges: 1,
+                received_at: Some(
+                    UnixTimestamp(4711),
+                ),
             },
         }
         "###);
@@ -1242,7 +1283,10 @@ mod tests {
             .collect::<Result<Vec<_>, _>>()
             .unwrap();
 
-        let json_metrics: Vec<Bucket> = serde_json::from_str(json).unwrap();
+        let mut json_metrics: Vec<Bucket> = serde_json::from_str(json).unwrap();
+        for json_metric in &mut json_metrics {
+            json_metric.metadata.received_at = Some(timestamp);
+        }
 
         assert_eq!(statsd_metrics, json_metrics);
     }
@@ -1254,7 +1298,8 @@ mod tests {
 
         let timestamp = UnixTimestamp::from_secs(1615889449);
         let statsd_metric = Bucket::parse(text.as_bytes(), timestamp).unwrap();
-        let json_metric: Bucket = serde_json::from_str(json).unwrap();
+        let mut json_metric: Bucket = serde_json::from_str(json).unwrap();
+        json_metric.metadata.received_at = Some(timestamp);
 
         assert_eq!(statsd_metric, json_metric);
     }
@@ -1275,7 +1320,9 @@ mod tests {
           }
         ]"#;
 
-        let buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
+        let mut buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
+        buckets[0].metadata = BucketMetadata::new(UnixTimestamp::from_secs(1615889440));
+
         insta::assert_debug_snapshot!(buckets, @r###"
         [
             Bucket {
@@ -1297,6 +1344,9 @@ mod tests {
                 },
                 metadata: BucketMetadata {
                     merges: 1,
+                    received_at: Some(
+                        UnixTimestamp(1615889440),
+                    ),
                 },
             },
         ]
@@ -1315,7 +1365,9 @@ mod tests {
           }
         ]"#;
 
-        let buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
+        let mut buckets = serde_json::from_str::<Vec<Bucket>>(json).unwrap();
+        buckets[0].metadata = BucketMetadata::new(UnixTimestamp::from_secs(1615889440));
+
         insta::assert_debug_snapshot!(buckets, @r###"
         [
             Bucket {
@@ -1330,6 +1382,9 @@ mod tests {
                 tags: {},
                 metadata: BucketMetadata {
                     merges: 1,
+                    received_at: Some(
+                        UnixTimestamp(1615889440),
+                    ),
                 },
             },
         ]