CRAB/python/Status.py

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

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

        if common.scheduler.name().upper() == 'CONDOR_G':
            # create hash of cfg file
            self.hash = makeCksum(common.work_space.cfgFileName())
        else:
            self.hash = ''

        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.compute()
        self.PrintReport_()
        stop = time.time()
        common.logger.debug(1, "Status Time: "+str(stop - start))
        common.logger.write("Status Time: "+str(stop - start))
        pass

    def compute(self):
        """
        compute the status
        """

        common.logger.message("Checking the status of all jobs: please wait")
        task = common._db.getTask()
        up_task = common.scheduler.queryEverything(task['id']) ## NeW BL--DS
        toPrint=[]
        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=''
        
#            if jobStatus == 'Done (Success)' or jobStatus == 'Cleared' or jobStatus == 'Done (Aborted)':
            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)
#           elif jobStatus == 'Created':
#               printline+="%-8s %-18s %-40s %-13s %-15s" % (bossid,'Created',dest,'','')
#               pass
#           else:
#               printline+="%-8s %-18s %-40s %-13s %-15s" % (bossid,jobStatus,dest,'','')
#               toPrint.append(printline)
            resFlag = 0

## Here to be implemented.. maybe putting stuff in a dedicated funcion.... better if not needed
#"""     
#            if jobStatus != 'Created'  and jobStatus != 'Unknown':
#                jid1 = string.strip(jobAttributes[bossid]['SCHED_ID'])
#
#                jobId = ''
#                if common.scheduler.name().upper() == 'CONDOR_G':
#                    jobId = str(bossid) + '_' + self.hash + '_' + string.strip(jobAttributes[bossid]['SCHED_ID'])
#                    common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
#                else:
#                    jobId = str(bossid) + '_' + string.strip(jobAttributes[bossid]['SCHED_ID'])
#                    if common.scheduler.name() == 'lsf' or common.scheduler.name() == 'caf':
#                        jobId=str(bossid)+"_https://"+common.scheduler.name()+":/"+string.strip(jobAttributes[bossid]['SCHED_ID'])+"-"+string.replace(common.taskDB.dict('taskId'),"_","-")
#                        common.logger.debug(5,'JobID for ML monitoring is created for LSF scheduler:'+jobId)
#                    pass
#                pass
#
#                common.logger.debug(5,"sending info to ML")
#                params = {}
#                if RB != None:
#                    params = {'taskId': common.taskDB.dict('taskId'), \
#                    'jobId': jobId,\
#                    'sid': string.strip(jobAttributes[bossid]['SCHED_ID']), \
#                    'StatusValueReason': job_status_reason, \
#                    'StatusValue': jobStatus, \
#                    'StatusEnterTime': job_last_time, \
#                    'StatusDestination': dest, \
#                    'RBname': RB }
#                else:
#                    params = {'taskId': common.taskDB.dict('taskId'), \
#                    'jobId': jobId,\
#                    'sid': string.strip(jobAttributes[bossid]['SCHED_ID']), \
#                    'StatusValueReason': job_status_reason, \
#                    'StatusValue': jobStatus, \
#                    'StatusEnterTime': job_last_time, \
#                    'StatusDestination': dest }
#                common.logger.debug(5,str(params))
#
#                common.apmon.sendToML(params)
##            if printline != '':
##                print printline

        self.detailedReport(toPrint)
  #      self.update_(for_summary)
        return

    def detailedReport(self, lines):

        counter = 0
        printline = ''
        printline+= "%-8s %-18s %-40s %-13s %-15s" % ('ID','STATUS','E_HOST','EXE_EXIT_CODE','JOB_EXIT_STATUS')
        print printline
        print '---------------------------------------------------------------------------------------------------'

        for i in range(len(lines)):
            if counter != 0 and counter%10 == 0 :
                print '---------------------------------------------------------------------------------------------------'
            print lines[i]
            counter += 1

    def PrintReport_(self):

        # query sui distinct statusScheduler
        #distinct_status =  common._db.queryDistJob('dlsDestination')
        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 joinIntArray_(self,array) :
        output = ''
        for item in array :
            output += str(item)+','
        if output[-1] == ',' :
            output = output[:-1]
        return output

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