getsentry · iambriccardo · May 22, 2024 · May 7, 2024 · May 7, 2024 · May 7, 2024
@@ -21,6 +21,7 @@
 - Emit negative outcomes for denied metrics. ([#3508](https://github.com/getsentry/relay/pull/3508))
 - Increase size limits for internal batch endpoints. ([#3562](https://github.com/getsentry/relay/pull/3562))
 - Emit negative outcomes when metrics are rejected because of a disabled namespace. ([#3544](https://github.com/getsentry/relay/pull/3544))
+- Forward `received_at` timestamp for buckets sent to Kafka. ([#3561](https://github.com/getsentry/relay/pull/3561))
 
 ## 24.4.2
 

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -134,7 +134,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 }

@@ -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)]

@@ -0,0 +1,3 @@
+from .time import time_after, time_within, time_within_delta
+
+__all__ = ["time_after", "time_within", "time_within_delta"]
@@ -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"]