Switch trace compression from gzip to Zstandard (.zst)

Replace gzip with Zstandard for all trace output:
- compress_log now writes .zst via a shared compress_file_zstd helper
  (stream compression); per-stream logs upload as .csv.zst.
- compress_dir writes the final session archive as .tar.zst (tar streamed
  through a zstd stream_writer).
- Multi-part filesystem snapshots compress to .csv.zst.
- utils gains compress_file_zstd / create_tar_zst and a ZSTD_LEVEL constant;
  zstandard is imported lazily so importing the modules never requires the
  dependency (matches the requests pattern). Drop the now-unused gzip imports.

- Add requirements.txt (requests, zstandard) and install zstandard in the
  unit-tests workflow so the compression round-trip is actually exercised.
- Tests assert .zst output and round-trip via zstandard, skipped when the
  optional dependency is absent.

Author: claude

.github/workflows/unit-tests.yml

@@ -27,5 +27,8 @@ jobs:
         with:
           python-version: '3.11'
 
+      - name: Install zstandard (used by the compression tests)
+        run: python3 -m pip install --upgrade "zstandard>=0.21"
+
       - name: Run unit tests
         run: python3 -m unittest discover -s tests -v

requirements.txt

@@ -0,0 +1,4 @@
+# Runtime dependencies for the io-tracer userspace tooling.
+# Note: bcc (python3-bpfcc) is installed via the system package manager, not pip.
+requests>=2.25
+zstandard>=0.21

src/tracer/ObjectStorageManager.py

@@ -13,7 +13,7 @@
 Example:
     manager = ObjectStorageManager(version="vRelease")
     manager.test_connection()  # Check if server is reachable
-    manager.put_object("/path/to/trace.tar.gz")  # Upload a file
+    manager.put_object("/path/to/trace.tar.zst")  # Upload a file
 """
 
 import mimetypes

src/tracer/WriterManager.py

@@ -5,7 +5,7 @@
 - Creating output directory structure
 - Buffering trace events for different subsystems
 - Writing events to CSV files
-- Compressing output files with gzip
+- Compressing output files with Zstandard (.zst)
 - Optionally uploading files to cloud storage
 
 The manager uses adaptive buffering to handle high event rates and
@@ -29,10 +29,12 @@
 import tarfile
 
 from .ObjectStorageManager import ObjectStorageManager
-from ..utility.utils import logger, create_tar_gz, capture_machine_id, compress_log
+from ..utility.utils import (
+    logger, create_tar_zst, capture_machine_id, compress_log,
+    compress_file_zstd, ZSTD_LEVEL,
+)
 import threading
 from collections import deque
-import gzip
 import shutil
 import time
 
@@ -501,7 +503,7 @@ def flush_fssnap_only(self):
         Flush filesystem snapshot buffer to a multi-part file.
 
         Writes buffer to filesystem_snapshot_part####_TIMESTAMP_DEVICEID.csv,
-        compresses it with gzip, and increments the part counter.
+        compresses it with Zstandard, and increments the part counter.
         """
         with self._stream_locks['fs_snap']:
             if not self.fs_snap_buffer:
@@ -533,15 +535,13 @@ def flush_fssnap_only(self):
             self._fs_snap_handle.close()
             self._fs_snap_handle = None
 
-            # Compress with gzip
+            # Compress with Zstandard
             if os.path.exists(part_filepath):
                 # Don't log or count each part - we'll log when snapshot is complete
-                with open(part_filepath, "rb") as f_in:
-                    with gzip.open(part_filepath + ".gz", "wb") as f_out:
-                        shutil.copyfileobj(f_in, f_out)
+                compress_file_zstd(part_filepath, part_filepath + ".zst")
 
                 os.remove(part_filepath)
-                compressed_file = part_filepath + ".gz"
+                compressed_file = part_filepath + ".zst"
 
                 # Store for later upload (after snapshot completion and final part rename)
                 if self.automatic_upload:
@@ -593,15 +593,15 @@ def mark_fs_snapshot_complete(self):
         old_filename = (
             f"filesystem_snapshot_part{last_part_str}_"
             f"{self.fs_snapshot_timestamp}_"
-            f"{self.fs_snapshot_device_id}.csv.gz"
+            f"{self.fs_snapshot_device_id}.csv.zst"
         )
         old_filepath = f"{self.output_dir}/filesystem_snapshot/{old_filename}"
 
         # Construct new filename with completion marker
         new_filename = (
             f"filesystem_snapshot_part{last_part_str}_"
             f"{self.fs_snapshot_timestamp}_"
-            f"{self.fs_snapshot_device_id}_complete_parts{total_parts}.csv.gz"
+            f"{self.fs_snapshot_device_id}_complete_parts{total_parts}.csv.zst"
         )
         new_filepath = f"{self.output_dir}/filesystem_snapshot/{new_filename}"
 
@@ -922,22 +922,20 @@ def write_pagefault():
 
     def compress_log(self, input_file: str):
         """
-        Compress a log file with gzip and optionally upload.
-        
+        Compress a log file with Zstandard and optionally upload.
+
         Args:
             input_file: Path to the file to compress
         """
         try:
             src = input_file
-            dst = input_file + ".gz"
-            
+            dst = input_file + ".zst"
+
             # Check if file exists (may already be compressed for multi-part files)
             if not os.path.exists(src):
                 return
-            
-            with open(src, "rb") as f_in:
-                with gzip.open(dst, "wb") as f_out:
-                    shutil.copyfileobj(f_in, f_out) # type: ignore
+
+            compress_file_zstd(src, dst)
 
             if self.automatic_upload:
                 self.created_files += 1
@@ -951,17 +949,21 @@ def compress_log(self, input_file: str):
 
     def compress_dir(self, input_dir: str):
         """
-        Compress a directory to tar.gz and optionally upload.
-        
+        Compress a directory to tar.zst and optionally upload.
+
         Args:
             input_dir: Path to the directory to compress
         """
         try:
+            import zstandard
             src = input_dir
-            dst = input_dir.rstrip("/").rstrip("\\") + ".tar.gz"
+            dst = input_dir.rstrip("/").rstrip("\\") + ".tar.zst"
 
-            with tarfile.open(dst, "w:gz") as tar:
-                tar.add(src, arcname=os.path.basename(src))
+            cctx = zstandard.ZstdCompressor(level=ZSTD_LEVEL)
+            with open(dst, "wb") as f_out:
+                with cctx.stream_writer(f_out) as compressor:
+                    with tarfile.open(mode="w|", fileobj=compressor) as tar:
+                        tar.add(src, arcname=os.path.basename(src))
 
             if self.automatic_upload:
                 self.created_files += 1

src/tracer/snappers/FilesystemSnapper.py

@@ -20,7 +20,6 @@
 from ..WriterManager import WriteManager
 from pathlib import Path
 from datetime import datetime
-import gzip
 import shutil
 import os
 import time

src/utility/utils.py

@@ -28,8 +28,6 @@
 import time
 import datetime
 import tarfile
-import gzip
-import shutil
 import hashlib
 import socket
 import struct
@@ -194,34 +192,58 @@ def logger(error_scale: str, string: str, timestamp: bool = False):
         logo += f" [{formatted_time}]" 
     print(logo + " " + string)
 
-def create_tar_gz(output_filename: str, files_to_archive: list[str]):
+# Zstandard compression level. 3 is the library default — a good
+# speed/ratio tradeoff for streaming large trace logs.
+ZSTD_LEVEL = 3
+
+
+def compress_file_zstd(src: str, dst: str, level: int = ZSTD_LEVEL):
     """
-    Create a gzipped tar archive from a list of files.
-    
+    Stream-compress a file to Zstandard.
+
     Args:
-        output_filename: Name of the output .tar.gz file
+        src: Path to the source file
+        dst: Path to write the compressed (.zst) output
+        level: Zstandard compression level
+
+    zstandard is imported lazily so environments that never compress (and the
+    pure-Python unit tests) don't require the dependency at import time.
+    """
+    import zstandard
+    cctx = zstandard.ZstdCompressor(level=level)
+    with open(src, "rb") as f_in, open(dst, "wb") as f_out:
+        cctx.copy_stream(f_in, f_out)
+
+
+def create_tar_zst(output_filename: str, files_to_archive: list[str], level: int = ZSTD_LEVEL):
+    """
+    Create a Zstandard-compressed tar archive from a list of files.
+
+    Args:
+        output_filename: Name of the output .tar.zst file
         files_to_archive: List of file paths to include
+        level: Zstandard compression level
     """
-    with tarfile.open(output_filename, "w:gz") as tar:
-        for file_path in files_to_archive:
-            tar.add(file_path, arcname=os.path.basename(file_path))
-    logger("info", f"Created tar.gz archive: {output_filename}")
+    import zstandard
+    cctx = zstandard.ZstdCompressor(level=level)
+    with open(output_filename, "wb") as f_out:
+        with cctx.stream_writer(f_out) as compressor:
+            with tarfile.open(mode="w|", fileobj=compressor) as tar:
+                for file_path in files_to_archive:
+                    tar.add(file_path, arcname=os.path.basename(file_path))
+    logger("info", f"Created tar.zst archive: {output_filename}")
+
 
 def compress_log(input_file: str):
     """
-    Compress a log file using gzip.
-    
+    Compress a log file using Zstandard.
+
     Args:
         input_file: Path to the file to compress
-        
-    Creates input_file.gz and removes the original.
-    """
-    src = input_file
-    dst = input_file + ".gz"
-    with open(src, "rb") as f_in:
-        with gzip.open(dst, "wb") as f_out:
-            shutil.copyfileobj(f_in, f_out) # type: ignore
 
+    Creates input_file.zst and removes the original.
+    """
+    compress_file_zstd(input_file, input_file + ".zst")
     os.remove(input_file)
 
 def capture_machine_id() -> str:

tests/test_writer_upload.py

@@ -11,7 +11,6 @@
     pytest tests/
 """
 
-import gzip
 import os
 import sys
 import tempfile
@@ -29,6 +28,22 @@
     except ModuleNotFoundError:
         sys.modules["requests"] = types.ModuleType("requests")
 
+# Compression uses Zstandard. Round-trip assertions are skipped when the
+# optional dependency isn't installed (CI installs it explicitly).
+try:
+    import zstandard
+    HAS_ZSTD = True
+except ModuleNotFoundError:
+    HAS_ZSTD = False
+
+
+def _zstd_read_text(path):
+    """Decompress a .zst file to text for round-trip assertions."""
+    dctx = zstandard.ZstdDecompressor()
+    with open(path, "rb") as f:
+        return dctx.stream_reader(f).read().decode()
+
+
 from src.tracer.WriterManager import WriteManager
 
 
@@ -78,22 +93,23 @@ def _make_log(self, subdir, name, text):
             f.write(text)
         return path
 
-    def test_compress_log_uploads_individual_gz(self):
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
+    def test_compress_log_uploads_individual_zst(self):
         src = self._make_log("fs", "fs_x.csv", "a,b,c\n1,2,3\n")
         self.wm.compress_log(src)
 
         # Exactly one upload, the compressed file — no tar bundle.
         self.assertEqual(len(self.upload.uploaded), 1)
         uploaded = self.upload.uploaded[0]
-        self.assertTrue(uploaded.endswith(".csv.gz"))
+        self.assertTrue(uploaded.endswith(".csv.zst"))
         self.assertFalse(uploaded.endswith(".tar"))
         self.assertTrue(os.path.exists(uploaded))
         # Source .csv is removed once compressed.
         self.assertFalse(os.path.exists(src))
-        # Content round-trips through gzip.
-        with gzip.open(uploaded, "rt") as f:
-            self.assertEqual(f.read(), "a,b,c\n1,2,3\n")
+        # Content round-trips through Zstandard.
+        self.assertEqual(_zstd_read_text(uploaded), "a,b,c\n1,2,3\n")
 
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
     def test_upload_preserves_subdirectory(self):
         # The backend file_type is derived from the parent directory, so each
         # stream must stay under its own subdir (fs, ds, cache, ...).
@@ -105,6 +121,7 @@ def test_upload_preserves_subdirectory(self):
         parents = {os.path.basename(os.path.dirname(p)) for p in self.upload.uploaded}
         self.assertEqual(parents, {"fs", "ds", "cache", "process"})
 
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
     def test_no_upload_when_automatic_disabled(self):
         self.wm.automatic_upload = False
         src = self._make_log("fs", "fs_x.csv", "row\n")
@@ -149,6 +166,7 @@ def _write_current(self, file_attr, text):
             f.write(text)
         return path
 
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
     def test_size_triggers_rotation(self):
         self.wm.max_file_bytes = 50
         self.wm.max_file_age = 10**9  # disable age trigger
@@ -158,12 +176,13 @@ def test_size_triggers_rotation(self):
 
         self.assertEqual(len(self.upload.uploaded), 1)
         uploaded = self.upload.uploaded[0]
-        self.assertTrue(uploaded.endswith(".csv.gz"))
+        self.assertTrue(uploaded.endswith(".csv.zst"))
         self.assertEqual(os.path.basename(os.path.dirname(uploaded)), "fs")
         # Rotated to a fresh file; the old .csv is gone (compressed away).
         self.assertNotEqual(self.wm.output_vfs_file, path)
         self.assertFalse(os.path.exists(path))
 
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
     def test_age_triggers_rotation(self):
         self.wm.max_file_bytes = 10**12  # disable size trigger
         self._write_current("output_block_file", "row\n")
@@ -188,15 +207,15 @@ def test_missing_or_empty_file_is_skipped(self):
         self.wm._maybe_rotate_stale_logs()
         self.assertEqual(self.upload.uploaded, [])
 
+    @unittest.skipUnless(HAS_ZSTD, "zstandard not installed")
     def test_rotate_flushes_buffered_rows(self):
         self.wm.vfs_buffer.append("a,b,c")
         self.wm.vfs_buffer.append("d,e,f")
 
         self.wm._rotate_stream("vfs")
 
         self.assertEqual(len(self.upload.uploaded), 1)
-        with gzip.open(self.upload.uploaded[0], "rt") as f:
-            content = f.read()
+        content = _zstd_read_text(self.upload.uploaded[0])
         self.assertIn("a,b,c", content)
         self.assertIn("d,e,f", content)
         self.assertEqual(len(self.wm.vfs_buffer), 0)