From e98af175bec1dc8fb87f969f551cccfa087a31c7 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 11:48:02 +0200 Subject: [PATCH 1/4] feat: introduce DataStore dedicated to specific bucketing types --- .../AdministratorGuide/Systems/Accounting/index.rst | 8 ++++++++ src/DIRAC/AccountingSystem/ConfigTemplate.cfg | 3 +++ src/DIRAC/AccountingSystem/DB/AccountingDB.py | 9 ++++++++- src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py | 7 +++++-- src/DIRAC/AccountingSystem/Service/DataStoreHandler.py | 5 ++++- 5 files changed, 28 insertions(+), 4 deletions(-) diff --git a/docs/source/AdministratorGuide/Systems/Accounting/index.rst b/docs/source/AdministratorGuide/Systems/Accounting/index.rst index b77ea651f38..5fb68f54b8c 100644 --- a/docs/source/AdministratorGuide/Systems/Accounting/index.rst +++ b/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 diff --git a/src/DIRAC/AccountingSystem/ConfigTemplate.cfg b/src/DIRAC/AccountingSystem/ConfigTemplate.cfg index c96d34a6158..2ec8331bd21 100644 --- a/src/DIRAC/AccountingSystem/ConfigTemplate.cfg +++ b/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 diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 87fd4dee9f8..d5f933da4cd 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -16,11 +16,12 @@ 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) 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 +130,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] @@ -279,6 +283,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"]: diff --git a/src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py b/src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py index 64b1c85be6d..6e11d08a26d 100644 --- a/src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py @@ -6,10 +6,11 @@ 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() @@ -17,7 +18,9 @@ def __init__(self, csPath, readOnly=False): 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}") diff --git a/src/DIRAC/AccountingSystem/Service/DataStoreHandler.py b/src/DIRAC/AccountingSystem/Service/DataStoreHandler.py index 69516b1b0e8..ed4f9ab36b4 100644 --- a/src/DIRAC/AccountingSystem/Service/DataStoreHandler.py +++ b/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: From 9fc2cd0fc2ed11c0b56457f067dd78188f5fbc2e Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:03:16 +0200 Subject: [PATCH 2/4] feat (AccountingDB): cache some of the calls for optimization --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 25 ++++++++++++++++--- .../AccountingSystem/Test_AccountingDB.py | 4 +-- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index d5f933da4cd..97a6e2230e7 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/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,10 +17,20 @@ gSynchro = ThreadSafe.Synchronizer() +ADDKEYVALUE_CACHE_SIZE = 2048 +GETTABLENAME_CACHE_SIZE = 128 + class AccountingDB(DB): 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 @@ -163,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() @@ -447,6 +462,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 @@ -573,10 +589,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) @@ -591,12 +609,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]" @@ -1299,6 +1315,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 diff --git a/tests/Integration/AccountingSystem/Test_AccountingDB.py b/tests/Integration/AccountingSystem/Test_AccountingDB.py index 468eb50a008..4b85faa85cf 100644 --- a/tests/Integration/AccountingSystem/Test_AccountingDB.py +++ b/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 From fa38473ec8d21d123ac5f167407af72d9195c074 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 16:05:07 +0200 Subject: [PATCH 3/4] fix (AccountingDB): reset pending lock time when encountering an error --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 97a6e2230e7..437e3dc04dc 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -217,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"] From 4440a4d97debc4fa8a544bf05ee7e4e975cc6e2a Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 16:52:34 +0200 Subject: [PATCH 4/4] fix (AccountingDB): escape column name --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 437e3dc04dc..abc4ac85c51 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -205,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()] @@ -713,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: @@ -734,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 "