[pbs-devel] [PATCH proxmox-backup v8 14/45] api: reader: fetch chunks based on datastore backend

[Lukas Wagner]

One comment inline, but nothing prohibitive of a R-b:

Reviewed-by: Lukas Wagner <l.wagner at proxmox.com>


On  2025-07-15 14:53, Christian Ebner wrote:
> Read the chunk based on the datastores backend, reading from local
> filesystem or fetching from S3 object store.
> 
> Signed-off-by: Christian Ebner <c.ebner at proxmox.com>
> ---
> changes since version 7:
> - no changes
> 
>  src/api2/reader/environment.rs | 12 ++++++----
>  src/api2/reader/mod.rs         | 41 +++++++++++++++++++++++-----------
>  2 files changed, 36 insertions(+), 17 deletions(-)
> 
> diff --git a/src/api2/reader/environment.rs b/src/api2/reader/environment.rs
> index 3b2f06f43..8924352b0 100644
> --- a/src/api2/reader/environment.rs
> +++ b/src/api2/reader/environment.rs
> @@ -1,13 +1,14 @@
>  use std::collections::HashSet;
>  use std::sync::{Arc, RwLock};
>  
> +use anyhow::Error;
>  use serde_json::{json, Value};
>  
>  use proxmox_router::{RpcEnvironment, RpcEnvironmentType};
>  
>  use pbs_api_types::Authid;
>  use pbs_datastore::backup_info::BackupDir;
> -use pbs_datastore::DataStore;
> +use pbs_datastore::{DataStore, DatastoreBackend};
>  use proxmox_rest_server::formatter::*;
>  use proxmox_rest_server::WorkerTask;
>  use tracing::info;
> @@ -23,6 +24,7 @@ pub struct ReaderEnvironment {
>      pub worker: Arc<WorkerTask>,
>      pub datastore: Arc<DataStore>,
>      pub backup_dir: BackupDir,
> +    pub backend: DatastoreBackend,
>      allowed_chunks: Arc<RwLock<HashSet<[u8; 32]>>>,
>  }
>  
> @@ -33,8 +35,9 @@ impl ReaderEnvironment {
>          worker: Arc<WorkerTask>,
>          datastore: Arc<DataStore>,
>          backup_dir: BackupDir,
> -    ) -> Self {
> -        Self {
> +    ) -> Result<Self, Error> {
> +        let backend = datastore.backend()?;
> +        Ok(Self {
>              result_attributes: json!({}),
>              env_type,
>              auth_id,
> @@ -43,8 +46,9 @@ impl ReaderEnvironment {
>              debug: tracing::enabled!(tracing::Level::DEBUG),
>              formatter: JSON_FORMATTER,
>              backup_dir,
> +            backend,
>              allowed_chunks: Arc::new(RwLock::new(HashSet::new())),
> -        }
> +        })
>      }
>  
>      pub fn log<S: AsRef<str>>(&self, msg: S) {
> diff --git a/src/api2/reader/mod.rs b/src/api2/reader/mod.rs
> index a77216043..997d9ca77 100644
> --- a/src/api2/reader/mod.rs
> +++ b/src/api2/reader/mod.rs
> @@ -3,6 +3,7 @@
>  use anyhow::{bail, format_err, Context, Error};
>  use futures::*;
>  use hex::FromHex;
> +use http_body_util::BodyExt;
>  use hyper::body::Incoming;
>  use hyper::header::{self, HeaderValue, CONNECTION, UPGRADE};
>  use hyper::http::request::Parts;
> @@ -27,8 +28,9 @@ use pbs_api_types::{
>  };
>  use pbs_config::CachedUserInfo;
>  use pbs_datastore::index::IndexFile;
> -use pbs_datastore::{DataStore, PROXMOX_BACKUP_READER_PROTOCOL_ID_V1};
> +use pbs_datastore::{DataStore, DatastoreBackend, PROXMOX_BACKUP_READER_PROTOCOL_ID_V1};
>  use pbs_tools::json::required_string_param;
> +use proxmox_s3_client::S3Client;
>  
>  use crate::api2::backup::optional_ns_param;
>  use crate::api2::helpers;
> @@ -162,7 +164,7 @@ fn upgrade_to_backup_reader_protocol(
>                      worker.clone(),
>                      datastore,
>                      backup_dir,
> -                );
> +                )?;
>  
>                  env.debug = debug;
>  
> @@ -323,17 +325,10 @@ fn download_chunk(
>              ));
>          }
>  
> -        let (path, _) = env.datastore.chunk_path(&digest);
> -        let path2 = path.clone();
> -
> -        env.debug(format!("download chunk {:?}", path));
> -
> -        let data =
> -            proxmox_async::runtime::block_in_place(|| std::fs::read(path)).map_err(move |err| {
> -                http_err!(BAD_REQUEST, "reading file {:?} failed: {}", path2, err)
> -            })?;
> -
> -        let body = Body::from(data);
> +        let body = match &env.backend {
> +            DatastoreBackend::Filesystem => load_from_filesystem(env, &digest)?,
> +            DatastoreBackend::S3(s3_client) => fetch_from_object_store(s3_client, &digest).await?,
> +        };
>  
>          // fixme: set other headers ?
>          Ok(Response::builder()
> @@ -345,6 +340,26 @@ fn download_chunk(
>      .boxed()
>  }
>  
> +async fn fetch_from_object_store(s3_client: &S3Client, digest: &[u8; 32]) -> Result<Body, Error> {
> +    let object_key = pbs_datastore::s3::object_key_from_digest(digest)?;
> +    if let Some(response) = s3_client.get_object(object_key).await? {

^ Do we maybe want some kind of retry-logic for retrieving objects as well? Disregard
in case you implement it in a later patch, I'm reviewing this series patch by patch.

> +        let data = response.content.collect().await?.to_bytes();
> +        return Ok(Body::from(data));
> +    }
> +    bail!("cannot find chunk with digest {}", hex::encode(digest));
> +}
> +
> +fn load_from_filesystem(env: &ReaderEnvironment, digest: &[u8; 32]) -> Result<Body, Error> {
> +    let (path, _) = env.datastore.chunk_path(digest);
> +    let path2 = path.clone();
> +
> +    env.debug(format!("download chunk {path:?}"));
> +
> +    let data = proxmox_async::runtime::block_in_place(|| std::fs::read(path))
> +        .map_err(move |err| http_err!(BAD_REQUEST, "reading file {path2:?} failed: {err}"))?;
> +    Ok(Body::from(data))
> +}
> +
>  /* this is too slow
>  fn download_chunk_old(
>      _parts: Parts,

-- 
- Lukas