| 2 |
|
|
| 3 |
|
import sys, getopt, string |
| 4 |
|
|
| 5 |
< |
from FwkJobRep.ReportParser import readJobReport |
| 5 |
> |
from ProdCommon.FwkJobRep.ReportParser import readJobReport |
| 6 |
> |
from DashboardAPI import apmonSend, apmonFree |
| 7 |
> |
|
| 8 |
|
|
| 9 |
|
def main(argv) : |
| 10 |
|
""" |
| 11 |
|
parseCrabFjr |
| 12 |
|
|
| 13 |
< |
parse CRAB FrameworkJobReport on WN and return parameters to WN wrapper script |
| 14 |
< |
|
| 15 |
< |
prints information separated by semi-colon in fixed order: |
| 14 |
< |
|
| 15 |
< |
1. ExitStatus (0 or ExitStatus from CMSSW) |
| 13 |
> |
- 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 |
|
|
| 17 |
|
required parameters: |
| 18 |
|
--input : input FJR xml file |
| 19 |
+ |
--MonitorID : DashBoard MonitorID |
| 20 |
+ |
--MonitorJobID : DashBoard MonitorJobID |
| 21 |
|
|
| 22 |
|
optional parameters: |
| 23 |
|
--help : help |
| 27 |
|
|
| 28 |
|
# defaults |
| 29 |
|
input = '' |
| 30 |
+ |
MonitorID = '' |
| 31 |
+ |
MonitorJobID = '' |
| 32 |
|
debug = 0 |
| 33 |
|
|
| 34 |
|
try: |
| 35 |
< |
opts, args = getopt.getopt(argv, "", ["input=", "debug", "help"]) |
| 35 |
> |
opts, args = getopt.getopt(argv, "", ["input=", "MonitorID=", "MonitorJobID=", "debug", "help"]) |
| 36 |
|
except getopt.GetoptError: |
| 37 |
|
print main.__doc__ |
| 38 |
|
sys.exit(2) |
| 44 |
|
sys.exit() |
| 45 |
|
elif opt == "--input" : |
| 46 |
|
input = arg |
| 47 |
+ |
elif opt == "--MonitorID" : |
| 48 |
+ |
MonitorID = arg |
| 49 |
+ |
elif opt == "--MonitorJobID" : |
| 50 |
+ |
MonitorJobID = arg |
| 51 |
|
elif opt == "--debug" : |
| 52 |
|
debug = 1 |
| 53 |
|
|
| 54 |
< |
if input == '': |
| 54 |
> |
if input == '' or MonitorID == '' or MonitorJobID == '': |
| 55 |
|
print main.__doc__ |
| 56 |
|
sys.exit() |
| 57 |
|
|
| 58 |
|
# load FwkJobRep |
| 59 |
|
jobReport = readJobReport(input)[0] |
| 60 |
|
|
| 61 |
< |
report = [] |
| 61 |
> |
exit_status = '' |
| 62 |
|
|
| 63 |
< |
# get ExitStatus of last error |
| 64 |
< |
if len(jobReport.errors) != 0 : |
| 65 |
< |
report.append(str(jobReport.errors[-1]['ExitStatus'])) |
| 66 |
< |
else : |
| 67 |
< |
report.append(str(0)) |
| 63 |
> |
##### 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 |
|
|
| 76 |
|
# get i/o statistics |
| 77 |
|
storageStatistics = str(jobReport.storageStatistics) |
| 78 |
|
|
| 79 |
< |
print ';'.join(report) |
| 79 |
> |
# dashboard report dictionary |
| 80 |
> |
dashboard_report = {} |
| 81 |
> |
|
| 82 |
> |
##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 |
> |
# 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 |
> |
if data_field == ' ' or not data_field or data_field == '': |
| 107 |
> |
continue |
| 108 |
> |
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 |
> |
report[protocol] = {action : [attempted,succeeded,total_size,total_time,min_time,max_time] } |
| 127 |
> |
|
| 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 |
> |
for action in report[protocol].keys() : |
| 141 |
> |
try: size = float(report[protocol][action][2]) |
| 142 |
> |
except: size = 'NULL' |
| 143 |
> |
try: time = float(report[protocol][action][3])/1000 |
| 144 |
> |
except: time = 'NULL' |
| 145 |
> |
dashboard_report['io_'+protocol+'_'+action] = str(size)+'_'+str(time) |
| 146 |
> |
|
| 147 |
> |
if debug : |
| 148 |
> |
ordered = dashboard_report.keys() |
| 149 |
> |
ordered.sort() |
| 150 |
> |
for key in ordered: |
| 151 |
> |
print key,'=',dashboard_report[key] |
| 152 |
> |
|
| 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 |
|
|
| 163 |
|
|
| 164 |
+ |
return exit_string |
| 165 |
+ |
|
| 166 |
|
|
| 167 |
|
if __name__ == '__main__' : |
| 168 |
< |
main(sys.argv[1:]) |
| 168 |
> |
exit_status = main(sys.argv[1:]) |
| 169 |
> |
# output for wrapper script |
| 170 |
> |
print exit_status |
| 171 |
|
|
| 172 |
|
|
