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.101 by slacapra, Tue Oct 17 09:52:57 2006 UTC vs.
Revision 1.153 by mcinquil, Mon Nov 12 15:09:14 2007 UTC

# Line 3 | Line 3 | from crab_logger import Logger
3   from crab_exceptions import *
4   from crab_util import *
5   from EdgConfig import *
6 + from BlackWhiteListParser import BlackWhiteListParser
7   import common
8  
9   import os, sys, time
# Line 22 | Line 23 | class SchedulerEdg(Scheduler):
23  
24      def configure(self, cfg_params):
25  
26 +        # init BlackWhiteListParser
27 +        self.blackWhiteListParser = BlackWhiteListParser(cfg_params)
28 +
29 +        self.proxyValid=0
30 +        try: self.dontCheckProxy=int(cfg_params["EDG.dont_check_proxy"])
31 +        except KeyError: self.dontCheckProxy = 0
32 +
33          try:
34              RB=cfg_params["EDG.rb"]
35              self.rb_param_file=self.rb_configure(RB)
# Line 44 | Line 52 | class SchedulerEdg(Scheduler):
52          except KeyError:
53              self.role = None
54              
55 <        try: self.LCG_version = cfg_params["EDG.lcg_version"]
56 <        except KeyError: self.LCG_version = '2'
55 >        #try: self.LCG_version = cfg_params["EDG.lcg_version"]
56 >        #except KeyError: self.LCG_version = '2'
57  
58          try:
59              self.EDG_ce_black_list = cfg_params['EDG.ce_black_list']
# Line 59 | Line 67 | class SchedulerEdg(Scheduler):
67          try: self.VO = cfg_params['EDG.virtual_organization']
68          except KeyError: self.VO = 'cms'
69  
70 +        try: self.copy_input_data = cfg_params["USER.copy_input_data"]
71 +        except KeyError: self.copy_input_data = 0
72 +
73          try: self.return_data = cfg_params['USER.return_data']
74          except KeyError: self.return_data = 0
75  
# Line 76 | Line 87 | class SchedulerEdg(Scheduler):
87          except KeyError: self.copy_data = 0
88  
89          if ( int(self.return_data) == 0 and int(self.copy_data) == 0 ):
90 <           msg = 'Warning: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
90 >           msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
91             msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
92             raise CrabException(msg)
93  
94 <        try:
95 <            self.lfc_host = cfg_params['EDG.lfc_host']
96 <        except KeyError:
97 <            msg = "Error. The [EDG] section does not have 'lfc_host' value"
98 <            msg = msg + " it's necessary to know the LFC host name"
99 <            common.logger.message(msg)
89 <            raise CrabException(msg)
90 <        try:
91 <            self.lcg_catalog_type = cfg_params['EDG.lcg_catalog_type']
92 <        except KeyError:
93 <            msg = "Error. The [EDG] section does not have 'lcg_catalog_type' value"
94 <            msg = msg + " it's necessary to know the catalog type"
95 <            common.logger.message(msg)
96 <            raise CrabException(msg)
97 <        try:
98 <            self.lfc_home = cfg_params['EDG.lfc_home']
99 <        except KeyError:
100 <            msg = "Error. The [EDG] section does not have 'lfc_home' value"
101 <            msg = msg + " it's necessary to know the home catalog dir"
102 <            common.logger.message(msg)
103 <            raise CrabException(msg)
104 <      
94 >        if ( int(self.return_data) == 1 and int(self.copy_data) == 1 ):
95 >           msg = 'Error: return_data and copy_data cannot be set both to 1\n'
96 >           msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
97 >           raise CrabException(msg)
98 >
99 >        ########### FEDE FOR DBS2 ##############################
100          try:
101 <            self.register_data = cfg_params["USER.register_data"]
102 <            if int(self.register_data) == 1:
101 >            self.publish_data = cfg_params["USER.publish_data"]
102 >            self.checkProxy()
103 >            if int(self.publish_data) == 1:
104                  try:
105 <                    self.LFN = cfg_params['USER.lfn_dir']
105 >                    self.publish_data_name = cfg_params['USER.publish_data_name']
106                  except KeyError:
107 <                    msg = "Error. The [USER] section does not have 'lfn_dir' value"
108 <                    msg = msg + " it's necessary for LCF registration"
109 <                    common.logger.message(msg)
107 >                    msg = "Error. The [USER] section does not have 'publish_data_name'"
108 >                    raise CrabException(msg)
109 >                try:
110 >                    tmp = runCommand("voms-proxy-info -identity")
111 >                    tmp = string.split(tmp,'/')
112 >                    reCN=re.compile(r'CN=')
113 >                    for t in tmp:
114 >                        if reCN.match(t):
115 >                            self.UserGridName=string.strip((t.replace('CN=','')).replace(' ',''))
116 >                        
117 >                    #self.UserGridName = string.strip(runCommand("voms-proxy-info -identity | awk -F\'CN\' \'{print $2$3$4}\' | tr -d \'=/ \'"))
118 >                except:
119 >                    msg = "Error. Problem with voms-proxy-info -identity command"
120                      raise CrabException(msg)
121 <        except KeyError: self.register_data = 0
121 >        except KeyError: self.publish_data = 0
122  
123 <        if ( int(self.copy_data) == 0 and int(self.register_data) == 1 ):
124 <           msg = 'Warning: register_data = 1 must be used with copy_data = 1\n'
123 >        if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
124 >           msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
125             msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
126             common.logger.message(msg)
127             raise CrabException(msg)
128 +        #################################################
129 +
130 +        #try:
131 +        #    self.lfc_host = cfg_params['EDG.lfc_host']
132 +        #except KeyError:
133 +        #    msg = "Error. The [EDG] section does not have 'lfc_host' value"
134 +        #    msg = msg + " it's necessary to know the LFC host name"
135 +        #    common.logger.message(msg)
136 +        #    raise CrabException(msg)
137 +        #try:
138 +        #    self.lcg_catalog_type = cfg_params['EDG.lcg_catalog_type']
139 +        #except KeyError:
140 +        #    msg = "Error. The [EDG] section does not have 'lcg_catalog_type' value"
141 +        #    msg = msg + " it's necessary to know the catalog type"
142 +        #    common.logger.message(msg)
143 +        #    raise CrabException(msg)
144 +        #try:
145 +        #    self.lfc_home = cfg_params['EDG.lfc_home']
146 +        #except KeyError:
147 +        #    msg = "Error. The [EDG] section does not have 'lfc_home' value"
148 +        #    msg = msg + " it's necessary to know the home catalog dir"
149 +        #    common.logger.message(msg)
150 +        #    raise CrabException(msg)
151 +      
152 +        #try:
153 +        #    self.register_data = cfg_params["USER.register_data"]
154 +        #    if int(self.register_data) == 1:
155 +        #        try:
156 +        #            self.LFN = cfg_params['USER.lfn_dir']
157 +        #        except KeyError:
158 +        #            msg = "Error. The [USER] section does not have 'lfn_dir' value"
159 +        #            msg = msg + " it's necessary for LCF registration"
160 +        #            common.logger.message(msg)
161 +        #            raise CrabException(msg)
162 +        #except KeyError: self.register_data = 0
163 +
164 +        #if ( int(self.copy_data) == 0 and int(self.register_data) == 1 ):
165 +        #   msg = 'Warning: register_data = 1 must be used with copy_data = 1\n'
166 +        #   msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
167 +        #   common.logger.message(msg)
168 +        #   raise CrabException(msg)
169  
170          try: self.EDG_requirements = cfg_params['EDG.requirements']
171          except KeyError: self.EDG_requirements = ''
# Line 151 | Line 198 | class SchedulerEdg(Scheduler):
198          libPath=os.path.join(path, "lib", "python")
199          sys.path.append(libPath)
200  
154 –        self.proxyValid=0
155 –
201          try:
202              self._taskId = cfg_params['taskId']
203          except:
# Line 177 | Line 222 | class SchedulerEdg(Scheduler):
222          self.edg_config_vo = edgConfig.configVO()
223  
224          if (self.edg_config and self.edg_config_vo != ''):
225 <            self.rb_param_file = 'RBconfig = "'+self.edg_config+'";\nRBconfigVO = "'+self.edg_config_vo+'";'
225 >            self.rb_param_file = 'RBconfig = "'+self.edg_config+'";\nRBconfigVO = "'+self.edg_config_vo+'";\n'
226              #print "rb_param_file = ", self.rb_param_file
227          return self.rb_param_file
228        
# Line 212 | Line 257 | class SchedulerEdg(Scheduler):
257              for i in range(len(ce_white_list)):
258                  if i == 0:
259                      if (req == ' '):
260 <                        req = req + '((RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
260 >                        req = req + '((RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
261                      else:
262 <                        req = req +  ' && ((RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
262 >                        req = req +  ' && ((RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
263                      pass
264                  else:
265 <                    req = req +  ' || (RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
265 >                    req = req +  ' || (RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
266              req = req + ')'
267          
268          if self.EDG_ce_black_list:
269              ce_black_list = string.split(self.EDG_ce_black_list,',')
270              for ce in ce_black_list:
271                  if (req == ' '):
272 <                    req = req + '(!RegExp("' + ce + '", other.GlueCEUniqueId))'
272 >                    req = req + '(!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
273                  else:
274 <                    req = req +  ' && (!RegExp("' + ce + '", other.GlueCEUniqueId))'
274 >                    req = req +  ' && (!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
275                  pass
276          if self.EDG_clock_time:
277              if (req == ' '):
# Line 241 | Line 286 | class SchedulerEdg(Scheduler):
286                  req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
287                  
288          for i in range(len(first)): # Add loop DS
289 +            groupReq = req
290              self.param='sched_param_'+str(i)+'.clad'
291              param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')
292  
293              itr4=self.findSites_(first[i])
294              for arg in itr4:
295 <                req = req + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
296 <            param_file.write('Requirements = '+req +';\n')  
295 >                groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
296 >            param_file.write('Requirements = '+groupReq +';\n')  
297    
298              if (self.rb_param_file != ''):
299                  param_file.write(self.rb_param_file)  
# Line 281 | Line 327 | class SchedulerEdg(Scheduler):
327          txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
328          txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
329  
330 <        txt += 'echo "middleware discovery " \n'
330 >        txt += 'echo "middleware discovery: " \n'
331          txt += 'if [ $GRID3_APP_DIR ]; then\n'
332          txt += '    middleware=OSG \n'
333 <        txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
333 >        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
334 >        txt += '        SyncCE="$OSG_JOB_CONTACT"; \n'
335 >        txt += '        echo "SyncCE=$SyncCE" | tee -a $RUNTIME_AREA/$repo ;\n'
336 >        txt += '    else\n'
337 >        txt += '        echo "not reporting SyncCE";\n'
338 >        txt += '    fi\n';
339          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
340 <        txt += '    echo "middleware =$middleware" \n'
340 >        txt += '    echo ">>> middleware =$middleware" \n'
341          txt += 'elif [ $OSG_APP ]; then \n'
342          txt += '    middleware=OSG \n'
343 <        txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
343 >        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
344 >        txt += '        SyncCE="$OSG_JOB_CONTACT"; \n'
345 >        txt += '        echo "SyncCE=$SyncCE" | tee -a $RUNTIME_AREA/$repo ;\n'
346 >        txt += '    else\n'
347 >        txt += '        echo "not reporting SyncCE";\n'
348 >        txt += '    fi\n';
349          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
350 <        txt += '    echo "middleware =$middleware" \n'
350 >        txt += '    echo ">>> middleware =$middleware" \n'
351          txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
352          txt += '    middleware=LCG \n'
353 <        txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
353 >   #     txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
354 >        txt += '    echo "SyncCE=`glite-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
355          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
356 <        txt += '    echo "middleware =$middleware" \n'
356 >        txt += '    echo ">>> middleware =$middleware" \n'
357          txt += 'else \n'
358          txt += '    echo "SET_CMS_ENV 10030 ==> middleware not identified" \n'
359          txt += '    echo "JOB_EXIT_STATUS = 10030" \n'
360          txt += '    echo "JobExitCode=10030" | tee -a $RUNTIME_AREA/$repo \n'
361          txt += '    dumpStatus $RUNTIME_AREA/$repo \n'
305 –        txt += '    rm -f $RUNTIME_AREA/$repo \n'
306 –        txt += '    echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
307 –        txt += '    echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
362          txt += '    exit 1 \n'
363          txt += 'fi \n'
364  
311 –        txt += '# report first time to DashBoard \n'
365          txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
313 –        txt += 'rm -f $RUNTIME_AREA/$repo \n'
314 –        txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
315 –        txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
366          
367          txt += '\n\n'
368  
319 –        if int(self.copy_data) == 1:
320 –           if self.SE:
321 –              txt += 'export SE='+self.SE+'\n'
322 –              txt += 'echo "SE = $SE"\n'
323 –           if self.SE_PATH:
324 –              if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
325 –              txt += 'export SE_PATH='+self.SE_PATH+'\n'
326 –              txt += 'echo "SE_PATH = $SE_PATH"\n'
327 –
369          txt += 'export VO='+self.VO+'\n'
329 –        ### add some line for LFC catalog setting
330 –        txt += 'if [ $middleware == LCG ]; then \n'
331 –        txt += '    if [[ $LCG_CATALOG_TYPE != \''+self.lcg_catalog_type+'\' ]]; then\n'
332 –        txt += '        export LCG_CATALOG_TYPE='+self.lcg_catalog_type+'\n'
333 –        txt += '    fi\n'
334 –        txt += '    if [[ $LFC_HOST != \''+self.lfc_host+'\' ]]; then\n'
335 –        txt += '        export LFC_HOST='+self.lfc_host+'\n'
336 –        txt += '    fi\n'
337 –        txt += '    if [[ $LFC_HOME != \''+self.lfc_home+'\' ]]; then\n'
338 –        txt += '        export LFC_HOME='+self.lfc_home+'\n'
339 –        txt += '    fi\n'
340 –        txt += 'elif [ $middleware == OSG ]; then\n'
341 –        txt += '    echo "LFC catalog setting to be implemented for OSG"\n'
342 –        txt += 'fi\n'
343 –        #####
344 –        if int(self.register_data) == 1:
345 –           txt += 'if [ $middleware == LCG ]; then \n'
346 –           txt += '    export LFN='+self.LFN+'\n'
347 –           txt += '    lfc-ls $LFN\n'
348 –           txt += '    result=$?\n'
349 –           txt += '    echo $result\n'
350 –           ### creation of LFN dir in LFC catalog, under /grid/cms dir  
351 –           txt += '    if [ $result != 0 ]; then\n'
352 –           txt += '       lfc-mkdir $LFN\n'
353 –           txt += '       result=$?\n'
354 –           txt += '       echo $result\n'
355 –           txt += '    fi\n'
356 –           txt += 'elif [ $middleware == OSG ]; then\n'
357 –           txt += '    echo " Files registration to be implemented for OSG"\n'
358 –           txt += 'fi\n'
359 –           txt += '\n'
360 –
361 –           if self.VO:
362 –              txt += 'export VO='+self.VO+'\n'
363 –           if self.LFN:
364 –              txt += 'if [ $middleware == LCG ]; then \n'
365 –              txt += '    export LFN='+self.LFN+'\n'
366 –              txt += 'fi\n'
367 –              txt += '\n'
368 –
370          txt += 'if [ $middleware == LCG ]; then\n'
371 <        txt += '    CloseCEs=`edg-brokerinfo getCE`\n'
371 >        txt += '    CloseCEs=`glite-brokerinfo getCE`\n'
372          txt += '    echo "CloseCEs = $CloseCEs"\n'
373          txt += '    CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
374          txt += '    echo "CE = $CE"\n'
# Line 379 | Line 380 | class SchedulerEdg(Scheduler):
380          txt += '        echo "JOB_EXIT_STATUS = 10099" \n'
381          txt += '        echo "JobExitCode=10099" | tee -a $RUNTIME_AREA/$repo \n'
382          txt += '        dumpStatus $RUNTIME_AREA/$repo \n'
382 –        txt += '        rm -f $RUNTIME_AREA/$repo \n'
383 –        txt += '        echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
384 –        txt += '        echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
383          txt += '        exit 1 \n'
384          txt += '    fi \n'
385          txt += 'fi \n'
# Line 393 | Line 391 | class SchedulerEdg(Scheduler):
391          Copy input data from SE to WN    
392          """
393          txt = ''
394 +        if not self.copy_input_data: return txt
395  
396          ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
397          txt += 'if [ $middleware == OSG ]; then\n'
# Line 440 | Line 439 | class SchedulerEdg(Scheduler):
439          Write a CopyResults part of a job script, e.g.
440          to copy produced output into a storage element.
441          """
442 <        txt = ''
444 <        if int(self.copy_data) == 1:
445 <           txt += '#\n'
446 <           txt += '#   Copy output to SE = $SE\n'
447 <           txt += '#\n'
448 <           txt += '    if [ $middleware == OSG ]; then\n'
449 <           txt += '        echo "X509_USER_PROXY = $X509_USER_PROXY"\n'
450 <           txt += '        echo "source $OSG_APP/glite/setup_glite_ui.sh"\n'
451 <           txt += '        source $OSG_APP/glite/setup_glite_ui.sh\n'
452 <           txt += '        export X509_CERT_DIR=$OSG_APP/glite/etc/grid-security/certificates\n'
453 <           txt += '        echo "export X509_CERT_DIR=$X509_CERT_DIR"\n'
454 <           txt += '    fi \n'
455 <
456 <           txt += '    for out_file in $file_list ; do\n'
457 <           txt += '        echo "Trying to copy output file to $SE using srmcp"\n'
458 <           txt += '        echo "mkdir -p $HOME/.srmconfig"\n'
459 <           txt += '        mkdir -p $HOME/.srmconfig\n'
460 <           txt += '        if [ $middleware == LCG ]; then\n'
461 <           txt += '           echo "srmcp -retry_num 3 -retry_timeout 480000 file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
462 <           txt += '           exitstring=`srmcp -retry_num 3 -retry_timeout 480000 file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
463 <           txt += '        elif [ $middleware == OSG ]; then\n'
464 <           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'
465 <           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'
466 <           txt += '        fi \n'
467 <           txt += '        copy_exit_status=$?\n'
468 <           txt += '        echo "COPY_EXIT_STATUS for srmcp = $copy_exit_status"\n'
469 <           txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
470 <
471 <           txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
472 <           txt += '            echo "Possible problem with SE = $SE"\n'
473 <           txt += '            echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
474 <           txt += '            echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
475 <           txt += '            echo "srmcp failed, attempting lcg-cp."\n'
476 <           if common.logger.debugLevel() >= 5:
477 <               txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
478 <               txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
479 <           else:              
480 <               txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
481 <               txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
482 <           txt += '            copy_exit_status=$?\n'
483 <           txt += '            echo "COPY_EXIT_STATUS for lcg-cp = $copy_exit_status"\n'
484 <           txt += '            echo "STAGE_OUT = $copy_exit_status"\n'
485 <
486 <           txt += '            if [ $copy_exit_status -ne 0 ]; then\n'
487 <           txt += '               echo "Problems with SE = $SE"\n'
488 <           txt += '               echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
489 <           txt += '               echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
490 <           txt += '               echo "srmcp and lcg-cp and failed!"\n'
491 <           txt += '            else\n'
492 <           txt += '               echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
493 <           txt += '               echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
494 <           txt += '               echo "output copied into $SE/$SE_PATH directory"\n'
495 <           txt += '               echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
496 <           txt += '               echo "lcg-cp succeeded"\n'
497 <           txt += '            fi\n'
498 <           txt += '        else\n'
499 <           txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
500 <           txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
501 <           txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
502 <           txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
503 <           txt += '            echo "srmcp succeeded"\n'
504 <           txt += '         fi\n'
505 <           txt += '     done\n'
506 <        return txt
442 >        txt = '\n'
443  
444 <    def wsRegisterOutput(self):
445 <        """
446 <        Returns part of a job script which does scheduler-specific work.
511 <        """
444 >        txt += '#\n'
445 >        txt += '# COPY OUTPUT FILE TO SE\n'
446 >        txt += '#\n\n'
447  
448 <        txt = ''
449 <        if int(self.register_data) == 1:
450 <        ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
451 <           txt += 'if [ $middleware == OSG ]; then\n'
452 <           txt += '   #\n'
453 <           txt += '   #   Register output to LFC deactivated in OSG mode\n'
454 <           txt += '   #\n'
455 <           txt += '   echo "Register output to LFC deactivated in OSG mode"\n'
456 <           txt += 'elif [ $middleware == LCG ]; then \n'
457 <           txt += '#\n'
458 <           txt += '#  Register output to LFC\n'
459 <           txt += '#\n'
460 <           txt += '   if [ $copy_exit_status -eq 0 ]; then\n'
461 <           txt += '      for out_file in $file_list ; do\n'
462 <           txt += '         echo "Trying to register the output file into LFC"\n'
463 <           txt += '         echo "lcg-rf -l $LFN/$out_file --vo $VO -t 1200 sfn://$SE$SE_PATH/$out_file 2>&1"\n'
464 <           txt += '         lcg-rf -l $LFN/$out_file --vo $VO -t 1200 sfn://$SE$SE_PATH/$out_file 2>&1 \n'
465 <           txt += '         register_exit_status=$?\n'
466 <           txt += '         echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
467 <           txt += '         echo "STAGE_OUT = $register_exit_status"\n'
468 <           txt += '         if [ $register_exit_status -ne 0 ]; then \n'
469 <           txt += '            echo "Problems with the registration to LFC" \n'
470 <           txt += '            echo "Try with srm protocol" \n'
471 <           txt += '            echo "lcg-rf -l $LFN/$out_file --vo $VO -t 1200 srm://$SE$SE_PATH/$out_file 2>&1"\n'
472 <           txt += '            lcg-rf -l $LFN/$out_file --vo $VO -t 1200 srm://$SE$SE_PATH/$out_file 2>&1 \n'
473 <           txt += '            register_exit_status=$?\n'
474 <           txt += '            echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
475 <           txt += '            echo "STAGE_OUT = $register_exit_status"\n'
476 <           txt += '            if [ $register_exit_status -ne 0 ]; then \n'
477 <           txt += '               echo "Problems with the registration into LFC" \n'
478 <           txt += '            fi \n'
479 <           txt += '         else \n'
480 <           txt += '            echo "output registered to LFC"\n'
481 <           txt += '         fi \n'
482 <           txt += '         echo "StageOutExitStatus = $register_exit_status" | tee -a $RUNTIME_AREA/$repo\n'
483 <           txt += '      done\n'
484 <           txt += '   else \n'
485 <           txt += '      echo "Trying to copy output file to CloseSE"\n'
486 <           txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
487 <           txt += '      for out_file in $file_list ; do\n'
488 <           txt += '         echo "lcg-cr -v -l lfn:${LFN}/$out_file -d $CLOSE_SE -P $LFN/$out_file --vo $VO file://$RUNTIME_AREA/$out_file 2>&1" \n'
489 <           txt += '         lcg-cr -v -l lfn:${LFN}/$out_file -d $CLOSE_SE -P $LFN/$out_file --vo $VO file://$RUNTIME_AREA/$out_file 2>&1 \n'
490 <           txt += '         register_exit_status=$?\n'
491 <           txt += '         echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
492 <           txt += '         echo "STAGE_OUT = $register_exit_status"\n'
493 <           txt += '         if [ $register_exit_status -ne 0 ]; then \n'
494 <           txt += '            echo "Problems with CloseSE or Catalog" \n'
495 <           txt += '         else \n'
561 <           txt += '            echo "The program was successfully executed"\n'
562 <           txt += '            echo "SE = $CLOSE_SE"\n'
563 <           txt += '            echo "LFN for the file is LFN=${LFN}/$out_file"\n'
564 <           txt += '         fi \n'
565 <           txt += '         echo "StageOutExitStatus = $register_exit_status" | tee -a $RUNTIME_AREA/$repo\n'
566 <           txt += '      done\n'
567 <           txt += '   fi \n'
568 <           txt += '   exit_status=$register_exit_status\n'
569 <           txt += 'fi \n'
448 >        SE_PATH=''
449 >        if int(self.copy_data) == 1:
450 >            if self.SE:
451 >                txt += 'export SE='+self.SE+'\n'
452 >                txt += 'echo "SE = $SE"\n'
453 >            if self.SE_PATH:
454 >                if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
455 >                SE_PATH=self.SE_PATH
456 >            if int(self.publish_data) == 1:
457 >                txt += '### publish_data = 1 so the SE path where to copy the output is: \n'
458 >                path_add = self.UserGridName + '/' + self.publish_data_name +'_${PSETHASH}/'
459 >                SE_PATH = SE_PATH + path_add
460 >            txt += 'export SE_PATH='+SE_PATH+'\n'
461 >            txt += 'echo "SE_PATH = $SE_PATH"\n'
462 >            
463 >            txt += 'echo ">>> Copy output files from WN = `hostname` to SE = $SE :"\n'
464 >            
465 >            txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
466 >            txt += '    echo "--> No output file to copy to $SE"\n'
467 >            txt += '    copy_exit_status=$output_exit_status\n'
468 >            txt += '    echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
469 >            txt += 'else\n'
470 >            txt += '    for out_file in $file_list ; do\n'
471 >            txt += '        echo "Trying to copy output file to $SE"\n'
472 >            txt += '        cmscp $out_file ${SE} ${SE_PATH} $out_file $middleware\n'
473 >            txt += '        copy_exit_status=$?\n'
474 >            txt += '        echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
475 >            txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
476 >            txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
477 >            txt += '            echo "Problem copying $out_file to $SE $SE_PATH"\n'
478 >            txt += '            echo "StageOutExitStatus = $copy_exit_status " | tee -a $RUNTIME_AREA/$repo\n'
479 >            txt += '            copy_exit_status=60307\n'
480 >            txt += '        else\n'
481 >            txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
482 >            txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
483 >            txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
484 >            txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
485 >            txt += '        fi\n'
486 >            txt += '    done\n'
487 >            txt += '    if [ $copy_exit_status -ne 0 ]; then\n'
488 >            txt += '        SE=""\n'
489 >            txt += '        echo "SE = $SE"\n'
490 >            txt += '        SE_PATH=""\n'
491 >            txt += '        echo "SE_PATH = $SE_PATH"\n'
492 >            txt += '    fi\n'
493 >            txt += 'fi\n'
494 >            txt += 'exit_status=$copy_exit_status\n'
495 >            pass
496          return txt
497  
498      def loggingInfo(self, id):
# Line 575 | Line 501 | class SchedulerEdg(Scheduler):
501          """
502          self.checkProxy()
503          cmd = 'edg-job-get-logging-info -v 2 ' + id
578 –        #cmd_out = os.popen(cmd)
504          cmd_out = runCommand(cmd)
505          return cmd_out
506  
582 –    def getExitStatus(self, id):
583 –        return self.getStatusAttribute_(id, 'exit_code')
584 –
585 –    def queryStatus(self, id):
586 –        return self.getStatusAttribute_(id, 'status')
587 –
588 –    def queryDest(self, id):  
589 –        return self.getStatusAttribute_(id, 'destination')
590 –
591 –
592 –    def getStatusAttribute_(self, id, attr):
593 –        """ Query a status of the job with id """
594 –
595 –        self.checkProxy()
596 –        hstates = {}
597 –        Status = importName('edg_wl_userinterface_common_LbWrapper', 'Status')
598 –        # Bypass edg-job-status interfacing directly to C++ API
599 –        # Job attribute vector to retrieve status without edg-job-status
600 –        level = 0
601 –        # Instance of the Status class provided by LB API
602 –        jobStat = Status()
603 –        st = 0
604 –        #print id, level, attr, self.states.index(attr)
605 –        jobStat.getStatus(id, level)
606 –        #print jobStat.loadStatus(st)
607 –        err, apiMsg = jobStat.get_error()
608 –        if err:
609 –            common.logger.debug(5,'Error caught' + apiMsg)
610 –            return None
611 –        else:
612 –            for i in range(len(self.states)):
613 –                # Fill an hash table with all information retrieved from LB API
614 –                hstates[ self.states[i] ] = jobStat.loadStatus(st)[i]
615 –                #print i, jobStat.loadStatus(st)[i]
616 –            result = jobStat.loadStatus(st)[self.states.index(attr)]
617 –            #print str(result)
618 –            return result
619 –
507      def queryDetailedStatus(self, id):
508          """ Query a detailed status of the job with id """
509          cmd = 'edg-job-status '+id
510          cmd_out = runCommand(cmd)
511          return cmd_out
512  
626 –    ##### FEDE ######        
513      def findSites_(self, n):
514          itr4 =[]
515 +
516          sites = common.jobDB.destination(n)
517 <        if len(sites)>0 and sites[0]=="Any":
517 >
518 >        if len(sites)>0 and sites[0]=="":
519              return itr4
520 +
521          itr = ''
522          if sites != [""]:#CarlosDaniele
523 <            for site in sites:
523 >            ##Addedd Daniele
524 >            replicas = self.blackWhiteListParser.checkBlackList(sites,n)
525 >            if len(replicas)!=0:
526 >                replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
527 >              
528 >            if len(replicas)==0:
529 >                itr = itr + 'target.GlueSEUniqueID=="NONE" '
530 >                #msg = 'No sites remaining that host any part of the requested data! Exiting... '
531 >                #raise CrabException(msg)
532 >            #####        
533 >           # for site in sites:
534 >            for site in replicas:
535                  #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
536                  itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
537              itr = itr[0:-4]
# Line 639 | Line 539 | class SchedulerEdg(Scheduler):
539          return itr4
540  
541      def createXMLSchScript(self, nj, argsList):
642 –   # def createXMLSchScript(self, nj):
542        
543          """
544          Create a XML-file for BOSS4.
# Line 653 | Line 552 | class SchedulerEdg(Scheduler):
552          index = nj - 1
553          job = common.job_list[index]
554          jbt = job.type()
656 –        
555          inp_sandbox = jbt.inputSandbox(index)
556 <        out_sandbox = jbt.outputSandbox(index)
556 >        #out_sandbox = jbt.outputSandbox(index)
557          """
558          [end] FIX-ME
559          """
# Line 671 | Line 569 | class SchedulerEdg(Scheduler):
569          dir = string.split(common.work_space.topDir(), '/')
570          taskName = dir[len(dir)-2]
571    
674 –        to_writeReq = ''
572          to_write = ''
573  
574          req=' '
# Line 726 | Line 623 | class SchedulerEdg(Scheduler):
623          to_write = to_write + 'MyProxyServer = "&quot;' + self.proxyServer + '&quot;"\n'
624          to_write = to_write + 'VirtualOrganisation = "&quot;' + self.VO + '&quot;"\n'
625  
626 <                #TaskName  
626 >        #TaskName  
627          dir = string.split(common.work_space.topDir(), '/')
628          taskName = dir[len(dir)-2]
629  
630          xml.write(str(title))
631 <        xml.write('<task name="' +str(taskName)+'">\n')
631 >
632 >        #First check the X509_USER_PROXY. In not there use the default
633 >        try:    
634 >            x509=os.environ['X509_USER_PROXY']
635 >        except Exception, ex:
636 >            import traceback
637 >            common.logger.debug( 6, str(ex) )
638 >            common.logger.debug( 6, traceback.format_exc() )
639 >            x509_cmd = 'ls /tmp/x509up_u`id -u`'
640 >            x509=runCommand(x509_cmd).strip()
641 >        xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + ' task_info="' + str(x509) + '">\n')
642          xml.write(jt_string)
643          
644          if (to_write != ''):
# Line 758 | Line 665 | class SchedulerEdg(Scheduler):
665          indy: here itr4
666          '''
667          
668 <
669 <        xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
668 >        xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.schedulerName)+'">\n')
669 >       # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
670          xml.write(jt_string)
671  
672          #executable
# Line 772 | Line 679 | class SchedulerEdg(Scheduler):
679          xml.write('<program>\n')
680          xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
681          xml.write(jt_string)
682 <    
682 >
683          xml.write('<args> <![CDATA[\n _ITR2_ \n]]> </args>\n')
684          xml.write('<program_types> crabjob </program_types>\n')
685 <        inp_box = script + ','
685 >        inp_box = common.work_space.pathForTgz() + 'job/' + jbt.scriptName + ','
686  
687          if inp_sandbox != None:
688              for fl in inp_sandbox:
# Line 783 | Line 690 | class SchedulerEdg(Scheduler):
690                  pass
691              pass
692  
693 <        inp_box = inp_box + os.path.abspath(os.environ['CRABDIR']+'/python/'+'report.py') + ',' +\
694 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'DashboardAPI.py') + ','+\
695 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'Logger.py') + ','+\
696 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'ProcInfo.py') + ','+\
697 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'apmon.py') + ','+\
698 <                  os.path.abspath(os.environ['CRABDIR']+'/python/'+'parseCrabFjr.py')
792 <
793 <        if (not jbt.additional_inbox_files == []):
794 <            inp_box = inp_box + ','
795 <            for addFile in jbt.additional_inbox_files:
796 <                addFile = os.path.abspath(addFile)
797 <                inp_box = inp_box+''+addFile+','
798 <                pass
693 > #        if (not jbt.additional_inbox_files == []):
694 > #            inp_box = inp_box + ','
695 > #            for addFile in jbt.additional_inbox_files:
696 > #                #addFile = os.path.abspath(addFile)
697 > #                inp_box = inp_box+''+addFile+','
698 > #                pass
699  
700          if inp_box[-1] == ',' : inp_box = inp_box[:-1]
701          inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
# Line 826 | Line 726 | class SchedulerEdg(Scheduler):
726          INDY
727          something similar should be also done for infiles (if it makes sense!)
728          """
729 +        # Stuff to be returned _always_ via sandbox
730 +        for fl in jbt.output_file_sandbox:
731 +            out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
732 +            pass
733 +        pass
734 +
735 +        # via sandbox iif required return_data
736          if int(self.return_data) == 1:
737              for fl in jbt.output_file:
738                  out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
# Line 854 | Line 761 | class SchedulerEdg(Scheduler):
761          Function to check the Globus proxy.
762          """
763          if (self.proxyValid): return
764 <        timeleft = -999
764 >
765 >        ### Just return if asked to do so
766 >        if (self.dontCheckProxy==1):
767 >            self.proxyValid=1
768 >            return
769 >
770          minTimeLeft=10*3600 # in seconds
771  
772          minTimeLeftServer = 100 # in hours
# Line 899 | Line 811 | class SchedulerEdg(Scheduler):
811              common.logger.message('No credential delegated to myproxy server '+self.proxyServer+' will do now')
812              renewProxy = 1
813          else:
814 <            # if myproxy exist but not long enough, renew
815 <            reTime = re.compile( r'timeleft: (\d+)' )
816 <            #print "<"+str(reTime.search( cmd_out ).group(1))+">"
817 <            if reTime.match( cmd_out ):
818 <                time = reTime.search( line ).group(1)
819 <                if time < minTimeLeftServer:
820 <                    renewProxy = 1
821 <                    common.logger.message('No credential delegation will expire in '+time+' hours: renew it')
814 >            ## minimum time: 5 days
815 >            minTime = 4 * 24 * 3600
816 >            ## regex to extract the right information
817 >            myproxyRE = re.compile("timeleft: (?P<hours>[\\d]*):(?P<minutes>[\\d]*):(?P<seconds>[\\d]*)")
818 >            for row in cmd_out.split("\n"):
819 >                g = myproxyRE.search(row)
820 >                if g:
821 >                    hours = g.group("hours")
822 >                    minutes = g.group("minutes")
823 >                    seconds = g.group("seconds")
824 >                    timeleft = int(hours)*3600 + int(minutes)*60 + int(seconds)
825 >                    if timeleft < minTime:
826 >                        renewProxy = 1
827 >                        common.logger.message('Your proxy will expire in:\n\t'+hours+' hours '+minutes+' minutes '+seconds+' seconds\n')
828 >                        common.logger.message('Need to renew it:')
829 >                    pass
830                  pass
831              pass
832          
# Line 929 | Line 849 | class SchedulerEdg(Scheduler):
849          if self.edg_config_vo:
850              edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
851          return edg_ui_cfg_opt
852 +
853 +    def tOut(self, list):
854 +        return 120
855 +
856 +

Diff Legend

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