[pbs-devel] [PATCH proxmox-backup] proxmox-rrd: define and implement new ZSTD compressed format

[pbs-devel] [PATCH proxmox-backup] proxmox-rrd: define and implement new ZSTD compressed format

[Dietmar Maurer]

This further reduce the amount of data written.
---
 proxmox-rrd/Cargo.toml   |  1 +
 proxmox-rrd/src/cache.rs | 38 ++++++++++++++++---------------
 proxmox-rrd/src/rrd.rs   | 49 ++++++++++++++++++++++++++++++++++++----
 3 files changed, 65 insertions(+), 23 deletions(-)

diff --git a/proxmox-rrd/Cargo.toml b/proxmox-rrd/Cargo.toml
index 1c75b2e1..b095bd6f 100644
--- a/proxmox-rrd/Cargo.toml
+++ b/proxmox-rrd/Cargo.toml
@@ -10,6 +10,7 @@ anyhow = "1.0"
 bitflags = "1.2.1"
 log = "0.4"
 nix = "0.19.1"
+zstd = { version = "0.6", features = [ "bindgen" ] }
 
 proxmox = { version = "0.13.5", features = ["api-macro"] }
 
diff --git a/proxmox-rrd/src/cache.rs b/proxmox-rrd/src/cache.rs
index 57cbb394..bf075161 100644
--- a/proxmox-rrd/src/cache.rs
+++ b/proxmox-rrd/src/cache.rs
@@ -125,6 +125,21 @@ impl RRDCache {
         self.apply_journal_locked(&mut state)
     }
 
+    fn load_rrd(path: &Path, dst: DST) -> RRD {
+        match RRD::load(path) {
+            Ok(mut rrd) => {
+                rrd.set_compression(true);
+                rrd
+            }
+            Err(err) => {
+                if err.kind() != std::io::ErrorKind::NotFound {
+                    log::warn!("overwriting RRD file {:?}, because of load error: {}", path, err);
+                }
+                RRD::new(dst, true)
+            },
+        }
+    }
+
     fn apply_journal_locked(&self, state: &mut RRDCacheState) -> Result<(), Error> {
 
         log::info!("applying rrd journal");
@@ -173,15 +188,8 @@ impl RRDCache {
                 path.push(&entry.rel_path);
                 create_path(path.parent().unwrap(), Some(self.dir_options.clone()), Some(self.dir_options.clone()))?;
 
-                let mut rrd = match RRD::load(&path) {
-                    Ok(rrd) => rrd,
-                    Err(err) => {
-                        if err.kind() != std::io::ErrorKind::NotFound {
-                            log::warn!("overwriting RRD file {:?}, because of load error: {}", path, err);
-                        }
-                        RRD::new(entry.dst)
-                    },
-                };
+                let mut rrd = Self::load_rrd(&path, entry.dst);
+
                 if entry.time > get_last_update(&entry.rel_path, &rrd) {
                     rrd.update(entry.time, entry.value);
                 }
@@ -240,16 +248,10 @@ impl RRDCache {
             let mut path = self.basedir.clone();
             path.push(rel_path);
             create_path(path.parent().unwrap(), Some(self.dir_options.clone()), Some(self.dir_options.clone()))?;
-            let mut rrd = match RRD::load(&path) {
-                Ok(rrd) => rrd,
-                Err(err) => {
-                    if err.kind() != std::io::ErrorKind::NotFound {
-                        log::warn!("overwriting RRD file {:?}, because of load error: {}", path, err);
-                    }
-                    RRD::new(dst)
-                },
-            };
+
+            let mut rrd = Self::load_rrd(&path, dst);
             rrd.update(now, value);
+
             state.rrd_map.insert(rel_path.into(), rrd);
         }
 
diff --git a/proxmox-rrd/src/rrd.rs b/proxmox-rrd/src/rrd.rs
index 026498ed..f2558646 100644
--- a/proxmox-rrd/src/rrd.rs
+++ b/proxmox-rrd/src/rrd.rs
@@ -17,6 +17,10 @@ pub const RRD_DATA_ENTRIES: usize = 70;
 // openssl::sha::sha256(b"Proxmox Round Robin Database file v1.0")[0..8];
 pub const PROXMOX_RRD_MAGIC_1_0: [u8; 8] =  [206, 46, 26, 212, 172, 158, 5, 186];
 
+/// Proxmox RRD file magic number for zstd compressed RRD files
+//openssl::sha::sha256(b"Proxmox Round Robin Database file v1.0 (zstd compressed)")[0..8];
+pub const PROXMOX_ZSTD_RRD_MAGIC_1_0: [u8; 8] =  [84, 61, 44, 56, 140, 132, 5, 11];
+
 use crate::DST;
 
 bitflags!{
@@ -198,14 +202,14 @@ pub struct RRD {
 impl RRD {
 
     /// Create a new empty instance
-    pub fn new(dst: DST) -> Self {
+    pub fn new(dst: DST, compress: bool) -> Self {
         let flags = match dst {
             DST::Gauge => RRAFlags::DST_GAUGE,
             DST::Derive => RRAFlags::DST_DERIVE,
         };
 
         Self {
-            magic: PROXMOX_RRD_MAGIC_1_0,
+            magic: if compress { PROXMOX_ZSTD_RRD_MAGIC_1_0 } else { PROXMOX_RRD_MAGIC_1_0 },
             hour_avg: RRA::new(
                 flags | RRAFlags::CF_AVERAGE,
                 RRDTimeFrameResolution::Hour as u64,
@@ -301,8 +305,28 @@ impl RRD {
     }
 
     /// Create instance from raw data, testing data len and magic number
-    pub fn from_raw(mut raw: &[u8]) -> Result<Self, std::io::Error> {
+    pub fn from_raw(raw: &[u8]) -> Result<Self, std::io::Error> {
+
+        if raw.len() < 8 {
+            let msg = format!("not an rrd file - file is too small ({})", raw.len());
+             return Err(std::io::Error::new(std::io::ErrorKind::Other, msg));
+        }
+
+        if raw[0..8] == PROXMOX_ZSTD_RRD_MAGIC_1_0 {
+            let mut data = Vec::new();
+            data.extend(&PROXMOX_ZSTD_RRD_MAGIC_1_0);
+            data.extend(zstd::block::decompress(&raw[8..], std::mem::size_of::<RRD>())?);
+            Self::from_uncompressed_raw(&data)
+        } else {
+            Self::from_uncompressed_raw(raw)
+        }
+
+    }
+
+    fn from_uncompressed_raw(mut raw: &[u8]) -> Result<Self, std::io::Error> {
+
         let expected_len = std::mem::size_of::<RRD>();
+
         if raw.len() != expected_len {
             let msg = format!("wrong data size ({} != {})", raw.len(), expected_len);
             return Err(std::io::Error::new(std::io::ErrorKind::Other, msg));
@@ -314,7 +338,7 @@ impl RRD {
             raw.read_exact(rrd_slice)?;
         }
 
-        if rrd.magic != PROXMOX_RRD_MAGIC_1_0 {
+        if !(rrd.magic == PROXMOX_RRD_MAGIC_1_0 || rrd.magic == PROXMOX_ZSTD_RRD_MAGIC_1_0) {
             let msg = "wrong magic number".to_string();
             return Err(std::io::Error::new(std::io::ErrorKind::Other, msg));
         }
@@ -322,6 +346,13 @@ impl RRD {
         Ok(rrd)
     }
 
+    /// Set compression flag
+    ///
+    /// This just changes the magic number. Actual compression is done in [Self::save].
+    pub fn set_compression(&mut self, compress: bool) {
+        self.magic = if compress { PROXMOX_ZSTD_RRD_MAGIC_1_0 } else { PROXMOX_RRD_MAGIC_1_0 };
+    }
+
     /// Load data from a file
     pub fn load(path: &Path) -> Result<Self, std::io::Error> {
         let raw = std::fs::read(path)?;
@@ -333,7 +364,15 @@ impl RRD {
         let rrd_slice = unsafe {
             std::slice::from_raw_parts(self as *const _ as *const u8, std::mem::size_of::<RRD>())
         };
-        replace_file(filename, rrd_slice, options)
+
+        if rrd_slice[0..8] == PROXMOX_ZSTD_RRD_MAGIC_1_0  {
+            let mut data = Vec::new();
+            data.extend(&PROXMOX_ZSTD_RRD_MAGIC_1_0);
+            data.extend(zstd::block::compress(&rrd_slice[8..], 1)?);
+            replace_file(filename, &data, options)
+        } else {
+            replace_file(filename, rrd_slice, options)
+        }
     }
 
     pub fn last_update(&self) -> f64 {
-- 
2.30.2