feat(quotas): Add ability to use namespace in non-global quotas. (#3090)

#2716

With this PR, we can also use namespace in non-global quotas. 

The reason for this PR was the requirement to be able to ratelimit based
on namespace per org, but in reality it's pretty flexible so we can do
namespace per project as well.

CHANGELOG.md

@@ -30,6 +30,7 @@
 
 **Internal**:
 
+- Add ability to use namespace in non-global quotas. ([#3090](https://github.com/getsentry/relay/pull/3090))
 - Set the span op on segments. ([#3082](https://github.com/getsentry/relay/pull/3082))
 - Skip profiles without required measurements. ([#3112](https://github.com/getsentry/relay/pull/3112))
 - Push metrics summaries to their own topic. ([#3045](https://github.com/getsentry/relay/pull/3045))

relay-quotas/src/redis.rs

@@ -140,10 +140,11 @@ impl<'a> RedisQuota<'a> {
         let org = self.scoping.organization_id;
 
         format!(
-            "quota:{id}{{{org}}}{subscope}:{slot}",
+            "quota:{id}{{{org}}}{subscope}{namespace}:{slot}",
             id = self.prefix,
             org = org,
             subscope = OptionalDisplay(subscope),
+            namespace = OptionalDisplay(self.namespace),
             slot = self.slot(),
         )
     }
@@ -309,6 +310,7 @@ mod tests {
     use std::sync::Arc;
     use std::time::{SystemTime, UNIX_EPOCH};
 
+    use relay_base_schema::metrics::MetricNamespace;
     use relay_base_schema::project::{ProjectId, ProjectKey};
     use relay_redis::redis::Commands;
     use relay_redis::RedisConfigOptions;
@@ -383,6 +385,76 @@ mod tests {
         );
     }
 
+    /// Tests that a quota with and without namespace are counted separately.
+    #[test]
+    fn test_non_global_namespace_quota() {
+        let quota_limit = 5;
+        let get_quota = |namespace: Option<MetricNamespace>| -> Quota {
+            Quota {
+                id: Some(format!("test_simple_quota_{}", uuid::Uuid::new_v4())),
+                categories: DataCategories::new(),
+                scope: QuotaScope::Organization,
+                scope_id: None,
+                limit: Some(quota_limit),
+                window: Some(600),
+                reason_code: Some(ReasonCode::new(format!("ns: {:?}", namespace))),
+                namespace,
+            }
+        };
+
+        let quotas = &[get_quota(None)];
+        let quota_with_namespace = &[get_quota(Some(MetricNamespace::Transactions))];
+
+        let scoping = ItemScoping {
+            category: DataCategory::Error,
+            scoping: &Scoping {
+                organization_id: 42,
+                project_id: ProjectId::new(43),
+                project_key: ProjectKey::parse("a94ae32be2584e0bbd7a4cbb95971fee").unwrap(),
+                key_id: Some(44),
+            },
+            namespace: Some(MetricNamespace::Transactions),
+        };
+
+        let rate_limiter = build_rate_limiter();
+
+        // First confirm normal behaviour without namespace.
+        for i in 0..10 {
+            let rate_limits: Vec<RateLimit> = rate_limiter
+                .is_rate_limited(quotas, scoping, 1, false)
+                .expect("rate limiting failed")
+                .into_iter()
+                .collect();
+
+            if i < quota_limit {
+                assert_eq!(rate_limits, vec![]);
+            } else {
+                assert_eq!(
+                    rate_limits[0].reason_code,
+                    Some(ReasonCode::new("ns: None"))
+                );
+            }
+        }
+
+        // Then, send identical quota with namespace and confirm it counts separately.
+        for i in 0..10 {
+            let rate_limits: Vec<RateLimit> = rate_limiter
+                .is_rate_limited(quota_with_namespace, scoping, 1, false)
+                .expect("rate limiting failed")
+                .into_iter()
+                .collect();
+
+            if i < quota_limit {
+                assert_eq!(rate_limits, vec![]);
+            } else {
+                assert_eq!(
+                    rate_limits[0].reason_code,
+                    Some(ReasonCode::new("ns: Some(Transactions)"))
+                );
+            }
+        }
+    }
+
     #[test]
     fn test_simple_quota() {
         let quotas = &[Quota {