From bfbb398ad325fb69d85e80f21a686bd838cefe1a Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 11:48:02 +0200 Subject: [PATCH 01/11] feat: introduce dedicated bucketing types --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 8 +++++++- src/DIRAC/AccountingSystem/DB/MultiAccountingDB.py | 7 +++++-- src/DIRAC/AccountingSystem/Service/DataStoreHandler.py | 5 ++++- 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 01389df7171..1f9afb122bc 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -15,11 +15,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 @@ -128,6 +129,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] @@ -278,6 +282,8 @@ 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) 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 b2688159ab726ef58ba61a9218e99c168f26d991 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 13:33:49 +0200 Subject: [PATCH 02/11] fixup! feat: introduce dedicated bucketing types --- 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 1f9afb122bc..35641dbafbb 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -284,6 +284,7 @@ def registerType(self, name, definitionKeyFields, definitionAccountingFields, bu """ 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"]: From a1ec638119c9444b877c77cd8cd02a083dafe28d Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:03:16 +0200 Subject: [PATCH 03/11] feat: optimize --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 35641dbafbb..33e0fa18960 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 @@ -56,6 +59,12 @@ def __init__(self, name="Accounting/AccountingDB", readOnly=False, parentLogger= self.__lastCompactionEpoch = TimeUtilities.toEpoch(lcd) self.__registerTypes() + # Cached method + self._addkeyvalue_cache = LRUCache(maxsize=1024) + self._addkeyvalue_lock = threading.Lock() + self._gettablename_cache = LRUCache(maxsize=128) + self._gettablename_lock = threading.Lock() + def __loadTablesCreated(self): result = self._query("show tables") if not result["OK"]: # pylint: disable=invalid-sequence-index @@ -162,6 +171,8 @@ def loadPendingRecords(self): """ Load all records pending to insertion and generate threaded jobs """ + print(f"CHRIS {self.__addKeyValue.cache_info()=}") + print(f"CHRIS {self._getTableName.cache_info()=}") gSynchro.lock() try: now = time.time() @@ -446,6 +457,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 @@ -572,10 +584,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) @@ -590,12 +604,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]" @@ -1298,6 +1310,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 From 0f91cbbb220f8cd931a5584498becfb8ebc0d4c7 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:03:27 +0200 Subject: [PATCH 04/11] feat: optimize test --- tests/Integration/AccountingSystem/Test_AccountingDB.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 a0ce9fe4cd1ce0bc637c53ddafc9b5cf168cc856 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:05:38 +0200 Subject: [PATCH 05/11] fixup! feat: optimize --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 33e0fa18960..4232b5fcdb4 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -20,6 +20,13 @@ 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=1024) + self._addkeyvalue_lock = threading.Lock() + self._gettablename_cache = LRUCache(maxsize=128) + self._gettablename_lock = threading.Lock() + self.maxBucketTime = 604800 # 1 w self.autoCompact = False self.__readOnly = readOnly @@ -59,12 +66,6 @@ def __init__(self, name="Accounting/AccountingDB", readOnly=False, parentLogger= self.__lastCompactionEpoch = TimeUtilities.toEpoch(lcd) self.__registerTypes() - # Cached method - self._addkeyvalue_cache = LRUCache(maxsize=1024) - self._addkeyvalue_lock = threading.Lock() - self._gettablename_cache = LRUCache(maxsize=128) - self._gettablename_lock = threading.Lock() - def __loadTablesCreated(self): result = self._query("show tables") if not result["OK"]: # pylint: disable=invalid-sequence-index From 12ad2581e4f244189440b30e5d24fafdf3ef43e9 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:39:19 +0200 Subject: [PATCH 06/11] fixup! feat: optimize --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 4232b5fcdb4..be23a3c512e 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -172,8 +172,8 @@ def loadPendingRecords(self): """ Load all records pending to insertion and generate threaded jobs """ - print(f"CHRIS {self.__addKeyValue.cache_info()=}") - print(f"CHRIS {self._getTableName.cache_info()=}") + print(f"CHRIS {self.__addKeyValue.cache=}") + print(f"CHRIS {self._getTableName.cache=}") gSynchro.lock() try: now = time.time() From f092729f1cef6e111a284c07435890b4b0894589 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 14:53:45 +0200 Subject: [PATCH 07/11] fixup! feat: optimize --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index be23a3c512e..7e5dcc6033e 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -172,8 +172,8 @@ def loadPendingRecords(self): """ Load all records pending to insertion and generate threaded jobs """ - print(f"CHRIS {self.__addKeyValue.cache=}") - print(f"CHRIS {self._getTableName.cache=}") + print(f"CHRIS addkeyvalue cache {self.__addKeyValue.cache.currsize}/{self.__addKeyValue.cache.maxsize}") + print(f"CHRIS gettablename{self._getTableName.cache.currsize}/{self._getTableName.cache.maxsize}") gSynchro.lock() try: now = time.time() From 2ed02903fdbcf8a586da008a349b0e9861334ac9 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 16:05:07 +0200 Subject: [PATCH 08/11] 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 7e5dcc6033e..2aede98a4b9 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -213,6 +213,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 d30dcb67c1162fc179d90b3b94edf7eb8ebca8db Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 16:52:34 +0200 Subject: [PATCH 09/11] fix (AccountingDB): escape column name --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 2aede98a4b9..cf6e3f2de1d 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -201,7 +201,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()] From e39ca1b514e79bfc1cb4a191d920c6f838c7ab03 Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 17:14:20 +0200 Subject: [PATCH 10/11] fixup! fix (AccountingDB): escape column name --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index cf6e3f2de1d..9431f9ec484 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -730,7 +730,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 " From 410699c426fbf07bebf16649233790173e62a3ed Mon Sep 17 00:00:00 2001 From: Christophe Haen Date: Thu, 23 Jul 2026 17:29:42 +0200 Subject: [PATCH 11/11] fixup! fixup! fix (AccountingDB): escape column name --- src/DIRAC/AccountingSystem/DB/AccountingDB.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/DIRAC/AccountingSystem/DB/AccountingDB.py b/src/DIRAC/AccountingSystem/DB/AccountingDB.py index 9431f9ec484..7828cdb7996 100644 --- a/src/DIRAC/AccountingSystem/DB/AccountingDB.py +++ b/src/DIRAC/AccountingSystem/DB/AccountingDB.py @@ -709,7 +709,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: