diff --git a/CHANGELOG.md b/CHANGELOG.md index c4804ed17e5..a0a568f060b 100644 --- a/CHANGELOG.md +++ b/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**: 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 160714dd52a..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. @@ -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#"{ 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()