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.131 by corvo, Wed Aug 15 13:22:52 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 +            if int(self.publish_data) == 1:
98 +                try:
99 +                    self.publish_data_name = cfg_params['USER.publish_data_name']
100 +                except KeyError:
101 +                    msg = "Error. The [USER] section does not have 'publish_data_name'"
102 +                    raise CrabException(msg)
103 +                try:
104 +                    tmp = runCommand("voms-proxy-info -identity")
105 +                    tmp = string.split(tmp,'/')
106 +                    reCN=re.compile(r'CN=')
107 +                    for t in tmp:
108 +                        if reCN.match(t):
109 +                            self.UserGridName=string.strip((t.replace('CN=','')).replace(' ',''))
110 +                        
111 +                    #self.UserGridName = string.strip(runCommand("voms-proxy-info -identity | awk -F\'CN\' \'{print $2$3$4}\' | tr -d \'=/ \'"))
112 +                except:
113 +                    self.checkProxy()
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 302 | 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 324 | 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 375 | 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 450 | 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 464 | 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 483 | Line 543 | class SchedulerEdg(Scheduler):
543             txt += '            echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
544             txt += '            echo "srmcp failed, attempting lcg-cp."\n'
545             if common.logger.debugLevel() >= 5:
546 +               ########### FEDE CHANGES TO WRITE IN SRM LNL.INFN.IT #################
547                 txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
548                 txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
549 +               #txt += '            echo "lcg-cp --vo $VO -t 2400 --verbose file://`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
550 +               #txt += '            exitstring=`lcg-cp --vo $VO -t 2400 --verbose file://\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
551             else:              
552                 txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file gsiftp://${SE}${SE_PATH}$out_file"\n'
553                 txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file gsiftp://${SE}${SE_PATH}$out_file 2>&1`\n'
554 +               #txt += '            echo "lcg-cp --vo $VO -t 2400 file://`pwd`/$out_file srm://${SE}:8443${SE_PATH}$out_file"\n'
555 +               #txt += '            exitstring=`lcg-cp --vo $VO -t 2400 file://\`pwd\`/$out_file srm://${SE}:8443${SE_PATH}$out_file 2>&1`\n'
556             txt += '            copy_exit_status=$?\n'
557             txt += '            echo "COPY_EXIT_STATUS for lcg-cp = $copy_exit_status"\n'
558             txt += '            echo "STAGE_OUT = $copy_exit_status"\n'
# Line 497 | Line 562 | class SchedulerEdg(Scheduler):
562             txt += '               echo "StageOutExitStatus = 198" | tee -a $RUNTIME_AREA/$repo\n'
563             txt += '               echo "StageOutExitStatusReason = $exitstring" | tee -a $RUNTIME_AREA/$repo\n'
564             txt += '               echo "srmcp and lcg-cp and failed!"\n'
565 +           txt += '               SE=""\n'
566 +           txt += '               echo "SE = $SE"\n'
567 +           txt += '               SE_PATH=""\n'
568 +           txt += '               echo "SE_PATH = $SE_PATH"\n'
569             txt += '            else\n'
570             txt += '               echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
571             txt += '               echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
# Line 512 | Line 581 | class SchedulerEdg(Scheduler):
581             txt += '            echo "srmcp succeeded"\n'
582             txt += '         fi\n'
583             txt += '     done\n'
584 +           txt += '     exit_status=$copy_exit_status\n'
585          return txt
586  
587      def wsRegisterOutput(self):
# Line 557 | Line 627 | class SchedulerEdg(Scheduler):
627             txt += '      done\n'
628             txt += '   else \n'
629             txt += '      echo "Trying to copy output file to CloseSE"\n'
630 <           txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
630 > #          txt += '      CLOSE_SE=`edg-brokerinfo getCloseSEs | head -1`\n'
631 >           txt += '      CLOSE_SE=`glite-brokerinfo getCloseSEs | head -1`\n'
632             txt += '      for out_file in $file_list ; do\n'
633             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'
634             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 593 | Line 664 | class SchedulerEdg(Scheduler):
664          cmd_out = runCommand(cmd)
665          return cmd_out
666  
596 –    ##### FEDE ######        
667      def findSites_(self, n):
668          itr4 =[]
669 +
670          sites = common.jobDB.destination(n)
671 <        if len(sites)>0 and sites[0]=="Any":
671 >
672 >        if len(sites)>0 and sites[0]=="":
673              return itr4
674 +
675          itr = ''
676          if sites != [""]:#CarlosDaniele
677 <            for site in sites:
677 >            ##Addedd Daniele
678 >            replicas = self.blackWhiteListParser.checkBlackList(sites,n)
679 >            if len(replicas)!=0:
680 >                replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
681 >              
682 >            if len(replicas)==0:
683 >                msg = 'No sites remaining that host any part of the requested data! Exiting... '
684 >                raise CrabException(msg)
685 >            #####        
686 >           # for site in sites:
687 >            for site in replicas:
688                  #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
689                  itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
690              itr = itr[0:-4]
# Line 624 | Line 707 | class SchedulerEdg(Scheduler):
707          jbt = job.type()
708          
709          inp_sandbox = jbt.inputSandbox(index)
710 <        out_sandbox = jbt.outputSandbox(index)
710 >        #out_sandbox = jbt.outputSandbox(index)
711          """
712          [end] FIX-ME
713          """
# Line 640 | Line 723 | class SchedulerEdg(Scheduler):
723          dir = string.split(common.work_space.topDir(), '/')
724          taskName = dir[len(dir)-2]
725    
643 –        to_writeReq = ''
726          to_write = ''
727  
728          req=' '
# Line 700 | Line 782 | class SchedulerEdg(Scheduler):
782          taskName = dir[len(dir)-2]
783  
784          xml.write(str(title))
785 <        xml.write('<task name="' +str(taskName)+'" sub_path="' + common.work_space.bossCache() + '">\n')
785 >        #xml.write('<task name="' +str(taskName)+'" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache">\n')
786 >
787 >        #xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + '" task_info="' + os.path.expandvars('X509_USER_PROXY') + '">\n')
788 >        xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + ' task_info="' + os.environ["X509_USER_PROXY"] + '">\n')
789          xml.write(jt_string)
790          
791          if (to_write != ''):
# Line 727 | Line 812 | class SchedulerEdg(Scheduler):
812          indy: here itr4
813          '''
814          
815 <
816 <        xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
815 >        xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.schedulerName)+'">\n')
816 >       # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
817          xml.write(jt_string)
818  
819          #executable
# Line 752 | Line 837 | class SchedulerEdg(Scheduler):
837                  pass
838              pass
839  
840 <        if (not jbt.additional_inbox_files == []):
841 <            inp_box = inp_box + ','
842 <            for addFile in jbt.additional_inbox_files:
843 <                addFile = os.path.abspath(addFile)
844 <                inp_box = inp_box+''+addFile+','
845 <                pass
840 > #        if (not jbt.additional_inbox_files == []):
841 > #            inp_box = inp_box + ','
842 > #            for addFile in jbt.additional_inbox_files:
843 > #                #addFile = os.path.abspath(addFile)
844 > #                inp_box = inp_box+''+addFile+','
845 > #                pass
846  
847          if inp_box[-1] == ',' : inp_box = inp_box[:-1]
848          inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
# Line 825 | Line 910 | class SchedulerEdg(Scheduler):
910          if (self.proxyValid): return
911  
912          ### Just return if asked to do so
913 <        if (self.dontCheckProxy):
913 >        if (self.dontCheckProxy==1):
914              self.proxyValid=1
915              return
916  
832 –        timeleft = -999
917          minTimeLeft=10*3600 # in seconds
918  
919          minTimeLeftServer = 100 # in hours
# Line 878 | Line 962 | class SchedulerEdg(Scheduler):
962              reTime = re.compile( r'timeleft: (\d+)' )
963              #print "<"+str(reTime.search( cmd_out ).group(1))+">"
964              if reTime.match( cmd_out ):
965 <                time = reTime.search( line ).group(1)
965 >                time = reTime.search( cmd_out ).group(1)
966                  if time < minTimeLeftServer:
967                      renewProxy = 1
968                      common.logger.message('No credential delegation will expire in '+time+' hours: renew it')
# Line 904 | Line 988 | class SchedulerEdg(Scheduler):
988          if self.edg_config_vo:
989              edg_ui_cfg_opt += ' --config-vo ' + self.edg_config_vo + ' '
990          return edg_ui_cfg_opt
991 +
992 +    def submitTout(self, list):
993 +        return 120
994 +
995 +

Diff Legend

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