CRAB/python/Status.py

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

class Status(Actor):
    def __init__(self, cfg_params, nj_list):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        self.countDone = 0
        self.countReady = 0
        self.countSched = 0
        self.countRun = 0
        self.countCleared = 0
        self.countToTjob = 0
        
        Status = crab_util.importName('edg_wl_userinterface_common_LbWrapper', 'Status')
        # Bypass edg-job-status interfacing directly to C++ API
        # Job attribute vector to retrieve status without edg-job-status
        self.level = 0
        # Instance of the Status class provided by LB API
        self.jobStat = Status()

        self.states = [ "Acl", "cancelReason", "cancelling","ce_node","children", \
          "children_hist","children_num","children_states","condorId","condor_jdl", \
          "cpuTime","destination", "done_code","exit_code","expectFrom", \
          "expectUpdate","globusId","jdl","jobId","jobtype", \
          "lastUpdateTime","localId","location", "matched_jdl","network_server", \
          "owner","parent_job", "reason","resubmitted","rsl","seed",\
          "stateEnterTime","stateEnterTimes","subjob_failed", \
          "user tags" , "status" , "status_code","hierarchy"]
        self.hstates = {}
        for key in self.states:
            self.hstates[key]=''

        return

    def run(self):
        """
        The main method of the class.
        """
        common.logger.debug(5, "Status::run() called")

        common.jobDB.load()
        for nj in self.nj_list:
            st = common.jobDB.status(nj)
            self.countToTjob = self.countToTjob + 1
            jid = common.jobDB.jobId(nj)
            if st == 'S':
                result = common.scheduler.queryStatus(jid)
                self.processResult_(nj, result)
                exit = ''
                if result == 'Done':
                    exit = common.scheduler.getExitStatus(jid)
                    pass
                print 'Job %03d:'%(nj+1),jid,result,exit
                pass
            else:
                exit = ''
                if st == 'D':
                    exit = common.scheduler.getExitStatus(jid)
                    pass
                print 'Job %03d:'%(nj+1),jid,crab_util.crabJobStatusToString(st),exit
                pass
            pass

        common.jobDB.save()

        self.Report_()
        pass


    def processResult_(self, nj, result):
        #######################################################################################
        # self.hstates['destination'] = self.hstates['destination'].strip()
        # destination = self.hstates['destination'].split(":")[0]
        # self.hstates['jobId'] =  self.hstates['jobId'].strip()
        # ID3 =  self.hstates['jobId'].split("/")[3]
        # brokTmp = self.hstates['jobId'].split("/")[2]
        # broker = brokTmp.split(":")[0]
        # self.hstates['destination'] =  self.hstates['destination'].strip()
        # destination =  self.hstates['destination'].split(":")[0]
        # resFlag = 0
        #######################################################################################
        
        ### TODO: set relevant status also to DB

        try: 
            if result == 'Done': 
                self.countDone = self.countDone + 1
                exCode = self.hstates['exit_code']
                common.jobDB.setStatus(nj, 'D')
                #statistic.notify('checkstatus',resFlag,exCode,dataset,owner,destination,broker,ID3,ID1,NJC)
            elif result == 'Ready':
                self.countReady = self.countReady + 1
                #statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
            elif result == 'Scheduled':
                self.countSched = self.countSched + 1
                #statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
            elif result == 'Running':
                self.countRun = self.countRun + 1
                #statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
            elif result == 'Aborted':
                common.jobDB.setStatus(nj, 'A')
                #job.saveJobStatus()
                #statistic.notify('checkstatus',resFlag,'abort',dataset,owner,destination,broker,ID3,ID1,NJC)
                pass
            elif result == 'Cancelled':
                common.jobDB.setStatus(nj, 'K')
                #job.setStatus('K')
                #job.saveJobStatus()
                #statistic.notify('checkstatus',resFlag,'cancel',dataset,owner,destination,broker,ID3,ID1,NJC)
                pass
            elif result == 'Cleared':
                #job.setStatus('P')
                exCode = self.hstates['exit_code']
                #statistic.notify('checkstatus',resFlag,exCode,dataset,owner,destination,broker,ID3,ID1,NJC)
                #job.saveJobStatus()
                self.countCleared = self.countCleared + 1
        except UnboundLocalError:
            common.logger.message('ERROR: UnboundLocalError with ')

    def Report_(self) :

        """ Report #jobs for each status  """  

        #job_stat = common.job_list.loadStatus()

        print ''
        print ">>>>>>>>> %i Total Jobs " % (self.countToTjob)

        if (self.countReady != 0):
            print ''
            print ">>>>>>>>> %i Jobs Ready" % (self.countReady)
        if (self.countSched != 0):
            print ''
            print ">>>>>>>>> %i Jobs Scheduled" % (self.countSched)
        if (self.countRun != 0):
            print ''
            print ">>>>>>>>> %i Jobs Running" % (self.countRun)
        if (self.countCleared != 0):
            print ''
            print ">>>>>>>>> %i Jobs Retrieved (=Cleared)" % (self.countCleared)
            print "          You can resubmit them specifying JOB numbers: crab.py -resubmit JOB_number (or range of JOB) -continue" 
            print "          (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"       
        # if job_stat[6] or job_stat[7]:
        #     print ''
        #     print ">>>>>>>>> %i Jobs aborted or killed(=cancelled by user)" % (job_stat[6] + job_stat[7])
        #     print "          Resubmit them with: crab.py -resubmit -continue to resubmit all" 
        #     print "          or specifying JOB numbers (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"       
        #     print "           "       
        if (self.countDone != 0):
            print ">>>>>>>>> %i Jobs Done" % (self.countDone)
            print "          Retrieve them with: crab.py -getoutput -continue to retrieve all" 
            print "          or specifying JOB numbers (i.e -getoutput 1-3 => 1 and 2 and 3 or -getoutput 1,3 => 1 and 3)"
            print('\n')  
        pass


    # def getJobStatus_(self, sid, attr):
    #     result = ''
    #     st = 0
    #     self.jobStat.getStatus(sid, self.level)
    #     (err, apiMsg) = self.jobStat.get_error()
    #     if err:
    #         common.logger.message(apiMsg)
    #         return None
    #     else:
    #         for i in range(len(self.states)):
    #             #print "states = ", self.states
    #             # Fill an hash table with all information retrieved from LB API
    #             self.hstates[ self.states[i] ] = self.jobStat.loadStatus(st)[i]
    #         result = self.jobStat.loadStatus(st)[ self.states.index(attr) ]
    #     return result
Revision:	1.8
Committed:	Tue Aug 23 12:42:05 2005 UTC (19 years, 8 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_0_0_pre3
Changes since 1.7:	+9 -9 lines
Log Message:	recomment...
#	User	Rev	Content
1	slacapra	1.1	from Actor import *
2			import common, crab_util
3			import string, os
4
5			class Status(Actor):
6			def __init__(self, cfg_params, nj_list):
7			self.cfg_params = cfg_params
8			self.nj_list = nj_list
9			self.countDone = 0
10			self.countReady = 0
11			self.countSched = 0
12			self.countRun = 0
13			self.countCleared = 0
14			self.countToTjob = 0
15
16			Status = crab_util.importName('edg_wl_userinterface_common_LbWrapper', 'Status')
17			# Bypass edg-job-status interfacing directly to C++ API
18			# Job attribute vector to retrieve status without edg-job-status
19			self.level = 0
20			# Instance of the Status class provided by LB API
21			self.jobStat = Status()
22
23			self.states = [ "Acl", "cancelReason", "cancelling","ce_node","children", \
24			"children_hist","children_num","children_states","condorId","condor_jdl", \
25			"cpuTime","destination", "done_code","exit_code","expectFrom", \
26			"expectUpdate","globusId","jdl","jobId","jobtype", \
27			"lastUpdateTime","localId","location", "matched_jdl","network_server", \
28	slacapra	1.5	"owner","parent_job", "reason","resubmitted","rsl","seed",\
29			"stateEnterTime","stateEnterTimes","subjob_failed", \
30	slacapra	1.1	"user tags" , "status" , "status_code","hierarchy"]
31			self.hstates = {}
32	slacapra	1.6	for key in self.states:
33			self.hstates[key]=''
34	slacapra	1.1
35			return
36
37			def run(self):
38			"""
39			The main method of the class.
40			"""
41			common.logger.debug(5, "Status::run() called")
42
43			common.jobDB.load()
44			for nj in self.nj_list:
45			st = common.jobDB.status(nj)
46			self.countToTjob = self.countToTjob + 1
47			jid = common.jobDB.jobId(nj)
48			if st == 'S':
49	slacapra	1.5	result = common.scheduler.queryStatus(jid)
50	slacapra	1.1	self.processResult_(nj, result)
51	slacapra	1.4	exit = ''
52			if result == 'Done':
53	slacapra	1.5	exit = common.scheduler.getExitStatus(jid)
54	slacapra	1.4	pass
55			print 'Job %03d:'%(nj+1),jid,result,exit
56	slacapra	1.1	pass
57			else:
58	slacapra	1.4	exit = ''
59			if st == 'D':
60	slacapra	1.5	exit = common.scheduler.getExitStatus(jid)
61	slacapra	1.4	pass
62			print 'Job %03d:'%(nj+1),jid,crab_util.crabJobStatusToString(st),exit
63	slacapra	1.1	pass
64			pass
65
66	slacapra	1.2	common.jobDB.save()
67
68	slacapra	1.1	self.Report_()
69			pass
70
71
72			def processResult_(self, nj, result):
73			#######################################################################################
74	slacapra	1.8	# self.hstates['destination'] = self.hstates['destination'].strip()
75			# destination = self.hstates['destination'].split(":")[0]
76			# self.hstates['jobId'] = self.hstates['jobId'].strip()
77			# ID3 = self.hstates['jobId'].split("/")[3]
78			# brokTmp = self.hstates['jobId'].split("/")[2]
79			# broker = brokTmp.split(":")[0]
80			# self.hstates['destination'] = self.hstates['destination'].strip()
81			# destination = self.hstates['destination'].split(":")[0]
82			# resFlag = 0
83	slacapra	1.1	#######################################################################################
84	slacapra	1.3
85			### TODO: set relevant status also to DB
86
87	slacapra	1.1	try:
88			if result == 'Done':
89			self.countDone = self.countDone + 1
90			exCode = self.hstates['exit_code']
91	slacapra	1.3	common.jobDB.setStatus(nj, 'D')
92	slacapra	1.1	#statistic.notify('checkstatus',resFlag,exCode,dataset,owner,destination,broker,ID3,ID1,NJC)
93			elif result == 'Ready':
94			self.countReady = self.countReady + 1
95			#statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
96			elif result == 'Scheduled':
97			self.countSched = self.countSched + 1
98			#statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
99			elif result == 'Running':
100			self.countRun = self.countRun + 1
101			#statistic.notify('checkstatus',resFlag,'-----',dataset,owner,destination,broker,ID3,ID1,NJC)
102			elif result == 'Aborted':
103	slacapra	1.2	common.jobDB.setStatus(nj, 'A')
104	slacapra	1.1	#job.saveJobStatus()
105			#statistic.notify('checkstatus',resFlag,'abort',dataset,owner,destination,broker,ID3,ID1,NJC)
106			pass
107			elif result == 'Cancelled':
108	slacapra	1.3	common.jobDB.setStatus(nj, 'K')
109	slacapra	1.1	#job.setStatus('K')
110			#job.saveJobStatus()
111			#statistic.notify('checkstatus',resFlag,'cancel',dataset,owner,destination,broker,ID3,ID1,NJC)
112			pass
113			elif result == 'Cleared':
114			#job.setStatus('P')
115			exCode = self.hstates['exit_code']
116			#statistic.notify('checkstatus',resFlag,exCode,dataset,owner,destination,broker,ID3,ID1,NJC)
117			#job.saveJobStatus()
118			self.countCleared = self.countCleared + 1
119			except UnboundLocalError:
120			common.logger.message('ERROR: UnboundLocalError with ')
121
122			def Report_(self) :
123
124			""" Report #jobs for each status """
125
126			#job_stat = common.job_list.loadStatus()
127
128			print ''
129			print ">>>>>>>>> %i Total Jobs " % (self.countToTjob)
130
131			if (self.countReady != 0):
132			print ''
133			print ">>>>>>>>> %i Jobs Ready" % (self.countReady)
134			if (self.countSched != 0):
135			print ''
136			print ">>>>>>>>> %i Jobs Scheduled" % (self.countSched)
137			if (self.countRun != 0):
138			print ''
139			print ">>>>>>>>> %i Jobs Running" % (self.countRun)
140			if (self.countCleared != 0):
141			print ''
142			print ">>>>>>>>> %i Jobs Retrieved (=Cleared)" % (self.countCleared)
143			print " You can resubmit them specifying JOB numbers: crab.py -resubmit JOB_number (or range of JOB) -continue"
144			print " (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"
145			# if job_stat[6] or job_stat[7]:
146			# print ''
147			# print ">>>>>>>>> %i Jobs aborted or killed(=cancelled by user)" % (job_stat[6] + job_stat[7])
148			# print " Resubmit them with: crab.py -resubmit -continue to resubmit all"
149			# print " or specifying JOB numbers (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"
150			# print " "
151			if (self.countDone != 0):
152			print ">>>>>>>>> %i Jobs Done" % (self.countDone)
153			print " Retrieve them with: crab.py -getoutput -continue to retrieve all"
154			print " or specifying JOB numbers (i.e -getoutput 1-3 => 1 and 2 and 3 or -getoutput 1,3 => 1 and 3)"
155			print('\n')
156			pass
157
158
159	slacapra	1.5	# def getJobStatus_(self, sid, attr):
160			# result = ''
161			# st = 0
162			# self.jobStat.getStatus(sid, self.level)
163			# (err, apiMsg) = self.jobStat.get_error()
164			# if err:
165			# common.logger.message(apiMsg)
166			# return None
167			# else:
168			# for i in range(len(self.states)):
169			# #print "states = ", self.states
170			# # Fill an hash table with all information retrieved from LB API
171			# self.hstates[ self.states[i] ] = self.jobStat.loadStatus(st)[i]
172			# result = self.jobStat.loadStatus(st)[ self.states.index(attr) ]
173			# return result