Repo cleaner improvements (#1128)

* make sure we don't fail silently if we cannot load a module

* Add a parameter for production to say whether to delete the objects without runs or not

* fix the logger in the rules

* logger in ccdb is set externally

* Doc for accessing qcdb

* fix

* leftover

Author: Barthelemy

Framework/script/RepoCleaner/Ccdb.py

@@ -7,6 +7,9 @@
 import requests
 
 
+logger = logging  # default logger
+
+
 class ObjectVersion:
     '''
     A version of an object in the CCDB. 
@@ -50,7 +53,7 @@ class Ccdb:
     counter_preserved: int = 0
 
     def __init__(self, url):
-        logging.info(f"Instantiate CCDB at {url}")
+        logger.info(f"Instantiate CCDB at {url}")
         self.url = url
 
     def getObjectsList(self, added_since: int = 0) -> List[str]:
@@ -60,16 +63,16 @@ def getObjectsList(self, added_since: int = 0) -> List[str]:
         :param added_since: if specified, only return objects added since this timestamp in epoch milliseconds.
         :return A list of strings, each containing a path to an object in the CCDB.
         '''
-        logging.debug(f"added_since : {added_since}")
+        logger.debug(f"added_since : {added_since}")
         url_for_all_obj = self.url + '/latest/.*'
-        logging.debug(f"Ccdb::getObjectsList -> {url_for_all_obj}")
+        logger.debug(f"Ccdb::getObjectsList -> {url_for_all_obj}")
         headers = {'Accept':'application/json', 'If-Not-Before':str(added_since)}
         r = requests.get(url_for_all_obj, headers=headers)
         r.raise_for_status()
         try:
             json = r.json()
         except JSONDecodeError as err:
-            logging.error(f"JSON decode error: {err}")
+            logger.error(f"JSON decode error: {err}")
             raise
         paths = []
         for item in json['objects']:
@@ -83,7 +86,7 @@ def getVersionsList(self, object_path: str) -> List[ObjectVersion]:
         :return A list of ObjectVersion.
         '''
         url_browse_all_versions = self.url + '/browse/' + object_path
-        logging.debug(f"Ccdb::getVersionsList -> {url_browse_all_versions}")
+        logger.debug(f"Ccdb::getVersionsList -> {url_browse_all_versions}")
         headers = {'Accept':'application/json', 'Connection': 'close'}
         r = requests.get(url_browse_all_versions, headers=headers)
         r.raise_for_status()
@@ -106,7 +109,7 @@ def deleteVersion(self, version: ObjectVersion):
         :param version: The version of the object to delete, as an instance of ObjectVersion.
         '''
         url_delete = self.url + '/' + version.path + '/' + str(version.validFrom) + '/' + version.uuid
-        logging.debug(f"Delete version at url {url_delete}")
+        logger.debug(f"Delete version at url {url_delete}")
         headers = {'Connection': 'close'}
         try:
             r = requests.delete(url_delete, headers=headers)
@@ -126,9 +129,9 @@ def updateValidity(self, version: ObjectVersion, valid_from: int, valid_to: int,
         :param metadata: Add or modify metadata
         '''
         full_path = self.url + '/' + version.path + '/' + str(valid_from) + '/' + str(valid_to) + '/' + str(version.uuid) + '?'
-        logging.debug(f"Update end limit validity of {version.path} ({version.uuid}) from {version.validTo} to {valid_to}")
+        logger.debug(f"Update end limit validity of {version.path} ({version.uuid}) from {version.validTo} to {valid_to}")
         if metadata is not None:
-            logging.debug(f"{metadata}")
+            logger.debug(f"{metadata}")
             for key in metadata:
                 full_path += key + "=" + metadata[key] + "&"
         try:
@@ -140,7 +143,7 @@ def updateValidity(self, version: ObjectVersion, valid_from: int, valid_to: int,
             print(e)
             sys.exit(1)  # really ?
 
-def putVersion(self, version: ObjectVersion, data):
+    def putVersion(self, version: ObjectVersion, data):
         '''
         :param version: An ObjectVersion that describes the data to be uploaded.
         :param data: the actual data to send. E.g.:{'somekey': 'somevalue'}
@@ -150,17 +153,17 @@ def putVersion(self, version: ObjectVersion, data):
         if version.metadata is not None:
             for key in version.metadata:
                 full_path += key + "=" + version.metadata[key] + "/"
-        logging.debug(f"fullpath: {full_path}")
+        logger.debug(f"fullpath: {full_path}")
         headers = {'Connection': 'close'}
         r = requests.post(full_path, files=data, headers=headers)
         if r.ok:
-            logging.debug(f"Version pushed to {version.path}")
+            logger.debug(f"Version pushed to {version.path}")
         else:
-            logging.error(f"Could not post a new version of {version.path}: {r.text}")
+            logger.error(f"Could not post a new version of {version.path}: {r.text}")
 
 def main():
-    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%d-%b-%y %H:%M:%S')
-    logging.getLogger().setLevel(int(10))
+    logger.basicConfig(level=logger.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%d-%b-%y %H:%M:%S')
+    logger.getLogger().setLevel(int(10))
 
     ccdb = Ccdb('http://ccdb-test.cern.ch:8080')
 

Framework/script/RepoCleaner/README.md

@@ -42,3 +42,9 @@ Most of the classes and Rules have a main to help test them. To run do e.g. `pyt
 
 ## Installation
 CMake will install the python scripts in bin and the config file in etc.
+
+## Example
+
+```
+PYTHONPATH=./rules:$PYTHONPATH ./o2-qc-repo-cleaner --dry-run --config config-test.yaml --dry-run --only-path qc/DAQ --log-level 10
+```

Framework/script/RepoCleaner/config-test.yaml

@@ -5,6 +5,10 @@ Rules:
   - object_path: QcTask/example
     delay: 120
     policy: 1_per_hour
-    
+  - object_path: qc/TST/MO/repo/test/.*
+    delay: 60
+    policy: production
+    delete_when_no_run: True
+
 Ccdb:
    Url: http://ccdb-test.cern.ch:8080

Framework/script/RepoCleaner/o2-qc-repo-cleaner

@@ -17,13 +17,14 @@
 #
 # Usage
 #         # run with debug logs and don't actually touch the database
-#         ./repoCleaner --dry-run --log-level 10 
+#         PYTHONPATH=./rules:$PYTHONPATH./o2-qc-repo-cleaner --dry-run --log-level 10
 
 import argparse
 import logging
 import requests
 import re
 import sys
+import importlib
 from typing import List
 import tempfile
 import socket
@@ -289,17 +290,25 @@ def read_config(args):
     return ccdb_url, rules
 
 
-def process_object(object_path, rules, ccdb):
+def process_object(object_path, rules, ccdb, args):
     logger = create_parallel_logger()
+    logger.setLevel(int(args.log_level))
     logger.info(f"Processing {object_path}")
 
     # Take the first matching rule, if any
     rule = findMatchingRule(rules, object_path)
+
     if rule is None:
+        logger.info(f"    no matching rule")
         return
 
     # Apply rule on object (find the plug-in script and apply)
-    module = __import__(rule.policy)
+    try:
+        module = importlib.import_module(rule.policy)
+        module.logger = logger
+    except ModuleNotFoundError as err:
+        logger.error(f"could not load module {rule.policy}")
+        return
     stats = module.process(ccdb, object_path, int(rule.delay),  # rule.migration == "True",
                            rule.extra_params)
     logger.info(f"{rule.policy} applied on {object_path}: {stats}")
@@ -309,17 +318,17 @@ def run(args, ccdb_url, rules):
 
     # Get list of objects from CCDB
     ccdb = Ccdb(ccdb_url)
+    ccdb.logger = logging.getLogger
     paths = ccdb.getObjectsList(getTimestampLastExecution())
     if args.only_path != '':
         paths = [item for item in paths if item is not None and item.startswith(args.only_path)]
     logging.debug(paths)
-    logging.debug(len(paths))
     # For each object call the first matching rule
     logging.info("Loop through the objects and apply first matching rule.")
 
     logging.info(f"workers: {args.workers}")
     pool = mp.Pool(int(args.workers))
-    [pool.apply_async(process_object, args=(object_path, rules, ccdb)) for object_path in paths]
+    [pool.apply_async(process_object, args=(object_path, rules, ccdb, args)) for object_path in paths]
     pool.close()
     pool.join()
 

Framework/script/RepoCleaner/rules/1_per_hour.py

@@ -1,11 +1,14 @@
 from datetime import datetime
 from datetime import timedelta
-import logging
+import logger
 from typing import Dict
 
 from Ccdb import Ccdb, ObjectVersion
 
 
+logger = logging  # default logger
+
+
 def process(ccdb: Ccdb, object_path: str, delay: int, #migration: bool,
             extra_params: Dict[str, str]):
     '''
@@ -23,7 +26,7 @@ def process(ccdb: Ccdb, object_path: str, delay: int, #migration: bool,
     :return a dictionary with the number of deleted, preserved and updated versions. Total = deleted+preserved.
     '''
 
-    logging.debug(f"Plugin 1_per_hour processing {object_path}")
+    logger.debug(f"Plugin 1_per_hour processing {object_path}")
 
     versions = ccdb.getVersionsList(object_path)
 
@@ -41,22 +44,22 @@ def process(ccdb: Ccdb, object_path: str, delay: int, #migration: bool,
         else:
             if v.validFromAsDt < datetime.now() - timedelta(minutes=delay):
                 deletion_list.append(v)
-                logging.debug(f"not in the grace period, we delete {v}")
+                logger.debug(f"not in the grace period, we delete {v}")
                 ccdb.deleteVersion(v)
             else:
                 preservation_list.append(v)
 
-    logging.debug("deleted : ")
+    logger.debug("deleted : ")
     for v in deletion_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
-    logging.debug("preserved : ")
+    logger.debug("preserved : ")
     for v in preservation_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
-    logging.debug("updated : ")
+    logger.debug("updated : ")
     for v in update_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
     return {"deleted" : len(deletion_list), "preserved": len(preservation_list), "updated" : len(update_list)}
 

Framework/script/RepoCleaner/rules/1_per_run.py

@@ -6,9 +6,14 @@
 import dryable
 from typing import Dict
 
+
+logger = logging  # default logger
+
+
 def in_grace_period(version: ObjectVersion, delay: int):
     return not (version.validFromAsDt < datetime.now() - timedelta(minutes=delay))
 
+
 def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, str]):
     '''
     Process this deletion rule on the object. We use the CCDB passed by argument.
@@ -29,41 +34,41 @@ def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, st
     :return a dictionary with the number of deleted, preserved and updated versions. Total = deleted+preserved.
     '''
 
-    logging.debug(f"Plugin 1_per_run processing {object_path}")
+    logger.debug(f"Plugin 1_per_run processing {object_path}")
 
     preservation_list: List[ObjectVersion] = []
     deletion_list: List[ObjectVersion] = []
     update_list: List[ObjectVersion] = []
     runs_dict: DefaultDict[str, List[ObjectVersion]] = defaultdict(list)
 
     delete_when_no_run = (extra_params.get("delete_when_no_run", False) == True)
-    logging.debug(f"delete_when_no_run : {delete_when_no_run}")
+    logger.debug(f"delete_when_no_run : {delete_when_no_run}")
 
     # Find all the runs and group the versions
     versions = ccdb.getVersionsList(object_path)
     for v in versions:
-        logging.debug(f"Processing {v}")
+        logger.debug(f"Processing {v}")
         if "Run" in v.metadata:
             runs_dict[v.metadata['Run']].append(v)
         elif "RunNumber" in v.metadata:
             runs_dict[v.metadata['RunNumber']].append(v)
         else:
             runs_dict[-1].append(v) # the ones with no run specified
 
-    logging.debug(f"Number of runs : {len(runs_dict)}")
-    logging.debug(f"Number of versions without runs : {len(runs_dict[-1])}")
+    logger.debug(f"Number of runs : {len(runs_dict)}")
+    logger.debug(f"Number of versions without runs : {len(runs_dict[-1])}")
 
     # if we should not touch the files with no runs, let's just remove them from the map
     if not delete_when_no_run:
         del runs_dict[-1]
 
     # Dispatch the versions to deletion and preservation lists
     for r, run_versions in runs_dict.items():
-        # logging.debug(f"- run {r}")
+        # logger.debug(f"- run {r}")
 
         freshest: ObjectVersion = None
         for v in run_versions:
-            # logging.debug(f"  - version {v}")
+            # logger.debug(f"  - version {v}")
             if freshest is None or freshest.validFromAsDt < v.validFromAsDt:
                 if freshest is not None:
                     if in_grace_period(freshest, delay):
@@ -82,23 +87,23 @@ def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, st
     for d in deletion_list:
         ccdb.deleteVersion(d)
 
-    logging.debug(f"deleted ({len(deletion_list)}) : ")
+    logger.debug(f"deleted ({len(deletion_list)}) : ")
     for v in deletion_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
-    logging.debug(f"preserved ({len(preservation_list)}) : ")
+    logger.debug(f"preserved ({len(preservation_list)}) : ")
     for v in preservation_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
-    logging.debug(f"updated ({len(update_list)}) : ")
+    logger.debug(f"updated ({len(update_list)}) : ")
     for v in update_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
     return {"deleted" : len(deletion_list), "preserved": len(preservation_list), "updated" : len(update_list)}
 
 
 def main():
-    logging.getLogger().setLevel(int(10))
+    logger.getLogger().setLevel(int(10))
     dryable.set( True )
     ccdb = Ccdb('http://ccdb-test.cern.ch:8080')
     process(ccdb, "qc/testRunCleanup", 0)

Framework/script/RepoCleaner/rules/last_only.py

@@ -6,6 +6,9 @@
 from Ccdb import Ccdb, ObjectVersion
 
 
+logger = logging  # default logger
+
+
 def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, str]):
     '''
     Process this deletion rule on the object. We use the CCDB passed by argument.
@@ -19,7 +22,7 @@ def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, st
     :return a dictionary with the number of deleted, preserved and updated versions. Total = deleted+preserved.
     '''
 
-    logging.debug(f"Plugin last_only processing {object_path}")
+    logger.debug(f"Plugin last_only processing {object_path}")
 
     versions = ccdb.getVersionsList(object_path)
 
@@ -30,7 +33,7 @@ def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, st
     for v in versions:
         if earliest == None or v.validFromAsDt > earliest.validFromAsDt:
             earliest = v
-    logging.debug(f"earliest : {earliest}")
+    logger.debug(f"earliest : {earliest}")
 
     # delete the non-earliest if we are not in the grace period
     for v in versions:
@@ -44,13 +47,13 @@ def process(ccdb: Ccdb, object_path: str, delay: int, extra_params: Dict[str, st
         else:
             preservation_list.append(v)
 
-    logging.debug("deleted : ")
+    logger.debug("deleted : ")
     for v in deletion_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
-    logging.debug("preserved : ")
+    logger.debug("preserved : ")
     for v in preservation_list:
-        logging.debug(f"   {v}")
+        logger.debug(f"   {v}")
 
     return {"deleted" : len(deletion_list), "preserved": len(preservation_list)}