feat(metrics): Route to different aggregators (#2341)

Place a router in front of the `relay_metrics::Aggregator` service, to
allow different aggregator configurations for different use cases. This
will allow us to

1. change the bucket interval for span metrics from 10 seconds to 1
minute to improve aggregation,
2. set a different maximum length for tag values for span
metrics.

---------

Co-authored-by: Tor <[email protected]>

Author: jjbayer

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+**Internal**:
+
+- Add capability to configure metrics aggregators per use case. ([#2341](https://github.com/getsentry/relay/pull/2341))
+
 ## 23.7.0
 
 **Bug Fixes**:

Cargo.lock

@@ -14,7 +14,7 @@ use relay_common::{Dsn, Uuid};
 use relay_kafka::{
     ConfigError as KafkaConfigError, KafkaConfig, KafkaConfigParam, KafkaTopic, TopicAssignments,
 };
-use relay_metrics::AggregatorConfig;
+use relay_metrics::{AggregatorConfig, ScopedAggregatorConfig};
 use relay_redis::RedisConfig;
 use serde::de::{DeserializeOwned, Unexpected, Visitor};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -1220,6 +1220,8 @@ struct ConfigValues {
     #[serde(default)]
     aggregator: AggregatorConfig,
     #[serde(default)]
+    secondary_aggregators: Vec<ScopedAggregatorConfig>,
+    #[serde(default)]
     auth: AuthConfig,
     #[serde(default)]
     aws: AwsConfig,
@@ -1963,6 +1965,11 @@ impl Config {
         &self.values.aggregator
     }
 
+    /// Returns configuration for non-default metrics [aggregators](relay_metrics::Aggregator).
+    pub fn secondary_aggregator_configs(&self) -> &Vec<ScopedAggregatorConfig> {
+        &self.values.secondary_aggregators
+    }
+
     /// Return the statically configured Relays.
     pub fn static_relays(&self) -> &HashMap<RelayId, RelayInfo> {
         &self.values.auth.static_relays

relay-config/src/config.rs

@@ -13,6 +13,7 @@ publish = false
 float-ord = "0.3.1"
 fnv = "1.0.7"
 hash32 = "0.3.1"
+itertools = "0.10.5"
 relay-common = { path = "../relay-common" }
 relay-log = { path = "../relay-log" }
 relay-statsd = { path = "../relay-statsd" }

relay-metrics/Cargo.toml

@@ -1056,7 +1056,7 @@ impl AggregatorConfig {
         if !self.timestamp_range().contains(&output_timestamp) {
             let delta = (ts as i64) - (UnixTimestamp::now().as_secs() as i64);
             relay_statsd::metric!(
-                histogram(MetricHistograms::InvalidBucketTimestamp) = delta as f64
+                histogram(MetricHistograms::InvalidBucketTimestamp) = delta as f64,
             );
             return Err(AggregateMetricsErrorKind::InvalidTimestamp(timestamp).into());
         }
@@ -1092,7 +1092,7 @@ impl AggregatorConfig {
         let delay = UnixTimestamp::now().as_secs() as i64 - bucket_timestamp.as_secs() as i64;
         relay_statsd::metric!(
             histogram(MetricHistograms::BucketsDelay) = delay as f64,
-            backedated = if flush.is_none() { "true" } else { "false" },
+            backdated = if flush.is_none() { "true" } else { "false" },
         );
 
         // If the initial flush time has passed or cannot be represented, debounce future flushes
@@ -1173,7 +1173,6 @@ impl Ord for QueuedBucket {
 /// This is cheaper to pass around than a (BucketKey, Bucket) pair.
 pub struct HashedBucket {
     // This is only public because pop_flush_buckets is used in benchmark.
-    // TODO: Find better name for this struct
     hashed_key: u64,
     bucket: Bucket,
 }
@@ -1372,8 +1371,8 @@ pub struct FlushBuckets {
 /// A message containing a list of [`Metric`]s to be inserted into the aggregator.
 #[derive(Debug)]
 pub struct InsertMetrics {
-    project_key: ProjectKey,
-    metrics: Vec<Metric>,
+    pub(crate) project_key: ProjectKey,
+    pub(crate) metrics: Vec<Metric>,
 }
 
 impl InsertMetrics {
@@ -1403,8 +1402,8 @@ impl InsertMetrics {
 /// A message containing a list of [`Bucket`]s to be inserted into the aggregator.
 #[derive(Debug)]
 pub struct MergeBuckets {
-    project_key: ProjectKey,
-    buckets: Vec<Bucket>,
+    pub(crate) project_key: ProjectKey,
+    pub(crate) buckets: Vec<Bucket>,
 }
 
 impl MergeBuckets {
@@ -1506,6 +1505,7 @@ impl FromMessage<MergeBuckets> for Aggregator {
 ///
 /// Receivers must implement a handler for the [`FlushBuckets`] message.
 pub struct AggregatorService {
+    name: String,
     config: AggregatorConfig,
     buckets: HashMap<BucketKey, QueuedBucket>,
     receiver: Option<Recipient<FlushBuckets, NoResponse>>,
@@ -1521,8 +1521,18 @@ impl AggregatorService {
     pub fn new(
         config: AggregatorConfig,
         receiver: Option<Recipient<FlushBuckets, NoResponse>>,
+    ) -> Self {
+        Self::named("default".to_owned(), config, receiver)
+    }
+
+    /// Like [`Self::new`], but with a provided name.
+    pub(crate) fn named(
+        name: String,
+        config: AggregatorConfig,
+        receiver: Option<Recipient<FlushBuckets, NoResponse>>,
     ) -> Self {
         Self {
+            name,
             config,
             buckets: HashMap::new(),
             receiver,
@@ -1703,6 +1713,7 @@ impl AggregatorService {
             Entry::Occupied(mut entry) => {
                 relay_statsd::metric!(
                     counter(MetricCounters::MergeHit) += 1,
+                    aggregator = &self.name,
                     metric_name = metric_name_tag(&entry.key().metric_name),
                 );
                 let bucket_value = &mut entry.get_mut().value;
@@ -1714,10 +1725,12 @@ impl AggregatorService {
             Entry::Vacant(entry) => {
                 relay_statsd::metric!(
                     counter(MetricCounters::MergeMiss) += 1,
+                    aggregator = &self.name,
                     metric_name = metric_name_tag(&entry.key().metric_name),
                 );
                 relay_statsd::metric!(
                     set(MetricSets::UniqueBucketsCreated) = entry.key().hash64() as i64, // 2-complement
+                    aggregator = &self.name,
                     metric_name = metric_name_tag(&entry.key().metric_name),
                 );
 
@@ -1743,6 +1756,7 @@ impl AggregatorService {
     ) -> Result<(), AggregateMetricsError> {
         relay_statsd::metric!(
             counter(MetricCounters::InsertMetric) += 1,
+            aggregator = &self.name,
             metric_type = metric.value.ty().as_str(),
         );
         let key = BucketKey {
@@ -1797,7 +1811,10 @@ impl AggregatorService {
     ///
     /// Note that this function is primarily intended for tests.
     pub fn pop_flush_buckets(&mut self) -> HashMap<ProjectKey, Vec<HashedBucket>> {
-        relay_statsd::metric!(gauge(MetricGauges::Buckets) = self.buckets.len() as u64);
+        relay_statsd::metric!(
+            gauge(MetricGauges::Buckets) = self.buckets.len() as u64,
+            aggregator = &self.name,
+        );
 
         // We only emit statsd metrics for the cost on flush (and not when merging the buckets),
         // assuming that this gives us more than enough data points.
@@ -1811,37 +1828,41 @@ impl AggregatorService {
 
         let mut stats = BTreeMap::new();
 
-        relay_statsd::metric!(timer(MetricTimers::BucketsScanDuration), {
-            let bucket_interval = self.config.bucket_interval;
-            let cost_tracker = &mut self.cost_tracker;
-            self.buckets.retain(|key, entry| {
-                if force || entry.elapsed() {
-                    // Take the value and leave a placeholder behind. It'll be removed right after.
-                    let value = mem::replace(&mut entry.value, BucketValue::Counter(0.0));
-                    cost_tracker.subtract_cost(key.project_key, key.cost());
-                    cost_tracker.subtract_cost(key.project_key, value.cost());
-
-                    let (bucket_count, item_count) = stats
-                        .entry((metric_type_tag(&value), metric_name_tag(&key.metric_name)))
-                        .or_insert((0usize, 0usize));
-                    *bucket_count += 1;
-                    *item_count += value.len();
-
-                    let bucket = Bucket::from_parts(key.clone(), bucket_interval, value);
-                    buckets
-                        .entry(key.project_key)
-                        .or_default()
-                        .push(HashedBucket {
-                            hashed_key: key.hash64(),
-                            bucket,
-                        });
-
-                    false
-                } else {
-                    true
-                }
-            });
-        });
+        relay_statsd::metric!(
+            timer(MetricTimers::BucketsScanDuration),
+            aggregator = &self.name,
+            {
+                let bucket_interval = self.config.bucket_interval;
+                let cost_tracker = &mut self.cost_tracker;
+                self.buckets.retain(|key, entry| {
+                    if force || entry.elapsed() {
+                        // Take the value and leave a placeholder behind. It'll be removed right after.
+                        let value = mem::replace(&mut entry.value, BucketValue::Counter(0.0));
+                        cost_tracker.subtract_cost(key.project_key, key.cost());
+                        cost_tracker.subtract_cost(key.project_key, value.cost());
+
+                        let (bucket_count, item_count) = stats
+                            .entry((metric_type_tag(&value), metric_name_tag(&key.metric_name)))
+                            .or_insert((0usize, 0usize));
+                        *bucket_count += 1;
+                        *item_count += value.len();
+
+                        let bucket = Bucket::from_parts(key.clone(), bucket_interval, value);
+                        buckets
+                            .entry(key.project_key)
+                            .or_default()
+                            .push(HashedBucket {
+                                hashed_key: key.hash64(),
+                                bucket,
+                            });
+
+                        false
+                    } else {
+                        true
+                    }
+                });
+            }
+        );
 
         for ((ty, name), (bucket_count, item_count)) in stats.into_iter() {
             relay_statsd::metric!(
@@ -1876,7 +1897,8 @@ impl AggregatorService {
 
             // Log the distribution of buckets over partition key
             relay_statsd::metric!(
-                histogram(MetricHistograms::PartitionKeys) = partition_key as f64
+                histogram(MetricHistograms::PartitionKeys) = partition_key as f64,
+                aggregator = &self.name,
             );
         }
         partitions
@@ -1895,13 +1917,15 @@ impl AggregatorService {
             .map(|batch| {
                 relay_statsd::metric!(
                     histogram(MetricHistograms::BucketsPerBatch) = batch.len() as f64,
+                    aggregator = &self.name,
                 );
                 process(batch);
             })
             .count();
 
         relay_statsd::metric!(
             histogram(MetricHistograms::BatchesPerPartition) = num_batches as f64,
+            aggregator = &self.name,
         );
     }
 
@@ -1923,7 +1947,8 @@ impl AggregatorService {
         for (project_key, project_buckets) in flush_buckets.into_iter() {
             let bucket_count = project_buckets.len() as u64;
             relay_statsd::metric!(
-                histogram(MetricHistograms::BucketsFlushedPerProject) = bucket_count
+                histogram(MetricHistograms::BucketsFlushedPerProject) = bucket_count,
+                aggregator = &self.name,
             );
             total_bucket_count += bucket_count;
 
@@ -1941,7 +1966,10 @@ impl AggregatorService {
                 });
             }
         }
-        relay_statsd::metric!(histogram(MetricHistograms::BucketsFlushed) = total_bucket_count);
+        relay_statsd::metric!(
+            histogram(MetricHistograms::BucketsFlushed) = total_bucket_count,
+            aggregator = &self.name,
+        );
     }
 
     fn handle_accepts_metrics(&self, sender: Sender<bool>) {
@@ -2007,9 +2035,8 @@ impl Service for AggregatorService {
         tokio::spawn(async move {
             let mut ticker = tokio::time::interval(FLUSH_INTERVAL);
             let mut shutdown = Controller::shutdown_handle();
-            relay_log::info!("aggregator started");
 
-            // Note that currently this loop never exists and will run till the tokio runtime shuts
+            // Note that currently this loop never exits and will run till the tokio runtime shuts
             // down. This is about to change with the refactoring for the shutdown process.
             loop {
                 tokio::select! {
@@ -2022,7 +2049,6 @@ impl Service for AggregatorService {
                     else => break,
                 }
             }
-            relay_log::info!("aggregator stopped");
         });
     }
 }
@@ -2033,7 +2059,8 @@ impl Drop for AggregatorService {
         if remaining_buckets > 0 {
             relay_log::error!("metrics aggregator dropping {remaining_buckets} buckets");
             relay_statsd::metric!(
-                counter(MetricCounters::BucketsDropped) += remaining_buckets as i64
+                counter(MetricCounters::BucketsDropped) += remaining_buckets as i64,
+                aggregator = &self.name,
             );
         }
     }
@@ -2064,8 +2091,6 @@ mod tests {
 
     #[derive(Clone, Default)]
     struct TestReceiver {
-        // TODO: Better way to communicate with service after it's started?
-        // Messages, maybe?
         data: Arc<RwLock<ReceivedData>>,
         reject_all: bool,
     }
@@ -2380,7 +2405,6 @@ mod tests {
         MetricValue::Distribution(2.0)
             .merge_into(&mut value)
             .unwrap();
-        // TODO: This should be ordered
         assert_eq!(value, BucketValue::Distribution(dist![1., 2., 3., 2.]));
     }
 
@@ -3124,8 +3148,8 @@ mod tests {
         let output = run_test_bucket_partitioning(None);
         insta::assert_debug_snapshot!(output, @r###"
         [
-            "metrics.buckets.per_batch:2|h",
-            "metrics.buckets.batches_per_partition:1|h",
+            "metrics.buckets.per_batch:2|h|#aggregator:default",
+            "metrics.buckets.batches_per_partition:1|h|#aggregator:default",
         ]
         "###);
     }
@@ -3138,17 +3162,17 @@ mod tests {
         let (partition_keys, tail) = output.split_at(2);
         insta::assert_debug_snapshot!(BTreeSet::from_iter(partition_keys), @r###"
         {
-            "metrics.buckets.partition_keys:59|h",
-            "metrics.buckets.partition_keys:62|h",
+            "metrics.buckets.partition_keys:59|h|#aggregator:default",
+            "metrics.buckets.partition_keys:62|h|#aggregator:default",
         }
         "###);
 
         insta::assert_debug_snapshot!(tail, @r###"
         [
-            "metrics.buckets.per_batch:1|h",
-            "metrics.buckets.batches_per_partition:1|h",
-            "metrics.buckets.per_batch:1|h",
-            "metrics.buckets.batches_per_partition:1|h",
+            "metrics.buckets.per_batch:1|h|#aggregator:default",
+            "metrics.buckets.batches_per_partition:1|h|#aggregator:default",
+            "metrics.buckets.per_batch:1|h|#aggregator:default",
+            "metrics.buckets.batches_per_partition:1|h|#aggregator:default",
         ]
         "###);
     }

relay-metrics/src/aggregation.rs

@@ -102,7 +102,9 @@
 
 mod aggregation;
 mod protocol;
+mod router;
 mod statsd;
 
 pub use aggregation::*;
 pub use protocol::*;
+pub use router::*;

relay-metrics/src/lib.rs

@@ -159,7 +159,7 @@ impl fmt::Display for ParseMetricError {
 /// Right now this successfully deserializes any kind of string, but in reality only `"sessions"`
 /// (for release health) and `"transactions"` (for metrics-enhanced performance) is supported.
 /// Everything else is dropped both in the metrics aggregator and in the store service.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub enum MetricNamespace {
     /// Metrics extracted from sessions.
     Sessions,