CRAB/python/StatusServer.py

from Actor import *
from crab_util import *
import common
from ApmonIf import ApmonIf
import Statistic
import time
from ProgressBar import ProgressBar
from TerminalController import TerminalController

import xml.dom.minidom
import xml.dom.ext
import TaskDB

class StatusServer(Actor):
 
    def __init__(self, cfg_params,):
        self.cfg_params = cfg_params

        self.countSubmit = 0
        self.countSubmitting = 0
        self.countDone = 0
        self.countReady = 0
        self.countSched = 0
        self.countRun = 0
        self.countAbort = 0
        self.countCancel = 0
        self.countCleared = 0
        self.countToTjob = 0

        try:  
            self.server_name = self.cfg_params['CRAB.server_name'] # gsiftp://pcpg01.cern.ch/data/SEDir/
        except KeyError:
            msg = 'No server selected ...' 
            msg = msg + 'Please specify a server in the crab cfg file' 
            raise CrabException(msg) 

        return

    def translateStatus(self, status):
        """
        simmetric as server
        """

        stateConverting = {'Running': 'R', 'Aborted': 'A', 'Done': 'D', 'Done (Failed)': 'D',\
                           'Cleared': 'D', 'Cancelled': 'K', 'Killed': 'K'}

        if status in stateConverting:
            return stateConverting[status]
        return None


    def run(self):
        """
        The main method of the class: check the status of the task
        """
        common.logger.debug(5, "status server::run() called")
        start = time.time()

        totalCreatedJobs = 0
        flagSubmit = 1
        for nj in range(common.jobDB.nJobs()):
            if (common.jobDB.status(nj)!='S'):
                totalCreatedJobs +=1
        #        flagSubmit = 0

        if not flagSubmit:
            common.logger.message("Not all jobs are submitted: before checking the status submit all the jobs.")
            return

        common.scheduler.checkProxy()

        common.taskDB.load()
        WorkDirName =os.path.basename(os.path.split(common.work_space.topDir())[0])
        projectUniqName = 'crab_'+str(WorkDirName)+'_'+common.taskDB.dict('TasKUUID')     
        try: 
            common.logger.message ("Checking the status...\n")
            cmd = 'lcg-cp --vo cms  gsiftp://' + str(self.server_name) + str(projectUniqName)+'/res/xmlReportFile.xml file://'+common.work_space.resDir()+'xmlReportFile.xml'
            common.logger.debug(6, cmd)
            os.system(cmd +' >& /dev/null')  

        except: 
            msg = ("task status not yet available")
            raise CrabException(msg)

        try:     
            file = open(common.work_space.resDir()+"xmlReportFile.xml", "r")
            doc = xml.dom.minidom.parse(common.work_space.resDir()+ "xmlReportFile.xml" )
        
        except: 
            msg = ("problems reading report file")
            raise CrabException(msg)

        ###  <Job status='Submitted' job_exit='NULL' id='1' exe_exit='NULL'/>

        task     = doc.childNodes[0].childNodes[1].getAttribute("taskName")
        self.countToTjob = int(doc.childNodes[0].childNodes[1].getAttribute("totJob") )
       
        addTree = 3

        common.jobDB.load()

        if doc.childNodes[0].childNodes[3].getAttribute("id") == "all":
            if doc.childNodes[0].childNodes[3].getAttribute("status") == "Submitted":
                self.countSubmitting = common.jobDB.nJobs()
                for nj in range(common.jobDB.nJobs()):
                    common.jobDB.setStatus(nj, 'S')
            else:
                self.countAbort = common.jobDB.nJobs()
                for nj in range(common.jobDB.nJobs()):
                    common.jobDB.setStatus(nj, 'A')
            self.countToTjob = common.jobDB.nJobs()
        else:
            printline = ''
            printline+= "%-10s %-20s %-20s %-25s" % ('JOBID','STATUS','EXE_EXIT_CODE','JOB_EXIT_STATUS')
            print printline
            print '-------------------------------------------------------------------------------------'

            for job in range( self.countToTjob ):
                idJob = doc.childNodes[0].childNodes[job+addTree].getAttribute("id")
                stato = doc.childNodes[0].childNodes[job+addTree].getAttribute("status")
                exe_exit_code = doc.childNodes[0].childNodes[job+addTree].getAttribute("job_exit")
                job_exit_status = doc.childNodes[0].childNodes[job+addTree].getAttribute("exe_exit")
                jobDbStatus = self.translateStatus(stato)
 
                if jobDbStatus != None:
                    common.logger.debug(5, '*** Updating jobdb for job %s ***' %idJob)
                    if common.jobDB.status( str(int(idJob)-1) ) != "Y":
                        common.jobDB.setStatus( str(int(idJob)-1), self.translateStatus(stato) )
                    else:
                        stato = "Cleared"
                    common.jobDB.setExitStatus(  str(int(idJob)-1), job_exit_status )
                if stato != "Done" and stato != "Cleared" and stato != "Aborted" and stato != "Done (Failed)":
                    print "%-10s %-20s" % (idJob,stato)
                else:
                    print "%-10s %-20s %-20s %-25s" % (idJob,stato,exe_exit_code,job_exit_status)

                if stato == 'Running':
                    self.countRun += 1
                elif stato == 'Aborted':
                    self.countAbort += 1
                elif stato == 'Done':
                    self.countDone += 1
                elif stato == 'Cancelled':
                    self.countCancel += 1
                elif stato == 'Submitted':
                    self.countSubmit += 1
                elif stato == 'Submitting':
                    self.countSubmitting += 1
                elif stato == 'Ready':
                    self.countReady += 1
                elif stato == 'Scheduled':
                    self.countSched += 1
                elif stato == 'Cleared':
                    self.countCleared += 1

                addTree += 1
        common.jobDB.save()

        self.PrintReport_()


    def PrintReport_(self) :

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


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

        if (self.countReady != 0):
            print ">>>>>>>>> %i Jobs Ready" % (self.countReady)
        if (self.countSubmitting != 0) :
            print ">>>>>>>>> %i Jobs Submitting by the server" % (self.countSubmitting)
        if (self.countSubmit != 0):
            print ">>>>>>>>> %i Jobs Submitted" % (self.countSubmit)
        if (self.countAbort != 0):
            print ">>>>>>>>> %i Jobs Aborted" % (self.countAbort)
        if (self.countSched != 0):
            print ">>>>>>>>> %i Jobs Scheduled" % (self.countSched)
        if (self.countRun != 0):
            print ">>>>>>>>> %i Jobs Running" % (self.countRun)
        if (self.countCleared != 0):
            print ">>>>>>>>> %i Jobs Cleared" % (self.countRun)
        if (self.countDone != 0):
            print ">>>>>>>>> %i Jobs Done" % (self.countDone)
            print "          Retrieve them with: crab.py -getoutput -continue"
        print ''
        pass

Revision:	1.4
Committed:	Wed May 23 07:15:55 2007 UTC (17 years, 11 months ago) by mcinquil
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_5_2, CRAB_2_0_0_pre1, CRAB_1_5_1
Changes since 1.3:	+2 -2 lines
Log Message:	More detailed status; bug fix
#	User	Rev	Content
1	spiga	1.1	from Actor import *
2			from crab_util import *
3			import common
4			from ApmonIf import ApmonIf
5			import Statistic
6			import time
7			from ProgressBar import ProgressBar
8			from TerminalController import TerminalController
9
10			import xml.dom.minidom
11			import xml.dom.ext
12			import TaskDB
13
14			class StatusServer(Actor):
15
16			def __init__(self, cfg_params,):
17			self.cfg_params = cfg_params
18	mcinquil	1.2
19			self.countSubmit = 0
20			self.countSubmitting = 0
21			self.countDone = 0
22			self.countReady = 0
23			self.countSched = 0
24			self.countRun = 0
25			self.countAbort = 0
26			self.countCancel = 0
27			self.countCleared = 0
28			self.countToTjob = 0
29
30	spiga	1.3	try:
31			self.server_name = self.cfg_params['CRAB.server_name'] # gsiftp://pcpg01.cern.ch/data/SEDir/
32			except KeyError:
33			msg = 'No server selected ...'
34			msg = msg + 'Please specify a server in the crab cfg file'
35			raise CrabException(msg)
36
37	spiga	1.1	return
38	mcinquil	1.2
39			def translateStatus(self, status):
40			"""
41			simmetric as server
42			"""
43
44	mcinquil	1.4	stateConverting = {'Running': 'R', 'Aborted': 'A', 'Done': 'D', 'Done (Failed)': 'D',\
45	mcinquil	1.2	'Cleared': 'D', 'Cancelled': 'K', 'Killed': 'K'}
46
47			if status in stateConverting:
48			return stateConverting[status]
49			return None
50
51
52	spiga	1.1	def run(self):
53			"""
54			The main method of the class: check the status of the task
55			"""
56			common.logger.debug(5, "status server::run() called")
57			start = time.time()
58
59			totalCreatedJobs = 0
60			flagSubmit = 1
61			for nj in range(common.jobDB.nJobs()):
62			if (common.jobDB.status(nj)!='S'):
63			totalCreatedJobs +=1
64	mcinquil	1.2	# flagSubmit = 0
65	spiga	1.1
66			if not flagSubmit:
67			common.logger.message("Not all jobs are submitted: before checking the status submit all the jobs.")
68			return
69
70			common.scheduler.checkProxy()
71
72			common.taskDB.load()
73			WorkDirName =os.path.basename(os.path.split(common.work_space.topDir())[0])
74			projectUniqName = 'crab_'+str(WorkDirName)+'_'+common.taskDB.dict('TasKUUID')
75			try:
76			common.logger.message ("Checking the status...\n")
77	spiga	1.3	cmd = 'lcg-cp --vo cms gsiftp://' + str(self.server_name) + str(projectUniqName)+'/res/xmlReportFile.xml file://'+common.work_space.resDir()+'xmlReportFile.xml'
78	mcinquil	1.2	common.logger.debug(6, cmd)
79	spiga	1.1	os.system(cmd +' >& /dev/null')
80
81			except:
82			msg = ("task status not yet available")
83			raise CrabException(msg)
84
85			try:
86			file = open(common.work_space.resDir()+"xmlReportFile.xml", "r")
87			doc = xml.dom.minidom.parse(common.work_space.resDir()+ "xmlReportFile.xml" )
88
89			except:
90			msg = ("problems reading report file")
91			raise CrabException(msg)
92
93	mcinquil	1.2	### <Job status='Submitted' job_exit='NULL' id='1' exe_exit='NULL'/>
94
95	spiga	1.1	task = doc.childNodes[0].childNodes[1].getAttribute("taskName")
96	mcinquil	1.2	self.countToTjob = int(doc.childNodes[0].childNodes[1].getAttribute("totJob") )
97
98			addTree = 3
99
100			common.jobDB.load()
101
102			if doc.childNodes[0].childNodes[3].getAttribute("id") == "all":
103			if doc.childNodes[0].childNodes[3].getAttribute("status") == "Submitted":
104			self.countSubmitting = common.jobDB.nJobs()
105			for nj in range(common.jobDB.nJobs()):
106			common.jobDB.setStatus(nj, 'S')
107			else:
108			self.countAbort = common.jobDB.nJobs()
109			for nj in range(common.jobDB.nJobs()):
110			common.jobDB.setStatus(nj, 'A')
111			self.countToTjob = common.jobDB.nJobs()
112	spiga	1.1	else:
113	mcinquil	1.2	printline = ''
114			printline+= "%-10s %-20s %-20s %-25s" % ('JOBID','STATUS','EXE_EXIT_CODE','JOB_EXIT_STATUS')
115			print printline
116			print '-------------------------------------------------------------------------------------'
117
118			for job in range( self.countToTjob ):
119			idJob = doc.childNodes[0].childNodes[job+addTree].getAttribute("id")
120			stato = doc.childNodes[0].childNodes[job+addTree].getAttribute("status")
121			exe_exit_code = doc.childNodes[0].childNodes[job+addTree].getAttribute("job_exit")
122			job_exit_status = doc.childNodes[0].childNodes[job+addTree].getAttribute("exe_exit")
123			jobDbStatus = self.translateStatus(stato)
124	spiga	1.1
125	mcinquil	1.2	if jobDbStatus != None:
126			common.logger.debug(5, '* Updating jobdb for job %s *' %idJob)
127			if common.jobDB.status( str(int(idJob)-1) ) != "Y":
128			common.jobDB.setStatus( str(int(idJob)-1), self.translateStatus(stato) )
129			else:
130			stato = "Cleared"
131			common.jobDB.setExitStatus( str(int(idJob)-1), job_exit_status )
132	mcinquil	1.4	if stato != "Done" and stato != "Cleared" and stato != "Aborted" and stato != "Done (Failed)":
133	mcinquil	1.2	print "%-10s %-20s" % (idJob,stato)
134			else:
135			print "%-10s %-20s %-20s %-25s" % (idJob,stato,exe_exit_code,job_exit_status)
136
137			if stato == 'Running':
138			self.countRun += 1
139			elif stato == 'Aborted':
140			self.countAbort += 1
141			elif stato == 'Done':
142			self.countDone += 1
143			elif stato == 'Cancelled':
144			self.countCancel += 1
145			elif stato == 'Submitted':
146			self.countSubmit += 1
147			elif stato == 'Submitting':
148			self.countSubmitting += 1
149			elif stato == 'Ready':
150			self.countReady += 1
151			elif stato == 'Scheduled':
152			self.countSched += 1
153			elif stato == 'Cleared':
154			self.countCleared += 1
155
156			addTree += 1
157			common.jobDB.save()
158
159			self.PrintReport_()
160
161
162			def PrintReport_(self) :
163
164			""" Report #jobs for each status """
165
166
167			print ''
168			print ">>>>>>>>> %i Total Jobs " % (self.countToTjob)
169			print ''
170
171			if (self.countReady != 0):
172			print ">>>>>>>>> %i Jobs Ready" % (self.countReady)
173			if (self.countSubmitting != 0) :
174			print ">>>>>>>>> %i Jobs Submitting by the server" % (self.countSubmitting)
175			if (self.countSubmit != 0):
176			print ">>>>>>>>> %i Jobs Submitted" % (self.countSubmit)
177			if (self.countAbort != 0):
178			print ">>>>>>>>> %i Jobs Aborted" % (self.countAbort)
179			if (self.countSched != 0):
180			print ">>>>>>>>> %i Jobs Scheduled" % (self.countSched)
181			if (self.countRun != 0):
182			print ">>>>>>>>> %i Jobs Running" % (self.countRun)
183			if (self.countCleared != 0):
184			print ">>>>>>>>> %i Jobs Cleared" % (self.countRun)
185			if (self.countDone != 0):
186			print ">>>>>>>>> %i Jobs Done" % (self.countDone)
187			print " Retrieve them with: crab.py -getoutput -continue"
188			print ''
189			pass
190	spiga	1.1