Merge pull request DIRACGrid#8216 from fstagni/90_accounting_simplif

[9.0] accounting and monitoring simplifications

Author: fstagni

src/DIRAC/AccountingSystem/Client/AccountingCLI.py

@@ -57,99 +57,6 @@ def showTraceback(self):
         traceback.print_tb(sys.exc_info()[2])
         print("________________________\n")
 
-    def do_registerType(self, args):
-        """
-        Registers a new accounting type
-          Usage : registerType <typeName>
-          <DIRACRoot>/DIRAC/AccountingSystem/Client/Types/<typeName>
-           should exist and inherit the base type
-        """
-        try:
-            argList = args.split()
-            if argList:
-                typeName = argList[0].strip()
-            else:
-                gLogger.error("No type name specified")
-                return
-            # Try to import the type
-            result = self.objectLoader.loadObject(f"DIRAC.AccountingSystem.Client.Types.{typeName}")
-            if not result["OK"]:
-                return result
-            typeClass = result["Value"]
-
-            gLogger.info(f"Loaded type {typeClass.__name__}")
-            typeDef = typeClass().getDefinition()
-            acClient = DataStoreClient()
-            retVal = acClient.registerType(*typeDef)
-            if retVal["OK"]:
-                gLogger.info("Type registered successfully")
-            else:
-                gLogger.error(f"Error: {retVal['Message']}")
-        except Exception:
-            self.showTraceback()
-
-    def do_resetBucketLength(self, args):
-        """
-        Set the bucket Length. Will trigger a recalculation of buckets. Can take a while.
-          Usage : resetBucketLength <typeName>
-          <DIRACRoot>/DIRAC/AccountingSystem/Client/Types/<typeName>
-           should exist and inherit the base type
-        """
-        try:
-            argList = args.split()
-            if argList:
-                typeName = argList[0].strip()
-            else:
-                gLogger.error("No type name specified")
-                return
-
-            # Try to import the type
-            result = self.objectLoader.loadObject(f"DIRAC.AccountingSystem.Client.Types.{typeName}")
-            if not result["OK"]:
-                return result
-            typeClass = result["Value"]
-            gLogger.info(f"Loaded type {typeClass.__name__}")
-            typeDef = typeClass().getDefinition()
-            acClient = DataStoreClient()
-            retVal = acClient.setBucketsLength(typeDef[0], typeDef[3])
-            if retVal["OK"]:
-                gLogger.info("Type registered successfully")
-            else:
-                gLogger.error(f"Error: {retVal['Message']}")
-        except Exception:
-            self.showTraceback()
-
-    def do_regenerateBuckets(self, args):
-        """
-        Regenerate buckets for type. Can take a while.
-          Usage : regenerateBuckets <typeName>
-          <DIRACRoot>/DIRAC/AccountingSystem/Client/Types/<typeName>
-           should exist and inherit the base type
-        """
-        try:
-            argList = args.split()
-            if argList:
-                typeName = argList[0].strip()
-            else:
-                gLogger.error("No type name specified")
-                return
-
-            # Try to import the type
-            result = self.objectLoader.loadObject(f"DIRAC.AccountingSystem.Client.Types.{typeName}")
-            if not result["OK"]:
-                return result
-            typeClass = result["Value"]
-            gLogger.info(f"Loaded type {typeClass.__name__}")
-            typeDef = typeClass().getDefinition()
-            acClient = DataStoreClient()
-            retVal = acClient.regenerateBuckets(typeDef[0])
-            if retVal["OK"]:
-                gLogger.info("Buckets recalculated!")
-            else:
-                gLogger.error(f"Error: {retVal['Message']}")
-        except Exception:
-            self.showTraceback()
-
     def do_showRegisteredTypes(self, args):
         """
         Get a list of registered types
@@ -170,50 +77,3 @@ def do_showRegisteredTypes(self, args):
                 print(" Value fields:\n  %s" % "\n  ".join(typeList[2]))
         except Exception:
             self.showTraceback()
-
-    def do_deleteType(self, args):
-        """
-        Delete a registered accounting type.
-          Usage : deleteType <typeName>
-          WARN! It will delete all data associated to that type! VERY DANGEROUS!
-          If you screw it, you'll discover a new dimension of pain and doom! :)
-        """
-        try:
-            argList = args.split()
-            if argList:
-                typeName = argList[0].strip()
-            else:
-                gLogger.error("No type name specified")
-                return
-
-            choice = input(
-                f"Are you completely sure you want to delete type {typeName} and all it's data? yes/no [no]: "
-            )
-            choice = choice.lower()
-            if choice not in ("yes", "y"):
-                print("Delete aborted")
-                return
-
-            acClient = DataStoreClient()
-            retVal = acClient.deleteType(typeName)
-            if not retVal["OK"]:
-                gLogger.error(f"Error: {retVal['Message']}")
-                return
-            print("Hope you meant it, because it's done")
-        except Exception:
-            self.showTraceback()
-
-    def do_compactBuckets(self, args):
-        """
-        Compact buckets table
-          Usage : compactBuckets
-        """
-        try:
-            acClient = DataStoreClient()
-            retVal = acClient.compactDB()
-            if not retVal["OK"]:
-                gLogger.error(f"Error: {retVal['Message']}")
-                return
-            gLogger.info("Done")
-        except Exception:
-            self.showTraceback()

src/DIRAC/AccountingSystem/Client/DataStoreClient.py

@@ -122,19 +122,6 @@ def delayedCommit(self):
 
         return S_OK()
 
-    def remove(self, register):
-        """
-        Remove a Register from the Accounting DataStore
-        """
-        if not self.__checkBaseType(register.__class__):
-            return S_ERROR("register is not a valid type (has to inherit from BaseAccountingType")
-        retVal = register.checkValues()
-        if not retVal["OK"]:
-            return retVal
-        if gConfig.getValue("/LocalSite/DisableAccounting", False):
-            return S_OK()
-        return self._getRPC().remove(*register.getValues())
-
 
 def _sendToFailover(rpcStub):
     """Create a ForwardDISET operation for failover"""

src/DIRAC/AccountingSystem/Client/Types/BaseAccountingType.py

@@ -161,13 +161,6 @@ def getContents(self):
                 cD[self.fieldsList[iPos]] = self.valuesList[iPos]
         return cD
 
-    def registerToServer(self):
-        """
-        Register type in server
-        """
-        rpcClient = Client(url="Accounting/DataStore")
-        return rpcClient.registerType(*self.getDefinition())
-
     def commit(self):
         """
         Commit register to server

src/DIRAC/AccountingSystem/ConfigTemplate.cfg

@@ -9,11 +9,6 @@ Services
     Authorization
     {
       Default = authenticated
-      compactDB = ServiceAdministrator
-      deleteType = ServiceAdministrator
-      registerType = ServiceAdministrator
-      setBucketsLength = ServiceAdministrator
-      regenerateBuckets = ServiceAdministrator
     }
   }
   ##END

src/DIRAC/AccountingSystem/Service/DataStoreHandler.py

@@ -14,7 +14,6 @@
 from DIRAC import S_ERROR, S_OK
 from DIRAC.AccountingSystem.DB.MultiAccountingDB import MultiAccountingDB
 from DIRAC.ConfigurationSystem.Client import PathFinder
-from DIRAC.Core.Base.Client import Client
 from DIRAC.Core.DISET.RequestHandler import RequestHandler, getServiceOption
 from DIRAC.Core.Utilities import TimeUtilities
 from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
@@ -39,30 +38,6 @@ def initializeHandler(cls, svcInfoDict):
             gThreadScheduler.addPeriodicTask(60, cls.__acDB.loadPendingRecords)
         return S_OK()
 
-    types_registerType = [str, list, list, list]
-
-    def export_registerType(self, typeName, definitionKeyFields, definitionAccountingFields, bucketsLength):
-        """
-        Register a new type. (Only for all powerful admins)
-        """
-        return self.__acDB.registerType(typeName, definitionKeyFields, definitionAccountingFields, bucketsLength)
-
-    types_setBucketsLength = [str, list]
-
-    def export_setBucketsLength(self, typeName, bucketsLength):
-        """
-        Change the buckets Length. (Only for all powerful admins)
-        """
-        return self.__acDB.changeBucketsLength(typeName, bucketsLength)
-
-    types_regenerateBuckets = [str]
-
-    def export_regenerateBuckets(self, typeName):
-        """
-        Recalculate buckets. (Only for all powerful admins)
-        """
-        return self.__acDB.regenerateBuckets(typeName)
-
     types_getRegisteredTypes = []
 
     def export_getRegisteredTypes(self):
@@ -106,51 +81,4 @@ def export_commitRegisters(self, entriesList):
             records.append((entry[0], startTime, endTime, entry[3]))
         return self.__acDB.insertRecordBundleThroughQueue(records)
 
-    types_compactDB = []
-
-    def export_compactDB(self):
-        """
-        Compact the db by grouping buckets
-        """
-        # if we are running workers (not only one service) we can redirect the request to the master
-        # For more information please read the Administrative guide Accounting part!
-        # ADVICE: If you want to trigger the bucketing, please make sure the bucketing is not running!!!!
-        if self.runBucketing:
-            return self.__acDB.compactBuckets()
-
-        return Client(url="Accounting/DataStoreMaster").compactDB()
-
     types_remove = [str, datetime.datetime, datetime.datetime, list]
-
-    def export_remove(self, typeName, startTime, endTime, valuesList):
-        """
-        Remove a record for a type
-        """
-        startTime = int(TimeUtilities.toEpoch(startTime))
-        endTime = int(TimeUtilities.toEpoch(endTime))
-        return self.__acDB.deleteRecord(typeName, startTime, endTime, valuesList)
-
-    types_removeRegisters = [list]
-
-    def export_removeRegisters(self, entriesList):
-        """
-        Remove a record for a type
-        """
-        expectedTypes = [str, datetime.datetime, datetime.datetime, list]
-        for entry in entriesList:
-            if len(entry) != 4:
-                return S_ERROR("Invalid records")
-            for i, en in enumerate(entry):
-                if not isinstance(en, expectedTypes[i]):
-                    return S_ERROR(f"{i} field in the records should be {expectedTypes[i]}")
-        ok = 0
-        for entry in entriesList:
-            startTime = int(TimeUtilities.toEpoch(entry[1]))
-            endTime = int(TimeUtilities.toEpoch(entry[2]))
-            record = entry[3]
-            result = self.__acDB.deleteRecord(entry[0], startTime, endTime, record)
-            if not result["OK"]:
-                return S_OK(ok)
-            ok += 1
-
-        return S_OK(ok)

src/DIRAC/MonitoringSystem/DB/MonitoringDB.py

@@ -447,26 +447,6 @@ def __getRawData(self, typeName, condDict, size=-1):
                 records.append({paramName: getattr(resObj["_source"], paramName) for paramName in paramNames})
             return S_OK(records)
 
-    def getLastDayData(self, typeName, condDict):
-        """
-        It returns the last day data for a given monitoring type.
-
-        :returns: for example
-
-          .. code-block:: python
-
-           {'sort': [{'timestamp': {'order': 'desc'}}],
-            'query': {'bool': {'must': [{'match': {'host': 'dzmathe.cern.ch'}},
-                                        {'match': {'component': 'Bookkeeping_BookkeepingManager'}}]}}}
-
-        :param str typeName: name of the monitoring type
-        :param dict condDict: conditions for the query
-
-                      * key -> name of the field
-                      * value -> list of possible values
-        """
-        return self.__getRawData(typeName, condDict)
-
     def getLimitedData(self, typeName, condDict, size=10):
         """
         Returns a list of records for a given selection.

src/DIRAC/MonitoringSystem/Service/MonitoringHandler.py

@@ -244,24 +244,6 @@ def export_getReport(self, reportRequest):
         reportRequest["generatePlot"] = False
         return reporter.generate(reportRequest)
 
-    types_addMonitoringRecords = [str, list]
-
-    def export_addMonitoringRecords(self, monitoringtype, data):
-        """
-        Bulk insert data directly to the given monitoring type.
-
-        :param str monitoringtype: monitoring type name
-        :param list data: list of documents
-        :returns: S_OK or S_ERROR
-        """
-
-        retVal = self.__db.getIndexName(monitoringtype)
-        if not retVal["OK"]:
-            return retVal
-        prefix = retVal["Value"]
-        gLogger.debug("addMonitoringRecords:", prefix)
-        return self.__db.bulk_index(prefix, data)
-
     types_addRecords = [str, str, list]
 
     def export_addRecords(self, indexname, monitoringType, data):
@@ -290,21 +272,6 @@ def export_deleteIndex(self, indexName):
         gLogger.debug("delete index:", indexName)
         return self.__db.deleteIndex(indexName)
 
-    types_getLastDayData = [str, dict]
-
-    def export_getLastDayData(self, typeName, condDict):
-        """
-        It returns the data from the last day index. Note: we create daily indexes.
-
-        :param str typeName: name of the monitoring type
-        :param dict condDict: conditions for the query
-
-                       * key -> name of the field
-                       * value -> list of possible values
-        """
-
-        return self.__db.getLastDayData(typeName, condDict)
-
     types_getLimitedDat = [str, dict, int]
 
     def export_getLimitedData(self, typeName, condDict, size):

tests/Integration/AccountingSystem/Test_DataStoreClient.py

@@ -28,6 +28,3 @@ def test_addAndRemoveStorageOccupancy():
     assert res["OK"], res["Message"]
     res = gDataStoreClient.commit()
     assert res["OK"], res["Message"]
-    # now removing that record
-    res = gDataStoreClient.remove(record)
-    assert res["OK"], res["Message"]

tests/Integration/AccountingSystem/Test_ReportsClient.py

@@ -49,7 +49,3 @@ def test_addAndRemoveStorageOccupancy():
         "StorageElement",
     )
     assert res["OK"], res["Message"]
-
-    # now removing that record
-    res = gDataStoreClient.remove(record)
-    assert res["OK"], res["Message"]