From c51663073b75d2dd9663637a7c0ba80c345c8503 Mon Sep 17 00:00:00 2001
From: Maximilien R <maximilien@deuxfleurs.fr>
Date: Tue, 15 Feb 2022 20:09:43 +0100
Subject: [PATCH] Implement basic metrics in table

---
 Cargo.lock           |  1 +
 src/admin/metrics.rs |  6 ------
 src/garage/server.rs |  2 +-
 src/table/Cargo.toml |  2 ++
 src/table/data.rs    |  6 ++++++
 src/table/lib.rs     |  1 +
 src/table/metrics.rs | 25 +++++++++++++++++++++++++
 7 files changed, 36 insertions(+), 7 deletions(-)
 create mode 100644 src/table/metrics.rs

diff --git a/Cargo.lock b/Cargo.lock
index 00e7777b..7ca4c2fc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -902,6 +902,7 @@ dependencies = [
  "garage_util 0.6.0",
  "hexdump",
  "log",
+ "opentelemetry",
  "rand",
  "rmp-serde 0.15.5",
  "serde",
diff --git a/src/admin/metrics.rs b/src/admin/metrics.rs
index 547ee4c8..ccc26d26 100644
--- a/src/admin/metrics.rs
+++ b/src/admin/metrics.rs
@@ -76,7 +76,6 @@ struct AdminServerMetrics {
 	http_counter: BoundCounter<u64>,
 	http_body_gauge: BoundValueRecorder<u64>,
 	http_req_histogram: BoundValueRecorder<f64>,
-	bucket_v2_merkle_updater_todo_queue_length: BoundValueRecorder<f64>,
 }
 
 impl AdminServer {
@@ -102,11 +101,6 @@ impl AdminServer {
 					.with_description("The HTTP request latencies in seconds.")
 					.init()
 					.bind(HANDLER_ALL.as_ref()),
-				bucket_v2_merkle_updater_todo_queue_length: meter
-					.f64_value_recorder("bucket_v2.merkle_updater.todo_queue_length")
-					.with_description("Bucket merkle updater TODO queue length.")
-					.init()
-					.bind(HANDLER_ALL.as_ref()),
 			},
 		}
 	}
diff --git a/src/garage/server.rs b/src/garage/server.rs
index 923df1cd..cd32d708 100644
--- a/src/garage/server.rs
+++ b/src/garage/server.rs
@@ -35,7 +35,7 @@ pub async fn run_server(config_file: PathBuf) -> Result<(), Error> {
 		.open()
 		.expect("Unable to open sled DB");
 
-	info!("Configure and run admin web server...");
+	info!("Initialize admin web server and metric backend...");
 	let admin_server_init = AdminServer::init();
 
 	info!("Initializing background runner...");
diff --git a/src/table/Cargo.toml b/src/table/Cargo.toml
index 91d71ddd..f6f9974d 100644
--- a/src/table/Cargo.toml
+++ b/src/table/Cargo.toml
@@ -17,6 +17,8 @@ path = "lib.rs"
 garage_rpc = { version = "0.6.0", path = "../rpc" }
 garage_util = { version = "0.6.0", path = "../util" }
 
+opentelemetry = "0.17"
+
 async-trait = "0.1.7"
 bytes = "1.0"
 hexdump = "0.1"
diff --git a/src/table/data.rs b/src/table/data.rs
index d7787b6b..a5209c26 100644
--- a/src/table/data.rs
+++ b/src/table/data.rs
@@ -13,6 +13,7 @@ use garage_rpc::system::System;
 
 use crate::crdt::Crdt;
 use crate::gc::GcTodoEntry;
+use crate::metrics::*;
 use crate::replication::*;
 use crate::schema::*;
 
@@ -28,6 +29,8 @@ pub struct TableData<F: TableSchema, R: TableReplication> {
 	pub(crate) merkle_todo: sled::Tree,
 	pub(crate) merkle_todo_notify: Notify,
 	pub(crate) gc_todo: sled::Tree,
+
+	pub(crate) metrics: TableMetrics,
 }
 
 impl<F, R> TableData<F, R>
@@ -51,6 +54,8 @@ where
 			.open_tree(&format!("{}:gc_todo_v2", F::TABLE_NAME))
 			.expect("Unable to open DB tree");
 
+		let metrics = TableMetrics::new(F::TABLE_NAME, merkle_todo.clone());
+
 		Arc::new(Self {
 			system,
 			instance,
@@ -60,6 +65,7 @@ where
 			merkle_todo,
 			merkle_todo_notify: Notify::new(),
 			gc_todo,
+			metrics,
 		})
 	}
 
diff --git a/src/table/lib.rs b/src/table/lib.rs
index d6c19f1b..e0920357 100644
--- a/src/table/lib.rs
+++ b/src/table/lib.rs
@@ -4,6 +4,7 @@
 #[macro_use]
 extern crate log;
 
+mod metrics;
 pub mod schema;
 pub mod util;
 
diff --git a/src/table/metrics.rs b/src/table/metrics.rs
new file mode 100644
index 00000000..38e93904
--- /dev/null
+++ b/src/table/metrics.rs
@@ -0,0 +1,25 @@
+use opentelemetry::{global, metrics::*, KeyValue};
+
+/// TableMetrics reference all counter used for metrics
+pub struct TableMetrics {
+	merkle_updater_todo_queue_length: ValueObserver<u64>,
+}
+impl TableMetrics {
+	pub fn new(table_name: &'static str, merkle_todo: sled::Tree) -> Self {
+		let meter = global::meter(table_name);
+		TableMetrics {
+			merkle_updater_todo_queue_length: meter
+				.u64_value_observer(
+					format!("merkle_updater_todo_queue_length"),
+					move |observer| {
+						observer.observe(
+							merkle_todo.len() as u64,
+							&[KeyValue::new("table_name", table_name)],
+						)
+					},
+				)
+				.with_description("Bucket merkle updater TODO queue length")
+				.init(),
+		}
+	}
+}