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

Comparing COMP/CRAB/python/SchedulerEdg.py (file contents):
Revision 1.73.2.5.2.7 by spiga, Wed Sep 13 17:28:40 2006 UTC vs.
Revision 1.106.2.1 by slacapra, Wed Dec 6 10:33:05 2006 UTC

# Line 18 | Line 18 | class SchedulerEdg(Scheduler):
18                        "owner","parent_job", "reason","resubmitted","rsl","seed",\
19                        "stateEnterTime","stateEnterTimes","subjob_failed", \
20                        "user tags" , "status" , "status_code","hierarchy"]
21 +        
22          return
23  
24      def configure(self, cfg_params):
25  
26          try:
27 <            RB = cfg_params["EDG.rb"]
28 <            edgConfig = EdgConfig(RB)
28 <            self.edg_config = edgConfig.config()
29 <            self.edg_config_vo = edgConfig.configVO()
27 >            RB=cfg_params["EDG.rb"]
28 >            self.rb_param_file=self.rb_configure(RB)
29          except KeyError:
30 <            self.edg_config = ''
31 <            self.edg_config_vo = ''
33 <
30 >            self.rb_param_file=''
31 >            pass
32          try:
33              self.proxyServer = cfg_params["EDG.proxy_server"]
34          except KeyError:
35              self.proxyServer = 'myproxy.cern.ch'
36          common.logger.debug(5,'Setting myproxy server to '+self.proxyServer)
37  
38 +        try:
39 +            self.group = cfg_params["EDG.group"]
40 +        except KeyError:
41 +            self.group = None
42 +            
43 +        try:
44 +            self.role = cfg_params["EDG.role"]
45 +        except KeyError:
46 +            self.role = None
47 +            
48          try: self.LCG_version = cfg_params["EDG.lcg_version"]
49          except KeyError: self.LCG_version = '2'
50  
43 –        try: self.EDG_requirements = cfg_params['EDG.requirements']
44 –        except KeyError: self.EDG_requirements = ''
45 –
46 –        try: self.EDG_retry_count = cfg_params['EDG.retry_count']
47 –        except KeyError: self.EDG_retry_count = ''
48 –
51          try:
52              self.EDG_ce_black_list = cfg_params['EDG.ce_black_list']
53          except KeyError:
# Line 58 | Line 60 | class SchedulerEdg(Scheduler):
60          try: self.VO = cfg_params['EDG.virtual_organization']
61          except KeyError: self.VO = 'cms'
62  
63 <        try: self.return_data = cfg_params['USER.return_data']
62 <        except KeyError: self.return_data = 1
63 <
64 <        try:
65 <             self.copy_input_data = common.analisys_common_info['copy_input_data']
63 >        try: self.copy_input_data = cfg_params["USER.copy_input_data"]
64          except KeyError: self.copy_input_data = 0
65  
66 +        try: self.return_data = cfg_params['USER.return_data']
67 +        except KeyError: self.return_data = 0
68 +
69          try:
70              self.copy_data = cfg_params["USER.copy_data"]
71              if int(self.copy_data) == 1:
# Line 125 | Line 126 | class SchedulerEdg(Scheduler):
126  
127          try: self.EDG_requirements = cfg_params['EDG.requirements']
128          except KeyError: self.EDG_requirements = ''
129 <                                                                                                                                                            
129 >
130 >        try: self.EDG_addJdlParam = string.split(cfg_params['EDG.additional_jdl_parameters'],',')
131 >        except KeyError: self.EDG_addJdlParam = []
132 >
133          try: self.EDG_retry_count = cfg_params['EDG.retry_count']
134          except KeyError: self.EDG_retry_count = ''
135 <                                                                                                                                                            
135 >
136 >        try: self.EDG_shallow_retry_count= cfg_params['EDG.shallow_retry_count']
137 >        except KeyError: self.EDG_shallow_retry_count = ''
138 >
139          try: self.EDG_clock_time = cfg_params['EDG.max_wall_clock_time']
140          except KeyError: self.EDG_clock_time= ''
141 <                                                                                                                                                            
141 >
142          try: self.EDG_cpu_time = cfg_params['EDG.max_cpu_time']
143          except KeyError: self.EDG_cpu_time = ''
144  
145          # Add EDG_WL_LOCATION to the python path
146 +
147          try:
148              path = os.environ['EDG_WL_LOCATION']
149          except:
# Line 160 | Line 168 | class SchedulerEdg(Scheduler):
168          try: self.schedulerName = cfg_params['CRAB.scheduler']
169          except KeyError: self.scheduler = ''
170  
171 +        try: self.dontCheckProxy=cfg_params["EDG.dont_check_proxy"]
172 +        except KeyError: self.dontCheckProxy = 0
173 +
174          return
175      
176  
177 +    def rb_configure(self, RB):
178 +        self.edg_config = ''
179 +        self.edg_config_vo = ''
180 +        self.rb_param_file = ''
181 +
182 +        edgConfig = EdgConfig(RB)
183 +        self.edg_config = edgConfig.config()
184 +        self.edg_config_vo = edgConfig.configVO()
185 +
186 +        if (self.edg_config and self.edg_config_vo != ''):
187 +            self.rb_param_file = 'RBconfig = "'+self.edg_config+'";\nRBconfigVO = "'+self.edg_config_vo+'";'
188 +            #print "rb_param_file = ", self.rb_param_file
189 +        return self.rb_param_file
190 +      
191 +
192      def sched_parameter(self):
193          """
194 <        Returns file with scheduler-specific parameters
194 >        Returns file with requirements and scheduler-specific parameters
195          """
196 <        if (self.edg_config and self.edg_config_vo != ''):
197 <            self.param='sched_param.clad'
196 >        index = int(common.jobDB.nJobs()) - 1
197 >        job = common.job_list[index]
198 >        jbt = job.type()
199 >        
200 >        lastBlock=-1
201 >        first = []
202 >        for n in range(common.jobDB.nJobs()):
203 >            currBlock=common.jobDB.block(n)
204 >            if (currBlock!=lastBlock):
205 >                lastBlock = currBlock
206 >                first.append(n)
207 >  
208 >        req = ''
209 >        req = req + jbt.getRequirements()
210 >    
211 >        if self.EDG_requirements:
212 >            if (req == ' '):
213 >                req = req + self.EDG_requirements
214 >            else:
215 >                req = req +  ' && ' + self.EDG_requirements
216 >
217 >        if self.EDG_ce_white_list:
218 >            ce_white_list = string.split(self.EDG_ce_white_list,',')
219 >            for i in range(len(ce_white_list)):
220 >                if i == 0:
221 >                    if (req == ' '):
222 >                        req = req + '((RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
223 >                    else:
224 >                        req = req +  ' && ((RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
225 >                    pass
226 >                else:
227 >                    req = req +  ' || (RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
228 >            req = req + ')'
229 >        
230 >        if self.EDG_ce_black_list:
231 >            ce_black_list = string.split(self.EDG_ce_black_list,',')
232 >            for ce in ce_black_list:
233 >                if (req == ' '):
234 >                    req = req + '(!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
235 >                else:
236 >                    req = req +  ' && (!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
237 >                pass
238 >        if self.EDG_clock_time:
239 >            if (req == ' '):
240 >                req = req + 'other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
241 >            else:
242 >                req = req + ' && other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
243 >
244 >        if self.EDG_cpu_time:
245 >            if (req == ' '):
246 >                req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
247 >            else:
248 >                req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
249 >                
250 >        for i in range(len(first)): # Add loop DS
251 >            groupReq = req
252 >            self.param='sched_param_'+str(i)+'.clad'
253              param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')
254 <            param_file.write('RBconfig = "'+self.edg_config+'";\n')  
255 <            param_file.write('RBconfigVO = "'+self.edg_config_vo+'";')
254 >
255 >            itr4=self.findSites_(first[i])
256 >            for arg in itr4:
257 >                groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
258 >            param_file.write('Requirements = '+groupReq +';\n')  
259 >  
260 >            if (self.rb_param_file != ''):
261 >                param_file.write(self.rb_param_file)  
262 >
263 >            if len(self.EDG_addJdlParam):
264 >                for p in self.EDG_addJdlParam:
265 >                    param_file.write(p)
266 >
267              param_file.close()  
268 <            return 1
177 <        else:
178 <            return 0
268 >
269  
270      def wsSetupEnvironment(self):
271          """
272          Returns part of a job script which does scheduler-specific work.
273          """
274          txt = ''
275 +        txt += '# strip arguments\n'
276 +        txt += 'echo "strip arguments"\n'
277 +        txt += 'args=("$@")\n'
278 +        txt += 'nargs=$#\n'
279 +        txt += 'shift $nargs\n'
280          txt += "# job number (first parameter for job wrapper)\n"
281 <        txt += "NJob=$1\n"
281 >        #txt += "NJob=$1\n"
282 >        txt += "NJob=${args[0]}\n"
283  
284          txt += '# job identification to DashBoard \n'
285          txt += 'MonitorJobID=`echo ${NJob}_$EDG_WL_JOBID`\n'
# Line 194 | Line 290 | class SchedulerEdg(Scheduler):
290          txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
291  
292          txt += 'echo "middleware discovery " \n'
293 <        txt += 'if [ $VO_CMS_SW_DIR ]; then \n'
198 <        txt += '    middleware=LCG \n'
199 <        txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
200 <        txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
201 <        txt += '    echo "middleware =$middleware" \n'
202 <        txt += 'elif [ $GRID3_APP_DIR ]; then\n'
293 >        txt += 'if [ $GRID3_APP_DIR ]; then\n'
294          txt += '    middleware=OSG \n'
295          txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
296          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
# Line 209 | Line 300 | class SchedulerEdg(Scheduler):
300          txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
301          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
302          txt += '    echo "middleware =$middleware" \n'
303 +        txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
304 +        txt += '    middleware=LCG \n'
305 +        txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
306 +        txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
307 +        txt += '    echo "middleware =$middleware" \n'
308          txt += 'else \n'
309          txt += '    echo "SET_CMS_ENV 10030 ==> middleware not identified" \n'
310          txt += '    echo "JOB_EXIT_STATUS = 10030" \n'
# Line 238 | Line 334 | class SchedulerEdg(Scheduler):
334                txt += 'echo "SE_PATH = $SE_PATH"\n'
335  
336          txt += 'export VO='+self.VO+'\n'
337 <        ### FEDE: add some line for LFC catalog setting
337 >        ### add some line for LFC catalog setting
338          txt += 'if [ $middleware == LCG ]; then \n'
339          txt += '    if [[ $LCG_CATALOG_TYPE != \''+self.lcg_catalog_type+'\' ]]; then\n'
340          txt += '        export LCG_CATALOG_TYPE='+self.lcg_catalog_type+'\n'
# Line 305 | Line 401 | class SchedulerEdg(Scheduler):
401          Copy input data from SE to WN    
402          """
403          txt = ''
404 <        try:
405 <            self.copy_input_data = common.analisys_common_info['copy_input_data']
310 <        except KeyError: self.copy_input_data = 0
311 <        if int(self.copy_input_data) == 1:
404 >        if not self.copy_input_data: return txt
405 >
406          ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
407 <           txt += 'if [ $middleware == OSG ]; then\n'
408 <           txt += '   #\n'
409 <           txt += '   #   Copy Input Data from SE to this WN deactivated in OSG mode\n'
410 <           txt += '   #\n'
411 <           txt += '   echo "Copy Input Data from SE to this WN deactivated in OSG mode"\n'
412 <           txt += 'elif [ $middleware == LCG ]; then \n'
413 <           txt += '   #\n'
414 <           txt += '   #   Copy Input Data from SE to this WN\n'
415 <           txt += '   #\n'
416 <           ### changed by georgia (put a loop copying more than one input files per jobs)          
417 <           txt += '   for input_file in $cur_file_list \n'
418 <           txt += '   do \n'
419 <           txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$input_lfn/$input_file file:`pwd`/$input_file 2>&1\n'
420 <           txt += '      copy_input_exit_status=$?\n'
421 <           txt += '      echo "COPY_INPUT_EXIT_STATUS = $copy_input_exit_status"\n'
422 <           txt += '      if [ $copy_input_exit_status -ne 0 ]; then \n'
423 <           txt += '         echo "Problems with copying to WN" \n'
424 <           txt += '      else \n'
425 <           txt += '         echo "input copied into WN" \n'
426 <           txt += '      fi \n'
427 <           txt += '   done \n'
428 <           ### copy a set of PU ntuples (same for each jobs -- but accessed randomly)
429 <           txt += '   for file in $cur_pu_list \n'
430 <           txt += '   do \n'
431 <           txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$pu_lfn/$file file:`pwd`/$file 2>&1\n'
432 <           txt += '      copy_input_pu_exit_status=$?\n'
433 <           txt += '      echo "COPY_INPUT_PU_EXIT_STATUS = $copy_input_pu_exit_status"\n'
434 <           txt += '      if [ $copy_input_pu_exit_status -ne 0 ]; then \n'
435 <           txt += '         echo "Problems with copying pu to WN" \n'
436 <           txt += '      else \n'
437 <           txt += '         echo "input pu files copied into WN" \n'
438 <           txt += '      fi \n'
439 <           txt += '   done \n'
440 <           txt += '   \n'
441 <           txt += '   ### Check SCRATCH space available on WN : \n'
442 <           txt += '   df -h \n'
443 <           txt += 'fi \n'
407 >        txt += 'if [ $middleware == OSG ]; then\n'
408 >        txt += '   #\n'
409 >        txt += '   #   Copy Input Data from SE to this WN deactivated in OSG mode\n'
410 >        txt += '   #\n'
411 >        txt += '   echo "Copy Input Data from SE to this WN deactivated in OSG mode"\n'
412 >        txt += 'elif [ $middleware == LCG ]; then \n'
413 >        txt += '   #\n'
414 >        txt += '   #   Copy Input Data from SE to this WN\n'
415 >        txt += '   #\n'
416 >        ### changed by georgia (put a loop copying more than one input files per jobs)          
417 >        txt += '   for input_file in $cur_file_list \n'
418 >        txt += '   do \n'
419 >        txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$input_lfn/$input_file file:`pwd`/$input_file 2>&1\n'
420 >        txt += '      copy_input_exit_status=$?\n'
421 >        txt += '      echo "COPY_INPUT_EXIT_STATUS = $copy_input_exit_status"\n'
422 >        txt += '      if [ $copy_input_exit_status -ne 0 ]; then \n'
423 >        txt += '         echo "Problems with copying to WN" \n'
424 >        txt += '      else \n'
425 >        txt += '         echo "input copied into WN" \n'
426 >        txt += '      fi \n'
427 >        txt += '   done \n'
428 >        ### copy a set of PU ntuples (same for each jobs -- but accessed randomly)
429 >        txt += '   for file in $cur_pu_list \n'
430 >        txt += '   do \n'
431 >        txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$pu_lfn/$file file:`pwd`/$file 2>&1\n'
432 >        txt += '      copy_input_pu_exit_status=$?\n'
433 >        txt += '      echo "COPY_INPUT_PU_EXIT_STATUS = $copy_input_pu_exit_status"\n'
434 >        txt += '      if [ $copy_input_pu_exit_status -ne 0 ]; then \n'
435 >        txt += '         echo "Problems with copying pu to WN" \n'
436 >        txt += '      else \n'
437 >        txt += '         echo "input pu files copied into WN" \n'
438 >        txt += '      fi \n'
439 >        txt += '   done \n'
440 >        txt += '   \n'
441 >        txt += '   ### Check SCRATCH space available on WN : \n'
442 >        txt += '   df -h \n'
443 >        txt += 'fi \n'
444            
445          return txt
446  
# Line 367 | Line 461 | class SchedulerEdg(Scheduler):
461             txt += '        export X509_CERT_DIR=$OSG_APP/glite/etc/grid-security/certificates\n'
462             txt += '        echo "export X509_CERT_DIR=$X509_CERT_DIR"\n'
463             txt += '    fi \n'
464 +
465             txt += '    for out_file in $file_list ; do\n'
466 <           txt += '        echo "Trying to copy output file to $SE using lcg-cp"\n'
467 <           txt += '        echo "lcg-cp --vo $VO -t 1200 --verbose file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
468 <           txt += '        exitstring=`lcg-cp --vo $VO -t 1200 --verbose file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
466 >           txt += '        echo "Trying to copy output file to $SE using srmcp"\n'
467 >           txt += '        echo "mkdir -p $HOME/.srmconfig"\n'
468 >           txt += '        mkdir -p $HOME/.srmconfig\n'
469 >           txt += '        if [ $middleware == LCG ]; then\n'
470 >           txt += '           echo "srmcp -retry_num 3 -retry_timeout 480000 file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
471 >           txt += '           exitstring=`srmcp -retry_num 3 -retry_timeout 480000 file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
472 >           txt += '        elif [ $middleware == OSG ]; then\n'
473 >           txt += '           echo "srmcp -retry_num 3 -retry_timeout 240000 -x509_user_trusted_certificates $X509_CERT_DIR file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
474 >           txt += '           exitstring=`srmcp -retry_num 3 -retry_timeout 240000 -x509_user_trusted_certificates $X509_CERT_DIR file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
475 >           txt += '        fi \n'
476             txt += '        copy_exit_status=$?\n'
477 <           txt += '        echo "COPY_EXIT_STATUS for lcg-cp = $copy_exit_status"\n'
477 >           txt += '        echo "COPY_EXIT_STATUS for srmcp = $copy_exit_status"\n'
478             txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
479 +
480             txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
481             txt += '            echo "Possible problem with SE = $SE"\n'
482             txt += '            echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
483             txt += '            echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
484 <           txt += '            echo "lcg-cp failed, attempting srmcp"\n'
485 <           txt += '            echo "mkdir -p $HOME/.srmconfig"\n'
486 <           txt += '            mkdir -p $HOME/.srmconfig\n'
487 <           txt += '            if [ $middleware == LCG ]; then\n'
488 <           txt += '               echo "srmcp -retry_num 5 -retry_timeout 240000 file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
489 <           txt += '               exitstring=`srmcp -retry_num 5 -retry_timeout 240000 file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
490 <           txt += '            elif [ $middleware == OSG ]; then\n'
388 <           txt += '               echo "srmcp -retry_num 5 -retry_timeout 240000 -x509_user_trusted_certificates $OSG_APP/glite/etc/grid-security/certificates file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
389 <           txt += '               exitstring=`srmcp -retry_num 5 -retry_timeout 240000 -x509_user_trusted_certificates $OSG_APP/glite/etc/grid-security/certificates file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
390 <           txt += '            fi \n'
484 >           txt += '            echo "srmcp failed, attempting lcg-cp."\n'
485 >           if common.logger.debugLevel() >= 5:
486 >               txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
487 >               txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
488 >           else:              
489 >               txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
490 >               txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
491             txt += '            copy_exit_status=$?\n'
492 <           txt += '            echo "COPY_EXIT_STATUS for srm = $copy_exit_status"\n'
492 >           txt += '            echo "COPY_EXIT_STATUS for lcg-cp = $copy_exit_status"\n'
493             txt += '            echo "STAGE_OUT = $copy_exit_status"\n'
494 +
495             txt += '            if [ $copy_exit_status -ne 0 ]; then\n'
496             txt += '               echo "Problems with SE = $SE"\n'
497             txt += '               echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
498             txt += '               echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
499 <           txt += '               echo "lcg-cp and srm failed"\n'
399 <           txt += '               echo "If storage_path in your config file contains a ? you may need a \? instead."\n'
499 >           txt += '               echo "srmcp and lcg-cp and failed!"\n'
500             txt += '            else\n'
501             txt += '               echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
502             txt += '               echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
503             txt += '               echo "output copied into $SE/$SE_PATH directory"\n'
504             txt += '               echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
505 <           txt += '               echo "srmcp succeeded"\n'
505 >           txt += '               echo "lcg-cp succeeded"\n'
506             txt += '            fi\n'
507             txt += '        else\n'
508             txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
509             txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
510             txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
511             txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
512 <           txt += '            echo "lcg-cp succeeded"\n'
512 >           txt += '            echo "srmcp succeeded"\n'
513             txt += '         fi\n'
514             txt += '     done\n'
515          return txt
# Line 421 | Line 521 | class SchedulerEdg(Scheduler):
521  
522          txt = ''
523          if int(self.register_data) == 1:
524 <           ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
524 >        ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
525             txt += 'if [ $middleware == OSG ]; then\n'
526             txt += '   #\n'
527             txt += '   #   Register output to LFC deactivated in OSG mode\n'
# Line 484 | Line 584 | class SchedulerEdg(Scheduler):
584          """
585          self.checkProxy()
586          cmd = 'edg-job-get-logging-info -v 2 ' + id
487 –        #cmd_out = os.popen(cmd)
587          cmd_out = runCommand(cmd)
588          return cmd_out
589  
491 –  #  def listMatch(self, nj):
492 –  #      """
493 –  #      Check the compatibility of available resources
494 –  #      """
495 –  #      self.checkProxy()
496 –  #      jdl = common.work_space.shareDir()+"fake.jdl"
497 –  #      cmd = 'edg-job-list-match ' + self.configOpt_() + str(jdl)
498 –  #      cmd_out = runCommand(cmd,0,10)
499 –  #      if not cmd_out:
500 –  #          raise CrabException("ERROR: "+cmd+" failed!")
501 – #
502 – #       return self.parseListMatch_(cmd_out, jdl)
503 –
504 –  #  def parseListMatch_(self, out, jdl):
505 –  #      """
506 –  #      Parse the f* output of edg-list-match and produce something sensible
507 –  #      """
508 –  #      reComment = re.compile( r'^\**$' )
509 –  #      reEmptyLine = re.compile( r'^$' )
510 –  #      reVO = re.compile( r'Selected Virtual Organisation name.*' )
511 –  #      reLine = re.compile( r'.*')
512 –  #      reCE = re.compile( r'(.*:.*)')
513 –  #      reCEId = re.compile( r'CEId.*')
514 –  #      reNO = re.compile( r'No Computing Element matching' )
515 –  #      reRB = re.compile( r'Connecting to host' )
516 –  #      next = 0
517 –  #      CEs=[]
518 –  #      Match=0
519 – #
520 – #        #print out
521 – #        lines = reLine.findall(out)
522 –
523 – #       i=0
524 – #       CEs=[]
525 – #       for line in lines:
526 – #           string.strip(line)
527 – #           #print line
528 – #           if reNO.match( line ):
529 – #               common.logger.debug(5,line)
530 – #               return 0
531 – #               pass
532 – #           if reVO.match( line ):
533 – #               VO =reVO.match( line ).group()
534 – #               common.logger.debug(5,"VO "+VO)
535 – #               pass
536 –
537 – #           if reRB.match( line ):
538 – #               RB = reRB.match(line).group()
539 – #               common.logger.debug(5,"RB "+RB)
540 – #               pass
541 –
542 – #           if reCEId.search( line ):
543 – #               for lineCE in lines[i:-1]:
544 – #                   if reCE.match( lineCE ):
545 – #                       CE = string.strip(reCE.search(lineCE).group(1))
546 – #                       CEs.append(CE.split(':')[0])
547 – #                       pass
548 – #                   pass
549 – #               pass
550 – #           i=i+1
551 – #           pass
552 –
553 – #       common.logger.debug(5,"All CE :"+str(CEs))
554 –
555 – #       sites = []
556 – #       [sites.append(it) for it in CEs if not sites.count(it)]
557 –
558 – #       common.logger.debug(5,"All Sites :"+str(sites))
559 – #       common.logger.message("Matched Sites :"+str(sites))
560 – #       return len(sites)
561 –
562 –
563 –    ################################################################ To remove when Boss4 store this info  DS. (start)
564 –    def getExitStatus(self, id):
565 –        return self.getStatusAttribute_(id, 'exit_code')
566 –
567 –    def queryStatus(self, id):
568 –        return self.getStatusAttribute_(id, 'status')
569 –
570 –    def queryDest(self, id):  
571 –        return self.getStatusAttribute_(id, 'destination')
572 –
573 –
574 –    def getStatusAttribute_(self, id, attr):
575 –        """ Query a status of the job with id """
576 –
577 –        self.checkProxy()
578 –        hstates = {}
579 –        Status = importName('edg_wl_userinterface_common_LbWrapper', 'Status')
580 –        # Bypass edg-job-status interfacing directly to C++ API
581 –        # Job attribute vector to retrieve status without edg-job-status
582 –        level = 0
583 –        # Instance of the Status class provided by LB API
584 –        jobStat = Status()
585 –        st = 0
586 –        jobStat.getStatus(id, level)
587 –        err, apiMsg = jobStat.get_error()
588 –        if err:
589 –            common.logger.debug(5,'Error caught' + apiMsg)
590 –            return None
591 –        else:
592 –           for i in range(len(self.states)):
593 –                # Fill an hash table with all information retrieved from LB API
594 –                hstates[ self.states[i] ] = jobStat.loadStatus(st)[i]
595 –           result = jobStat.loadStatus(st)[self.states.index(attr)]
596 –           return result
597 –
598 –
590      def queryDetailedStatus(self, id):
591          """ Query a detailed status of the job with id """
592          cmd = 'edg-job-status '+id
593          cmd_out = runCommand(cmd)
594          return cmd_out
595  
605 –
606 –  #  def createFakeJdl(self,nj):  # TMP Just waiting listmatch functionalitly  
607 –  #                                # implementation into BOSS4   Daniele
608 –  #      """                          
609 –  #      Create a fake jdl considering
610 –  #      only requirements  
611 –  #      """
612 –  #      job = common.job_list[0]
613 –  #      jbt = job.type()
614 –  #      inp_storage_subdir = ''
615 –  #      
616 –  #      
617 –  #      SPL = inp_storage_subdir
618 –  #      if ( SPL and SPL[-1] != '/' ) : SPL = SPL + '/'
619 – #
620 – #        jdl = open(common.work_space.shareDir()+"fake.jdl","w")
621 – #
622 – #        script = job.scriptFilename()
623 – #        jdl.write('Executable = "' + os.path.basename(script) +'";\n')
624 – #
625 – #        req='Requirements = '
626 – #        noreq=req
627 – #        ##### 8_9_06 FEDE
628 – #        req = req + jbt.getRequirements(nj)
629 – #        #req = req + jbt.getRequirements()
630 – #        ########
631 – #        #### and USER REQUIREMENT
632 – #       if self.EDG_requirements:
633 – #           if (req != noreq):
634 – #               req = req +  ' && '
635 – #           req = req + self.EDG_requirements
636 – #       #### FEDE #####
637 – #       if self.EDG_ce_white_list:
638 – #           ce_white_list = string.split(self.EDG_ce_white_list,',')
639 – #           #print "req = ", req
640 – #           for i in range(len(ce_white_list)):
641 – #               if i == 0:
642 – #                   if (req != noreq):
643 – #                       req = req +  ' && '
644 – #                   req = req + '((RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
645 –  #                  pass
646 –  #              else:
647 –  #                  req = req +  ' || (RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
648 –  #          req = req + ')'
649 –  #      
650 –  #      if self.EDG_ce_black_list:
651 –  #          ce_black_list = string.split(self.EDG_ce_black_list,',')
652 –  #          for ce in ce_black_list:
653 –  #              if (req != noreq):
654 –  #                  req = req +  ' && '
655 –  #              req = req + '(!RegExp("' + ce + '", other.GlueCEUniqueId))'
656 –  #              pass
657 –  #
658 –  #      ###############
659 –  #      clockTime=480
660 –  #      if self.EDG_clock_time:
661 –  #          clockTime= self.EDG_clock_time
662 –  #      if (req != noreq):
663 –  #          req = req + ' && '
664 –  #      req = req + '((other.GlueCEPolicyMaxWallClockTime == 0) || (other.GlueCEPolicyMaxWallClockTime>='+str(clockTime)+'))'
665 –  #
666 –  #      cpuTime=1000
667 –  #      if self.EDG_cpu_time:
668 –  #          cpuTime=self.EDG_cpu_time
669 –  #      if (req != noreq):
670 –  #          req = req + ' && '
671 –   #     req = req + '((other.GlueCEPolicyMaxCPUTime == 0) || (other.GlueCEPolicyMaxCPUTime>='+str(cpuTime)+'))'
672 –   #
673 –   #     if (req != noreq):
674 –   #         req = req + ';\n'
675 –   #         jdl.write(req)
676 –   #                                                                                                                                                          
677 –   #     jdl.write('VirtualOrganisation = "' + self.VO + '";\n')
678 –   #
679 –   #     if ( self.EDG_retry_count ):              
680 –   #         jdl.write('RetryCount = '+self.EDG_retry_count+';\n')
681 –   #         pass
682 –   #
683 –   #     jdl.write('MyProxyServer = "' + self.proxyServer + '";\n')
684 – #
685 – #        jdl.close()
686 – #        return
687 –  
596      ##### FEDE ######        
597 <    def findSites_(self, n_tot_job):
598 <        itr4=''
599 <        #print "n_tot_job = ", n_tot_job
600 <        for n in range(n_tot_job):
601 <            sites = common.jobDB.destination(n)
602 <            #job = common.job_list[n]
603 <            #jbt = job.type()
696 <        #    print "common.jobDB.destination(n) = ", common.jobDB.destination(n)
697 <        #    print "sites = ", sites
597 >    def findSites_(self, n):
598 >        itr4 =[]
599 >        sites = common.jobDB.destination(n)
600 >        if len(sites)>0 and sites[0]=="Any":
601 >            return itr4
602 >        itr = ''
603 >        if sites != [""]:#CarlosDaniele
604              for site in sites:
605 <                itr4 = itr4 + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
606 <                pass
607 <            # remove last ||
608 <            itr4 = itr4[0:-4]
703 <            itr4 = itr4 + ','
704 <        # remove last ,
705 <        itr4 = itr4[0:-1]
706 <        #print "itr4 = ", itr4
605 >                #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
606 >                itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
607 >            itr = itr[0:-4]
608 >            itr4.append( itr )
609          return itr4
610  
611 <    def createXMLSchScript(self, nj, argsList, jobList):
612 <   # def createXMLSchScript(self, nj):
611 >    def createXMLSchScript(self, nj, argsList):
612 >      
613          """
614          Create a XML-file for BOSS4.
615          """
616    #      job = common.job_list[nj]
617          """
618          INDY
619 <        [begin] da rivedere:
620 <        in particolare passerei il jobType ed eliminerei le dipendenze da job
619 >        [begin] FIX-ME:
620 >        I would pass jobType instead of job
621          """
622          index = nj - 1
623          job = common.job_list[index]
# Line 724 | Line 626 | class SchedulerEdg(Scheduler):
626          inp_sandbox = jbt.inputSandbox(index)
627          out_sandbox = jbt.outputSandbox(index)
628          """
629 <        [end] da rivedere
629 >        [end] FIX-ME
630          """
631  
632          
# Line 742 | Line 644 | class SchedulerEdg(Scheduler):
644          to_write = ''
645  
646          req=' '
647 <        req = req + jbt.getRequirements(nj)
647 >        req = req + jbt.getRequirements()
648  
747 –
748 –        #sites = common.jobDB.destination(nj)
749 –        #if len(sites)>0 and sites[0]!="Any":
750 –        #    req = req + ' && anyMatch(other.storage.CloseSEs, (_ITR4_))'
751 –        #req = req    
752 –    
649          if self.EDG_requirements:
650              if (req == ' '):
651                  req = req + self.EDG_requirements
# Line 787 | Line 683 | class SchedulerEdg(Scheduler):
683                  req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
684              else:
685                  req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
686 <
791 <        #if (req != ' '):
792 <        #    req = req + '\n'
793 <        #    to_writeReq = req
794 <                                                                                                                                                            
686 >                                                                                          
687          if ( self.EDG_retry_count ):              
688              to_write = to_write + 'RetryCount = "'+self.EDG_retry_count+'"\n'
689              pass
690  
691 +        if ( self.EDG_shallow_retry_count ):              
692 +            to_write = to_write + 'ShallowRetryCount = "'+self.EDG_shallow_retry_count+'"\n'
693 +            pass
694 +
695          to_write = to_write + 'MyProxyServer = "&quot;' + self.proxyServer + '&quot;"\n'
696          to_write = to_write + 'VirtualOrganisation = "&quot;' + self.VO + '&quot;"\n'
697  
802 –
698          #TaskName  
699          dir = string.split(common.work_space.topDir(), '/')
700          taskName = dir[len(dir)-2]
# Line 807 | Line 702 | class SchedulerEdg(Scheduler):
702          xml.write(str(title))
703          xml.write('<task name="' +str(taskName)+'">\n')
704          xml.write(jt_string)
705 <
811 <        xml.write('<iterator>\n')
812 <
813 <        #print str(nj)
814 <        xml.write('\t<iteratorRule name="ITR1" rule="1:'+ str(nj) + '" />\n')
815 <        #print argsList
816 <        xml.write('\t<iteratorRule name="ITR2" rule="'+ argsList + '" />\n')
817 <        #print jobList
818 <        xml.write('\t<iteratorRule name="ITR3" rule="'+ jobList + '" />\n')
819 <
820 <        #### FEDE #####
821 <        itr4=self.findSites_(nj)
822 <        #print "--->>> itr4 = ", itr4
823 <        if (itr4 != ''):
824 <           xml.write('\t<iteratorRule name="ITR4" rule="'+itr4+ '" />\n')
825 <           req = req + ' && anyMatch(other.storage.CloseSEs, (_ITR4_))'
826 <           pass
827 <    #    print "--->>> req= ", req        
828 <  
705 >        
706          if (to_write != ''):
707              xml.write('<extraTags\n')
708              xml.write(to_write)
709              xml.write('/>\n')
710              pass
711  
712 +        xml.write('<iterator>\n')
713 +        xml.write('\t<iteratorRule name="ITR1">\n')
714 +        xml.write('\t\t<ruleElement> 1:'+ str(nj) + ' </ruleElement>\n')
715 +        xml.write('\t</iteratorRule>\n')
716 +        xml.write('\t<iteratorRule name="ITR2">\n')
717 +        for arg in argsList:
718 +            xml.write('\t\t<ruleElement> <![CDATA[\n'+ arg + '\n\t\t]]> </ruleElement>\n')
719 +            pass
720 +        xml.write('\t</iteratorRule>\n')
721 +        #print jobList
722 +        xml.write('\t<iteratorRule name="ITR3">\n')
723 +        xml.write('\t\t<ruleElement> 1:'+ str(nj) + ':1:6 </ruleElement>\n')
724 +        xml.write('\t</iteratorRule>\n')
725 +
726 +        '''
727 +        indy: here itr4
728 +        '''
729 +        
730 +
731          xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
732          xml.write(jt_string)
733  
838 –        if (req != ' '):
839 –            req = req + '\n'
840 –            xml.write('<extraTags>\n')
841 –            xml.write('<Requirements>\n')
842 –            xml.write('<![CDATA[\n')
843 –            xml.write(req)
844 –            xml.write(']]>\n')
845 –            xml.write('</Requirements>\n')
846 –            xml.write('</extraTags>\n')
847 –            pass
848 –
734          #executable
735  
736          """
737          INDY
738 <        script dipende dal jobType: dovrebbe essere semplice tirarlo fuori in altro modo
738 >        script depends on jobType: it should be probably get in a different way
739          """        
740          script = job.scriptFilename()
741 <        xml.write('<program exec="' + os.path.basename(script) +'"\n')
741 >        xml.write('<program>\n')
742 >        xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
743          xml.write(jt_string)
744      
745 <          
746 <        ### only one .sh  JDL has arguments:
747 <        ### Fabio
862 < #        xml.write('args = "' + str(nj+1)+' '+ jbt.getJobTypeArguments(nj, "EDG") +'"\n')
863 <        xml.write('args = "_ITR2_"\n')
864 <        xml.write('program_types="crabjob"\n')
865 <        inp_box = 'infiles="'
866 <        inp_box = inp_box + '' + script + ','
745 >        xml.write('<args> <![CDATA[\n _ITR2_ \n]]> </args>\n')
746 >        xml.write('<program_types> crabjob </program_types>\n')
747 >        inp_box = script + ','
748  
749          if inp_sandbox != None:
750              for fl in inp_sandbox:
# Line 875 | Line 756 | class SchedulerEdg(Scheduler):
756                    os.path.abspath(os.environ['CRABDIR']+'/python/'+'DashboardAPI.py') + ','+\
757                    os.path.abspath(os.environ['CRABDIR']+'/python/'+'Logger.py') + ','+\
758                    os.path.abspath(os.environ['CRABDIR']+'/python/'+'ProcInfo.py') + ','+\
759 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'apmon.py')
759 >                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'apmon.py') + ','+\
760 >                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'parseCrabFjr.py')
761  
762          if (not jbt.additional_inbox_files == []):
763 <            inp_box = inp_box + ', '
763 >            inp_box = inp_box + ','
764              for addFile in jbt.additional_inbox_files:
765                  addFile = os.path.abspath(addFile)
766                  inp_box = inp_box+''+addFile+','
767                  pass
768  
769          if inp_box[-1] == ',' : inp_box = inp_box[:-1]
770 <        inp_box = inp_box + ' "\n'
770 >        inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
771          xml.write(inp_box)
772          
773          base = jbt.name()
774          stdout = base + '__ITR3_.stdout'
775          stderr = base + '__ITR3_.stderr'
776          
777 <        xml.write('stderr="' + stderr + '"\n')
778 <        xml.write('stdout="' + stdout + '"\n')
777 >        xml.write('<stderr> ' + stderr + '</stderr>\n')
778 >        xml.write('<stdout> ' + stdout + '</stdout>\n')
779          
780  
781 <        out_box = 'outfiles="' + \
900 <                  stdout + ',' + \
781 >        out_box = stdout + ',' + \
782                    stderr + ',.BrokerInfo,'
783  
784          """
# Line 912 | Line 793 | class SchedulerEdg(Scheduler):
793  
794          """
795          INDY
796 <        qualcosa del genere andrebbe fatta per gli infiles
797 <        """        
796 >        something similar should be also done for infiles (if it makes sense!)
797 >        """
798 >        # Stuff to be returned _always_ via sandbox
799 >        for fl in jbt.output_file_sandbox:
800 >            out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
801 >            pass
802 >        pass
803 >
804 >        # via sandbox iif required return_data
805          if int(self.return_data) == 1:
806              for fl in jbt.output_file:
807                  out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
# Line 921 | Line 809 | class SchedulerEdg(Scheduler):
809              pass
810  
811          if out_box[-1] == ',' : out_box = out_box[:-1]
812 <        out_box = out_box + '"'
813 <        xml.write(out_box+'\n')
812 >        out_box = '<outfiles> <![CDATA[\n' + out_box + '\n]]></outfiles>\n'
813 >        xml.write(out_box)
814  
815 <        xml.write('group="'+taskName+'"\n')
928 <        xml.write('BossAttr="crabjob.INTERNAL_ID=_ITR1_"\n')
815 >        xml.write('<BossAttr> crabjob.INTERNAL_ID=_ITR1_ </BossAttr>\n')
816  
817 <        xml.write('/>\n')
817 >        xml.write('</program>\n')
818          xml.write('</chain>\n')
819  
820          xml.write('</iterator>\n')
821          xml.write('</task>\n')
822  
823          xml.close()
824 +      
825 +
826          return
827  
828      def checkProxy(self):
# Line 941 | Line 830 | class SchedulerEdg(Scheduler):
830          Function to check the Globus proxy.
831          """
832          if (self.proxyValid): return
833 +
834 +        ### Just return if asked to do so
835 +        if (self.dontCheckProxy):
836 +            self.proxyValid=1
837 +            return
838 +
839          timeleft = -999
840          minTimeLeft=10*3600 # in seconds
841  
# Line 961 | Line 856 | class SchedulerEdg(Scheduler):
856          pass
857  
858          if mustRenew:
859 <            common.logger.message( "No valid proxy found or remaining time of validity of already existing proxy shorter than 10 hours!\n Creating a user proxy with default length of 96h\n")
860 <            cmd = 'voms-proxy-init -voms cms -valid 96:00'
859 >            common.logger.message( "No valid proxy found or remaining time of validity of already existing proxy shorter than 10 hours!\n Creating a user proxy with default length of 192h\n")
860 >            cmd = 'voms-proxy-init -voms '+self.VO
861 >            if self.group:
862 >                cmd += ':/'+self.VO+'/'+self.group
863 >            if self.role:
864 >                cmd += '/role='+self.role
865 >            cmd += ' -valid 192:00'
866              try:
867                  # SL as above: damn it!
868 +                common.logger.debug(10,cmd)
869                  out = os.system(cmd)
870                  if (out>0): raise CrabException("Unable to create a valid proxy!\n")
871              except:
# Line 982 | Line 883 | class SchedulerEdg(Scheduler):
883          else:
884              # if myproxy exist but not long enough, renew
885              reTime = re.compile( r'timeleft: (\d+)' )
886 +            #print "<"+str(reTime.search( cmd_out ).group(1))+">"
887              if reTime.match( cmd_out ):
888                  time = reTime.search( line ).group(1)
889                  if time < minTimeLeftServer:

Diff Legend

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