From f9c1fc8657684b6b93f460350f653aa5655c0326 Mon Sep 17 00:00:00 2001
From: Bojan Serafimov <bojan.serafimov7@gmail.com>
Date: Tue, 24 May 2022 21:48:45 -0400
Subject: [PATCH] Add metric

---
 pageserver/src/bin/pageserver.rs         | 10 +++++-----
 pageserver/src/tenant_jobs/compaction.rs |  2 --
 pageserver/src/tenant_jobs/gc.rs         |  2 --
 pageserver/src/tenant_jobs/worker.rs     | 15 ++++++++++++++-
 4 files changed, 19 insertions(+), 10 deletions(-)

diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs
index 5a9fc0e57d..d5d69e26a7 100644
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -179,11 +179,11 @@ fn main() -> anyhow::Result<()> {
 
     // If failpoints are used, terminate the whole pageserver process if they are hit.
     let scenario = FailScenario::setup();
-    // if fail::has_failpoints() {
-    //     std::panic::set_hook(Box::new(|_| {
-    //         std::process::exit(1);
-    //     }));
-    // }
+    if fail::has_failpoints() {
+        std::panic::set_hook(Box::new(|_| {
+            std::process::exit(1);
+        }));
+    }
 
     // Basic initialization of things that don't change after startup
     virtual_file::init(conf.max_file_descriptors);
diff --git a/pageserver/src/tenant_jobs/compaction.rs b/pageserver/src/tenant_jobs/compaction.rs
index 5f80dd2775..c680d9fda6 100644
--- a/pageserver/src/tenant_jobs/compaction.rs
+++ b/pageserver/src/tenant_jobs/compaction.rs
@@ -31,5 +31,3 @@ impl Job for CompactionJob {
 }
 
 pub static COMPACTION_POOL: OnceCell<Pool<CompactionJob>> = OnceCell::new();
-
-// TODO spawn 20 worker threads
diff --git a/pageserver/src/tenant_jobs/gc.rs b/pageserver/src/tenant_jobs/gc.rs
index d8bfdd854a..cbb47bf439 100644
--- a/pageserver/src/tenant_jobs/gc.rs
+++ b/pageserver/src/tenant_jobs/gc.rs
@@ -31,5 +31,3 @@ impl Job for GcJob {
 }
 
 pub static GC_POOL: OnceCell<Pool<GcJob>> = OnceCell::new();
-
-// TODO spawn 20 worker threads
diff --git a/pageserver/src/tenant_jobs/worker.rs b/pageserver/src/tenant_jobs/worker.rs
index 309e9c9e34..0a490b0394 100644
--- a/pageserver/src/tenant_jobs/worker.rs
+++ b/pageserver/src/tenant_jobs/worker.rs
@@ -1,4 +1,15 @@
 use std::{any::Any, collections::{BinaryHeap, HashMap}, fmt::Debug, hash::Hash, ops::Add, panic::{self, AssertUnwindSafe}, sync::{Condvar, Mutex}, time::{Duration, Instant}};
+use lazy_static::lazy_static;
+use metrics::{register_gauge_vec, GaugeVec};
+
+lazy_static! {
+    static ref POOL_UTILIZATION_GAUGE: GaugeVec = register_gauge_vec!(
+        "pageserver_pool_utilization",
+        "Number of bysy workers",
+        &["pool_name"]
+    )
+    .expect("Failed to register safekeeper_commit_lsn gauge vec");
+}
 
 pub trait Job: std::fmt::Debug + Send + Clone + PartialOrd + Ord + Hash + 'static {
     type ErrorType;
@@ -104,9 +115,11 @@ impl<J: Job> Pool<J> where J::ErrorType: Debug {
 
                 // Run job without holding lock
                 drop(job_table);
+                POOL_UTILIZATION_GAUGE.with_label_values(&["todo_put_pool_name_here"]).inc();
                 let result = panic::catch_unwind(AssertUnwindSafe(|| {
                     job.run()
                 }));
+                POOL_UTILIZATION_GAUGE.with_label_values(&["todo_put_pool_name_here"]).dec();
                 job_table = self.job_table.lock().unwrap();
 
                 // Update job status
@@ -118,7 +131,7 @@ impl<J: Job> Pool<J> where J::ErrorType: Debug {
                         job_table.queue.push(Deadline {
                             job: job.clone(),
                             start_by: reschedule_for,
-                        })
+                        });
                     },
                     Ok(Ok(None)) => {
                         job_table.status.remove(&job);