From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from firstgate.proxmox.com (firstgate.proxmox.com [IPv6:2a01:7e0:0:424::9]) by lore.proxmox.com (Postfix) with ESMTPS id 8B4BC1FF185 for ; Mon, 21 Jul 2025 18:48:50 +0200 (CEST) Received: from firstgate.proxmox.com (localhost [127.0.0.1]) by firstgate.proxmox.com (Proxmox) with ESMTP id BFECC17118; Mon, 21 Jul 2025 18:49:57 +0200 (CEST) From: Christian Ebner To: pbs-devel@lists.proxmox.com Date: Mon, 21 Jul 2025 18:44:36 +0200 Message-ID: <20250721164507.1045869-19-c.ebner@proxmox.com> X-Mailer: git-send-email 2.47.2 In-Reply-To: <20250721164507.1045869-1-c.ebner@proxmox.com> References: <20250721164507.1045869-1-c.ebner@proxmox.com> MIME-Version: 1.0 X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1753116307744 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.045 Adjusted score from AWL reputation of From: address BAYES_00 -1.9 Bayes spam probability is 0 to 1% DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Subject: [pbs-devel] [PATCH proxmox-backup v10 15/46] datastore: local chunk reader: read chunks based on backend X-BeenThere: pbs-devel@lists.proxmox.com X-Mailman-Version: 2.1.29 Precedence: list List-Id: Proxmox Backup Server development discussion List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: Proxmox Backup Server development discussion Content-Type: text/plain; charset="us-ascii" Content-Transfer-Encoding: 7bit Errors-To: pbs-devel-bounces@lists.proxmox.com Sender: "pbs-devel" Get and store the datastore's backend on local chunk reader instantiantion and fetch chunks based on the variant from either the filesystem or the s3 object store. By storing the backend variant, the s3 client is instantiated only once and reused until the local chunk reader instance is dropped. Signed-off-by: Christian Ebner --- changes since version 9: - propagate backend creation error for local chunk reader pbs-datastore/Cargo.toml | 1 + pbs-datastore/src/local_chunk_reader.rs | 43 +++++++++++++++++++------ src/api2/admin/datastore.rs | 12 ++++--- src/server/pull.rs | 8 +++-- src/server/push.rs | 8 +++-- src/server/sync.rs | 22 +++++-------- 6 files changed, 63 insertions(+), 31 deletions(-) diff --git a/pbs-datastore/Cargo.toml b/pbs-datastore/Cargo.toml index 7e56dbd31..8ce930a94 100644 --- a/pbs-datastore/Cargo.toml +++ b/pbs-datastore/Cargo.toml @@ -13,6 +13,7 @@ crc32fast.workspace = true endian_trait.workspace = true futures.workspace = true hex = { workspace = true, features = [ "serde" ] } +http-body-util.workspace = true hyper.workspace = true libc.workspace = true log.workspace = true diff --git a/pbs-datastore/src/local_chunk_reader.rs b/pbs-datastore/src/local_chunk_reader.rs index 05a70c068..58a2fee8d 100644 --- a/pbs-datastore/src/local_chunk_reader.rs +++ b/pbs-datastore/src/local_chunk_reader.rs @@ -3,17 +3,21 @@ use std::pin::Pin; use std::sync::Arc; use anyhow::{bail, Error}; +use http_body_util::BodyExt; use pbs_api_types::CryptMode; use pbs_tools::crypt_config::CryptConfig; +use proxmox_s3_client::S3Client; use crate::data_blob::DataBlob; +use crate::datastore::DatastoreBackend; use crate::read_chunk::{AsyncReadChunk, ReadChunk}; use crate::DataStore; #[derive(Clone)] pub struct LocalChunkReader { store: Arc, + backend: DatastoreBackend, crypt_config: Option>, crypt_mode: CryptMode, } @@ -23,12 +27,14 @@ impl LocalChunkReader { store: Arc, crypt_config: Option>, crypt_mode: CryptMode, - ) -> Self { - Self { + ) -> Result { + let backend = store.backend()?; + Ok(Self { store, + backend, crypt_config, crypt_mode, - } + }) } fn ensure_crypt_mode(&self, chunk_mode: CryptMode) -> Result<(), Error> { @@ -47,10 +53,26 @@ impl LocalChunkReader { } } +async fn fetch(s3_client: Arc, digest: &[u8; 32]) -> Result { + let object_key = crate::s3::object_key_from_digest(digest)?; + if let Some(response) = s3_client.get_object(object_key).await? { + let bytes = response.content.collect().await?.to_bytes(); + DataBlob::from_raw(bytes.to_vec()) + } else { + bail!("no object with digest {}", hex::encode(digest)); + } +} + impl ReadChunk for LocalChunkReader { fn read_raw_chunk(&self, digest: &[u8; 32]) -> Result { - let chunk = self.store.load_chunk(digest)?; + let chunk = match &self.backend { + DatastoreBackend::Filesystem => self.store.load_chunk(digest)?, + DatastoreBackend::S3(s3_client) => { + proxmox_async::runtime::block_on(fetch(Arc::clone(s3_client), digest))? + } + }; self.ensure_crypt_mode(chunk.crypt_mode()?)?; + Ok(chunk) } @@ -69,11 +91,14 @@ impl AsyncReadChunk for LocalChunkReader { digest: &'a [u8; 32], ) -> Pin> + Send + 'a>> { Box::pin(async move { - let (path, _) = self.store.chunk_path(digest); - - let raw_data = tokio::fs::read(&path).await?; - - let chunk = DataBlob::load_from_reader(&mut &raw_data[..])?; + let chunk = match &self.backend { + DatastoreBackend::Filesystem => { + let (path, _) = self.store.chunk_path(digest); + let raw_data = tokio::fs::read(&path).await?; + DataBlob::load_from_reader(&mut &raw_data[..])? + } + DatastoreBackend::S3(s3_client) => fetch(Arc::clone(s3_client), digest).await?, + }; self.ensure_crypt_mode(chunk.crypt_mode()?)?; Ok(chunk) diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs index e6b8449d2..d742633cf 100644 --- a/src/api2/admin/datastore.rs +++ b/src/api2/admin/datastore.rs @@ -1520,7 +1520,8 @@ pub fn download_file_decoded( let (csum, size) = index.compute_csum(); manifest.verify_file(&file_name, &csum, size)?; - let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None); + let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None) + .context("creating local chunk reader failed")?; let reader = CachedChunkReader::new(chunk_reader, index, 1).seekable(); Body::wrap_stream(AsyncReaderStream::new(reader).map_err(move |err| { eprintln!("error during streaming of '{:?}' - {}", path, err); @@ -1535,7 +1536,8 @@ pub fn download_file_decoded( let (csum, size) = index.compute_csum(); manifest.verify_file(&file_name, &csum, size)?; - let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None); + let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None) + .context("creating local chunk reader failed")?; let reader = CachedChunkReader::new(chunk_reader, index, 1).seekable(); Body::wrap_stream( AsyncReaderStream::with_buffer_size(reader, 4 * 1024 * 1024).map_err( @@ -1739,7 +1741,8 @@ pub async fn catalog( let (csum, size) = index.compute_csum(); manifest.verify_file(&file_name, &csum, size)?; - let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None); + let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None) + .context("creating local chunk reader failed")?; let reader = BufferedDynamicReader::new(index, chunk_reader); let mut catalog_reader = CatalogReader::new(reader); @@ -1808,7 +1811,8 @@ fn get_local_pxar_reader( let (csum, size) = index.compute_csum(); manifest.verify_file(pxar_name, &csum, size)?; - let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None); + let chunk_reader = LocalChunkReader::new(datastore, None, CryptMode::None) + .context("creating local chunk reader failed")?; let reader = BufferedDynamicReader::new(index, chunk_reader); let archive_size = reader.archive_size(); diff --git a/src/server/pull.rs b/src/server/pull.rs index 775ed0c59..a4402b720 100644 --- a/src/server/pull.rs +++ b/src/server/pull.rs @@ -306,7 +306,9 @@ async fn pull_single_archive<'a>( info!("skipping chunk sync for same datastore"); } else { let stats = pull_index_chunks( - reader.chunk_reader(archive_info.crypt_mode), + reader + .chunk_reader(archive_info.crypt_mode) + .context("failed to get chunk reader")?, snapshot.datastore().clone(), index, downloaded_chunks, @@ -326,7 +328,9 @@ async fn pull_single_archive<'a>( info!("skipping chunk sync for same datastore"); } else { let stats = pull_index_chunks( - reader.chunk_reader(archive_info.crypt_mode), + reader + .chunk_reader(archive_info.crypt_mode) + .context("failed to get chunk reader")?, snapshot.datastore().clone(), index, downloaded_chunks, diff --git a/src/server/push.rs b/src/server/push.rs index e71012ed8..528eed9ff 100644 --- a/src/server/push.rs +++ b/src/server/push.rs @@ -890,7 +890,9 @@ pub(crate) async fn push_snapshot( .await; } let index = DynamicIndexReader::open(&path)?; - let chunk_reader = reader.chunk_reader(entry.chunk_crypt_mode()); + let chunk_reader = reader + .chunk_reader(entry.chunk_crypt_mode()) + .context("failed to get chunk reader")?; let sync_stats = push_index( &archive_name, index, @@ -914,7 +916,9 @@ pub(crate) async fn push_snapshot( .await; } let index = FixedIndexReader::open(&path)?; - let chunk_reader = reader.chunk_reader(entry.chunk_crypt_mode()); + let chunk_reader = reader + .chunk_reader(entry.chunk_crypt_mode()) + .context("failed to get chunk reader")?; let size = index.index_bytes(); let sync_stats = push_index( &archive_name, diff --git a/src/server/sync.rs b/src/server/sync.rs index 09814ef0c..9238a8626 100644 --- a/src/server/sync.rs +++ b/src/server/sync.rs @@ -87,7 +87,7 @@ impl SyncStats { /// and checking whether chunk sync should be skipped. pub(crate) trait SyncSourceReader: Send + Sync { /// Returns a chunk reader with the specified encryption mode. - fn chunk_reader(&self, crypt_mode: CryptMode) -> Arc; + fn chunk_reader(&self, crypt_mode: CryptMode) -> Result, Error>; /// Asynchronously loads a file from the source into a local file. /// `filename` is the name of the file to load from the source. @@ -113,13 +113,10 @@ pub(crate) struct LocalSourceReader { #[async_trait::async_trait] impl SyncSourceReader for RemoteSourceReader { - fn chunk_reader(&self, crypt_mode: CryptMode) -> Arc { - Arc::new(RemoteChunkReader::new( - self.backup_reader.clone(), - None, - crypt_mode, - HashMap::new(), - )) + fn chunk_reader(&self, crypt_mode: CryptMode) -> Result, Error> { + let chunk_reader = + RemoteChunkReader::new(self.backup_reader.clone(), None, crypt_mode, HashMap::new()); + Ok(Arc::new(chunk_reader)) } async fn load_file_into(&self, filename: &str, into: &Path) -> Result, Error> { @@ -190,12 +187,9 @@ impl SyncSourceReader for RemoteSourceReader { #[async_trait::async_trait] impl SyncSourceReader for LocalSourceReader { - fn chunk_reader(&self, crypt_mode: CryptMode) -> Arc { - Arc::new(LocalChunkReader::new( - self.datastore.clone(), - None, - crypt_mode, - )) + fn chunk_reader(&self, crypt_mode: CryptMode) -> Result, Error> { + let chunk_reader = LocalChunkReader::new(self.datastore.clone(), None, crypt_mode)?; + Ok(Arc::new(chunk_reader)) } async fn load_file_into(&self, filename: &str, into: &Path) -> Result, Error> { -- 2.47.2 _______________________________________________ pbs-devel mailing list pbs-devel@lists.proxmox.com https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel