diff --git a/Cargo.toml b/Cargo.toml index 2a9028fa..d8a20001 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ ssl = ["rdkafka/ssl"] [dependencies] chrono = "0.4.26" coarsetime = "0.1.33" +metrics = "0.24" once_cell = "1.18.0" rand = "0.8.5" rdkafka = { version = ">=0.37.0,<0.40", features = ["cmake-build", "tracing"] } diff --git a/rust-arroyo/src/backends/kafka/mod.rs b/rust-arroyo/src/backends/kafka/mod.rs index 55d21c77..d896e8ff 100644 --- a/rust-arroyo/src/backends/kafka/mod.rs +++ b/rust-arroyo/src/backends/kafka/mod.rs @@ -4,7 +4,6 @@ use super::CommitOffsets; use super::Consumer as ArroyoConsumer; use super::ConsumerError; use crate::backends::kafka::types::KafkaPayload; -use crate::gauge; use crate::types::{BrokerMessage, Partition, Topic}; use chrono::{DateTime, Utc}; use parking_lot::Mutex; @@ -218,18 +217,15 @@ impl ClientContext for CustomContext { } fn stats(&self, stats: Statistics) { - gauge!( - "arroyo.consumer.librdkafka.total_queue_size", - stats.replyq as u64, - ); + metrics::gauge!("arroyo.consumer.librdkafka.total_queue_size").set(stats.replyq as f64); for (topic_name, topic) in stats.topics.iter() { for (partition_num, partition) in topic.partitions.iter() { - gauge!( + metrics::gauge!( "arroyo.consumer.librdkafka.fetch_queue_count", - partition.fetchq_cnt as u64, - "topic" => topic_name, + "topic" => topic_name.clone(), "partition" => partition_num.to_string() - ); + ) + .set(partition.fetchq_cnt as f64); } } } diff --git a/rust-arroyo/src/backends/kafka/producer.rs b/rust-arroyo/src/backends/kafka/producer.rs index d25ea6bd..e9ff9d5e 100644 --- a/rust-arroyo/src/backends/kafka/producer.rs +++ b/rust-arroyo/src/backends/kafka/producer.rs @@ -5,8 +5,6 @@ use crate::backends::ProducerError; use crate::backends::{ AsyncProducer as ArroyoAsyncProducer, Producer as ArroyoProducer, ProducerFuture, }; -use crate::counter; -use crate::gauge; use crate::types::TopicOrPartition; use rdkafka::client::ClientContext; use rdkafka::config::ClientConfig; @@ -40,80 +38,80 @@ impl ClientContext for ProducerContext { // Record broker latency metrics if let Some(int_latency) = &broker_stats.int_latency { let p99_latency_ms = int_latency.p99 as f64 / 1000.0; - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.p99_int_latency", - p99_latency_ms as u64, "broker_id" => broker_id_str.clone(), - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(p99_latency_ms as u64 as f64); } if let Some(outbuf_latency) = &broker_stats.outbuf_latency { let p99_latency_ms = outbuf_latency.p99 as f64 / 1000.0; - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.p99_outbuf_latency", - p99_latency_ms as u64, "broker_id" => broker_id_str.clone(), - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(p99_latency_ms as u64 as f64); } if let Some(rtt) = &broker_stats.rtt { let p99_rtt_ms = rtt.p99 as f64 / 1000.0; - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.p99_rtt", - p99_rtt_ms as u64, "broker_id" => broker_id_str.clone(), - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(p99_rtt_ms as u64 as f64); } // Record broker transmission error metrics - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.broker_txerrs", - broker_stats.txerrs as i64, "broker_id" => broker_id_str.clone(), - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(broker_stats.txerrs as f64); - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.broker_txretries", - broker_stats.txretries as i64, "broker_id" => broker_id_str, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(broker_stats.txretries as f64); } // Record global producer metrics - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.message_count", - stats.msg_cnt as i64, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(stats.msg_cnt as f64); - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.message_count_max", - stats.msg_max as i64, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(stats.msg_max as f64); - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.message_size", - stats.msg_size as i64, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(stats.msg_size as f64); - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.message_size_max", - stats.msg_size_max as i64, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(stats.msg_size_max as f64); - gauge!( + metrics::gauge!( "arroyo.producer.librdkafka.reply_queue_size", - stats.replyq as i64, - "producer_name" => producer_name - ); + "producer_name" => producer_name.to_owned() + ) + .set(stats.replyq as f64); } } @@ -130,7 +128,12 @@ impl RdkafkaProducerContext for ProducerContext { Err((err, _)) => get_error_name(err), }; let producer_name = self.get_producer_name(); - counter!("arroyo.producer.produce_status", 1, "status" => result, "producer_name" => producer_name); + metrics::counter!( + "arroyo.producer.produce_status", + "status" => result, + "producer_name" => producer_name.to_owned() + ) + .increment(1); } } @@ -204,13 +207,25 @@ fn record_producer_error( let producer_error = ProducerError::ProducerFailure { error: error_name.clone(), }; - counter!("arroyo.producer.produce_status", 1, "status" => "error", "code" => error_name, "producer_name" => producer_name); + metrics::counter!( + "arroyo.producer.produce_status", + "status" => "error", + "code" => error_name, + "producer_name" => producer_name.to_owned() + ) + .increment(1); return producer_error; } let producer_error = ProducerError::ProducerFailure { error: default_error.to_string(), }; - counter!("arroyo.producer.produce_status", 1, "status" => "error", "code" => default_error, "producer_name" => producer_name); + metrics::counter!( + "arroyo.producer.produce_status", + "status" => "error", + "code" => default_error.to_owned(), + "producer_name" => producer_name.to_owned() + ) + .increment(1); producer_error } diff --git a/rust-arroyo/src/lib.rs b/rust-arroyo/src/lib.rs index a8f6e89c..23822a2a 100644 --- a/rust-arroyo/src/lib.rs +++ b/rust-arroyo/src/lib.rs @@ -1,5 +1,4 @@ pub mod backends; -pub mod metrics; pub mod processing; pub mod testutils; pub mod types; diff --git a/rust-arroyo/src/metrics/globals.rs b/rust-arroyo/src/metrics/globals.rs deleted file mode 100644 index 19764855..00000000 --- a/rust-arroyo/src/metrics/globals.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::sync::OnceLock; - -use super::Metric; - -/// The global [`Recorder`] which will receive [`Metric`] to be recorded. -pub trait Recorder { - /// Instructs the recorder to record the given [`Metric`]. - fn record_metric(&self, metric: Metric<'_>); -} - -impl Recorder for Box { - fn record_metric(&self, metric: Metric<'_>) { - (**self).record_metric(metric) - } -} - -static GLOBAL_RECORDER: OnceLock> = OnceLock::new(); - -/// Initialize the global [`Recorder`]. -/// -/// This will register the given `recorder` as the single global [`Recorder`] instance. -/// -/// This function can only be called once, and subsequent calls will return an -/// [`Err`] in case a global [`Recorder`] has already been initialized. -pub fn init(recorder: R) -> Result<(), R> { - let mut result = Err(recorder); - { - let result = &mut result; - let _ = GLOBAL_RECORDER.get_or_init(|| { - let recorder = std::mem::replace(result, Ok(())).unwrap_err(); - Box::new(recorder) - }); - } - result -} - -/// Records a [`Metric`] with the globally configured [`Recorder`]. -/// -/// This function will be a noop in case no global [`Recorder`] is configured. -pub fn record_metric(metric: Metric<'_>) { - if let Some(recorder) = GLOBAL_RECORDER.get() { - recorder.record_metric(metric) - } -} diff --git a/rust-arroyo/src/metrics/macros.rs b/rust-arroyo/src/metrics/macros.rs deleted file mode 100644 index 9e47500d..00000000 --- a/rust-arroyo/src/metrics/macros.rs +++ /dev/null @@ -1,62 +0,0 @@ -/// Create a [`Metric`]. -/// -/// Instead of creating metrics directly, it is recommended to immediately record -/// metrics using the [`counter!`], [`gauge!`] or [`distribution!`] macros. -/// -/// This is the recommended way to create a [`Metric`], as the -/// implementation details of it might change. -/// -/// [`Metric`]: crate::metrics::Metric -#[macro_export] -macro_rules! metric { - ($ty:ident: $key:expr, $value:expr - $(, $($tag_key:expr => $tag_val:expr),*)? - ) => {{ - $crate::metrics::Metric { - key: &$key, - ty: $crate::metrics::MetricType::$ty, - - tags: &[ - $($(($tag_key, &$tag_val),)*)? - ], - value: $value.into(), - - __private: (), - } - }}; -} - -/// Records a counter [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder). -#[macro_export] -macro_rules! counter { - ($expr:expr) => { - $crate::__record_metric!(Counter: $expr, 1); - }; - ($($tt:tt)+) => { - $crate::__record_metric!(Counter: $($tt)+); - }; -} - -/// Records a gauge [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder). -#[macro_export] -macro_rules! gauge { - ($($tt:tt)+) => { - $crate::__record_metric!(Gauge: $($tt)+); - }; -} - -/// Records a timer [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder). -#[macro_export] -macro_rules! timer { - ($($tt:tt)+) => { - $crate::__record_metric!(Timer: $($tt)+); - }; -} - -#[macro_export] -#[doc(hidden)] -macro_rules! __record_metric { - ($($tt:tt)+) => {{ - $crate::metrics::record_metric($crate::metric!($($tt)+)); - }}; -} diff --git a/rust-arroyo/src/metrics/mod.rs b/rust-arroyo/src/metrics/mod.rs deleted file mode 100644 index 975426c3..00000000 --- a/rust-arroyo/src/metrics/mod.rs +++ /dev/null @@ -1,8 +0,0 @@ -mod globals; -mod macros; -mod statsd; -mod types; - -pub use globals::*; -pub use statsd::*; -pub use types::*; diff --git a/rust-arroyo/src/metrics/statsd.rs b/rust-arroyo/src/metrics/statsd.rs deleted file mode 100644 index 55a09f45..00000000 --- a/rust-arroyo/src/metrics/statsd.rs +++ /dev/null @@ -1,112 +0,0 @@ -use std::cell::RefCell; -use std::fmt::{Debug, Display, Write}; - -use super::{Metric, Recorder}; - -thread_local! { - static STRING_BUFFER: RefCell = const { RefCell::new(String::new()) }; -} - -/// A generic sink used by the [`StatsdRecorder`]. -pub trait MetricSink { - /// Emits a StatsD-formatted `metric`. - fn emit(&self, metric: &str); -} - -/// A recorder emitting StatsD-formatted [`Metric`]s to a configured [`MetricSink`]. -pub struct StatsdRecorder { - prefix: String, - sink: S, - tags: String, -} - -impl Debug for StatsdRecorder { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("StatsdRecorder") - .field("prefix", &self.prefix) - .field("formatted tags", &self.tags) - .finish_non_exhaustive() - } -} - -impl StatsdRecorder { - /// Creates a new Recorder with the given `prefix` and `sink`. - /// - /// The recorder will emit [`Metric`]s to formatted in `statsd` format to the - /// configured [`MetricSink`]. - pub fn new(prefix: &str, sink: S) -> Self { - let prefix = if prefix.is_empty() { - String::new() - } else { - format!("{}.", prefix.trim_end_matches('.')) - }; - Self { - prefix, - sink, - tags: String::new(), - } - } - - /// Add a global tag (as key/value) to this Recorder. - pub fn with_tag(mut self, key: &'static str, value: impl Display) -> Self { - let t = &mut self.tags; - if t.is_empty() { - t.push_str("|#"); - } else { - t.push(','); - } - - t.push_str(key); - t.push(':'); - let _ = write!(t, "{value}"); - - self - } - - fn write_metric(&self, metric: Metric<'_>, s: &mut String) { - s.push_str(&self.prefix); - metric.write_base_metric(s); - - s.push_str(&self.tags); - if !metric.tags.is_empty() { - if self.tags.is_empty() { - s.push_str("|#"); - } else { - s.push(','); - } - - metric.write_tags(s); - } - } -} - -impl Recorder for StatsdRecorder { - fn record_metric(&self, metric: Metric<'_>) { - STRING_BUFFER.with_borrow_mut(|s| { - s.clear(); - s.reserve(256); - - self.write_metric(metric, s); - - self.sink.emit(s); - }); - } -} - -impl Metric<'_> { - pub(crate) fn write_base_metric(&self, s: &mut String) { - let _ = write!(s, "{}:{}|", self.key, self.value); - s.push_str(self.ty.as_str()); - } - - pub(crate) fn write_tags(&self, s: &mut String) { - for (i, &(key, value)) in self.tags.iter().enumerate() { - if i > 0 { - s.push(','); - } - s.push_str(key); - s.push(':'); - let _ = write!(s, "{value}"); - } - } -} diff --git a/rust-arroyo/src/metrics/types.rs b/rust-arroyo/src/metrics/types.rs deleted file mode 100644 index 0cae7dbc..00000000 --- a/rust-arroyo/src/metrics/types.rs +++ /dev/null @@ -1,110 +0,0 @@ -use core::fmt::{self, Display}; -use std::time::Duration; - -/// The Type of a Metric. -/// -/// Counters, Gauges and Distributions are supported, -/// with more types to be added later. -#[non_exhaustive] -#[derive(Debug)] -pub enum MetricType { - /// A counter metric, using the StatsD `c` type. - Counter, - /// A gauge metric, using the StatsD `g` type. - Gauge, - /// A timer metric, using the StatsD `ms` type. - Timer, - // Distribution, - // Meter, - // Histogram, - // Set, -} - -impl MetricType { - /// Returns the StatsD metrics type. - pub fn as_str(&self) -> &str { - match self { - MetricType::Counter => "c", - MetricType::Gauge => "g", - MetricType::Timer => "ms", - } - } -} - -impl Display for MetricType { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(self.as_str()) - } -} - -/// A Metric Value. -/// -/// This supports various numeric values for now, but might gain support for -/// `Duration` and other types later on. -#[non_exhaustive] -#[derive(Debug)] -pub enum MetricValue { - /// A signed value. - I64(i64), - /// An unsigned value. - U64(u64), - /// A floating-point value. - F64(f64), - /// A [`Duration`] value. - Duration(Duration), -} - -impl Display for MetricValue { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - MetricValue::I64(v) => v.fmt(f), - MetricValue::U64(v) => v.fmt(f), - MetricValue::F64(v) => v.fmt(f), - MetricValue::Duration(d) => d.as_millis().fmt(f), - } - } -} - -macro_rules! into_metric_value { - ($($from:path),+ => $variant:ident) => { - $( - impl From<$from> for MetricValue { - #[inline(always)] - fn from(f: $from) -> Self { - Self::$variant(f.into()) - } - } - )+ - }; -} - -into_metric_value!(i8, i16, i32, i64 => I64); -into_metric_value!(u8, u16, u32, u64 => U64); -into_metric_value!(f32, f64 => F64); -into_metric_value!(Duration => Duration); -into_metric_value!(coarsetime::Duration => Duration); - -/// An alias for a list of Metric tags. -pub type MetricTags<'a> = &'a [(&'static str, &'a dyn Display)]; - -/// A fully types Metric. -/// -/// Most importantly, the metric has a [`ty`](MetricType), a `key` and a [`value`](MetricValue). -/// It can also have a list of [`tags`](MetricTags). -/// -/// This struct might change in the future, and one should construct it via -/// the [`metric!`](crate::metric) macro instead. -pub struct Metric<'a> { - /// The key, or name, of the metric. - pub key: &'a dyn Display, - /// The type of metric. - pub ty: MetricType, - - /// A list of tags for this metric. - pub tags: MetricTags<'a>, - /// The metrics value. - pub value: MetricValue, - - #[doc(hidden)] - pub __private: (), -} diff --git a/rust-arroyo/src/processing/dlq.rs b/rust-arroyo/src/processing/dlq.rs index eaa90f14..7801bb27 100644 --- a/rust-arroyo/src/processing/dlq.rs +++ b/rust-arroyo/src/processing/dlq.rs @@ -11,8 +11,6 @@ use tokio::task::JoinHandle; use crate::backends::kafka::producer::KafkaProducer; use crate::backends::kafka::types::KafkaPayload; use crate::backends::Producer; -use crate::counter; -use crate::gauge; use crate::processing::strategies::InvalidMessageReason; use crate::types::{BrokerMessage, Partition, Topic, TopicOrPartition}; @@ -99,7 +97,7 @@ impl DlqProducer for KafkaDlqProducer { Box::pin(async move { if let Err(err) = producer.produce(&topic, payload) { - counter!("arroyo.consumer.dlq.produce_error", 1); + metrics::counter!("arroyo.consumer.dlq.produce_error").increment(1); tracing::error!("Failed to produce to DLQ: {:?}", err); } @@ -428,20 +426,17 @@ impl BufferedMessages { } // Number of partitions in the buffer map - gauge!( - "arroyo.consumer.dlq_buffer.assigned_partitions", - self.buffered_messages.len() as u64, - ); + metrics::gauge!("arroyo.consumer.dlq_buffer.assigned_partitions") + .set(self.buffered_messages.len() as f64); let buffered = self.buffered_messages.entry(message.partition).or_default(); if let Some(max) = self.max_per_partition { if buffered.len() >= max { - counter!( + metrics::counter!( "arroyo.consumer.dlq_buffer.exceeded", - 1, - - "partition_id" => message.partition.index - ); + "partition_id" => message.partition.index.to_string() + ) + .increment(1); buffered.pop_front(); } } @@ -451,27 +446,25 @@ impl BufferedMessages { } fn report_partition_metrics(partition_index: u16, buffered: &VecDeque) { - gauge!( + metrics::gauge!( "arroyo.consumer.dlq_buffer.capacity", - buffered.capacity() as u64, - "partition_id" => partition_index - ); + "partition_id" => partition_index.to_string() + ) + .set(buffered.capacity() as f64); - gauge!( + metrics::gauge!( "arroyo.consumer.dlq_buffer.len", - buffered.len() as u64, - "partition_id" => partition_index - ); + "partition_id" => partition_index.to_string() + ) + .set(buffered.len() as f64); } /// Return the message at the given offset or None if it is not found in the buffer. /// Messages up to the offset for the given partition are removed. pub fn pop(&mut self, partition: &Partition, offset: u64) -> Option> { // Number of partitions in the buffer map - gauge!( - "arroyo.consumer.dlq_buffer.assigned_partitions", - self.buffered_messages.len() as u64, - ); + metrics::gauge!("arroyo.consumer.dlq_buffer.assigned_partitions") + .set(self.buffered_messages.len() as f64); let messages = self.buffered_messages.get_mut(partition)?; while let Some(message) = messages.front() { diff --git a/rust-arroyo/src/processing/metrics_buffer.rs b/rust-arroyo/src/processing/metrics_buffer.rs index 984a22e0..f20e9e96 100644 --- a/rust-arroyo/src/processing/metrics_buffer.rs +++ b/rust-arroyo/src/processing/metrics_buffer.rs @@ -1,4 +1,3 @@ -use crate::timer; use crate::utils::timing::Deadline; use core::fmt::Debug; use std::collections::BTreeMap; @@ -38,7 +37,7 @@ impl MetricsBuffer { pub fn flush(&mut self) { let timers = mem::take(&mut self.timers); for (metric, duration) in timers { - timer!(&metric, duration); + metrics::histogram!(metric).record(duration.as_millis() as f64); } self.flush_deadline.restart(); diff --git a/rust-arroyo/src/processing/mod.rs b/rust-arroyo/src/processing/mod.rs index 2238c18f..ee971d87 100644 --- a/rust-arroyo/src/processing/mod.rs +++ b/rust-arroyo/src/processing/mod.rs @@ -17,7 +17,6 @@ use crate::processing::strategies::{ }; use crate::types::{InnerMessage, Message, Partition, Topic}; use crate::utils::timing::Deadline; -use crate::{counter, timer}; pub mod dlq; mod metrics_buffer; @@ -122,10 +121,8 @@ impl AssignmentCallbacks for Callbacks) { tracing::info!("New partitions assigned: {:?}", partitions); - counter!( - "arroyo.consumer.partitions_assigned.count", - partitions.len() as i64 - ); + metrics::counter!("arroyo.consumer.partitions_assigned.count") + .increment(partitions.len() as u64); let start = coarsetime::Instant::recent(); @@ -134,18 +131,14 @@ impl AssignmentCallbacks for Callbacks(&self, commit_offsets: C, partitions: Vec) { tracing::info!("Partitions to revoke: {:?}", partitions); - counter!( - "arroyo.consumer.partitions_revoked.count", - partitions.len() as i64, - ); + metrics::counter!("arroyo.consumer.partitions_revoked.count") + .increment(partitions.len() as u64); let start = coarsetime::Instant::recent(); @@ -185,7 +178,8 @@ impl AssignmentCallbacks for Callbacks StreamProcessor { } fn _run_once(&mut self) -> Result<(), RunError> { - counter!("arroyo.consumer.run.count"); + metrics::counter!("arroyo.consumer.run.count").increment(1); let consumer_is_paused = self.consumer_state.is_paused(); if consumer_is_paused { diff --git a/rust-arroyo/src/processing/strategies/commit_offsets.rs b/rust-arroyo/src/processing/strategies/commit_offsets.rs index ccf9a931..4f96907d 100644 --- a/rust-arroyo/src/processing/strategies/commit_offsets.rs +++ b/rust-arroyo/src/processing/strategies/commit_offsets.rs @@ -4,7 +4,6 @@ use std::time::Duration; use chrono::Utc; use crate::processing::strategies::{CommitRequest, ProcessingStrategy, SubmitError}; -use crate::timer; use crate::types::{Message, Partition}; use super::StrategyError; @@ -57,9 +56,11 @@ impl ProcessingStrategy for CommitOffsets { if now - self.last_record_time > coarsetime::Duration::from_secs(1) { if let Some(timestamp) = message.timestamp() { // FIXME: this used to be in seconds - timer!( - "arroyo.consumer.latency", - (Utc::now() - timestamp).to_std().unwrap_or_default() + metrics::histogram!("arroyo.consumer.latency").record( + (Utc::now() - timestamp) + .to_std() + .unwrap_or_default() + .as_millis() as f64, ); self.last_record_time = now; } diff --git a/rust-arroyo/src/processing/strategies/healthcheck.rs b/rust-arroyo/src/processing/strategies/healthcheck.rs index 0dc1dd5c..63f11f95 100644 --- a/rust-arroyo/src/processing/strategies/healthcheck.rs +++ b/rust-arroyo/src/processing/strategies/healthcheck.rs @@ -1,7 +1,6 @@ use std::path::PathBuf; use std::time::{Duration, SystemTime}; -use crate::counter; use crate::processing::strategies::{ CommitRequest, ProcessingStrategy, StrategyError, SubmitError, }; @@ -40,7 +39,7 @@ impl HealthCheck { tracing::error!(error); } - counter!("arroyo.processing.strategies.healthcheck.touch"); + metrics::counter!("arroyo.processing.strategies.healthcheck.touch").increment(1); self.deadline = now + self.interval; } } diff --git a/rust-arroyo/src/processing/strategies/reduce.rs b/rust-arroyo/src/processing/strategies/reduce.rs index 3d28a50e..3ac2a76f 100644 --- a/rust-arroyo/src/processing/strategies/reduce.rs +++ b/rust-arroyo/src/processing/strategies/reduce.rs @@ -2,7 +2,6 @@ use crate::processing::strategies::{ merge_commit_request, CommitRequest, MessageRejected, ProcessingStrategy, StrategyError, SubmitError, }; -use crate::timer; use crate::types::{Message, Partition}; use crate::utils::timing::Deadline; use std::collections::BTreeMap; @@ -206,11 +205,11 @@ impl Reduce { "force" }; - timer!( + metrics::histogram!( "arroyo.strategies.reduce.batch_time.ms", - batch_time, "flush_reason" => flush_reason - ); + ) + .record(batch_time.as_millis() as f64); let batch_state = mem::replace( &mut self.batch_state, diff --git a/rust-arroyo/src/processing/strategies/run_task_in_threads.rs b/rust-arroyo/src/processing/strategies/run_task_in_threads.rs index 4fbc6e6b..a0b83028 100644 --- a/rust-arroyo/src/processing/strategies/run_task_in_threads.rs +++ b/rust-arroyo/src/processing/strategies/run_task_in_threads.rs @@ -12,7 +12,6 @@ use crate::processing::strategies::{ }; use crate::types::Message; use crate::utils::timing::Deadline; -use crate::{counter, gauge, timer}; use super::StrategyError; @@ -140,21 +139,27 @@ where self.commit_request_carried_over = merge_commit_request(self.commit_request_carried_over.take(), commit_request); - gauge!("arroyo.strategies.run_task_in_threads.threads", - self.handles.len() as u64, + metrics::gauge!( + "arroyo.strategies.run_task_in_threads.threads", "strategy_name" => self.metric_strategy_name - ); - gauge!("arroyo.strategies.run_task_in_threads.concurrency", - self.concurrency as u64, + ) + .set(self.handles.len() as f64); + metrics::gauge!( + "arroyo.strategies.run_task_in_threads.concurrency", "strategy_name" => self.metric_strategy_name - ); + ) + .set(self.concurrency as f64); if let Some(message) = self.message_carried_over.take() { match self.next_step.submit(message) { Err(SubmitError::MessageRejected(MessageRejected { message: transformed_message, })) => { - counter!("arroyo.strategies.run_task_in_threads.got_backpressure", 1, "strategy_name" => self.metric_strategy_name); + metrics::counter!( + "arroyo.strategies.run_task_in_threads.got_backpressure", + "strategy_name" => self.metric_strategy_name + ) + .increment(1); self.message_carried_over = Some(transformed_message); } Err(SubmitError::InvalidMessage(invalid_message)) => { @@ -203,7 +208,11 @@ where fn submit(&mut self, message: Message) -> Result<(), SubmitError> { if self.message_carried_over.is_some() { - counter!("arroyo.strategies.run_task_in_threads.giving_backpressure", 1, "strategy_name" => self.metric_strategy_name); + metrics::counter!( + "arroyo.strategies.run_task_in_threads.giving_backpressure", + "strategy_name" => self.metric_strategy_name + ) + .increment(1); return Err(SubmitError::MessageRejected(MessageRejected { message })); } @@ -251,11 +260,11 @@ where } self.handles.clear(); - timer!( + metrics::histogram!( "arroyo.strategies.run_task_in_threads.join_time", - start.elapsed(), "strategy_name" => self.metric_strategy_name - ); + ) + .record(start.elapsed().as_millis() as f64); let next_commit = self.next_step.join(deadline.map(|d| d.remaining()))?;