diff --git a/AccountingSystem/Client/Types/PilotSubmission.py b/AccountingSystem/Client/Types/PilotSubmission.py new file mode 100644 index 00000000000..021ea1d0568 --- /dev/null +++ b/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() diff --git a/AccountingSystem/private/Plotters/PilotSubmissionPlotter.py b/AccountingSystem/private/Plotters/PilotSubmissionPlotter.py new file mode 100644 index 00000000000..433c6aebb5d --- /dev/null +++ b/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) diff --git a/WorkloadManagementSystem/Agent/SiteDirector.py b/WorkloadManagementSystem/Agent/SiteDirector.py index a1f30374bbd..c57847274c7 100644 --- a/WorkloadManagementSystem/Agent/SiteDirector.py +++ b/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 diff --git a/WorkloadManagementSystem/ConfigTemplate.cfg b/WorkloadManagementSystem/ConfigTemplate.cfg index 4fa7ca951ac..526f9798ec7 100644 --- a/WorkloadManagementSystem/ConfigTemplate.cfg +++ b/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