[PATCH proxmox-backup v5 19/20] datastore: add thresholds notification callback on datastore lookup

[Christian Ebner <c.ebner@proxmox.com>]

Pass in the callback to be executed when a datastore threshold value
exceeds its set limit. This callback is then callable by the s3
client request counters implementation, currently the only component
defining thresholds.

When a threshold is exceeded, the notification is generated by the
callback and queued for being send.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs       | 25 +++++++++++++++++++++++--
 pbs-datastore/src/snapshot_reader.rs |  1 +
 src/tools/mod.rs                     |  9 ++++++++-
 3 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index d60609124..f419a5cb7 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -204,11 +204,13 @@ impl DataStoreImpl {
 }
 
 pub type ProxyConfigCallback = fn() -> Option<ProxyConfig>;
+pub type ThresholdsExceededCallback = fn(&str, &str, u64, u64) -> Result<(), Error>;
 
 pub struct DataStoreLookup<'a> {
     name: &'a str,
     operation: Operation,
     proxy_config_callback: Arc<Option<ProxyConfigCallback>>,
+    thresholds_exceeded_callback: Arc<Option<ThresholdsExceededCallback>>,
 }
 
 impl<'a> DataStoreLookup<'a> {
@@ -216,11 +218,13 @@ impl<'a> DataStoreLookup<'a> {
         name: &'a str,
         operation: Operation,
         proxy_config_callback: Arc<Option<ProxyConfigCallback>>,
+        thresholds_exceeded_callback: Arc<Option<ThresholdsExceededCallback>>,
     ) -> Self {
         Self {
             name,
             operation,
             proxy_config_callback,
+            thresholds_exceeded_callback,
         }
     }
 }
@@ -574,7 +578,12 @@ impl DataStore {
             )?)
         };
 
-        let datastore = DataStore::with_store_and_config(chunk_store, config, gen_num)?;
+        let datastore = DataStore::with_store_and_config(
+            chunk_store,
+            config,
+            gen_num,
+            Arc::clone(&lookup.thresholds_exceeded_callback),
+        )?;
 
         let datastore = Arc::new(datastore);
         datastore_cache.insert(lookup.name.to_string(), datastore.clone());
@@ -610,7 +619,8 @@ impl DataStore {
         {
             // the datastore drop handler does the checking if tasks are running and clears the
             // cache entry, so we just have to trigger it here
-            let lookup = DataStoreLookup::with(name, Operation::Lookup, Arc::new(None));
+            let lookup =
+                DataStoreLookup::with(name, Operation::Lookup, Arc::new(None), Arc::new(None));
             let _ = DataStore::lookup_datastore(lookup);
         }
 
@@ -664,6 +674,7 @@ impl DataStore {
             Arc::new(chunk_store),
             config,
             None,
+            Arc::new(None),
         )?);
 
         if let Some(operation) = operation {
@@ -681,6 +692,7 @@ impl DataStore {
         chunk_store: Arc<ChunkStore>,
         config: DataStoreConfig,
         generation: Option<usize>,
+        thresholds_exceeded_callback: Arc<Option<ThresholdsExceededCallback>>,
     ) -> Result<DataStoreImpl, Error> {
         let mut gc_status_path = chunk_store.base_path();
         gc_status_path.push(".gc-status");
@@ -733,6 +745,15 @@ impl DataStore {
                 let mut request_counters = Self::request_counters(&config, &backend_config)?;
 
                 Self::update_notification_thresholds(&mut request_counters, &config)?;
+                SharedRequestCounters::update_thresholds_exceeded_callback(Box::new(
+                    move |counter_id, threshold, value| {
+                        thresholds_exceeded_callback.map(|cb| {
+                            if let Err(err) = cb(&config.name, counter_id, threshold, value) {
+                                log::error!("failed to send notification: {err:?}");
+                            }
+                        });
+                    },
+                ));
 
                 (Some(cache), Some(Arc::new(request_counters)))
             } else {
diff --git a/pbs-datastore/src/snapshot_reader.rs b/pbs-datastore/src/snapshot_reader.rs
index ddae632cc..88c186076 100644
--- a/pbs-datastore/src/snapshot_reader.rs
+++ b/pbs-datastore/src/snapshot_reader.rs
@@ -169,6 +169,7 @@ impl<F: Fn(&[u8; 32]) -> bool> Iterator for SnapshotChunkIterator<'_, F> {
                             self.snapshot_reader.datastore_name(),
                             Operation::Read,
                             Arc::clone(&self.proxy_config_callback),
+                            Arc::new(None),
                         );
                         let datastore = DataStore::lookup_datastore(lookup)?;
                         let order =
diff --git a/src/tools/mod.rs b/src/tools/mod.rs
index 5f3505417..37ad4e4aa 100644
--- a/src/tools/mod.rs
+++ b/src/tools/mod.rs
@@ -204,5 +204,12 @@ pub(super) fn node_proxy_config() -> Option<proxmox_http::ProxyConfig> {
 /// Read the nodes http proxy config from the node config.
 #[inline(always)]
 pub fn lookup_with<'a>(name: &'a str, operation: Operation) -> DataStoreLookup<'a> {
-    DataStoreLookup::with(name, operation, Arc::new(Some(node_proxy_config)))
+    DataStoreLookup::with(
+        name,
+        operation,
+        Arc::new(Some(node_proxy_config)),
+        Arc::new(Some(
+            crate::server::notifications::send_datastore_threshold_exceeded,
+        )),
+    )
 }
-- 
2.47.3