Merge pull request #5955 from rupozzi/dataop

[integration] Hackaton fix: DataOperation and PilotSubmission Monitoring issues

Author: fstagni

src/DIRAC/DataManagementSystem/Agent/FTS3Agent.py

@@ -649,7 +649,7 @@ def execute(self):
         return S_OK()
 
     def endExecution(self):
-        self.dataOpSender.concludeSending()
+        return self.dataOpSender.concludeSending()
 
     def __sendAccounting(self, ftsJob):
 

src/DIRAC/DataManagementSystem/Client/DataManager.py

@@ -1877,4 +1877,4 @@ def getReplica(self, lfn, storageElementName, localPath=False):
         return self.__executeIfReplicaExists(storageElementName, lfn, "getFile", localPath=localPath)
 
     def __del__(self):
-        self.dataOpSender.concludeSending()
+        return self.dataOpSender.concludeSending()

src/DIRAC/MonitoringSystem/Client/DataOperationSender.py

@@ -41,14 +41,15 @@ def sendData(self, baseDict, commitFlag=False, delayedCommit=False, startTime=Fa
         """
         if "Monitoring" in self.monitoringOption:
             baseDict["ExecutionSite"] = DIRAC.siteName()
+            baseDict["Channel"] = baseDict["Source"] + "->" + baseDict["Destination"]
             self.dataOperationReporter.addRecord(baseDict)
             if commitFlag or delayedCommit:
                 result = self.dataOperationReporter.commit()
                 sLog.debug("Committing data operation to monitoring")
                 if not result["OK"]:
                     sLog.error("Could not commit data operation to monitoring", result["Message"])
-                    return result
-                sLog.debug("Done committing to monitoring")
+                else:
+                    sLog.debug("Done committing to monitoring")
 
         if "Accounting" in self.monitoringOption:
             self.dataOp.setValuesFromDict(baseDict)
@@ -80,13 +81,20 @@ def sendData(self, baseDict, commitFlag=False, delayedCommit=False, startTime=Fa
 
         return S_OK()
 
-    # Call this method in order to commit all records added but not yet committed to Accounting
+    # Call this method in order to commit all records added but not yet committed to Accounting and Monitoring
     def concludeSending(self):
+        if "Monitoring" in self.monitoringOption:
+            result = self.dataOperationReporter.commit()
+            sLog.debug("Committing data operation to monitoring")
+            if not result["OK"]:
+                sLog.error("Could not commit data operation to monitoring", result["Message"])
+            else:
+                sLog.debug("Done committing to monitoring")
         if "Accounting" in self.monitoringOption:
             result = gDataStoreClient.commit()
             sLog.debug("Concluding the sending and committing data operation to accounting")
             if not result["OK"]:
                 sLog.error("Could not commit data operation to accounting", result["Message"])
                 return result
-        sLog.debug("Done committing to accounting")
-        return S_OK()
+            sLog.debug("Done committing to accounting")
+        return result

src/DIRAC/MonitoringSystem/Client/Types/DataOperation.py

@@ -18,6 +18,7 @@ def __init__(self):
             "Destination",
             "Protocol",
             "FinalStatus",
+            "Channel",
         ]
 
         self.monitoringFields = [

src/DIRAC/MonitoringSystem/Client/Types/PilotSubmissionMonitoring.py

@@ -1,5 +1,5 @@
 """
-Monitoring Type for Pilot Submission
+Monitoring Type for Pilot Submission.
 """
 
 from DIRAC.MonitoringSystem.Client.Types.BaseType import BaseType
@@ -32,7 +32,7 @@ def __init__(self):
                 "Queue": {"type": "keyword"},
                 "Status": {"type": "keyword"},
                 "NumTotal": {"type": "long"},
-                "NumSucceded": {"type": "long"},
+                "NumSucceeded": {"type": "long"},
             }
         )
 

src/DIRAC/MonitoringSystem/Service/MonitoringHandler.py

@@ -109,7 +109,7 @@ def transfer_toClient(self, fileId, token, fileHelper):
         if len(fileId) > 5 and fileId[1] == ":":
             gLogger.info("Seems the file request is a plot generation request!")
             try:
-                result = self.__generatePlotFromFileId(fileId)
+                result = self._generatePlotFromFileId(fileId)
             except Exception as e:  # pylint: disable=broad-except
                 gLogger.exception("Exception while generating plot", str(e))
                 result = S_ERROR("Error while generating plot: %s" % str(e))

src/DIRAC/MonitoringSystem/private/DBUtils.py

@@ -176,3 +176,47 @@ def _accumulate(granularity, startEpoch, endEpoch, dataDict):
                     lastValue += currentDict[timeEpoch]
                 currentDict[timeEpoch] = lastValue
         return dataDict
+
+    def stripDataField(self, dataDict, fieldId):
+        """
+        Strip <fieldId> data and sum the rest as it was data from one key
+
+        :param dict dataDict: dictionary of the form::
+
+            { 'key' : { <timeEpoch1>: [1, 2, 3],
+                        <timeEpoch2>: [3, 4, 5].. } }
+
+          The dataDict is modified in this function and the return structure is:
+
+          .. code-block :: python
+
+                  dataDict : { 'key' : { <timeEpoch1>: 1,
+                                         <timeEpoch2>: 3.. } }
+
+        :param int fieldId:
+
+        :returns: list of dictionaries
+
+          .. code-block:: python
+
+                     [ { <timeEpoch1>: 2, <timeEpoch2>: 4... }
+                       { <timeEpoch1>: 3, <timeEpoch2>): 5... } ]
+
+        :rtype: python:list
+
+        """
+        remainingData = [{}]  # Hack for empty data
+        for key in dataDict:
+            for timestamp in dataDict[key]:
+                for iPos in dataDict[key][timestamp]:
+                    remainingData.append({})
+                break
+            break
+        for key in dataDict:
+            for timestamp in dataDict[key]:
+                strippedField = float(dataDict[key][timestamp].pop(fieldId))
+                for iPos, value in enumerate(dataDict[key][timestamp]):
+                    remainingData[iPos].setdefault(timestamp, 0.0)
+                    remainingData[iPos] += float(value)
+                dataDict[key][timestamp] = strippedField
+        return remainingData

src/DIRAC/MonitoringSystem/private/Plotters/BasePlotter.py

@@ -1,5 +1,5 @@
 """
-It used to create several plots
+It is used to create several plots
 """
 import time
 import copy
@@ -107,7 +107,7 @@ def __init__(self, db, setup, extraArgs=None):
 
     def generate(self, reportRequest):
         """
-        It retrives the data from the database and create the plot
+        It retrives the data from the database and creates the plot
 
         :param dict reportRequest: contains the plot attributes
         """
@@ -347,7 +347,7 @@ def __checkThumbnailMetadata(self, metadata):
         return False
 
     def __plotData(self, filename, dataDict, metadata, funcToPlot):
-        """It create the plot.
+        """It creates the plot.
 
         :param str filename: the name of the file which contains the plot
         :param dict dataDict: data used to crate the plot
@@ -379,7 +379,7 @@ def _generateTimedStackedBarPlot(self, filename, dataDict, metadata):
 
     def _generateQualityPlot(self, filename, dataDict, metadata):
         """
-        it create a quality plot
+        it creates a quality plot
         """
         return self.__plotData(filename, dataDict, metadata, generateQualityPlot)
 
@@ -397,7 +397,7 @@ def _generatePiePlot(self, filename, dataDict, metadata):
 
     def _generateStackedLinePlot(self, filename, dataDict, metadata):
         """
-        It create a stacked lien plot
+        It creates a stacked lien plot
         """
         return self.__plotData(filename, dataDict, metadata, generateStackedLinePlot)
 
@@ -413,3 +413,28 @@ def _fillWithZero(self, granularity, startEpoch, endEpoch, dataDict):
                 if timeEpoch not in currentDict:
                     currentDict[timeEpoch] = 0
         return dataDict
+
+    def _calculateEfficiencyDict(self, totDataDict, dataDict):
+        """
+        It returns a dict with efficiency calculated from each key in totDataDict and dataDict
+        """
+        for item, val in dataDict.items():
+            totVal = totDataDict.get(item, {})
+            try:
+                dataDict[item] = {key: (float(val[key]) / float(totVal[key])) * 100 for key in val if key in totVal}
+            except (ZeroDivisionError, TypeError):
+                gLogger.warn("Error in ", val)
+                gLogger.warn("Dividing by zero or using None type field. Skipping the key of this dict item...")
+        return dataDict
+
+    def _sumDictValues(self, dataDict):
+        """
+        Sums the values of each item in `dataDict`.
+        Returns the dictionary with the same keys and the values replaced by their sum.
+        """
+        for key, values in dataDict.items():
+            sum = 0
+            for val in values.values():
+                sum += val
+            dataDict[key] = sum
+        return dataDict