CRAB/python/Submitter.py

from Actor import *
from crab_util import *
import common
from ApmonIf import ApmonIf
import Statistic
from random import random
import time

class Submitter(Actor):
    def __init__(self, cfg_params, nj_list):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        
        try:
            self.ML = int(cfg_params['USER.activate_monalisa'])
        except KeyError:
            self.ML = 0
            pass

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

        totalCreatedJobs= 0
        listCE = ""
        start = time.time()
        for nj in range(common.jobDB.nJobs()):
            if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
            pass

        if (totalCreatedJobs==0):
            common.logger.message("No jobs to be submitted: first create them")
            return
        
        firstJob=self.nj_list[0]
        match = common.scheduler.listMatch(firstJob)
        if match:
            common.logger.message("Found "+str(match)+" compatible site(s)")
        else:
            raise CrabException("No compatible site found!")
        #########
        # Loop over jobs
        njs = 0
        try:
            for nj in self.nj_list:
                st = common.jobDB.status(nj)
#                print "nj = ", nj 
#                print "st = ", st
                if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
                    long_st = crabJobStatusToString(st)
                    msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
                    common.logger.message(msg)
                    continue

                common.logger.message("Submitting job # "+`(nj+1)`)
                jid = common.scheduler.submit(nj)

                common.jobDB.setStatus(nj, 'S')
                common.jobDB.setJobId(nj, jid)
                common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
                njs += 1

                ############################################   

                if st == 'C':
                    resFlag = 0
                elif st == 'RC':
                    resFlag = 2
                else:            
                    resFlag = 0
                    pass
                  
                try:
                    Statistic.Monitor('submit',resFlag,jid,'-----')
                except:
                    pass
                
                fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
                self.cfg_params['sid'] = jid
                nevtJob = common.jobDB.maxEvents(nj)

                # OLI: JobID treatment, special for Condor-G scheduler
                jobId = ''
                if common.scheduler.boss_scheduler_name == 'condor_g':
                    # create hash of cfg file
                    hash = makeCksum(common.work_space.cfgFileName())
                    jobId = str(nj + 1) + '_' + hash + '_' + self.cfg_params['sid']
                    common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
                else:
                    jobId = str(nj + 1) + '_' + self.cfg_params['sid']
                    common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)

                if ( jid.find(":") != -1 ) :
                    rb = jid.split(':')[1]
                    self.cfg_params['rb'] = rb.replace('//', '')
                else :
                    self.cfg_params['rb'] = 'OSG'

                params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
                          'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
                for i in fl.readlines():
                    val = i.split(':')
                    params[val[0]] = string.strip(val[1])
                self.cfg_params['apmon'].sendToML(params)
        except:
            exctype, value = sys.exc_info()[:2]
            print "Type: %s Value: %s"%(exctype, value)
            common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
            common.jobDB.save()
        
        stop = time.time()
        common.logger.debug(5, "Submission Time: "+str(stop - start))
        #print "Submission Time: %d "%(stop - start)
        common.jobDB.save()
            
        msg = '\nTotal of %d jobs submitted'%njs
        if njs != len(self.nj_list) :
            msg += ' (from %d requested).'%(len(self.nj_list))
            pass
        else:
            msg += '.'
            pass
        common.logger.message(msg)
        return
Revision:	1.46
Committed:	Wed May 24 21:04:52 2006 UTC (18 years, 11 months ago) by gutsche
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	pre_cmssw_integration_20060527
Changes since 1.45:	+14 -1 lines
Log Message:	added monitoring for OSG using EDG and Condor-G scheduler
#	User	Rev	Content
1	nsmirnov	1.1	from Actor import *
2	nsmirnov	1.6	from crab_util import *
3	nsmirnov	1.2	import common
4	corvo	1.9	from ApmonIf import ApmonIf
5	spiga	1.10	import Statistic
6	corvo	1.27	from random import random
7	corvo	1.30	import time
8	nsmirnov	1.1
9			class Submitter(Actor):
10	corvo	1.40	def __init__(self, cfg_params, nj_list):
11	nsmirnov	1.1	self.cfg_params = cfg_params
12	nsmirnov	1.4	self.nj_list = nj_list
13	corvo	1.35
14	slacapra	1.25	try:
15			self.ML = int(cfg_params['USER.activate_monalisa'])
16			except KeyError:
17			self.ML = 0
18			pass
19	corvo	1.30
20	nsmirnov	1.1	return
21
22			def run(self):
23	nsmirnov	1.2	"""
24			The main method of the class.
25			"""
26			common.logger.debug(5, "Submitter::run() called")
27	slacapra	1.24
28	gutsche	1.42	totalCreatedJobs= 0
29	corvo	1.37	listCE = ""
30	corvo	1.33	start = time.time()
31	slacapra	1.24	for nj in range(common.jobDB.nJobs()):
32	spiga	1.29	if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
33	slacapra	1.24	pass
34
35			if (totalCreatedJobs==0):
36			common.logger.message("No jobs to be submitted: first create them")
37	slacapra	1.28	return
38	slacapra	1.24
39	slacapra	1.7	firstJob=self.nj_list[0]
40			match = common.scheduler.listMatch(firstJob)
41	slacapra	1.8	if match:
42	slacapra	1.43	common.logger.message("Found "+str(match)+" compatible site(s)")
43	slacapra	1.8	else:
44	slacapra	1.28	raise CrabException("No compatible site found!")
45	fanzago	1.16	#########
46	nsmirnov	1.2	# Loop over jobs
47	nsmirnov	1.6	njs = 0
48	corvo	1.15	try:
49	corvo	1.17	for nj in self.nj_list:
50			st = common.jobDB.status(nj)
51	corvo	1.21	# print "nj = ", nj
52			# print "st = ", st
53	corvo	1.17	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
54			long_st = crabJobStatusToString(st)
55	slacapra	1.19	msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
56	slacapra	1.18	common.logger.message(msg)
57	corvo	1.17	continue
58
59			common.logger.message("Submitting job # "+`(nj+1)`)
60			jid = common.scheduler.submit(nj)
61
62			common.jobDB.setStatus(nj, 'S')
63			common.jobDB.setJobId(nj, jid)
64	corvo	1.33	common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
65	corvo	1.17	njs += 1
66
67			############################################
68
69			if st == 'C':
70			resFlag = 0
71			elif st == 'RC':
72			resFlag = 2
73			else:
74			resFlag = 0
75			pass
76	spiga	1.26
77	spiga	1.38	try:
78			Statistic.Monitor('submit',resFlag,jid,'-----')
79			except:
80			pass
81	corvo	1.17
82	corvo	1.44	fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
83			self.cfg_params['sid'] = jid
84			nevtJob = common.jobDB.maxEvents(nj)
85	gutsche	1.46
86			# OLI: JobID treatment, special for Condor-G scheduler
87			jobId = ''
88			if common.scheduler.boss_scheduler_name == 'condor_g':
89			# create hash of cfg file
90			hash = makeCksum(common.work_space.cfgFileName())
91			jobId = str(nj + 1) + '_' + hash + '_' + self.cfg_params['sid']
92			common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
93			else:
94			jobId = str(nj + 1) + '_' + self.cfg_params['sid']
95			common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
96
97	corvo	1.44	if ( jid.find(":") != -1 ) :
98			rb = jid.split(':')[1]
99			self.cfg_params['rb'] = rb.replace('//', '')
100			else :
101			self.cfg_params['rb'] = 'OSG'
102	gutsche	1.46
103			params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
104	corvo	1.44	'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
105			for i in fl.readlines():
106			val = i.split(':')
107			params[val[0]] = string.strip(val[1])
108	corvo	1.45	self.cfg_params['apmon'].sendToML(params)
109	corvo	1.17	except:
110	corvo	1.27	exctype, value = sys.exc_info()[:2]
111	corvo	1.30	print "Type: %s Value: %s"%(exctype, value)
112	corvo	1.27	common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
113	corvo	1.17	common.jobDB.save()
114	corvo	1.30
115			stop = time.time()
116	fanzago	1.32	common.logger.debug(5, "Submission Time: "+str(stop - start))
117			#print "Submission Time: %d "%(stop - start)
118	corvo	1.17	common.jobDB.save()
119	corvo	1.15
120	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
121			if njs != len(self.nj_list) :
122			msg += ' (from %d requested).'%(len(self.nj_list))
123			pass
124			else:
125			msg += '.'
126			pass
127	nsmirnov	1.2	common.logger.message(msg)
128	nsmirnov	1.1	return