core: services: commander: Delete gz logs first

[patrickelectric]

Streamed unlink of every rotated .gz is too slow.
Fix #3588

[github-actions]

Automated PR Review

0. Summary

• Verdict: MINOR SUGGESTIONS ✏️

Adds a fast-path bulk delete of every *.gz under LOG_FOLDER_PATH via find ... -delete at the top of the remove_log_services_stream handler, before the existing per-file streaming deletion runs. Addresses #3588 where streamed unlinks of many rotated .gz files were too slow. The subprocess is offloaded to a thread and LOG_FOLDER_PATH is a server-side env-var (not caller-supplied), so there is no shell-injection surface.

1. Correctness & Implementation Bugs

• 1.1 [minor] core/services/commander/main.py:216 — no timeout= on subprocess.run. If find hangs (e.g. stuck NFS/overlayfs, unresponsive block device on the flash), the thread from asyncio.to_thread will block forever, the HTTP request will never move on to deletion_stream_response, and the client will get no output at all (streaming hasn't started yet). Adding a bounded timeout (e.g. a few minutes) and logging on subprocess.TimeoutExpired would keep the endpoint responsive.
• 1.2 [minor] core/services/commander/main.py:216 — find will exit non-zero if LOG_FOLDER_PATH doesn't exist (fresh installs, unmounted volume). With check=False and no post-inspection, the failure is silent. Consider either capturing stderr and logging it, or handling FileNotFoundError/non-zero returncode explicitly so operators can tell why the fast path did nothing.

5. UI / UX

• 5.1 [minor] The endpoint's contract is "stream real-time updates about each file being deleted" (docstring at main.py:214). With this change, all .gz files are deleted before any streaming output is emitted, so callers now see a long unexplained silent gap followed by progress only for non-.gz files. If the frontend uses the stream to render a progress bar, that bar will jump instead of ramping. Worth either emitting a synthetic progress event before/after the bulk step, or at minimum noting the behavior change in the docstring.

6. Code Quality & Style

• 6.1 [nit] core/services/commander/main.py:217 — check=False is already the default for subprocess.run; passing it explicitly is redundant. Fine to keep for readability, but if you want to be consistent with other subprocess calls in this file (e.g. main.py:313 uses check=True), signalling the intent is a wash.

8. Documentation

• 8.1 [nit] core/services/commander/main.py:216 — logger.info("Delete all gzs since they are good to go") — the "good to go" phrasing doesn't explain why gzs are safe to bulk-delete without progress reporting. A clearer message like "Bulk-deleting rotated .gz logs before streamed deletion (fast path for #3588)" would age better in journalctl.

Generated by PR Review Bot. This is advisory, a human reviewer must still approve.

[joaoantoniocardoso]

Since #4123, the current speed is around 1 GB/s; did you measure what we are gaining here?

Looking at the code, it seems the wrong abstraction: we have a stream, and this patch moves the heavy work outside of it, removing feedback.

[patrickelectric]

Just to be sure how did you test ? how many files and what are the size of the files ?

[joaoantoniocardoso]

I previously measured on a single system that had a lot of non-gzipped extension logs from my debug builds, so it was a bit above 1GB/s because of that skew.

I now replicated a per-service averaged distribution from 6 system logs from different clients, then scaled its gz distribution to fill ~16GB.

I remeasured it, and deletion is at 0.35GB/s on 1.4-dev. Master barely has logs, though.

The averaged distributions:

service,n_files,n_log,n_gz,total_bytes,file_median,file_p90
linux2rest,67,1,66,73778733,662006,1409084
mavlink-camera-manager,67,1,66,5940545,25255,263037
beacon,38,1,37,2761092,26908,321340
wifi-manager,28,1,27,2590191,32855,204937
bag-of-holding,29,1,28,1711641,14234,157940
version-chooser,16,1,15,1508972,12786,156380
ping,46,1,45,1241154,27820,148025
cable-guy,35,1,34,1046160,4752,108954
helper,27,1,26,691286,9527,33219
ardupilot-manager,30,1,29,431237,5295,88333
major_tom,6,1,5,112401,5922,53842
blueos_startup_update,24,1,23,109652,3435,4160
commander,66,1,65,92852,1166,3184
kraken,28,1,27,86463,1663,7430
bridget,72,1,71,63833,709,804
nmea-injector,72,1,71,59706,714,911
log-zipper,18,1,17,55071,1202,2801
telemetry,6,1,5,28518,1271,5646
bootstrap,32,1,31,23087,670,745
pardal,72,1,71,23044,330,486

[patrickelectric]

@joaoantoniocardoso yeah, this deletes 32GB in less than 1 second.

[joaoantoniocardoso]

@joaoantoniocardoso yeah, this deletes 32GB in less than 1 second.

Awesome! Can that step run inside the stream? Since gz won't ever be open, we may just skip the lsof forgas in the stream task?

[patrickelectric]

@joaoantoniocardoso yeah, this deletes 32GB in less than 1 second.

Awesome! Can that step run inside the stream? Since gz won't ever be open, we may just skip the lsof forgas in the stream task?

the delete stream is generic, the code is specific for the logs.

[joaoantoniocardoso]

@joaoantoniocardoso yeah, this deletes 32GB in less than 1 second.

Awesome! Can that step run inside the stream? Since gz won't ever be open, we may just skip the lsof forgas in the stream task?

the delete stream is generic, the code is specific for the logs.

Yes, and we should keep it generic. Perhaps:

diff --git a/core/services/commander/main.py b/core/services/commander/main.py
index 0c157992c..636696849 100755
--- a/core/services/commander/main.py
+++ b/core/services/commander/main.py
@@ -12,7 +12,12 @@ from typing import Any, AsyncGenerator, Dict
 import appdirs
 from commonwealth.utils.apis import GenericErrorHandlingRoute
 from commonwealth.utils.commands import run_command
-from commonwealth.utils.general import delete_everything, delete_everything_stream
+from commonwealth.utils.general import (
+    bulk_find_delete_stream,
+    delete_everything,
+    delete_everything_stream,
+    open_files_under,
+)
 from commonwealth.utils.logs import InterceptHandler, init_logger
 from commonwealth.utils.sentry_config import init_sentry_async
 from commonwealth.utils.streaming import streamer
@@ -56,13 +61,13 @@ def check_what_i_am_doing(i_know_what_i_am_doing: bool = False) -> None:
         )
 
 
-def deletion_stream_response(path: Path) -> StreamingResponse:
+def deletion_stream_response(stream: AsyncGenerator[dict[str, Any], None]) -> StreamingResponse:
     async def generate() -> AsyncGenerator[str, None]:
         try:
-            async for info in delete_everything_stream(path):
+            async for info in stream:
                 yield json.dumps(info)
         except Exception as error:
-            logger.error(f"Error during deletion stream of {path}: {error}")
+            logger.error(f"Error during deletion stream: {error}")
             raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(error)) from error
 
     return StreamingResponse(
@@ -77,6 +82,14 @@ def deletion_stream_response(path: Path) -> StreamingResponse:
     )
 
 
+async def service_log_deletion_stream(root: Path) -> AsyncGenerator[dict[str, Any], None]:
+    open_files = await asyncio.to_thread(open_files_under, root)
+    async for info in bulk_find_delete_stream(root, "*.gz"):
+        yield info
+    async for info in delete_everything_stream(root, open_files=open_files):
+        yield info
+
+
 @app.post("/command/host", status_code=status.HTTP_200_OK)
 @version(1, 0)
 async def command_host(command: str, i_know_what_i_am_doing: bool = False) -> Any:
@@ -213,7 +226,7 @@ async def remove_log_services(i_know_what_i_am_doing: bool = False) -> Any:
 async def remove_log_services_stream(i_know_what_i_am_doing: bool = False) -> StreamingResponse:
     """Stream the deletion of log files, providing real-time updates about each file being deleted."""
     check_what_i_am_doing(i_know_what_i_am_doing)
-    return deletion_stream_response(Path(LOG_FOLDER_PATH))
+    return deletion_stream_response(service_log_deletion_stream(Path(LOG_FOLDER_PATH)))
 
 
 @app.post("/services/remove_mavlink_log", status_code=status.HTTP_200_OK)
@@ -228,7 +241,7 @@ async def remove_mavlink_log_services(i_know_what_i_am_doing: bool = False) -> A
 async def remove_mavlink_log_services_stream(i_know_what_i_am_doing: bool = False) -> StreamingResponse:
     """Stream the deletion of MAVLink log files, providing real-time updates about each file being deleted."""
     check_what_i_am_doing(i_know_what_i_am_doing)
-    return deletion_stream_response(Path(MAVLINK_LOG_FOLDER_PATH))
+    return deletion_stream_response(delete_everything_stream(Path(MAVLINK_LOG_FOLDER_PATH)))
 
 
 @app.get("/services/check_log_folder_size", status_code=status.HTTP_200_OK)