perf(crons): Enforce crons rate limiting at the Relay level

CHANGELOG.md

@@ -7,6 +7,8 @@
 - Extract OTLP spans' client sample rate from TraceState. ([#6312](https://github.com/getsentry/relay/pull/6312))
 - Raise the size limit for the flags context to 128 KiB. ([#6310](https://github.com/getsentry/relay/pull/6310))
 - Raise the size limit for logs to 2 MiB. ([#6316](https://github.com/getsentry/relay/pull/6316))
+- Include the environment in the cron check-in routing key so a monitor's environments no longer share a single Kafka partition. ([#6331](https://github.com/getsentry/relay/pull/6331))
+- Rate limit cron check-ins per monitor environment, controlled by the `relay.cron-monitor-rate-limit` option. ([#6330](https://github.com/getsentry/relay/pull/6330))
 
 **Bug Fixes**:
 

relay-dynamic-config/src/global.rs

@@ -156,6 +156,14 @@ pub struct Options {
     )]
     pub sessions_eap_rollout_rate: f32,
 
+    /// Cron check-in messages accepted per monitor environment per minute.
+    #[serde(
+        rename = "relay.cron-monitor-rate-limit",
+        deserialize_with = "default_on_error",
+        skip_serializing_if = "is_default"
+    )]
+    pub cron_monitor_rate_limit: Option<u64>,
+
     /// Kill-switch for fetching project configs in endpoints.
     #[serde(
         default = "default_killswitched",

relay-monitors/src/lib.rs

@@ -174,6 +174,16 @@ pub struct ProcessedCheckInResult {
 
     /// The JSON payload of the processed check-in.
     pub payload: Vec<u8>,
+
+    /// The normalized monitor slug, after trimming.
+    pub monitor_slug: String,
+
+    /// The environment the check-in is associated with.
+    ///
+    /// Normalized the same way as the routing key, so a per-monitor limiter keyed on this cannot
+    /// disagree with the partition the check-in is routed to. Absent and empty environments are
+    /// both reported as `production`, matching Sentry.
+    pub environment: String,
 }
 
 /// Normalizes a monitor check-in payload.
@@ -206,17 +216,36 @@ pub fn process_check_in(
     let namespace = NAMESPACE
         .get_or_init(|| Uuid::new_v5(&Uuid::NAMESPACE_URL, b"https://sentry.io/crons/#did"));
 
-    // Use the project_id + monitor_slug as the routing key hint. This helps ensure monitor
-    // check-ins are processed in order by consistently routing check-ins from the same monitor.
-
+    // Use the project_id + monitor_slug + monitor env as the routing key hint. This helps ensure
+    // monitor check-ins are processed in order by consistently routing check-ins from the same
+    // monitor + env combo.
+    //
+    // Keep this in sync with `CheckinItem.processing_key` in Sentry
+    // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/types.py
+    //
+    // Also keep the environment in sync with Sentry's `ensure_environment`
+    // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/models.py
+    // We translate empty environments to `production`. This needs to be consistent here or we can
+    // end up with checkins for the same monitor/env routed to different partitions.
+    //
+    // Only the routing key is normalized here, the payload is forwarded untouched.
     let slug = &check_in.monitor_slug;
-    let project_id_slug_key = format!("{project_id}:{slug}");
+    let environment = match check_in.environment.as_deref() {
+        Some(environment) if !environment.is_empty() => environment,
+        _ => "production",
+    };
+    let routing_key = format!("{project_id}:{slug}:{environment}");
+
+    let routing_hint = Uuid::new_v5(namespace, routing_key.as_bytes());
 
-    let routing_hint = Uuid::new_v5(namespace, project_id_slug_key.as_bytes());
+    let monitor_slug = check_in.monitor_slug.clone();
+    let environment = environment.to_owned();
 
     Ok(ProcessedCheckInResult {
         routing_hint,
         payload: serde_json::to_vec(&check_in)?,
+        monitor_slug,
+        environment,
     })
 }
 
@@ -342,8 +371,8 @@ mod tests {
 
         let result = process_check_in(json.as_bytes(), ProjectId::new(1));
 
-        // The routing_hint should be consistent for the (project_id, monitor_slug)
-        let expected_uuid = Uuid::parse_str("66e5c5fa-b1b9-5980-8d85-432c1874521a").unwrap();
+        // The routing_hint should be consistent for the (project_id, monitor_slug, environment)
+        let expected_uuid = Uuid::parse_str("9aa99731-a8e3-5594-9f00-c3e8a62c2b11").unwrap();
 
         if let Ok(processed_result) = result {
             assert_eq!(String::from_utf8(processed_result.payload).unwrap(), json);
@@ -353,6 +382,47 @@ mod tests {
         }
     }
 
+    #[test]
+    fn routing_hint_splits_environments() {
+        let hint = |env: &str| {
+            let json = format!(
+                r#"{{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"my-monitor","environment":"{env}","status":"ok"}}"#
+            );
+            process_check_in(json.as_bytes(), ProjectId::new(1))
+                .unwrap()
+                .routing_hint
+        };
+
+        // The consumer groups on (project, slug, environment) and only guarantees order within a
+        // group, so environments of one monitor do not need to share a partition.
+        assert_ne!(hint("prod"), hint("dev"));
+        assert_eq!(hint("prod"), hint("prod"));
+        assert_eq!(
+            hint("prod"),
+            Uuid::parse_str("f97ad155-c5c6-57f4-b748-03a301a14e54").unwrap()
+        );
+    }
+
+    #[test]
+    fn routing_hint_treats_missing_environment_as_production() {
+        let hint = |env: Option<&str>| {
+            let json = match env {
+                Some(env) => format!(
+                    r#"{{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"my-monitor","environment":"{env}","status":"ok"}}"#
+                ),
+                None => r#"{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"my-monitor","status":"ok"}"#.to_owned(),
+            };
+            process_check_in(json.as_bytes(), ProjectId::new(1))
+                .unwrap()
+                .routing_hint
+        };
+
+        // Sentry resolves all three to the same monitor environment, so they have to share a
+        // partition or their check-ins can be processed out of order.
+        assert_eq!(hint(None), hint(Some("")));
+        assert_eq!(hint(None), hint(Some("production")));
+    }
+
     #[test]
     fn process_empty_slug() {
         let json = r#"{

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

@@ -0,0 +1,210 @@
+use std::sync::{Arc, OnceLock};
+
+use relay_quotas::{DataCategories, DataCategory, Quota, QuotaScope, ReasonCode};
+use uuid::Uuid;
+
+/// Reason code reported on outcomes for check-ins dropped by this limiter.
+pub const REASON_CODE: &str = "monitor_rate_limit";
+
+/// Default number of heck-in messages permitted per monitor environment per window
+/// Mirrors `crons.per_monitor_rate_limit` in Sentry
+pub const DEFAULT_LIMIT: u64 = 6;
+
+/// Length of the window in seconds.
+pub const DEFAULT_WINDOW: u64 = 60;
+
+/// Builds a quota that counts a single monitor environment.
+///
+/// `environment` is expected to already be normalized by [`relay_monitors::process_check_in`],
+/// which is also what the Kafka routing key is derived from, so the two cannot disagree.
+pub fn monitor_quota(slug: &str, environment: &str, limit: u64, window: u64) -> Quota {
+    static NAMESPACE: OnceLock<Uuid> = OnceLock::new();
+    let namespace = NAMESPACE
+        .get_or_init(|| Uuid::new_v5(&Uuid::NAMESPACE_URL, b"https://sentry.io/crons/#rl"));
+
+    let key = format!("{}:{slug}:{environment}", slug.len());
+    let id = format!(
+        "monitor:{}",
+        Uuid::new_v5(namespace, key.as_bytes()).simple()
+    );
+
+    Quota {
+        id: Some(Arc::from(id)),
+        categories: DataCategories::from_slice(&[DataCategory::Monitor]),
+        scope: QuotaScope::Project,
+        scope_id: None,
+        limit: Some(limit),
+        window: Some(window),
+        namespace: None,
+        reason_code: Some(ReasonCode::new(REASON_CODE)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn quota(slug: &str, environment: &str) -> Quota {
+        monitor_quota(slug, environment, DEFAULT_LIMIT, DEFAULT_WINDOW)
+    }
+
+    #[test]
+    fn test_id_is_stable() {
+        assert_eq!(quota("nightly", "prod").id, quota("nightly", "prod").id);
+        assert_eq!(
+            quota("nightly", "prod").id.as_deref(),
+            Some("monitor:7cdfae8f0da55aecba3b846aa8c31c14")
+        );
+    }
+
+    #[test]
+    fn test_environments_do_not_share_a_counter() {
+        assert_ne!(quota("job", "prod").id, quota("job", "stg").id);
+    }
+
+    #[test]
+    fn test_separator_in_slug_does_not_collide() {
+        assert_ne!(quota("job", "a:b").id, quota("job:a", "b").id);
+    }
+
+    #[test]
+    fn test_applies_to_check_ins_in_any_project() {
+        let quota = quota("job", "production");
+
+        assert_eq!(quota.scope, QuotaScope::Project);
+        assert_eq!(quota.scope_id, None);
+        assert!(quota.categories.contains(&DataCategory::Monitor));
+        assert!(quota.id.is_some(), "an id is required to count in redis");
+    }
+}
+
+/// Tests against a live redis
+///
+/// Every test derives a unique slug so counters from an earlier run cannot leak into a later one.
+#[cfg(test)]
+mod redis_tests {
+    use std::time::{SystemTime, UNIX_EPOCH};
+
+    use relay_base_schema::organization::OrganizationId;
+    use relay_base_schema::project::{ProjectId, ProjectKey};
+    use relay_quotas::{RedisRateLimiter, Scoping};
+    use relay_redis::{AsyncRedisClient, RedisConfigOptions};
+
+    use super::*;
+
+    fn build_limiter() -> RedisRateLimiter {
+        let url = std::env::var("RELAY_REDIS_URL")
+            .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_owned());
+        let client =
+            AsyncRedisClient::single("test", &url, &RedisConfigOptions::default()).unwrap();
+
+        RedisRateLimiter::new(client)
+    }
+
+    fn scoping(project_id: u64) -> Scoping {
+        Scoping {
+            organization_id: OrganizationId::new(42),
+            project_id: ProjectId::new(project_id),
+            project_key: ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap(),
+            key_id: None,
+        }
+    }
+
+    fn unique_slug(name: &str) -> String {
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+
+        format!("{name}-{nanos}")
+    }
+
+    async fn check(
+        limiter: &RedisRateLimiter,
+        slug: &str,
+        environment: &str,
+        project_id: u64,
+    ) -> bool {
+        let quota = monitor_quota(slug, environment, DEFAULT_LIMIT, DEFAULT_WINDOW);
+        let scoping = scoping(project_id);
+
+        limiter
+            .is_rate_limited(&[quota], scoping.item(DataCategory::Monitor), 1, false)
+            .await
+            .unwrap()
+            .is_limited()
+    }
+
+    #[tokio::test]
+    async fn test_limits_once_the_allowance_is_used_up() {
+        let limiter = build_limiter();
+        let slug = unique_slug("noisy");
+
+        for i in 0..DEFAULT_LIMIT {
+            assert!(
+                !check(&limiter, &slug, "production", 1).await,
+                "check-in {i} passes"
+            );
+        }
+
+        assert!(
+            check(&limiter, &slug, "production", 1).await,
+            "the next is limited"
+        );
+        assert!(
+            check(&limiter, &slug, "production", 1).await,
+            "and stays limited"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_reports_the_expected_reason_code() {
+        let limiter = build_limiter();
+        let slug = unique_slug("reason");
+        let quota = monitor_quota(&slug, "production", 0, DEFAULT_WINDOW);
+        let scoping = scoping(1);
+
+        let limits = limiter
+            .is_rate_limited(&[quota], scoping.item(DataCategory::Monitor), 1, false)
+            .await
+            .unwrap();
+
+        let reason = limits.longest().and_then(|limit| limit.reason_code.clone());
+        assert_eq!(reason.as_ref().map(|r| r.as_str()), Some(REASON_CODE));
+    }
+
+    #[tokio::test]
+    async fn test_environments_do_not_share_an_allowance() {
+        let limiter = build_limiter();
+        let slug = unique_slug("shared");
+
+        for _ in 0..DEFAULT_LIMIT {
+            assert!(!check(&limiter, &slug, "prod", 1).await);
+        }
+        assert!(check(&limiter, &slug, "prod", 1).await, "prod limited");
+
+        assert!(
+            !check(&limiter, &slug, "staging", 1).await,
+            "staging has its own allowance"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_projects_do_not_share_an_allowance() {
+        let limiter = build_limiter();
+        let slug = unique_slug("cross-project");
+
+        for _ in 0..DEFAULT_LIMIT {
+            assert!(!check(&limiter, &slug, "production", 1).await);
+        }
+        assert!(
+            check(&limiter, &slug, "production", 1).await,
+            "project 1 limited"
+        );
+
+        assert!(
+            !check(&limiter, &slug, "production", 2).await,
+            "the same slug in another project is unaffected"
+        );
+    }
+}