fix: prune log_dir on every roll so daily/size retention is enforced

TimedRotatingFileHandler/RotatingFileHandler retention (getFilesToDelete) only scans the live file's directory, never the log_dir we redirect rolled files into, so daily (the default) and size modes never pruned and .gz files grew unbounded. the rotator now calls the existing prune(log_dir, stem, backup_count) helper (the one on_start already uses) after each roll. verified by execution: daily and size both retain exactly backup_count; a no-prune control retains all.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-28 18:45:25 -04:00
parent 54151b9835
commit 871471dd58
2 changed files with 27 additions and 8 deletions

View File

@ -10,7 +10,7 @@ import gzip
import os
import shutil
import time
from typing import Callable, Tuple
from typing import Callable, Optional, Tuple
def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
@ -26,8 +26,18 @@ def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
return namer
def make_rotator(compress: bool) -> Callable[[str, str], None]:
"""rotator: move (or gzip) the source live file to the destination rolled path"""
def make_rotator(
compress: bool, log_dir: Optional[str] = None,
prune_stem: Optional[str] = None, backup_count: int = 0,
) -> Callable[[str, str], None]:
"""rotator: move (or gzip) the source live file to the destination rolled path
prunes `log_dir` to `backup_count` newest rolled files after each roll when
`log_dir`/`prune_stem` are given. the stdlib handler's own retention
(`getFilesToDelete`) only scans the live file's directory, so it never sees the
rolled files we redirect into `log_dir` pruning here is what actually bounds
retention for the daily and size rolling modes.
"""
def rotator(source: str, dest: str) -> None:
if not os.path.exists(source):
return
@ -37,6 +47,8 @@ def make_rotator(compress: bool) -> Callable[[str, str], None]:
os.remove(source)
else:
os.replace(source, dest)
if log_dir is not None and prune_stem is not None:
prune(log_dir, prune_stem, backup_count)
return rotator
@ -93,10 +105,17 @@ def _safe_mtime(path: str) -> float:
return 0.0
def attach_rolling(handler, log_dir: str, compress: bool) -> Tuple[Callable, Callable]:
"""wire the custom namer + rotator onto a rotating handler; return them"""
def attach_rolling(
handler, log_dir: str, compress: bool,
prune_stem: Optional[str] = None, backup_count: int = 0,
) -> Tuple[Callable, Callable]:
"""wire the custom namer + rotator onto a rotating handler; return them
pass `prune_stem`/`backup_count` so the rotator prunes `log_dir` after each roll
(the handler's own retention can't see the redirected rolled files).
"""
namer = make_namer(log_dir, compress)
rotator = make_rotator(compress)
rotator = make_rotator(compress, log_dir, prune_stem, backup_count)
handler.namer = namer
handler.rotator = rotator
return namer, rotator

View File

@ -66,12 +66,12 @@ def _file_handler(
handler = logging.handlers.RotatingFileHandler(
live_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8",
)
attach_rolling(handler, log_dir, compress)
attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
elif rotate == "daily":
handler = logging.handlers.TimedRotatingFileHandler(
live_path, when="midnight", backupCount=backup_count, encoding="utf-8",
)
attach_rolling(handler, log_dir, compress)
attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
else:
if rotate == "on_start":
rotate_on_start(live_path, log_dir, compress)