add metrics to workers

2 files changed, 26 insertions, 2 deletions

diff --git a/src/db/metric_proxy.rs b/src/db/metric_proxy.rs
index d246e7fe..e5221a7f 100644
--- a/src/db/metric_proxy.rs
+++ b/src/db/metric_proxy.rs
@@ -9,7 +9,7 @@ use crate::{
 };
 use opentelemetry::{
 	global,
-	metrics::{Counter, Unit, ValueRecorder},
+	metrics::{Unit, ValueRecorder},
 	KeyValue,
 };
 
@@ -21,7 +21,7 @@ pub struct MetricDbProxy {
 
 impl MetricDbProxy {
 	pub fn init(db: LmdbDb) -> Db {
-		let meter = global::meter("garage/web");
+		let meter = global::meter("garage/db");
 		let s = Self {
 			db,
 			op: meter
diff --git a/src/util/background/worker.rs b/src/util/background/worker.rs
index 8165e2cb..22f747b6 100644
--- a/src/util/background/worker.rs
+++ b/src/util/background/worker.rs
@@ -1,11 +1,13 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
+use std::time::Instant;
 
 use async_trait::async_trait;
 use futures::future::*;
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
+use opentelemetry::{global, metrics::ValueRecorder, KeyValue};
 use serde::{Deserialize, Serialize};
 use tokio::select;
 use tokio::sync::{mpsc, watch};
@@ -62,6 +64,7 @@ pub(crate) struct WorkerProcessor {
 	stop_signal: watch::Receiver<bool>,
 	worker_chan: mpsc::UnboundedReceiver<Box<dyn Worker>>,
 	worker_info: Arc<std::sync::Mutex<HashMap<usize, WorkerInfo>>>,
+	metrics: ValueRecorder<f64>,
 }
 
 impl WorkerProcessor {
@@ -70,10 +73,15 @@ impl WorkerProcessor {
 		stop_signal: watch::Receiver<bool>,
 		worker_info: Arc<std::sync::Mutex<HashMap<usize, WorkerInfo>>>,
 	) -> Self {
+		let meter = global::meter("garage/util");
 		Self {
 			stop_signal,
 			worker_chan,
 			worker_info,
+			metrics: meter
+				.f64_value_recorder("util.worker_step")
+				.with_description("Duration and amount of worker steps executed")
+				.init(),
 		}
 	}
 
@@ -103,6 +111,7 @@ impl WorkerProcessor {
 								errors: 0,
 								consecutive_errors: 0,
 								last_error: None,
+								metrics: self.metrics.clone(),
 							};
 						workers.push(async move {
 							worker.step().await;
@@ -183,10 +192,13 @@ struct WorkerHandler {
 	errors: usize,
 	consecutive_errors: usize,
 	last_error: Option<(String, u64)>,
+	metrics: ValueRecorder<f64>,
 }
 
 impl WorkerHandler {
 	async fn step(&mut self) {
+		let request_start = Instant::now();
+		//@FIXME we also want to track errors in metrics but I don't know how yet.
 		match self.state {
 			WorkerState::Busy => match self.worker.work(&mut self.stop_signal).await {
 				Ok(s) => {
@@ -229,5 +241,17 @@ impl WorkerHandler {
 			}
 			WorkerState::Done => unreachable!(),
 		}
+
+		// metrics
+		let metric_tags = [
+			KeyValue::new("state", self.state.to_string()),
+			KeyValue::new("name", self.worker.name()),
+			KeyValue::new("id", format!("{}", self.task_id)),
+		];
+
+		let delay_secs = Instant::now()
+			.saturating_duration_since(request_start)
+			.as_secs_f64();
+		self.metrics.record(delay_secs, &metric_tags);
 	}
 }