feat (TS): clean TransformationAgent cache upon every loop

Author: chaen

src/DIRAC/TransformationSystem/Agent/TransformationAgent.py

@@ -1,4 +1,4 @@
-"""  TransformationAgent processes transformations found in the transformation database.
+"""TransformationAgent processes transformations found in the transformation database.
 
 The following options can be set for the TransformationAgent.
 
@@ -8,13 +8,15 @@
   :dedent: 2
   :caption: TransformationAgent options
 """
+
 from importlib import import_module
 
 import time
 import os
 import datetime
 import pickle
 import concurrent.futures
+from pathlib import Path
 
 from DIRAC import S_OK, S_ERROR
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
@@ -127,6 +129,9 @@ def execute(self):
         if not res["OK"]:
             self._logError("Failed to obtain transformations:", res["Message"])
             return S_OK()
+
+        active_trans_ids = [t["TransformationID"] for t in res["Value"]]
+        self.cleanOldTransformationCache(active_trans_ids)
         # Process the transformations
         count = 0
         future_to_transID = {}
@@ -164,6 +169,22 @@ def execute(self):
 
         return S_OK()
 
+    def cleanOldTransformationCache(self, active_trans_ids: list[int]):
+        cache_filenames = {Path(self.__cacheFile(tid)) for tid in active_trans_ids}
+        existing_caches = set(Path(self.workDirectory).glob("*.pkl"))
+        useless_cache_files = existing_caches - cache_filenames
+
+        if useless_cache_files:
+            self._logInfo(f"Found potentially {len(useless_cache_files)} useless cache files")
+
+        # Since idle transformations aren't in active_trans_ids, let's filter it more
+        # and take only files that haven't been touched for 2 month
+        last_update_threshold = (datetime.datetime.utcnow() - datetime.timedelta(days=60)).timestamp()
+
+        for cache_file in useless_cache_files:
+            if Path(cache_file).stat().st_mtime < last_update_threshold:
+                cache_file.unlink()
+
     def getTransformations(self):
         """Obtain the transformations to be executed - this is executed at the start of every loop (it's really the
         only real thing in the execute()