CRAB/python/Status.py

from Actor import *
import common
import string, os, time
import sha
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",'')
        self.server_name = ''
 
        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_()
        ## TEMPORARY FIXME Ds
        msg = showWebMon(self.server_name)
        common.logger.message(msg)

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

    def query(self,display=True):
        """
        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,display)

    def compute(self, up_task, display=True ):

        toPrint=[]
        taskId = str(up_task['name'])
        task_unique_name = str(up_task['name'])
        ended = None

        self.wrapErrorList = []
        for job in up_task.jobs :
            id = str(job.runningJob['jobId'])
            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'])
            self.wrapErrorList.append(job_exit_code)
            ended = str(job['standardInput'])  
            printline=''
            if dest == 'None' :  dest = ''
            if exe_exit_code == 'None' :  exe_exit_code = ''
            if job_exit_code == 'None' :  job_exit_code = ''
            printline+="%-6s %-18s %-36s %-13s %-16s %-4s" % (id,jobStatus,dest,exe_exit_code,job_exit_code,ended)
            toPrint.append(printline)

            if jobStatus is not None:
                self.dataToDash(job,id,taskId,task_unique_name,dest,jobStatus)
        header = ''
        if ended != None and len(ended) > 0:
            header+= "%-6s %-18s %-36s %-13s %-16s %-4s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS','ENDED')
        else:
            header+= "%-6s %-18s %-36s %-13s %-16s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS')

        if display: displayReport(self,header,toPrint,self.xml)

        return

    def PrintReport_(self):


        possible_status = [
                         'Created',
                         'Undefined',
                         'Submitting',
                         'Submitted',
                         'Waiting',
                         'Ready',
                         'Scheduled',
                         'Running',
                         'Done',
                         'Killing',
                         'Killed',
                         'Aborted',
                         'Unknown',
                         'Done (Failed)',
                         'Cleared',
                         'retrieved'
                          ]

        jobs = common._db.nJobs('list')
        WrapExitCode = list(set(self.wrapErrorList))
        print ''
        print ">>>>>>>>> %i Total Jobs " % (len(jobs))
        print ''
        list_ID=[]
        for c in WrapExitCode:
            if c != 'None':
                self.reportCodes(c)
            else:
                for st in possible_status:
                    list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
                    if len(list_ID)>0:
                        if st == 'killed':
                            print ">>>>>>>>> %i Jobs %s  " % (len(list_ID), str(st))
                            print "          You can resubmit them specifying JOB numbers: crab -resubmit <List of jobs>"
                            print "          List of jobs: %s \n" % readableList(self,list_ID)
                        elif st == 'Aborted':
                            print ">>>>>>>>> %i Jobs %s  " % (len(list_ID), str(st))
                            print "          You can resubmit them specifying JOB numbers: crab -resubmit <List of jobs>"
                            print "          List of jobs: %s \n" % readableList(self,list_ID)
                        elif st == 'Done' or st == 'Done (Failed)' :
                            print ">>>>>>>>> %i Jobs %s  " % (len(list_ID), str(st))
                            print "          Retrieve them with: crab -getoutput <List of jobs>"
                            print "          List of jobs: %s \n" % readableList(self,list_ID)
                        else   :
                            print ">>>>>>>>> %i Jobs %s \n " % (len(list_ID), str(st))
                            print "          List of jobs %s: %s \n" % (str(st),readableList(self,list_ID))
                        pass
                    pass
                pass
            pass
        return

    def reportCodes(self,code): 
        """
        """
        list_ID = common._db.queryAttrRunJob({'wrapperReturnCode':code},'jobId')
        if len(list_ID)>0:
            print ">>>>>>>>> %i Jobs with Wrapper Exit Code : %s " % (len(list_ID), str(code))
            print "          List of jobs: %s" % readableList(self,list_ID)
            print " "

        return
 
    def dataToDash(self,job,id,taskId,task_unique_name,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() in ['CONDOR_G','GLIDEIN']:
            WMS = 'OSG'
            taskHash = sha.new(common._db.queryTask('name')).hexdigest()
            jobId = str(id) + '_https://' + common.scheduler.name() + '/' + taskHash + '/' + str(id)
            common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
        elif common.scheduler.name().upper() in ['LSF','CAF']:
            WMS = common.scheduler.name()
            jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(task_unique_name,"_","-")
            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

        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.61
Committed:	Mon Mar 9 14:23:32 2009 UTC (16 years, 1 month ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_6_0_pre1, CRAB_2_5_1, CRAB_2_5_1_pre4, CRAB_2_5_1_pre3, CRAB_2_5_1_pre2, CRAB_2_5_1_pre1, CRAB_2_5_0, CRAB_2_5_0_pre7
Changes since 1.60:	+2 -2 lines
Log Message:	fix for dashoboard
#	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	ewv	1.42	import sha
5	spiga	1.28	from crab_util import *
6	slacapra	1.1
7	mcinquil	1.53
8	slacapra	1.1	class Status(Actor):
9	spiga	1.24	def __init__(self, *args):
10			self.cfg_params = args[0]
11	spiga	1.29	self.xml = self.cfg_params.get("USER.xml_report",'')
12	spiga	1.56	self.server_name = ''
13
14	slacapra	1.1	return
15
16			def run(self):
17			"""
18	spiga	1.24	The main method of the class: compute the status and print a report
19	slacapra	1.1	"""
20			common.logger.debug(5, "Status::run() called")
21
22	spiga	1.24	start = time.time()
23	mcinquil	1.53
24	spiga	1.28	self.query()
25	slacapra	1.15	self.PrintReport_()
26	spiga	1.56	## TEMPORARY FIXME Ds
27			msg = showWebMon(self.server_name)
28	slacapra	1.59	common.logger.message(msg)
29	mcinquil	1.53
30	spiga	1.24	stop = time.time()
31			common.logger.debug(1, "Status Time: "+str(stop - start))
32			common.logger.write("Status Time: "+str(stop - start))
33	slacapra	1.15	pass
34
35	spiga	1.52	def query(self,display=True):
36	slacapra	1.15	"""
37	spiga	1.24	compute the status
38	slacapra	1.15	"""
39	spiga	1.24	common.logger.message("Checking the status of all jobs: please wait")
40			task = common._db.getTask()
41	ewv	1.33	upTask = common.scheduler.queryEverything(task['id'])
42	spiga	1.52	self.compute(upTask,display)
43	spiga	1.28
44	spiga	1.52	def compute(self, up_task, display=True ):
45	ewv	1.33
46	spiga	1.24	toPrint=[]
47	spiga	1.61	taskId = str(up_task['name'])
48			task_unique_name = str(up_task['name'])
49	mcinquil	1.58	ended = None
50	ewv	1.44
51	spiga	1.37	self.wrapErrorList = []
52	spiga	1.25	for job in up_task.jobs :
53	spiga	1.35	id = str(job.runningJob['jobId'])
54	spiga	1.24	jobStatus = str(job.runningJob['statusScheduler'])
55			dest = str(job.runningJob['destination']).split(':')[0]
56			exe_exit_code = str(job.runningJob['applicationReturnCode'])
57	ewv	1.33	job_exit_code = str(job.runningJob['wrapperReturnCode'])
58	spiga	1.37	self.wrapErrorList.append(job_exit_code)
59	mcinquil	1.54	ended = str(job['standardInput'])
60	spiga	1.24	printline=''
61			if dest == 'None' : dest = ''
62			if exe_exit_code == 'None' : exe_exit_code = ''
63			if job_exit_code == 'None' : job_exit_code = ''
64	mcinquil	1.54	printline+="%-6s %-18s %-36s %-13s %-16s %-4s" % (id,jobStatus,dest,exe_exit_code,job_exit_code,ended)
65	spiga	1.24	toPrint.append(printline)
66	corvo	1.16
67	spiga	1.28	if jobStatus is not None:
68	spiga	1.51	self.dataToDash(job,id,taskId,task_unique_name,dest,jobStatus)
69	spiga	1.28	header = ''
70	mcinquil	1.58	if ended != None and len(ended) > 0:
71	mcinquil	1.54	header+= "%-6s %-18s %-36s %-13s %-16s %-4s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS','ENDED')
72			else:
73	mcinquil	1.55	header+= "%-6s %-18s %-36s %-13s %-16s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS')
74	slacapra	1.1
75	spiga	1.52	if display: displayReport(self,header,toPrint,self.xml)
76	spiga	1.9
77	spiga	1.28	return
78	spiga	1.24
79			def PrintReport_(self):
80
81	spiga	1.39
82			possible_status = [
83	spiga	1.41	'Created',
84	spiga	1.39	'Undefined',
85	ewv	1.42	'Submitting',
86	spiga	1.39	'Submitted',
87			'Waiting',
88			'Ready',
89			'Scheduled',
90			'Running',
91			'Done',
92	ewv	1.42	'Killing',
93	spiga	1.41	'Killed',
94	spiga	1.39	'Aborted',
95			'Unknown',
96	spiga	1.49	'Done (Failed)',
97	spiga	1.47	'Cleared',
98	ewv	1.42	'retrieved'
99	spiga	1.39	]
100
101	spiga	1.37	jobs = common._db.nJobs('list')
102			WrapExitCode = list(set(self.wrapErrorList))
103	spiga	1.24	print ''
104	spiga	1.37	print ">>>>>>>>> %i Total Jobs " % (len(jobs))
105	spiga	1.24	print ''
106	ewv	1.33	list_ID=[]
107	spiga	1.37	for c in WrapExitCode:
108	spiga	1.49	if c != 'None':
109			self.reportCodes(c)
110			else:
111	spiga	1.39	for st in possible_status:
112			list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
113			if len(list_ID)>0:
114	ewv	1.42	if st == 'killed':
115	spiga	1.39	print ">>>>>>>>> %i Jobs %s " % (len(list_ID), str(st))
116	spiga	1.46	print " You can resubmit them specifying JOB numbers: crab -resubmit <List of jobs>"
117	spiga	1.47	print " List of jobs: %s \n" % readableList(self,list_ID)
118	ewv	1.42	elif st == 'Aborted':
119	spiga	1.39	print ">>>>>>>>> %i Jobs %s " % (len(list_ID), str(st))
120	spiga	1.46	print " You can resubmit them specifying JOB numbers: crab -resubmit <List of jobs>"
121	spiga	1.47	print " List of jobs: %s \n" % readableList(self,list_ID)
122	spiga	1.49	elif st == 'Done' or st == 'Done (Failed)' :
123	spiga	1.39	print ">>>>>>>>> %i Jobs %s " % (len(list_ID), str(st))
124			print " Retrieve them with: crab -getoutput <List of jobs>"
125	spiga	1.47	print " List of jobs: %s \n" % readableList(self,list_ID)
126	ewv	1.42	else :
127	spiga	1.39	print ">>>>>>>>> %i Jobs %s \n " % (len(list_ID), str(st))
128	slacapra	1.59	print " List of jobs %s: %s \n" % (str(st),readableList(self,list_ID))
129			pass
130			pass
131			pass
132			pass
133			return
134	spiga	1.39
135	spiga	1.48	def reportCodes(self,code):
136	spiga	1.47	"""
137			"""
138	spiga	1.48	list_ID = common._db.queryAttrRunJob({'wrapperReturnCode':code},'jobId')
139	spiga	1.47	if len(list_ID)>0:
140	spiga	1.48	print ">>>>>>>>> %i Jobs with Wrapper Exit Code : %s " % (len(list_ID), str(code))
141	spiga	1.47	print " List of jobs: %s" % readableList(self,list_ID)
142			print " "
143	spiga	1.24
144	spiga	1.47	return
145
146	spiga	1.51	def dataToDash(self,job,id,taskId,task_unique_name,dest,jobStatus):
147	spiga	1.28	jid = job.runningJob['schedulerId']
148			job_status_reason = str(job.runningJob['statusReason'])
149			job_last_time = str(job.runningJob['startTime'])
150	ewv	1.43	if common.scheduler.name().upper() in ['CONDOR_G','GLIDEIN']:
151	spiga	1.30	WMS = 'OSG'
152	ewv	1.42	taskHash = sha.new(common._db.queryTask('name')).hexdigest()
153	ewv	1.43	jobId = str(id) + '_https://' + common.scheduler.name() + '/' + taskHash + '/' + str(id)
154	spiga	1.28	common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
155	ewv	1.42	elif common.scheduler.name().upper() in ['LSF','CAF']:
156			WMS = common.scheduler.name()
157	spiga	1.51	jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(task_unique_name,"_","-")
158	ewv	1.42	common.logger.debug(5,'JobID for ML monitoring is created for Local scheduler:'+jobId)
159	spiga	1.28	else:
160	ewv	1.42	jobId = str(id) + '_' + str(jid)
161			WMS = job.runningJob['service']
162			common.logger.debug(5,'JobID for ML monitoring is created for gLite scheduler:'+jobId)
163	spiga	1.28	pass
164
165			common.logger.debug(5,"sending info to ML")
166			params = {}
167			if WMS != None:
168			params = {'taskId': taskId, \
169			'jobId': jobId,\
170			'sid': str(jid), \
171			'StatusValueReason': job_status_reason, \
172			'StatusValue': jobStatus, \
173			'StatusEnterTime': job_last_time, \
174			'StatusDestination': dest, \
175			'RBname': WMS }
176			else:
177			params = {'taskId': taskId, \
178			'jobId': jobId,\
179			'sid': str(jid), \
180			'StatusValueReason': job_status_reason, \
181			'StatusValue': jobStatus, \
182			'StatusEnterTime': job_last_time, \
183			'StatusDestination': dest }
184			common.logger.debug(5,str(params))
185			common.apmon.sendToML(params)
186
187			return
188	ewv	1.33
189	spiga	1.24	def joinIntArray_(self,array) :
190			output = ''
191			for item in array :
192			output += str(item)+','
193			if output[-1] == ',' :
194			output = output[:-1]
195			return output