Merge pull request #38 from cacheMon/claude/sharp-davinci-4rkawz

Upload trace logs individually, enlarge per-stream logs, add time/size rotation

Author: 1a1a11a

src/tracer/WriterManager.py

@@ -118,15 +118,19 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
             'io_uring': deque(maxlen=1000),
         }
 
-        # Dynamic thresholds (min, max)
+        # Dynamic thresholds (min, max). Raised roughly 10x from the original
+        # sizes so each rotated/compressed log file is meaningfully larger,
+        # producing fewer, larger per-stream uploads instead of many tiny ones.
+        # The min is the steady-state file size at low event rates; the max
+        # caps memory by bounding how many events a buffer holds in RAM.
         self.dynamic_limits = {
-            'vfs': (8000, 500000),
-            'block': (8000, 50000),
-            'cache': (20000, 1000000),
-            'fs_state': (8000, 20000),
-            'proc_state': (8000, 10000),  # Match new process_max_events threshold
-            'pagefault': (8000, 100000),
-            'io_uring': (8000, 200000),
+            'vfs': (80000, 800000),
+            'block': (80000, 400000),
+            'cache': (100000, 1000000),
+            'fs_state': (80000, 200000),
+            'proc_state': (80000, 100000),
+            'pagefault': (80000, 400000),
+            'io_uring': (80000, 400000),
         }
 
         # Start adaptive sizing thread
@@ -140,14 +144,15 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self.periodic_flush_thread.start()
 
 
-        # Buffer flush thresholds
-        self.cache_max_events = 20000
-        self.vfs_max_events = 8000
-        self.block_max_events = 8000
-        self.process_max_events = 8000  # Large enough to fit entire hourly snapshot
-        self.fs_snap_max_events = 8000
-        self.pagefault_max_events = 8000
-        self.io_uring_max_events = 8000
+        # Buffer flush thresholds. Raised ~10x so each rotated log file holds
+        # more events and uploads larger (kept in sync with dynamic_limits).
+        self.cache_max_events = 100000
+        self.vfs_max_events = 80000
+        self.block_max_events = 80000
+        self.process_max_events = 80000  # Large enough to fit entire hourly snapshot
+        self.fs_snap_max_events = 80000
+        self.pagefault_max_events = 80000
+        self.io_uring_max_events = 80000
 
         # Per-stream locks. Buffer flushes are triggered both from the
         # perf-callback (polling) thread via append_*_log -> flush_*_only and
@@ -173,6 +178,31 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self._fs_snap_handle = None
         self._io_uring_handle = None
 
+        # Registry of the continuous event streams that support generic
+        # rotation. Snapshots (process, fs_snap) are intentionally excluded:
+        # rotating one mid-session would split a single logical snapshot.
+        # Each entry names the attributes that hold its buffer, file handle,
+        # and current output path, plus its output subdir/prefix and log label.
+        self._streams = {
+            'vfs':       {'subdir': 'fs',        'prefix': 'fs',        'buf': 'vfs_buffer',       'handle': '_vfs_handle',       'file': 'output_vfs_file',       'log': 'VFS'},
+            'block':     {'subdir': 'ds',        'prefix': 'ds',        'buf': 'block_buffer',     'handle': '_block_handle',     'file': 'output_block_file',     'log': 'Block'},
+            'cache':     {'subdir': 'cache',     'prefix': 'cache',     'buf': 'cache_buffer',     'handle': '_cache_handle',     'file': 'output_cache_file',     'log': 'Cache'},
+            'pagefault': {'subdir': 'pagefault', 'prefix': 'pagefault', 'buf': 'pagefault_buffer', 'handle': '_pagefault_handle', 'file': 'output_pagefault_file', 'log': 'PageFault'},
+            'io_uring':  {'subdir': 'io_uring',  'prefix': 'io_uring',  'buf': 'io_uring_buffer',  'handle': '_io_uring_handle',  'file': 'output_io_uring_file',  'log': 'IO_Uring'},
+        }
+
+        # Time/size based rotation so a slow stream's log doesn't wait until
+        # shutdown to upload. A stream's current file is rotated (compressed +
+        # queued for upload) once it is older than max_file_age or larger than
+        # max_file_bytes, even if it never reaches its event-count threshold.
+        # Monotonic open-times so the age check ignores wall-clock jumps.
+        self.max_file_age = 20 * 60               # 20 minutes
+        self.max_file_bytes = 100 * 1024 * 1024   # 100 MB (uncompressed on disk)
+        self._stream_opened = {key: time.monotonic() for key in self._streams}
+        # Per-stream rotation sequence, appended to each rotated filename so two
+        # rotations in the same millisecond can't collide on the timestamp.
+        self._stream_seq = {key: 0 for key in self._streams}
+
         # Cache sampling configuration
         self.cache_sample_rate = 1  # Can be increased to reduce cache event volume
         self.cache_event_counter = 0
@@ -187,12 +217,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         # Process snapshot session tracking
         self.process_snapshot_session_active = False
 
-        # Bundle upload tracking: accumulate files and upload as a single tar
-        self._pending_bundle: list[str] = []
-        self._bundle_lock = threading.Lock()
-        self._bundle_counter = 0
-        self.bundle_size = 5
-
     def _calculate_event_rate(self, event_type: str) -> float:
         """
         Calculate the event rate for a given event type.
@@ -281,6 +305,13 @@ def _periodic_flush(self):
                 except Exception as e:
                     logger("error", f"Error in periodic flush: {e}")
 
+            # Rotate+upload any log that has grown too large or aged too long,
+            # so slow streams don't defer their upload to shutdown.
+            try:
+                self._maybe_rotate_stale_logs()
+            except Exception as e:
+                logger("error", f"Error rotating stale logs: {e}")
+
     def _reset_flush_timer(self):
         """Reset the periodic flush timer (called after manual flushes)."""
         self._last_flush_time = time.time()
@@ -489,7 +520,7 @@ def direct_write(self, output_path: str, spec_str: str):
             with open(dst, 'w') as f:
                 f.write(spec_str)
             if self.automatic_upload:
-                self._add_to_bundle(dst)
+                self.upload_manager.append_object(dst)
         except Exception as e:
             logger("error", f"Error writing device spec to {output_path}: {e}")
 
@@ -625,7 +656,7 @@ def mark_fs_snapshot_complete(self):
                     logger('info', f"Files Created: {str(self.created_files)} (filesystem snapshot with {num_parts} parts)", True)
                 for part_file in self.fs_snapshot_parts_pending_upload:
                     if os.path.exists(part_file):
-                        self._add_to_bundle(part_file)
+                        self.upload_manager.append_object(part_file)
                 self.fs_snapshot_parts_pending_upload.clear()
 
         except Exception as e:
@@ -662,99 +693,99 @@ def flush_process_state_only(self):
             self.compress_log(rotated)
 
     def flush_cache_only(self):
-        """Flush cache buffer to file."""
-        rotated = None
-        with self._stream_locks['cache']:
-            if self.cache_buffer:
-                if self._cache_handle is None:
-                    self._cache_handle = open(self.output_cache_file, 'a', buffering=8192)
-                self.current_datetime = datetime.now()
+        """Flush cache buffer to file (rotate + compress + upload)."""
+        self._rotate_stream('cache')
 
-                self._write_buffer_to_file(self.cache_buffer, self._cache_handle, "Cache")
-                self._cache_handle.close()
-                rotated = self.output_cache_file
-                self.output_cache_file = f"{self.output_dir}/cache/cache_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-                self._cache_handle = open(self.output_cache_file, 'a', buffering=8192)
-                self._reset_flush_timer()
-        if rotated is not None:
-            self.compress_log(rotated)
+    def flush_vfs_only(self):
+        """Flush VFS buffer to file (rotate + compress + upload)."""
+        self._rotate_stream('vfs')
 
+    def flush_block_only(self):
+        """Flush block buffer to file (rotate + compress + upload)."""
+        self._rotate_stream('block')
 
-    def flush_vfs_only(self):
-        """Flush VFS buffer to file."""
-        rotated = None
-        with self._stream_locks['vfs']:
-            if self.vfs_buffer:
-                if self._vfs_handle is None:
-                    self._vfs_handle = open(self.output_vfs_file, 'a', buffering=8192)
-                self.current_datetime = datetime.now()
+    def flush_pagefault_only(self):
+        """Flush pagefault buffer to file (rotate + compress + upload)."""
+        self._rotate_stream('pagefault')
 
-                self._write_buffer_to_file(self.vfs_buffer, self._vfs_handle, "VFS")
-                # Close before compressing so we never gzip/delete a file that
-                # still has an open descriptor; rotate to a fresh output file.
-                self._vfs_handle.close()
-                rotated = self.output_vfs_file
-                self.output_vfs_file = f"{self.output_dir}/fs/fs_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-                self._vfs_handle = open(self.output_vfs_file, 'a', buffering=8192)
-                self._reset_flush_timer()
-        # Compress outside the lock: gzip + disk I/O is slow and must not block
-        # the perf-callback flush or the periodic writer for this stream.
-        if rotated is not None:
-            self.compress_log(rotated)
+    def flush_io_uring_only(self):
+        """Flush io_uring buffer to file (rotate + compress + upload)."""
+        self._rotate_stream('io_uring')
 
-    def flush_block_only(self):
-        """Flush block buffer to file."""
+    def _rotate_stream(self, key: str):
+        """Rotate one continuous stream's current log and queue it for upload.
+
+        Flushes any buffered rows into the current file, closes it, swaps in a
+        fresh timestamped output file, and compresses/uploads the rotated one.
+        Works whether the rows are still buffered (event-count flush) or were
+        already written to disk by the periodic writer (time/size rotation).
+        A no-op when there is nothing on disk or buffered to rotate.
+        """
+        s = self._streams[key]
         rotated = None
-        with self._stream_locks['block']:
-            if self.block_buffer:
-                if self._block_handle is None:
-                    self._block_handle = open(self.output_block_file, 'a', buffering=8192)
+        with self._stream_locks[key]:
+            buf = getattr(self, s['buf'])
+            handle = getattr(self, s['handle'])
+            cur_file = getattr(self, s['file'])
+
+            # Land any buffered rows in the current file first.
+            if buf:
+                if handle is None:
+                    handle = open(cur_file, 'a', buffering=8192)
+                    setattr(self, s['handle'], handle)
                 self.current_datetime = datetime.now()
+                self._write_buffer_to_file(buf, handle, s['log'])
 
-                self._write_buffer_to_file(self.block_buffer, self._block_handle, "Block")
-                self._block_handle.close()
-                rotated = self.output_block_file
-                self.output_block_file = f"{self.output_dir}/ds/ds_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-                self._block_handle = open(self.output_block_file, 'a', buffering=8192)
-                self._reset_flush_timer()
-        if rotated is not None:
-            self.compress_log(rotated)
+            # Close before compressing so we never gzip/delete an open file.
+            if handle is not None:
+                handle.close()
+                setattr(self, s['handle'], None)
 
-    def flush_pagefault_only(self):
-        """Flush pagefault buffer to file."""
-        rotated = None
-        with self._stream_locks['pagefault']:
-            if self.pagefault_buffer:
-                if self._pagefault_handle is None:
-                    self._pagefault_handle = open(self.output_pagefault_file, 'a', buffering=8192)
-                self.current_datetime = datetime.now()
+            # Skip rotating an empty file (avoids zero-byte uploads); just keep
+            # appending to it.
+            if not os.path.exists(cur_file) or os.path.getsize(cur_file) == 0:
+                setattr(self, s['handle'], open(cur_file, 'a', buffering=8192))
+                return
 
-                self._write_buffer_to_file(self.pagefault_buffer, self._pagefault_handle, "PageFault")
-                self._pagefault_handle.close()
-                rotated = self.output_pagefault_file
-                self.output_pagefault_file = f"{self.output_dir}/pagefault/pagefault_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-                self._pagefault_handle = open(self.output_pagefault_file, 'a', buffering=8192)
-                self._reset_flush_timer()
+            rotated = cur_file
+            self._stream_seq[key] += 1
+            new_file = (
+                f"{self.output_dir}/{s['subdir']}/{s['prefix']}_"
+                f"{datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]}_"
+                f"{self._stream_seq[key]:04d}.csv"
+            )
+            setattr(self, s['file'], new_file)
+            setattr(self, s['handle'], open(new_file, 'a', buffering=8192))
+            self._stream_opened[key] = time.monotonic()
+            self._reset_flush_timer()
+        # Compress outside the lock: gzip + disk I/O is slow and must not block
+        # the perf-callback flush or the periodic writer for this stream.
         if rotated is not None:
             self.compress_log(rotated)
 
-    def flush_io_uring_only(self):
-        """Flush io_uring buffer to file."""
-        rotated = None
-        with self._stream_locks['io_uring']:
-            if self.io_uring_buffer:
-                if self._io_uring_handle is None:
-                    self._io_uring_handle = open(self.output_io_uring_file, 'a', buffering=8192)
-                self.current_datetime = datetime.now()
+    def _maybe_rotate_stale_logs(self, now: float | None = None):
+        """Rotate continuous logs that have grown too large or aged too long.
 
-                self._write_buffer_to_file(self.io_uring_buffer, self._io_uring_handle, "IO_Uring")
-                self._io_uring_handle.close()
-                rotated = self.output_io_uring_file
-                self.output_io_uring_file = f"{self.output_dir}/io_uring/io_uring_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-                self._io_uring_handle = open(self.output_io_uring_file, 'a', buffering=8192)
-                self._reset_flush_timer()
-        if rotated is not None:
-            self.compress_log(rotated)
+        Lets a slow stream upload mid-trace instead of waiting until shutdown:
+        any stream whose current file exceeds ``max_file_bytes`` or has been
+        open longer than ``max_file_age`` is rotated and queued for upload.
+        ``now`` is an injectable ``time.monotonic()`` reading for testing.
+        """
+        if now is None:
+            now = time.monotonic()
+        for key, s in self._streams.items():
+            cur_file = getattr(self, s['file'])
+            try:
+                if not os.path.exists(cur_file):
+                    continue
+                size = os.path.getsize(cur_file)
+            except OSError:
+                continue
+            if size <= 0:
+                continue
+            age = now - self._stream_opened.get(key, now)
+            if size >= self.max_file_bytes or age >= self.max_file_age:
+                self._rotate_stream(key)
 
     def force_flush(self):
         """Flush all buffers and compress all output files."""
@@ -797,8 +828,6 @@ def force_flush(self):
 
         self.compress_log(self.output_pagefault_file)
         self.compress_log(self.output_io_uring_file)
-        if self.automatic_upload:
-            self._flush_bundle()
         self.compress_dir(self.output_dir)
 
 
@@ -937,52 +966,6 @@ def write_io_uring():
 
         self.clear_events()
 
-    def _add_to_bundle(self, file_path: str):
-        """Queue a compressed file for bundled upload."""
-        with self._bundle_lock:
-            self._pending_bundle.append(file_path)
-            should_flush = len(self._pending_bundle) >= self.bundle_size
-        if should_flush:
-            self._flush_bundle()
-
-    def _flush_bundle(self):
-        """Pack all pending files into one tar and queue it for upload."""
-        with self._bundle_lock:
-            if not self._pending_bundle:
-                return
-            files_to_bundle = list(self._pending_bundle)
-            self._pending_bundle.clear()
-            self._bundle_counter += 1
-            counter = self._bundle_counter
-
-        bundle_dir = os.path.dirname(self.output_dir.rstrip("/\\"))
-        bundle_ts = datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]
-        bundle_path = os.path.join(bundle_dir, f"bundle_{counter:04d}_{bundle_ts}.tar")
-
-        try:
-            with tarfile.open(bundle_path, "w") as tar:
-                for f in files_to_bundle:
-                    if os.path.exists(f):
-                        tar.add(f, arcname=os.path.relpath(f, bundle_dir))
-            # Tar closed successfully — safe to delete sources now
-            for f in files_to_bundle:
-                if os.path.exists(f):
-                    try:
-                        os.remove(f)
-                    except OSError as rm_err:
-                        logger("warning", f"Failed to remove bundled file {f}: {rm_err}")
-            self.upload_manager.append_object(bundle_path)
-        except Exception as e:
-            logger("error", f"Failed to create upload bundle: {e}")
-            if os.path.exists(bundle_path):
-                try:
-                    os.remove(bundle_path)
-                except OSError:
-                    pass
-            for f in files_to_bundle:
-                if os.path.exists(f):
-                    self.upload_manager.append_object(f)
-
     def compress_log(self, input_file: str):
         """
         Compress a log file with gzip and optionally upload.
@@ -1005,7 +988,9 @@ def compress_log(self, input_file: str):
             if self.automatic_upload:
                 self.created_files += 1
                 logger('info', f"Files Created: {str(self.created_files)}", True)
-                self._add_to_bundle(dst)
+                # Upload each compressed log individually, preserving its
+                # subdirectory (fs, ds, cache, process, ...) on the backend.
+                self.upload_manager.append_object(dst)
             os.remove(src)
         except Exception as e:
             logger("error", f"Failed compressing log {input_file}: {e}")