feat(server): Send metrics via the global endpoint

relay-metrics/src/aggregator.rs

@@ -71,13 +71,9 @@ impl BucketKey {
     ///
     /// This is used for partition key computation and statsd logging.
     fn hash64(&self) -> u64 {
-        BucketKeyRef {
-            project_key: self.project_key,
-            timestamp: self.timestamp,
-            metric_name: &self.metric_name,
-            tags: &self.tags,
-        }
-        .hash64()
+        let mut hasher = FnvHasher::default();
+        std::hash::Hash::hash(self, &mut hasher);
+        hasher.finish()
     }
 
     /// Estimates the number of bytes needed to encode the bucket key.
@@ -97,29 +93,6 @@ impl BucketKey {
     }
 }
 
-/// Pendant to [`BucketKey`] for referenced data, not owned data.
-///
-/// This makes it possible to compute a hash for a [`Bucket`]
-/// without destructing the bucket into a [`BucketKey`].
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-struct BucketKeyRef<'a> {
-    project_key: ProjectKey,
-    timestamp: UnixTimestamp,
-    metric_name: &'a str,
-    tags: &'a BTreeMap<String, String>,
-}
-
-impl<'a> BucketKeyRef<'a> {
-    /// Creates a 64-bit hash of the bucket key using FnvHasher.
-    ///
-    /// This is used for partition key computation and statsd logging.
-    fn hash64(&self) -> u64 {
-        let mut hasher = FnvHasher::default();
-        std::hash::Hash::hash(self, &mut hasher);
-        hasher.finish()
-    }
-}
-
 /// Estimates the number of bytes needed to encode the tags.
 ///
 /// Note that this does not necessarily match the exact memory footprint of the tags,
@@ -888,36 +861,6 @@ impl fmt::Debug for Aggregator {
     }
 }
 
-/// Splits buckets into N logical partitions, determined by the bucket key.
-pub fn partition_buckets(
-    project_key: ProjectKey,
-    buckets: impl IntoIterator<Item = Bucket>,
-    flush_partitions: Option<u64>,
-) -> BTreeMap<Option<u64>, Vec<Bucket>> {
-    let flush_partitions = match flush_partitions {
-        None => return BTreeMap::from([(None, buckets.into_iter().collect())]),
-        Some(x) => x.max(1), // handle 0,
-    };
-    let mut partitions = BTreeMap::<_, Vec<Bucket>>::new();
-    for bucket in buckets {
-        let key = BucketKeyRef {
-            project_key,
-            timestamp: bucket.timestamp,
-            metric_name: &bucket.name,
-            tags: &bucket.tags,
-        };
-
-        let partition_key = key.hash64() % flush_partitions;
-        partitions
-            .entry(Some(partition_key))
-            .or_default()
-            .push(bucket);
-
-        relay_statsd::metric!(histogram(MetricHistograms::PartitionKeys) = partition_key);
-    }
-    partitions
-}
-
 #[cfg(test)]
 mod tests {
 

relay-metrics/src/statsd.rs

@@ -124,13 +124,6 @@ pub enum MetricHistograms {
     ///    time period (`false`) or after the initial delay has expired (`true`).
     BucketsDelay,
 
-    ///
-    /// Distribution of flush buckets over partition keys.
-    ///
-    /// The distribution of buckets should be even.
-    /// If it is not, this metric should expose it.
-    PartitionKeys,
-
     /// Distribution of invalid bucket timestamps observed, relative to the time of observation.
     ///
     /// This is a temporary metric to better understand why we see so many invalid timestamp errors.
@@ -143,7 +136,6 @@ impl HistogramMetric for MetricHistograms {
             Self::BucketsFlushed => "metrics.buckets.flushed",
             Self::BucketsFlushedPerProject => "metrics.buckets.flushed_per_project",
             Self::BucketsDelay => "metrics.buckets.delay",
-            Self::PartitionKeys => "metrics.buckets.partition_keys",
             Self::InvalidBucketTimestamp => "metrics.buckets.invalid_timestamp",
         }
     }

relay-metrics/src/view.rs

@@ -276,14 +276,10 @@ impl<'a> Iterator for BucketsViewBySizeIter<'a> {
                 }
                 SplitDecision::MoveToNextBatch => break,
                 SplitDecision::Split(at) => {
-                    // Only certain buckets can be split, if the bucket can't be split,
-                    // move it to the next batch.
-                    if bucket.can_split() {
-                        self.current = Index {
-                            slice: self.current.slice,
-                            bucket: self.current.bucket + at,
-                        };
-                    }
+                    self.current = Index {
+                        slice: self.current.slice,
+                        bucket: self.current.bucket + at,
+                    };
                     break;
                 }
             }
@@ -332,6 +328,7 @@ impl<'a> Serialize for BucketsView<'a> {
 /// ```
 ///
 /// A view can be split again into multiple smaller views.
+#[derive(Clone)]
 pub struct BucketView<'a> {
     /// The source bucket.
     inner: &'a Bucket,
@@ -427,6 +424,46 @@ impl<'a> BucketView<'a> {
         Some(self)
     }
 
+    /// Estimates the number of bytes needed to serialize the bucket without value.
+    ///
+    /// Note that this does not match the exact size of the serialized payload. Instead, the size is
+    /// approximated through tags and a static overhead.
+    fn estimated_base_size(&self) -> usize {
+        50 + self.name().len() + aggregator::tags_cost(self.tags())
+    }
+
+    /// Estimates the number of bytes needed to serialize the bucket.
+    ///
+    /// Note that this does not match the exact size of the serialized payload. Instead, the size is
+    /// approximated through the number of contained values, assuming an average size of serialized
+    /// values.
+    pub fn estimated_size(&self) -> usize {
+        self.estimated_base_size() + self.len() * AVG_VALUE_SIZE
+    }
+
+    /// Calculates a split for this bucket if its estimated serialization size exceeds a threshold.
+    ///
+    /// There are three possible return values:
+    ///  - `(Some, None)` if the bucket fits entirely into the size budget. There is no split.
+    ///  - `(None, Some)` if the size budget cannot even hold the bucket name and tags. There is no
+    ///    split, the entire bucket is moved.
+    ///  - `(Some, Some)` if the bucket fits partially. Remaining values are moved into a new bucket
+    ///    with all other information cloned.
+    ///
+    /// This is an approximate function. The bucket is not actually serialized, but rather its
+    /// footprint is estimated through the number of data points contained. See
+    /// [`estimated_size`](Self::estimated_size) for more information.
+    pub fn split(self, size: usize, max_size: Option<usize>) -> (Option<Self>, Option<Self>) {
+        match split_at(&self, size, max_size.unwrap_or(0) / BUCKET_SPLIT_FACTOR) {
+            SplitDecision::BucketFits(_) => (Some(self), None),
+            SplitDecision::MoveToNextBatch => (None, Some(self)),
+            SplitDecision::Split(at) => {
+                let Range { start, end } = self.range.clone();
+                (self.clone().select(start..at), self.select(at..end))
+            }
+        }
+    }
+
     /// Whether the bucket can be split into multiple.
     ///
     /// Only set and distribution buckets can be split.
@@ -624,14 +661,18 @@ enum SplitDecision {
 /// `estimate_size` for more information.
 fn split_at(bucket: &BucketView<'_>, max_size: usize, min_split_size: usize) -> SplitDecision {
     // If there's enough space for the entire bucket, do not perform a split.
-    let bucket_size = estimate_size(bucket);
+    let bucket_size = bucket.estimated_size();
     if max_size >= bucket_size {
         return SplitDecision::BucketFits(bucket_size);
     }
 
+    if !bucket.can_split() {
+        return SplitDecision::MoveToNextBatch;
+    }
+
     // If the bucket key can't even fit into the remaining length, move the entire bucket into
     // the right-hand side.
-    let own_size = estimate_base_size(bucket);
+    let own_size = bucket.estimated_base_size();
     if max_size < (own_size + AVG_VALUE_SIZE) {
         // split_at must not be zero
         return SplitDecision::MoveToNextBatch;
@@ -644,27 +685,9 @@ fn split_at(bucket: &BucketView<'_>, max_size: usize, min_split_size: usize) ->
     // Perform a split with the remaining space after adding the key. We assume an average
     // length of 8 bytes per value and compute the number of items fitting into the left side.
     let split_at = (max_size - own_size) / AVG_VALUE_SIZE;
-
     SplitDecision::Split(split_at)
 }
 
-/// Estimates the number of bytes needed to serialize the bucket without value.
-///
-/// Note that this does not match the exact size of the serialized payload. Instead, the size is
-/// approximated through tags and a static overhead.
-fn estimate_base_size(bucket: &BucketView<'_>) -> usize {
-    50 + bucket.name().len() + aggregator::tags_cost(bucket.tags())
-}
-
-/// Estimates the number of bytes needed to serialize the bucket.
-///
-/// Note that this does not match the exact size of the serialized payload. Instead, the size is
-/// approximated through the number of contained values, assuming an average size of serialized
-/// values.
-fn estimate_size(bucket: &BucketView<'_>) -> usize {
-    estimate_base_size(bucket) + bucket.len() * AVG_VALUE_SIZE
-}
-
 #[cfg(test)]
 mod tests {
     use insta::assert_json_snapshot;
@@ -919,7 +942,7 @@ b3:42:75|s"#;
             .by_size(100)
             .map(|bv| {
                 let len: usize = bv.iter().map(|b| b.len()).sum();
-                let size: usize = bv.iter().map(|b| estimate_size(&b)).sum();
+                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
                 (len, size)
             })
             .collect::<Vec<_>>();
@@ -945,7 +968,7 @@ b3:42:75|s"#;
             .by_size(250)
             .map(|bv| {
                 let len: usize = bv.iter().map(|b| b.len()).sum();
-                let size: usize = bv.iter().map(|b| estimate_size(&b)).sum();
+                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
                 (len, size)
             })
             .collect::<Vec<_>>();
@@ -971,7 +994,7 @@ b3:42:75|s"#;
             .by_size(500)
             .map(|bv| {
                 let len: usize = bv.iter().map(|b| b.len()).sum();
-                let size: usize = bv.iter().map(|b| estimate_size(&b)).sum();
+                let size: usize = bv.iter().map(|b| b.estimated_size()).sum();
                 (len, size)
             })
             .collect::<Vec<_>>();

relay-server/Cargo.toml

@@ -50,6 +50,7 @@ bytes = { version = "1.4.0" }
 chrono = { workspace = true, features = ["clock"] }
 data-encoding = "2.3.3"
 flate2 = "1.0.19"
+fnv = "1.0.7"
 futures = { workspace = true }
 hash32 = { workspace = true }
 hashbrown = { workspace = true }
@@ -95,7 +96,7 @@ rmp-serde = "1.1.1"
 rust-embed = { version = "8.0.0", optional = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-smallvec = { workspace = true, features = ["drain_filter"]  }
+smallvec = { workspace = true, features = ["drain_filter"] }
 sqlx = { version = "0.7.0", features = [
     "macros",
     "migrate",