feat: Adding channel to monitoring records and fixed the plotters

Author: rupozzi

src/DIRAC/MonitoringSystem/Client/DataOperationSender.py

@@ -4,6 +4,7 @@
 
 """
 
+from email.mime import base
 import DIRAC
 from DIRAC import S_OK, gLogger
 
@@ -41,6 +42,7 @@ 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()
@@ -80,13 +82,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/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/Plotters/BasePlotter.py

@@ -1,5 +1,5 @@
 """
-It used to create several plots
+It is used to creates 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,26 @@ 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 (totItem, totVal), (item, val) in zip(totDataDict.items(), dataDict.items()):
+            try:
+                dataDict[item] = {key: (float(val[key]) / float(totVal[key])) * 100 for key in val if key in totVal}
+            except ZeroDivisionError and TypeError:
+                gLogger.warn("Dividing by zero or using None type field. Skipping the key of this dict item...")
+        return dataDict
+
+    def _sumDictValues(self, dataDict):
+        """
+        Sums all the values of the keys of each item in the dict.
+        Returns a dict with item-value pair and removes the keys.
+        """
+        for item, key in dataDict.items():
+            sum = 0
+            for val in key:
+                sum += key[val]
+            dataDict[item] = sum
+        return dataDict

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

@@ -1,5 +1,5 @@
 """
-This class is used to define the plot using the plot attributes.
+this class is used to define the plot using the plot attributes.
 """
 
 from DIRAC import S_OK
@@ -19,6 +19,7 @@ class DataOperationPlotter(BasePlotter):
     """
 
     _typeName = "DataOperation"
+
     _typeKeyFields = DataOperation().keyFields
 
     _reportSuceededTransfersName = "Successful transfers"
@@ -32,6 +33,11 @@ def _reportFailedTransfers(self, reportRequest):
         return self.__reportTransfers(reportRequest, "Failed", ("Succeeded", 1))
 
     def __reportTransfers(self, reportRequest, titleType, togetherFieldsToPlot):
+        """It is used to retrieve the data from the database.
+
+        :param dict reportRequest: contains attributes used to create the plot.
+        :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
+        """
         retVal = self._getTimedData(
             reportRequest["startTime"],
             reportRequest["endTime"],
@@ -68,38 +74,51 @@ def __plotTransfers(self, reportRequest, plotInfo, filename, titleType, together
         }
         return self._generateTimedStackedBarPlot(filename, plotInfo["graphDataDict"], metadata)
 
-    _reportQualityName = "Efficiency by protocol"
+    _reportQualityName = "Transfer Efficiency"
 
     def _reportQuality(self, reportRequest):
-        selectFields = ["TransferOK", "TransferTotal"]
+        """It is used to retrieve the data from the database.
+
+        :param dict reportRequest: contains attributes used to create the plot.
+        :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
+        """
+        # Retrieve the number of succeded transfers
         retVal = self._getTimedData(
             reportRequest["startTime"],
             reportRequest["endTime"],
-            selectFields,
+            "TransferOK",
+            preCondDict=reportRequest["condDict"],
+            metadataDict=None,
+        )
+        # Retrieve the number of total transfers
+        retTotVal = self._getTimedData(
+            reportRequest["startTime"],
+            reportRequest["endTime"],
+            "TransferTotal",
             preCondDict=reportRequest["condDict"],
             metadataDict=None,
         )
         if not retVal["OK"]:
             return retVal
+        if not retTotVal["OK"]:
+            return retTotVal
         dataDict, granularity = retVal["Value"]
-        if len(dataDict) > 1:
-            # Get the total for the plot
-            selectFields = ["TransferOK", "TransferTotal"]
-            retVal = self._getTimedData(
-                reportRequest["startTime"],
-                reportRequest["endTime"],
-                selectFields,
-                preCondDict=reportRequest["condDict"],
-                metadataDict=None,
-            )
-            if not retVal["OK"]:
-                return retVal
-            totalDict = retVal["Value"][0]
-            for key in totalDict:
-                dataDict[key] = totalDict[key]
-        return S_OK({"data": dataDict, "granularity": granularity})
+        totDataDict, granularity = retTotVal["Value"]
+        # Check that the dicts are not empty
+        if bool(dataDict) and bool(totDataDict):
+            # Return the efficiency in dataDict
+            effDict = self._calculateEfficiencyDict(totDataDict, dataDict)
+        return S_OK({"data": effDict, "granularity": granularity})
 
     def _plotQuality(self, reportRequest, plotInfo, filename):
+        """
+        Make 2 dimensional pilotSubmission efficiency plot
+
+        :param dict reportRequest: Condition to select data
+        :param dict plotInfo: Data for plot.
+        :param str  filename: File name
+        """
+
         metadata = {
             "title": "Transfer quality by %s" % reportRequest["grouping"],
             "starttime": reportRequest["startTime"],
@@ -111,6 +130,11 @@ def _plotQuality(self, reportRequest, plotInfo, filename):
     _reportTransferedDataName = "Cumulative transferred data"
 
     def _reportTransferedData(self, reportRequest):
+        """It is used to retrieve the data from the database.
+
+        :param dict reportRequest: contains attributes used to create the plot.
+        :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
+        """
         retVal = self._getTimedData(
             reportRequest["startTime"],
             reportRequest["endTime"],
@@ -131,6 +155,13 @@ def _reportTransferedData(self, reportRequest):
         )
 
     def _plotTransferedData(self, reportRequest, plotInfo, filename):
+        """It creates the plot.
+
+        :param dict reportRequest: plot attributes
+        :param dict plotInfo: contains all the data which are used to create the plot
+        :param str filename:
+        :return: S_OK or S_ERROR { 'plot' : value1, 'thumbnail' : value2 } value1 and value2 are TRUE/FALSE
+        """
         metadata = {
             "title": "Transfered data by %s" % reportRequest["grouping"],
             "starttime": reportRequest["startTime"],
@@ -142,6 +173,11 @@ def _plotTransferedData(self, reportRequest, plotInfo, filename):
         return self._generateCumulativePlot(filename, plotInfo["graphDataDict"], metadata)
 
     def _reportThroughput(self, reportRequest):
+        """It is used to retrieve the data from the database.
+
+        :param dict reportRequest: contains attributes used to create the plot.
+        :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
+        """
         retVal = self._getTimedData(
             reportRequest["startTime"],
             reportRequest["endTime"],
@@ -162,6 +198,13 @@ def _reportThroughput(self, reportRequest):
         )
 
     def _plotThroughput(self, reportRequest, plotInfo, filename):
+        """It creates the plot.
+
+        :param dict reportRequest: plot attributes
+        :param dict plotInfo: contains all the data which are used to create the plot
+        :param str filename:
+        :return: S_OK or S_ERROR { 'plot' : value1, 'thumbnail' : value2 } value1 and value2 are TRUE/FALSE
+        """
         metadata = {
             "title": "Throughput by %s" % reportRequest["grouping"],
             "ylabel": plotInfo["unit"],
@@ -174,21 +217,34 @@ def _plotThroughput(self, reportRequest, plotInfo, filename):
     _reportDataTransferedName = "Pie chart of transferred data"
 
     def _reportDataTransfered(self, reportRequest):
-        retVal = self._getSummaryData(
+        """It is used to retrieve the data from the database.
+
+        :param dict reportRequest: contains attributes used to create the plot.
+        :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
+        """
+        retVal = self._getTimedData(
             reportRequest["startTime"],
             reportRequest["endTime"],
-            "Transfersize",
+            "TransferSize",
             preCondDict=reportRequest["condDict"],
             metadataDict=None,
         )
         if not retVal["OK"]:
             return retVal
-        dataDict = retVal["Value"]
+        dataDict, granularity = retVal["Value"]
+        dataDict = self._sumDictValues(dataDict)
         for key in dataDict:
             dataDict[key] = int(dataDict[key])
         return S_OK({"data": dataDict})
 
     def _plotDataTransfered(self, reportRequest, plotInfo, filename):
+        """It creates the plot.
+
+        :param dict reportRequest: plot attributes
+        :param dict plotInfo: contains all the data which are used to create the plot
+        :param str filename:
+        :return: S_OK or S_ERROR { 'plot' : value1, 'thumbnail' : value2 } value1 and value2 are TRUE/FALSE
+        """
         metadata = {
             "title": "Total data transfered by %s" % reportRequest["grouping"],
             "ylabel": "bytes",

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

@@ -9,7 +9,6 @@
 
 
 class PilotSubmissionMonitoringPlotter(BasePlotter):
-
     """
     .. class:: PilotSubmissionMonitoringPlotter
 
@@ -22,6 +21,8 @@ class PilotSubmissionMonitoringPlotter(BasePlotter):
     _typeName = "PilotSubmissionMonitoring"
     _typeKeyFields = PilotSubmissionMonitoring().keyFields
 
+    _reportNumberOfSubmissions = "Total Number of Submission"
+
     def _reportNumberOfSubmissions(self, reportRequest):
         """It is used to retrieve the data from the database.
 
@@ -66,12 +67,15 @@ def _plotNumberOfSubmissions(self, reportRequest, plotInfo, filename):
 
         return self._generateStackedLinePlot(filename=filename, dataDict=plotInfo["data"], metadata=metadata)
 
+    _reportNumSucceededName = "Submission Efficiency"
+
     def _reportNumSucceeded(self, reportRequest):
         """It is used to retrieve the data from the database.
 
         :param dict reportRequest: contains attributes used to create the plot.
         :return: S_OK or S_ERROR {'data':value1, 'granularity':value2} value1 is a dictionary, value2 is the bucket length
         """
+        # Retrieve the number of succeded submissions
         retVal = self._getTimedData(
             startTime=reportRequest["startTime"],
             endTime=reportRequest["endTime"],
@@ -81,8 +85,25 @@ def _reportNumSucceeded(self, reportRequest):
         )
         if not retVal["OK"]:
             return retVal
+
+        # Retrieve the number of total submissions
+        retTotVal = self._getTimedData(
+            startTime=reportRequest["startTime"],
+            endTime=reportRequest["endTime"],
+            selectField="NumTotal",
+            preCondDict=reportRequest["condDict"],
+            metadataDict=None,
+        )
+        if not retTotVal["OK"]:
+            return retTotVal
+
         dataDict, granularity = retVal["Value"]
-        return S_OK({"data": dataDict, "granularity": granularity})
+        totDataDict, granularity = retTotVal["Value"]
+        # Check that the dicts are not empty
+        if bool(dataDict) and bool(totDataDict):
+            # Retrun the efficiency in dataDict
+            effDict = self._calculateEfficiencyDict(totDataDict, dataDict)
+        return S_OK({"data": effDict, "granularity": granularity})
 
     def _plotNumSucceeded(self, reportRequest, plotInfo, filename):
         """
@@ -92,12 +113,10 @@ def _plotNumSucceeded(self, reportRequest, plotInfo, filename):
         :param dict plotInfo: Data for plot.
         :param str  filename: File name
         """
-
         metadata = {
             "title": "Pilot Submission efficiency by %s" % reportRequest["grouping"],
             "starttime": reportRequest["startTime"],
             "endtime": reportRequest["endTime"],
             "span": plotInfo["granularity"],
         }
-
         return self._generateQualityPlot(filename, plotInfo["data"], metadata)