[pdm-devel] [RFC PATCH datacenter-manager 1/2] server: add new streaming 'content' api call for pbs

[Dominik Csapak]

that makes use of 'application/json-seq' mime type to stream data from
pbs to pdm, removing the need to buffer the data on pdm in the api call.

The pbs-client code is more or less a copy of the current snapshot call
list, but if needed this method can be refactored.

This requires the new 'content' api call for pbs [0]

0: https://lore.proxmox.com/pbs-devel/20251008134344.3512958-1-d.csapak@proxmox.com/

Signed-off-by: Dominik Csapak <d.csapak at proxmox.com>
---
 server/src/api/pbs/mod.rs | 40 +++++++++++++++++++++++++++++-
 server/src/pbs_client.rs  | 51 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 90 insertions(+), 1 deletion(-)

diff --git a/server/src/api/pbs/mod.rs b/server/src/api/pbs/mod.rs
index dc31f620..6e53ce6f 100644
--- a/server/src/api/pbs/mod.rs
+++ b/server/src/api/pbs/mod.rs
@@ -1,7 +1,7 @@
 use anyhow::{format_err, Error};
 use futures::StreamExt;
 
-use proxmox_router::{list_subdirs_api_method, Permission, Router, SubdirMap};
+use proxmox_router::{list_subdirs_api_method, Permission, Record, Router, SubdirMap};
 use proxmox_schema::api;
 use proxmox_schema::property_string::PropertyString;
 use proxmox_sortable_macro::sortable;
@@ -63,6 +63,7 @@ const DATASTORE_ITEM_SUBDIRS: SubdirMap = &sorted!([
         "snapshots",
         &Router::new().get(&API_METHOD_LIST_SNAPSHOTS_2)
     ),
+    ("content", &Router::new().get(&API_METHOD_LIST_CONTENT)),
 ]);
 
 #[api(
@@ -175,6 +176,43 @@ async fn list_snapshots_2(
     .into())
 }
 
+#[api(
+    stream: true,
+    input: {
+        properties: {
+            remote: { schema: REMOTE_ID_SCHEMA },
+            datastore: { schema: pbs_api_types::DATASTORE_SCHEMA },
+            ns: {
+                schema: pbs_api_types::BACKUP_NAMESPACE_SCHEMA,
+                optional: true,
+            },
+            "max-depth": {
+                schema: pbs_api_types::NS_MAX_DEPTH_SCHEMA,
+                optional: true,
+            }
+        },
+    },
+    returns: pbs_api_types::ADMIN_DATASTORE_LIST_CONTENT_RETURN_TYPE,
+    access: {
+        permission: &Permission::Privilege(&["resource", "{remote}", "datastore", "{datastore}"], PRIV_RESOURCE_AUDIT, false),
+    },
+)]
+/// List the PBS remote's datastores.
+async fn list_content(
+    remote: String,
+    datastore: String,
+    ns: Option<String>,
+    max_depth: Option<usize>,
+) -> Result<proxmox_router::Stream, Error> {
+    let (remotes, _) = pdm_config::remotes::config()?;
+    let remote = get_remote(&remotes, &remote)?;
+    let snapshots = connection::make_pbs_client(remote)?
+        .list_content(&datastore, ns.as_deref(), max_depth)
+        .await?
+        .map(Record::from_result);
+    Ok(snapshots.into())
+}
+
 #[api(
     input: {
         properties: {
diff --git a/server/src/pbs_client.rs b/server/src/pbs_client.rs
index d8278c8a..76041090 100644
--- a/server/src/pbs_client.rs
+++ b/server/src/pbs_client.rs
@@ -190,6 +190,57 @@ impl PbsClient {
         }
     }
 
+    /// List a datastore's content.
+    pub async fn list_content(
+        &self,
+        datastore: &str,
+        namespace: Option<&str>,
+        max_depth: Option<usize>,
+    ) -> Result<JsonRecords<pbs_api_types::DatastoreContent>, anyhow::Error> {
+        let path = ApiPathBuilder::new(format!("/api2/json/admin/datastore/{datastore}/content"))
+            .maybe_arg("ns", &namespace)
+            .maybe_arg("max-depth", &max_depth)
+            .build();
+        let response = self
+            .0
+            .streaming_request(http::Method::GET, &path, None::<()>)
+            .await?;
+
+        let body = response
+            .body
+            .ok_or_else(|| Error::Other("missing response body"))?;
+
+        if response.status == 200 {
+            if response
+                .content_type
+                .is_some_and(|c| c.starts_with("application/json-seq"))
+            {
+                Ok(JsonRecords::from_body(body))
+            } else {
+                let response: JsonData<_> = serde_json::from_slice(
+                    &body
+                        .collect()
+                        .await
+                        .map_err(|err| {
+                            Error::Anyhow(Box::new(err).context("failed to retrieve response body"))
+                        })?
+                        .to_bytes(),
+                )?;
+                Ok(JsonRecords::from_vec(response.data))
+            }
+        } else {
+            let data = body
+                .collect()
+                .await
+                .map_err(|err| {
+                    Error::Anyhow(Box::new(err).context("failed to retrieve response body"))
+                })?
+                .to_bytes();
+            let error = String::from_utf8_lossy(&data).into_owned();
+            Err(anyhow::Error::msg(error))
+        }
+    }
+
     /// create an API-Token on the PBS remote and give the token admin ACL on everything.
     pub async fn create_admin_token(
         &self,
-- 
2.47.3