ref(rate-limits): Rate limit envelopes instead of metrics for sampled…

…/indexed items (#3716)

Partially Implements: #3662 (does not move rate limits to after
aggregation).

Changes:
- Data in an envelope is always considered to be both Indexed and
Non-Indexed, for example: Until a transaction is removed from an
envelope it is considered to be both `Transaction` and
`TransactionIndexed`. Only dynamic sampling, when dropping the
transaction from the envelope, emits one outcome with the
`TransactionIndexed` category.
- A few tests asserted the old/wrong behaviour of only emitting a single
outcome, e.g. when an inbound filter filters a transaction we used to
only get a `Transaction` outcome but not a `TransactionIndexed`, these
tests have been aligned.
- When checking rate limits for an envelope, quota on the 'base'
category (e.g. `Transaction`) is now consumed.
- Metrics now remember whether they were extracted from a sampled
envelope item.
- Metrics are immediately rate limited using the same rate limits as the
ones used for the envelope.
- Metrics from sampled envelopes will not be rate limited again.
- Improved cached rate limits with a new type struct, which surfaced a
bug that they weren't always correctly expired (fixed).

Future Improvements:
- Move metrics rate limiting to after aggregation
- Stream line metrics rate limiting from 4 (cached transactions/spans,
transactions/spans, cached metric_buckets, metric_bucket) different
checks to 2 (cached, non-cached)

CHANGELOG.md

@@ -12,6 +12,7 @@
 **Internal**:
 
 - Treat arrays of pairs as key-value mappings during PII scrubbing. ([#3639](https://github.com/getsentry/relay/pull/3639))
+- Rate limit envelopes instead of metrics for sampled/indexed items. ([#3716](https://github.com/getsentry/relay/pull/3716))
 - Improve flush time calculation in metrics aggregator. ([#3726](https://github.com/getsentry/relay/pull/3726))
 
 ## 24.5.1

relay-base-schema/src/data_category.rs

@@ -169,9 +169,19 @@ impl DataCategory {
     pub fn index_category(self) -> Option<Self> {
         match self {
             Self::Transaction => Some(Self::TransactionIndexed),
+            Self::Span => Some(Self::SpanIndexed),
+            Self::Profile => Some(Self::ProfileIndexed),
             _ => None,
         }
     }
+
+    /// Returns `true` if this data category is an indexed data category.
+    pub fn is_indexed(self) -> bool {
+        matches!(
+            self,
+            Self::TransactionIndexed | Self::SpanIndexed | Self::ProfileIndexed
+        )
+    }
 }
 
 impl fmt::Display for DataCategory {

relay-metrics/src/aggregator.rs

@@ -64,6 +64,7 @@ struct BucketKey {
     timestamp: UnixTimestamp,
     metric_name: MetricName,
     tags: BTreeMap<String, String>,
+    extracted_from_indexed: bool,
 }
 
 impl BucketKey {
@@ -725,6 +726,7 @@ impl Aggregator {
             timestamp,
             metric_name: bucket.name,
             tags: bucket.tags,
+            extracted_from_indexed: bucket.metadata.extracted_from_indexed,
         };
         let key = validate_bucket_key(key, &self.config)?;
 
@@ -1012,6 +1014,7 @@ mod tests {
                         "c:transactions/foo@none",
                     ),
                     tags: {},
+                    extracted_from_indexed: false,
                 },
                 Counter(
                     85.0,
@@ -1061,10 +1064,11 @@ mod tests {
                 ("hello".to_owned(), "world".to_owned()),
                 ("answer".to_owned(), "42".to_owned()),
             ]),
+            extracted_from_indexed: false,
         };
         assert_eq!(
             bucket_key.cost(),
-            80 + // BucketKey
+            88 + // BucketKey
             5 + // name
             (5 + 5 + 6 + 2) // tags
         );
@@ -1107,6 +1111,7 @@ mod tests {
                         "c:transactions/foo@none",
                     ),
                     tags: {},
+                    extracted_from_indexed: false,
                 },
                 Counter(
                     84.0,
@@ -1120,6 +1125,7 @@ mod tests {
                         "c:transactions/foo@none",
                     ),
                     tags: {},
+                    extracted_from_indexed: false,
                 },
                 Counter(
                     42.0,
@@ -1242,6 +1248,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/foo@none".into(),
             tags: BTreeMap::new(),
+            extracted_from_indexed: false,
         };
         let fixed_cost = bucket_key.cost() + mem::size_of::<BucketValue>();
         for (metric_name, metric_value, expected_added_cost) in [
@@ -1402,6 +1409,7 @@ mod tests {
                 tags.insert("another\0garbage".to_owned(), "bye".to_owned());
                 tags
             },
+            extracted_from_indexed: false,
         };
 
         let mut bucket_key = validate_bucket_key(bucket_key, &test_config()).unwrap();
@@ -1427,6 +1435,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/a_short_metric".into(),
             tags: BTreeMap::new(),
+            extracted_from_indexed: false,
         };
         assert!(validate_bucket_key(short_metric, &test_config()).is_ok());
 
@@ -1435,6 +1444,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/long_name_a_very_long_name_its_super_long_really_but_like_super_long_probably_the_longest_name_youve_seen_and_even_the_longest_name_ever_its_extremly_long_i_cant_tell_how_long_it_is_because_i_dont_have_that_many_fingers_thus_i_cant_count_the_many_characters_this_long_name_is".into(),
             tags: BTreeMap::new(),
+            extracted_from_indexed: false,
         };
         let validation = validate_bucket_key(long_metric, &test_config());
 
@@ -1450,6 +1460,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/a_short_metric_with_long_tag_key".into(),
             tags: BTreeMap::from([("i_run_out_of_creativity_so_here_we_go_Lorem_Ipsum_is_simply_dummy_text_of_the_printing_and_typesetting_industry_Lorem_Ipsum_has_been_the_industrys_standard_dummy_text_ever_since_the_1500s_when_an_unknown_printer_took_a_galley_of_type_and_scrambled_it_to_make_a_type_specimen_book".into(), "tag_value".into())]),
+            extracted_from_indexed: false,
         };
         let validation = validate_bucket_key(short_metric_long_tag_key, &test_config()).unwrap();
         assert_eq!(validation.tags.len(), 0);
@@ -1459,6 +1470,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/a_short_metric_with_long_tag_value".into(),
             tags: BTreeMap::from([("tag_key".into(), "i_run_out_of_creativity_so_here_we_go_Lorem_Ipsum_is_simply_dummy_text_of_the_printing_and_typesetting_industry_Lorem_Ipsum_has_been_the_industrys_standard_dummy_text_ever_since_the_1500s_when_an_unknown_printer_took_a_galley_of_type_and_scrambled_it_to_make_a_type_specimen_book".into())]),
+            extracted_from_indexed: false,
         };
         let validation = validate_bucket_key(short_metric_long_tag_value, &test_config()).unwrap();
         assert_eq!(validation.tags.len(), 0);
@@ -1476,6 +1488,7 @@ mod tests {
             timestamp: UnixTimestamp::now(),
             metric_name: "c:transactions/a_short_metric".into(),
             tags: BTreeMap::from([("foo".into(), tag_value.clone())]),
+            extracted_from_indexed: false,
         };
         let validated_bucket = validate_metric_tags(short_metric, &test_config());
         assert_eq!(validated_bucket.tags["foo"], tag_value);
@@ -1583,6 +1596,7 @@ mod tests {
                 received_at: Some(
                     UnixTimestamp(999994711),
                 ),
+                extracted_from_indexed: false,
             },
         ]
         "###);
@@ -1609,6 +1623,7 @@ mod tests {
             timestamp,
             metric_name: "c:transactions/foo".into(),
             tags: BTreeMap::new(),
+            extracted_from_indexed: false,
         };
 
         // Second bucket has a timestamp in this hour.
@@ -1618,6 +1633,7 @@ mod tests {
             timestamp,
             metric_name: "c:transactions/foo".into(),
             tags: BTreeMap::new(),
+            extracted_from_indexed: false,
         };
 
         let flush_time_1 = get_flush_time(&config, reference_time, &bucket_key_1);

relay-metrics/src/bucket.rs

@@ -759,6 +759,17 @@ pub struct BucketMetadata {
     /// This field should be set to the time in which the first metric of a specific bucket was
     /// received in the outermost internal Relay.
     pub received_at: Option<UnixTimestamp>,
+
+    /// Is `true` if this metric was extracted from a sampled/indexed envelope item.
+    ///
+    /// The final dynamic sampling decision is always made in processing Relays.
+    /// If a metric was extracted from an item which is sampled (i.e. retained by dynamic sampling), this flag is `true`.
+    ///
+    /// Since these metrics from samples carry additional information, e.g. they don't
+    /// require rate limiting since the sample they've been extracted from was already
+    /// rate limited, this flag must be included in the aggregation key when aggregation buckets.
+    #[serde(skip)]
+    pub extracted_from_indexed: bool,
 }
 
 impl BucketMetadata {
@@ -769,6 +780,7 @@ impl BucketMetadata {
         Self {
             merges: 1,
             received_at: Some(received_at),
+            extracted_from_indexed: false,
         }
     }
 
@@ -793,6 +805,7 @@ impl Default for BucketMetadata {
         Self {
             merges: 1,
             received_at: None,
+            extracted_from_indexed: false,
         }
     }
 }
@@ -927,6 +940,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -961,6 +975,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -1013,6 +1028,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -1073,6 +1089,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -1103,6 +1120,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -1127,6 +1145,7 @@ mod tests {
             metadata: BucketMetadata {
                 merges: 1,
                 received_at: None,
+                extracted_from_indexed: false,
             },
         }
         "###);
@@ -1330,6 +1349,7 @@ mod tests {
                     received_at: Some(
                         UnixTimestamp(1615889440),
                     ),
+                    extracted_from_indexed: false,
                 },
             },
         ]
@@ -1371,6 +1391,7 @@ mod tests {
                     received_at: Some(
                         UnixTimestamp(1615889440),
                     ),
+                    extracted_from_indexed: false,
                 },
             },
         ]
@@ -1459,7 +1480,8 @@ mod tests {
             metadata,
             BucketMetadata {
                 merges: 2,
-                received_at: None
+                received_at: None,
+                extracted_from_indexed: false,
             }
         );
 
@@ -1469,7 +1491,8 @@ mod tests {
             metadata,
             BucketMetadata {
                 merges: 3,
-                received_at: Some(UnixTimestamp::from_secs(10))
+                received_at: Some(UnixTimestamp::from_secs(10)),
+                extracted_from_indexed: false,
             }
         );
 
@@ -1479,7 +1502,8 @@ mod tests {
             metadata,
             BucketMetadata {
                 merges: 4,
-                received_at: Some(UnixTimestamp::from_secs(10))
+                received_at: Some(UnixTimestamp::from_secs(10)),
+                extracted_from_indexed: false,
             }
         );
     }

relay-metrics/src/view.rs

@@ -478,7 +478,7 @@ impl<'a> BucketView<'a> {
 
         BucketMetadata {
             merges,
-            received_at: self.inner.metadata.received_at,
+            ..self.inner.metadata
         }
     }