[PATCH proxmox-backup v6 18/22] datastore: add thresholds notification callback on datastore lookup

Christian Ebner
2026-03-19T09:42:09Z

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