feat(metrics): Add a metrics aggregator

CHANGELOG.md

@@ -10,9 +10,8 @@
 **Internal**:
 
 - Emit the `quantity` field for outcomes of events. This field describes the total size in bytes for attachments or the event count for all other categories. A separate outcome is emitted for attachments in a rejected envelope, if any, in addition to the event outcome. ([#942](https://github.com/getsentry/relay/pull/942))
-- Add experimental metrics ingestion without bucketing or pre-aggregation. ([#948](https://github.com/getsentry/relay/pull/948))
+- Add experimental metrics ingestion with bucketing or pre-aggregation. ([#948](https://github.com/getsentry/relay/pull/948), [#952](https://github.com/getsentry/relay/pull/952), [#958](https://github.com/getsentry/relay/pull/958))
 - Change HTTP response for upstream timeouts from 502 to 504. ([#859](https://github.com/getsentry/relay/pull/859))
-- Disallow empty metric names, require alphabetic start. ([#952](https://github.com/getsentry/relay/pull/952))
 - Add rule id to outcomes coming from transaction sampling. ([#953](https://github.com/getsentry/relay/pull/953))
 - Add support for breakdowns ingestion. ([#934](https://github.com/getsentry/relay/pull/934))
 

relay-common/src/time.rs

@@ -15,8 +15,22 @@ pub fn instant_to_date_time(instant: Instant) -> chrono::DateTime<chrono::Utc> {
     instant_to_system_time(instant).into()
 }
 
-/// A unix timestap (full seconds elapsed since 1970).
-#[derive(Clone, Copy, Eq, Ord, PartialEq, PartialOrd)]
+/// The conversion result of [`UnixTimestamp::to_instant`].
+///
+/// If the time is outside of what can be represented in an [`Instant`], this is `Past` or
+/// `Future`.
+#[derive(Clone, Copy, Debug)]
+pub enum MonotonicResult {
+    /// A time before the earliest representable `Instant`.
+    Past,
+    /// A representable `Instant`.
+    Instant(Instant),
+    /// A time after the latest representable `Instant`.
+    Future,
+}
+
+/// A unix timestamp (full seconds elapsed since 1970-01-01 00:00 UTC).
+#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub struct UnixTimestamp(u64);
 
 impl UnixTimestamp {
@@ -54,6 +68,41 @@ impl UnixTimestamp {
     pub fn as_secs(self) -> u64 {
         self.0
     }
+
+    /// Converts the UNIX timestamp into an `Instant` based on the current system timestamp.
+    ///
+    /// Returns [`MonotonicResult::Instant`] if the timestamp can be represented. Otherwise, returns
+    /// [`MonotonicResult::Past`] or [`MonotonicResult::Future`].
+    ///
+    /// Note that the system time is subject to skew, so subsequent calls to `to_instant` may return
+    /// different values.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use std::time::{Duration, Instant};
+    /// use relay_common::{MonotonicResult, UnixTimestamp};
+    ///
+    /// let timestamp = UnixTimestamp::now();
+    /// if let MonotonicResult::Instant(instant) = timestamp.to_instant() {
+    ///    assert!((Instant::now() - instant) < Duration::from_millis(1));
+    /// }
+    /// ```
+    pub fn to_instant(self) -> MonotonicResult {
+        let now = Self::now();
+
+        if self > now {
+            match Instant::now().checked_add(self - now) {
+                Some(instant) => MonotonicResult::Instant(instant),
+                None => MonotonicResult::Future,
+            }
+        } else {
+            match Instant::now().checked_sub(now - self) {
+                Some(instant) => MonotonicResult::Instant(instant),
+                None => MonotonicResult::Past,
+            }
+        }
+    }
 }
 
 impl fmt::Debug for UnixTimestamp {

relay-metrics/Cargo.toml

@@ -10,9 +10,14 @@ license-file = "../LICENSE"
 publish = false
 
 [dependencies]
+actix = "0.7.9"
+hash32 = "0.1.1"
 relay-common = { path = "../relay-common" }
+relay-log = { path = "../relay-log" }
 serde = { version = "1.0.114", features = ["derive"] }
+serde_json = "1.0.55"
 
 [dev-dependencies]
+futures = "0.1.28"
 insta = "1.1.0"
-serde_json = "1.0.55"
+relay-test = { path = "../relay-test" }