ref(check-ins): Send checkins without envelope to store

CHANGELOG.md

@@ -37,6 +37,7 @@
 - Update sentry-conventions to 0.13.0. ([#6139](https://github.com/getsentry/relay/pull/6139))
 - Upgrade release image to Debian 13. ([#6110](https://github.com/getsentry/relay/pull/6110))
 - No longer serialize transactions into envelopes when storing them. ([#6193](https://github.com/getsentry/relay/pull/6193))
+- No longer serialize check-ins into envelopes when storing them. ([#6196](https://github.com/getsentry/relay/pull/6196))
 - Prefix upload location query params for forward compatibility. ([#6076](https://github.com/getsentry/relay/pull/6076))
 - Add config option to bypass the kafka fallback for objectstore uploads. ([#6127](https://github.com/getsentry/relay/pull/6127))
 - Use upstream descriptor in upload requests. ([#6128](https://github.com/getsentry/relay/pull/6128))

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

@@ -121,10 +121,18 @@ impl Forward for CheckInsOutput {
         s: processing::StoreHandle<'_>,
         ctx: processing::ForwardContext<'_>,
     ) -> Result<(), Rejected<()>> {
-        let envelope = self.serialize_envelope(ctx)?;
-        let envelope = ManagedEnvelope::from(envelope);
+        use crate::services::store::StoreCheckIn;
 
-        s.send_to_store(crate::services::store::StoreEnvelope { envelope });
+        let sdk = self.0.headers.meta().client().map(str::to_owned);
+        let retention_days = ctx.event_retention().standard;
+
+        for check_in in self.0.split(|work| work.check_ins.into_iter()) {
+            s.send_to_store(check_in.map(|check_in, _| StoreCheckIn {
+                check_in,
+                sdk: sdk.clone(),
+                retention_days,
+            }));
+        }
 
         Ok(())
     }

relay-server/src/services/store.rs

@@ -283,6 +283,23 @@ impl Counted for StoreProfile {
     }
 }
 
+/// A monitor check-in to be stored to Kafka.
+#[derive(Debug)]
+pub struct StoreCheckIn {
+    /// The serialized check-in.
+    pub check_in: Item,
+    /// The SDK client which produced the check-in.
+    pub sdk: Option<String>,
+    /// Check-in retention in days.
+    pub retention_days: u16,
+}
+
+impl Counted for StoreCheckIn {
+    fn quantities(&self) -> Quantities {
+        self.check_in.quantities()
+    }
+}
+
 /// The asynchronous thread pool used for scheduling storing tasks in the envelope store.
 pub type StoreServicePool = AsyncPool<StoreTask>;
 
@@ -315,6 +332,8 @@ pub enum Store {
     UserReport(Managed<StoreUserReport>),
     /// A single profile.
     Profile(Managed<StoreProfile>),
+    /// A singular monitor check-in.
+    CheckIn(Managed<StoreCheckIn>),
 }
 
 impl Store {
@@ -331,6 +350,7 @@ impl Store {
             Store::Attachment(_) => "attachment",
             Store::UserReport(_) => "user_report",
             Store::Profile(_) => "profile",
+            Store::CheckIn(_) => "check_in",
         }
     }
 }
@@ -417,6 +437,14 @@ impl FromMessage<Managed<StoreProfile>> for Store {
     }
 }
 
+impl FromMessage<Managed<StoreCheckIn>> for Store {
+    type Response = NoResponse;
+
+    fn from_message(message: Managed<StoreCheckIn>, _: ()) -> Self {
+        Self::CheckIn(message)
+    }
+}
+
 /// Service implementing the [`Store`] interface.
 pub struct StoreService {
     pool: StoreServicePool,
@@ -460,6 +488,7 @@ impl StoreService {
                 Store::Attachment(message) => self.handle_store_attachment(message),
                 Store::UserReport(message) => self.handle_user_report(message),
                 Store::Profile(message) => self.handle_profile(message),
+                Store::CheckIn(message) => self.handle_check_in(message),
             };
             if let Err(error) = result {
                 relay_log::error!(
@@ -630,18 +659,7 @@ impl StoreService {
                     retention,
                     item,
                 )?,
-                ItemType::CheckIn => {
-                    let client = envelope.meta().client();
-                    self.produce_check_in(
-                        scoping.project_id,
-                        scoping.organization_id,
-                        received_at,
-                        client,
-                        retention,
-                        item,
-                    )?
-                }
-                ty @ (ItemType::Log | ItemType::Span) => {
+                ty @ (ItemType::Log | ItemType::Span | ItemType::CheckIn) => {
                     debug_assert!(
                         false,
                         "received {ty} through an envelope, \
@@ -987,6 +1005,26 @@ impl StoreService {
         })
     }
 
+    fn handle_check_in(&self, message: Managed<StoreCheckIn>) -> Result<(), Rejected<StoreError>> {
+        let scoping = message.scoping();
+        let received_at = message.received_at();
+
+        message.try_accept(|check_in| {
+            let message = KafkaMessage::CheckIn(CheckInKafkaMessage {
+                message_type: CheckInMessageType::CheckIn,
+                project_id: scoping.project_id,
+                org_id: scoping.organization_id,
+                retention_days: check_in.retention_days,
+                start_time: safe_timestamp(received_at),
+                sdk: check_in.sdk,
+                payload: check_in.check_in.payload(),
+                routing_key_hint: check_in.check_in.routing_hint(),
+            });
+
+            self.produce(KafkaTopic::Monitors, message)
+        })
+    }
+
     fn create_metric_message<'a>(
         &self,
         scoping: &Scoping,
@@ -1355,31 +1393,6 @@ impl StoreService {
         self.produce(KafkaTopic::Profiles, KafkaMessage::Profile(message))?;
         Ok(())
     }
-
-    fn produce_check_in(
-        &self,
-        project_id: ProjectId,
-        org_id: OrganizationId,
-        received_at: DateTime<Utc>,
-        client: Option<&str>,
-        retention_days: u16,
-        item: &Item,
-    ) -> Result<(), StoreError> {
-        let message = KafkaMessage::CheckIn(CheckInKafkaMessage {
-            message_type: CheckInMessageType::CheckIn,
-            project_id,
-            retention_days,
-            start_time: safe_timestamp(received_at),
-            sdk: client.map(str::to_owned),
-            payload: item.payload(),
-            routing_key_hint: item.routing_hint(),
-            org_id,
-        });
-
-        self.produce(KafkaTopic::Monitors, message)?;
-
-        Ok(())
-    }
 }
 
 impl Service for StoreService {