ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/SchedulerGlite.py
(Generate patch)

Comparing COMP/CRAB/python/SchedulerGlite.py (file contents):
Revision 1.40 by spiga, Tue Mar 18 11:03:10 2008 UTC vs.
Revision 1.55 by afanfani, Tue May 20 11:28:49 2008 UTC

# Line 3 | Line 3 | from crab_logger import Logger
3   from crab_exceptions import *
4   from crab_util import *
5   from GliteConfig import *
6 + import EdgLoggingInfo
7   import common
8  
9   import os, sys, time
# Line 11 | Line 12 | class SchedulerGlite(SchedulerGrid):
12      def __init__(self, name="GLITE"):
13          SchedulerGrid.__init__(self,name)
14  
15 +        self.OSBsize = 55000000
16 +
17      def configure(self,cfg_params):
18          SchedulerGrid.configure(self, cfg_params)
19 +        self.checkProxy()
20          self.environment_unique_identifier = 'GLITE_WMS_JOBID'
21  
22 +    def realSchedParams(self,cfg_params):
23 +        """
24 +        Return dictionary with specific parameters, to use
25 +        with real scheduler  
26 +        """
27 +        self.rb_param_file=''
28 +        if (cfg_params.has_key('EDG.rb')):
29 +            self.rb_param_file=common.scheduler.rb_configure(cfg_params.get("EDG.rb"))
30 +        self.wms_service=cfg_params.get("EDG.wms_service",'')
31 +        self.skipWMSAuth=cfg_params.get("EDG.skipwmsauth",0)
32 +        params = { 'service' : self.wms_service, \
33 +                   'config' : self.rb_param_file, \
34 +                   'skipWMSAuth' : self.skipWMSAuth
35 +                 }
36 +        return  params
37 +      
38 +
39      def rb_configure(self, RB):
40          if not RB: return None
41          glite_config = None
# Line 29 | Line 50 | class SchedulerGlite(SchedulerGrid):
50  
51      def ce_list(self):
52          """
53 <        Returns string with requirement CE related    
53 >        Returns string with requirement CE related
54          """
55 <        req = ''  
55 >        req = ''
56          if self.EDG_ce_white_list:
57              ce_white_list = self.EDG_ce_white_list
58              tmpCe=[]
# Line 62 | Line 83 | class SchedulerGlite(SchedulerGrid):
83          # requirement added to skip gliteCE
84          req += '&& (!RegExp("blah", other.GlueCEUniqueId))'
85  
86 <        return req
86 >        return req,self.EDG_ce_white_list,self.EDG_ce_black_list
87  
88      def se_list(self, id, dest):
89          """
90 <        Returns string with requirement SE related    
91 <        """  
90 >        Returns string with requirement SE related
91 >        """
92          hostList=self.findSites_(id,dest)
93          req=''
94          reqtmp=[]
# Line 82 | Line 103 | class SchedulerGlite(SchedulerGrid):
103  
104      def jdlParam(self):
105          """
106 <        Returns
107 <        """
106 >        Returns
107 >        """
108          req=''
109          if self.EDG_addJdlParam:
110 <            if self.EDG_addJdlParam[-1] == '': self.EDG_addJdlParam= self.EDG_addJdlParam[:-1]
110 >            if self.EDG_addJdlParam[-1] == '': self.EDG_addJdlParam= self.EDG_addJdlParam[:-1]
111              for p in self.EDG_addJdlParam:
112               #   param_file.write(string.strip(p)+';\n')
113                  req+=string.strip(p)+';\n' ## BL--DS
# Line 95 | Line 116 | class SchedulerGlite(SchedulerGrid):
116      def specific_req(self):
117          """
118          Returns string with specific requirements
119 <        """
119 >        """
120          req=''
121          if self.EDG_clock_time:
122              if (not req == ' '): req = req + ' && '
# Line 104 | Line 125 | class SchedulerGlite(SchedulerGrid):
125          if self.EDG_cpu_time:
126              if (not req == ' '): req = req + ' && '
127              req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
128 <        
128 >
129          return req
130  
131      def sched_fix_parameter(self):
132          """
133          Returns string with requirements and scheduler-specific parameters
134          """
135 <        index = int(common._db.nJobs())
135 >        index = int(common._db.nJobs())
136          job = common.job_list[index-1]
137          jbt = job.type()
138          req = ''
# Line 122 | Line 143 | class SchedulerGlite(SchedulerGrid):
143              req = req + self.EDG_requirements
144  
145          Task_Req={'jobType':req}## DS--BL
146 <        common._db.updateTask_(Task_Req)        
146 >        common._db.updateTask_(Task_Req)
147  
148      def sched_parameter(self,i,task):
149          """
150          Returns string with requirements and scheduler-specific parameters
151          """
152 <        dest=  eval(task.jobs[i]['dlsDestination']) ## DS--BL
152 >        dest=  task.jobs[i-1]['dlsDestination'] ## DS--BL
153  
154          req=''
155          req +=task['jobType']
156  
157          sched_param=''
158          sched_param+='Requirements = ' + req +self.specific_req() + self.se_list(i,dest) +\
159 <                                        self.ce_list() +';\n' ## BL--DS
159 >                                        self.ce_list()[0] +';\n' ## BL--DS
160          if self.EDG_addJdlParam: sched_param+=self.jdlParam() ## BL--DS
161          sched_param+='MyProxyServer = "' + self.proxyServer + '";\n'
162          sched_param+='VirtualOrganisation = "' + self.VO + '";\n'
# Line 143 | Line 164 | class SchedulerGlite(SchedulerGrid):
164          sched_param+='ShallowRetryCount = '+str(self.EDG_shallow_retry_count)+';\n'
165  
166          return sched_param
146 –  
147 –    def wsSetupEnvironment(self):
148 –        """
149 –        Returns part of a job script which does scheduler-specific work.
150 –        """
151 –        txt = ''
152 –        txt += '# strip arguments\n'
153 –        txt += 'echo "strip arguments"\n'
154 –        txt += 'args=("$@")\n'
155 –        txt += 'nargs=$#\n'
156 –        txt += 'shift $nargs\n'
157 –        txt += "# job number (first parameter for job wrapper)\n"
158 –        txt += "NJob=${args[0]}; export NJob\n"
159 –
160 –        txt += '# job identification to DashBoard \n'
161 –        #txt += 'MonitorJobID=`echo ${NJob}_$GLITE_WMS_JOBID`\n'
162 –        #txt += 'SyncGridJobId=`echo $GLITE_WMS_JOBID`\n'
163 –        #txt += 'MonitorID=`echo ' + self._taskId + '`\n'
164 –        txt += 'MonitorJobID=${NJob}_$GLITE_WMS_JOBID \n'
165 –        txt += 'SyncGridJobId=$GLITE_WMS_JOBID \n'
166 –        txt += 'MonitorID='+self._taskId+' \n'
167 –        txt += 'echo "MonitorJobID=$MonitorJobID" > $RUNTIME_AREA/$repo \n'
168 –        txt += 'echo "SyncGridJobId=$SyncGridJobId" >> $RUNTIME_AREA/$repo \n'
169 –        txt += 'echo "MonitorID=$MonitorID" >> $RUNTIME_AREA/$repo\n'
170 –        #txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
171 –        #txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
172 –        #txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
173 –
174 –        #txt += 'echo "middleware discovery: " \n'
175 –        txt += 'echo ">>> GridFlavour discovery: " \n'
176 –        txt += 'if [ $VO_CMS_SW_DIR ]; then \n'
177 –        txt += '    middleware=LCG\n'
178 –        #txt += '    echo "SyncCE=`glite-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
179 –        txt += '    echo "SyncCE=`glite-brokerinfo getCE`" >> $RUNTIME_AREA/$repo \n'
180 –        #txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
181 –        txt += '    echo "GridFlavour=$middleware" | tee -a $RUNTIME_AREA/$repo \n'
182 –        #txt += '    echo ">>> middleware =$middleware" \n'
183 –        txt += 'elif [ $OSG_APP ]; then \n'
184 –        txt += '    middleware=OSG \n'
185 –        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
186 –        txt += '        SyncCE="$OSG_JOB_CONTACT"; \n'
187 –        #txt += '        echo "SyncCE=$SyncCE" | tee -a $RUNTIME_AREA/$repo ;\n'
188 –        txt += '        echo "SyncCE=$SyncCE" >> $RUNTIME_AREA/$repo ;\n'
189 –        txt += '    else\n'
190 –        txt += '        echo "not reporting SyncCE";\n'
191 –        txt += '    fi\n';
192 –        txt += '    echo "GridFlavour=$middleware" | tee -a $RUNTIME_AREA/$repo \n'
193 –        #txt += '    echo ">>> middleware =$middleware" \n'
194 –        txt += 'else \n'
195 –        txt += '    echo "ERROR ==> GridFlavour not identified" \n'
196 –        txt += '    job_exit_code=10030\n'
197 –        txt += '    func_exit \n'
198 –        #txt += '    echo "SET_CMS_ENV 10030 ==> middleware not identified" \n'
199 –        #txt += '    echo "JOB_EXIT_STATUS = 10030" \n'
200 –        #txt += '    echo "JobExitCode=10030" | tee -a $RUNTIME_AREA/$repo \n'
201 –        #txt += '    dumpStatus $RUNTIME_AREA/$repo \n'
202 –        #txt += '    exit 1 \n'
203 –        txt += 'fi \n'
204 –
205 –        txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
206 –
207 –        txt += '\n\n'
208 –
209 –        txt += 'export VO='+self.VO+'\n'
210 –        txt += 'if [ $middleware == LCG ]; then\n'
211 –        txt += '    CloseCEs=`glite-brokerinfo getCE`\n'
212 –        txt += '    echo "CloseCEs = $CloseCEs"\n'
213 –        txt += '    CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
214 –        txt += '    echo "CE = $CE"\n'
215 –        txt += 'elif [ $middleware == OSG ]; then \n'
216 –        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
217 –        txt += '        CE=`echo $OSG_JOB_CONTACT | /usr/bin/awk -F\/ \'{print $1}\'` \n'
218 –        txt += '    else \n'
219 –        txt += '        echo "ERROR ==> OSG mode in setting CE name from OSG_JOB_CONTACT" \n'
220 –        txt += '        job_exit_code=10099\n'
221 –        txt += '        func_exit \n'
222 –        #txt += '        echo "SET_CMS_ENV 10099 ==> OSG mode: ERROR in setting CE name from OSG_JOB_CONTACT" \n'
223 –        #txt += '        echo "JOB_EXIT_STATUS = 10099" \n'
224 –        #txt += '        echo "JobExitCode=10099" | tee -a $RUNTIME_AREA/$repo \n'
225 –        #txt += '        dumpStatus $RUNTIME_AREA/$repo \n'
226 –        #txt += '        exit 1 \n'
227 –        txt += '    fi \n'
228 –        txt += 'fi \n'
229 –
230 –        return txt
167  
168 <    def loggingInfo(self, id):
168 >    def decodeLogInfo(self, file):
169          """
170 <        retrieve the logging info from logging and bookkeeping and return it
170 >        Parse logging info file and return main info
171          """
172 <        self.checkProxy()
173 <        cmd = 'glite-job-logging-info -v 3 ' + id
174 <        cmd_out = runCommand(cmd)
239 <        return cmd_out
240 <
241 <    def queryDetailedStatus(self, id):
242 <        """ Query a detailed status of the job with id """
243 <        cmd = 'glite-job-status '+id
244 <        cmd_out = runCommand(cmd)
245 <        return cmd_out
172 >        loggingInfo = EdgLoggingInfo.EdgLoggingInfo()
173 >        reason = loggingInfo.decodeReason(file)
174 >        return reason
175  
176      def findSites_(self, n, sites):
177          itr4 =[]
# Line 253 | Line 182 | class SchedulerGlite(SchedulerGrid):
182              replicas = self.blackWhiteListParser.checkBlackList(sites,n)
183              if len(replicas)!=0:
184                  replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
185 <        
185 >
186              itr4 = replicas
187              #####
188          return itr4
189  
190 +    
191 +    def wsExitFunc(self):
192 +        """
193 +        """
194 +        txt = '\n'
195  
196 +        txt += '#\n'
197 +        txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
198 +        txt += '#\n\n'
199 +
200 +        txt += 'func_exit() { \n'
201 +        txt += self.wsExitFunc_common()
202 +        ### specific Glite check for OSB
203 +        txt += '    tar zcvf ${out_files}.tgz  ${final_list}\n'
204 +        txt += '    tmp_size=`ls -gGrta ${out_files}.tgz | awk \'{ print $3 }\'`\n'
205 +        txt += '    rm ${out_files}.tgz\n'  
206 +        txt += '    size=`expr $tmp_size`\n'
207 +        txt += '    echo "Total Output dimension: $size"\n'
208 +        txt += '    limit='+str(self.OSBsize) +' \n'  
209 +        txt += '    echo "WARNING: output files size limit is set to: $limit"\n'
210 +        txt += '    if [ "$limit" -lt "$size" ]; then\n'
211 +        txt += '        exceed=1\n'
212 +        txt += '        job_exit_code=70000\n'
213 +        txt += '        echo "Output Sanbox too big. Produced output is lost "\n'
214 +        txt += '    else\n'
215 +        txt += '        exceed=0\n'
216 +        txt += '        echo "Total Output dimension $size is fine."\n'
217 +        txt += '    fi\n'
218 +
219 +        txt += '    echo "JOB_EXIT_STATUS = $job_exit_code"\n'
220 +        txt += '    echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n'
221 +        txt += '    dumpStatus $RUNTIME_AREA/$repo\n'
222 +        txt += '    if [ $exceed -ne 1 ]; then\n'
223 +        txt += '        tar zcvf ${out_files}.tgz  ${final_list}\n'
224 +        txt += '    else\n'
225 +        txt += '        tar zcvf ${out_files}.tgz CMSSW_${NJob}.stdout CMSSW_${NJob}.stderr\n'
226 +        txt += '    fi\n'
227 +        txt += '    exit $job_exit_code\n'
228  
229 +        txt += '}\n'
230 +        return txt
231  
232 <    def tOut(self, list):
233 <        return 180
232 >    def userName(self):
233 >        """ return the user name """
234 >        tmp=runCommand("voms-proxy-info -identity")
235 >        return tmp.strip()

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines