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(str(up_task['name']).split('_')[:-1]))
        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)
            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):

        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':
                list_ID = common._db.queryAttrRunJob({'wrapperReturnCode':c},'jobId')
                if len(list_ID)>0:
                    print ">>>>>>>>> %i Jobs with Wrapper Exit Code : %s " % (len(list_ID), str(c))#,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" % self.readableList(list_ID)
                    print " "

    def readableList(self,rawList):
      listString = str(rawList[0])
      endRange = ''
      for i in range(1,len(rawList)):
        if rawList[i] == rawList[i-1]+1:
          endRange = str(rawList[i])
        else:
          if endRange:
            listString += '-' + endRange + ',' + str(rawList[i])
            endRange = ''
          else:
            listString += ',' + str(rawList[i])
      if endRange:
        listString += '-' + endRange
        endRange = ''

      return listString


    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.38
Committed:	Thu Jun 12 18:53:12 2008 UTC (16 years, 10 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_3_1_pre3, CRAB_2_3_1_pre2, CRAB_2_3_1_pre1, CRAB_2_3_0, CRAB_2_3_0_pre6, CRAB_2_3_0_pre1
Branch point for:	CRAB_2_3_0_br
Changes since 1.37:	+9 -8 lines
Log Message:	first check for empty code in status summary
#	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	ewv	1.33	upTask = common.scheduler.queryEverything(task['id'])
35	spiga	1.28	self.compute(upTask)
36
37			def compute(self, up_task):
38	ewv	1.33
39	spiga	1.24	toPrint=[]
40	spiga	1.32	taskId= str("_".join(str(up_task['name']).split('_')[:-1]))
41	spiga	1.37	self.wrapErrorList = []
42	spiga	1.25	for job in up_task.jobs :
43	spiga	1.35	id = str(job.runningJob['jobId'])
44	spiga	1.24	jobStatus = str(job.runningJob['statusScheduler'])
45			dest = str(job.runningJob['destination']).split(':')[0]
46			exe_exit_code = str(job.runningJob['applicationReturnCode'])
47	ewv	1.33	job_exit_code = str(job.runningJob['wrapperReturnCode'])
48	spiga	1.37	self.wrapErrorList.append(job_exit_code)
49	spiga	1.24	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	ewv	1.33
60	spiga	1.28	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	spiga	1.37	jobs = common._db.nJobs('list')
70
71			WrapExitCode = list(set(self.wrapErrorList))
72	slacapra	1.1
73	spiga	1.24	print ''
74	spiga	1.37	print ">>>>>>>>> %i Total Jobs " % (len(jobs))
75	spiga	1.24	print ''
76	ewv	1.33	list_ID=[]
77	spiga	1.37	for c in WrapExitCode:
78	spiga	1.38	if c != 'None':
79			list_ID = common._db.queryAttrRunJob({'wrapperReturnCode':c},'jobId')
80			if len(list_ID)>0:
81			print ">>>>>>>>> %i Jobs with Wrapper Exit Code : %s " % (len(list_ID), str(c))#,len(list_ID)
82			# if st == 'killed' or st == 'Aborted': print " You can resubmit them specifying JOB numbers: crab -resubmit JOB_number <Jobs list>"
83			# if st == 'Done' : print " Retrieve them with: crab -getoutput <Jobs list>"
84			# if st == 'Cleared': print " %i Jobs with EXE_EXIT_CODE: %s" % (len(common._db.queryDistJob('wrapperReturnCode')))
85			print " List of jobs: %s" % self.readableList(list_ID)
86			print " "
87	spiga	1.24
88	ewv	1.33	def readableList(self,rawList):
89			listString = str(rawList[0])
90			endRange = ''
91			for i in range(1,len(rawList)):
92			if rawList[i] == rawList[i-1]+1:
93			endRange = str(rawList[i])
94			else:
95			if endRange:
96	ewv	1.34	listString += '-' + endRange + ',' + str(rawList[i])
97	ewv	1.33	endRange = ''
98			else:
99	ewv	1.34	listString += ',' + str(rawList[i])
100	ewv	1.33	if endRange:
101			listString += '-' + endRange
102			endRange = ''
103
104			return listString
105
106	slacapra	1.1
107	spiga	1.28	def dataToDash(self,job,id,taskId,dest,jobStatus):
108	ewv	1.33
109	slacapra	1.1
110	spiga	1.28	jid = job.runningJob['schedulerId']
111			job_status_reason = str(job.runningJob['statusReason'])
112			job_last_time = str(job.runningJob['startTime'])
113			if common.scheduler.name().upper() == 'CONDOR_G':
114	spiga	1.30	WMS = 'OSG'
115	spiga	1.28	self.hash = makeCksum(common.work_space.cfgFileName())
116			jobId = str(id) + '_' + self.hash + '_' + str(jid)
117			common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
118			else:
119			if common.scheduler.name() in ['lsf','caf']:
120	spiga	1.30	WMS = common.scheduler.name()
121			jobId=str(id)+"_https://"+common.scheduler.name()+":/"+str(jid)+"-"+string.replace(taskId,"_","-")
122	spiga	1.28	common.logger.debug(5,'JobID for ML monitoring is created for Local scheduler:'+jobId)
123	spiga	1.30	else:
124			jobId = str(id) + '_' + str(jid)
125			WMS = job.runningJob['service']
126			common.logger.debug(5,'JobID for ML monitoring is created for gLite scheduler:'+jobId)
127	spiga	1.28	pass
128			pass
129
130			common.logger.debug(5,"sending info to ML")
131			params = {}
132			if WMS != None:
133			params = {'taskId': taskId, \
134			'jobId': jobId,\
135			'sid': str(jid), \
136			'StatusValueReason': job_status_reason, \
137			'StatusValue': jobStatus, \
138			'StatusEnterTime': job_last_time, \
139			'StatusDestination': dest, \
140			'RBname': WMS }
141			else:
142			params = {'taskId': taskId, \
143			'jobId': jobId,\
144			'sid': str(jid), \
145			'StatusValueReason': job_status_reason, \
146			'StatusValue': jobStatus, \
147			'StatusEnterTime': job_last_time, \
148			'StatusDestination': dest }
149			common.logger.debug(5,str(params))
150			common.apmon.sendToML(params)
151
152			return
153	ewv	1.33
154	spiga	1.24	def joinIntArray_(self,array) :
155			output = ''
156			for item in array :
157			output += str(item)+','
158			if output[-1] == ',' :
159			output = output[:-1]
160			return output
161	slacapra	1.1