Implement basic metrics in table

Cargo.lock

@@ -902,6 +902,7 @@ dependencies = [
  "garage_util 0.6.0",
  "hexdump",
  "log",
+ "opentelemetry",
  "rand",
  "rmp-serde 0.15.5",
  "serde",

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()),
 			},
 		}
 	}

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...");

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"

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,
 		})
 	}
 

src/table/lib.rs

@@ -4,6 +4,7 @@
 #[macro_use]
 extern crate log;
 
+mod metrics;
 pub mod schema;
 pub mod util;
 

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(),
+		}
+	}
+}