[pbs-devel] [PATCH proxmox-backup 6/6] api: admin: datastore: implement streaming content api call

[Dominik Csapak <d.csapak@proxmox.com>]

this is a new api call that utilizes `async-stream` together with
`proxmox_router::Stream` to provide a streaming interface to querying
the datastore content.

This can be done when a client reuqests this api call with the
`application/json-seq` Accept header.

In contrast to the existing api calls, this one
* returns all types of content items (namespaces, groups, snapshots; can
  be filtered with a parameter)
* iterates over them recursively (with the range that is given with the
  parameter)

The api call returns the data in the following order:
* first all visible namespaces
* then for each ns in order
  * each group
  * each snapshot

This is done so that we can have a good way of building a tree view in
the ui.

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
This should be thouroughly checked for permission checks. I did it to
the best of my ability, but of course some bug/issue could have crept in.

interesting side node, in my rather large setup with ~600 groups and ~1000
snapshosts per group, streaming this is faster than using the current
`snapshot` api (by a lot):
* `snapshot` api -> ~3 min
* `content` api with streaming -> ~2:11 min
* `content` api without streaming -> ~3 min

It seems that either collecting such a 'large' api response (~200MiB)
is expensive. My guesses what happens here are either:
* frequent (re)allocation of the resulting vec
* or serde's serializing code

but the cost seems still pretty high for that.
LMK if i should further investigate this.

 Cargo.toml                  |   2 +
 src/api2/admin/datastore.rs | 201 +++++++++++++++++++++++++++++++-----
 2 files changed, 176 insertions(+), 27 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index b3f55b4db..21eb293ba 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -114,6 +114,7 @@ pbs-tools = { path = "pbs-tools" }
 
 # regular crates
 anyhow = "1.0"
+async-stream = "0.3"
 async-trait = "0.1.56"
 apt-pkg-native = "0.3.2"
 bitflags = "2.4"
@@ -168,6 +169,7 @@ zstd-safe = "7"
 [dependencies]
 anyhow.workspace = true
 async-trait.workspace = true
+async-stream.workspace = true
 bytes.workspace = true
 cidr.workspace = true
 const_format.workspace = true
diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index 2252dcfa4..bf94f6400 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -23,7 +23,7 @@ use proxmox_compression::zstd::ZstdEncoder;
 use proxmox_log::LogContext;
 use proxmox_router::{
     http_err, list_subdirs_api_method, ApiHandler, ApiMethod, ApiResponseFuture, Permission,
-    Router, RpcEnvironment, RpcEnvironmentType, SubdirMap,
+    Record, Router, RpcEnvironment, RpcEnvironmentType, SubdirMap,
 };
 use proxmox_rrd_api_types::{RrdMode, RrdTimeframe};
 use proxmox_schema::*;
@@ -39,15 +39,16 @@ use pxar::EntryKind;
 
 use pbs_api_types::{
     print_ns_and_snapshot, print_store_and_ns, ArchiveType, Authid, BackupArchiveName,
-    BackupContent, BackupGroupDeleteStats, BackupNamespace, BackupType, Counts, CryptMode,
-    DataStoreConfig, DataStoreListItem, DataStoreMountStatus, DataStoreStatus,
-    GarbageCollectionJobStatus, GroupListItem, JobScheduleStatus, KeepOptions, MaintenanceMode,
-    MaintenanceType, Operation, PruneJobOptions, SnapshotListItem, SyncJobConfig,
-    BACKUP_ARCHIVE_NAME_SCHEMA, BACKUP_ID_SCHEMA, BACKUP_NAMESPACE_SCHEMA, BACKUP_TIME_SCHEMA,
-    BACKUP_TYPE_SCHEMA, CATALOG_NAME, CLIENT_LOG_BLOB_NAME, DATASTORE_SCHEMA,
-    IGNORE_VERIFIED_BACKUPS_SCHEMA, MAX_NAMESPACE_DEPTH, NS_MAX_DEPTH_SCHEMA, PRIV_DATASTORE_AUDIT,
-    PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_MODIFY, PRIV_DATASTORE_PRUNE, PRIV_DATASTORE_READ,
-    PRIV_DATASTORE_VERIFY, PRIV_SYS_MODIFY, UPID, UPID_SCHEMA, VERIFICATION_OUTDATED_AFTER_SCHEMA,
+    BackupContent, BackupGroupDeleteStats, BackupNamespace, BackupType, ContentListItem,
+    ContentType, Counts, CryptMode, DataStoreConfig, DataStoreListItem, DataStoreMountStatus,
+    DataStoreStatus, GarbageCollectionJobStatus, GroupListItem, JobScheduleStatus, KeepOptions,
+    MaintenanceMode, MaintenanceType, NamespaceListItem, Operation, PruneJobOptions,
+    SnapshotListItem, SyncJobConfig, BACKUP_ARCHIVE_NAME_SCHEMA, BACKUP_ID_SCHEMA,
+    BACKUP_NAMESPACE_SCHEMA, BACKUP_TIME_SCHEMA, BACKUP_TYPE_SCHEMA, CATALOG_NAME,
+    CLIENT_LOG_BLOB_NAME, DATASTORE_SCHEMA, IGNORE_VERIFIED_BACKUPS_SCHEMA, MAX_NAMESPACE_DEPTH,
+    NS_MAX_DEPTH_SCHEMA, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_MODIFY,
+    PRIV_DATASTORE_PRUNE, PRIV_DATASTORE_READ, PRIV_DATASTORE_VERIFY, PRIV_SYS_MODIFY, UPID,
+    UPID_SCHEMA, VERIFICATION_OUTDATED_AFTER_SCHEMA,
 };
 use pbs_client::pxar::{create_tar, create_zip};
 use pbs_config::CachedUserInfo;
@@ -70,7 +71,10 @@ use proxmox_rest_server::{formatter, worker_is_active, WorkerTask};
 
 use crate::api2::backup::optional_ns_param;
 use crate::api2::node::rrd::create_value_from_rrd;
-use crate::backup::{check_ns_privs_full, ListAccessibleBackupGroups, VerifyWorker, NS_PRIVS_OK};
+use crate::backup::{
+    can_access_any_namespace_in_range, check_ns_privs, check_ns_privs_full,
+    ListAccessibleBackupGroups, VerifyWorker, NS_PRIVS_OK,
+};
 use crate::server::jobstate::{compute_schedule_status, Job, JobState};
 use crate::tools::{backup_info_to_snapshot_list_item, get_all_snapshot_files, read_backup_index};
 
@@ -396,7 +400,7 @@ pub async fn delete_snapshot(
 }
 
 #[api(
-    serializing: true,
+    stream: true,
     input: {
         properties: {
             store: { schema: DATASTORE_SCHEMA },
@@ -404,40 +408,137 @@ pub async fn delete_snapshot(
                 type: BackupNamespace,
                 optional: true,
             },
-            "backup-type": {
+            "max-depth": {
+                schema: NS_MAX_DEPTH_SCHEMA,
                 optional: true,
-                type: BackupType,
             },
-            "backup-id": {
+            "content-type": {
                 optional: true,
-                schema: BACKUP_ID_SCHEMA,
+                type: ContentType,
             },
         },
     },
-    returns: pbs_api_types::ADMIN_DATASTORE_LIST_SNAPSHOTS_RETURN_TYPE,
     access: {
         permission: &Permission::Anybody,
         description: "Requires on /datastore/{store}[/{namespace}] either DATASTORE_AUDIT for any \
             or DATASTORE_BACKUP and being the owner of the group",
     },
 )]
-/// List backup snapshots.
-pub async fn list_snapshots(
+/// List datastore content, recursively through all namespaces.
+pub async fn list_content(
     store: String,
     ns: Option<BackupNamespace>,
-    backup_type: Option<BackupType>,
-    backup_id: Option<String>,
+    max_depth: Option<usize>,
+    content_type: Option<ContentType>,
     _param: Value,
     _info: &ApiMethod,
     rpcenv: &mut dyn RpcEnvironment,
-) -> Result<Vec<SnapshotListItem>, Error> {
+) -> Result<proxmox_router::Stream, Error> {
+    let (sender, mut receiver) = tokio::sync::mpsc::channel(128);
+
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
+    let user_info = CachedUserInfo::new()?;
 
-    tokio::task::spawn_blocking(move || unsafe {
-        list_snapshots_blocking(store, ns, backup_type, backup_id, auth_id)
-    })
-    .await
-    .map_err(|err| format_err!("failed to await blocking task: {err}"))?
+    let datastore = DataStore::lookup_datastore(&store, Some(Operation::Read))?;
+    if !can_access_any_namespace_in_range(
+        datastore.clone(),
+        &auth_id,
+        &user_info,
+        ns.clone(),
+        max_depth,
+    ) {
+        proxmox_router::http_bail!(FORBIDDEN, "permission check failed");
+    }
+
+    let ns = ns.unwrap_or_default();
+
+    let (list_ns, list_group, list_snapshots) = match content_type {
+        Some(ContentType::Namespace) => (true, false, false),
+        Some(ContentType::Group) => (false, true, false),
+        Some(ContentType::Snapshot) => (false, false, true),
+        None => (true, true, true),
+    };
+
+    tokio::spawn(async move {
+        tokio::task::spawn_blocking(move || {
+            if list_ns {
+                for ns in datastore.recursive_iter_backup_ns_ok(ns.clone(), max_depth)? {
+                    match check_ns_privs(&store, &ns, &auth_id, NS_PRIVS_OK) {
+                        Ok(_) => sender.blocking_send(Record::new(ContentListItem::from(
+                            NamespaceListItem { ns, comment: None },
+                        )))?,
+                        Err(_) => continue,
+                    }
+                }
+            }
+
+            if !list_group && !list_snapshots {
+                return Ok(());
+            }
+
+            for ns in datastore.recursive_iter_backup_ns_ok(ns, max_depth)? {
+                let list_all = match check_ns_privs_full(
+                    &store,
+                    &ns,
+                    &auth_id,
+                    PRIV_DATASTORE_AUDIT,
+                    PRIV_DATASTORE_BACKUP,
+                ) {
+                    Ok(requires_owner) => !requires_owner,
+                    Err(_) => continue,
+                };
+                if list_group {
+                    for group in datastore.iter_backup_groups(ns.clone())? {
+                        let group = group?;
+                        let group = backup_group_to_group_list_item(
+                            datastore.clone(),
+                            group,
+                            &ns,
+                            &auth_id,
+                            list_all,
+                        );
+
+                        if let Some(group) = group {
+                            sender.blocking_send(Record::new(ContentListItem::from((
+                                ns.clone(),
+                                group,
+                            ))))?;
+                        }
+                    }
+                }
+
+                if !list_snapshots {
+                    continue;
+                }
+
+                for group in datastore.iter_backup_groups(ns.clone())? {
+                    let group = group?;
+                    let owner = match get_group_owner(&store, &ns, &group) {
+                        Some(auth_id) => auth_id,
+                        None => continue,
+                    };
+                    for snapshot in group.iter_snapshots()? {
+                        let snapshot = BackupInfo::new(snapshot?)?;
+                        let snapshot = backup_info_to_snapshot_list_item(&snapshot, &owner);
+                        sender.blocking_send(Record::new(ContentListItem::from((
+                            ns.clone(),
+                            snapshot,
+                        ))))?;
+                    }
+                }
+            }
+            Ok::<_, Error>(())
+        })
+        .await??;
+
+        Ok::<_, Error>(())
+    });
+    Ok(async_stream::try_stream! {
+        while let Some(elem) = receiver.recv().await {
+            yield elem;
+        }
+    }
+    .into())
 }
 
 fn get_group_owner(
@@ -523,6 +624,51 @@ unsafe fn list_snapshots_blocking(
     })
 }
 
+#[api(
+    serializing: true,
+    input: {
+        properties: {
+            store: { schema: DATASTORE_SCHEMA },
+            ns: {
+                type: BackupNamespace,
+                optional: true,
+            },
+            "backup-type": {
+                optional: true,
+                type: BackupType,
+            },
+            "backup-id": {
+                optional: true,
+                schema: BACKUP_ID_SCHEMA,
+            },
+        },
+    },
+    returns: pbs_api_types::ADMIN_DATASTORE_LIST_SNAPSHOTS_RETURN_TYPE,
+    access: {
+        permission: &Permission::Anybody,
+        description: "Requires on /datastore/{store}[/{namespace}] either DATASTORE_AUDIT for any \
+            or DATASTORE_BACKUP and being the owner of the group",
+    },
+)]
+/// List backup snapshots.
+pub async fn list_snapshots(
+    store: String,
+    ns: Option<BackupNamespace>,
+    backup_type: Option<BackupType>,
+    backup_id: Option<String>,
+    _param: Value,
+    _info: &ApiMethod,
+    rpcenv: &mut dyn RpcEnvironment,
+) -> Result<Vec<SnapshotListItem>, Error> {
+    let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
+
+    tokio::task::spawn_blocking(move || unsafe {
+        list_snapshots_blocking(store, ns, backup_type, backup_id, auth_id)
+    })
+    .await
+    .map_err(|err| format_err!("failed to await blocking task: {err}"))?
+}
+
 async fn get_snapshots_count(
     store: &Arc<DataStore>,
     owner: Option<&Authid>,
@@ -2773,6 +2919,7 @@ const DATASTORE_INFO_SUBDIRS: SubdirMap = &[
         "change-owner",
         &Router::new().post(&API_METHOD_SET_BACKUP_OWNER),
     ),
+    ("content", &Router::new().get(&API_METHOD_LIST_CONTENT)),
     (
         "download",
         &Router::new().download(&API_METHOD_DOWNLOAD_FILE),
-- 
2.47.3



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