[pbs-devel] [PATCH proxmox-backup 2/7] pull: allow pulling groups selectively

["Fabian Grünbichler" <f.gruenbichler@proxmox.com>]

without requiring workarounds based on ownership and limited
visibility/access.

if a group filter is set, remove_vanished will only consider filtered
groups for removal to prevent concurrent disjunct filters from trashing
eachother's synced groups.

Signed-off-by: Fabian Grünbichler <f.gruenbichler@proxmox.com>
---
 src/api2/pull.rs                  | 29 +++++++++++++++++++++++++----
 src/bin/proxmox-backup-manager.rs | 14 ++++++++++++++
 src/server/pull.rs                | 24 ++++++++++++++++++++++--
 3 files changed, 61 insertions(+), 6 deletions(-)

diff --git a/src/api2/pull.rs b/src/api2/pull.rs
index 4893c9fb..36149761 100644
--- a/src/api2/pull.rs
+++ b/src/api2/pull.rs
@@ -1,5 +1,6 @@
 //! Sync datastore from remote server
 use std::sync::{Arc};
+use std::str::FromStr;
 
 use anyhow::{format_err, Error};
 use futures::{select, future::FutureExt};
@@ -10,9 +11,9 @@ use proxmox::api::{ApiMethod, Router, RpcEnvironment, Permission};
 use pbs_client::{HttpClient, BackupRepository};
 
 use crate::server::{WorkerTask, jobstate::Job, pull::pull_store};
-use crate::backup::DataStore;
+use crate::backup::{BackupGroup, DataStore};
 use crate::api2::types::{
-    DATASTORE_SCHEMA, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA, Authid,
+    BACKUP_GROUP_SCHEMA, DATASTORE_SCHEMA, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA, Authid,
 };
 use crate::config::{
     remote,
@@ -101,7 +102,7 @@ pub fn do_sync_job(
                 worker.log(format!("Sync datastore '{}' from '{}/{}'",
                         sync_job.store, sync_job.remote, sync_job.remote_store));
 
-                pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, sync_owner).await?;
+                pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, sync_owner, None).await?;
 
                 worker.log(format!("sync job '{}' end", &job_id));
 
@@ -152,6 +153,14 @@ pub fn do_sync_job(
                 schema: REMOVE_VANISHED_BACKUPS_SCHEMA,
                 optional: true,
             },
+            "groups": {
+                type: Array,
+                description: "List of group identifiers to filter for. All if unspecified.",
+                items: {
+                    schema: BACKUP_GROUP_SCHEMA,
+                },
+                optional: true,
+            },
         },
     },
     access: {
@@ -169,6 +178,7 @@ async fn pull (
     remote: String,
     remote_store: String,
     remove_vanished: Option<bool>,
+    groups: Option<Vec<String>>,
     _info: &ApiMethod,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<String, Error> {
@@ -185,7 +195,18 @@ async fn pull (
 
         worker.log(format!("sync datastore '{}' start", store));
 
-        let pull_future = pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, auth_id);
+        let groups = match groups {
+            Some(filter) => {
+                let mut groups = std::collections::HashSet::with_capacity(filter.len());
+                for group in filter {
+                    let group = BackupGroup::from_str(&group)?;
+                    groups.insert(group);
+                }
+                Some(groups)
+            },
+            None => None,
+        };
+        let pull_future = pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, auth_id, groups);
         let future = select!{
             success = pull_future.fuse() => success,
             abort = worker.abort_future().map(|_| Err(format_err!("pull aborted"))) => abort,
diff --git a/src/bin/proxmox-backup-manager.rs b/src/bin/proxmox-backup-manager.rs
index 93d6de57..6844a1ab 100644
--- a/src/bin/proxmox-backup-manager.rs
+++ b/src/bin/proxmox-backup-manager.rs
@@ -230,6 +230,15 @@ fn task_mgmt_cli() -> CommandLineInterface {
                 schema: REMOVE_VANISHED_BACKUPS_SCHEMA,
                 optional: true,
             },
+            "groups": {
+                type: Array,
+                description: "List of group identifiers to filter for. All if unspecified.",
+                items: {
+                    type: String,
+                    description: "Backup group identifier",
+                },
+                optional: true,
+            },
             "output-format": {
                 schema: OUTPUT_FORMAT,
                 optional: true,
@@ -243,6 +252,7 @@ async fn pull_datastore(
     remote_store: String,
     local_store: String,
     remove_vanished: Option<bool>,
+    groups: Option<Vec<String>>,
     param: Value,
 ) -> Result<Value, Error> {
 
@@ -256,6 +266,10 @@ async fn pull_datastore(
         "remote-store": remote_store,
     });
 
+    if groups.is_some() {
+        args["groups"] = json!(groups);
+    }
+
     if let Some(remove_vanished) = remove_vanished {
         args["remove-vanished"] = Value::from(remove_vanished);
     }
diff --git a/src/server/pull.rs b/src/server/pull.rs
index 5214a218..2904c37f 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -631,6 +631,7 @@ pub async fn pull_store(
     tgt_store: Arc<DataStore>,
     delete: bool,
     auth_id: Authid,
+    group_filter: Option<HashSet<BackupGroup>>,
 ) -> Result<(), Error> {
     // explicit create shared lock to prevent GC on newly created chunks
     let _shared_store_lock = tgt_store.try_shared_chunk_store_lock()?;
@@ -644,8 +645,7 @@ pub async fn pull_store(
 
     let mut list: Vec<GroupListItem> = serde_json::from_value(result["data"].take())?;
 
-    worker.log(format!("found {} groups to sync", list.len()));
-
+    let total_count = list.len();
     list.sort_unstable_by(|a, b| {
         let type_order = a.backup_type.cmp(&b.backup_type);
         if type_order == std::cmp::Ordering::Equal {
@@ -655,6 +655,21 @@ pub async fn pull_store(
         }
     });
 
+
+    let list = if let Some(ref group_filter) = &group_filter {
+        let list:Vec<GroupListItem> = list
+            .into_iter()
+            .filter(|group| {
+                group_filter.contains(&BackupGroup::new(&group.backup_type, &group.backup_id))
+            })
+            .collect();
+        worker.log(format!("found {} groups to sync (out of {} requested by filter)", list.len(), group_filter.len()));
+        list
+    } else {
+        worker.log(format!("found {} groups to sync", total_count));
+        list
+    };
+
     let mut errors = false;
 
     let mut new_groups = std::collections::HashSet::new();
@@ -717,6 +732,11 @@ pub async fn pull_store(
                 if new_groups.contains(&local_group) {
                     continue;
                 }
+                if let Some(ref group_filter) = &group_filter {
+                    if !group_filter.contains(&local_group) {
+                        continue;
+                    }
+                }
                 worker.log(format!(
                     "delete vanished group '{}/{}'",
                     local_group.backup_type(),
-- 
2.30.2