[pbs-devel] [PATCH proxmox-backup v11 30/46] datastore: add local datastore cache for network attached storages

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

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).

Cached chunks are stored on the local datastore cache, already
containing the datastore's contents metadata (namespace, group,
snapshot, owner, index files, ecc..), used to perform fast lookups.
The cache itself only stores chunk digests, not the raw data itself.
When payload data is required, contents are looked up and read from
the local datastore cache filesystem, including fallback to fetch from
the backend if the presumably cached entry is not found.

The cacher allows to fetch cache items on cache misses via the access
method.

The capacity of the cache is derived from the local datastore cache
filesystem, or by the user configured value, whichever is smalller.
The capacity is only set on instantiation of the store, and the current
value kept as long as the datastore remains cached in the datastore
cache. To change the value, the store has to be either be set to offline
mode and back, or the services restarted.

Basic performance tests:

Backup and upload of contents of linux git repository to AWS S3,
snapshots removed in-between each backup run to avoid other chunk reuse
optimization of PBS.

no-cache:
    had to backup 5.069 GiB of 5.069 GiB (compressed 3.718 GiB) in 50.76 s (average 102.258 MiB/s)
empty-cache:
    had to backup 5.069 GiB of 5.069 GiB (compressed 3.718 GiB) in 50.42 s (average 102.945 MiB/s)
all-cached:
    had to backup 5.069 GiB of 5.069 GiB (compressed 3.718 GiB) in 43.78 s (average 118.554 MiB/s)

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
Reviewed-by: Lukas Wagner <l.wagner@proxmox.com>
Reviewed-by: Hannes Laimer <h.laimer@proxmox.com>
---
changes since version 10:
 - no changes

 pbs-datastore/src/datastore.rs                |  81 +++++++-
 pbs-datastore/src/lib.rs                      |   3 +
 .../src/local_datastore_lru_cache.rs          | 180 ++++++++++++++++++
 3 files changed, 260 insertions(+), 4 deletions(-)
 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 e105e606e..13934f6a4 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()));
@@ -139,6 +140,7 @@ pub struct DataStoreImpl {
     last_digest: Option<[u8; 32]>,
     sync_level: DatastoreFSyncLevel,
     backend_config: DatastoreBackendConfig,
+    lru_store_caching: Option<LocalDatastoreLruCache>,
 }
 
 impl DataStoreImpl {
@@ -154,6 +156,7 @@ impl DataStoreImpl {
             last_digest: None,
             sync_level: Default::default(),
             backend_config: Default::default(),
+            lru_store_caching: None,
         })
     }
 }
@@ -258,6 +261,39 @@ 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(())
+    }
+
+    /// Returns the cacher for datastores backed by S3 object stores.
+    /// This allows to fetch chunks to the local cache store on-demand.
+    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>,
@@ -440,6 +476,33 @@ impl DataStore {
                 .parse_property_string(config.backend.as_deref().unwrap_or(""))?,
         )?;
 
+        let lru_store_caching = if DatastoreBackendType::S3 == backend_config.ty.unwrap_or_default()
+        {
+            let mut cache_capacity = 0;
+            if let Ok(fs_info) = proxmox_sys::fs::fs_info(&chunk_store.base_path()) {
+                cache_capacity = fs_info.available / (16 * 1024 * 1024);
+            }
+            if let Some(max_cache_size) = backend_config.max_cache_size {
+                info!(
+                    "Got requested max cache size {max_cache_size} for store {}",
+                    config.name
+                );
+                let max_cache_capacity = max_cache_size.as_u64() / (16 * 1024 * 1024);
+                cache_capacity = cache_capacity.min(max_cache_capacity);
+            }
+            let cache_capacity = usize::try_from(cache_capacity).unwrap_or_default();
+
+            info!(
+                "Using datastore cache with capacity {cache_capacity} for store {}",
+                config.name
+            );
+
+            let cache = LocalDatastoreLruCache::new(cache_capacity, chunk_store.clone());
+            Some(cache)
+        } else {
+            None
+        };
+
         Ok(DataStoreImpl {
             chunk_store,
             gc_mutex: Mutex::new(()),
@@ -449,6 +512,7 @@ impl DataStore {
             last_digest,
             sync_level: tuning.sync_level.unwrap_or_default(),
             backend_config,
+            lru_store_caching,
         })
     }
 
@@ -1710,7 +1774,7 @@ impl DataStore {
         delete_list: &mut Vec<S3ObjectKey>,
         gc_status: &mut GarbageCollectionStatus,
     ) -> Result<bool, Error> {
-        let chunk_path = match self.chunk_path_from_object_key(&object_key) {
+        let (chunk_path, digest) = match self.chunk_path_from_object_key(&object_key) {
             Some(path) => path,
             None => return Ok(false),
         };
@@ -1731,6 +1795,10 @@ impl DataStore {
         let bad = chunk_path.as_path().ends_with(".bad");
 
         if atime < min_atime {
+            if let Some(cache) = self.cache() {
+                // ignore errors, phase 3 will retry cleanup anyways
+                let _ = cache.remove(&digest);
+            }
             delete_list.push(object_key.clone());
             if bad {
                 gc_status.removed_bad += 1;
@@ -1756,7 +1824,7 @@ impl DataStore {
     }
 
     // Check and generate a chunk path from given object key
-    fn chunk_path_from_object_key(&self, object_key: &S3ObjectKey) -> Option<PathBuf> {
+    fn chunk_path_from_object_key(&self, object_key: &S3ObjectKey) -> Option<(PathBuf, [u8; 32])> {
         // Check object is actually a chunk
         let digest = match Path::new::<str>(object_key).file_name() {
             Some(file_name) => file_name,
@@ -1779,7 +1847,12 @@ impl DataStore {
         chunk_path.push(hexdigit_prefix);
         chunk_path.push(digest);
 
-        Some(chunk_path)
+        let mut digest_bytes = [0u8; 32];
+        let digest = digest.as_bytes();
+        // safe to unwrap as already checked above
+        hex::decode_to_slice(&digest[..64], &mut digest_bytes).unwrap();
+
+        Some((chunk_path, digest_bytes))
     }
 
     pub fn try_shared_chunk_store_lock(&self) -> Result<ProcessLockSharedGuard, Error> {
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..c0edd3619
--- /dev/null
+++ b/pbs-datastore/src/local_datastore_lru_cache.rs
@@ -0,0 +1,180 @@
+//! 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)]
+/// Cacher to fetch chunks from the S3 object store and insert them in the local cache store.
+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 = Arc::clone(&self.client);
+        let store = Arc::clone(&self.store);
+        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 {
+    /// Create a new cache instance storing up to given capacity chunks in the local cache store.
+    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)
+    }
+
+    /// Access the locally cached chunk or fetch it from the S3 object store via the provided
+    /// cacher instance.
+    ///
+    /// For evicted cache nodes, clear the chunk file contents but leave the empty marker file
+    /// behind so garbage collection doesn't clean it while in use.
+    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)
+        }
+    }
+
+    /// Checks if the given digest is stored in the datastores LRU cache
+    pub fn contains(&self, digest: &[u8; 32]) -> bool {
+        self.cache.contains(*digest)
+    }
+}
-- 
2.47.2



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel