[pbs-devel] [RFC proxmox-backup 1/4] api: config/sync: add optional group-sync-tasks property

[Christian Ebner]

Allow to configure from 1 up to 8 concurrent tasks to perform
multiple group syncs concurrently.
The property is exposed via the sync job config and passed to
the pull parameters for the sync job to setup and execute the tasks
accordingly.

Implements the schema definitions and includes the new property to
the `SyncJobConfig` and `PullParameters`.

Signed-off-by: Christian Ebner <c.ebner at proxmox.com>
---
 pbs-api-types/src/jobs.rs | 14 ++++++++++++++
 src/api2/config/sync.rs   | 10 ++++++++++
 src/api2/pull.rs          | 13 ++++++++++---
 src/server/pull.rs        |  4 ++++
 4 files changed, 38 insertions(+), 3 deletions(-)

diff --git a/pbs-api-types/src/jobs.rs b/pbs-api-types/src/jobs.rs
index 868702bc0..5e58787f7 100644
--- a/pbs-api-types/src/jobs.rs
+++ b/pbs-api-types/src/jobs.rs
@@ -64,6 +64,14 @@ pub const REMOVE_VANISHED_BACKUPS_SCHEMA: Schema = BooleanSchema::new(
 .default(false)
 .schema();
 
+const MAX_GROUP_SYNC_TASKS: usize = 8;
+pub const GROUP_SYNC_TASKS_SCHEMA: Schema =
+    IntegerSchema::new("Number of concurrent group pull tasks for sync jobs")
+        .minimum(1)
+        .maximum(MAX_GROUP_SYNC_TASKS as isize)
+        .default(1)
+        .schema();
+
 #[api(
     properties: {
         "next-run": {
@@ -552,6 +560,10 @@ pub const TRANSFER_LAST_SCHEMA: Schema =
             schema: TRANSFER_LAST_SCHEMA,
             optional: true,
         },
+        "group-sync-tasks": {
+            schema: GROUP_SYNC_TASKS_SCHEMA,
+            optional: true,
+        },
     }
 )]
 #[derive(Serialize, Deserialize, Clone, Updater, PartialEq)]
@@ -585,6 +597,8 @@ pub struct SyncJobConfig {
     pub limit: RateLimitConfig,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub transfer_last: Option<usize>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group_sync_tasks: Option<usize>,
 }
 
 impl SyncJobConfig {
diff --git a/src/api2/config/sync.rs b/src/api2/config/sync.rs
index 6fdc69a9e..b6cf81328 100644
--- a/src/api2/config/sync.rs
+++ b/src/api2/config/sync.rs
@@ -231,6 +231,8 @@ pub enum DeletableProperty {
     MaxDepth,
     /// Delete the transfer_last property,
     TransferLast,
+    /// Delete the group_sync_tasks property,
+    GroupSyncTasks,
 }
 
 #[api(
@@ -331,6 +333,9 @@ pub fn update_sync_job(
                 DeletableProperty::TransferLast => {
                     data.transfer_last = None;
                 }
+                DeletableProperty::GroupSyncTasks => {
+                    data.group_sync_tasks = None;
+                }
             }
         }
     }
@@ -369,6 +374,10 @@ pub fn update_sync_job(
         data.transfer_last = Some(transfer_last);
     }
 
+    if let Some(group_sync_tasks) = update.group_sync_tasks {
+        data.group_sync_tasks = Some(group_sync_tasks);
+    }
+
     if update.limit.rate_in.is_some() {
         data.limit.rate_in = update.limit.rate_in;
     }
@@ -533,6 +542,7 @@ acl:1:/remote/remote1/remotestore1:write at pbs:RemoteSyncOperator
         schedule: None,
         limit: pbs_api_types::RateLimitConfig::default(), // no limit
         transfer_last: None,
+        group_sync_tasks: None,
     };
 
     // should work without ACLs
diff --git a/src/api2/pull.rs b/src/api2/pull.rs
index e733c9839..0756e0a51 100644
--- a/src/api2/pull.rs
+++ b/src/api2/pull.rs
@@ -8,9 +8,9 @@ use proxmox_schema::api;
 
 use pbs_api_types::{
     Authid, BackupNamespace, GroupFilter, RateLimitConfig, SyncJobConfig, DATASTORE_SCHEMA,
-    GROUP_FILTER_LIST_SCHEMA, NS_MAX_DEPTH_REDUCED_SCHEMA, PRIV_DATASTORE_BACKUP,
-    PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA,
-    TRANSFER_LAST_SCHEMA,
+    GROUP_FILTER_LIST_SCHEMA, GROUP_SYNC_TASKS_SCHEMA, NS_MAX_DEPTH_REDUCED_SCHEMA,
+    PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ, REMOTE_ID_SCHEMA,
+    REMOVE_VANISHED_BACKUPS_SCHEMA, TRANSFER_LAST_SCHEMA,
 };
 use pbs_config::CachedUserInfo;
 use proxmox_human_byte::HumanByte;
@@ -89,6 +89,7 @@ impl TryFrom<&SyncJobConfig> for PullParameters {
             sync_job.group_filter.clone(),
             sync_job.limit.clone(),
             sync_job.transfer_last,
+            sync_job.group_sync_tasks,
         )
     }
 }
@@ -240,6 +241,10 @@ pub fn do_sync_job(
                 schema: TRANSFER_LAST_SCHEMA,
                 optional: true,
             },
+            "group-sync-tasks": {
+                schema: GROUP_SYNC_TASKS_SCHEMA,
+                optional: true,
+            },
         },
     },
     access: {
@@ -264,6 +269,7 @@ async fn pull(
     group_filter: Option<Vec<GroupFilter>>,
     limit: RateLimitConfig,
     transfer_last: Option<usize>,
+    group_sync_tasks: Option<usize>,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<String, Error> {
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
@@ -301,6 +307,7 @@ async fn pull(
         group_filter,
         limit,
         transfer_last,
+        group_sync_tasks,
     )?;
 
     // fixme: set to_stdout to false?
diff --git a/src/server/pull.rs b/src/server/pull.rs
index 823515e9a..80443132e 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -499,6 +499,8 @@ pub(crate) struct PullParameters {
     group_filter: Vec<GroupFilter>,
     /// How many snapshots should be transferred at most (taking the newest N snapshots)
     transfer_last: Option<usize>,
+    /// Number of concurrent group pull tasks for sync job
+    group_sync_tasks: Option<usize>,
 }
 
 impl PullParameters {
@@ -516,6 +518,7 @@ impl PullParameters {
         group_filter: Option<Vec<GroupFilter>>,
         limit: RateLimitConfig,
         transfer_last: Option<usize>,
+        group_sync_tasks: Option<usize>,
     ) -> Result<Self, Error> {
         if let Some(max_depth) = max_depth {
             ns.check_max_depth(max_depth)?;
@@ -560,6 +563,7 @@ impl PullParameters {
             max_depth,
             group_filter,
             transfer_last,
+            group_sync_tasks,
         })
     }
 }
-- 
2.39.2