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.107 by corvo, Mon Dec 4 11:11:35 2006 UTC vs.
Revision 1.133 by fanzago, Wed Aug 15 14:08:18 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 84 | Line 91 | class SchedulerEdg(Scheduler):
91             msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
92             raise CrabException(msg)
93  
94 +        ########### FEDE FOR DBS2 ##############################
95 +        try:
96 +            self.publish_data = cfg_params["USER.publish_data"]
97 +            self.checkProxy()
98 +            if int(self.publish_data) == 1:
99 +                try:
100 +                    self.publish_data_name = cfg_params['USER.publish_data_name']
101 +                except KeyError:
102 +                    msg = "Error. The [USER] section does not have 'publish_data_name'"
103 +                    raise CrabException(msg)
104 +                try:
105 +                    tmp = runCommand("voms-proxy-info -identity")
106 +                    tmp = string.split(tmp,'/')
107 +                    reCN=re.compile(r'CN=')
108 +                    for t in tmp:
109 +                        if reCN.match(t):
110 +                            self.UserGridName=string.strip((t.replace('CN=','')).replace(' ',''))
111 +                        
112 +                    #self.UserGridName = string.strip(runCommand("voms-proxy-info -identity | awk -F\'CN\' \'{print $2$3$4}\' | tr -d \'=/ \'"))
113 +                except:
114 +                    msg = "Error. Problem with voms-proxy-info -identity command"
115 +                    raise CrabException(msg)
116 +        except KeyError: self.publish_data = 0
117 +
118 +        if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
119 +           msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
120 +           msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
121 +           common.logger.message(msg)
122 +           raise CrabException(msg)
123 +        #################################################
124 +
125          try:
126              self.lfc_host = cfg_params['EDG.lfc_host']
127          except KeyError:
# Line 155 | Line 193 | class SchedulerEdg(Scheduler):
193          libPath=os.path.join(path, "lib", "python")
194          sys.path.append(libPath)
195  
158 –        self.proxyValid=0
159 –
196          try:
197              self._taskId = cfg_params['taskId']
198          except:
# Line 168 | Line 204 | class SchedulerEdg(Scheduler):
204          try: self.schedulerName = cfg_params['CRAB.scheduler']
205          except KeyError: self.scheduler = ''
206  
171 –        try: self.dontCheckProxy=cfg_params["EDG.dont_check_proxy"]
172 –        except KeyError: self.dontCheckProxy = 0
173 –
207          return
208      
209  
# Line 184 | Line 217 | class SchedulerEdg(Scheduler):
217          self.edg_config_vo = edgConfig.configVO()
218  
219          if (self.edg_config and self.edg_config_vo != ''):
220 <            self.rb_param_file = 'RBconfig = "'+self.edg_config+'";\nRBconfigVO = "'+self.edg_config_vo+'";'
220 >            self.rb_param_file = 'RBconfig = "'+self.edg_config+'";\nRBconfigVO = "'+self.edg_config_vo+'";\n'
221              #print "rb_param_file = ", self.rb_param_file
222          return self.rb_param_file
223        
# Line 248 | Line 281 | class SchedulerEdg(Scheduler):
281                  req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
282                  
283          for i in range(len(first)): # Add loop DS
284 +            groupReq = req
285              self.param='sched_param_'+str(i)+'.clad'
286              param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')
287  
288              itr4=self.findSites_(first[i])
289              for arg in itr4:
290 <                req = req + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
291 <            param_file.write('Requirements = '+req +';\n')  
290 >                groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
291 >            param_file.write('Requirements = '+groupReq +';\n')  
292    
293              if (self.rb_param_file != ''):
294                  param_file.write(self.rb_param_file)  
# Line 301 | Line 335 | class SchedulerEdg(Scheduler):
335          txt += '    echo "middleware =$middleware" \n'
336          txt += 'elif [ $VO_CMS_SW_DIR ]; then \n'
337          txt += '    middleware=LCG \n'
338 <        txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
338 >   #     txt += '    echo "SyncCE=`edg-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
339 >        txt += '    echo "SyncCE=`glite-brokerinfo getCE`" | tee -a $RUNTIME_AREA/$repo \n'
340          txt += '    echo "GridFlavour=`echo $middleware`" | tee -a $RUNTIME_AREA/$repo \n'
341          txt += '    echo "middleware =$middleware" \n'
342          txt += 'else \n'
# Line 323 | Line 358 | class SchedulerEdg(Scheduler):
358          
359          txt += '\n\n'
360  
361 <        if int(self.copy_data) == 1:
362 <           if self.SE:
363 <              txt += 'export SE='+self.SE+'\n'
364 <              txt += 'echo "SE = $SE"\n'
365 <           if self.SE_PATH:
366 <              if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
367 <              txt += 'export SE_PATH='+self.SE_PATH+'\n'
368 <              txt += 'echo "SE_PATH = $SE_PATH"\n'
361 > #        if int(self.copy_data) == 1:
362 > #           if self.SE:
363 > #              txt += 'export SE='+self.SE+'\n'
364 > #              txt += 'echo "SE = $SE"\n'
365 > #           if self.SE_PATH:
366 > #              if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
367 > #              txt += 'export SE_PATH='+self.SE_PATH+'\n'
368 > #              txt += 'echo "SE_PATH = $SE_PATH"\n'
369  
370          txt += 'export VO='+self.VO+'\n'
371          ### add some line for LFC catalog setting
# Line 374 | Line 409 | class SchedulerEdg(Scheduler):
409                txt += '\n'
410  
411          txt += 'if [ $middleware == LCG ]; then\n'
412 <        txt += '    CloseCEs=`edg-brokerinfo getCE`\n'
412 >    #    txt += '    CloseCEs=`edg-brokerinfo getCE`\n'
413 >        txt += '    CloseCEs=`glite-brokerinfo getCE`\n'
414          txt += '    echo "CloseCEs = $CloseCEs"\n'
415          txt += '    CE=`echo $CloseCEs | sed -e "s/:.*//"`\n'
416          txt += '    echo "CE = $CE"\n'
# Line 449 | Line 485 | class SchedulerEdg(Scheduler):
485          to copy produced output into a storage element.
486          """
487          txt = ''
488 +
489 +        ##### FEDE MOVED FROM SET_ENVIRONMENT ##############
490 +        
491 +        SE_PATH=''
492          if int(self.copy_data) == 1:
493 +           if self.SE:
494 +              txt += 'export SE='+self.SE+'\n'
495 +              txt += 'echo "SE = $SE"\n'
496 +           if self.SE_PATH:
497 +              if ( self.SE_PATH[-1] != '/' ) : self.SE_PATH = self.SE_PATH + '/'
498 +              SE_PATH=self.SE_PATH
499 +              ####### FEDE FOR DBS2
500 +              if int(self.publish_data) == 1:
501 +                  txt += '### publish_data = 1 so the SE path where to copy the output is: \n'
502 +                  #txt += 'subject=`voms-proxy-info -subject | awk -F\'CN\' \'{print $2$3$4}\' | tr -d \'=/ \'` \n'
503 +                  #txt += 'echo "subject = $subject" \n'
504 +                  #path_add = '${subject}/'+ self.publish_data_name +'_${PSETHASH}/'
505 +                  path_add = self.UserGridName + '/' + self.publish_data_name +'_${PSETHASH}/'
506 +                  SE_PATH = SE_PATH + path_add
507 +
508 +              txt += 'export SE_PATH='+SE_PATH+'\n'
509 +              txt += 'echo "SE_PATH = $SE_PATH"\n'
510 +
511 +        ##########################################################  
512 +
513 +        #if int(self.copy_data) == 1:
514             txt += '#\n'
515             txt += '#   Copy output to SE = $SE\n'
516             txt += '#\n'
# Line 463 | Line 524 | class SchedulerEdg(Scheduler):
524  
525             txt += '    for out_file in $file_list ; do\n'
526             txt += '        echo "Trying to copy output file to $SE using srmcp"\n'
527 <           txt += '        echo "mkdir -p $HOME/.srmconfig"\n'
528 <           txt += '        mkdir -p $HOME/.srmconfig\n'
527 >           # txt += '        echo "mkdir -p $HOME/.srmconfig"\n'
528 >           # txt += '        mkdir -p $HOME/.srmconfig\n'
529             txt += '        if [ $middleware == LCG ]; then\n'
530 <           txt += '           echo "srmcp -retry_num 3 -retry_timeout 480000 file:////`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
531 <           txt += '           exitstring=`srmcp -retry_num 3 -retry_timeout 480000 file:////\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
530 >           txt += '           echo "srmcp -retry_num 3 -retry_timeout 480000 file:///`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
531 >           txt += '           exitstring=`srmcp -retry_num 3 -retry_timeout 480000 file:///\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
532             txt += '        elif [ $middleware == OSG ]; then\n'
533 <           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'
534 <           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'
533 >           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'
534 >           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'
535             txt += '        fi \n'
536             txt += '        copy_exit_status=$?\n'
537             txt += '        echo "COPY_EXIT_STATUS for srmcp = $copy_exit_status"\n'
# Line 484 | Line 545 | class SchedulerEdg(Scheduler):
545             if common.logger.debugLevel() >= 5:
546                 txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
547                 txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
548 +               #txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
549 +               #txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
550             else:              
551                 txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
552                 txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
553 +               #txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
554 +               #txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
555             txt += '            copy_exit_status=$?\n'
556             txt += '            echo "COPY_EXIT_STATUS for lcg-cp = $copy_exit_status"\n'
557             txt += '            echo "STAGE_OUT = $copy_exit_status"\n'
# Line 496 | Line 561 | class SchedulerEdg(Scheduler):
561             txt += '               echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
562             txt += '               echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
563             txt += '               echo "srmcp and lcg-cp and failed!"\n'
564 +           ################### FEDE moved out of for ##############
565 +           #txt += '               SE=""\n'
566 +           #txt += '               echo "SE = $SE"\n'
567 +           #txt += '               SE_PATH=""\n'
568 +           #txt += '               echo "SE_PATH = $SE_PATH"\n'
569 +           ########################################################
570             txt += '            else\n'
571             txt += '               echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
572             txt += '               echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
# Line 511 | Line 582 | class SchedulerEdg(Scheduler):
582             txt += '            echo "srmcp succeeded"\n'
583             txt += '         fi\n'
584             txt += '     done\n'
585 +           txt += '     if [ $copy_exit_status -ne 0 ]; then\n'
586 +           txt += '           SE=""\n'
587 +           txt += '           echo "SE = $SE"\n'
588 +           txt += '           SE_PATH=""\n'
589 +           txt += '           echo "SE_PATH = $SE_PATH"\n'
590 +           txt += '     fi\n'
591 +           txt += '     exit_status=$copy_exit_status\n'
592          return txt
593  
594      def wsRegisterOutput(self):
# Line 556 | Line 634 | class SchedulerEdg(Scheduler):
634             txt += '      done\n'
635             txt += '   else \n'
636             txt += '      echo "Trying to copy output file to CloseSE"\n'
637 <           txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
637 > #          txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
638 >           txt += '      CLOSE_SE=`glite-brokerinfo getCloseSEs | head -1`\n'
639             txt += '      for out_file in $file_list ; do\n'
640             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'
641             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'
# Line 592 | Line 671 | class SchedulerEdg(Scheduler):
671          cmd_out = runCommand(cmd)
672          return cmd_out
673  
595 –    ##### FEDE ######        
674      def findSites_(self, n):
675          itr4 =[]
676 +
677          sites = common.jobDB.destination(n)
678 <        if len(sites)>0 and sites[0]=="Any":
678 >
679 >        if len(sites)>0 and sites[0]=="":
680              return itr4
681 +
682          itr = ''
683          if sites != [""]:#CarlosDaniele
684 <            for site in sites:
684 >            ##Addedd Daniele
685 >            replicas = self.blackWhiteListParser.checkBlackList(sites,n)
686 >            if len(replicas)!=0:
687 >                replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
688 >              
689 >            if len(replicas)==0:
690 >                msg = 'No sites remaining that host any part of the requested data! Exiting... '
691 >                raise CrabException(msg)
692 >            #####        
693 >           # for site in sites:
694 >            for site in replicas:
695                  #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
696                  itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
697              itr = itr[0:-4]
# Line 623 | Line 714 | class SchedulerEdg(Scheduler):
714          jbt = job.type()
715          
716          inp_sandbox = jbt.inputSandbox(index)
717 <        out_sandbox = jbt.outputSandbox(index)
717 >        #out_sandbox = jbt.outputSandbox(index)
718          """
719          [end] FIX-ME
720          """
# Line 639 | Line 730 | class SchedulerEdg(Scheduler):
730          dir = string.split(common.work_space.topDir(), '/')
731          taskName = dir[len(dir)-2]
732    
642 –        to_writeReq = ''
733          to_write = ''
734  
735          req=' '
# Line 699 | Line 789 | class SchedulerEdg(Scheduler):
789          taskName = dir[len(dir)-2]
790  
791          xml.write(str(title))
792 <        xml.write('<task name="' +str(taskName)+'" sub_path="' + common.work_space.bossCache() + '">\n')
792 >        #xml.write('<task name="' +str(taskName)+'" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache">\n')
793 >
794 >        #xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + '" task_info="' + os.path.expandvars('X509_USER_PROXY') + '">\n')
795 >        xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + ' task_info="' + os.environ["X509_USER_PROXY"] + '">\n')
796          xml.write(jt_string)
797          
798          if (to_write != ''):
# Line 726 | Line 819 | class SchedulerEdg(Scheduler):
819          indy: here itr4
820          '''
821          
822 <
823 <        xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
822 >        xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.schedulerName)+'">\n')
823 >       # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
824          xml.write(jt_string)
825  
826          #executable
# Line 751 | Line 844 | class SchedulerEdg(Scheduler):
844                  pass
845              pass
846  
847 <        if (not jbt.additional_inbox_files == []):
848 <            inp_box = inp_box + ','
849 <            for addFile in jbt.additional_inbox_files:
850 <                addFile = os.path.abspath(addFile)
851 <                inp_box = inp_box+''+addFile+','
852 <                pass
847 > #        if (not jbt.additional_inbox_files == []):
848 > #            inp_box = inp_box + ','
849 > #            for addFile in jbt.additional_inbox_files:
850 > #                #addFile = os.path.abspath(addFile)
851 > #                inp_box = inp_box+''+addFile+','
852 > #                pass
853  
854          if inp_box[-1] == ',' : inp_box = inp_box[:-1]
855          inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
# Line 824 | Line 917 | class SchedulerEdg(Scheduler):
917          if (self.proxyValid): return
918  
919          ### Just return if asked to do so
920 <        if (self.dontCheckProxy):
920 >        if (self.dontCheckProxy==1):
921              self.proxyValid=1
922              return
923  
831 –        timeleft = -999
924          minTimeLeft=10*3600 # in seconds
925  
926          minTimeLeftServer = 100 # in hours
# Line 877 | Line 969 | class SchedulerEdg(Scheduler):
969              reTime = re.compile( r'timeleft: (\d+)' )
970              #print "<"+str(reTime.search( cmd_out ).group(1))+">"
971              if reTime.match( cmd_out ):
972 <                time = reTime.search( line ).group(1)
972 >                time = reTime.search( cmd_out ).group(1)
973                  if time < minTimeLeftServer:
974                      renewProxy = 1
975                      common.logger.message('No credential delegation will expire in '+time+' hours: renew it')
# Line 903 | Line 995 | class SchedulerEdg(Scheduler):
995          if self.edg_config_vo:
996              edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
997          return edg_ui_cfg_opt
998 +
999 +    def submitTout(self, list):
1000 +        return 120
1001 +
1002 +

Diff Legend

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