From 4c811fbee4b4b35bced53a99933a29320cd3f84d Mon Sep 17 00:00:00 2001 From: Dan Fuller Date: Wed, 26 Aug 2026 13:19:54 -0700 Subject: [PATCH 1/3] perf(crons): Include environment in checkin Kafka routing key We already route by this key in the consumer when we put things into batches, but missed it here. This has the effect of causing extremely hot partitions for any monitor that has a lot of envs. This can also be exacerbated if they're also sending more than the rate limit. A missing or empty environment is normalized to "production" for the routing key, matching how Sentry resolves it in `MonitorEnvironment.objects.ensure_environment`. Without that, check-ins omitting the environment and those sending "production" describe one monitor environment but would route to different partitions, which have independent lag. Only the routing key is normalized; the payload is forwarded untouched. --- CHANGELOG.md | 1 + relay-monitors/src/lib.rs | 71 +++++++++++++++++++++++++++++++++++---- 2 files changed, 65 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4804ed17e5..b899b614a09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - 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)) **Bug Fixes**: diff --git a/relay-monitors/src/lib.rs b/relay-monitors/src/lib.rs index 160714dd52a..75c6e6477ff 100644 --- a/relay-monitors/src/lib.rs +++ b/relay-monitors/src/lib.rs @@ -206,13 +206,29 @@ 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. + // + // This has to stay in sync with `CheckinItem.processing_key` in Sentry, which is the key the + // monitors consumer groups check-ins on and the only unit it serializes them within: + // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/types.py + // + // Sentry also resolves a missing or empty environment to "production", so both forms name the + // same monitor environment and have to land on the same partition, otherwise their check-ins + // are processed by different consumers with independent lag: + // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/models.py + // (`MonitorEnvironmentManager.ensure_environment`) + // + // 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, project_id_slug_key.as_bytes()); + let routing_hint = Uuid::new_v5(namespace, routing_key.as_bytes()); Ok(ProcessedCheckInResult { routing_hint, @@ -342,8 +358,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 +369,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#"{ From cef17f02f60f27411c685df1c6a11ce86eeb0c79 Mon Sep 17 00:00:00 2001 From: Dan Fuller Date: Wed, 26 Aug 2026 14:22:22 -0700 Subject: [PATCH 2/3] clean up comments --- relay-monitors/src/lib.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/relay-monitors/src/lib.rs b/relay-monitors/src/lib.rs index 75c6e6477ff..126bb0033bf 100644 --- a/relay-monitors/src/lib.rs +++ b/relay-monitors/src/lib.rs @@ -210,15 +210,13 @@ pub fn process_check_in( // monitor check-ins are processed in order by consistently routing check-ins from the same // monitor + env combo. // - // This has to stay in sync with `CheckinItem.processing_key` in Sentry, which is the key the - // monitors consumer groups check-ins on and the only unit it serializes them within: + // Keep this in sync with `CheckinItem.processing_key` in Sentry // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/types.py // - // Sentry also resolves a missing or empty environment to "production", so both forms name the - // same monitor environment and have to land on the same partition, otherwise their check-ins - // are processed by different consumers with independent lag: + // Also keep the environment in sync with Sentry's `ensure_environment` // https://github.com/getsentry/sentry/blob/master/src/sentry/monitors/models.py - // (`MonitorEnvironmentManager.ensure_environment`) + // 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; From 2ba7aa5bb3dfc1996b402f30fc6e733c6ffcf25d Mon Sep 17 00:00:00 2001 From: Dan Fuller Date: Wed, 26 Aug 2026 12:52:11 -0700 Subject: [PATCH 3/3] perf(crons): Enforce crons rate limiting at the Relay level This enforces the 6 checkin per monitor/env limit at the Relay level. It's also configurable via options automator using the relay.cron-monitor-rate-limit option. We care about this because we partition checkins in the Kafka topic by monitor environment. So when we have someone sending a large number of checkins to a particular monitorenv, we end up with hot partitions that cause the consumer to lag behind. Ideally this should mean we don't see any rate limit events happen on the consumer, and we should see more stability there. --- CHANGELOG.md | 1 + relay-dynamic-config/src/global.rs | 8 + relay-monitors/src/lib.rs | 15 ++ .../src/processing/check_ins/limiter.rs | 210 ++++++++++++++++++ relay-server/src/processing/check_ins/mod.rs | 91 +++++++- .../src/processing/check_ins/process.rs | 102 ++++++++- relay-server/src/processing/relay.rs | 7 +- relay-server/src/services/processor.rs | 4 +- tests/integration/fixtures/mini_sentry.py | 1 + tests/integration/test_monitors.py | 51 +++++ 10 files changed, 481 insertions(+), 9 deletions(-) create mode 100644 relay-server/src/processing/check_ins/limiter.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index b899b614a09..a0a568f060b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ - 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**: diff --git a/relay-dynamic-config/src/global.rs b/relay-dynamic-config/src/global.rs index 0017e6c6ca0..1ef43d8594e 100644 --- a/relay-dynamic-config/src/global.rs +++ b/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, + /// Kill-switch for fetching project configs in endpoints. #[serde( default = "default_killswitched", diff --git a/relay-monitors/src/lib.rs b/relay-monitors/src/lib.rs index 126bb0033bf..5ec37ac7c04 100644 --- a/relay-monitors/src/lib.rs +++ b/relay-monitors/src/lib.rs @@ -174,6 +174,16 @@ pub struct ProcessedCheckInResult { /// The JSON payload of the processed check-in. pub payload: Vec, + + /// 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. @@ -228,9 +238,14 @@ pub fn process_check_in( let routing_hint = Uuid::new_v5(namespace, routing_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, }) } diff --git a/relay-server/src/processing/check_ins/limiter.rs b/relay-server/src/processing/check_ins/limiter.rs new file mode 100644 index 00000000000..70f5ef2518e --- /dev/null +++ b/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 = 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" + ); + } +} diff --git a/relay-server/src/processing/check_ins/mod.rs b/relay-server/src/processing/check_ins/mod.rs index 32276f8b29d..f6324a69f81 100644 --- a/relay-server/src/processing/check_ins/mod.rs +++ b/relay-server/src/processing/check_ins/mod.rs @@ -1,5 +1,7 @@ use std::sync::Arc; +#[cfg(feature = "processing")] +use futures::future; use relay_cogs::{AppFeature, FeatureWeights}; use relay_quotas::{DataCategory, RateLimits}; @@ -9,6 +11,8 @@ use crate::managed::{Counted, Managed, ManagedEnvelope, OutcomeError, Quantities use crate::processing::{self, Context, CountRateLimited, Forward, Output, QuotaRateLimiter}; use crate::services::outcome::{DiscardReason, Outcome}; +#[cfg(feature = "processing")] +mod limiter; mod process; type Result = std::result::Result; @@ -50,12 +54,21 @@ impl From for Error { /// A processor for Check-Ins. pub struct CheckInsProcessor { limiter: Arc, + #[cfg(feature = "processing")] + redis: Option>, } impl CheckInsProcessor { /// Creates a new [`Self`]. - pub fn new(limiter: Arc) -> Self { - Self { limiter } + pub fn new( + limiter: Arc, + #[cfg(feature = "processing")] redis: Option>, + ) -> Self { + Self { + limiter, + #[cfg(feature = "processing")] + redis, + } } } @@ -89,16 +102,82 @@ impl processing::Processor for CheckInsProcessor { mut check_ins: Managed, ctx: Context<'_>, ) -> Result, Rejected> { - if ctx.is_processing() { - process::normalize(&mut check_ins); + #[cfg_attr(not(feature = "processing"), allow(unused_variables))] + let monitors = ctx + .is_processing() + .then(|| process::normalize(&mut check_ins)); + + #[cfg_attr(not(feature = "processing"), allow(unused_mut))] + let mut check_ins = self.limiter.enforce_quotas(check_ins, ctx).await?; + + #[cfg(feature = "processing")] + if let (Some(monitors), Some(redis)) = (monitors, &self.redis) { + self.enforce_monitor_limits(&mut check_ins, &monitors, redis, ctx) + .await; } - let check_ins = self.limiter.enforce_quotas(check_ins, ctx).await?; - Ok(Output::just(CheckInsOutput(check_ins))) } } +#[cfg(feature = "processing")] +impl CheckInsProcessor { + /// Drops check-ins whose monitor has exceeded its own limit. + async fn enforce_monitor_limits( + &self, + check_ins: &mut Managed, + monitors: &[(String, String)], + redis: &relay_quotas::RedisRateLimiter, + ctx: Context<'_>, + ) { + let limit = ctx + .global_config + .options + .cron_monitor_rate_limit + .unwrap_or(limiter::DEFAULT_LIMIT); + + if limit == 0 { + return; + } + + let item_scoping = check_ins.scoping().item(DataCategory::Monitor); + let mut limited = future::join_all(monitors.iter().map(|(slug, environment)| { + let quota = limiter::monitor_quota(slug, environment, limit, limiter::DEFAULT_WINDOW); + + async move { + match redis + .is_rate_limited(&[quota], item_scoping, 1, false) + .await + { + Ok(limits) => limits.is_limited().then_some(limits), + Err(err) => { + relay_log::error!( + error = &err as &dyn std::error::Error, + "failed to check monitor check-in rate limit" + ); + None + } + } + } + })) + .await; + + let mut i = 0; + check_ins.retain( + |check_ins| &mut check_ins.check_ins, + |_check_in, _| { + let limits = limited.get_mut(i).and_then(Option::take); + i += 1; + + match limits { + Some(limits) => Err(Error::RateLimited(limits)), + None => Ok(()), + } + }, + ); + } +} + /// Output produced by the [`CheckInsProcessor`]. #[derive(Debug)] pub struct CheckInsOutput(Managed); diff --git a/relay-server/src/processing/check_ins/process.rs b/relay-server/src/processing/check_ins/process.rs index dc1a10270ff..b08f2fbf030 100644 --- a/relay-server/src/processing/check_ins/process.rs +++ b/relay-server/src/processing/check_ins/process.rs @@ -5,8 +5,11 @@ use crate::processing::check_ins::{Error, SerializedCheckIns}; /// Normalizes all check-ins using the [`relay_monitors`] module. /// /// Individual, invalid check-ins will be discarded. -pub fn normalize(check_ins: &mut Managed) { +/// +/// Returns the monitor slug and environment of each valid check-in +pub fn normalize(check_ins: &mut Managed) -> Vec<(String, String)> { let scoping = check_ins.scoping(); + let mut monitors = Vec::new(); check_ins.retain( |check_ins| &mut check_ins.check_ins, @@ -23,7 +26,104 @@ pub fn normalize(check_ins: &mut Managed) { check_in.set_routing_hint(result.routing_hint); check_in.set_payload(ContentType::Json, result.payload); + monitors.push((result.monitor_slug, result.environment)); Ok::<_, Error>(()) }, ); + + monitors +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + + use crate::envelope::{Envelope, Item, ItemType}; + use relay_quotas::DataCategory; + + use crate::managed::ManagedTestHandle; + use crate::services::outcome::{DiscardReason, Outcome}; + + use super::*; + + fn check_in_item(payload: &str) -> Item { + let mut item = Item::new(ItemType::CheckIn); + item.set_payload(ContentType::Json, payload.to_owned()); + item + } + + fn managed(payloads: &[&str]) -> (Managed, ManagedTestHandle) { + let bytes = Bytes::from( + "{\"dsn\":\"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42\"}".to_owned(), + ); + let headers = Envelope::parse_bytes(bytes).unwrap().headers().clone(); + let check_ins = payloads.iter().copied().map(check_in_item).collect(); + + Managed::for_test(SerializedCheckIns { headers, check_ins }).build() + } + + #[test] + fn test_returned_monitors_line_up_with_the_kept_check_ins() { + let (mut check_ins, mut handle) = managed(&[ + r#"{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"first","status":"ok"}"#, + // Dropped: an empty slug is rejected. + r#"{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"","status":"ok"}"#, + r#"{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"second","environment":"prod","status":"ok"}"#, + // Dropped: not valid json. + r#"{"#, + r#"{"check_in_id":"a460c25ff2554577b920fcfacae4e5eb","monitor_slug":"third","status":"ok"}"#, + ]); + + let monitors = normalize(&mut check_ins); + + assert_eq!( + monitors, + vec![ + // The absent environments are normalized to "production", the same value the + // routing key is built from. + ("first".to_owned(), "production".to_owned()), + ("second".to_owned(), "prod".to_owned()), + ("third".to_owned(), "production".to_owned()), + ] + ); + assert_eq!(monitors.len(), check_ins.check_ins.len()); + + drop(check_ins); + handle.assert_outcome( + &Outcome::Invalid(DiscardReason::InvalidCheckIn), + DataCategory::Monitor, + 1, + ); + handle.assert_outcome( + &Outcome::Invalid(DiscardReason::InvalidJson), + DataCategory::Monitor, + 1, + ); + handle.assert_internal_outcome(DataCategory::Monitor, 3); + } + + #[test] + fn test_no_monitors_when_every_check_in_is_invalid() { + let (mut check_ins, mut handle) = + managed(&[r#"{"#, r#"{"monitor_slug":"","status":"ok"}"#]); + + let monitors = normalize(&mut check_ins); + + assert!(monitors.is_empty()); + assert!(check_ins.check_ins.is_empty()); + + drop(check_ins); + handle.assert_outcome( + &Outcome::Invalid(DiscardReason::InvalidJson), + DataCategory::Monitor, + 1, + ); + handle.assert_outcome( + &Outcome::Invalid(DiscardReason::InvalidCheckIn), + DataCategory::Monitor, + 1, + ); + // Dropping an emptied `Managed` still reports, with nothing left to report on. + handle.assert_internal_outcome(DataCategory::Monitor, 0); + } } diff --git a/relay-server/src/processing/relay.rs b/relay-server/src/processing/relay.rs index e6b13e6e866..df58453c6e0 100644 --- a/relay-server/src/processing/relay.rs +++ b/relay-server/src/processing/relay.rs @@ -58,6 +58,7 @@ impl RelayProcessor { quota_limiter: &Arc, geoip_lookup: &GeoIpLookup, outcome_aggregator: Addr, + #[cfg(feature = "processing")] rate_limiter: Option>, ) -> Self { // Just so everything fits in a single line. let ql = || Arc::clone(quota_limiter); @@ -66,7 +67,11 @@ impl RelayProcessor { cogs, attachments: AttachmentProcessor::new(ql()), - check_ins: CheckInsProcessor::new(ql()), + check_ins: CheckInsProcessor::new( + ql(), + #[cfg(feature = "processing")] + rate_limiter, + ), client_reports: ClientReportsProcessor::new(outcome_aggregator), errors: ErrorsProcessor::new(ql(), geoip_lookup.clone()), forward_unknown: ForwardUnknownProcessor::new(), diff --git a/relay-server/src/services/processor.rs b/relay-server/src/services/processor.rs index 9fd3c1b5526..586272096ac 100644 --- a/relay-server/src/services/processor.rs +++ b/relay-server/src/services/processor.rs @@ -605,12 +605,14 @@ impl EnvelopeProcessorService { global_config, project_cache, #[cfg(feature = "processing")] - rate_limiter, + rate_limiter: rate_limiter.clone(), processor: RelayProcessor::new( cogs.clone(), "a_limiter, &geoip_lookup, addrs.outcome_aggregator.clone(), + #[cfg(feature = "processing")] + rate_limiter, ), cogs, addrs, diff --git a/tests/integration/fixtures/mini_sentry.py b/tests/integration/fixtures/mini_sentry.py index 2b8264f6a13..3ffe4c948c3 100644 --- a/tests/integration/fixtures/mini_sentry.py +++ b/tests/integration/fixtures/mini_sentry.py @@ -571,6 +571,7 @@ def reraise_test_failures(): "relay.span-usage-metric": True, "relay.session.processing.rollout": 1.0, "relay.endpoint-fetch-config.enabled": True, + "relay.cron-monitor-rate-limit": 0, }, } diff --git a/tests/integration/test_monitors.py b/tests/integration/test_monitors.py index ec6cfd304e8..7c1e7afc9db 100644 --- a/tests/integration/test_monitors.py +++ b/tests/integration/test_monitors.py @@ -181,3 +181,54 @@ def test_monitor_post_json_body(mini_sentry, relay): "timezone": "America/Los_Angles", }, } + + +def test_monitor_rate_limit_with_processing( + mini_sentry, relay_with_processing, monitors_consumer, outcomes_consumer +): + project_id = 44 + mini_sentry.add_basic_project_config(project_id) + # Relay only refetches global config every `cache.global_config_fetch_interval` seconds + # (10 by default), so the option has to be set before it starts. + mini_sentry.set_global_config_option("relay.cron-monitor-rate-limit", 2) + relay = relay_with_processing() + monitors_consumer = monitors_consumer() + outcomes_consumer = outcomes_consumer() + + for _ in range(4): + relay.send_check_in(project_id, generate_check_in("limited-monitor")) + + for _ in range(2): + check_in, _ = monitors_consumer.get_check_in() + assert check_in["monitor_slug"] == "limited-monitor" + monitors_consumer.assert_empty() + + outcomes_consumer.assert_rate_limited( + "monitor_rate_limit", categories=["monitor"], quantity=2 + ) + + +def test_monitor_rate_limit_is_per_environment_with_processing( + mini_sentry, relay_with_processing, monitors_consumer, outcomes_consumer +): + project_id = 45 + mini_sentry.add_basic_project_config(project_id) + # Relay only refetches global config every `cache.global_config_fetch_interval` seconds + # (10 by default), so the option has to be set before it starts. + mini_sentry.set_global_config_option("relay.cron-monitor-rate-limit", 1) + relay = relay_with_processing() + monitors_consumer = monitors_consumer() + outcomes_consumer = outcomes_consumer() + + for environment in ["prod", "staging"]: + check_in = generate_check_in("shared-monitor") + check_in["environment"] = environment + relay.send_check_in(project_id, check_in) + + environments = { + monitors_consumer.get_check_in()[0]["environment"] for _ in range(2) + } + assert environments == {"prod", "staging"} + + monitors_consumer.assert_empty() + outcomes_consumer.assert_empty()