[PATCH proxmox-backup v6 18/22] 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       | 36 ++++++++++++++++++++++++----
 pbs-datastore/src/snapshot_reader.rs |  1 +
 src/tools/mod.rs                     |  9 ++++++-
 3 files changed, 41 insertions(+), 5 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 379cf8b73..ec0ffd713 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -202,14 +202,25 @@ impl DataStoreImpl {
     }
 }
 
+pub type ThresholdsExceededCallback = fn(&str, &str, u64, u64) -> Result<(), Error>;
+
 pub struct DataStoreLookup<'a> {
     name: &'a str,
     operation: Operation,
+    thresholds_exceeded_callback: Arc<Option<ThresholdsExceededCallback>>,
 }
 
 impl<'a> DataStoreLookup<'a> {
-    pub fn with(name: &'a str, operation: Operation) -> Self {
-        Self { name, operation }
+    pub fn with(
+        name: &'a str,
+        operation: Operation,
+        thresholds_exceeded_callback: Arc<Option<ThresholdsExceededCallback>>,
+    ) -> Self {
+        Self {
+            name,
+            operation,
+            thresholds_exceeded_callback,
+        }
     }
 }
 
@@ -556,7 +567,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());
@@ -591,7 +607,7 @@ 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);
+            let lookup = DataStoreLookup::with(name, Operation::Lookup, Arc::new(None));
             let _ = DataStore::lookup_datastore(lookup);
         }
 
@@ -645,6 +661,7 @@ impl DataStore {
             Arc::new(chunk_store),
             config,
             None,
+            Arc::new(None),
         )?);
 
         if let Some(operation) = operation {
@@ -658,6 +675,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");
@@ -710,6 +728,16 @@ impl DataStore {
                 let mut request_counters = Self::request_counters(&config, &backend_config)?;
 
                 Self::update_notification_thresholds(&mut request_counters, &config)?;
+                request_counters.set_thresholds_exceeded_callback(
+                    config.name,
+                    Box::new(move |label, counter_id, threshold, value| {
+                        thresholds_exceeded_callback.map(|cb| {
+                            if let Err(err) = cb(label, 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 d522a02d7..2c8f19c17 100644
--- a/pbs-datastore/src/snapshot_reader.rs
+++ b/pbs-datastore/src/snapshot_reader.rs
@@ -166,6 +166,7 @@ impl<F: Fn(&[u8; 32]) -> bool> Iterator for SnapshotChunkIterator<'_, F> {
                         let lookup = DataStoreLookup::with(
                             self.snapshot_reader.datastore_name(),
                             Operation::Read,
+                            Arc::new(None),
                         );
                         let datastore = DataStore::lookup_datastore(lookup)?;
                         let order =
diff --git a/src/tools/mod.rs b/src/tools/mod.rs
index a82aaf136..2dfbac885 100644
--- a/src/tools/mod.rs
+++ b/src/tools/mod.rs
@@ -4,6 +4,7 @@
 
 use anyhow::{bail, Error};
 use std::collections::HashSet;
+use std::sync::Arc;
 
 use pbs_api_types::{
     Authid, BackupContent, CryptMode, Operation, SnapshotListItem, SnapshotVerifyState,
@@ -192,5 +193,11 @@ pub(crate) fn backup_info_to_snapshot_list_item(
 /// Lookup the datastore by name with given operation.
 #[inline(always)]
 pub fn lookup_with<'a>(name: &'a str, operation: Operation) -> DataStoreLookup<'a> {
-    DataStoreLookup::with(name, operation)
+    DataStoreLookup::with(
+        name,
+        operation,
+        Arc::new(Some(
+            crate::server::notifications::send_datastore_threshold_exceeded,
+        )),
+    )
 }
-- 
2.47.3