[pbs-devel] [PATCH proxmox-backup v6 29/37] datastore: add local datastore cache for network attached storages
Christian Ebner
c.ebner at proxmox.com
Tue Jul 8 19:01:06 CEST 2025
Use a local datastore as cache using LRU cache replacement policy for
operations on a datastore backed by a network, e.g. by an S3 object
store backend. The goal is to reduce number of requests to the
backend and thereby save costs (monetary as well as time).
The cacher allows to fetch cache items on cache misses via the access
method.
Signed-off-by: Christian Ebner <c.ebner at proxmox.com>
---
pbs-datastore/src/datastore.rs | 53 +++++-
pbs-datastore/src/lib.rs | 3 +
.../src/local_datastore_lru_cache.rs | 172 ++++++++++++++++++
3 files changed, 227 insertions(+), 1 deletion(-)
create mode 100644 pbs-datastore/src/local_datastore_lru_cache.rs
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index c1ba2dcea..ef146e84a 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -40,9 +40,10 @@ use crate::dynamic_index::{DynamicIndexReader, DynamicIndexWriter};
use crate::fixed_index::{FixedIndexReader, FixedIndexWriter};
use crate::hierarchy::{ListGroups, ListGroupsType, ListNamespaces, ListNamespacesRecursive};
use crate::index::IndexFile;
+use crate::local_datastore_lru_cache::S3Cacher;
use crate::s3::S3_CONTENT_PREFIX;
use crate::task_tracking::{self, update_active_operations};
-use crate::DataBlob;
+use crate::{DataBlob, LocalDatastoreLruCache};
static DATASTORE_MAP: LazyLock<Mutex<HashMap<String, Arc<DataStoreImpl>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
@@ -136,6 +137,7 @@ pub struct DataStoreImpl {
last_digest: Option<[u8; 32]>,
sync_level: DatastoreFSyncLevel,
backend_config: DatastoreBackendConfig,
+ lru_store_caching: Option<LocalDatastoreLruCache>,
}
impl DataStoreImpl {
@@ -151,6 +153,7 @@ impl DataStoreImpl {
last_digest: None,
sync_level: Default::default(),
backend_config: Default::default(),
+ lru_store_caching: None,
})
}
}
@@ -255,6 +258,37 @@ impl DataStore {
Ok(backend_type)
}
+ pub fn cache(&self) -> Option<&LocalDatastoreLruCache> {
+ self.inner.lru_store_caching.as_ref()
+ }
+
+ /// Check if the digest is present in the local datastore cache.
+ /// Always returns false if there is no cache configured for this datastore.
+ pub fn cache_contains(&self, digest: &[u8; 32]) -> bool {
+ if let Some(cache) = self.inner.lru_store_caching.as_ref() {
+ return cache.contains(digest);
+ }
+ false
+ }
+
+ /// Insert digest as most recently used on in the cache.
+ /// Returns with success if there is no cache configured for this datastore.
+ pub fn cache_insert(&self, digest: &[u8; 32], chunk: &DataBlob) -> Result<(), Error> {
+ if let Some(cache) = self.inner.lru_store_caching.as_ref() {
+ return cache.insert(digest, chunk);
+ }
+ Ok(())
+ }
+
+ pub fn cacher(&self) -> Result<Option<S3Cacher>, Error> {
+ self.backend().map(|backend| match backend {
+ DatastoreBackend::S3(s3_client) => {
+ Some(S3Cacher::new(s3_client, self.inner.chunk_store.clone()))
+ }
+ DatastoreBackend::Filesystem => None,
+ })
+ }
+
pub fn lookup_datastore(
name: &str,
operation: Option<Operation>,
@@ -437,6 +471,16 @@ impl DataStore {
.parse_property_string(config.backend.as_deref().unwrap_or(""))?,
)?;
+ const LOCAL_DATASTORE_CACHE_SIZE: usize = 10_000_000;
+ let lru_store_caching = if DatastoreBackendType::S3 == backend_config.ty.unwrap_or_default()
+ {
+ let cache =
+ LocalDatastoreLruCache::new(LOCAL_DATASTORE_CACHE_SIZE, chunk_store.clone());
+ Some(cache)
+ } else {
+ None
+ };
+
Ok(DataStoreImpl {
chunk_store,
gc_mutex: Mutex::new(()),
@@ -446,6 +490,7 @@ impl DataStore {
last_digest,
sync_level: tuning.sync_level.unwrap_or_default(),
backend_config,
+ lru_store_caching,
})
}
@@ -1579,6 +1624,12 @@ impl DataStore {
chunk_count += 1;
if atime < min_atime {
+ if let Some(cache) = self.cache() {
+ let mut digest_bytes = [0u8; 32];
+ hex::decode_to_slice(digest.as_bytes(), &mut digest_bytes)?;
+ // ignore errors, phase 3 will retry cleanup anyways
+ let _ = cache.remove(&digest_bytes);
+ }
delete_list.push(content.key);
if bad {
gc_status.removed_bad += 1;
diff --git a/pbs-datastore/src/lib.rs b/pbs-datastore/src/lib.rs
index ca6fdb7d8..b9eb035c2 100644
--- a/pbs-datastore/src/lib.rs
+++ b/pbs-datastore/src/lib.rs
@@ -217,3 +217,6 @@ pub use snapshot_reader::SnapshotReader;
mod local_chunk_reader;
pub use local_chunk_reader::LocalChunkReader;
+
+mod local_datastore_lru_cache;
+pub use local_datastore_lru_cache::LocalDatastoreLruCache;
diff --git a/pbs-datastore/src/local_datastore_lru_cache.rs b/pbs-datastore/src/local_datastore_lru_cache.rs
new file mode 100644
index 000000000..bb64c52f3
--- /dev/null
+++ b/pbs-datastore/src/local_datastore_lru_cache.rs
@@ -0,0 +1,172 @@
+//! Use a local datastore as cache for operations on a datastore attached via
+//! a network layer (e.g. via the S3 backend).
+
+use std::future::Future;
+use std::sync::Arc;
+
+use anyhow::{bail, Error};
+use http_body_util::BodyExt;
+
+use pbs_tools::async_lru_cache::{AsyncCacher, AsyncLruCache};
+use proxmox_s3_client::S3Client;
+
+use crate::ChunkStore;
+use crate::DataBlob;
+
+#[derive(Clone)]
+pub struct S3Cacher {
+ client: Arc<S3Client>,
+ store: Arc<ChunkStore>,
+}
+
+impl AsyncCacher<[u8; 32], ()> for S3Cacher {
+ fn fetch(
+ &self,
+ key: [u8; 32],
+ ) -> Box<dyn Future<Output = Result<Option<()>, Error>> + Send + 'static> {
+ let client = self.client.clone();
+ let store = self.store.clone();
+ Box::new(async move {
+ let object_key = crate::s3::object_key_from_digest(&key)?;
+ match client.get_object(object_key).await? {
+ None => bail!("could not fetch object with key {}", hex::encode(key)),
+ Some(response) => {
+ let bytes = response.content.collect().await?.to_bytes();
+ let chunk = DataBlob::from_raw(bytes.to_vec())?;
+ store.insert_chunk(&chunk, &key)?;
+ Ok(Some(()))
+ }
+ }
+ })
+ }
+}
+
+impl S3Cacher {
+ pub fn new(client: Arc<S3Client>, store: Arc<ChunkStore>) -> Self {
+ Self { client, store }
+ }
+}
+
+/// LRU cache using local datastore for caching chunks
+///
+/// Uses a LRU cache, but without storing the values in-memory but rather
+/// on the filesystem
+pub struct LocalDatastoreLruCache {
+ cache: AsyncLruCache<[u8; 32], ()>,
+ store: Arc<ChunkStore>,
+}
+
+impl LocalDatastoreLruCache {
+ pub fn new(capacity: usize, store: Arc<ChunkStore>) -> Self {
+ Self {
+ cache: AsyncLruCache::new(capacity),
+ store,
+ }
+ }
+
+ /// Insert a new chunk into the local datastore cache.
+ ///
+ /// Fails if the chunk cannot be inserted successfully.
+ pub fn insert(&self, digest: &[u8; 32], chunk: &DataBlob) -> Result<(), Error> {
+ self.store.insert_chunk(chunk, digest)?;
+ self.cache.insert(*digest, (), |digest| {
+ let (path, _digest_str) = self.store.chunk_path(&digest);
+ // Truncate to free up space but keep the inode around, since that
+ // is used as marker for chunks in use by garbage collection.
+ if let Err(err) = nix::unistd::truncate(&path, 0) {
+ if err != nix::errno::Errno::ENOENT {
+ return Err(Error::from(err));
+ }
+ }
+ Ok(())
+ })
+ }
+
+ /// Remove a chunk from the local datastore cache.
+ ///
+ /// Fails if the chunk cannot be deleted successfully.
+ pub fn remove(&self, digest: &[u8; 32]) -> Result<(), Error> {
+ self.cache.remove(*digest);
+ let (path, _digest_str) = self.store.chunk_path(digest);
+ std::fs::remove_file(path).map_err(Error::from)
+ }
+
+ pub async fn access(
+ &self,
+ digest: &[u8; 32],
+ cacher: &mut S3Cacher,
+ ) -> Result<Option<DataBlob>, Error> {
+ if self
+ .cache
+ .access(*digest, cacher, |digest| {
+ let (path, _digest_str) = self.store.chunk_path(&digest);
+ // Truncate to free up space but keep the inode around, since that
+ // is used as marker for chunks in use by garbage collection.
+ if let Err(err) = nix::unistd::truncate(&path, 0) {
+ if err != nix::errno::Errno::ENOENT {
+ return Err(Error::from(err));
+ }
+ }
+ Ok(())
+ })
+ .await?
+ .is_some()
+ {
+ let (path, _digest_str) = self.store.chunk_path(digest);
+ let mut file = match std::fs::File::open(&path) {
+ Ok(file) => file,
+ Err(err) => {
+ // Expected chunk to be present since LRU cache has it, but it is missing
+ // locally, try to fetch again
+ if err.kind() == std::io::ErrorKind::NotFound {
+ let object_key = crate::s3::object_key_from_digest(digest)?;
+ match cacher.client.get_object(object_key).await? {
+ None => {
+ bail!("could not fetch object with key {}", hex::encode(digest))
+ }
+ Some(response) => {
+ let bytes = response.content.collect().await?.to_bytes();
+ let chunk = DataBlob::from_raw(bytes.to_vec())?;
+ self.store.insert_chunk(&chunk, digest)?;
+ std::fs::File::open(&path)?
+ }
+ }
+ } else {
+ return Err(Error::from(err));
+ }
+ }
+ };
+ let chunk = match DataBlob::load_from_reader(&mut file) {
+ Ok(chunk) => chunk,
+ Err(err) => {
+ use std::io::Seek;
+ // Check if file is empty marker file, try fetching content if so
+ if file.seek(std::io::SeekFrom::End(0))? == 0 {
+ let object_key = crate::s3::object_key_from_digest(digest)?;
+ match cacher.client.get_object(object_key).await? {
+ None => {
+ bail!("could not fetch object with key {}", hex::encode(digest))
+ }
+ Some(response) => {
+ let bytes = response.content.collect().await?.to_bytes();
+ let chunk = DataBlob::from_raw(bytes.to_vec())?;
+ self.store.insert_chunk(&chunk, digest)?;
+ let mut file = std::fs::File::open(&path)?;
+ DataBlob::load_from_reader(&mut file)?
+ }
+ }
+ } else {
+ return Err(err);
+ }
+ }
+ };
+ Ok(Some(chunk))
+ } else {
+ Ok(None)
+ }
+ }
+
+ pub fn contains(&self, digest: &[u8; 32]) -> bool {
+ self.cache.contains(*digest)
+ }
+}
--
2.47.2
More information about the pbs-devel
mailing list