feat(monitors): Ingestion for basic check-in payloads

relay-common/src/constants.rs

@@ -120,6 +120,8 @@ pub enum DataCategory {
     /// This is the category for transaction payloads that were accepted and stored in full. In
     /// contrast, `transaction` only guarantees that metrics have been accepted for the transaction.
     TransactionIndexed = 9,
+    /// Cron monitor checkins.
+    Cron = 10,
     //
     // IMPORTANT: After adding a new entry to DataCategory, go to the `relay-cabi` subfolder and run
     // `make header` to regenerate the C-binding. This allows using the data category from Python.
@@ -145,6 +147,7 @@ impl DataCategory {
             "replay" => Self::Replay,
             "transaction_processed" => Self::TransactionProcessed,
             "transaction_indexed" => Self::TransactionIndexed,
+            "cron" => Self::Cron,
             _ => Self::Unknown,
         }
     }
@@ -163,6 +166,7 @@ impl DataCategory {
             Self::Replay => "replay",
             Self::TransactionProcessed => "transaction_processed",
             Self::TransactionIndexed => "transaction_indexed",
+            Self::Cron => "cron",
             Self::Unknown => "unknown",
         }
     }

relay-config/src/config.rs

@@ -524,6 +524,8 @@ struct Limits {
     max_attachments_size: ByteSize,
     /// The maximum combined size for all client reports in an envelope or request.
     max_client_reports_size: ByteSize,
+    /// The maximum payload size for a cron monitor checkin.
+    max_checkin_size: ByteSize,
     /// The maximum payload size for an entire envelopes. Individual limits still apply.
     max_envelope_size: ByteSize,
     /// The maximum number of session items per envelope.
@@ -571,6 +573,7 @@ impl Default for Limits {
             max_attachment_size: ByteSize::mebibytes(100),
             max_attachments_size: ByteSize::mebibytes(100),
             max_client_reports_size: ByteSize::kibibytes(4),
+            max_checkin_size: ByteSize::kibibytes(100),
             max_envelope_size: ByteSize::mebibytes(100),
             max_session_count: 100,
             max_api_payload_size: ByteSize::mebibytes(20),
@@ -1690,6 +1693,11 @@ impl Config {
         self.values.limits.max_client_reports_size.as_bytes()
     }
 
+    /// Returns the maxmium payload size of a cron monitor checkin in bytes.
+    pub fn max_checkin_size(&self) -> usize {
+        self.values.limits.max_checkin_size.as_bytes()
+    }
+
     /// Returns the maximum size of an envelope payload in bytes.
     ///
     /// Individual item size limits still apply.

relay-crons/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "relay-crons"
+authors = ["Sentry <oss@sentry.io>"]
+description = "Crons processing for Relay"
+homepage = "https://getsentry.github.io/relay/"
+repository = "https://github.com/getsentry/relay"
+version = "23.2.0"
+edition = "2021"
+license-file = "../LICENSE"
+publish = false
+
+[dependencies]
+relay-common = { path = "../relay-common" }
+serde = { version = "1.0.114", features = ["derive"] }
+serde_json = "1.0.55"
+thiserror = "1.0.38"
+
+[dev-dependencies]
+similar-asserts = "1.4.2"

relay-crons/src/lib.rs

@@ -0,0 +1,108 @@
+//! Crons protocol and processing for Sentry.
+//!
+//! [Cron Monitors] allow you to monitor the uptime and performance of any scheduled, recurring job
+//! in Sentry. Once implemented, it'll allow you to get alerts and metrics to help you solve errors,
+//! detect timeouts, and prevent disruptions to your service.
+//!
+//! # API
+//!
+//! The public API documentation is available on [Sentry Docs](https://docs.sentry.io/api/crons/).
+//!
+//! [cron monitors]: https://docs.sentry.io/product/crons/
+
+#![doc(
+    html_logo_url = "https://raw.githubusercontent.com/getsentry/relay/master/artwork/relay-icon.png",
+    html_favicon_url = "https://raw.githubusercontent.com/getsentry/relay/master/artwork/relay-icon.png"
+)]
+#![warn(missing_docs)]
+
+use relay_common::Uuid;
+use serde::{Deserialize, Serialize};
+
+/// Error returned from [`process_checkin`].
+#[derive(Debug, thiserror::Error)]
+pub enum ProcessCheckinError {
+    /// Failed to deserialize the payload.
+    #[error("failed to deserialize checkin")]
+    Json(#[from] serde_json::Error),
+}
+
+///
+#[derive(Clone, Copy, Debug, PartialEq, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum CheckinStatus {
+    /// Checkin had no issues during execution.
+    Ok,
+    /// Checkin failed or otherwise had some issues.
+    Error,
+    /// Checkin is expectred to complete.
+    InProgress,
+    /// Monitor did not check in on time.
+    Missed,
+    /// No status was passed.
+    #[serde(other)]
+    Unknown,
+}
+
+fn uuid_simple<S>(uuid: &Uuid, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    uuid.as_simple().serialize(serializer)
+}
+
+/// The cron monitor checkin payload.
+#[derive(Debug, Deserialize, Serialize)]
+struct Checkin {
+    /// Unique identifier of this checkin.
+    #[serde(serialize_with = "uuid_simple")]
+    checkin_id: Uuid, // TODO(ja): roundtrip without slashes
+
+    /// Identifier of the monitor for this checkin.
+    #[serde(serialize_with = "uuid_simple")]
+    monitor_id: Uuid,
+
+    /// Status of this checkin. Defaults to `"unknown"`.
+    status: CheckinStatus,
+
+    /// Duration of this check since it has started in seconds.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    duration: Option<f64>,
+}
+
+/// Normalizes a cron monitor checkin payload.
+///
+/// Returns `None` if the payload was valid and does not have to be changed. Returns `Some` for
+/// valid payloads that were normalized.
+pub fn process_checkin(payload: &[u8]) -> Result<Vec<u8>, ProcessCheckinError> {
+    let mut checkin = serde_json::from_slice::<Checkin>(payload)?;
+
+    // Missed status cannot be ingested, this is computed on the server.
+    if checkin.status == CheckinStatus::Missed {
+        checkin.status = CheckinStatus::Unknown;
+    }
+
+    let serialized = serde_json::to_vec(&checkin)?;
+    Ok(serialized)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use similar_asserts::assert_eq;
+
+    #[test]
+    fn test_json_roundtrip() {
+        let json = r#"{
+  "checkin_id": "a460c25ff2554577b920fcfacae4e5eb",
+  "monitor_id": "4dc8556e039245c7bd569f8cf513ea42",
+  "status": "in_progress",
+  "duration": 21.0
+}"#;
+
+        let checkin = serde_json::from_str::<Checkin>(json).unwrap();
+        let serialized = serde_json::to_string_pretty(&checkin).unwrap();
+
+        assert_eq!(json, serialized);
+    }
+}

relay-kafka/src/config.rs

@@ -39,14 +39,16 @@ pub enum KafkaTopic {
     ReplayEvents,
     /// ReplayRecordings, large blobs sent by the replay sdk
     ReplayRecordings,
+    /// Cron monitor checkins.
+    Crons,
 }
 
 impl KafkaTopic {
     /// Returns iterator over the variants of [`KafkaTopic`].
     /// It will have to be adjusted if the new variants are added.
     pub fn iter() -> std::slice::Iter<'static, Self> {
         use KafkaTopic::*;
-        static TOPICS: [KafkaTopic; 11] = [
+        static TOPICS: [KafkaTopic; 12] = [
             Events,
             Attachments,
             Transactions,
@@ -58,6 +60,7 @@ impl KafkaTopic {
             Profiles,
             ReplayEvents,
             ReplayRecordings,
+            Crons,
         ];
         TOPICS.iter()
     }
@@ -93,6 +96,8 @@ pub struct TopicAssignments {
     pub replay_events: TopicAssignment,
     /// Recordings topic name.
     pub replay_recordings: TopicAssignment,
+    /// Cron monitor checkins.
+    pub crons: TopicAssignment,
 }
 
 impl TopicAssignments {
@@ -113,6 +118,7 @@ impl TopicAssignments {
             KafkaTopic::Profiles => &self.profiles,
             KafkaTopic::ReplayEvents => &self.replay_events,
             KafkaTopic::ReplayRecordings => &self.replay_recordings,
+            KafkaTopic::Crons => &self.crons,
         }
     }
 }
@@ -132,6 +138,7 @@ impl Default for TopicAssignments {
             profiles: "profiles".to_owned().into(),
             replay_events: "ingest-replay-events".to_owned().into(),
             replay_recordings: "ingest-replay-recordings".to_owned().into(),
+            crons: "ingest-crons".to_owned().into(),
         }
     }
 }

relay-quotas/src/quota.rs

@@ -107,7 +107,8 @@ impl CategoryUnit {
             | DataCategory::Security
             | DataCategory::Profile
             | DataCategory::TransactionProcessed
-            | DataCategory::TransactionIndexed => Some(Self::Count),
+            | DataCategory::TransactionIndexed
+            | DataCategory::Cron => Some(Self::Count),
             DataCategory::Attachment => Some(Self::Bytes),
             DataCategory::Session => Some(Self::Batched),
             DataCategory::Unknown => None,

relay-server/Cargo.toml

@@ -20,6 +20,7 @@ ssl = [
 ]
 processing = [
     "dep:minidump",
+    "dep:relay-crons",
     "dep:relay-profiling",
     "dep:symbolic-common",
     "dep:symbolic-unreal",
@@ -57,6 +58,7 @@ relay-auth = { path = "../relay-auth" }
 relay-aws-extension = { path = "../relay-aws-extension" }
 relay-common = { path = "../relay-common" }
 relay-config = { path = "../relay-config" }
+relay-crons = { path = "../relay-crons", optional = true }
 relay-filter = { path = "../relay-filter" }
 relay-general = { path = "../relay-general" }
 relay-kafka = { path = "../relay-kafka", optional = true }

relay-server/src/actors/processor.rs

@@ -1020,7 +1020,7 @@ impl EnvelopeProcessorService {
         }
     }
 
-    /// Remove profiles from the envelope if the feature flag is not enabled
+    /// Remove profiles from the envelope if the feature flag is not enabled.
     fn filter_profiles(&self, state: &mut ProcessEnvelopeState) {
         let profiling_enabled = state.project_state.has_feature(Feature::Profiling);
         state.envelope.retain_items(|item| match item.ty() {
@@ -1029,6 +1029,28 @@ impl EnvelopeProcessorService {
         });
     }
 
+    /// Normalize cron monitor checkins and remove invalid ones.
+    #[cfg(feature = "processing")]
+    fn process_checkins(&self, state: &mut ProcessEnvelopeState) {
+        state.envelope.retain_items(|item| {
+            if item.ty() != &ItemType::Checkin {
+                return true;
+            }
+
+            match relay_crons::process_checkin(&item.payload()) {
+                Ok(processed) => {
+                    item.set_payload(ContentType::Json, processed);
+                    true
+                }
+                Err(error) => {
+                    // TODO: Track an outcome.
+                    relay_log::debug!("dropped invalid cron checkin: {}", LogError(&error));
+                    false
+                }
+            }
+        })
+    }
+
     /// Process profiles and set the profile ID in the profile context on the transaction if successful
     #[cfg(feature = "processing")]
     fn process_profiles(&self, state: &mut ProcessEnvelopeState) {
@@ -1567,6 +1589,8 @@ impl EnvelopeProcessorService {
             ItemType::Profile => false,
             ItemType::ReplayEvent => false,
             ItemType::ReplayRecording => false,
+            ItemType::Checkin => false,
+
             // Without knowing more, `Unknown` items are allowed to be repeated
             ItemType::Unknown(_) => false,
         }
@@ -2235,6 +2259,7 @@ impl EnvelopeProcessorService {
             self.enforce_quotas(state)?;
             // We need the event parsed in order to set the profile context on it
             self.process_profiles(state);
+            self.process_checkins(state);
         });
 
         if state.has_event() {