feat(cardinality): Implement name based cardinality limits (#3313)

Implements metric name based cardinality limits, this does not yet use a
Redis pipeline.

Author: Dav1dde

CHANGELOG.md

@@ -19,6 +19,7 @@
 - Extract metrics from transaction spans. ([#3273](https://github.com/getsentry/relay/pull/3273), [#3324](https://github.com/getsentry/relay/pull/3324))
 - Implement volume metric stats. ([#3281](https://github.com/getsentry/relay/pull/3281))
 - Scrub transactions before enforcing quotas. ([#3248](https://github.com/getsentry/relay/pull/3248))
+- Implement metric name based cardinality limits. ([#3313](https://github.com/getsentry/relay/pull/3313))
 - Kafka topic config supports default topic names as keys. ([#3282](https://github.com/getsentry/relay/pull/3282))
 - Set all span tags on the transaction span. ([#3310](https://github.com/getsentry/relay/pull/3310))
 - Collect duration for all spans. ([#3322](https://github.com/getsentry/relay/pull/3322))

Cargo.lock

@@ -18,6 +18,7 @@ redis = ["relay-redis/impl"]
 workspace = true
 
 [dependencies]
+hash32 = { workspace = true }
 hashbrown = { workspace = true }
 parking_lot = { workspace = true }
 relay-base-schema = { workspace = true }

relay-cardinality/Cargo.toml

@@ -73,16 +73,23 @@ impl Params {
     }
 
     #[inline(always)]
-    fn run(&self, limiter: &RedisSetLimiter, entries: impl IntoIterator<Item = Entry>) {
+    fn run<'a>(&self, limiter: &RedisSetLimiter, entries: impl IntoIterator<Item = Entry<'a>>) {
         limiter
             .check_cardinality_limits(self.scoping, &self.limits, entries, &mut NoopRejections)
             .unwrap();
     }
 
     /// Every round contains the same hashes.
-    fn rounds(&self) -> Vec<Vec<Entry>> {
+    fn rounds(&self) -> Vec<Vec<Entry<'static>>> {
         let entries = (0..self.num_hashes)
-            .map(|i| Entry::new(EntryId(i), MetricNamespace::Custom, u32::MAX - (i as u32)))
+            .map(|i| {
+                Entry::new(
+                    EntryId(i),
+                    MetricNamespace::Custom,
+                    "foo",
+                    u32::MAX - (i as u32),
+                )
+            })
             .collect::<Vec<_>>();
 
         (0..self.rounds)
@@ -91,7 +98,7 @@ impl Params {
     }
 
     /// High cardinality, every round contains unique hashes.
-    fn rounds_unique(&self) -> Vec<Vec<Entry>> {
+    fn rounds_unique(&self) -> Vec<Vec<Entry<'static>>> {
         let hash = AtomicU32::new(u32::MAX);
 
         (0..self.rounds)
@@ -101,6 +108,7 @@ impl Params {
                         Entry::new(
                             EntryId(i),
                             MetricNamespace::Custom,
+                            "foo",
                             hash.fetch_sub(1, Ordering::SeqCst),
                         )
                     })
@@ -110,17 +118,18 @@ impl Params {
     }
 
     /// Entry which is never generated by either [`Self::rounds`] or [`Self::rounds_unique`].
-    fn never_entry() -> Entry {
-        Entry::new(EntryId(usize::MAX), MetricNamespace::Custom, 0)
+    fn never_entry() -> Entry<'static> {
+        Entry::new(EntryId(usize::MAX), MetricNamespace::Custom, "foo", 0)
     }
 
     /// A vector of entries which is never generated by either [`Self::rounds`] or [`Self::rounds_unique`].
-    fn never_entries(&self) -> Vec<Entry> {
+    fn never_entries(&self) -> Vec<Entry<'static>> {
         (0..self.limits[0].limit)
             .map(|i| {
                 Entry::new(
                     EntryId(usize::MAX - i as usize),
                     MetricNamespace::Custom,
+                    "foo",
                     i as u32,
                 )
             })

relay-cardinality/benches/redis_impl.rs

@@ -44,6 +44,9 @@ pub enum CardinalityScope {
     /// The limit will be enforced for a specific project.
     Project,
 
+    /// A per metric name cardinality limit.
+    Name,
+
     /// Any other scope that is not known by this Relay.
     #[serde(other)]
     Unknown,

relay-cardinality/src/config.rs

@@ -30,15 +30,15 @@ pub trait Limiter {
     /// Verifies cardinality limits.
     ///
     /// Returns an iterator containing only accepted entries.
-    fn check_cardinality_limits<'a, E, R>(
+    fn check_cardinality_limits<'a, 'b, E, R>(
         &self,
         scoping: Scoping,
         limits: &'a [CardinalityLimit],
         entries: E,
         rejections: &mut R,
     ) -> Result<()>
     where
-        E: IntoIterator<Item = Entry>,
+        E: IntoIterator<Item = Entry<'b>>,
         R: Rejections<'a>;
 }
 
@@ -51,16 +51,21 @@ pub trait CardinalityItem {
     ///
     /// If this method returns `None` the item is automatically rejected.
     fn namespace(&self) -> Option<MetricNamespace>;
+
+    /// Name of the item.
+    fn name(&self) -> &str;
 }
 
 /// A single entry to check cardinality for.
 #[derive(Clone, Copy, Debug)]
-pub struct Entry {
+pub struct Entry<'a> {
     /// Opaque entry Id, used to keep track of indices and buckets.
     pub id: EntryId,
 
-    /// Metric namespace to which the cardinality limit is scoped.
+    /// Metric namespace to which the cardinality limit can be scoped.
     pub namespace: MetricNamespace,
+    /// Name to which the cardinality limit can be scoped.
+    pub name: &'a str,
     /// Hash of the metric name and tags.
     pub hash: u32,
 }
@@ -73,12 +78,13 @@ pub struct Entry {
 #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
 pub struct EntryId(pub usize);
 
-impl Entry {
+impl<'a> Entry<'a> {
     /// Creates a new entry.
-    pub fn new(id: EntryId, namespace: MetricNamespace, hash: u32) -> Self {
+    pub fn new(id: EntryId, namespace: MetricNamespace, name: &'a str, hash: u32) -> Self {
         Self {
             id,
             namespace,
+            name,
             hash,
         }
     }
@@ -112,7 +118,12 @@ impl<T: Limiter> CardinalityLimiter<T> {
 
         metric!(timer(CardinalityLimiterTimers::CardinalityLimiter), {
             let entries = items.iter().enumerate().filter_map(|(id, item)| {
-                Some(Entry::new(EntryId(id), item.namespace()?, item.to_hash()))
+                Some(Entry::new(
+                    EntryId(id),
+                    item.namespace()?,
+                    item.name(),
+                    item.to_hash(),
+                ))
             });
 
             let mut rejections = RejectionTracker::default();
@@ -245,6 +256,10 @@ mod tests {
         fn namespace(&self) -> Option<MetricNamespace> {
             self.namespace
         }
+
+        fn name(&self) -> &str {
+            "foobar"
+        }
     }
 
     fn build_limits() -> [CardinalityLimit; 1] {
@@ -314,15 +329,15 @@ mod tests {
         struct RejectAllLimiter;
 
         impl Limiter for RejectAllLimiter {
-            fn check_cardinality_limits<'a, I, T>(
+            fn check_cardinality_limits<'a, 'b, I, T>(
                 &self,
                 _scoping: Scoping,
                 limits: &'a [CardinalityLimit],
                 entries: I,
                 rejections: &mut T,
             ) -> Result<()>
             where
-                I: IntoIterator<Item = Entry>,
+                I: IntoIterator<Item = Entry<'b>>,
                 T: Rejections<'a>,
             {
                 for entry in entries {
@@ -356,15 +371,15 @@ mod tests {
         struct AcceptAllLimiter;
 
         impl Limiter for AcceptAllLimiter {
-            fn check_cardinality_limits<'a, I, T>(
+            fn check_cardinality_limits<'a, 'b, I, T>(
                 &self,
                 _scoping: Scoping,
                 _limits: &'a [CardinalityLimit],
                 _entries: I,
                 _rejections: &mut T,
             ) -> Result<()>
             where
-                I: IntoIterator<Item = Entry>,
+                I: IntoIterator<Item = Entry<'b>>,
                 T: Rejections<'a>,
             {
                 Ok(())
@@ -390,15 +405,15 @@ mod tests {
         struct RejectEvenLimiter;
 
         impl Limiter for RejectEvenLimiter {
-            fn check_cardinality_limits<'a, I, T>(
+            fn check_cardinality_limits<'a, 'b, I, T>(
                 &self,
                 scoping: Scoping,
                 limits: &'a [CardinalityLimit],
                 entries: I,
                 rejections: &mut T,
             ) -> Result<()>
             where
-                I: IntoIterator<Item = Entry>,
+                I: IntoIterator<Item = Entry<'b>>,
                 T: Rejections<'a>,
             {
                 assert_eq!(scoping, build_scoping());
@@ -445,15 +460,15 @@ mod tests {
         struct RejectLimits;
 
         impl Limiter for RejectLimits {
-            fn check_cardinality_limits<'a, I, T>(
+            fn check_cardinality_limits<'a, 'b, I, T>(
                 &self,
                 _scoping: Scoping,
                 limits: &'a [CardinalityLimit],
                 entries: I,
                 rejections: &mut T,
             ) -> Result<()>
             where
-                I: IntoIterator<Item = Entry>,
+                I: IntoIterator<Item = Entry<'b>>,
                 T: Rejections<'a>,
             {
                 for entry in entries {

relay-cardinality/src/limiter.rs

@@ -196,14 +196,17 @@ impl ScopedCache {
 
 #[cfg(test)]
 mod tests {
+    use relay_base_schema::metrics::MetricNamespace;
     use relay_base_schema::project::ProjectId;
 
+    use crate::limiter::{Entry, EntryId};
+    use crate::redis::quota::PartialQuotaScoping;
     use crate::{CardinalityLimit, CardinalityScope, OrganizationId, Scoping, SlidingWindow};
 
     use super::*;
 
     fn build_scoping(organization_id: OrganizationId, window: SlidingWindow) -> QuotaScoping {
-        QuotaScoping::new(
+        PartialQuotaScoping::new(
             Scoping {
                 organization_id,
                 project_id: ProjectId::new(1),
@@ -218,6 +221,12 @@ mod tests {
             },
         )
         .unwrap()
+        .complete(Entry {
+            id: EntryId(0),
+            namespace: MetricNamespace::Spans,
+            name: "foobar",
+            hash: 123,
+        })
     }
 
     #[test]