feat: introduce dedicated bucketing types

docs/source/AdministratorGuide/Systems/Accounting/index.rst

@@ -74,6 +74,14 @@ For instance::
 With the previous configuration all accounting data will be stored and retrieved from the usual database except for the _WMSHistory_ type that will be stored and retrieved from the _Acc2_ database.
 
 
+Dedicated DataStore accounting
+==============================
+
+It is possible to run multiple DataStore, each bucketing only certain type of accounting. It can be useful in case of backlog for certain types of accounting. It is however not recommended to run like that.
+
+For that you just have to define ``AccountingTypes=DataOperation,DataStore,etc`` for all the DataStore services. This has the drawback of needing you to make sure that each type of accounting is defined exactly once (i.e. no multiple agents operate on the same type, and every type appears in one agent).
+
+
 .. _datastorehelpers:
 
 DataStore Helpers

src/DIRAC/AccountingSystem/ConfigTemplate.cfg

@@ -10,6 +10,9 @@ Services
     {
       Default = authenticated
     }
+    # Specify which bucketing to run. Not recommended, see the docs
+    # for more details.
+    # AccountingTypes = DataStorage, DataOperation
   }
   ##END
   ##BEGIN ReportGenerator

src/DIRAC/AccountingSystem/DB/AccountingDB.py

@@ -5,6 +5,9 @@
 import threading
 import time
 
+from cachetools import cachedmethod, LRUCache, TTLCache, cached
+
+
 from DIRAC import S_ERROR, S_OK
 from DIRAC.Core.Base.DB import DB
 from DIRAC.Core.Utilities import DEncode, List, ThreadSafe, TimeUtilities
@@ -14,13 +17,24 @@
 
 gSynchro = ThreadSafe.Synchronizer()
 
+ADDKEYVALUE_CACHE_SIZE = 2048
+GETTABLENAME_CACHE_SIZE = 128
+
 
 class AccountingDB(DB):
-    def __init__(self, name="Accounting/AccountingDB", readOnly=False, parentLogger=None):
+    def __init__(self, name="Accounting/AccountingDB", readOnly=False, parentLogger=None, accounting_types=None):
         DB.__init__(self, "AccountingDB", name, parentLogger=parentLogger)
+
+        # Cached method
+        self._addkeyvalue_cache = LRUCache(maxsize=ADDKEYVALUE_CACHE_SIZE)
+        self._addkeyvalue_lock = threading.Lock()
+        self._gettablename_cache = LRUCache(maxsize=GETTABLENAME_CACHE_SIZE)
+        self._gettablename_lock = threading.Lock()
+
         self.maxBucketTime = 604800  # 1 w
         self.autoCompact = False
         self.__readOnly = readOnly
+        self.__accounting_types = accounting_types if accounting_types else []
         self.__doingCompaction = False
         self.__doingPendingLockTime = 0
         self.__deadLockRetries = 2
@@ -129,6 +143,9 @@ def __loadCatalogFromDB(self):
             raise Exception(retVal["Message"])
         for typesEntry in retVal["Value"]:
             typeName = typesEntry[0]
+            if self.__accounting_types and typeName not in self.__accounting_types:
+                self.log.info("Ignoring accounting type as not in the list", typeName)
+                continue
             keyFields = List.fromChar(typesEntry[1], ",")
             valueFields = List.fromChar(typesEntry[2], ",")
             bucketsLength = DEncode.decode(typesEntry[3].encode())[0]
@@ -159,6 +176,8 @@ def loadPendingRecords(self):
         """
         Load all records pending to insertion and generate threaded jobs
         """
+        self.log.info("addkeyvalue cache", f"{self.__addKeyValue.cache.currsize}/{self.__addKeyValue.cache.maxsize}")
+        self.log.info("gettablename cache", f"{self._getTableName.cache.currsize}/{self._getTableName.cache.maxsize}")
         gSynchro.lock()
         try:
             now = time.time()
@@ -186,7 +205,7 @@ def loadPendingRecords(self):
                 % self.getWaitingRecordsLifeTime()
             )
             req = "SELECT "
-            req += ",".join(sqlFields)
+            req += ", ".join([f"`{f}`" for f in sqlFields])
             req += f" FROM {sqlTableName} "
             req += "WHERE taken = 0 or TIMESTAMPDIFF( SECOND, takenSince, UTC_TIMESTAMP() ) > %s "
             args = [self.getWaitingRecordsLifeTime()]
@@ -198,6 +217,7 @@ def loadPendingRecords(self):
                     "[PENDING] Error when trying to get pending records",
                     f"for {typeName} : {result['Message']}",
                 )
+                self.__doingPendingLockTime = 0
                 return result
             self.log.info(f"[PENDING] Got {len(result['Value'])} pending records for type {typeName}")
             dbData = result["Value"]
@@ -279,6 +299,9 @@ def registerType(self, name, definitionKeyFields, definitionAccountingFields, bu
         """
         Register a new type
         """
+        if self.__accounting_types and name not in self.__accounting_types:
+            self.log.info("Not registering accounting type as not in the list", name)
+            return S_OK(False)
 
         result = self.__loadTablesCreated()
         if not result["OK"]:
@@ -440,6 +463,7 @@ def __getIdForKeyValue(self, typeName, keyName, keyValue, conn=False):
             return S_OK(retVal["Value"][0][0])
         return S_ERROR(f"Key id {keyName} for value {keyValue} does not exist although it should")
 
+    @cachedmethod(lambda self: self._addkeyvalue_cache, lock=lambda self: self._addkeyvalue_lock)
     def __addKeyValue(self, typeName, keyName, keyValue):
         """
         Adds a key value to a key table if not existant
@@ -566,10 +590,12 @@ def __insertFromINTable(self, recordTuples):
         """
         Do the real insert and delete from the in buffer table
         """
+        if self.__readOnly:
+            return S_ERROR("ReadOnly mode enabled. No modification allowed")
         self.log.verbose("Received bundle to process", f"of {len(recordTuples)} elements")
         for record in recordTuples:
             iD, typeName, startTime, endTime, valuesList, insertionEpoch = record
-            result = self.insertRecordDirectly(typeName, startTime, endTime, valuesList)
+            result = self._insertRecordDirectly(typeName, startTime, endTime, valuesList)
             if not result["OK"]:
                 req = "UPDATE "
                 req += self._getTableName("in", typeName)
@@ -584,12 +610,10 @@ def __insertFromINTable(self, recordTuples):
             if not result["OK"]:
                 self.log.error("Can't delete row from the IN table", result["Message"])
 
-    def insertRecordDirectly(self, typeName, startTime, endTime, valuesList):
+    def _insertRecordDirectly(self, typeName, startTime, endTime, valuesList):
         """
         Add an entry to the type contents
         """
-        if self.__readOnly:
-            return S_ERROR("ReadOnly mode enabled. No modification allowed")
         self.log.info(
             "Adding record",
             "for type %s\n [%s -> %s]"
@@ -689,7 +713,7 @@ def __writeBuckets(self, typeName, buckets, keyValues, valuesList, connObj=False
         sqlFields.extend(self.dbCatalog[typeName]["keys"])
         sqlFields.extend(self.dbCatalog[typeName]["values"])
         sqlUpData = ["entriesInBucket=entriesInBucket+VALUES(entriesInBucket)"]
-        sqlUpData.extend([f"{x}={x}+VALUES({x})" for x in self.dbCatalog[typeName]["values"]])
+        sqlUpData.extend([f"`{x}`=`{x}`+VALUES(`{x}`)" for x in self.dbCatalog[typeName]["values"]])
         valueGroups = []
         sqlValues = []
         for bucketInfo in buckets:
@@ -710,7 +734,7 @@ def __writeBuckets(self, typeName, buckets, keyValues, valuesList, connObj=False
         req = "INSERT INTO "
         req += self._getTableName("bucket", typeName)
         req += " ("
-        req += ",".join(sqlFields)
+        req += ", ".join([f"`{f}`" for f in sqlFields])
         req += ") VALUES "
         req += ",".join(valueGroups)
         req += " ON DUPLICATE KEY UPDATE "
@@ -1292,6 +1316,7 @@ def __commitTransaction(self, connObj):
     def __rollbackTransaction(self, connObj):
         return self._query("ROLLBACK", conn=connObj)
 
+    @cachedmethod(lambda self: self._gettablename_cache, lock=lambda self: self._gettablename_lock)
     def _getTableName(self, tableType, typeName, keyName=None):
         """
         Generate table name

src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py

@@ -6,18 +6,21 @@
 
 
 class MultiAccountingDB:
-    def __init__(self, csPath, readOnly=False):
+    def __init__(self, csPath, readOnly=False, accounting_types=None):
         self.__csPath = csPath
         self.__readOnly = readOnly
         self.__dbByType = {}
+        self.__accountingTypes = accounting_types if accounting_types else []
         self.__defaultDB = "AccountingDB/AccountingDB"
         self.__log = gLogger.getSubLogger(self.__class__.__name__)
         self.__generateDBs()
         self.__registerMethods()
 
     def __generateDBs(self):
         self.__log.notice("Creating default AccountingDB...")
-        self.__allDBs = {self.__defaultDB: AccountingDB(readOnly=self.__readOnly)}
+        self.__allDBs = {
+            self.__defaultDB: AccountingDB(readOnly=self.__readOnly, accounting_types=self.__accountingTypes)
+        }
         result = gConfig.getOptionsDict(self.__csPath)
         if not result["OK"]:
             gLogger.verbose("No extra databases defined", f"in {self.__csPath}")

src/DIRAC/AccountingSystem/Service/DataStoreHandler.py

@@ -28,7 +28,10 @@ class DataStoreHandler(RequestHandler):
     @classmethod
     def initializeHandler(cls, svcInfoDict):
         multiPath = PathFinder.getDatabaseSection("Accounting/MultiDB")
-        cls.__acDB = MultiAccountingDB(multiPath)
+        # we can focus on only some of the accoutning type
+        cls.accounting_types = getServiceOption(svcInfoDict, "AccountingTypes", [])
+
+        cls.__acDB = MultiAccountingDB(multiPath, accounting_types=cls.accounting_types)
         # we can run multiple services in read only mode. In that case we do not bucket
         cls.runBucketing = getServiceOption(svcInfoDict, "RunBucketing", True)
         if cls.runBucketing:

tests/Integration/AccountingSystem/Test_AccountingDB.py

@@ -42,10 +42,10 @@
 
 @pytest.fixture
 def inout():
-    res = acDB.insertRecordDirectly("Pilot", startTime, middleTime, keyValues_1 + nonKeyValue_1)
+    res = acDB._insertRecordDirectly("Pilot", startTime, middleTime, keyValues_1 + nonKeyValue_1)
     assert res["OK"], res["Message"]
 
-    res = acDB.insertRecordDirectly("Pilot", middleTime, endTime, keyValues_2 + nonKeyValue_2)
+    res = acDB._insertRecordDirectly("Pilot", middleTime, endTime, keyValues_2 + nonKeyValue_2)
     assert res["OK"], res["Message"]
 
     yield