ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/SchedulerGlite.py
Revision: 1.4
Committed: Wed Oct 11 15:45:53 2006 UTC (18 years, 6 months ago) by fanzago
Content type: text/x-python
Branch: MAIN
Changes since 1.3: +21 -4 lines
Log Message:
changes to switch glite broker

File Contents

# User Rev Content
1 spiga 1.2 #from Scheduler import Scheduler
2     from SchedulerEdg import SchedulerEdg
3     from crab_logger import Logger
4     from crab_exceptions import *
5     from crab_util import *
6 fanzago 1.4 #from EdgConfig import *
7     from GliteConfig import *
8 spiga 1.2 import common
9    
10     import os, sys, time
11    
12     class SchedulerGlite(SchedulerEdg):
13     def __init__(self):
14 slacapra 1.3 SchedulerEdg.__init__(self)
15    
16 fanzago 1.4 def rb_configure(self, RB):
17     self.glite_config = ''
18     self.rb_param_file = ''
19    
20     gliteConfig = GliteConfig(RB)
21     self.glite_config = gliteConfig.config()
22    
23     if (self.glite_config != ''):
24     self.rb_param_file = 'WMSconfig = '+self.glite_config+';'
25     #print "rb_param_file = ", self.rb_param_file
26     return self.rb_param_file
27    
28 spiga 1.2 def sched_parameter(self):
29     """
30     Returns file with requirements and scheduler-specific parameters
31     """
32     index = int(common.jobDB.nJobs()) - 1
33     job = common.job_list[index]
34     jbt = job.type()
35    
36     lastDest=''
37     first = []
38     last = []
39     for n in range(common.jobDB.nJobs()):
40     currDest=common.jobDB.destination(n)
41     if (currDest!=lastDest):
42     lastDest = currDest
43     first.append(n)
44     if n != 0:last.append(n-1)
45     if len(first)>len(last) :last.append(common.jobDB.nJobs())
46    
47     req = ''
48     req = req + jbt.getRequirements()
49    
50     if self.EDG_requirements:
51     if (req == ' '):
52     req = req + self.EDG_requirements
53     else:
54     req = req + ' && ' + self.EDG_requirements
55     if self.EDG_ce_white_list:
56     ce_white_list = string.split(self.EDG_ce_white_list,',')
57     for i in range(len(ce_white_list)):
58     if i == 0:
59     if (req == ' '):
60     req = req + '((RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
61     else:
62     req = req + ' && ((RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
63     pass
64     else:
65     req = req + ' || (RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
66     req = req + ')'
67    
68     if self.EDG_ce_black_list:
69     ce_black_list = string.split(self.EDG_ce_black_list,',')
70     for ce in ce_black_list:
71     if (req == ' '):
72     req = req + '(!RegExp("' + ce + '", other.GlueCEUniqueId))'
73     else:
74     req = req + ' && (!RegExp("' + ce + '", other.GlueCEUniqueId))'
75     pass
76     if self.EDG_clock_time:
77     if (req == ' '):
78     req = req + 'other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
79     else:
80     req = req + ' && other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
81    
82     if self.EDG_cpu_time:
83     if (req == ' '):
84     req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
85     else:
86     req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
87    
88     for i in range(len(first)): # Add loop DS
89     self.param='sched_param_'+str(i)+'.clad'
90     param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')
91    
92     itr4=self.findSites_(first[i])
93 slacapra 1.3 req1=[]
94     j = 0
95     concString = '||'
96     for arg in itr4:
97     #############
98     # MC Changed matching syntax to avoid gang matching
99     #############
100     req1.append(' Member("'+arg+'" , other.GlueCESEBindGroupSEUniqueID) ')
101     if len(req1): req = req + " && (" + concString.join(req1) + ')'
102     req = req + ';\n'
103     param_file.write('Requirements = ' + req )
104 spiga 1.2
105 fanzago 1.4 # if (self.edg_config and self.edg_config_vo != ''):
106     # param_file.write('RBconfig = "'+self.edg_config+'";\n')
107     # param_file.write('RBconfigVO = "'+self.edg_config_vo+'";')
108    
109     if (self.rb_param_file != ''):
110     param_file.write(self.rb_param_file)
111    
112 spiga 1.2
113     param_file.close()
114    
115     def wsSetupEnvironment(self):
116     """
117     Returns part of a job script which does scheduler-specific work.
118     """
119     txt = ''
120     txt += '# strip arguments\n'
121     txt += 'echo "strip arguments"\n'
122     txt += 'args=("$@")\n'
123     txt += 'nargs=$#\n'
124     txt += 'shift $nargs\n'
125     txt += "# job number (first parameter for job wrapper)\n"
126     txt += "NJob=${args[0]}\n"
127    
128     txt += '# job identification to DashBoard \n'
129     txt += 'MonitorJobID=`echo ${NJob}_$GLITE_WMS_JOBID`\n'
130     txt += 'SyncGridJobId=`echo $GLITE_WMS_JOBID`\n'
131     txt += 'MonitorID=`echo ' + self._taskId + '`\n'
132     txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
133     txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
134     txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
135    
136     txt += 'echo "middleware discovery " \n'
137     txt += 'if [ $VO_CMS_SW_DIR ]; then \n'
138     txt += ' middleware=LCG \n'
139     txt += ' echo "SyncCE=`glite-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
140     txt += ' echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
141     txt += ' echo "middleware =$middleware" \n'
142     txt += 'elif [ $GRID3_APP_DIR ]; then\n'
143     txt += ' middleware=OSG \n'
144     txt += ' echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
145     txt += ' echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
146     txt += ' echo "middleware =$middleware" \n'
147     txt += 'elif [ $OSG_APP ]; then \n'
148     txt += ' middleware=OSG \n'
149     txt += ' echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
150     txt += ' echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
151     txt += ' echo "middleware =$middleware" \n'
152     txt += 'else \n'
153     txt += ' echo "SET_CMS_ENV 10030 ==> middleware not identified" \n'
154     txt += ' echo "JOB_EXIT_STATUS = 10030" \n'
155     txt += ' echo "JobExitCode=10030" | tee -a $RUNTIME_AREA/$repo \n'
156     txt += ' dumpStatus $RUNTIME_AREA/$repo \n'
157     txt += ' rm -f $RUNTIME_AREA/$repo \n'
158     txt += ' echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
159     txt += ' echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
160     txt += ' exit 1 \n'
161     txt += 'fi \n'
162    
163     txt += '# report first time to DashBoard \n'
164     txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
165     txt += 'rm -f $RUNTIME_AREA/$repo \n'
166     txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
167     txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
168    
169     txt += '\n\n'
170    
171     if int(self.copy_data) == 1:
172     if self.SE:
173     txt += 'export SE='+self.SE+'\n'
174     txt += 'echo "SE = $SE"\n'
175     if self.SE_PATH:
176     if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
177     txt += 'export SE_PATH='+self.SE_PATH+'\n'
178     txt += 'echo "SE_PATH = $SE_PATH"\n'
179    
180     txt += 'export VO='+self.VO+'\n'
181     ### some line for LFC catalog setting
182     txt += 'if [ $middleware == LCG ]; then \n'
183     txt += ' if [[ $LCG_CATALOG_TYPE != \''+self.lcg_catalog_type+'\' ]]; then\n'
184     txt += ' export LCG_CATALOG_TYPE='+self.lcg_catalog_type+'\n'
185     txt += ' fi\n'
186     txt += ' if [[ $LFC_HOST != \''+self.lfc_host+'\' ]]; then\n'
187     txt += ' export LFC_HOST='+self.lfc_host+'\n'
188     txt += ' fi\n'
189     txt += ' if [[ $LFC_HOME != \''+self.lfc_home+'\' ]]; then\n'
190     txt += ' export LFC_HOME='+self.lfc_home+'\n'
191     txt += ' fi\n'
192     txt += 'elif [ $middleware == OSG ]; then\n'
193     txt += ' echo "LFC catalog setting to be implemented for OSG"\n'
194     txt += 'fi\n'
195     #####
196     if int(self.register_data) == 1:
197     txt += 'if [ $middleware == LCG ]; then \n'
198     txt += ' export LFN='+self.LFN+'\n'
199     txt += ' lfc-ls $LFN\n'
200     txt += ' result=$?\n'
201     txt += ' echo $result\n'
202     ### creation of LFN dir in LFC catalog, under /grid/cms dir
203     txt += ' if [ $result != 0 ]; then\n'
204     txt += ' lfc-mkdir $LFN\n'
205     txt += ' result=$?\n'
206     txt += ' echo $result\n'
207     txt += ' fi\n'
208     txt += 'elif [ $middleware == OSG ]; then\n'
209     txt += ' echo " Files registration to be implemented for OSG"\n'
210     txt += 'fi\n'
211     txt += '\n'
212    
213     if self.VO:
214     txt += 'export VO='+self.VO+'\n'
215     if self.LFN:
216     txt += 'if [ $middleware == LCG ]; then \n'
217     txt += ' export LFN='+self.LFN+'\n'
218     txt += 'fi\n'
219     txt += '\n'
220    
221     txt += 'if [ $middleware == LCG ]; then\n'
222     txt += ' CloseCEs=`glite-brokerinfo getCE`\n'
223     txt += ' echo "CloseCEs = $CloseCEs"\n'
224     txt += ' CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
225     txt += ' echo "CE = $CE"\n'
226     txt += 'elif [ $middleware == OSG ]; then \n'
227     txt += ' if [ $OSG_JOB_CONTACT ]; then \n'
228     txt += ' CE=`echo $OSG_JOB_CONTACT | /usr/bin/awk -F\/ \'{print $1}\'` \n'
229     txt += ' else \n'
230     txt += ' echo "SET_CMS_ENV 10099 ==> OSG mode: ERROR in setting CE name from OSG_JOB_CONTACT" \n'
231     txt += ' echo "JOB_EXIT_STATUS = 10099" \n'
232     txt += ' echo "JobExitCode=10099" | tee -a $RUNTIME_AREA/$repo \n'
233     txt += ' dumpStatus $RUNTIME_AREA/$repo \n'
234     txt += ' rm -f $RUNTIME_AREA/$repo \n'
235     txt += ' echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
236     txt += ' echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
237     txt += ' exit 1 \n'
238     txt += ' fi \n'
239     txt += 'fi \n'
240    
241     return txt
242    
243     def loggingInfo(self, id):
244     """
245     retrieve the logging info from logging and bookkeeping and return it
246     """
247     self.checkProxy()
248     cmd = 'glite-job-logging-info -v 2 ' + id
249     cmd_out = runCommand(cmd)
250     return cmd_out
251    
252     def queryDetailedStatus(self, id):
253     """ Query a detailed status of the job with id """
254     cmd = 'glite-job-status '+id
255     cmd_out = runCommand(cmd)
256     return cmd_out
257    
258     def findSites_(self, n):
259     sites = common.jobDB.destination(n)
260 slacapra 1.3 if len(sites)>0 and sites[0]=="":
261     return []
262     return sites