ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/SchedulerGlite.py
Revision: 1.17.2.3
Committed: Mon Oct 22 16:58:29 2007 UTC (17 years, 6 months ago) by fanzago
Content type: text/x-python
Branch: CRAB_1_5_4_SLC3_start
CVS Tags: CRAB_1_5_4_SLC3, CRAB_1_5_4_SLC3_pre4
Branch point for: CRAB_1_5_4_SLC3_pre4_br
Changes since 1.17.2.2: +11 -6 lines
Log Message:
Using OSG_JOB_CONTACT instead of GLOBUS_GRAM_JOB_CONTACT

File Contents

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