Used labels and service while calling metrics-rs macros

src/metrics/perf_event.rs

@@ -1,6 +1,5 @@
 use crate::prelude::alloc::fmt::Formatter;
-use metrics::register_histogram;
-use perf_event::{events::Hardware, Counter, Group};
+use perf_event::events::Hardware;
 use serde::Deserialize;
 use std::fmt;
 
@@ -66,26 +65,3 @@ impl PerfEvents {
  self.counters.push(hardware_metric_kind);
  }
 }
-
-pub struct HardwareMetricGroup {
- pub group: Group,
- pub counters: Vec<(String, Counter)>,
-}
-
-impl HardwareMetricGroup {
- pub(crate) fn register_hardware_metric_gauges(
- &mut self,
- node_name: String,
- perf_events: PerfEvents,
- ) -> std::io::Result<()> {
- let iterator = perf_events.counters.iter();
- for value in iterator {
- register_histogram!(self.get_field_gauge_name(&node_name, &value.to_string()));
- }
- self.group.enable()
- }
-
- pub fn get_field_gauge_name(&self, field_name: &str, node_name: &str) -> String {
- format!("{}_{}", node_name, field_name)
- }
-}

src/metrics/runtime_metrics.rs

@@ -1,18 +1,12 @@
 use crate::metrics::meter::Meter;
-use metrics::{register_counter, register_gauge, register_histogram};
 
 #[derive(Clone, Debug, Default)]
 pub struct NodeMetrics {
  pub inbound_throughput: Meter,
 }
 
 impl NodeMetrics {
- pub fn new(node_name: &str) -> NodeMetrics {
- register_gauge!(format!("{}_{}", node_name, "inbound_throughput"));
- register_counter!(format!("{}_{}", node_name, "epoch_counter"));
- register_counter!(format!("{}_{}", node_name, "watermark_counter"));
- register_histogram!(format!("{}_{}", node_name, "batch_execution_time"));
-
+ pub fn new() -> NodeMetrics {
  NodeMetrics {
  inbound_throughput: Meter::new(),
  }
@@ -22,9 +16,7 @@ pub struct SourceMetrics {
  pub incoming_message_rate: Meter,
 }
 impl SourceMetrics {
- pub fn new(source_node_name: &str) -> SourceMetrics {
- register_gauge!(format!("{}_{}", source_node_name, "incoming_message_rate"));
- register_counter!(format!("{}_{}", source_node_name, "error_counter"));
+ pub fn new() -> SourceMetrics {
  SourceMetrics {
  incoming_message_rate: Meter::new(),
  }

src/stream/node/mod.rs

@@ -7,7 +7,9 @@ pub mod debug;
 pub mod source;
 
 #[cfg(feature = "metrics")]
-use metrics::{gauge, histogram, increment_counter};
+use metrics::{
+ gauge, histogram, increment_counter, register_counter, register_gauge, register_histogram,
+};
 
 #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
 use perf_event::{Builder, Group};
@@ -38,7 +40,7 @@ use std::{cell::UnsafeCell, sync::Arc};
 pub type NodeDescriptor = String;
 
 #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
-use crate::metrics::perf_event::{HardwareMetricGroup, PerfEvents};
+use crate::metrics::perf_event::PerfEvents;
 
 #[cfg(feature = "metrics")]
 use crate::metrics::runtime_metrics::NodeMetrics;
@@ -140,7 +142,7 @@ where
  logger: ArconLogger,
 
  #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
- hardware_metric_group: HardwareMetricGroup,
+ perf_events: PerfEvents,
 
  #[cfg(feature = "metrics")]
  node_metrics: NodeMetrics,
@@ -168,32 +170,19 @@ where
  let timer = ArconTimer::new(timer_id, backend);
 
  #[cfg(feature = "metrics")]
- let borrowed_descriptor: &str = &descriptor.clone();
+ {
+ register_gauge!("inbound_throughput", "node" => descriptor.clone());
+ register_counter!("epoch_counter", "node" => descriptor.clone());
+ register_counter!("watermark_counter", "node" => descriptor.clone());
+ register_histogram!("batch_execution_time","execution time per events batch","node" => descriptor.clone());
+ }
 
  #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
- let hardware_metric_group = {
- let hardware_metrics_group = Group::new().unwrap();
- let mut hardware_metric_group = HardwareMetricGroup {
- group: hardware_metrics_group,
- counters: vec![],
- };
-
- let iterator = perf_events.counters.iter();
- for val in iterator {
- hardware_metric_group.counters.push((
- val.to_string(),
- Builder::new()
- .group(&mut hardware_metric_group.group)
- .kind(val.get_hardware_kind())
- .build()
- .unwrap(),
- ));
+ {
+ for value in perf_events.counters.iter() {
+ register_histogram!(value.to_string(),"node" => descriptor.clone());
  }
- hardware_metric_group
- .register_hardware_metric_gauges(descriptor.clone(), perf_events)
- .ok();
- hardware_metric_group
- };
+ }
 
  Node {
  ctx: ComponentContext::uninitialised(),
@@ -207,10 +196,10 @@ where
  logger,
 
  #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
- hardware_metric_group,
+ perf_events,
 
  #[cfg(feature = "metrics")]
- node_metrics: NodeMetrics::new(borrowed_descriptor),
+ node_metrics: NodeMetrics::new(),
  }
  }
 
@@ -236,40 +225,50 @@ where
  return Ok(());
  }
 
+ #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
+ let (mut group, counters) = {
+ let mut group = Group::new()?;
+ let mut counters = Vec::with_capacity(self.perf_events.counters.len());
+ for hardware_counter in self.perf_events.counters.iter() {
+ let counter = Builder::new()
+ .group(&mut group)
+ .kind(hardware_counter.get_hardware_kind())
+ .build()?;
+
+ counters.push((hardware_counter.to_string(), counter));
+ }
+ (group, counters)
+ };
+
  #[cfg(feature = "metrics")]
  let start_time = Instant::now();
+
+ #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
+ group.enable()?;
+
  match message {
  MessageContainer::Raw(r) => self.handle_events(r.sender, r.events)?,
  MessageContainer::Local(l) => self.handle_events(l.sender, l.events)?,
  }
 
+ #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
+ group.disable()?;
+
  #[cfg(feature = "metrics")]
  {
  let elapsed = start_time.elapsed();
- histogram!(
- format!("{}_{}", &self.descriptor, "batch_execution_time"),
- elapsed.as_micros() as f64
- );
+ histogram!("batch_execution_time", elapsed.as_micros() as f64,"node" => self.descriptor.clone());
  }
 
  #[cfg(feature = "metrics")]
- gauge!(
- format!("{}_{}", &self.descriptor, "inbound_throughput"),
- self.node_metrics.inbound_throughput.get_one_min_rate()
- );
+ gauge!("inbound_throughput", self.node_metrics.inbound_throughput.get_one_min_rate(), "node" => self.descriptor.clone());
 
  #[cfg(all(feature = "hardware_counters", target_os = "linux", not(test)))]
  {
- let counts = self.hardware_metric_group.group.read()?;
- let counter_iterator = self.hardware_metric_group.counters.iter();
- for (metric_name, counter) in counter_iterator {
- histogram!(
- self.hardware_metric_group
- .get_field_gauge_name(metric_name, &self.descriptor),
- counts[counter] as f64
- );
+ let counts = group.read()?;
+ for (metric_name, counter) in counters.iter() {
+ histogram!(String::from(metric_name), counts[counter] as f64, "node" => self.descriptor.clone());
  }
- self.hardware_metric_group.group.reset()?;
  }
 
  Ok(())
@@ -342,7 +341,7 @@ where
  };
 
  #[cfg(feature = "metrics")]
- increment_counter!(format!("{}_{}", &self.descriptor, "watermark_counter"));
+ increment_counter!("watermark_counter", "node" => self.descriptor.clone());
 
  // Forward the watermark
  unsafe {
@@ -408,7 +407,8 @@ where
  #[inline]
  fn complete_epoch(&mut self) -> ArconResult<()> {
  #[cfg(feature = "metrics")]
- increment_counter!(format!("{}_{}", &self.descriptor, "epoch_counter"));
+ increment_counter!("epoch_counter", "node" => self.descriptor.clone());
+
  // flush the blocked_channels list
  self.node_state.blocked_channels().clear();
 

src/stream/node/source.rs

@@ -1,7 +1,7 @@
 // Copyright (c) 2020, KTH Royal Institute of Technology.
 // SPDX-License-Identifier: AGPL-3.0-only
 #[cfg(feature = "metrics")]
-use metrics::{gauge, increment_counter};
+use metrics::{gauge, increment_counter, register_counter, register_gauge};
 
 #[cfg(feature = "metrics")]
 use crate::metrics::runtime_metrics::SourceMetrics;
@@ -74,6 +74,13 @@ where
  logger: ArconLogger,
  ) -> Self {
  let borrowed_source_name: &str = &conf.name.clone();
+
+ #[cfg(feature = "metrics")]
+ {
+ register_gauge!("incoming_message_rate", "source" => conf.name.clone());
+ register_counter!("error_counter", "source" => conf.name.clone());
+ }
+
  Self {
  ctx: ComponentContext::uninitialised(),
  manager_port: RequiredPort::uninitialised(),
@@ -88,31 +95,23 @@ where
  logger,
 
  #[cfg(feature = "metrics")]
- source_metrics: SourceMetrics::new(borrowed_source_name),
+ source_metrics: SourceMetrics::new(),
 
  descriptor: String::from(borrowed_source_name),
  }
  }
- pub fn process(&mut self) -> ArconResult<()> {
+ pub fn process(&mut self) -> ArconResult<usize> {
  let mut counter = 0;
 
  loop {
  if counter >= self.conf.batch_size {
- return Ok(());
+ return Ok(counter);
  }
 
  let poll = self.source.poll_next()?;
 
  match poll {
  Ok(Poll::Ready(record)) => {
- #[cfg(feature = "metrics")]
- self.source_metrics.incoming_message_rate.mark_n(1);
-
- #[cfg(feature = "metrics")]
- gauge!(
- format!("{}_{}", &self.descriptor, "incoming_message_rate"),
- self.source_metrics.incoming_message_rate.get_one_min_rate()
- );
  match self.conf.time {
  ArconTime::Event => match &self.conf.extractor {
  Some(extractor) => {
@@ -129,17 +128,25 @@ where
  }
  Ok(Poll::Pending) => {
  // nothing to collect, reschedule...
- return Ok(());
+ return Ok(counter);
  }
  Ok(Poll::Done) => {
  // signal end..
  self.ended = true;
- return Ok(());
+ return Ok(counter);
  }
  Err(error) => {
  #[cfg(feature = "metrics")]
- increment_counter!(format!("{}_{}", &self.descriptor, "error_counter"),);
- return self.handle_source_error(error);
+ increment_counter!("error_counter", "source" => self.descriptor.clone());
+
+ match self.handle_source_error(error) {
+ Ok(_) => {
+ counter += 1;
+ }
+ Err(err) => {
+ return Err(err);
+ }
+ }
  }
  }
  }
@@ -153,7 +160,7 @@ where
  KafkaError::Canceled | KafkaError::ConsumerCommit(_) => {
  return Err(crate::error::Error::Unsupported {
  msg: error.to_string(),
- })
+ });
  }
  _ => (),
  }
@@ -235,12 +242,23 @@ where
  S: Source,
 {
  fn handle(&mut self, _event: ProcessSource) -> Handled {
- if let Err(error) = self.process() {
- // fatal error, must shutdown..
- // TODO: coordinate shutdown of the application..
- error!(self.logger, "{}", error);
- }
+ match self.process() {
+ #[cfg(not(feature = "metrics"))]
+ Ok(_) => (),
+ #[cfg(feature = "metrics")]
+ Ok(polled_records) => {
+ self.source_metrics
+ .incoming_message_rate
+ .mark_n(polled_records as u64);
+ gauge!("incoming_message_rate", self.source_metrics.incoming_message_rate.get_one_min_rate(), "source" => self.descriptor.clone());
+ }
 
+ Err(error) => {
+ // fatal error, must shutdown..
+ // TODO: coordinate shutdown of the application..
+ error!(self.logger, "{}", error);
+ }
+ }
  if self.ended {
  self.manager_port.trigger(SourceManagerEvent::End);
  } else {