chore(deps): Bump sentry-arroyo to 2.47.0

sentry_streams/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 [dependencies]
 pyo3 = { version = "0.29.0" }
 serde = { version = "1.0", features = ["derive"] }
-sentry_arroyo = { version = ">=2.40.0, <2.44.0", features = ["ssl"] }
+sentry_arroyo = { version = "2.47.0", features = ["ssl"] }
 chrono = "0.4.40"
 tracing = "0.1.40"
 tracing-subscriber = "0.3.20"

sentry_streams/pyproject.toml

@@ -18,7 +18,7 @@ classifiers = [
 
 dependencies = [
     "requests>=2.32.3",
-    "sentry-arroyo>=2.39.2,<2.44.0",
+    "sentry-arroyo>=2.47.0",
     "pyyaml>=6.0.2",
     "jsonschema>=4.23.0",
     "sentry-kafka-schemas>=2.2.0",

sentry_streams/src/consumer.rs

@@ -231,7 +231,8 @@ pub fn build_dlq_policy(
 
             // Create Kafka producer for DLQ
             let producer_config = dlq_config.producer_config.clone().into();
-            let kafka_producer = KafkaProducer::new(producer_config);
+            let kafka_producer =
+                KafkaProducer::new(producer_config).expect("Failed to create DLQ Kafka producer");
             let dlq_producer = KafkaDlqProducer::new(kafka_producer, Topic::new(&dlq_config.topic));
 
             // Use default DLQ limits (no limits) and no max buffered messages

sentry_streams/src/metrics.rs

@@ -1,43 +1,73 @@
 use crate::metrics_config::PyMetricConfig;
-use metrics::Label;
+use metrics::{
+    Counter, Gauge, Histogram, Key, KeyName, Label, Metadata, Recorder, SharedString, Unit,
+};
 use metrics_exporter_dogstatsd::DogStatsDBuilder;
-use sentry_arroyo::metrics::{init as arroyo_init, Metric, MetricType, Recorder};
 use std::time::Duration;
-use tracing::{error, info, warn};
-
-struct MetricsFacadeRecorder;
-
-impl Recorder for MetricsFacadeRecorder {
-    fn record_metric(&self, metric: Metric<'_>) {
-        let key = format!("streams.pipeline.{}", metric.key);
-        let value_f64 = match metric.value {
-            sentry_arroyo::metrics::MetricValue::I64(v) => v as f64,
-            sentry_arroyo::metrics::MetricValue::U64(v) => v as f64,
-            sentry_arroyo::metrics::MetricValue::F64(v) => v,
-            sentry_arroyo::metrics::MetricValue::Duration(d) => d.as_millis() as f64,
-            _ => return,
-        };
+use tracing::{error, info};
 
-        let labels: Vec<(String, String)> = metric
-            .tags
-            .iter()
-            .map(|(key, value_dyn)| (key.to_string(), format!("{}", value_dyn)))
-            .collect();
+const ARROYO_METRIC_PREFIX: &str = "arroyo.";
+const STREAMS_METRIC_PREFIX: &str = "streams.pipeline.";
 
-        match metric.ty {
-            MetricType::Counter => {
-                let counter = metrics::counter!(key, &labels);
-                counter.increment(value_f64 as u64);
-            }
-            MetricType::Gauge => {
-                let gauge = metrics::gauge!(key, &labels);
-                gauge.set(value_f64);
-            }
-            MetricType::Timer => {
-                let histogram = metrics::histogram!(key, &labels);
-                histogram.record(value_f64);
-            }
-            _ => {}
+/// Arroyo emits its own metrics (`arroyo.*`) through the global metrics-rs
+/// recorder. This wrapper namespaces them under `streams.pipeline.` so they
+/// keep the same names as before and match the Python side, while leaving
+/// streams' own metrics (already `streams.pipeline.*`) untouched.
+struct ArroyoPrefixRecorder<R> {
+    inner: R,
+}
+
+fn prefixed_name(name: &str) -> Option<String> {
+    name.starts_with(ARROYO_METRIC_PREFIX)
+        .then(|| format!("{STREAMS_METRIC_PREFIX}{name}"))
+}
+
+fn prefix_key_name(key: KeyName) -> KeyName {
+    match prefixed_name(key.as_str()) {
+        Some(name) => KeyName::from(name),
+        None => key,
+    }
+}
+
+fn prefix_key(key: &Key) -> Option<Key> {
+    prefixed_name(key.name())
+        .map(|name| Key::from_parts(name, key.labels().cloned().collect::<Vec<_>>()))
+}
+
+impl<R: Recorder> Recorder for ArroyoPrefixRecorder<R> {
+    fn describe_counter(&self, key: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.inner
+            .describe_counter(prefix_key_name(key), unit, description)
+    }
+
+    fn describe_gauge(&self, key: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.inner
+            .describe_gauge(prefix_key_name(key), unit, description)
+    }
+
+    fn describe_histogram(&self, key: KeyName, unit: Option<Unit>, description: SharedString) {
+        self.inner
+            .describe_histogram(prefix_key_name(key), unit, description)
+    }
+
+    fn register_counter(&self, key: &Key, metadata: &Metadata<'_>) -> Counter {
+        match prefix_key(key) {
+            Some(key) => self.inner.register_counter(&key, metadata),
+            None => self.inner.register_counter(key, metadata),
+        }
+    }
+
+    fn register_gauge(&self, key: &Key, metadata: &Metadata<'_>) -> Gauge {
+        match prefix_key(key) {
+            Some(key) => self.inner.register_gauge(&key, metadata),
+            None => self.inner.register_gauge(key, metadata),
+        }
+    }
+
+    fn register_histogram(&self, key: &Key, metadata: &Metadata<'_>) -> Histogram {
+        match prefix_key(key) {
+            Some(key) => self.inner.register_histogram(&key, metadata),
+            None => self.inner.register_histogram(key, metadata),
         }
     }
 }
@@ -77,21 +107,52 @@ pub fn configure_metrics(metric_config: Option<PyMetricConfig>) {
             builder = builder.with_flush_interval(duration);
         }
 
-        match builder.install() {
+        let recorder = match builder.build() {
+            Ok(recorder) => recorder,
+            Err(e) => {
+                error!("Failed to build DogStatsD recorder: {}", e);
+                return;
+            }
+        };
+
+        match metrics::set_global_recorder(ArroyoPrefixRecorder { inner: recorder }) {
             Ok(_) => {
                 info!("Successfully initialized DogStatsD metrics exporter");
             }
             Err(e) => {
                 error!("Failed to install DogStatsD recorder: {}", e);
-                return;
             }
         }
+    }
+}
 
-        if arroyo_init(MetricsFacadeRecorder).is_err() {
-            warn!("Arroyo metrics recorder already initialized, skipping");
-            return;
-        }
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_prefixed_name() {
+        assert_eq!(
+            prefixed_name("arroyo.consumer.run.count").as_deref(),
+            Some("streams.pipeline.arroyo.consumer.run.count")
+        );
+        assert_eq!(prefixed_name("streams.pipeline.input.messages"), None);
+    }
 
-        info!("Successfully initialized arroyo metrics");
+    #[test]
+    fn test_prefix_key_keeps_labels() {
+        let key = Key::from_parts(
+            "arroyo.producer.produce_status",
+            vec![Label::new("status", "success")],
+        );
+        let prefixed = prefix_key(&key).unwrap();
+        assert_eq!(
+            prefixed.name(),
+            "streams.pipeline.arroyo.producer.produce_status"
+        );
+        assert_eq!(
+            prefixed.labels().collect::<Vec<_>>(),
+            key.labels().collect::<Vec<_>>()
+        );
     }
 }

sentry_streams/src/operators.rs

@@ -166,7 +166,8 @@ pub fn build(
             topic_name,
             kafka_config,
         } => {
-            let producer = KafkaProducer::new(kafka_config.clone().into());
+            let producer = KafkaProducer::new(kafka_config.clone().into())
+                .expect("Failed to create StreamSink Kafka producer");
             Box::new(StreamSink::new(
                 route.clone(),
                 producer,