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.108 by slacapra, Wed Dec 6 10:27:30 2006 UTC vs.
Revision 1.156 by fanzago, Mon Dec 10 18:28:57 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, gzip
9 > import os, sys, time
10  
11   class SchedulerEdg(Scheduler):
12      def __init__(self):
# Line 18 | Line 19 | class SchedulerEdg(Scheduler):
19                        "owner","parent_job", "reason","resubmitted","rsl","seed",\
20                        "stateEnterTime","stateEnterTimes","subjob_failed", \
21                        "user tags" , "status" , "status_code","hierarchy"]
21 –        
22          return
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 39 | Line 46 | class SchedulerEdg(Scheduler):
46              self.group = cfg_params["EDG.group"]
47          except KeyError:
48              self.group = None
49 <            
49 >
50          try:
51              self.role = cfg_params["EDG.role"]
52          except KeyError:
53              self.role = None
47 –            
48 –        try: self.LCG_version = cfg_params["EDG.lcg_version"]
49 –        except KeyError: self.LCG_version = '2'
54  
55 <        try:
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']
60 <        except KeyError:
60 >        except KeyError:
61              self.EDG_ce_black_list  = ''
62  
63 <        try:
63 >        try:
64              self.EDG_ce_white_list = cfg_params['EDG.ce_white_list']
65          except KeyError: self.EDG_ce_white_list = ''
66  
# Line 66 | Line 73 | class SchedulerEdg(Scheduler):
73          try: self.return_data = cfg_params['USER.return_data']
74          except KeyError: self.return_data = 0
75  
76 <        try:
76 >        try:
77              self.copy_data = cfg_params["USER.copy_data"]
78              if int(self.copy_data) == 1:
79                  try:
# Line 77 | Line 84 | class SchedulerEdg(Scheduler):
84                      msg = msg + " and/or 'storage_path' entries, necessary to copy the output"
85                      common.logger.message(msg)
86                      raise CrabException(msg)
87 <        except KeyError: self.copy_data = 0
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'
91 <           msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\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 +        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.lfc_host = cfg_params['EDG.lfc_host']
102 <        except KeyError:
103 <            msg = "Error. The [EDG] section does not have 'lfc_host' value"
91 <            msg = msg + " it's necessary to know the LFC host name"
92 <            common.logger.message(msg)
93 <            raise CrabException(msg)
94 <        try:
95 <            self.lcg_catalog_type = cfg_params['EDG.lcg_catalog_type']
96 <        except KeyError:
97 <            msg = "Error. The [EDG] section does not have 'lcg_catalog_type' value"
98 <            msg = msg + " it's necessary to know the catalog type"
99 <            common.logger.message(msg)
100 <            raise CrabException(msg)
101 <        try:
102 <            self.lfc_home = cfg_params['EDG.lfc_home']
103 <        except KeyError:
104 <            msg = "Error. The [EDG] section does not have 'lfc_home' value"
105 <            msg = msg + " it's necessary to know the home catalog dir"
106 <            common.logger.message(msg)
107 <            raise CrabException(msg)
108 <      
109 <        try:
110 <            self.register_data = cfg_params["USER.register_data"]
111 <            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"
116 <                    msg = msg + " it's necessary for LCF registration"
117 <                    common.logger.message(msg)
107 >                    msg = "Error. The [USER] section does not have 'publish_data_name'"
108                      raise CrabException(msg)
109 <        except KeyError: self.register_data = 0
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.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'
125 <           msg = msg + 'Please modify copy_data value in your crab.cfg file\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 = ''
172  
173 <        try: self.EDG_addJdlParam = string.split(cfg_params['EDG.additional_jdl_parameters'],',')
173 >        try: self.EDG_addJdlParam = string.split(cfg_params['EDG.additional_jdl_parameters'],';')
174          except KeyError: self.EDG_addJdlParam = []
175  
176          try: self.EDG_retry_count = cfg_params['EDG.retry_count']
# Line 155 | Line 198 | class SchedulerEdg(Scheduler):
198          libPath=os.path.join(path, "lib", "python")
199          sys.path.append(libPath)
200  
158 –        self.proxyValid=0
159 –
201          try:
202              self._taskId = cfg_params['taskId']
203          except:
# Line 164 | Line 205 | class SchedulerEdg(Scheduler):
205  
206          try: self.jobtypeName = cfg_params['CRAB.jobtype']
207          except KeyError: self.jobtypeName = ''
208 <
208 >
209          try: self.schedulerName = cfg_params['CRAB.scheduler']
210          except KeyError: self.scheduler = ''
211  
171 –        try: self.dontCheckProxy=cfg_params["EDG.dont_check_proxy"]
172 –        except KeyError: self.dontCheckProxy = 0
173 –
212          return
213 <    
213 >
214  
215      def rb_configure(self, RB):
216          self.edg_config = ''
# Line 184 | 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 <      
228 >
229  
230      def sched_parameter(self):
231          """
# Line 196 | Line 234 | class SchedulerEdg(Scheduler):
234          index = int(common.jobDB.nJobs()) - 1
235          job = common.job_list[index]
236          jbt = job.type()
237 <        
237 >
238          lastBlock=-1
239          first = []
240          for n in range(common.jobDB.nJobs()):
# Line 204 | Line 242 | class SchedulerEdg(Scheduler):
242              if (currBlock!=lastBlock):
243                  lastBlock = currBlock
244                  first.append(n)
245 <  
245 >
246          req = ''
247          req = req + jbt.getRequirements()
248 <    
248 >
249          if self.EDG_requirements:
250              if (req == ' '):
251                  req = req + self.EDG_requirements
# Line 226 | Line 264 | class SchedulerEdg(Scheduler):
264                  else:
265                      req = req +  ' || (RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
266              req = req + ')'
267 <        
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:
# Line 246 | Line 284 | class SchedulerEdg(Scheduler):
284                  req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
285              else:
286                  req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
287 <                
287 >
288          for i in range(len(first)): # Add loop DS
289              groupReq = req
290              self.param='sched_param_'+str(i)+'.clad'
# Line 255 | Line 293 | class SchedulerEdg(Scheduler):
293              itr4=self.findSites_(first[i])
294              for arg in itr4:
295                  groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
296 <            param_file.write('Requirements = '+groupReq +';\n')  
297 <  
296 >            param_file.write('Requirements = '+groupReq +';\n')
297 >
298              if (self.rb_param_file != ''):
299 <                param_file.write(self.rb_param_file)  
299 >                param_file.write(self.rb_param_file)
300  
301              if len(self.EDG_addJdlParam):
302 +                if self.EDG_addJdlParam[-1] == '': self.EDG_addJdlParam= self.EDG_addJdlParam[:-1]
303                  for p in self.EDG_addJdlParam:
304 <                    param_file.write(p)
304 >                    param_file.write(string.strip(p)+';\n')
305  
306 <            param_file.close()  
306 >            param_file.close()
307  
308  
309      def wsSetupEnvironment(self):
# Line 289 | Line 328 | class SchedulerEdg(Scheduler):
328          txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
329          txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
330  
331 <        txt += 'echo "middleware discovery " \n'
332 <        txt += 'if [ $GRID3_APP_DIR ]; then\n'
331 >        txt += 'echo "middleware discovery: " \n'
332 >        txt += 'if [ $OSG_APP ]; then \n'
333          txt += '    middleware=OSG \n'
334 <        txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
335 <        txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
336 <        txt += '    echo "middleware =$middleware" \n'
337 <        txt += 'elif [ $OSG_APP ]; then \n'
338 <        txt += '    middleware=OSG \n'
339 <        txt += '    echo "SyncCE=`echo $EDG_WL_LOG_DESTINATION`" | tee -a $RUNTIME_AREA/$repo \n'
334 >        txt += '    if [ $OSG_JOB_CONTACT ]; then \n'
335 >        txt += '        SyncCE="$OSG_JOB_CONTACT"; \n'
336 >        txt += '        echo "SyncCE=$SyncCE" | tee -a $RUNTIME_AREA/$repo ;\n'
337 >        txt += '    else\n'
338 >        txt += '        echo "not reporting SyncCE";\n'
339 >        txt += '    fi\n';
340          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
341 <        txt += '    echo "middleware =$middleware" \n'
341 >        txt += '    echo ">>> middleware =$middleware" \n'
342          txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
343          txt += '    middleware=LCG \n'
344 <        txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
344 >   #     txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
345 >        txt += '    echo "SyncCE=`glite-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
346          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
347 <        txt += '    echo "middleware =$middleware" \n'
347 >        txt += '    echo ">>> middleware =$middleware" \n'
348          txt += 'else \n'
349          txt += '    echo "SET_CMS_ENV 10030 ==> middleware not identified" \n'
350          txt += '    echo "JOB_EXIT_STATUS = 10030" \n'
351          txt += '    echo "JobExitCode=10030" | tee -a $RUNTIME_AREA/$repo \n'
352          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'
353          txt += '    exit 1 \n'
354          txt += 'fi \n'
355  
319 –        txt += '# report first time to DashBoard \n'
356          txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
321 –        txt += 'rm -f $RUNTIME_AREA/$repo \n'
322 –        txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
323 –        txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
324 –        
325 –        txt += '\n\n'
357  
358 <        if int(self.copy_data) == 1:
328 <           if self.SE:
329 <              txt += 'export SE='+self.SE+'\n'
330 <              txt += 'echo "SE = $SE"\n'
331 <           if self.SE_PATH:
332 <              if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
333 <              txt += 'export SE_PATH='+self.SE_PATH+'\n'
334 <              txt += 'echo "SE_PATH = $SE_PATH"\n'
358 >        txt += '\n\n'
359  
360          txt += 'export VO='+self.VO+'\n'
361 <        ### add some line for LFC catalog setting
362 <        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'
341 <        txt += '    fi\n'
342 <        txt += '    if [[ $LFC_HOST != \''+self.lfc_host+'\' ]]; then\n'
343 <        txt += '        export LFC_HOST='+self.lfc_host+'\n'
344 <        txt += '    fi\n'
345 <        txt += '    if [[ $LFC_HOME != \''+self.lfc_home+'\' ]]; then\n'
346 <        txt += '        export LFC_HOME='+self.lfc_home+'\n'
347 <        txt += '    fi\n'
348 <        txt += 'elif [ $middleware == OSG ]; then\n'
349 <        txt += '    echo "LFC catalog setting to be implemented for OSG"\n'
350 <        txt += 'fi\n'
351 <        #####
352 <        if int(self.register_data) == 1:
353 <           txt += 'if [ $middleware == LCG ]; then \n'
354 <           txt += '    export LFN='+self.LFN+'\n'
355 <           txt += '    lfc-ls $LFN\n'
356 <           txt += '    result=$?\n'
357 <           txt += '    echo $result\n'
358 <           ### creation of LFN dir in LFC catalog, under /grid/cms dir  
359 <           txt += '    if [ $result != 0 ]; then\n'
360 <           txt += '       lfc-mkdir $LFN\n'
361 <           txt += '       result=$?\n'
362 <           txt += '       echo $result\n'
363 <           txt += '    fi\n'
364 <           txt += 'elif [ $middleware == OSG ]; then\n'
365 <           txt += '    echo " Files registration to be implemented for OSG"\n'
366 <           txt += 'fi\n'
367 <           txt += '\n'
368 <
369 <           if self.VO:
370 <              txt += 'export VO='+self.VO+'\n'
371 <           if self.LFN:
372 <              txt += 'if [ $middleware == LCG ]; then \n'
373 <              txt += '    export LFN='+self.LFN+'\n'
374 <              txt += 'fi\n'
375 <              txt += '\n'
376 <
377 <        txt += 'if [ $middleware == LCG ]; then\n'
378 <        txt += '    CloseCEs=`edg-brokerinfo getCE`\n'
361 >        txt += 'if [ $middleware == LCG ]; then\n'
362 >        txt += '    CloseCEs=`glite-brokerinfo getCE`\n'
363          txt += '    echo "CloseCEs = $CloseCEs"\n'
364          txt += '    CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
365          txt += '    echo "CE = $CE"\n'
# Line 387 | Line 371 | class SchedulerEdg(Scheduler):
371          txt += '        echo "JOB_EXIT_STATUS = 10099" \n'
372          txt += '        echo "JobExitCode=10099" | tee -a $RUNTIME_AREA/$repo \n'
373          txt += '        dumpStatus $RUNTIME_AREA/$repo \n'
390 –        txt += '        rm -f $RUNTIME_AREA/$repo \n'
391 –        txt += '        echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
392 –        txt += '        echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
374          txt += '        exit 1 \n'
375          txt += '    fi \n'
376 <        txt += 'fi \n'
376 >        txt += 'fi \n'
377  
378          return txt
379  
380      def wsCopyInput(self):
381          """
382 <        Copy input data from SE to WN    
382 >        Copy input data from SE to WN
383          """
384          txt = ''
385          if not self.copy_input_data: return txt
386  
387          ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
388 <        txt += 'if [ $middleware == OSG ]; then\n'
388 >        txt += 'if [ $middleware == OSG ]; then\n'
389          txt += '   #\n'
390          txt += '   #   Copy Input Data from SE to this WN deactivated in OSG mode\n'
391          txt += '   #\n'
# Line 413 | Line 394 | class SchedulerEdg(Scheduler):
394          txt += '   #\n'
395          txt += '   #   Copy Input Data from SE to this WN\n'
396          txt += '   #\n'
397 <        ### changed by georgia (put a loop copying more than one input files per jobs)          
397 >        ### changed by georgia (put a loop copying more than one input files per jobs)
398          txt += '   for input_file in $cur_file_list \n'
399          txt += '   do \n'
400          txt += '      lcg-cp --vo $VO --verbose -t 1200 lfn:$input_lfn/$input_file file:`pwd`/$input_file 2>&1\n'
# Line 440 | Line 421 | class SchedulerEdg(Scheduler):
421          txt += '   \n'
422          txt += '   ### Check SCRATCH space available on WN : \n'
423          txt += '   df -h \n'
424 <        txt += 'fi \n'
425 <          
424 >        txt += 'fi \n'
425 >
426          return txt
427  
428      def wsCopyOutput(self):
# Line 449 | Line 430 | class SchedulerEdg(Scheduler):
430          Write a CopyResults part of a job script, e.g.
431          to copy produced output into a storage element.
432          """
433 <        txt = ''
453 <        if int(self.copy_data) == 1:
454 <           txt += '#\n'
455 <           txt += '#   Copy output to SE = $SE\n'
456 <           txt += '#\n'
457 <           txt += '    if [ $middleware == OSG ]; then\n'
458 <           txt += '        echo "X509_USER_PROXY = $X509_USER_PROXY"\n'
459 <           txt += '        echo "source $OSG_APP/glite/setup_glite_ui.sh"\n'
460 <           txt += '        source $OSG_APP/glite/setup_glite_ui.sh\n'
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 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 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 "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 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 "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 "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 "srmcp succeeded"\n'
513 <           txt += '         fi\n'
514 <           txt += '     done\n'
515 <        return txt
433 >        txt = '\n'
434  
435 <    def wsRegisterOutput(self):
436 <        """
437 <        Returns part of a job script which does scheduler-specific work.
520 <        """
435 >        txt += '#\n'
436 >        txt += '# COPY OUTPUT FILE TO SE\n'
437 >        txt += '#\n\n'
438  
439 <        txt = ''
440 <        if int(self.register_data) == 1:
441 <        ## OLI_Daniele deactivate for OSG (wait for LCG UI installed on OSG)
442 <           txt += 'if [ $middleware == OSG ]; then\n'
443 <           txt += '   #\n'
444 <           txt += '   #   Register output to LFC deactivated in OSG mode\n'
445 <           txt += '   #\n'
446 <           txt += '   echo "Register output to LFC deactivated in OSG mode"\n'
447 <           txt += 'elif [ $middleware == LCG ]; then \n'
448 <           txt += '#\n'
449 <           txt += '#  Register output to LFC\n'
450 <           txt += '#\n'
451 <           txt += '   if [ $copy_exit_status -eq 0 ]; then\n'
452 <           txt += '      for out_file in $file_list ; do\n'
453 <           txt += '         echo "Trying to register the output file into LFC"\n'
454 <           txt += '         echo "lcg-rf -l $LFN/$out_file --vo $VO -t 1200 sfn://$SE$SE_PATH/$out_file 2>&1"\n'
455 <           txt += '         lcg-rf -l $LFN/$out_file --vo $VO -t 1200 sfn://$SE$SE_PATH/$out_file 2>&1 \n'
456 <           txt += '         register_exit_status=$?\n'
457 <           txt += '         echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
458 <           txt += '         echo "STAGE_OUT = $register_exit_status"\n'
459 <           txt += '         if [ $register_exit_status -ne 0 ]; then \n'
460 <           txt += '            echo "Problems with the registration to LFC" \n'
461 <           txt += '            echo "Try with srm protocol" \n'
462 <           txt += '            echo "lcg-rf -l $LFN/$out_file --vo $VO -t 1200 srm://$SE$SE_PATH/$out_file 2>&1"\n'
463 <           txt += '            lcg-rf -l $LFN/$out_file --vo $VO -t 1200 srm://$SE$SE_PATH/$out_file 2>&1 \n'
464 <           txt += '            register_exit_status=$?\n'
465 <           txt += '            echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
466 <           txt += '            echo "STAGE_OUT = $register_exit_status"\n'
467 <           txt += '            if [ $register_exit_status -ne 0 ]; then \n'
468 <           txt += '               echo "Problems with the registration into LFC" \n'
469 <           txt += '            fi \n'
470 <           txt += '         else \n'
471 <           txt += '            echo "output registered to LFC"\n'
472 <           txt += '         fi \n'
473 <           txt += '         echo "StageOutExitStatus = $register_exit_status" | tee -a $RUNTIME_AREA/$repo\n'
474 <           txt += '      done\n'
475 <           txt += '   else \n'
476 <           txt += '      echo "Trying to copy output file to CloseSE"\n'
477 <           txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
478 <           txt += '      for out_file in $file_list ; do\n'
479 <           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'
480 <           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'
481 <           txt += '         register_exit_status=$?\n'
482 <           txt += '         echo "REGISTER_EXIT_STATUS = $register_exit_status"\n'
483 <           txt += '         echo "STAGE_OUT = $register_exit_status"\n'
484 <           txt += '         if [ $register_exit_status -ne 0 ]; then \n'
485 <           txt += '            echo "Problems with CloseSE or Catalog" \n'
486 <           txt += '         else \n'
487 <           txt += '            echo "The program was successfully executed"\n'
571 <           txt += '            echo "SE = $CLOSE_SE"\n'
572 <           txt += '            echo "LFN for the file is LFN=${LFN}/$out_file"\n'
573 <           txt += '         fi \n'
574 <           txt += '         echo "StageOutExitStatus = $register_exit_status" | tee -a $RUNTIME_AREA/$repo\n'
575 <           txt += '      done\n'
576 <           txt += '   fi \n'
577 <           txt += '   exit_status=$register_exit_status\n'
578 <           txt += 'fi \n'
439 >        SE_PATH=''
440 >        if int(self.copy_data) == 1:
441 >            if self.SE:
442 >                txt += 'export SE='+self.SE+'\n'
443 >                txt += 'echo "SE = $SE"\n'
444 >            if self.SE_PATH:
445 >                if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
446 >                SE_PATH=self.SE_PATH
447 >            if int(self.publish_data) == 1:
448 >                txt += '### publish_data = 1 so the SE path where to copy the output is: \n'
449 >                path_add = self.UserGridName + '/' + self.publish_data_name +'_${PSETHASH}/'
450 >                SE_PATH = SE_PATH + path_add
451 >            txt += 'export SE_PATH='+SE_PATH+'\n'
452 >            txt += 'echo "SE_PATH = $SE_PATH"\n'
453 >
454 >            txt += 'echo ">>> Copy output files from WN = `hostname` to SE = $SE :"\n'
455 >
456 >            txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
457 >            txt += '    echo "--> No output file to copy to $SE"\n'
458 >            txt += '    copy_exit_status=$output_exit_status\n'
459 >            txt += '    echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
460 >            txt += 'else\n'
461 >            txt += '    for out_file in $file_list ; do\n'
462 >            txt += '        echo "Trying to copy output file to $SE"\n'
463 >            txt += '        cmscp $SOFTWARE_DIR/$out_file ${SE} ${SE_PATH} $out_file $middleware\n'
464 >            #txt += '        cmscp $out_file ${SE} ${SE_PATH} $out_file $middleware\n'
465 >            txt += '        copy_exit_status=$?\n'
466 >            txt += '        echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
467 >            txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
468 >            txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
469 >            txt += '            echo "Problem copying $out_file to $SE $SE_PATH"\n'
470 >            txt += '            echo "StageOutExitStatus = $copy_exit_status " | tee -a $RUNTIME_AREA/$repo\n'
471 >            txt += '            copy_exit_status=60307\n'
472 >            txt += '        else\n'
473 >            txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
474 >            txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
475 >            txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
476 >            txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
477 >            txt += '        fi\n'
478 >            txt += '    done\n'
479 >            txt += '    if [ $copy_exit_status -ne 0 ]; then\n'
480 >            txt += '        SE=""\n'
481 >            txt += '        echo "SE = $SE"\n'
482 >            txt += '        SE_PATH=""\n'
483 >            txt += '        echo "SE_PATH = $SE_PATH"\n'
484 >            txt += '    fi\n'
485 >            txt += 'fi\n'
486 >            txt += 'exit_status=$copy_exit_status\n'
487 >            pass
488          return txt
489  
490      def loggingInfo(self, id):
# Line 593 | Line 502 | class SchedulerEdg(Scheduler):
502          cmd_out = runCommand(cmd)
503          return cmd_out
504  
596 –    ##### FEDE ######        
505      def findSites_(self, n):
506 <        itr4 =[]
506 >        itr4 =[]
507 >
508          sites = common.jobDB.destination(n)
509 <        if len(sites)>0 and sites[0]=="Any":
509 >
510 >        if len(sites)>0 and sites[0]=="":
511              return itr4
512 +
513          itr = ''
514 <        if sites != [""]:#CarlosDaniele
515 <            for site in sites:
514 >        if sites != [""]:#CarlosDaniele
515 >            ##Addedd Daniele
516 >            replicas = self.blackWhiteListParser.checkBlackList(sites,n)
517 >            if len(replicas)!=0:
518 >                replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
519 >
520 >            if len(replicas)==0:
521 >                itr = itr + 'target.GlueSEUniqueID=="NONE" '
522 >                #msg = 'No sites remaining that host any part of the requested data! Exiting... '
523 >                #raise CrabException(msg)
524 >            #####
525 >           # for site in sites:
526 >            for site in replicas:
527                  #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
528                  itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
529              itr = itr[0:-4]
# Line 609 | Line 531 | class SchedulerEdg(Scheduler):
531          return itr4
532  
533      def createXMLSchScript(self, nj, argsList):
534 <      
534 >
535          """
536          Create a XML-file for BOSS4.
537          """
# Line 622 | Line 544 | class SchedulerEdg(Scheduler):
544          index = nj - 1
545          job = common.job_list[index]
546          jbt = job.type()
625 –        
547          inp_sandbox = jbt.inputSandbox(index)
548 <        out_sandbox = jbt.outputSandbox(index)
548 >        #out_sandbox = jbt.outputSandbox(index)
549          """
550          [end] FIX-ME
551          """
552  
553 <        
553 >
554          title = '<?xml version="1.0" encoding="UTF-8" standalone="no"?>\n'
555          jt_string = ''
556 <        
556 >
557          xml_fname = str(self.jobtypeName)+'.xml'
558          xml = open(common.work_space.shareDir()+'/'+xml_fname, 'a')
559  
560 <        #TaskName  
560 >        #TaskName
561          dir = string.split(common.work_space.topDir(), '/')
562          taskName = dir[len(dir)-2]
563 <  
643 <        to_writeReq = ''
563 >
564          to_write = ''
565  
566          req=' '
# Line 663 | Line 583 | class SchedulerEdg(Scheduler):
583                  else:
584                      req = req +  ' || (RegExp("' + ce_white_list[i] + '", other.GlueCEUniqueId))'
585              req = req + ')'
586 <        
586 >
587          if self.EDG_ce_black_list:
588              ce_black_list = string.split(self.EDG_ce_black_list,',')
589              for ce in ce_black_list:
# Line 683 | Line 603 | class SchedulerEdg(Scheduler):
603                  req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
604              else:
605                  req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
606 <                                                                                          
607 <        if ( self.EDG_retry_count ):              
606 >
607 >        if ( self.EDG_retry_count ):
608              to_write = to_write + 'RetryCount = "'+self.EDG_retry_count+'"\n'
609              pass
610  
611 <        if ( self.EDG_shallow_retry_count ):              
611 >        if ( self.EDG_shallow_retry_count ):
612              to_write = to_write + 'ShallowRetryCount = "'+self.EDG_shallow_retry_count+'"\n'
613              pass
614  
615          to_write = to_write + 'MyProxyServer = "&quot;' + self.proxyServer + '&quot;"\n'
616          to_write = to_write + 'VirtualOrganisation = "&quot;' + self.VO + '&quot;"\n'
617  
618 <        #TaskName  
618 >        #TaskName
619          dir = string.split(common.work_space.topDir(), '/')
620          taskName = dir[len(dir)-2]
621  
622          xml.write(str(title))
623 <        xml.write('<task name="' +str(taskName)+'" sub_path="' + common.work_space.bossCache() + '">\n')
623 >
624 >        #First check the X509_USER_PROXY. In not there use the default
625 >        try:
626 >            x509=os.environ['X509_USER_PROXY']
627 >        except Exception, ex:
628 >            import traceback
629 >            common.logger.debug( 6, str(ex) )
630 >            common.logger.debug( 6, traceback.format_exc() )
631 >            x509_cmd = 'ls /tmp/x509up_u`id -u`'
632 >            x509=runCommand(x509_cmd).strip()
633 >        xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + ' task_info="' + str(x509) + '">\n')
634          xml.write(jt_string)
635 <        
635 >
636          if (to_write != ''):
637              xml.write('<extraTags\n')
638              xml.write(to_write)
# Line 726 | Line 656 | class SchedulerEdg(Scheduler):
656          '''
657          indy: here itr4
658          '''
729 –        
659  
660 <        xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
660 >        xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.schedulerName)+'">\n')
661 >       # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
662          xml.write(jt_string)
663  
664          #executable
# Line 736 | Line 666 | class SchedulerEdg(Scheduler):
666          """
667          INDY
668          script depends on jobType: it should be probably get in a different way
669 <        """        
669 >        """
670          script = job.scriptFilename()
671          xml.write('<program>\n')
672          xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
# Line 752 | Line 682 | class SchedulerEdg(Scheduler):
682                  pass
683              pass
684  
685 <        if (not jbt.additional_inbox_files == []):
686 <            inp_box = inp_box + ','
687 <            for addFile in jbt.additional_inbox_files:
688 <                addFile = os.path.abspath(addFile)
689 <                inp_box = inp_box+''+addFile+','
690 <                pass
685 > #        if (not jbt.additional_inbox_files == []):
686 > #            inp_box = inp_box + ','
687 > #            for addFile in jbt.additional_inbox_files:
688 > #                #addFile = os.path.abspath(addFile)
689 > #                inp_box = inp_box+''+addFile+','
690 > #                pass
691  
692          if inp_box[-1] == ',' : inp_box = inp_box[:-1]
693          inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
694          xml.write(inp_box)
695 <        
695 >
696          base = jbt.name()
697          stdout = base + '__ITR3_.stdout'
698          stderr = base + '__ITR3_.stderr'
699 <        
699 >
700          xml.write('<stderr> ' + stderr + '</stderr>\n')
701          xml.write('<stdout> ' + stdout + '</stdout>\n')
702 <        
702 >
703  
704          out_box = stdout + ',' + \
705                    stderr + ',.BrokerInfo,'
# Line 804 | Line 734 | class SchedulerEdg(Scheduler):
734          if out_box[-1] == ',' : out_box = out_box[:-1]
735          out_box = '<outfiles> <![CDATA[\n' + out_box + '\n]]></outfiles>\n'
736          xml.write(out_box)
737 <
737 >
738          xml.write('<BossAttr> crabjob.INTERNAL_ID=_ITR1_ </BossAttr>\n')
739  
740          xml.write('</program>\n')
# Line 814 | Line 744 | class SchedulerEdg(Scheduler):
744          xml.write('</task>\n')
745  
746          xml.close()
747 <      
747 >
748  
749          return
750  
# Line 825 | Line 755 | class SchedulerEdg(Scheduler):
755          if (self.proxyValid): return
756  
757          ### Just return if asked to do so
758 <        if (self.dontCheckProxy):
758 >        if (self.dontCheckProxy==1):
759              self.proxyValid=1
760              return
761  
832 –        timeleft = -999
762          minTimeLeft=10*3600 # in seconds
763  
764          minTimeLeftServer = 100 # in hours
# Line 874 | Line 803 | class SchedulerEdg(Scheduler):
803              common.logger.message('No credential delegated to myproxy server '+self.proxyServer+' will do now')
804              renewProxy = 1
805          else:
806 <            # if myproxy exist but not long enough, renew
807 <            reTime = re.compile( r'timeleft: (\d+)' )
808 <            #print "<"+str(reTime.search( cmd_out ).group(1))+">"
809 <            if reTime.match( cmd_out ):
810 <                time = reTime.search( line ).group(1)
811 <                if time < minTimeLeftServer:
812 <                    renewProxy = 1
813 <                    common.logger.message('No credential delegation will expire in '+time+' hours: renew it')
806 >            ## minimum time: 5 days
807 >            minTime = 4 * 24 * 3600
808 >            ## regex to extract the right information
809 >            myproxyRE = re.compile("timeleft: (?P<hours>[\\d]*):(?P<minutes>[\\d]*):(?P<seconds>[\\d]*)")
810 >            for row in cmd_out.split("\n"):
811 >                g = myproxyRE.search(row)
812 >                if g:
813 >                    hours = g.group("hours")
814 >                    minutes = g.group("minutes")
815 >                    seconds = g.group("seconds")
816 >                    timeleft = int(hours)*3600 + int(minutes)*60 + int(seconds)
817 >                    if timeleft < minTime:
818 >                        renewProxy = 1
819 >                        common.logger.message('Your proxy will expire in:\n\t'+hours+' hours '+minutes+' minutes '+seconds+' seconds\n')
820 >                        common.logger.message('Need to renew it:')
821 >                    pass
822                  pass
823              pass
824 <        
824 >
825          # if not, create one.
826          if renewProxy:
827              cmd = 'myproxy-init -d -n -s '+self.proxyServer
# Line 901 | Line 838 | class SchedulerEdg(Scheduler):
838          edg_ui_cfg_opt = ' '
839          if self.edg_config:
840              edg_ui_cfg_opt = ' -c ' + self.edg_config + ' '
841 <        if self.edg_config_vo:
841 >        if self.edg_config_vo:
842              edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
843          return edg_ui_cfg_opt
844 +
845 +    def tOut(self, list):
846 +        return 120
847 +
848 +

Diff Legend

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