[v6r22] Pilot Submission statistics is counted by SiteDirector based on Accounting System

AccountingSystem/Client/Types/PilotSubmission.py

@@ -0,0 +1,32 @@
+''' Accounting Type for Pilot Submission
+'''
+
+from DIRAC.AccountingSystem.Client.Types.BaseAccountingType import BaseAccountingType
+
+__RCSID__ = "$Id$"
+
+
+class PilotSubmission(BaseAccountingType):
+  ''' Accounting Type class for Pilot Submission
+  '''
+
+  def __init__(self):
+    BaseAccountingType.__init__(self)
+
+    self.definitionKeyFields = [('HostName', 'VARCHAR(100)'),
+                                ('SiteDirector', 'VARCHAR(100)'),
+                                ('Site', 'VARCHAR(100)'),
+                                ('CE', 'VARCHAR(100)'),
+                                ('Queue', 'VARCHAR(100)'),
+                                ('Status', 'VARCHAR(100)')]
+    self.definitionAccountingFields = [('NumTotal', "INT UNSIGNED"),
+                                       ('NumSucceeded', 'INT UNSIGNED')]
+
+    self.bucketsLength = [(86400 * 2, 900),         # <2d = 15m
+                          (86400 * 10, 9000),       # <10d = 2.5h
+                          (86400 * 35, 18000),      # <35d = 5h
+                          (86400 * 30 * 6, 86400),  # >5d <6m = 1d
+                          (86400 * 600, 604800)]    # >6m = 1w
+
+    self.dataTimespan = 86400 * 30 * 14  # Only keep the last 14 months of data
+    self.checkType()

AccountingSystem/private/Plotters/PilotSubmissionPlotter.py

@@ -0,0 +1,123 @@
+
+from DIRAC import S_OK, S_ERROR
+from DIRAC.AccountingSystem.private.Plotters.BaseReporter import BaseReporter
+from DIRAC.AccountingSystem.Client.Types.PilotSubmission import PilotSubmission
+
+__RCSID__ = "$Id$"
+
+
+class PilotSubmissionPlotter(BaseReporter):
+  '''
+  Plotter for the Pilot Submission Accounting
+  '''
+
+  _typeName = "PilotSubmission"
+  _typeKeyFields = [dF[0] for dF in PilotSubmission().definitionKeyFields]
+
+  def _reportSubmission(self, reportRequest):
+    '''
+    Get data for Submission plot
+
+    :param dict reportRequest: Condition to select data
+    '''
+
+    selectFields = (self._getSelectStringForGrouping(reportRequest['groupingFields']) + ", %s, %s, SUM(%s)",
+                    reportRequest['groupingFields'][1] + ['startTime', 'bucketLength', 'NumTotal'])
+    retVal = self._getTimedData(reportRequest['startTime'],
+                                reportRequest['endTime'],
+                                selectFields,
+                                reportRequest['condDict'],
+                                reportRequest['groupingFields'],
+                                {})
+    if not retVal['OK']:
+      return retVal
+    dataDict, granularity = retVal['Value']
+    self.stripDataField(dataDict, 0)
+    dataDict, _ = self._divideByFactor(dataDict, granularity)
+    dataDict = self._fillWithZero(granularity, reportRequest['startTime'], reportRequest['endTime'], dataDict)
+    baseDataDict, graphDataDict, _, unit = self._findSuitableRateUnit(dataDict,
+                                                                      self._getAccumulationMaxValue(dataDict),
+                                                                      "jobs")
+    return S_OK({'data': baseDataDict, 'graphDataDict': graphDataDict,
+                 'granularity': granularity, 'unit': unit})
+
+  def _plotSubmission(self, reportRequest, plotInfo, filename):
+    '''
+    Make 1 dimensional pilotSubmission plot
+
+    :param dict reportRequest: Condition to select data
+    :param dict plotInfo: Data for plot
+    :param str  filename: File name
+    '''
+
+    metadata = {'title': 'Number of Submission by %s' % reportRequest['grouping'],
+                'starttime': reportRequest['startTime'],
+                'endtime': reportRequest['endTime'],
+                'span': plotInfo['granularity'],
+                'ylabel': plotInfo['unit']}
+    return self._generateTimedStackedBarPlot(filename, plotInfo['graphDataDict'], metadata)
+
+  _reportSubmissionEfficiencyName = "Submission efficiency"
+
+  def _reportSubmissionEfficiency(self, reportRequest):
+    '''
+    Get data for Submission Efficiency plot
+
+    :param dict reportRequest: Condition to select data
+    '''
+
+    selectFields = (self._getSelectStringForGrouping(reportRequest['groupingFields']) + ", %s, %s, SUM(%s), SUM(%s)",
+                    reportRequest['groupingFields'][1] + ['startTime', 'bucketLength',
+                                                          'NumTotal', 'NumSucceeded'])
+
+    retVal = self._getTimedData(reportRequest['startTime'],
+                                reportRequest['endTime'],
+                                selectFields,
+                                reportRequest['condDict'],
+                                reportRequest['groupingFields'],
+                                {'checkNone': True,
+                                 'convertToGranularity': 'sum',
+                                 'calculateProportionalGauges': False,
+                                 'consolidationFunction': self._efficiencyConsolidation})
+    if not retVal['OK']:
+      return retVal
+    dataDict, granularity = retVal['Value']
+    self.stripDataField(dataDict, 0)
+    if len(dataDict) > 1:
+      # Get the total for the plot
+      selectFields = ("'Total', %s, %s, SUM(%s),SUM(%s)",
+                      ['startTime', 'bucketLength',
+                       'NumSucceeded', 'NumTotal'])
+
+      retVal = self._getTimedData(reportRequest['startTime'],
+                                  reportRequest['endTime'],
+                                  selectFields,
+                                  reportRequest['condDict'],
+                                  reportRequest['groupingFields'],
+                                  {'scheckNone': True,
+                                   'convertToGranularity': 'sum',
+                                   'calculateProportionalGauges': False,
+                                   'consolidationFunction': self._efficiencyConsolidation})
+      if not retVal['OK']:
+        return retVal
+      totalDict = retVal['Value'][0]
+      self.stripDataField(totalDict, 0)
+      for key in totalDict:
+        dataDict[key] = totalDict[key]
+    return S_OK({'data': dataDict, 'granularity': granularity})
+
+  def _plotSubmissionEfficiency(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': 'Pilot Submission efficiency by %s' % reportRequest['grouping'],
+                'starttime': reportRequest['startTime'],
+                'endtime': reportRequest['endTime'],
+                'span': plotInfo['granularity']}
+
+    return self._generateQualityPlot(filename, plotInfo['data'], metadata)

WorkloadManagementSystem/Agent/SiteDirector.py

@@ -34,6 +34,7 @@
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.FrameworkSystem.Client.ProxyManagerClient import gProxyManager
 from DIRAC.AccountingSystem.Client.Types.Pilot import Pilot as PilotAccounting
+from DIRAC.AccountingSystem.Client.Types.PilotSubmission import PilotSubmission as PilotSubmissionAccounting
 from DIRAC.AccountingSystem.Client.DataStoreClient import gDataStoreClient
 from DIRAC.WorkloadManagementSystem.Client.MatcherClient import MatcherClient
 from DIRAC.WorkloadManagementSystem.Client.ServerUtils import pilotAgentsDB
@@ -116,6 +117,7 @@ def __init__(self, *args, **kwargs):
     self.updateStatus = True
     self.getOutput = False
     self.sendAccounting = True
+    self.sendSubmissionAccounting = True
 
     self.pilot3 = False
     self.pilotFiles = []  # files whose content will be compressed and sent together with the pilot wrapper
@@ -214,6 +216,7 @@ def beginExecution(self):
     self.updateStatus = self.am_getOption('UpdatePilotStatus', self.updateStatus)
     self.getOutput = self.am_getOption('GetPilotOutput', self.getOutput)
     self.sendAccounting = self.am_getOption('SendPilotAccounting', self.sendAccounting)
+    self.sendSubmissionAccounting = self.am_getOption('SendPilotSubmissionAccounting', self.sendSubmissionAccounting)
 
     # Get the site description dictionary
     siteNames = None
@@ -257,6 +260,8 @@ def beginExecution(self):
       self.log.always('Pilot output retrieval requested')
     if self.sendAccounting:
       self.log.always('Pilot accounting sending requested')
+    if self.sendSubmissionAccounting:
+      self.log.always('Pilot submission accounting sending requested')
 
     self.log.always('MaxPilotsToSubmit:', self.maxPilotsToSubmit)
     self.log.always('MaxJobsInFillMode:', self.maxJobsInFillMode)
@@ -850,6 +855,15 @@ def _submitPilotsToQueue(self, pilotsToSubmit, ce, queue):
     if not submitResult['OK']:
       self.log.error("Failed submission to queue",
                      "Queue %s:\n, %s" % (queue, submitResult['Message']))
+
+      if self.sendSubmissionAccounting:
+        self.sendPilotSubmissionAccounting(self.queueDict[queue]['Site'],
+                                           self.queueDict[queue]['CEName'],
+                                           self.queueDict[queue]['QueueName'],
+                                           pilotsToSubmit,
+                                           0,
+                                           'Failed')
+
       pilotsToSubmit = 0
       self.failedQueues[queue] += 1
       return submitResult
@@ -863,6 +877,13 @@ def _submitPilotsToQueue(self, pilotsToSubmit, ce, queue):
                                                     self.queueDict[queue]['QueueName'],
                                                     self.queueDict[queue]['CEName']))
     stampDict = submitResult.get('PilotStampDict', {})
+    if self.sendSubmissionAccounting:
+      self.sendPilotSubmissionAccounting(self.queueDict[queue]['Site'],
+                                         self.queueDict[queue]['CEName'],
+                                         self.queueDict[queue]['QueueName'],
+                                         len(pilotList),
+                                         len(pilotList),
+                                         'Succeeded')
 
     return S_OK((pilotsToSubmit, pilotList, stampDict))
 
@@ -1454,3 +1475,47 @@ def sendPilotAccounting(self, pilotDict):
       return result
 
     return S_OK()
+
+  def sendPilotSubmissionAccounting(self,
+                                    siteName,
+                                    ceName,
+                                    queueName,
+                                    numTotal,
+                                    numSucceeded,
+                                    status):
+
+    """ Send pilot submission accounting record
+        :param str siteName:     Site name
+        :param str ceName:       CE name
+        :param str queueName:    queue Name
+        :param int numTotal:     Total number of submission
+        :param int numSucceeded: Total number of submission succeeded
+        :param str status:       'Succeeded' or 'Failed'
+    """
+
+    pA = PilotSubmissionAccounting()
+    pA.setStartTime(dateTime())
+    pA.setEndTime(dateTime())
+    pA.setValueByKey('HostName', DIRAC.siteName())
+    if hasattr(self, "_AgentModule__moduleProperties"):
+      pA.setValueByKey('SiteDirector', self.am_getModuleParam('agentName'))
+    else:  # In case it is not executed as agent
+      pA.setValueByKey('SiteDirector', 'Client')
+
+    pA.setValueByKey('Site', siteName)
+    pA.setValueByKey('CE', ceName)
+    pA.setValueByKey('Queue', queueName)
+    pA.setValueByKey('Status', status)
+    pA.setValueByKey('NumTotal', numTotal)
+    pA.setValueByKey('NumSucceeded', numSucceeded)
+    result = gDataStoreClient.addRegister(pA)
+
+    if not result['OK']:
+      self.log.warn('Error in add Register:' + result['Message'])
+      return result
+
+    self.log.verbose("Begin commit")
+    result = gDataStoreClient.delayedCommit()
+    if not result['OK']:
+      self.log.error('Error in Commit:' + result['Message'])
+      return result

WorkloadManagementSystem/ConfigTemplate.cfg

@@ -213,8 +213,10 @@ Agents
     UpdatePilotStatus = True
     # Boolean value used to indicate if the pilot output will be or not retrieved
     GetPilotOutput = False
-    # Boolean value than indicates if the pilot job will send information for accounting
+    # Boolean value that indicates if the pilot job will send information for accounting
     SendPilotAccounting = True
+    # Boolean value that indicates if the pilot submission statistics will be sended for accounting
+    SendPilotSubmissionAccounting = True
   }
   ##END
   MultiProcessorSiteDirector