feat: New cleanDirectory utility function

Author: sfayer

src/DIRAC/Core/Utilities/File.py

@@ -13,7 +13,10 @@
 import errno
 import stat
 import tempfile
+import fnmatch
+import time
 from contextlib import contextmanager
+from pathlib import Path
 
 # Translation table of a given unit to Bytes
 # I know, it should be kB...
@@ -256,6 +259,68 @@ def convertSizeUnits(size, srcUnit, dstUnit):
         return -sys.maxsize
 
 
+def cleanDirectory(workDir, maxSecs=None, filePatterns=[], maxDepth=0, callbackFn=None, delEmptyDirs=False):
+    """Recursively clean files older than a threshold.
+
+    Walks ``workDir`` bottom-up and deletes (or invokes ``callbackFn`` on)
+    regular files that are older than ``maxSecs`` seconds which match the ``filePatterns``
+    glob. Empty directories can also be removed with ``delEmptyDirs``=``True``.
+
+    :param workDir: directory to scan
+    :param maxSecs: age threshold in seconds (files older are deleted); pass
+                    ``None`` to skip the age filter entirely
+    :param filePatterns: list of globs, only files matching this will be considered
+    :param maxDepth: maximum directory depth to process (0 = unlimited, 1 = root only)
+    :param callbackFn: If ``None`` files will be unlinked, otherwise this function
+                       will be called for matchin files instead. Function should take a
+                       single``Path`` object argument. Returning True indicates the
+                       file was processed without error. Returning False will add the
+                       path onto the list of failed files.
+    :param delEmptyDirs: if ``True``, delete directories that are empty after
+                           file cleanup. All directories are considered, the filePatterns
+                           glob is not used to filter these.
+    :returns: list of file (and dir) paths that could not be deleted (empty on success)
+    """
+    errFiles = []
+    timeThresh = time.time() - maxSecs if maxSecs is not None else None
+    rootPath = Path(workDir)
+
+    for curRoot, dirs, files in os.walk(workDir, topdown=False):
+        curPath = Path(curRoot)
+        depth = len(curPath.relative_to(rootPath).parts)
+
+        # Only process files if we're within maxDepth
+        if not maxDepth or depth < maxDepth:
+            for fileName in files:
+                if any(fnmatch.fnmatch(fileName, p) for p in filePatterns):
+                    filePath = curPath / fileName
+                    try:
+                        if not filePath.is_file() or filePath.is_symlink():
+                            continue  # Not a regular file
+                        if timeThresh is not None and filePath.stat().st_mtime >= timeThresh:
+                            continue  # file is not old enough
+                        if callbackFn:
+                            if not callbackFn(filePath):
+                                errFiles.append(str(filePath))
+                                continue
+                        else:
+                            filePath.unlink()
+                            continue
+                    except OSError:
+                        errFiles.append(str(filePath))
+                        continue
+
+            # Now files are processed, see if dir is empty if we're deleting empty dirs
+            if delEmptyDirs and curPath != rootPath:
+                try:
+                    if not any(curPath.iterdir()):
+                        curPath.rmdir()
+                except OSError:
+                    errFiles.append(str(curPath))
+
+    return errFiles
+
+
 @contextmanager
 def secureOpenForWrite(filename=None, *, text=True):
     """Securely open a file for writing.

src/DIRAC/Core/Utilities/test/Test_File.py

@@ -7,18 +7,22 @@
 
 # imports
 import os
+import time
 from os.path import abspath
 import re
 import sys
+from pathlib import Path
 
 from hypothesis import given
 from hypothesis.strategies import floats
 
 from pytest import mark
+from unittest.mock import MagicMock
 
 # sut
 from DIRAC.Core.Utilities.File import (
     checkGuid,
+    cleanDirectory,
     makeGuid,
     getSize,
     getMD5ForFiles,
@@ -134,3 +138,182 @@ def test_convert_loop(nb, srcUnit, dstUnit):
     # We exclude the infinity case
     if converted != float("Inf"):
         assert converted == nb
+
+
+def _set_old_mtime(path: str, age_seconds: int = 3600) -> None:
+    """Set a file's mtime to `age_seconds` ago."""
+    old_time = time.time() - age_seconds
+    os.utime(path, (old_time, old_time))
+
+
+def _build_tree(tmp_path, files, subdirs=None):
+    """Create a file tree and return a mapping of name -> pathlib.Path."""
+    mapping = {}
+    if subdirs:
+        for sd in subdirs:
+            (tmp_path / sd).mkdir(parents=True, exist_ok=True)
+    for name, content in files.items():
+        mapping[name] = tmp_path / name
+        mapping[name].write_text(content)
+    return mapping
+
+
+def test_clean_directory_basic(tmp_path):
+    """Check old file removed, newer file kept and old-non-matching kept."""
+    files = _build_tree(
+        tmp_path,
+        {"DIRAC_old_job": "abc", "DIRAC_new_job": "def", "other.log": "ghi"},
+    )
+    _set_old_mtime(str(files["DIRAC_old_job"]))
+    _set_old_mtime(str(files["other.log"]))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"])
+    assert err == []
+    assert not files["DIRAC_old_job"].exists()
+    assert files["DIRAC_new_job"].exists()
+    assert files["other.log"].exists()
+
+
+def test_clean_directory_empty_dir(tmp_path):
+    assert cleanDirectory(str(tmp_path)) == []
+
+
+def test_clean_directory_maxDepth_restricts_recursion(tmp_path):
+    files = _build_tree(
+        tmp_path,
+        {"root": "a", "DIRAC_root_job": "b"},
+        subdirs=["subdir"],
+    )
+    _set_old_mtime(str(files["root"]))
+    _set_old_mtime(str(files["DIRAC_root_job"]))
+    deep = tmp_path / "subdir" / "DIRAC_deep"
+    deep.write_text("c")
+    _set_old_mtime(str(deep))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"], maxDepth=1)
+    assert err == []
+    assert not files["DIRAC_root_job"].exists()
+    assert (tmp_path / "subdir" / "DIRAC_deep").exists()
+
+
+def test_clean_directory_recursive_without_maxDepth(tmp_path):
+    nested = tmp_path / "a" / "b"
+    nested.mkdir(parents=True)
+    deep = nested / "DIRAC_nested.log"
+    deep.write_text("data")
+    _set_old_mtime(str(deep))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["*.log"])
+    assert err == []
+    assert not deep.exists()
+
+
+def test_clean_directory_symlinks_not_deleted(tmp_path):
+    target = tmp_path / "real"
+    target.write_text("data")
+    symlink = tmp_path / "DIRAC_link"
+    symlink.symlink_to(target)
+    _set_old_mtime(str(symlink))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"])
+    assert err == []
+    assert symlink.exists()
+    assert target.exists()
+
+
+def test_clean_directory_multiple_patterns(tmp_path):
+    files = _build_tree(
+        tmp_path,
+        {"a.out": "x", "b.err": "y", "c.txt": "z"},
+    )
+    for f in files.values():
+        _set_old_mtime(str(f))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["*.out", "*.err"])
+    assert err == []
+    assert not files["a.out"].exists()
+    assert not files["b.err"].exists()
+    assert files["c.txt"].exists()
+
+
+def test_clean_directory_returns_errors_on_unable_delete(mocker, tmp_path):
+    """When unlink raises OSError, errors are collected."""
+    old_file = tmp_path / "DIRAC_blocked"
+    old_file.write_text("data")
+    _set_old_mtime(str(old_file))
+
+    mocker.patch("pathlib.Path.unlink", side_effect=PermissionError("Denied"))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"])
+    assert len(err) == 1
+    assert "DIRAC_blocked" in err[0]
+
+
+def test_clean_directory_empty_dirs_removed(tmp_path):
+    """With delEmptyDirs=True, empty directories are removed after file cleanup."""
+    subdir = tmp_path / "subdir" / "nested"
+    subdir.mkdir(parents=True)
+    log_file = subdir / "old.log"
+    log_file.write_text("data")
+    _set_old_mtime(str(log_file))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["*.log"], delEmptyDirs=True)
+    assert err == []
+    assert not subdir.exists()
+
+
+def test_clean_directory_empty_dirs_kept_by_default(tmp_path):
+    """Without delEmptyDirs, empty directories remain after file cleanup."""
+    subdir = tmp_path / "subdir"
+    subdir.mkdir(parents=True)
+    log_file = subdir / "old.log"
+    log_file.write_text("data")
+    _set_old_mtime(str(log_file))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["*.log"])
+    assert err == []
+    assert subdir.exists()
+
+
+def test_clean_directory_callback_fn(mocker, tmp_path):
+    """When callbackFn is provided, it replaces unlink."""
+    callback = mocker.Mock(side_effect=lambda p: (p.unlink(), True)[1])
+
+    old_file = tmp_path / "DIRAC_test"
+    old_file.write_text("data")
+    _set_old_mtime(str(old_file))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"], callbackFn=callback)
+    assert err == []
+    assert not old_file.exists()
+    callback.assert_called_once()
+
+
+def test_clean_directory_callback_false_recorded_as_error(tmp_path):
+    """Callback returning False records the file as an error; file is left untouched."""
+    def skip_all(_path):
+        return False
+
+    old_file = tmp_path / "DIRAC_test"
+    old_file.write_text("data")
+    _set_old_mtime(str(old_file))
+
+    err = cleanDirectory(str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"], callbackFn=skip_all)
+    assert len(err) == 1
+    assert "DIRAC_test" in err[0]
+    assert old_file.exists()
+
+
+def test_clean_directory_callback_oserror_recorded(mocker, tmp_path):
+    """Callback raising OSError is treated as a deletion error."""
+    mocker.patch("pathlib.Path.unlink", side_effect=PermissionError("Denied"))
+
+    old_file = tmp_path / "DIRAC_test"
+    old_file.write_text("data")
+    _set_old_mtime(str(old_file))
+
+    err = cleanDirectory(
+        str(tmp_path), maxSecs=60, filePatterns=["DIRAC_*"],
+        callbackFn=lambda p: (p.unlink(), True)[1],
+    )
+    assert len(err) == 1

src/DIRAC/FrameworkSystem/private/SecurityFileLog.py

@@ -1,13 +1,12 @@
 import os
-import re
 import time
 import gzip
 import queue
 import shutil
 import threading
 from DIRAC import gLogger, S_OK, S_ERROR
 from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
-from DIRAC.Core.Utilities.File import mkDir
+from DIRAC.Core.Utilities.File import cleanDirectory, mkDir
 
 
 class SecurityFileLog(threading.Thread):
@@ -52,24 +51,31 @@ def run(self):
             fd.close()
 
     def __launchCleaningOldLogFiles(self):
-        nowEpoch = time.time()
-        self.__walkOldLogs(self.__basePath, nowEpoch, re.compile(r"^\d*\.security\.log\.csv$"), 86400, self.__zipOldLog)
-        self.__walkOldLogs(
-            self.__basePath,
-            nowEpoch,
-            re.compile(r"^\d*\.security\.log\.csv\.gz$"),
-            self.__secsToLog,
-            self.__unlinkOldLog,
+        self._cleanupLogs(
+            self.__basePath, 86400, self.__zipOldLog, "*.security.log.csv"
         )
+        self._cleanupLogs(
+            self.__basePath, self.__secsToLog, self.__unlinkOldLog, "*.security.log.csv.gz"
+        )
+
+    def _cleanupLogs(self, basePath, maxSecs, functor, pattern):
+        """Clean old logs matching a pattern, optionally zipping first."""
+
+        errFiles = cleanDirectory(
+            basePath, maxSecs=maxSecs, filePatterns=[pattern], maxDepth=0, callbackFn=functor, delEmptyDirs=True
+        )
+        if errFiles:
+            for fp in errFiles:
+                gLogger.error("Failed to clean security log", fp)
 
     def __unlinkOldLog(self, filePath):
         try:
             gLogger.info(f"Unlinking file {filePath}")
             os.unlink(filePath)
         except Exception as e:
             gLogger.error("Can't unlink old log file", f"{filePath}: {str(e)}")
-            return 1
-        return 0
+            return False
+        return True
 
     def __zipOldLog(self, filePath):
         try:
@@ -79,30 +85,8 @@ def __zipOldLog(self, filePath):
                     shutil.copyfileobj(f_in, f_out)
         except Exception:
             gLogger.exception("Can't compress old log file", filePath)
-            return 1
-        return self.__unlinkOldLog(filePath) + 1
-
-    def __walkOldLogs(self, path, nowEpoch, reLog, executionInSecs, functor):
-        initialEntries = os.listdir(path)
-        numEntries = 0
-        for entry in initialEntries:
-            entryPath = os.path.join(path, entry)
-            if os.path.isdir(entryPath):
-                numEntries += 1
-                numEntriesSubDir = self.__walkOldLogs(entryPath, nowEpoch, reLog, executionInSecs, functor)
-                if numEntriesSubDir == 0:
-                    gLogger.info(f"Removing dir {entryPath}")
-                    try:
-                        os.rmdir(entryPath)
-                        numEntries -= 1
-                    except Exception as e:
-                        gLogger.error("Can't delete directory", f"{entryPath}: {str(e)}")
-            elif os.path.isfile(entryPath):
-                numEntries += 1
-                if reLog.match(entry):
-                    if nowEpoch - os.stat(entryPath)[8] > executionInSecs:
-                        numEntries += functor(entryPath) - 1
-        return numEntries
+            return False
+        return self.__unlinkOldLog(filePath)
 
     def logAction(self, msg):
         if len(msg) != len(self.__requiredFields):