| 5 |
|
from SchedulerGrid import SchedulerGrid |
| 6 |
|
from crab_exceptions import CrabException |
| 7 |
|
from crab_util import runCommand |
| 8 |
+ |
from crab_util import gethnUserNameFromSiteDB |
| 9 |
|
from ServerConfig import * |
| 9 |
– |
from WMCore.SiteScreening.BlackWhiteListParser import SEBlackWhiteListParser |
| 10 |
|
import Scram |
| 11 |
|
|
| 12 |
|
import common |
| 39 |
|
self.OSBsize = 50*1000*1000 # 50 MB |
| 40 |
|
|
| 41 |
|
self.environment_unique_identifier = None |
| 42 |
< |
|
| 42 |
> |
self.submissionDay = time.strftime("%y%m%d",time.localtime()) |
| 43 |
> |
|
| 44 |
|
return |
| 45 |
|
|
| 46 |
|
|
| 48 |
|
""" |
| 49 |
|
Configure the scheduler with the config settings from the user |
| 50 |
|
""" |
| 51 |
< |
|
| 51 |
> |
|
| 52 |
> |
# this line needs to be before the call to SchedulerGrid.configure |
| 53 |
> |
# because that calls SchedulerRemoteglidin in turn and |
| 54 |
> |
# sshControlPersist needs to be defined then :-( |
| 55 |
> |
self.sshControlPersist = cfg_params.get('USER.ssh_control_persist','3600') |
| 56 |
> |
if self.sshControlPersist.lower() == "no" or \ |
| 57 |
> |
self.sshControlPersist.lower() == "yes" or \ |
| 58 |
> |
self.sshControlPersist.isdigit() : |
| 59 |
> |
pass |
| 60 |
> |
else: |
| 61 |
> |
msg = "Error: invalid value '%s' for USER.ssh_control_persist " % \ |
| 62 |
> |
self.sshControlPersist |
| 63 |
> |
raise CrabException(msg) |
| 64 |
> |
|
| 65 |
|
SchedulerGrid.configure(self, cfg_params) |
| 66 |
|
|
| 67 |
|
self.proxyValid=0 |
| 68 |
< |
self.dontCheckProxy=int(cfg_params.get("GRID.dont_check_proxy",0)) |
| 68 |
> |
self.dontCheckProxy=int(cfg_params.get("GRID.dont_check_proxy",'0')) |
| 69 |
|
self.space_token = cfg_params.get("USER.space_token",None) |
| 70 |
|
self.proxyServer= 'myproxy.cern.ch' |
| 71 |
|
self.group = cfg_params.get("GRID.group", None) |
| 72 |
|
self.role = cfg_params.get("GRID.role", None) |
| 73 |
|
self.VO = cfg_params.get('GRID.virtual_organization','cms') |
| 74 |
+ |
self.allowOverflow = cfg_params.get('GRID.allow_overflow', '1') |
| 75 |
+ |
self.max_rss = cfg_params.get('GRID.max_rss','2000') |
| 76 |
|
|
| 77 |
|
self.checkProxy() |
| 78 |
|
|
| 96 |
|
msg+="\n Use GRID.se_white_list and/or GRID.se_black_list instead" |
| 97 |
|
raise CrabException(msg) |
| 98 |
|
|
| 99 |
+ |
|
| 100 |
+ |
# make sure proxy FQAN has not changed since last time |
| 101 |
+ |
command = "voms-proxy-info -identity -fqan 2>/dev/null" |
| 102 |
+ |
command += " | head -2" |
| 103 |
+ |
identity = runCommand(command) |
| 104 |
+ |
idfile = common.work_space.shareDir() + "GridIdentity" |
| 105 |
+ |
if os.access(idfile, os.F_OK) : |
| 106 |
+ |
# identity file exists from previous commands |
| 107 |
+ |
f=open(idfile, 'r') |
| 108 |
+ |
idFromFile=f.read() |
| 109 |
+ |
f.close() |
| 110 |
+ |
else : |
| 111 |
+ |
# create it |
| 112 |
+ |
f=open(idfile, 'w') |
| 113 |
+ |
f.write(identity) |
| 114 |
+ |
f.close() |
| 115 |
+ |
idFromFile = identity |
| 116 |
+ |
|
| 117 |
+ |
if identity != idFromFile: |
| 118 |
+ |
msg = "Wrong Grid Credentials:\n%s" % identity |
| 119 |
+ |
msg += "\nMake sure you have " |
| 120 |
+ |
msg += " DN, FQAN =\n%s" % idFromFile |
| 121 |
+ |
raise CrabException(msg) |
| 122 |
+ |
|
| 123 |
|
return |
| 84 |
– |
|
| 85 |
– |
def userName(self): |
| 86 |
– |
""" return the user name """ |
| 87 |
– |
tmp=runCommand("voms-proxy-info -identity 2>/dev/null") |
| 88 |
– |
return tmp.strip() |
| 124 |
|
|
| 125 |
+ |
# |
| 126 |
|
def envUniqueID(self): |
| 127 |
|
taskHash = sha1(common._db.queryTask('name')).hexdigest() |
| 128 |
|
id = "https://" + socket.gethostname() + '/' + taskHash + "/${NJob}" |
| 134 |
|
by $CRABPYTHON/Scheduler.py |
| 135 |
|
""" |
| 136 |
|
|
| 137 |
< |
#SB paste from crab ScheduerGlidein |
| 137 |
> |
#SB paste from crab SchedulerGlidein |
| 138 |
|
|
| 139 |
|
jobParams = "" |
| 140 |
|
|
| 146 |
|
seDest = self.blackWhiteListParser.expandList("T") # all of SiteDB |
| 147 |
|
|
| 148 |
|
seString=self.blackWhiteListParser.cleanForBlackWhiteList(seDest) |
| 149 |
< |
|
| 149 |
> |
# beware SiteDB V2 API, explicely cast to string in case it is unicode |
| 150 |
> |
seString=str(seString) |
| 151 |
> |
|
| 152 |
|
jobParams += '+DESIRED_SEs = "'+seString+'"; ' |
| 153 |
|
|
| 154 |
|
scram = Scram.Scram(None) |
| 158 |
|
cmsver=re.split('_', cmsVersion) |
| 159 |
|
numericCmsVersion = "%s%.2d%.2d" %(cmsver[1], int(cmsver[2]), int(cmsver[3])) |
| 160 |
|
|
| 161 |
+ |
# protect against datasetPath being None |
| 162 |
+ |
jobParams += '+DESIRED_CMSDataset ="' + str(self.datasetPath) + '";' |
| 163 |
+ |
|
| 164 |
|
jobParams += '+DESIRED_CMSVersion ="' +cmsVersion+'";' |
| 165 |
|
jobParams += '+DESIRED_CMSVersionNr ="' +numericCmsVersion+'";' |
| 166 |
|
jobParams += '+DESIRED_CMSScramArch ="' +scramArch+'";' |
| 167 |
+ |
|
| 168 |
+ |
userName = gethnUserNameFromSiteDB() |
| 169 |
+ |
jobParams += '+AccountingGroup ="' + userName+'";' |
| 170 |
|
|
| 171 |
|
myscheddName = self.remoteHost |
| 172 |
< |
jobParams += '+Glidein_MonitorID = "https://'+ myscheddName + '//$(Cluster).$(Process)"; ' |
| 172 |
> |
|
| 173 |
> |
jobParams += '+Glidein_MonitorID = "https://'+ myscheddName + \ |
| 174 |
> |
'//' + self.submissionDay + '//$(Cluster).$(Process)"; ' |
| 175 |
|
|
| 176 |
|
if (self.EDG_clock_time): |
| 177 |
< |
jobParams += '+MaxWallTimeMins = '+self.EDG_clock_time+'; ' |
| 177 |
> |
glideinTime = "%d" % (int(self.EDG_clock_time)+20) # 20 min to wrapup |
| 178 |
> |
jobParams += '+MaxWallTimeMins = '+ glideinTime + '; ' |
| 179 |
|
else: |
| 180 |
< |
jobParams += '+MaxWallTimeMins = %d; ' % (60*24) |
| 180 |
> |
jobParams += '+MaxWallTimeMins = %d; ' % (21*60+55) # 21:55h (unit = min) |
| 181 |
|
|
| 135 |
– |
common._db.updateTask_({'jobType':jobParams}) |
| 182 |
|
|
| 183 |
+ |
if self.max_rss : |
| 184 |
+ |
jobParams += 'request_memory = '+self.max_rss+';' |
| 185 |
+ |
|
| 186 |
+ |
if self.allowOverflow == "0": |
| 187 |
+ |
jobParams += '+CMS_ALLOW_OVERFLOW = False; ' |
| 188 |
+ |
|
| 189 |
+ |
if self.EDG_addJdlParam: |
| 190 |
+ |
if self.EDG_addJdlParam[-1] == '': |
| 191 |
+ |
self.EDG_addJdlParam = self.EDG_addJdlParam[:-1] |
| 192 |
+ |
for p in self.EDG_addJdlParam: |
| 193 |
+ |
jobParams += p.strip()+';\n' |
| 194 |
+ |
|
| 195 |
+ |
common._db.updateTask_({'jobType':jobParams}) |
| 196 |
|
|
| 197 |
|
return jobParams |
| 198 |
|
|
| 202 |
|
Return dictionary with specific parameters, to use with real scheduler |
| 203 |
|
is called when scheduler is initialized in Boss, i.e. at each crab command |
| 204 |
|
""" |
| 205 |
< |
#SB this method is used to pass directory names to Boss Scheduler |
| 205 |
> |
#SB this method is used to pass informatinos to Boss Scheduler |
| 206 |
|
# via params dictionary |
| 207 |
|
|
| 208 |
|
jobDir = common.work_space.jobDir() |
| 209 |
|
taskDir=common.work_space.topDir().split('/')[-2] |
| 210 |
|
shareDir = common.work_space.shareDir() |
| 152 |
– |
submissionDay = time.strftime("%y%m%d",time.localtime()) |
| 211 |
|
|
| 212 |
|
params = {'shareDir':shareDir, |
| 213 |
|
'jobDir':jobDir, |
| 214 |
|
'taskDir':taskDir, |
| 215 |
< |
'submissionDay':submissionDay} |
| 215 |
> |
'submissionDay':self.submissionDay, |
| 216 |
> |
'sshControlPersist':self.sshControlPersist} |
| 217 |
|
|
| 218 |
|
return params |
| 219 |
|
|
| 258 |
|
|
| 259 |
|
txt += 'func_exit() { \n' |
| 260 |
|
txt += self.wsExitFunc_common() |
| 202 |
– |
|
| 203 |
– |
txt += ' tar zcvf ${out_files}.tgz ${final_list}\n' |
| 204 |
– |
txt += ' tmp_size=`ls -gGrta ${out_files}.tgz | awk \'{ print $3 }\'`\n' |
| 205 |
– |
txt += ' rm ${out_files}.tgz\n' |
| 206 |
– |
txt += ' size=`expr $tmp_size`\n' |
| 207 |
– |
txt += ' echo "Total Output dimension: $size"\n' |
| 208 |
– |
txt += ' limit='+str(self.OSBsize) +' \n' |
| 209 |
– |
txt += ' echo "WARNING: output files size limit is set to: $limit"\n' |
| 210 |
– |
txt += ' if [ "$limit" -lt "$size" ]; then\n' |
| 211 |
– |
txt += ' exceed=1\n' |
| 212 |
– |
txt += ' job_exit_code=70000\n' |
| 213 |
– |
txt += ' echo "Output Sanbox too big. Produced output is lost "\n' |
| 214 |
– |
txt += ' else\n' |
| 215 |
– |
txt += ' exceed=0\n' |
| 216 |
– |
txt += ' echo "Total Output dimension $size is fine."\n' |
| 217 |
– |
txt += ' fi\n' |
| 218 |
– |
|
| 219 |
– |
txt += ' echo "JOB_EXIT_STATUS = $job_exit_code"\n' |
| 220 |
– |
txt += ' echo "JobExitCode=$job_exit_code" >> $RUNTIME_AREA/$repo\n' |
| 221 |
– |
txt += ' dumpStatus $RUNTIME_AREA/$repo\n' |
| 222 |
– |
txt += ' if [ $exceed -ne 1 ]; then\n' |
| 223 |
– |
txt += ' tar zcvf ${out_files}.tgz ${final_list}\n' |
| 224 |
– |
txt += ' else\n' |
| 225 |
– |
txt += ' tar zcvf ${out_files}.tgz CMSSW_${NJob}.stdout CMSSW_${NJob}.stderr\n' |
| 226 |
– |
txt += ' fi\n' |
| 227 |
– |
txt += ' python $RUNTIME_AREA/fillCrabFjr.py $RUNTIME_AREA/crab_fjr_$NJob.xml --errorcode $job_exit_code \n' |
| 228 |
– |
|
| 261 |
|
txt += ' exit $job_exit_code\n' |
| 262 |
|
txt += '}\n' |
| 263 |
|
|
| 285 |
|
remoteUserHost=str(task['serverName']) |
| 286 |
|
common.logger.info("serverName from Task DB is %s" % |
| 287 |
|
remoteUserHost) |
| 256 |
– |
if '@' in remoteUserHost: |
| 257 |
– |
remoteHost = remoteUserHost.split('@')[1] |
| 258 |
– |
else: |
| 259 |
– |
remoteHost = remoteUserHost |
| 288 |
|
else: |
| 289 |
|
if self.cfg_params.has_key('CRAB.submit_host'): |
| 290 |
|
# get a remote submission host from crab config file |
| 291 |
|
srvCfg=ServerConfig(self.cfg_params['CRAB.submit_host']).config() |
| 292 |
< |
remoteHost=srvCfg['serverName'] |
| 293 |
< |
common.logger.info("remotehost from crab.cfg = %s" % remoteHost) |
| 292 |
> |
remoteUserHost=srvCfg['serverName'] |
| 293 |
> |
common.logger.info("remotehost from crab.cfg = %s" % remoteUserHost) |
| 294 |
|
else: |
| 295 |
|
# pick from Available Servers List |
| 296 |
|
srvCfg=ServerConfig('default').config() |
| 297 |
< |
remoteHost = srvCfg['serverName'] |
| 298 |
< |
common.logger.info("remotehost from Avail.List = %s" % remoteHost) |
| 297 |
> |
remoteUserHost = srvCfg['serverName'] |
| 298 |
> |
common.logger.info("remotehost from Avail.List = %s" % remoteUserHost) |
| 299 |
|
|
| 300 |
< |
if not remoteHost: |
| 300 |
> |
if not remoteUserHost: |
| 301 |
|
raise CrabException('FATAL ERROR: remoteHost not defined') |
| 274 |
– |
|
| 275 |
– |
#common.logger.info("try to find out username for remote Host via uberftp ...") |
| 276 |
– |
#command="uberftp %s pwd|grep User|awk '{print $3}'" % remoteHost |
| 277 |
– |
#(status, output) = commands.getstatusoutput(command) |
| 278 |
– |
#if status == 0: |
| 279 |
– |
# remoteUser = output |
| 280 |
– |
# common.logger.info("remoteUser set to %s" % remoteUser) |
| 281 |
– |
# if remoteUser==None: |
| 282 |
– |
# raise CrabException('FATAL ERROR: REMOTE USER not defined') |
| 302 |
|
|
| 303 |
< |
#remoteUserHost = remoteUser + '@' + remoteHost |
| 304 |
< |
remoteUserHost = remoteHost |
| 303 |
> |
if '@' in remoteUserHost: |
| 304 |
> |
remoteHost = remoteUserHost.split('@')[1] |
| 305 |
> |
else: |
| 306 |
> |
remoteHost = remoteUserHost |
| 307 |
|
|
| 308 |
|
common._db.updateTask_({'serverName':remoteUserHost}) |
| 309 |
|
|
