CRAB/python/parseCrabFjr.py

#!/usr/bin/env python

import sys, getopt, string

from ProdCommon.FwkJobRep.ReportParser import readJobReport
from DashboardAPI import apmonSend, apmonFree


def main(argv) :
    """
    parseCrabFjr

    - parse CRAB FrameworkJobReport on WN: { 'protocol' : { 'action' : [attempted,succedeed,total-size,total-time,min-time,max-time] , ... } , ... }
    - report parameters to DashBoard using DashBoardApi.py: for all 'read' actions of all protocols, report MBPS
    - return ExitStatus and dump of DashBoard report separated by semi-colon to WN wrapper script

    required parameters:
    --input            :       input FJR xml file
    --MonitorID        :       DashBoard MonitorID
    --MonitorJobID     :       DashBoard MonitorJobID

    optional parameters:
    --help             :       help
    --debug            :       debug statements
    
    """

    # defaults
    input = ''
    MonitorID = ''
    MonitorJobID = ''
    debug = 0

    try:
        opts, args = getopt.getopt(argv, "", ["input=", "MonitorID=", "MonitorJobID=", "debug", "help"])
    except getopt.GetoptError:
        print main.__doc__
        sys.exit(2)

    # check command line parameter
    for opt, arg in opts :
        if opt  == "--help" :
            print main.__doc__
            sys.exit()
        elif opt == "--input" :
            input = arg
        elif opt == "--MonitorID" :
            MonitorID = arg
        elif opt == "--MonitorJobID" :
            MonitorJobID = arg
        elif opt == "--debug" :
            debug = 1
            
    if input == '' or MonitorID == '' or MonitorJobID == '':
        print main.__doc__
        sys.exit()

    # load FwkJobRep
    jobReport = readJobReport(input)[0]

    exit_status = ''
    
    ##### temporary fix for FJR incomplete ####
    fjr = open (input)
    len_fjr = len(fjr.readlines())
    if (len_fjr <= 6):
       ### 50115 - cmsRun did not produce a valid/readable job report at runtime
       exit_status = str(50115)
    else: 
        # get ExitStatus of last error
        if len(jobReport.errors) != 0 :
            exit_status = str(jobReport.errors[-1]['ExitStatus'])
        else :
            exit_status = str(0)

    # get i/o statistics
    storageStatistics = str(jobReport.storageStatistics)

    # dashboard report dictionary
    dashboard_report = {}

##Brian's patch to sent number of events procedded to the Dashboard 
    # Add NoEventsPerRun to the Dashboard report
    eventsPerRun = 0
    for inputFile in jobReport.inputFiles:
        try:
            eventsRead = str(inputFile.get('EventsRead', 0))
            eventsRead = int(eventsRead.strip())
        except:
            continue
        eventsPerRun += eventsRead
    dashboard_report['NoEventsPerRun'] = eventsPerRun
    dashboard_report['NbEvPerRun'] = eventsPerRun
    dashboard_report['NEventsProcessed'] = eventsPerRun
        #print "Total number of events:", eventsPerRun
##

    # check if storageStatistics is valid
    if storageStatistics.find('Storage statistics:') != -1 :
        # report form: { 'protocol' : { 'action' : [attempted,succedeed,total-size,total-time,min-time,max-time] , ... } , ... }
        report = {}
        data = storageStatistics.split('Storage statistics:')[1]
        data_fields = data.split(';')
        for data_field in data_fields:
            # parse: format protocol/action = attepted/succedeed/total-size/total-time/min-time/max-time
            if data_field == ' ' or not data_field or data_field == '':
               continue
            key = data_field.split('=')[0].strip()
            item = data_field.split('=')[1].strip()
            protocol = str(key.split('/')[0].strip())
            action = str(key.split('/')[1].strip())
            item_array = item.split('/')
            attempted = str(item_array[0].strip())
            succeeded = str(item_array[1].strip())
            total_size = str(item_array[2].strip().split('MB')[0])
            total_time = str(item_array[3].strip().split('ms')[0])
            min_time = str(item_array[4].strip().split('ms')[0])
            max_time = str(item_array[5].strip().split('ms')[0])
            # add to report
            if protocol in report.keys() :
                if action in report[protocol].keys() :
                    print 'protocol/action:',protocol,'/',action,'listed twice in report, taking the first'
                else :
                    report[protocol][action] = [attempted,succeeded,total_size,total_time,min_time,max_time]
            else :
                report[protocol] = {action : [attempted,succeeded,total_size,total_time,min_time,max_time] }

        if debug :
            for protocol in report.keys() :
                print 'protocol:',protocol
                for action in report[protocol].keys() :
                    print 'action:',action,'measurement:',report[protocol][action]

        # extract information to be sent to DashBoard
        # per protocol and for action=read, calculate MBPS
        # dashboard key is io_action
        dashboard_report['MonitorID'] = MonitorID
        dashboard_report['MonitorJobID'] = MonitorJobID
        for protocol in report.keys() :
            for action in report[protocol].keys() :
                try: size = float(report[protocol][action][2])
                except: size = 'NULL'
                try: time = float(report[protocol][action][3])/1000
                except: time = 'NULL'
                dashboard_report['io_'+protocol+'_'+action] = str(size)+'_'+str(time)

        if debug :
            ordered = dashboard_report.keys()
            ordered.sort()
            for key in ordered:
                print key,'=',dashboard_report[key]

        # send to DashBoard
        apmonSend(MonitorID, MonitorJobID, dashboard_report)
        apmonFree()

    # prepare exit string
    exit_string = str(exit_status)
    for key in dashboard_report.keys() :
        exit_string += ';' + str(key) + '=' + str(dashboard_report[key])

    
    return exit_string


if __name__ == '__main__' :
    exit_status = main(sys.argv[1:])
    # output for wrapper script
    print exit_status

    
Revision:	1.9
Committed:	Tue May 13 10:22:55 2008 UTC (16 years, 11 months ago) by afanfani
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	PRODCOMMON_0_10_7_testCS2, CRAB_2_2_1_pre3, CRAB_2_2_1_pre2, CRAB_2_2_1_pre1, CRAB_2_2_0, CRAB_2_2_0_pre21
Changes since 1.8:	+16 -0 lines
Log Message:	Add number of events processed to the Dashboard report
#	User	Rev	Content
1	gutsche	1.1	#!/usr/bin/env python
2
3			import sys, getopt, string
4
5	fanzago	1.6	from ProdCommon.FwkJobRep.ReportParser import readJobReport
6	gutsche	1.2	from DashboardAPI import apmonSend, apmonFree
7
8	gutsche	1.1
9			def main(argv) :
10			"""
11			parseCrabFjr
12
13	gutsche	1.2	- parse CRAB FrameworkJobReport on WN: { 'protocol' : { 'action' : [attempted,succedeed,total-size,total-time,min-time,max-time] , ... } , ... }
14			- report parameters to DashBoard using DashBoardApi.py: for all 'read' actions of all protocols, report MBPS
15			- return ExitStatus and dump of DashBoard report separated by semi-colon to WN wrapper script
16	gutsche	1.1
17			required parameters:
18			--input : input FJR xml file
19	gutsche	1.2	--MonitorID : DashBoard MonitorID
20			--MonitorJobID : DashBoard MonitorJobID
21	gutsche	1.1
22			optional parameters:
23			--help : help
24			--debug : debug statements
25
26			"""
27
28			# defaults
29			input = ''
30	gutsche	1.2	MonitorID = ''
31			MonitorJobID = ''
32	gutsche	1.1	debug = 0
33
34			try:
35	gutsche	1.2	opts, args = getopt.getopt(argv, "", ["input=", "MonitorID=", "MonitorJobID=", "debug", "help"])
36	gutsche	1.1	except getopt.GetoptError:
37			print main.__doc__
38			sys.exit(2)
39
40			# check command line parameter
41			for opt, arg in opts :
42			if opt == "--help" :
43			print main.__doc__
44			sys.exit()
45			elif opt == "--input" :
46			input = arg
47	gutsche	1.2	elif opt == "--MonitorID" :
48			MonitorID = arg
49			elif opt == "--MonitorJobID" :
50			MonitorJobID = arg
51	gutsche	1.1	elif opt == "--debug" :
52			debug = 1
53
54	gutsche	1.2	if input == '' or MonitorID == '' or MonitorJobID == '':
55	gutsche	1.1	print main.__doc__
56			sys.exit()
57
58			# load FwkJobRep
59			jobReport = readJobReport(input)[0]
60
61	gutsche	1.5	exit_status = ''
62	gutsche	1.1
63	fanzago	1.7	##### temporary fix for FJR incomplete ####
64			fjr = open (input)
65			len_fjr = len(fjr.readlines())
66			if (len_fjr <= 6):
67			### 50115 - cmsRun did not produce a valid/readable job report at runtime
68			exit_status = str(50115)
69			else:
70			# get ExitStatus of last error
71			if len(jobReport.errors) != 0 :
72			exit_status = str(jobReport.errors[-1]['ExitStatus'])
73			else :
74			exit_status = str(0)
75	gutsche	1.1
76			# get i/o statistics
77			storageStatistics = str(jobReport.storageStatistics)
78
79	gutsche	1.2	# dashboard report dictionary
80			dashboard_report = {}
81
82	afanfani	1.9	##Brian's patch to sent number of events procedded to the Dashboard
83			# Add NoEventsPerRun to the Dashboard report
84			eventsPerRun = 0
85			for inputFile in jobReport.inputFiles:
86			try:
87			eventsRead = str(inputFile.get('EventsRead', 0))
88			eventsRead = int(eventsRead.strip())
89			except:
90			continue
91			eventsPerRun += eventsRead
92			dashboard_report['NoEventsPerRun'] = eventsPerRun
93			dashboard_report['NbEvPerRun'] = eventsPerRun
94			dashboard_report['NEventsProcessed'] = eventsPerRun
95			#print "Total number of events:", eventsPerRun
96			##
97
98	gutsche	1.2	# check if storageStatistics is valid
99			if storageStatistics.find('Storage statistics:') != -1 :
100			# report form: { 'protocol' : { 'action' : [attempted,succedeed,total-size,total-time,min-time,max-time] , ... } , ... }
101			report = {}
102			data = storageStatistics.split('Storage statistics:')[1]
103			data_fields = data.split(';')
104			for data_field in data_fields:
105			# parse: format protocol/action = attepted/succedeed/total-size/total-time/min-time/max-time
106	afanfani	1.8	if data_field == ' ' or not data_field or data_field == '':
107			continue
108	gutsche	1.2	key = data_field.split('=')[0].strip()
109			item = data_field.split('=')[1].strip()
110			protocol = str(key.split('/')[0].strip())
111			action = str(key.split('/')[1].strip())
112			item_array = item.split('/')
113			attempted = str(item_array[0].strip())
114			succeeded = str(item_array[1].strip())
115			total_size = str(item_array[2].strip().split('MB')[0])
116			total_time = str(item_array[3].strip().split('ms')[0])
117			min_time = str(item_array[4].strip().split('ms')[0])
118			max_time = str(item_array[5].strip().split('ms')[0])
119			# add to report
120			if protocol in report.keys() :
121			if action in report[protocol].keys() :
122			print 'protocol/action:',protocol,'/',action,'listed twice in report, taking the first'
123			else :
124			report[protocol][action] = [attempted,succeeded,total_size,total_time,min_time,max_time]
125			else :
126	gutsche	1.3	report[protocol] = {action : [attempted,succeeded,total_size,total_time,min_time,max_time] }
127	gutsche	1.2
128			if debug :
129			for protocol in report.keys() :
130			print 'protocol:',protocol
131			for action in report[protocol].keys() :
132			print 'action:',action,'measurement:',report[protocol][action]
133
134			# extract information to be sent to DashBoard
135			# per protocol and for action=read, calculate MBPS
136			# dashboard key is io_action
137			dashboard_report['MonitorID'] = MonitorID
138			dashboard_report['MonitorJobID'] = MonitorJobID
139			for protocol in report.keys() :
140	gutsche	1.3	for action in report[protocol].keys() :
141			try: size = float(report[protocol][action][2])
142			except: size = 'NULL'
143	gutsche	1.4	try: time = float(report[protocol][action][3])/1000
144	gutsche	1.3	except: time = 'NULL'
145			dashboard_report['io_'+protocol+'_'+action] = str(size)+'_'+str(time)
146	gutsche	1.2
147			if debug :
148	gutsche	1.3	ordered = dashboard_report.keys()
149			ordered.sort()
150			for key in ordered:
151			print key,'=',dashboard_report[key]
152	gutsche	1.2
153			# send to DashBoard
154			apmonSend(MonitorID, MonitorJobID, dashboard_report)
155			apmonFree()
156
157			# prepare exit string
158			exit_string = str(exit_status)
159			for key in dashboard_report.keys() :
160			exit_string += ';' + str(key) + '=' + str(dashboard_report[key])
161
162	gutsche	1.5
163
164	gutsche	1.2	return exit_string
165	gutsche	1.1
166
167			if __name__ == '__main__' :
168	gutsche	1.2	exit_status = main(sys.argv[1:])
169			# output for wrapper script
170			print exit_status
171	gutsche	1.1
172