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.debug(1, 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 = uniqueTaskName(up_task['name'])
        task_unique_name = up_task['name']
        ended = ''

        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 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))

    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.57
Committed:	Tue Feb 10 14:36:54 2009 UTC (16 years, 2 months ago) by mcinquil
Content type:	text/x-python
Branch:	MAIN
Changes since 1.56:	+1 -0 lines
Log Message:	Fixed out of scope variable
#	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			common.logger.debug(1, 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	ewv	1.44	taskId = uniqueTaskName(up_task['name'])
48	spiga	1.51	task_unique_name = up_task['name']
49	mcinquil	1.57	ended = ''
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.54	if len(ended) > 0:
71			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
129	spiga	1.48	def reportCodes(self,code):
130	spiga	1.47	"""
131			"""
132	spiga	1.48	list_ID = common._db.queryAttrRunJob({'wrapperReturnCode':code},'jobId')
133	spiga	1.47	if len(list_ID)>0:
134	spiga	1.48	print ">>>>>>>>> %i Jobs with Wrapper Exit Code : %s " % (len(list_ID), str(code))
135	spiga	1.47	print " List of jobs: %s" % readableList(self,list_ID)
136			print " "
137	spiga	1.24
138	spiga	1.47	return
139
140	spiga	1.51	def dataToDash(self,job,id,taskId,task_unique_name,dest,jobStatus):
141	spiga	1.28	jid = job.runningJob['schedulerId']
142			job_status_reason = str(job.runningJob['statusReason'])
143			job_last_time = str(job.runningJob['startTime'])
144	ewv	1.43	if common.scheduler.name().upper() in ['CONDOR_G','GLIDEIN']:
145	spiga	1.30	WMS = 'OSG'
146	ewv	1.42	taskHash = sha.new(common._db.queryTask('name')).hexdigest()
147	ewv	1.43	jobId = str(id) + '_https://' + common.scheduler.name() + '/' + taskHash + '/' + str(id)
148	spiga	1.28	common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
149	ewv	1.42	elif common.scheduler.name().upper() in ['LSF','CAF']:
150			WMS = common.scheduler.name()
151	spiga	1.51	jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(task_unique_name,"_","-")
152	ewv	1.42	common.logger.debug(5,'JobID for ML monitoring is created for Local scheduler:'+jobId)
153	spiga	1.28	else:
154	ewv	1.42	jobId = str(id) + '_' + str(jid)
155			WMS = job.runningJob['service']
156			common.logger.debug(5,'JobID for ML monitoring is created for gLite scheduler:'+jobId)
157	spiga	1.28	pass
158
159			common.logger.debug(5,"sending info to ML")
160			params = {}
161			if WMS != None:
162			params = {'taskId': taskId, \
163			'jobId': jobId,\
164			'sid': str(jid), \
165			'StatusValueReason': job_status_reason, \
166			'StatusValue': jobStatus, \
167			'StatusEnterTime': job_last_time, \
168			'StatusDestination': dest, \
169			'RBname': WMS }
170			else:
171			params = {'taskId': taskId, \
172			'jobId': jobId,\
173			'sid': str(jid), \
174			'StatusValueReason': job_status_reason, \
175			'StatusValue': jobStatus, \
176			'StatusEnterTime': job_last_time, \
177			'StatusDestination': dest }
178			common.logger.debug(5,str(params))
179			common.apmon.sendToML(params)
180
181			return
182	ewv	1.33
183	spiga	1.24	def joinIntArray_(self,array) :
184			output = ''
185			for item in array :
186			output += str(item)+','
187			if output[-1] == ',' :
188			output = output[:-1]
189			return output