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 = {}

    # 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.8
Committed:	Mon Apr 7 18:14:28 2008 UTC (17 years ago) by afanfani
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_2_0_pre19, CRAB_2_2_0_pre18, CRAB_2_2_0_pre17, CRAB_2_2_0_pre16, CRAB_2_2_0_pre15, CRAB_2_2_0_pre13, CRAB_2_2_0_pre12, CRAB_2_2_0_pre11, CRAB_2_2_0_pre10, bp_osg_bdii, CRAB_2_2_0_pre9, CRAB_2_2_0_pre8, CRAB_2_2_0_pre7, CRAB_2_2_0_pre5
Branch point for:	osg_bdii
Changes since 1.7:	+2 -0 lines
Log Message:	cope with almost empty StorageStatistic tag
#	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			# check if storageStatistics is valid
83			if storageStatistics.find('Storage statistics:') != -1 :
84			# report form: { 'protocol' : { 'action' : [attempted,succedeed,total-size,total-time,min-time,max-time] , ... } , ... }
85			report = {}
86			data = storageStatistics.split('Storage statistics:')[1]
87			data_fields = data.split(';')
88			for data_field in data_fields:
89			# parse: format protocol/action = attepted/succedeed/total-size/total-time/min-time/max-time
90	afanfani	1.8	if data_field == ' ' or not data_field or data_field == '':
91			continue
92	gutsche	1.2	key = data_field.split('=')[0].strip()
93			item = data_field.split('=')[1].strip()
94			protocol = str(key.split('/')[0].strip())
95			action = str(key.split('/')[1].strip())
96			item_array = item.split('/')
97			attempted = str(item_array[0].strip())
98			succeeded = str(item_array[1].strip())
99			total_size = str(item_array[2].strip().split('MB')[0])
100			total_time = str(item_array[3].strip().split('ms')[0])
101			min_time = str(item_array[4].strip().split('ms')[0])
102			max_time = str(item_array[5].strip().split('ms')[0])
103			# add to report
104			if protocol in report.keys() :
105			if action in report[protocol].keys() :
106			print 'protocol/action:',protocol,'/',action,'listed twice in report, taking the first'
107			else :
108			report[protocol][action] = [attempted,succeeded,total_size,total_time,min_time,max_time]
109			else :
110	gutsche	1.3	report[protocol] = {action : [attempted,succeeded,total_size,total_time,min_time,max_time] }
111	gutsche	1.2
112			if debug :
113			for protocol in report.keys() :
114			print 'protocol:',protocol
115			for action in report[protocol].keys() :
116			print 'action:',action,'measurement:',report[protocol][action]
117
118			# extract information to be sent to DashBoard
119			# per protocol and for action=read, calculate MBPS
120			# dashboard key is io_action
121			dashboard_report['MonitorID'] = MonitorID
122			dashboard_report['MonitorJobID'] = MonitorJobID
123			for protocol in report.keys() :
124	gutsche	1.3	for action in report[protocol].keys() :
125			try: size = float(report[protocol][action][2])
126			except: size = 'NULL'
127	gutsche	1.4	try: time = float(report[protocol][action][3])/1000
128	gutsche	1.3	except: time = 'NULL'
129			dashboard_report['io_'+protocol+'_'+action] = str(size)+'_'+str(time)
130	gutsche	1.2
131			if debug :
132	gutsche	1.3	ordered = dashboard_report.keys()
133			ordered.sort()
134			for key in ordered:
135			print key,'=',dashboard_report[key]
136	gutsche	1.2
137			# send to DashBoard
138			apmonSend(MonitorID, MonitorJobID, dashboard_report)
139			apmonFree()
140
141			# prepare exit string
142			exit_string = str(exit_status)
143			for key in dashboard_report.keys() :
144			exit_string += ';' + str(key) + '=' + str(dashboard_report[key])
145
146	gutsche	1.5
147
148	gutsche	1.2	return exit_string
149	gutsche	1.1
150
151			if __name__ == '__main__' :
152	gutsche	1.2	exit_status = main(sys.argv[1:])
153			# output for wrapper script
154			print exit_status
155	gutsche	1.1
156