[pbs-devel] [RFC proxmox-backup 2/2] client: add --compresion flag to create_backup

[Maximiliano Sandoval R <m.sandoval@proxmox.com>]

This partially addresses
https://bugzilla.proxmox.com/show_bug.cgi?id=4900.

We take from borg the idea of passing compression levels in the form
`--compression=zstd,3`. This future proof in the sense that it can
accommodate future algorithms which might have more than one parameter,
it also works with negative arguments.

Signed-off-by: Maximiliano Sandoval R <m.sandoval@proxmox.com>
---
 pbs-datastore/src/data_blob.rs    | 40 +++++++++++++++++++++++++++++++
 proxmox-backup-client/src/main.rs | 29 ++++++++++++++++------
 2 files changed, 62 insertions(+), 7 deletions(-)

diff --git a/pbs-datastore/src/data_blob.rs b/pbs-datastore/src/data_blob.rs
index 765f5a98..2ea986af 100644
--- a/pbs-datastore/src/data_blob.rs
+++ b/pbs-datastore/src/data_blob.rs
@@ -4,6 +4,7 @@ use anyhow::{bail, Error};
 use openssl::symm::{decrypt_aead, Mode};
 
 use proxmox_io::{ReadExt, WriteExt};
+use proxmox_schema::{Schema, StringSchema};
 
 use pbs_api_types::CryptMode;
 use pbs_tools::crypt_config::CryptConfig;
@@ -24,6 +25,45 @@ impl Default for Compression {
     }
 }
 
+impl std::str::FromStr for Compression {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if s == "none" {
+            return Ok(Self::None);
+        }
+
+        if let Some(("zstd", level)) = s.split_once(',') {
+            let level = level.parse::<i32>()?;
+
+            if zstd::compression_level_range().contains(&level) {
+                Ok(Self::Zstd(level))
+            } else {
+                bail!("Invalid ZSTD compression level: {level}");
+            }
+        } else {
+            bail!("Unknown compression: {s}");
+        }
+    }
+}
+
+impl std::fmt::Display for Compression {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Compression::None => write!(f, "none"),
+            Compression::Zstd(level) => write!(f, "zstd,{level}"),
+        }
+    }
+}
+
+proxmox_serde::forward_deserialize_to_from_str!(Compression);
+proxmox_serde::forward_serialize_to_display!(Compression);
+
+impl proxmox_schema::ApiType for Compression {
+    const API_SCHEMA: Schema =
+        StringSchema::new("Compression (e.g. 'none', 'zstd,<level>')").schema();
+}
+
 /// Encoded data chunk with digest and positional information
 pub struct ChunkInfo {
     pub chunk: DataBlob,
diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs
index e5d67c7d..3aaf127e 100644
--- a/proxmox-backup-client/src/main.rs
+++ b/proxmox-backup-client/src/main.rs
@@ -526,6 +526,7 @@ struct CatalogUploadResult {
 fn spawn_catalog_upload(
     client: Arc<BackupWriter>,
     encrypt: bool,
+    compression: Compression,
 ) -> Result<CatalogUploadResult, Error> {
     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
     let catalog_stream = proxmox_async::blocking::StdChannelStream(catalog_rx);
@@ -540,7 +541,7 @@ fn spawn_catalog_upload(
 
     let upload_options = UploadOptions {
         encrypt,
-        compression: Compression::default(),
+        compression,
         ..UploadOptions::default()
     };
 
@@ -585,6 +586,10 @@ fn spawn_catalog_upload(
                    description: "Path to file.",
                }
            },
+           "compression": {
+               optional: true,
+               type: Compression,
+           },
            "all-file-systems": {
                type: Boolean,
                description: "Include all mounted subdirectories.",
@@ -720,6 +725,16 @@ async fn create_backup(
     let empty = Vec::new();
     let exclude_args = param["exclude"].as_array().unwrap_or(&empty);
 
+    let compression = match param["compression"].as_str() {
+        Some(c) => {
+            match Compression::from_str(c) {
+                Err(err) => bail!("Invalid compression '{c}': {err}"),
+                Ok(comp) => comp,
+            }
+        },
+        None => Compression::default(),
+    };
+
     let mut pattern_list = Vec::with_capacity(exclude_args.len());
     for entry in exclude_args {
         let entry = entry
@@ -948,7 +963,7 @@ async fn create_backup(
             // no dry-run
             (BackupSpecificationType::CONFIG, false) => {
                 let upload_options = UploadOptions {
-                    compression: Compression::default(),
+                    compression,
                     encrypt: crypto.mode == CryptMode::Encrypt,
                     ..UploadOptions::default()
                 };
@@ -962,7 +977,7 @@ async fn create_backup(
             (BackupSpecificationType::LOGFILE, false) => {
                 // fixme: remove - not needed anymore ?
                 let upload_options = UploadOptions {
-                    compression: Compression::default(),
+                    compression,
                     encrypt: crypto.mode == CryptMode::Encrypt,
                     ..UploadOptions::default()
                 };
@@ -977,7 +992,7 @@ async fn create_backup(
                 // start catalog upload on first use
                 if catalog.is_none() {
                     let catalog_upload_res =
-                        spawn_catalog_upload(client.clone(), crypto.mode == CryptMode::Encrypt)?;
+                        spawn_catalog_upload(client.clone(), crypto.mode == CryptMode::Encrypt, compression)?;
                     catalog = Some(catalog_upload_res.catalog_writer);
                     catalog_result_rx = Some(catalog_upload_res.result);
                 }
@@ -998,7 +1013,7 @@ async fn create_backup(
 
                 let upload_options = UploadOptions {
                     previous_manifest: previous_manifest.clone(),
-                    compression: Compression::default(),
+                    compression,
                     encrypt: crypto.mode == CryptMode::Encrypt,
                     ..UploadOptions::default()
                 };
@@ -1022,7 +1037,7 @@ async fn create_backup(
                 let upload_options = UploadOptions {
                     previous_manifest: previous_manifest.clone(),
                     fixed_size: Some(size),
-                    compression: Compression::default(),
+                    compression,
                     encrypt: crypto.mode == CryptMode::Encrypt,
                 };
 
@@ -1077,7 +1092,7 @@ async fn create_backup(
     log::debug!("Upload index.json to '{}'", repo);
 
     let options = UploadOptions {
-        compression: Compression::default(),
+        compression,
         encrypt: false,
         ..UploadOptions::default()
     };
-- 
2.39.2