ref(statsd): Use statsdproxy to pre-aggregate metrics in-memory

CHANGELOG.md

@@ -15,6 +15,7 @@
 
 - Add quotas to global config. ([#3086](https://github.com/getsentry/relay/pull/3086))
 - Adds support for dynamic metric bucket encoding. ([#3137](https://github.com/getsentry/relay/pull/3137))
+- Use statsdproxy to pre-aggregate metrics. ([#2425](https://github.com/getsentry/relay/pull/2425))
 
 ## 24.2.0
 

relay-config/src/config.rs

@@ -497,6 +497,10 @@ struct Metrics {
     ///
     /// Defaults to `true`.
     buffering: bool,
+    /// Emitted metrics will be aggregated to optimize bandwidth.
+    ///
+    /// Defaults to `true`.
+    aggregation: bool,
     /// Global sample rate for all emitted metrics between `0.0` and `1.0`.
     ///
     /// For example, a value of `0.3` means that only 30% of the emitted metrics will be sent.
@@ -512,6 +516,7 @@ impl Default for Metrics {
             default_tags: BTreeMap::new(),
             hostname_tag: None,
             buffering: true,
+            aggregation: true,
             sample_rate: 1.0,
         }
     }
@@ -1823,6 +1828,11 @@ impl Config {
         self.values.metrics.buffering
     }
 
+    /// Returns true if metrics aggregation is enabled, false otherwise.
+    pub fn metrics_aggregation(&self) -> bool {
+        self.values.metrics.aggregation
+    }
+
     /// Returns the global sample rate for all metrics.
     pub fn metrics_sample_rate(&self) -> f32 {
         self.values.metrics.sample_rate

relay-statsd/Cargo.toml

@@ -15,6 +15,7 @@ crossbeam-channel = "0.5.6"
 parking_lot = { workspace = true }
 rand = { workspace = true }
 relay-log = { path = "../relay-log" }
+statsdproxy = { version = "0.1.2", features = ["cadence-adapter"], default-features = false }
 
 [features]
 default = []

relay-statsd/src/lib.rs

@@ -23,7 +23,7 @@
 //! ```no_run
 //! # use std::collections::BTreeMap;
 //!
-//! relay_statsd::init("myprefix", "localhost:8125", BTreeMap::new(), true, 1.0);
+//! relay_statsd::init("myprefix", "localhost:8125", BTreeMap::new(), true, true, 1.0);
 //! ```
 //!
 //! ## Macro Usage
@@ -66,6 +66,8 @@ use cadence::{
 };
 use parking_lot::RwLock;
 use rand::distributions::{Distribution, Uniform};
+use statsdproxy::cadence::StatsdProxyMetricSink;
+use statsdproxy::config::AggregateMetricsConfig;
 
 /// Maximum number of metric events that can be queued before we start dropping them
 const METRICS_MAX_QUEUE_SIZE: usize = 100_000;
@@ -226,6 +228,7 @@ pub fn init<A: ToSocketAddrs>(
     host: A,
     default_tags: BTreeMap<String, String>,
     buffering: bool,
+    aggregating: bool,
     sample_rate: f32,
 ) {
     let addrs: Vec<_> = host.to_socket_addrs().unwrap().collect();
@@ -247,7 +250,24 @@ pub fn init<A: ToSocketAddrs>(
     let socket = UdpSocket::bind("0.0.0.0:0").unwrap();
     socket.set_nonblocking(true).unwrap();
 
-    let statsd_client = if buffering {
+    let statsd_client = if aggregating {
+        let host = host.to_socket_addrs().unwrap().next().unwrap();
+        let statsdproxy_sink = StatsdProxyMetricSink::new(move || {
+            let next_step = statsdproxy::middleware::upstream::Upstream::new(host)
+                .expect("failed to create statsdproxy metric sink");
+            statsdproxy::middleware::aggregate::AggregateMetrics::new(
+                AggregateMetricsConfig {
+                    aggregate_gauges: true,
+                    aggregate_counters: true,
+                    flush_interval: 1,
+                    flush_offset: 0,
+                    max_map_size: None,
+                },
+                next_step,
+            )
+        });
+        StatsdClient::from_sink(prefix, statsdproxy_sink)
+    } else if buffering {
         let udp_sink = BufferedUdpMetricSink::from(host, socket).unwrap();
         let queuing_sink = QueuingMetricSink::with_capacity(udp_sink, METRICS_MAX_QUEUE_SIZE);
         StatsdClient::from_sink(prefix, queuing_sink)

relay/src/setup.rs

@@ -69,6 +69,7 @@ pub fn init_metrics(config: &Config) -> Result<()> {
         &addrs[..],
         default_tags,
         config.metrics_buffering(),
+        config.metrics_aggregation(),
         config.metrics_sample_rate(),
     );