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" }, ]