ref(cardinality): Log Sentry errors for cardinality limits (#3053)

Dav1dde · web-flow · commit f2e83c397b28 · 2024-02-05T14:25:59.000Z
To also track which namespace/limit was affected, this PR also adds a
hashset of limit IDs to the cardinality limiter return value.
diff --git a/relay-cardinality/benches/redis_impl.rs b/relay-cardinality/benches/redis_impl.rs
@@ -36,8 +36,8 @@ fn build_limiter(redis: RedisPool, reset_redis: bool) -> RedisSetLimiter {
 
 struct NoopRejections;
 
-impl Rejections for NoopRejections {
-    fn reject(&mut self, _entry_id: EntryId) {}
+impl<'a> Rejections<'a> for NoopRejections {
+    fn reject(&mut self, _limit_id: &'a str, _entry_id: EntryId) {}
 }
 
 #[derive(Debug)]
diff --git a/relay-cardinality/src/limiter.rs b/relay-cardinality/src/limiter.rs
@@ -20,26 +20,26 @@ pub struct Scoping {
 }
 
 /// Accumulator of all cardinality limiter rejections.
-pub trait Rejections {
+pub trait Rejections<'a> {
     /// Called for ever [`Entry`] which was rejected from the [`Limiter`].
-    fn reject(&mut self, entry_id: EntryId);
+    fn reject(&mut self, limit_id: &'a str, entry_id: EntryId);
 }
 
 /// Limiter responsible to enforce limits.
 pub trait Limiter {
     /// Verifies cardinality limits.
     ///
     /// Returns an iterator containing only accepted entries.
-    fn check_cardinality_limits<E, R>(
+    fn check_cardinality_limits<'a, E, R>(
         &self,
         scoping: Scoping,
-        limits: &[CardinalityLimit],
+        limits: &'a [CardinalityLimit],
         entries: E,
         rejections: &mut R,
     ) -> Result<()>
     where
         E: IntoIterator<Item = Entry>,
-        R: Rejections;
+        R: Rejections<'a>;
 }
 
 /// Unit of operation for the cardinality limiter.
@@ -100,30 +100,30 @@ impl<T: Limiter> CardinalityLimiter<T> {
     /// Checks cardinality limits of a list of buckets.
     ///
     /// Returns an iterator of all buckets that have been accepted.
-    pub fn check_cardinality_limits<I: CardinalityItem>(
+    pub fn check_cardinality_limits<'a, I: CardinalityItem>(
         &self,
         scoping: Scoping,
-        limits: &[CardinalityLimit],
+        limits: &'a [CardinalityLimit],
         items: Vec<I>,
-    ) -> Result<CardinalityLimits<I>, (Vec<I>, Error)> {
+    ) -> Result<CardinalityLimits<'a, I>, (Vec<I>, Error)> {
         metric!(timer(CardinalityLimiterTimers::CardinalityLimiter), {
             let entries = items.iter().enumerate().filter_map(|(id, item)| {
                 Some(Entry::new(EntryId(id), item.namespace()?, item.to_hash()))
             });
 
-            let mut rejections = RejectedIds::default();
+            let mut rejections = RejectionTracker::default();
             if let Err(err) =
                 self.limiter
                     .check_cardinality_limits(scoping, limits, entries, &mut rejections)
             {
                 return Err((items, err));
             }
 
-            if !rejections.0.is_empty() {
+            if !rejections.entries.is_empty() {
                 relay_log::debug!(
                     scoping = ?scoping,
                     "rejected {} metrics due to cardinality limit",
-                    rejections.0.len(),
+                    rejections.entries.len(),
                 );
             }
 
@@ -136,30 +136,46 @@ impl<T: Limiter> CardinalityLimiter<T> {
 ///
 /// The result can be used directly by [`CardinalityLimits`].
 #[derive(Debug, Default)]
-struct RejectedIds(HashSet<usize>);
+struct RejectionTracker<'a> {
+    limits: HashSet<&'a str>,
+    entries: HashSet<usize>,
+}
 
-impl Rejections for RejectedIds {
+impl<'a> Rejections<'a> for RejectionTracker<'a> {
     #[inline(always)]
-    fn reject(&mut self, entry_id: EntryId) {
-        self.0.insert(entry_id.0);
+    fn reject(&mut self, limit_id: &'a str, entry_id: EntryId) {
+        self.limits.insert(limit_id);
+        self.entries.insert(entry_id.0);
     }
 }
 
 /// Result of [`CardinalityLimiter::check_cardinality_limits`].
 #[derive(Debug)]
-pub struct CardinalityLimits<T> {
+pub struct CardinalityLimits<'a, T> {
     source: Vec<T>,
     rejections: HashSet<usize>,
+    limits: HashSet<&'a str>,
 }
 
-impl<T> CardinalityLimits<T> {
-    fn new(source: Vec<T>, rejections: RejectedIds) -> Self {
+impl<'a, T> CardinalityLimits<'a, T> {
+    fn new(source: Vec<T>, rejections: RejectionTracker<'a>) -> Self {
         Self {
             source,
-            rejections: rejections.0,
+            rejections: rejections.entries,
+            limits: rejections.limits,
         }
     }
 
+    /// Returns `true` if any items have been rejected.
+    pub fn has_rejections(&self) -> bool {
+        !self.rejections.is_empty()
+    }
+
+    /// Returns all id's of cardinality limits which were exceeded.
+    pub fn enforced_limits(&self) -> &HashSet<&'a str> {
+        &self.limits
+    }
+
     /// Recovers the original list of items passed to the cardinality limiter.
     pub fn into_source(self) -> Vec<T> {
         self.source
@@ -259,21 +275,27 @@ mod tests {
         let limits = CardinalityLimits {
             source: vec!['a', 'b', 'c', 'd', 'e'],
             rejections: HashSet::from([0, 1, 3]),
+            limits: HashSet::new(),
         };
         assert_rejected(&limits, ['a', 'b', 'd']);
+        assert!(limits.has_rejections());
         assert_eq!(limits.into_accepted(), vec!['c', 'e']);
 
         let limits = CardinalityLimits {
             source: vec!['a', 'b', 'c', 'd', 'e'],
             rejections: HashSet::from([]),
+            limits: HashSet::new(),
         };
         assert_rejected(&limits, []);
+        assert!(!limits.has_rejections());
         assert_eq!(limits.into_accepted(), vec!['a', 'b', 'c', 'd', 'e']);
 
         let limits = CardinalityLimits {
             source: vec!['a', 'b', 'c', 'd', 'e'],
             rejections: HashSet::from([0, 1, 2, 3, 4]),
+            limits: HashSet::new(),
         };
+        assert!(limits.has_rejections());
         assert_rejected(&limits, ['a', 'b', 'c', 'd', 'e']);
         assert!(limits.into_accepted().is_empty());
     }
@@ -283,19 +305,19 @@ mod tests {
         struct RejectAllLimiter;
 
         impl Limiter for RejectAllLimiter {
-            fn check_cardinality_limits<I, T>(
+            fn check_cardinality_limits<'a, I, T>(
                 &self,
                 _scoping: Scoping,
-                _limits: &[CardinalityLimit],
+                _limits: &'a [CardinalityLimit],
                 entries: I,
                 outcomes: &mut T,
             ) -> Result<()>
             where
                 I: IntoIterator<Item = Entry>,
-                T: Rejections,
+                T: Rejections<'a>,
             {
                 for entry in entries {
-                    outcomes.reject(entry.id);
+                    outcomes.reject("test", entry.id);
                 }
 
                 Ok(())
@@ -304,17 +326,19 @@ mod tests {
 
         let limiter = CardinalityLimiter::new(RejectAllLimiter);
 
+        let limits = build_limits();
         let result = limiter
             .check_cardinality_limits(
                 build_scoping(),
-                &build_limits(),
+                &limits,
                 vec![
                     Item::new(0, MetricNamespace::Transactions),
                     Item::new(1, MetricNamespace::Transactions),
                 ],
             )
             .unwrap();
 
+        assert_eq!(result.enforced_limits(), &HashSet::from(["test"]));
         assert!(result.into_accepted().is_empty());
     }
 
@@ -323,16 +347,16 @@ mod tests {
         struct AcceptAllLimiter;
 
         impl Limiter for AcceptAllLimiter {
-            fn check_cardinality_limits<I, T>(
+            fn check_cardinality_limits<'a, I, T>(
                 &self,
                 _scoping: Scoping,
-                _limits: &[CardinalityLimit],
+                _limits: &'a [CardinalityLimit],
                 _entries: I,
                 _outcomes: &mut T,
             ) -> Result<()>
             where
                 I: IntoIterator<Item = Entry>,
-                T: Rejections,
+                T: Rejections<'a>,
             {
                 Ok(())
             }
@@ -344,8 +368,9 @@ mod tests {
             Item::new(0, MetricNamespace::Transactions),
             Item::new(1, MetricNamespace::Spans),
         ];
+        let limits = build_limits();
         let result = limiter
-            .check_cardinality_limits(build_scoping(), &build_limits(), items.clone())
+            .check_cardinality_limits(build_scoping(), &limits, items.clone())
             .unwrap();
 
         assert_eq!(result.into_accepted(), items);
@@ -356,7 +381,7 @@ mod tests {
         struct RejectEvenLimiter;
 
         impl Limiter for RejectEvenLimiter {
-            fn check_cardinality_limits<I, T>(
+            fn check_cardinality_limits<'a, I, T>(
                 &self,
                 scoping: Scoping,
                 limits: &[CardinalityLimit],
@@ -365,14 +390,14 @@ mod tests {
             ) -> Result<()>
             where
                 I: IntoIterator<Item = Entry>,
-                T: Rejections,
+                T: Rejections<'a>,
             {
                 assert_eq!(scoping, build_scoping());
                 assert_eq!(limits, &build_limits());
 
                 for entry in entries {
                     if entry.id.0 % 2 == 0 {
-                        outcomes.reject(entry.id);
+                        outcomes.reject("test", entry.id);
                     }
                 }
 
diff --git a/relay-cardinality/src/redis/limiter.rs b/relay-cardinality/src/redis/limiter.rs
@@ -90,16 +90,16 @@ impl RedisSetLimiter {
 }
 
 impl Limiter for RedisSetLimiter {
-    fn check_cardinality_limits<E, R>(
+    fn check_cardinality_limits<'a, E, R>(
         &self,
         scoping: Scoping,
-        limits: &[CardinalityLimit],
+        limits: &'a [CardinalityLimit],
         entries: E,
         rejections: &mut R,
     ) -> Result<()>
     where
         E: IntoIterator<Item = Entry>,
-        R: Rejections,
+        R: Rejections<'a>,
     {
         let timestamp = UnixTimestamp::now();
         // Allows to fast forward time in tests.
@@ -124,7 +124,7 @@ impl Limiter for RedisSetLimiter {
                     }
                     CacheOutcome::Rejected => {
                         // Rejected, add it to the rejected list and move on.
-                        rejections.reject(entry.id);
+                        rejections.reject(state.id, entry.id);
                         state.cache_hit();
                         state.rejected();
                     }
@@ -156,7 +156,7 @@ impl Limiter for RedisSetLimiter {
             let mut cache = self.cache.update(state.scope, timestamp); // Acquire a write lock.
             for (entry, status) in results {
                 if status.is_rejected() {
-                    rejections.reject(entry.id);
+                    rejections.reject(state.id, entry.id);
                     state.rejected();
                 } else {
                     cache.accept(entry.hash);
@@ -346,8 +346,8 @@ mod tests {
     #[derive(Debug, Default, PartialEq, Eq)]
     struct Rejections(HashSet<EntryId>);
 
-    impl super::Rejections for Rejections {
-        fn reject(&mut self, entry_id: EntryId) {
+    impl<'a> super::Rejections<'a> for Rejections {
+        fn reject(&mut self, _limit_id: &'a str, entry_id: EntryId) {
             self.0.insert(entry_id);
         }
     }
diff --git a/relay-dynamic-config/src/global.rs b/relay-dynamic-config/src/global.rs
@@ -38,11 +38,6 @@ impl GlobalConfig {
             Ok(None)
         }
     }
-
-    /// Returns the [`Options::cardinality_limiter_mode`] option.
-    pub fn cardinality_limiter_mode(&self) -> CardinalityLimiterMode {
-        self.options.cardinality_limiter_mode
-    }
 }
 
 /// All options passed down from Sentry to Relay.
@@ -79,6 +74,16 @@ pub struct Options {
     )]
     pub cardinality_limiter_mode: CardinalityLimiterMode,
 
+    /// Sample rate for Cardinality Limiter Sentry errors.
+    ///
+    /// Rate needs to be between `0.0` and `1.0`.
+    /// If set to `1.0` all cardinality limiter rejections will be logged as a Sentry error.
+    #[serde(
+        rename = "relay.cardinality-limiter.error-sample-rate",
+        skip_serializing_if = "is_default"
+    )]
+    pub cardinality_limiter_error_sample_rate: f32,
+
     /// Kill switch for disabling the span usage metric.
     ///
     /// This metric is converted into outcomes in a sentry-side consumer.
diff --git a/relay-server/src/services/processor.rs b/relay-server/src/services/processor.rs
@@ -1774,7 +1774,7 @@ impl EnvelopeProcessorService {
         mode: ExtractionMode,
     ) -> Vec<Bucket> {
         let global_config = self.inner.global_config.current();
-        let cardinality_limiter_mode = global_config.cardinality_limiter_mode();
+        let cardinality_limiter_mode = global_config.options.cardinality_limiter_mode;
 
         if matches!(cardinality_limiter_mode, CardinalityLimiterMode::Disabled) {
             return buckets;
@@ -1801,6 +1801,17 @@ impl EnvelopeProcessorService {
             }
         };
 
+        let error_sample_rate = global_config.options.cardinality_limiter_error_sample_rate;
+        if limits.has_rejections() && sample(error_sample_rate) {
+            for limit_id in limits.enforced_limits() {
+                relay_log::error!(
+                    tags.organization_id = scoping.organization_id,
+                    tags.limit_id = limit_id,
+                    "Cardinality Limit"
+                );
+            }
+        }
+
         if matches!(cardinality_limiter_mode, CardinalityLimiterMode::Passive) {
             return limits.into_source();
         }
@@ -2251,6 +2262,14 @@ impl UpstreamRequest for SendEnvelope {
     }
 }
 
+/// Returns `true` if the current item should be sampled.
+///
+/// The passed `rate` is expected to be `0 <= rate <= 1`.
+#[cfg(feature = "processing")]
+fn sample(rate: f32) -> bool {
+    (rate >= 1.0) || (rate > 0.0 && rand::random::<f32>() < rate)
+}
+
 /// Computes a stable partitioning key for sharded metric requests.
 fn partition_key(project_key: ProjectKey, bucket: &Bucket, partitions: Option<u64>) -> Option<u64> {
     use std::hash::{Hash, Hasher};

Original file line number	Diff line number	Diff line change
`@@ -36,8 +36,8 @@ fn build_limiter(redis: RedisPool, reset_redis: bool) -> RedisSetLimiter {`
`36`	`36`
`37`	`37`	`struct NoopRejections;`
`38`	`38`
`39`		`-impl Rejections for NoopRejections {`
`40`		`- fn reject(&mut self, _entry_id: EntryId) {}`
	`39`	`+impl<'a> Rejections<'a> for NoopRejections {`
	`40`	`+ fn reject(&mut self, _limit_id: &'a str, _entry_id: EntryId) {}`
`41`	`41`	`}`
`42`	`42`
`43`	`43`	`#[derive(Debug)]`