From 53c69f71c1fe06c3b07359f0adcfc30174675b80 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:26:14 +0000 Subject: [PATCH] chore(deps): Bump sentry-arroyo to 2.47.0 Bump both the Python package and the Rust crate. Arroyo 2.44.0 moved the Rust crate to metrics-rs and removed its own recorder API, so drop the arroyo Recorder adapter and instead wrap the DogStatsD recorder to keep arroyo metrics under the streams.pipeline namespace. Kafka producer constructors now return Result; keep the previous panic-on-failure behavior with expect(). Co-Authored-By: Markus Unterwaditzer --- sentry_streams/Cargo.lock | 5 +- sentry_streams/Cargo.toml | 2 +- sentry_streams/pyproject.toml | 2 +- sentry_streams/src/consumer.rs | 3 +- sentry_streams/src/metrics.rs | 145 +++++++++++++++++++++++--------- sentry_streams/src/operators.rs | 3 +- sentry_streams/uv.lock | 10 +-- 7 files changed, 117 insertions(+), 53 deletions(-) diff --git a/sentry_streams/Cargo.lock b/sentry_streams/Cargo.lock index 05fcefe3..fb9023f3 100644 --- a/sentry_streams/Cargo.lock +++ b/sentry_streams/Cargo.lock @@ -2431,12 +2431,13 @@ dependencies = [ [[package]] name = "sentry_arroyo" -version = "2.40.0" +version = "2.47.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ead64064bc67db796b37b2ad465c6798a80343a11fc23186cb78f06c116added" +checksum = "7529427a38f7928d90896498d6872c60a3797e3a523fa5334afbf596d7146f09" dependencies = [ "chrono", "coarsetime", + "metrics", "once_cell", "parking_lot", "rand 0.8.7", diff --git a/sentry_streams/Cargo.toml b/sentry_streams/Cargo.toml index 06c628f8..62d3db5b 100644 --- a/sentry_streams/Cargo.toml +++ b/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" diff --git a/sentry_streams/pyproject.toml b/sentry_streams/pyproject.toml index 1b430bb5..111e31aa 100644 --- a/sentry_streams/pyproject.toml +++ b/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", diff --git a/sentry_streams/src/consumer.rs b/sentry_streams/src/consumer.rs index 5ce4c884..2393c60f 100644 --- a/sentry_streams/src/consumer.rs +++ b/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 diff --git a/sentry_streams/src/metrics.rs b/sentry_streams/src/metrics.rs index 8749dc2f..62a08377 100644 --- a/sentry_streams/src/metrics.rs +++ b/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 { + inner: R, +} + +fn prefixed_name(name: &str) -> Option { + 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 { + prefixed_name(key.name()) + .map(|name| Key::from_parts(name, key.labels().cloned().collect::>())) +} + +impl Recorder for ArroyoPrefixRecorder { + fn describe_counter(&self, key: KeyName, unit: Option, description: SharedString) { + self.inner + .describe_counter(prefix_key_name(key), unit, description) + } + + fn describe_gauge(&self, key: KeyName, unit: Option, description: SharedString) { + self.inner + .describe_gauge(prefix_key_name(key), unit, description) + } + + fn describe_histogram(&self, key: KeyName, unit: Option, 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) { 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::>(), + key.labels().collect::>() + ); } } diff --git a/sentry_streams/src/operators.rs b/sentry_streams/src/operators.rs index 9b174c38..8132b3e9 100644 --- a/sentry_streams/src/operators.rs +++ b/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, diff --git a/sentry_streams/uv.lock b/sentry_streams/uv.lock index b01c10fd..0411006d 100644 --- a/sentry_streams/uv.lock +++ b/sentry_streams/uv.lock @@ -1154,14 +1154,14 @@ wheels = [ [[package]] name = "sentry-arroyo" -version = "2.40.3" +version = "2.47.0" source = { registry = "https://sfw.security.sentry.io/pypi/simple" } dependencies = [ { name = "confluent-kafka" }, ] -sdist = { url = "https://sfw.security.sentry.io/pypi/packages/3b/cd/9e37e2698f347081db478f043e7d8cda34afb33f0607c812b821ed8c736c/sentry_arroyo-2.40.3.tar.gz", hash = "sha256:be698f6b0549338911a6b69d99d21c01be9c6f68d2d88c0b824a8e40df74cbf2", size = 93992, upload-time = "2026-06-18T19:36:40.04Z" } +sdist = { url = "https://sfw.security.sentry.io/pypi/packages/ef/4b/98bb4cea86ba3ac0e9d9cf69c4a625e49a4fc9473a7fc091efc113cf166b/sentry_arroyo-2.47.0.tar.gz", hash = "sha256:65bc18aa4df8877ad9af95c353573e7986b3211d5d3870e3d3256e6eedcfb979", size = 100424, upload-time = "2026-10-19T13:58:24.075Z" } wheels = [ - { url = "https://sfw.security.sentry.io/pypi/packages/09/11/e82645edc7f2fd8fc97301102a9f189d0ecb2702635f30b72954adceda91/sentry_arroyo-2.40.3-py3-none-any.whl", hash = "sha256:d10b9767cbb572de86b8a41b3894bcb371d112ac5d6f70cd66f35cabdbf6ea9b", size = 121816, upload-time = "2026-06-18T19:36:38.906Z" }, + { url = "https://sfw.security.sentry.io/pypi/packages/59/d1/9ab7d10d40ed3576c9b67ba307ff7349cf1106f4737fe385476b39f7ed9a/sentry_arroyo-2.47.0-py3-none-any.whl", hash = "sha256:8839a5ab50898268757b985c4ec8f73896cfff3cf89aa5f46070185ec197b397", size = 128711, upload-time = "2026-10-19T13:58:22.768Z" }, ] [[package]] @@ -1223,7 +1223,7 @@ wheels = [ [[package]] name = "sentry-streams" -version = "0.0.65" +version = "0.0.67" source = { editable = "." } dependencies = [ { name = "click" }, @@ -1264,7 +1264,7 @@ requires-dist = [ { name = "polars", specifier = ">=1.31.0,<1.32" }, { name = "pyyaml", specifier = ">=6.0.2" }, { name = "requests", specifier = ">=2.32.3" }, - { name = "sentry-arroyo", specifier = ">=2.39.2" }, + { name = "sentry-arroyo", specifier = ">=2.47.0" }, { name = "sentry-kafka-schemas", specifier = ">=2.2.0" }, ]