feat(server): Org rate limit per metric bucket (#2758)

part of: #2716

in order to protect our kafka metric consumers, we want to have a way of
rate limiting based on the amount of buckets, as that's what decides the
load placed on our kafka topics.

We are starting out with just the org throughput limits but will be
expanded upon further as outlined in the linked epic.

CHANGELOG.md

@@ -23,6 +23,7 @@
 - Temporarily add metric summaries on spans and top-level transaction events to link DDM with performance monitoring. ([#2757](https://github.com/getsentry/relay/pull/2757))
 - Add size limits on metric related envelope items. ([#2800](https://github.com/getsentry/relay/pull/2800))
 - Include the size offending item in the size limit error message. ([#2801](https://github.com/getsentry/relay/pull/2801))
+- Org rate limit metrics per bucket. ([#2758](https://github.com/getsentry/relay/pull/2758))
 
 ## 23.11.2
 

py/sentry_relay/consts.py

@@ -25,6 +25,7 @@ class DataCategory(IntEnum):
  SPAN = 12
  MONITOR_SEAT = 13
  USER_REPORT_V2 = 14
+ METRIC_BUCKET = 15
  UNKNOWN = -1
  # end generated
 

relay-base-schema/src/data_category.rs

@@ -63,6 +63,8 @@ pub enum DataCategory {
  /// Currently standardized on name UserReportV2 to avoid clashing with the old UserReport.
  /// TODO(jferg): Rename this to UserFeedback once old UserReport is deprecated.
  UserReportV2 = 14,
+ /// Metric bucket.
+ MetricBucket = 15,
  //
  // IMPORTANT: After adding a new entry to DataCategory, go to the `relay-cabi` subfolder and run
  // `make header` to regenerate the C-binding. This allows using the data category from Python.
@@ -93,6 +95,7 @@ impl DataCategory {
  "span" => Self::Span,
  "monitor_seat" => Self::MonitorSeat,
  "feedback" => Self::UserReportV2,
+ "metric_bucket" => Self::MetricBucket,
  _ => Self::Unknown,
  }
  }
@@ -116,6 +119,7 @@ impl DataCategory {
  Self::Span => "span",
  Self::MonitorSeat => "monitor_seat",
  Self::UserReportV2 => "feedback",
+ Self::MetricBucket => "metric_bucket",
  Self::Unknown => "unknown",
  }
  }

relay-cabi/include/relay.h

@@ -15,8 +15,7 @@
 /**
  * Classifies the type of data that is being ingested.
  */
-enum RelayDataCategory
-{
+enum RelayDataCategory {
  /**
  * Reserved and unused.
  */
@@ -97,6 +96,10 @@ enum RelayDataCategory
  * TODO(jferg): Rename this to UserFeedback once old UserReport is deprecated.
  */
  RELAY_DATA_CATEGORY_USER_REPORT_V2 = 14,
+ /**
+ * Metric bucket.
+ */
+ RELAY_DATA_CATEGORY_METRIC_BUCKET = 15,
  /**
  * Any other data category not known by this Relay.
  */
@@ -107,8 +110,7 @@ typedef int8_t RelayDataCategory;
 /**
  * Controls the globbing behaviors.
  */
-enum GlobFlags
-{
+enum GlobFlags {
  /**
  * When enabled `**` matches over path separators and `*` does not.
  */
@@ -131,8 +133,7 @@ typedef uint32_t GlobFlags;
 /**
  * Represents all possible error codes.
  */
-enum RelayErrorCode
-{
+enum RelayErrorCode {
  RELAY_ERROR_CODE_NO_ERROR = 0,
  RELAY_ERROR_CODE_PANIC = 1,
  RELAY_ERROR_CODE_UNKNOWN = 2,
@@ -157,8 +158,7 @@ typedef uint32_t RelayErrorCode;
  * Values from <https://github.com/open-telemetry/opentelemetry-specification/blob/8fb6c14e4709e75a9aaa64b0dbbdf02a6067682a/specification/api-tracing.md#status>
  * Mapping to HTTP from <https://github.com/open-telemetry/opentelemetry-specification/blob/8fb6c14e4709e75a9aaa64b0dbbdf02a6067682a/specification/data-http.md#status>
  */
-enum RelaySpanStatus
-{
+enum RelaySpanStatus {
  /**
  * The operation completed successfully.
  *
@@ -283,8 +283,7 @@ typedef struct RelayStoreNormalizer RelayStoreNormalizer;
  * - When obtained as instance through return values, always free the string.
  * - When obtained as pointer through field access, never free the string.
  */
-typedef struct RelayStr
-{
+typedef struct RelayStr {
  /**
  * Pointer to the UTF-8 encoded string data.
  */
@@ -308,8 +307,7 @@ typedef struct RelayStr
  * - When obtained as instance through return values, always free the buffer.
  * - When obtained as pointer through field access, never free the buffer.
  */
-typedef struct RelayBuf
-{
+typedef struct RelayBuf {
  /**
  * Pointer to the raw data.
  */
@@ -327,8 +325,7 @@ typedef struct RelayBuf
 /**
  * Represents a key pair from key generation.
  */
-typedef struct RelayKeyPair
-{
+typedef struct RelayKeyPair {
  /**
  * The public key used for verifying Relay signatures.
  */
@@ -342,8 +339,7 @@ typedef struct RelayKeyPair
 /**
  * A 16-byte UUID.
  */
-typedef struct RelayUuid
-{
+typedef struct RelayUuid {
  /**
  * UUID bytes in network byte order (big endian).
  */

relay-quotas/src/quota.rs

@@ -111,6 +111,7 @@ impl CategoryUnit {
  | DataCategory::Span
  | DataCategory::MonitorSeat
  | DataCategory::Monitor
+ | DataCategory::MetricBucket
  | DataCategory::UserReportV2 => Some(Self::Count),
  DataCategory::Attachment => Some(Self::Bytes),
  DataCategory::Session => Some(Self::Batched),
@@ -295,16 +296,14 @@ impl Quota {
  // Check for a scope identifier constraint. If there is no constraint, this means that the
  // quota matches any scope. In case the scope is unknown, it will be coerced to the most
  // specific scope later.
- let scope_id = match self.scope_id {
- Some(ref scope_id) => scope_id,
- None => return true,
+ let Some(scope_id) = self.scope_id.as_ref() else {
+ return true;
  };
 
  // Check if the scope identifier in the quota is parseable. If not, this means we cannot
  // fulfill the constraint, so the quota does not match.
- let parsed = match scope_id.parse::<u64>() {
- Ok(parsed) => parsed,
- Err(_) => return false,
+ let Ok(parsed) = scope_id.parse::<u64>() else {
+ return false;
  };
 
  // At this stage, require that the scope is known since we have to fulfill the constraint.