CRAB/python/Status.py

from Actor import *
import common
import Statistic

class Status(Actor):
    def __init__(self, cfg_params, nj_list=[]):
        if nj_list==[]:
            self.nj_list = range(len(common.job_list))
        else:
            self.nj_list = nj_list
        self.countDone = 0
        self.countReady = 0
        self.countSched = 0
        self.countRun = 0
        self.countAbort = 0
        self.countCancel = 0
        self.countCleared = 0
        self.countToTjob = 0
        self.cfg_params = cfg_params
        

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

        self.compute()
        self.PrintReport_()
        pass

    def status(self) :
        """ Return #jobs for each status as a tuple"""
        return (self.countToTjob,self.countReady,self.countSched,self.countRun,self.countCleared,self.countAbort,self.countCancel,self.countDone)

    def compute(self):
        """
        Update the status to DB
        """
        bossTaskId = common.taskDB.dict('BossTaskId')
        statusList = common.scheduler.queryStatusList(bossTaskId, self.nj_list)

        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' or st == 'A' or st == 'D' or st == 'K':
                #result = common.scheduler.queryStatus(jid)
                result = statusList[jid]
                self.processResult_(nj, result, jid)
                exitStatus = common.jobDB.exitStatus(nj)
                print 'Job %03d:'%(nj+1),jid,result,exitStatus
                dest = common.scheduler.queryDest(jid)
                if ( dest.find(":") != -1 ) :
                    dest = dest.split(":")[0]
                if int(self.cfg_params['USER.activate_monalisa']) == 1:
                    params = {'taskId': 'JobStatus', 'jobId': jid, 'StatusValueReason': common.scheduler.getStatusAttribute_(jid, 'reason'), \
                    'StatusValue': st, 'StatusEnterTime': common.scheduler.getStatusAttribute_(jid, 'stateEnterTime'), 'StatusDestination': dest}
#                    self.cfg_params['apmon'].fillDict({'taskId': 'JobStatus', 'jobId': jid, \
#                                                   'StatusValueReason': common.scheduler.getStatusAttribute_(jid, 'reason'), \
#                                                   'StatusValue': st, 'StatusEnterTime': common.scheduler.getStatusAttribute_(jid, 'stateEnterTime'), 'StatusDestination': dest})
                    self.cfg_params['apmon'].sendToML(params)
                pass
            else:
                exitStatus = common.jobDB.exitStatus(nj)
                #print 'Job %03d:'%(nj+1),jid,crab_util.crabJobStatusToString(st),exitStatus
                pass

        common.jobDB.save()
        pass

    def processResult_(self, nj, result,jid):

        destination = common.scheduler.queryDest(jid)
        if ( destination.find(":") != -1 ) :
            destination = destination.split(":")[0]
            
        if ( jid.find(":") != -1 ) :
            #ID3 =  jid.split("/")[3]
            broker = jid.split("/")[2].split(":")[0]
        else :
            #ID3 = jid
            broker = 'OSG'
            
        resFlag = 0
        ### TODO: set relevant status also to DB

        try: 
            if result == 'Done': 
                self.countDone = self.countDone + 1
                exCode = common.scheduler.getExitStatus(jid)
                common.jobDB.setStatus(nj, 'D')
                jid = common.jobDB.jobId(nj)
                exitStatus = common.scheduler.getExitStatus(jid)
                common.jobDB.setExitStatus(nj, exitStatus)
                Statistic.Monitor('checkstatus',resFlag,jid,exCode,broker)
            elif result == 'Ready':
                self.countReady = self.countReady + 1
                Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
            elif result == 'Scheduled':
                self.countSched = self.countSched + 1
                Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
            elif result == 'Running':
                self.countRun = self.countRun + 1
                Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
            elif result == 'Aborted':
                common.jobDB.setStatus(nj, 'A')
                self.countAbort = self.countAbort + 1
                Statistic.Monitor('checkstatus',resFlag,jid,'abort',broker)
                pass
            elif result == 'Cancelled':
                common.jobDB.setStatus(nj, 'K')
                self.countCancel = self.countCancel + 1
                Statistic.Monitor('checkstatus',resFlag,jid,'cancel',broker)
                pass
            elif result == 'Cleared':
                exCode = common.scheduler.getExitStatus(jid) 
                Statistic.Monitor('checkstatus',resFlag,jid,exCode,broker) 
                self.countCleared = self.countCleared + 1
        except UnboundLocalError:
            common.logger.message('ERROR: UnboundLocalError with ')

    def PrintReport_(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

Revision:	1.21
Committed:	Wed Jan 17 18:17:59 2007 UTC (18 years, 3 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_0_4, CRAB_2_0_4_pre2, CRAB_2_0_4_pre1, CRAB_2_0_3, CRAB_2_0_3_pre1, CRAB_2_0_2, CRAB_2_0_2_pre6, CRAB_2_0_2_pre5, CRAB_2_0_2_pre4, CRAB_2_0_2_pre3, CRAB_1_5_4_SLC3, CRAB_1_5_4_SLC3_pre4, CRAB_2_0_2_pre2, CRAB_2_0_2_pre1, CRAB_1_5_4_SLC3_pre3, CRAB_2_0_1, CRAB_1_5_4_SLC3_pre2, CRAB_2_0_1_pre1, CRAB_1_5_4_SLC3_pre1, CRAB_2_0_0, CRAB_2_0_0_pre10, CRAB_2_0_0_pre9, CRAB_1_5_4, CRAB_1_5_4_pre2, CRAB_1_5_4_pre1, CRAB_2_0_0_pre7, CRAB_2_0_0_pre6, CRAB_1_5_3, CRAB_1_5_3_pre5, CRAB_1_5_3_pre4, CRAB_2_0_0_pre5, CRAB_1_5_3_pre3, configure, CRAB_2_0_0_pre4, CRAB_1_5_3_pre2, CRAB_1_5_3_pre1, CRAB_2_0_0_pre3, CRAB_1_5_2, CRAB_2_0_0_pre2, CRAB_2_0_0_pre1, CRAB_1_5_1, CRAB_1_5_1_pre4, CRAB_1_5_1_pre3, CRAB_1_5_1_pre2, CRAB_1_5_1_pre1, CRAB_1_5_0, CRAB_1_5_0_pre9, CRAB_1_5_0_pre8, CRAB_1_5_0_pre7, CRAB_1_5_0_pre6, CRAB_1_5_0_pre5, CRAB_1_5_0_pre4
Branch point for:	CRAB_1_5_4_SLC3_pre4_br, CRAB_1_5_4_SLC3_start
Changes since 1.20:	+18 -19 lines
Log Message:	many minor fixes reported by pychecker, mostly unsude import and unused variables
#	User	Rev	Content
1	slacapra	1.1	from Actor import *
2	slacapra	1.21	import common
3	spiga	1.9	import Statistic
4	slacapra	1.1
5			class Status(Actor):
6	corvo	1.16	def __init__(self, cfg_params, nj_list=[]):
7	slacapra	1.15	if nj_list==[]:
8			self.nj_list = range(len(common.job_list))
9			else:
10			self.nj_list = nj_list
11	slacapra	1.1	self.countDone = 0
12			self.countReady = 0
13			self.countSched = 0
14			self.countRun = 0
15	slacapra	1.15	self.countAbort = 0
16			self.countCancel = 0
17	slacapra	1.1	self.countCleared = 0
18			self.countToTjob = 0
19	corvo	1.16	self.cfg_params = cfg_params
20	slacapra	1.1
21	spiga	1.9
22	gutsche	1.18	#Status = crab_util.importName('edg_wl_userinterface_common_LbWrapper', 'Status')
23	slacapra	1.1	# Bypass edg-job-status interfacing directly to C++ API
24			# Job attribute vector to retrieve status without edg-job-status
25			self.level = 0
26			# Instance of the Status class provided by LB API
27	gutsche	1.18	#self.jobStat = Status()
28	slacapra	1.1
29			self.states = [ "Acl", "cancelReason", "cancelling","ce_node","children", \
30			"children_hist","children_num","children_states","condorId","condor_jdl", \
31			"cpuTime","destination", "done_code","exit_code","expectFrom", \
32			"expectUpdate","globusId","jdl","jobId","jobtype", \
33			"lastUpdateTime","localId","location", "matched_jdl","network_server", \
34	slacapra	1.5	"owner","parent_job", "reason","resubmitted","rsl","seed",\
35			"stateEnterTime","stateEnterTimes","subjob_failed", \
36	slacapra	1.1	"user tags" , "status" , "status_code","hierarchy"]
37			self.hstates = {}
38	slacapra	1.6	for key in self.states:
39			self.hstates[key]=''
40	slacapra	1.1
41			return
42
43			def run(self):
44			"""
45			The main method of the class.
46			"""
47			common.logger.debug(5, "Status::run() called")
48
49	slacapra	1.15	self.compute()
50			self.PrintReport_()
51			pass
52
53			def status(self) :
54			""" Return #jobs for each status as a tuple"""
55			return (self.countToTjob,self.countReady,self.countSched,self.countRun,self.countCleared,self.countAbort,self.countCancel,self.countDone)
56
57			def compute(self):
58			"""
59			Update the status to DB
60			"""
61	slacapra	1.20	bossTaskId = common.taskDB.dict('BossTaskId')
62	slacapra	1.21	statusList = common.scheduler.queryStatusList(bossTaskId, self.nj_list)
63	slacapra	1.15
64	slacapra	1.1	common.jobDB.load()
65			for nj in self.nj_list:
66			st = common.jobDB.status(nj)
67			self.countToTjob = self.countToTjob + 1
68			jid = common.jobDB.jobId(nj)
69	gutsche	1.18	if st == 'S' or st == 'A' or st == 'D' or st == 'K':
70	slacapra	1.20	#result = common.scheduler.queryStatus(jid)
71			result = statusList[jid]
72	slacapra	1.11	self.processResult_(nj, result, jid)
73	slacapra	1.21	exitStatus = common.jobDB.exitStatus(nj)
74			print 'Job %03d:'%(nj+1),jid,result,exitStatus
75	gutsche	1.18	dest = common.scheduler.queryDest(jid)
76			if ( dest.find(":") != -1 ) :
77	slacapra	1.21	dest = dest.split(":")[0]
78	corvo	1.16	if int(self.cfg_params['USER.activate_monalisa']) == 1:
79	corvo	1.19	params = {'taskId': 'JobStatus', 'jobId': jid, 'StatusValueReason': common.scheduler.getStatusAttribute_(jid, 'reason'), \
80			'StatusValue': st, 'StatusEnterTime': common.scheduler.getStatusAttribute_(jid, 'stateEnterTime'), 'StatusDestination': dest}
81			# self.cfg_params['apmon'].fillDict({'taskId': 'JobStatus', 'jobId': jid, \
82			# 'StatusValueReason': common.scheduler.getStatusAttribute_(jid, 'reason'), \
83			# 'StatusValue': st, 'StatusEnterTime': common.scheduler.getStatusAttribute_(jid, 'stateEnterTime'), 'StatusDestination': dest})
84			self.cfg_params['apmon'].sendToML(params)
85	slacapra	1.1	pass
86			else:
87	slacapra	1.21	exitStatus = common.jobDB.exitStatus(nj)
88			#print 'Job %03d:'%(nj+1),jid,crab_util.crabJobStatusToString(st),exitStatus
89	slacapra	1.1	pass
90	corvo	1.16
91	slacapra	1.2	common.jobDB.save()
92	slacapra	1.1	pass
93
94	spiga	1.9	def processResult_(self, nj, result,jid):
95
96	gutsche	1.18	destination = common.scheduler.queryDest(jid)
97			if ( destination.find(":") != -1 ) :
98			destination = destination.split(":")[0]
99
100			if ( jid.find(":") != -1 ) :
101	slacapra	1.21	#ID3 = jid.split("/")[3]
102	gutsche	1.18	broker = jid.split("/")[2].split(":")[0]
103			else :
104	slacapra	1.21	#ID3 = jid
105	gutsche	1.18	broker = 'OSG'
106
107	spiga	1.9	resFlag = 0
108	slacapra	1.3	### TODO: set relevant status also to DB
109
110	slacapra	1.1	try:
111			if result == 'Done':
112			self.countDone = self.countDone + 1
113	spiga	1.9	exCode = common.scheduler.getExitStatus(jid)
114	slacapra	1.3	common.jobDB.setStatus(nj, 'D')
115	slacapra	1.10	jid = common.jobDB.jobId(nj)
116	slacapra	1.21	exitStatus = common.scheduler.getExitStatus(jid)
117			common.jobDB.setExitStatus(nj, exitStatus)
118			Statistic.Monitor('checkstatus',resFlag,jid,exCode,broker)
119	slacapra	1.1	elif result == 'Ready':
120			self.countReady = self.countReady + 1
121	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
122	slacapra	1.1	elif result == 'Scheduled':
123			self.countSched = self.countSched + 1
124	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
125	slacapra	1.1	elif result == 'Running':
126			self.countRun = self.countRun + 1
127	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,'-----',broker)
128	slacapra	1.1	elif result == 'Aborted':
129	slacapra	1.2	common.jobDB.setStatus(nj, 'A')
130	slacapra	1.15	self.countAbort = self.countAbort + 1
131	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,'abort',broker)
132	slacapra	1.1	pass
133			elif result == 'Cancelled':
134	slacapra	1.3	common.jobDB.setStatus(nj, 'K')
135	slacapra	1.15	self.countCancel = self.countCancel + 1
136	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,'cancel',broker)
137	slacapra	1.1	pass
138			elif result == 'Cleared':
139	spiga	1.9	exCode = common.scheduler.getExitStatus(jid)
140	slacapra	1.21	Statistic.Monitor('checkstatus',resFlag,jid,exCode,broker)
141	slacapra	1.1	self.countCleared = self.countCleared + 1
142			except UnboundLocalError:
143			common.logger.message('ERROR: UnboundLocalError with ')
144
145	slacapra	1.15	def PrintReport_(self) :
146	slacapra	1.1
147			""" Report #jobs for each status """
148
149			#job_stat = common.job_list.loadStatus()
150
151			print ''
152			print ">>>>>>>>> %i Total Jobs " % (self.countToTjob)
153
154			if (self.countReady != 0):
155			print ''
156			print ">>>>>>>>> %i Jobs Ready" % (self.countReady)
157			if (self.countSched != 0):
158			print ''
159			print ">>>>>>>>> %i Jobs Scheduled" % (self.countSched)
160			if (self.countRun != 0):
161			print ''
162			print ">>>>>>>>> %i Jobs Running" % (self.countRun)
163			if (self.countCleared != 0):
164			print ''
165			print ">>>>>>>>> %i Jobs Retrieved (=Cleared)" % (self.countCleared)
166			print " You can resubmit them specifying JOB numbers: crab.py -resubmit JOB_number (or range of JOB) -continue"
167			print " (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"
168			# if job_stat[6] or job_stat[7]:
169			# print ''
170			# print ">>>>>>>>> %i Jobs aborted or killed(=cancelled by user)" % (job_stat[6] + job_stat[7])
171			# print " Resubmit them with: crab.py -resubmit -continue to resubmit all"
172			# print " or specifying JOB numbers (i.e -resubmit 1-3 => 1 and 2 and 3 or -resubmit 1,3 => 1 and 3)"
173			# print " "
174			if (self.countDone != 0):
175			print ">>>>>>>>> %i Jobs Done" % (self.countDone)
176			print " Retrieve them with: crab.py -getoutput -continue to retrieve all"
177			print " or specifying JOB numbers (i.e -getoutput 1-3 => 1 and 2 and 3 or -getoutput 1,3 => 1 and 3)"
178			print('\n')
179			pass
180