CRAB/python/Status.py

from Actor import *
import common
import string, os, time
from crab_util import *

class Status(Actor):
    def __init__(self, *args):
        self.cfg_params = args[0]

        self.xml = self.cfg_params.get("USER.xml_report",'')

        return

    def run(self):
        """
        The main method of the class: compute the status and print a report
        """
        common.logger.debug(5, "Status::run() called")

        start = time.time()
        self.query()
        self.PrintReport_()
        stop = time.time()
        common.logger.debug(1, "Status Time: "+str(stop - start))
        common.logger.write("Status Time: "+str(stop - start))
        pass

    def query(self):
        """
        compute the status
        """
        common.logger.message("Checking the status of all jobs: please wait")
        task = common._db.getTask()
        upTask = common.scheduler.queryEverything(task['id']) 
        self.compute(upTask)

    def compute(self, up_task):
 
        toPrint=[]

        taskId=str("_".join(up_task['name']).split('_')[:-1])

        for job in up_task.jobs :
            id = str(job.runningJob['id'])
            jobStatus =  str(job.runningJob['statusScheduler'])
            dest = str(job.runningJob['destination']).split(':')[0]
            exe_exit_code = str(job.runningJob['applicationReturnCode'])
            job_exit_code = str(job.runningJob['wrapperReturnCode']) 
            printline=''
            header = ''
            if dest == 'None' :  dest = ''
            if exe_exit_code == 'None' :  exe_exit_code = ''
            if job_exit_code == 'None' :  job_exit_code = ''
            printline+="%-8s %-18s %-40s %-13s %-15s" % (id,jobStatus,dest,exe_exit_code,job_exit_code)
            toPrint.append(printline)

            if jobStatus is not None:
                self.dataToDash(job,id,taskId,dest,jobStatus)
 
        header = ''
        header+= "%-8s %-18s %-40s %-13s %-15s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS')

        displayReport(self,header,toPrint,self.xml)

        return

    def PrintReport_(self):

        possible_status = [
                         'Undefined', 
                         'Submitted',
                         'Waiting',
                         'Ready', 
                         'Scheduled',
                         'Running',
                         'Done',
                         'Cancelled',
                         'Aborted',
                         'Unknown',
                         'Done(failed)'
                         'Cleared'
                          ]

        print ''
        print ">>>>>>>>> %i Total Jobs " % (common._db.nJobs())
        print ''
        list_ID=[] 
        for st in possible_status:
            list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
            if len(list_ID)>0:
                print ">>>>>>>>> %i Jobs  %s " % (len(list_ID), str(st))#,len(list_ID)
                if st == 'killed' or st == 'Aborted': print "          You can resubmit them specifying JOB numbers: crab -resubmit JOB_number <Jobs list>"
                if st == 'Done'   : print "          Retrieve them with: crab -getoutput <Jobs list>"
                if st == 'Cleared': print "          %i Jobs with EXE_EXIT_CODE: %s" % (len(common._db.queryDistJob('wrapperReturnCode')))
                print "          List of jobs: %s" % str(common._db.queryAttrRunJob({'statusScheduler':st},'jobId'))
                print " "

#        if (len(self.countCorrupt) != 0):
#            self.countCorrupt.sort()
#            print ''
#            print ">>>>>>>>> %i Jobs cleared with corrupt output" % len(self.countCorrupt)
#            print "          List of jobs: %s" % self.joinIntArray_(self.countCorrupt)
#            print "          You can resubmit them specifying JOB numbers: crab -resubmit JOB_number <Jobs list>"
#        if (len(self.countCleared.keys()) != 0):
#            total_size = 0
#            for key in self.countCleared.keys() :
#                total_size += len(self.countCleared[key])
#            print ''


    def dataToDash(self,job,id,taskId,dest,jobStatus):
        

        jid = job.runningJob['schedulerId']
        job_status_reason = str(job.runningJob['statusReason'])
        job_last_time = str(job.runningJob['startTime'])
        if common.scheduler.name().upper() == 'CONDOR_G':
            WMS = 'OSG'
            self.hash = makeCksum(common.work_space.cfgFileName())
            jobId = str(id) + '_' + self.hash + '_' + str(jid)
            common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
        else:
            if common.scheduler.name() in ['lsf','caf']:
                WMS = common.scheduler.name()
                jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(taskId,"_","-")
                common.logger.debug(5,'JobID for ML monitoring is created for Local scheduler:'+jobId)
            else:
                jobId = str(id) + '_' + str(jid)
                WMS = job.runningJob['service']
                common.logger.debug(5,'JobID for ML monitoring is created for gLite scheduler:'+jobId)
            pass
        pass

        common.logger.debug(5,"sending info to ML")
        params = {}
        if WMS != None:
            params = {'taskId': taskId, \
            'jobId': jobId,\
            'sid': str(jid), \
            'StatusValueReason': job_status_reason, \
            'StatusValue': jobStatus, \
            'StatusEnterTime': job_last_time, \
            'StatusDestination': dest, \
            'RBname': WMS }
        else:
            params = {'taskId': taskId, \
            'jobId': jobId,\
            'sid': str(jid), \
            'StatusValueReason': job_status_reason, \
            'StatusValue': jobStatus, \
            'StatusEnterTime': job_last_time, \
            'StatusDestination': dest }
        common.logger.debug(5,str(params))
        common.apmon.sendToML(params)

        return
 
    def joinIntArray_(self,array) :
        output = ''
        for item in array :
            output += str(item)+','
        if output[-1] == ',' :
            output = output[:-1]
        return output

Revision:	1.31
Committed:	Sun Apr 20 18:11:49 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.30:	+1 -1 lines
Log Message:	fix for info to the dashboard
#	User	Rev	Content
1	slacapra	1.1	from Actor import *
2	slacapra	1.21	import common
3	spiga	1.24	import string, os, time
4	spiga	1.28	from crab_util import *
5	slacapra	1.1
6			class Status(Actor):
7	spiga	1.24	def __init__(self, *args):
8			self.cfg_params = args[0]
9
10	spiga	1.29	self.xml = self.cfg_params.get("USER.xml_report",'')
11	slacapra	1.1
12			return
13
14			def run(self):
15			"""
16	spiga	1.24	The main method of the class: compute the status and print a report
17	slacapra	1.1	"""
18			common.logger.debug(5, "Status::run() called")
19
20	spiga	1.24	start = time.time()
21	spiga	1.28	self.query()
22	slacapra	1.15	self.PrintReport_()
23	spiga	1.24	stop = time.time()
24			common.logger.debug(1, "Status Time: "+str(stop - start))
25			common.logger.write("Status Time: "+str(stop - start))
26	slacapra	1.15	pass
27
28	spiga	1.28	def query(self):
29	slacapra	1.15	"""
30	spiga	1.24	compute the status
31	slacapra	1.15	"""
32	spiga	1.24	common.logger.message("Checking the status of all jobs: please wait")
33			task = common._db.getTask()
34	spiga	1.28	upTask = common.scheduler.queryEverything(task['id'])
35			self.compute(upTask)
36
37			def compute(self, up_task):
38
39	spiga	1.24	toPrint=[]
40	spiga	1.28
41	spiga	1.31	taskId=str("_".join(up_task['name']).split('_')[:-1])
42	spiga	1.28
43	spiga	1.25	for job in up_task.jobs :
44	spiga	1.27	id = str(job.runningJob['id'])
45	spiga	1.24	jobStatus = str(job.runningJob['statusScheduler'])
46			dest = str(job.runningJob['destination']).split(':')[0]
47			exe_exit_code = str(job.runningJob['applicationReturnCode'])
48			job_exit_code = str(job.runningJob['wrapperReturnCode'])
49			printline=''
50	spiga	1.28	header = ''
51	spiga	1.24	if dest == 'None' : dest = ''
52			if exe_exit_code == 'None' : exe_exit_code = ''
53			if job_exit_code == 'None' : job_exit_code = ''
54			printline+="%-8s %-18s %-40s %-13s %-15s" % (id,jobStatus,dest,exe_exit_code,job_exit_code)
55			toPrint.append(printline)
56	corvo	1.16
57	spiga	1.28	if jobStatus is not None:
58			self.dataToDash(job,id,taskId,dest,jobStatus)
59
60			header = ''
61			header+= "%-8s %-18s %-40s %-13s %-15s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS')
62	slacapra	1.1
63	spiga	1.28	displayReport(self,header,toPrint,self.xml)
64	spiga	1.9
65	spiga	1.28	return
66	spiga	1.24
67			def PrintReport_(self):
68
69			possible_status = [
70			'Undefined',
71			'Submitted',
72			'Waiting',
73			'Ready',
74			'Scheduled',
75			'Running',
76			'Done',
77			'Cancelled',
78			'Aborted',
79			'Unknown',
80			'Done(failed)'
81			'Cleared'
82			]
83	slacapra	1.1
84	spiga	1.24	print ''
85			print ">>>>>>>>> %i Total Jobs " % (common._db.nJobs())
86			print ''
87			list_ID=[]
88			for st in possible_status:
89			list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
90			if len(list_ID)>0:
91			print ">>>>>>>>> %i Jobs %s " % (len(list_ID), str(st))#,len(list_ID)
92			if st == 'killed' or st == 'Aborted': print " You can resubmit them specifying JOB numbers: crab -resubmit JOB_number <Jobs list>"
93			if st == 'Done' : print " Retrieve them with: crab -getoutput <Jobs list>"
94			if st == 'Cleared': print " %i Jobs with EXE_EXIT_CODE: %s" % (len(common._db.queryDistJob('wrapperReturnCode')))
95			print " List of jobs: %s" % str(common._db.queryAttrRunJob({'statusScheduler':st},'jobId'))
96			print " "
97
98			# if (len(self.countCorrupt) != 0):
99			# self.countCorrupt.sort()
100			# print ''
101			# print ">>>>>>>>> %i Jobs cleared with corrupt output" % len(self.countCorrupt)
102			# print " List of jobs: %s" % self.joinIntArray_(self.countCorrupt)
103			# print " You can resubmit them specifying JOB numbers: crab -resubmit JOB_number <Jobs list>"
104			# if (len(self.countCleared.keys()) != 0):
105			# total_size = 0
106			# for key in self.countCleared.keys() :
107			# total_size += len(self.countCleared[key])
108			# print ''
109	slacapra	1.1
110
111	spiga	1.28	def dataToDash(self,job,id,taskId,dest,jobStatus):
112
113	slacapra	1.1
114	spiga	1.28	jid = job.runningJob['schedulerId']
115			job_status_reason = str(job.runningJob['statusReason'])
116			job_last_time = str(job.runningJob['startTime'])
117			if common.scheduler.name().upper() == 'CONDOR_G':
118	spiga	1.30	WMS = 'OSG'
119	spiga	1.28	self.hash = makeCksum(common.work_space.cfgFileName())
120			jobId = str(id) + '_' + self.hash + '_' + str(jid)
121			common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
122			else:
123			if common.scheduler.name() in ['lsf','caf']:
124	spiga	1.30	WMS = common.scheduler.name()
125			jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(taskId,"_","-")
126	spiga	1.28	common.logger.debug(5,'JobID for ML monitoring is created for Local scheduler:'+jobId)
127	spiga	1.30	else:
128			jobId = str(id) + '_' + str(jid)
129			WMS = job.runningJob['service']
130			common.logger.debug(5,'JobID for ML monitoring is created for gLite scheduler:'+jobId)
131	spiga	1.28	pass
132			pass
133
134			common.logger.debug(5,"sending info to ML")
135			params = {}
136			if WMS != None:
137			params = {'taskId': taskId, \
138			'jobId': jobId,\
139			'sid': str(jid), \
140			'StatusValueReason': job_status_reason, \
141			'StatusValue': jobStatus, \
142			'StatusEnterTime': job_last_time, \
143			'StatusDestination': dest, \
144			'RBname': WMS }
145			else:
146			params = {'taskId': taskId, \
147			'jobId': jobId,\
148			'sid': str(jid), \
149			'StatusValueReason': job_status_reason, \
150			'StatusValue': jobStatus, \
151			'StatusEnterTime': job_last_time, \
152			'StatusDestination': dest }
153			common.logger.debug(5,str(params))
154			common.apmon.sendToML(params)
155
156			return
157
158	spiga	1.24	def joinIntArray_(self,array) :
159			output = ''
160			for item in array :
161			output += str(item)+','
162			if output[-1] == ',' :
163			output = output[:-1]
164			return output
165	slacapra	1.1