feat(spanv2): Implement store retentions (#5252)

Considered passing `Retention` to the `store`, but then I feel like
`Retention` should be a type 'owned' by the store and we should refactor
all usages. So might keep that for a follow-up and just keep it
consistent with the rest.

Author: Dav1dde

relay-server/src/processing/spans/mod.rs

@@ -188,7 +188,7 @@ impl Forward for SpanOutput {
     fn forward_store(
         self,
         s: &relay_system::Addr<crate::services::store::Store>,
-        _: processing::ForwardContext<'_>,
+        ctx: processing::ForwardContext<'_>,
     ) -> Result<(), Rejected<()>> {
         let spans = match self {
             SpanOutput::NotProcessed(spans) => {
@@ -199,17 +199,12 @@ impl Forward for SpanOutput {
             SpanOutput::Processed(spans) => spans,
         };
 
-        let (spans, server_sample_rate) =
-            spans.split_with_context(|spans| (spans.spans, spans.server_sample_rate));
-
         let ctx = store::Context {
-            server_sample_rate,
-            // TODO: retentions still need to be taken from the project info.
-            retention_days: crate::constants::DEFAULT_EVENT_RETENTION,
-            downsampled_retention_days: crate::constants::DEFAULT_EVENT_RETENTION,
+            server_sample_rate: spans.server_sample_rate,
+            retention: ctx.retention(|r| r.span.as_ref()),
         };
 
-        for span in spans {
+        for span in spans.split(|spans| spans.spans) {
             if let Ok(span) = span.try_map(|span, _| store::convert(span, &ctx)) {
                 s.send(span)
             };

relay-server/src/processing/spans/store.rs

@@ -4,6 +4,7 @@ use relay_event_schema::protocol::SpanV2;
 use relay_protocol::{Annotated, FiniteF64};
 
 use crate::envelope::WithHeader;
+use crate::processing::Retention;
 use crate::processing::spans::{Error, Result};
 use crate::services::outcome::DiscardReason;
 use crate::services::store::StoreSpanV2;
@@ -25,12 +26,10 @@ macro_rules! required {
 
 /// Context required for [`convert`].
 pub struct Context {
-    /// Span retention in days.
-    pub retention_days: u16,
-    /// Span downsampled retention in day.
-    pub downsampled_retention_days: u16,
     /// Server side applied sample rate.
     pub server_sample_rate: Option<f64>,
+    /// Item retention.
+    pub retention: Retention,
 }
 
 /// Converts a processed [`SpanV2`] into a [Kafka](crate::services::store::Store) compatible format.
@@ -43,8 +42,8 @@ pub fn convert(span: WithHeader<SpanV2>, ctx: &Context) -> Result<Box<StoreSpanV
 
     Ok(Box::new(StoreSpanV2 {
         routing_key,
-        retention_days: ctx.retention_days,
-        downsampled_retention_days: ctx.downsampled_retention_days,
+        retention_days: ctx.retention.standard,
+        downsampled_retention_days: ctx.retention.downsampled,
         item: span,
     }))
 }

tests/integration/test_spansv2.py

@@ -47,10 +47,15 @@ def test_spansv2_basic(
 
     project_id = 42
     project_config = mini_sentry.add_full_project_config(project_id)
-    project_config["config"]["features"] = [
-        "organizations:standalone-span-ingestion",
-        "projects:span-v2-experimental-processing",
-    ]
+    project_config["config"].update(
+        {
+            "features": [
+                "organizations:standalone-span-ingestion",
+                "projects:span-v2-experimental-processing",
+            ],
+            "retentions": {"span": {"standard": 42, "downsampled": 1337}},
+        }
+    )
 
     relay = relay(relay_with_processing(options=TEST_CONFIG), options=TEST_CONFIG)
 
@@ -102,8 +107,8 @@ def test_spansv2_basic(
         "end_timestamp": time_within(ts.timestamp() + 0.5),
         "is_remote": False,
         "status": "ok",
-        "retention_days": 90,
-        "downsampled_retention_days": 90,
+        "retention_days": 42,
+        "downsampled_retention_days": 1337,
         "key_id": 123,
         "organization_id": 1,
         "project_id": 42,