feat(metrics): Forward received_at field to Kafka (#3561)

getsentry · May 22, 2024 · 5b8198b · 5b8198b
1 parent 9d3c91c
commit 5b8198b
Show file tree

Hide file tree

Showing 7 changed files with 154 additions and 35 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,7 @@
 - Send microsecond precision timestamps. ([#3613](https://github.com/getsentry/relay/pull/3613))
 - Map outcome reasons for dynamic sampling to reduced set of values. ([#3623](https://github.com/getsentry/relay/pull/3623))
 - Extract status for spans. ([#3606](https://github.com/getsentry/relay/pull/3606))
+- Forward `received_at` timestamp for buckets sent to Kafka. ([#3561](https://github.com/getsentry/relay/pull/3561))
 
 ## 24.5.0
 

diff --git a/relay-server/src/services/store.rs b/relay-server/src/services/store.rs
@@ -469,8 +469,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,
@@ -479,6 +477,7 @@ impl StoreService {
             timestamp: view.timestamp(),
             tags: view.tags(),
             retention_days,
+            received_at: view.metadata().received_at,
         })
     }
 
@@ -1253,6 +1252,8 @@ struct MetricKafkaMessage<'a> {
     timestamp: UnixTimestamp,
     tags: &'a BTreeMap<String, String>,
     retention_days: u16,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    received_at: Option<UnixTimestamp>,
 }
 
 #[derive(Clone, Debug, Serialize)]

diff --git a/tests/integration/asserts/__init__.py b/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"]
diff --git a/tests/integration/asserts/time.py b/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)
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,3 @@
		from .time import time_after, time_within, time_within_delta

		__all__ = ["time_after", "time_within", "time_within_delta"]