feat(metrics): Forward received_at field to Kafka

Cargo.toml

@@ -135,7 +135,7 @@ rust-embed = "8.0.0"
 schemars = { version = "=0.8.10", features = ["uuid1", "chrono"] }
 sentry = "0.32.2"
 sentry-core = "0.32.2"
-sentry-kafka-schemas = { version = "0.1.79", default-features = false }
+sentry-kafka-schemas = { version = "0.1.80", default-features = false }
 sentry-release-parser = { version = "1.3.2", default-features = false }
 sentry-types = "0.32.2"
 sentry_usage_accountant = { version = "0.1.0", default-features = false }

relay-server/src/services/store.rs

@@ -507,8 +507,6 @@ impl StoreService {
             BucketViewValue::Gauge(g) => MetricValue::Gauge(g),
         };
 
-        // TODO: propagate the `received_at` field upstream.
-        // https://github.com/getsentry/relay/issues/3515
         Ok(MetricKafkaMessage {
             org_id: organization_id,
             project_id,
@@ -517,6 +515,7 @@ impl StoreService {
             timestamp: view.timestamp(),
             tags: view.tags(),
             retention_days,
+            received_at: view.metadata().received_at,
         })
     }
 
@@ -1290,6 +1289,8 @@ struct MetricKafkaMessage<'a> {
     timestamp: UnixTimestamp,
     tags: &'a BTreeMap<String, String>,
     retention_days: u16,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    received_at: Option<UnixTimestamp>,
 }
 
 #[derive(Clone, Debug, Serialize)]

tests/integration/asserts/__init__.py

@@ -0,0 +1,3 @@
+from .time import time_after, time_within, time_within_delta
+
+__all__ = ["time_after", "time_within", "time_within_delta"]

tests/integration/asserts/time.py

@@ -0,0 +1,35 @@
+from datetime import timedelta, datetime, timezone
+
+
+class _WithinBounds:
+
+    def __init__(self, lower_bound, upper_bound):
+        self._lower_bound = lower_bound
+        self._upper_bound = upper_bound
+
+    def __eq__(self, other):
+        assert isinstance(other, int)
+        return self._lower_bound <= other <= self._upper_bound
+
+    def __str__(self):
+        return f"{self._lower_bound} <= x <= {self._upper_bound}"
+
+
+def time_after(lower_bound):
+    upper_bound = int(datetime.now(tz=timezone.utc).timestamp())
+    return time_within(lower_bound, upper_bound)
+
+
+def time_within(lower_bound, upper_bound):
+    assert lower_bound <= upper_bound
+    return _WithinBounds(lower_bound, upper_bound)
+
+
+def time_within_delta(timestamp, delta=None):
+    if delta is None:
+        delta = timedelta(seconds=5)
+
+    lower_bound = (datetime.fromtimestamp(timestamp) - delta).timestamp()
+    upper_bound = (datetime.fromtimestamp(timestamp) + delta).timestamp()
+
+    return _WithinBounds(lower_bound, upper_bound)