#!/usr/bin/python # # dmsub: Submit a data movement job using a variety of methods. # Copyright 2009,2014 University of Tennessee # # License: GNU GPL v2; see ../COPYING for details. # Revision info: # $HeadURL$ # $Revision$ # $Date$ import getopt import datetime import os import shutil import subprocess import sys import uuid credcache = True debug = False hold = False blksize = 1048576 streams = 1 stripesize = 268787456 class TransferList(object): def __init__(self,destlist=None,options=None): if ( destlist is not None ): self._destlist = destlist else: self._destlist = None self._x509proxy = "default" self._altproto = None self._binary = True self._blksize = blksize self._tcpbuf = blksize self._allow3rdparty = True self._streams = streams self._stripesize = stripesize self._dcau = True self._nsimfiles = 1 self._srcsubj = None self._destsubj = None self._cleanup = False self._retries = 0 if ( options is not None ): self.setOptions(options) def setOptions(self,options): if ( isdict(options) ): for option in options.keys(): self.setOption(option,options[option]) def setOption(self,keyword,value): if ( keyword=="x509proxy" ): self._x509proxy = value elif ( keyword=="alt_protocols" ): self._altproto = value elif ( keyword=="binary" ): if ( value=="false" or value=="False" ): value = False else: value = True self._binary = value elif ( keyword=="block_size" ): self._blksize = value elif ( keyword=="stripe_size" ): self._stripsize = value elif ( keyword=="tcp_buffer" ): self._tcpbuf = value elif ( keyword=="no_tpt" ): if ( value=="false" or value=="False" ): value = False else: value = True self._allow3rdparty = not value elif ( keyword=="dcau" ): if ( value=="false" or value=="False" ): value = False else: value = True self._dcau = value elif ( keyword=="streams" ): self._nstreams = value elif ( keyword=="concurrency" ): self._nsimfiles = value elif ( keyword=="src_subject" ): self._srcsubj = value elif ( keyword=="dest_subject" ): self._destsubj = value elif ( keyword=="cleanup" ): if ( value=="false" or value=="False" ): value = False else: value = True self._cleanup = value elif ( keyword=="retries" ): self._retries = value def getOption(self,keyword): if ( keyword=="x509proxy" ): return self._x509proxy elif ( keyword=="alt_protocols" ): return self._altproto elif ( keyword=="binary" ): return self._binary elif ( keyword=="block_size" ): return self._blksize elif ( keyword=="stripe_size" ): return self._stripesize elif ( keyword=="tcp_buffer" ): return self._tcpbuf elif ( keyword=="no_tpt" ): return self._allow3rdparty elif ( keyword=="dcau" ): return self._dcau elif ( keyword=="streams" ): return self._streams elif ( keyword=="concurrency" ): return self._nsimfiles elif ( keyword=="src_subject" ): return self._srcsubj elif ( keyword=="dest_subject" ): return self._destsubj elif ( keyword=="cleanup" ): return self._cleanup elif ( keyword=="retries" ): return self._retries else: return None def addURLpair(self,src,dest): if ( self._destlist is None ): self._destlist = dict() self._destlist[src] = dest def getSrcURL(self,dest): src = None if ( self._destlist is not None and (dest in self._destlist.values()) ): for (mysrc,mydest) in self._destlist.items(): if ( dest==mydest ): src = mysrc break return src def getDestURL(self,src): dest = None if ( self._destlist is not None and self._destlist.has_key(src) ): dest = self._destlist[src] return dest def getURLdict(self): if ( self._destlist is not None ): return self._destlist else: return dict() def getAllSrcURLs(self): if ( self._destlist is not None ): return self._destlist.keys() else: return () class ScheduledDataTransfer(object): def __init__(self,filename=None,desctype="urlpair",movertype="supermover",batchtype="pbs",server=None,port=None,credcache=False): self.setName("dmsub") self.setWalltime("1:00:00") self.setQueue(None) self.setFileName(filename) self.setDescriptionType(desctype) self.setMoverType(movertype) self.setBatchType(batchtype) self.setServer(server) self.setPort(port) self.setCredCache(credcache) self._cachedlist = None self._needupdate = True self._errlog = None self._outlog = None self._joinlogs = None self._dep = None self._vars = None def setFileName(self,filename): self._file = filename self._needupdate = True def getFileName(self): return self._file def setName(self,name): self._name = name def getName(self): return self._name def setWalltime(self,walltime): self._walltime = walltime def getWalltime(self): return self._walltime def setQueue(self,queue): self._queue = queue def getQueue(self): return self._queue def setErrLog(self,errlog): self._errlog = errlog def getErrLog(self): return self._errlog def setOutLog(self,outlog): self._outlog = outlog def getOutLog(self): return self._outlog def setJoinLogs(self,joinlogs): self._joinlogs = joinlogs def getJoinLogs(self): return self._joinlogs def setDescriptionType(self,type): if ( type=="urlpair" or type=="dmover" or type=="rft" or type=="stork" ): self._desctype = type else: raise RuntimeError("invalid description type \""+type+"\"") self._needupdate = True def getDescriptionType(self): return self._desctype def setMoverType(self,type): if ( type=="supermover" or type=="guc" or type=="globus-url-copy" or type=="dmover" or type=="rft" or type=="stork" ): self._movertype = type else: raise RuntimeError("invalid mover type \""+type+"\"") def getMoverType(self): return self._movertype def setBatchType(self,type): if ( type=="pbs" or type=="torque" or type=="dmover" or type=="rft" or type=="stork" or type=="condor" ): self._batchtype = type else: raise RuntimeError("invalid batch type \""+type+"\"") def getBatchType(self): return self._batchtype def setDependency(self,dep): self._dep = dep def getDependency(self): return self._dep def setVars(self,vars): if ( self._vars is None ): self._vars = [] varlist = vars.split(",") for var in varlist: self._vars.append(var) def getVars(self): return self._vars def setServer(self,server): self._server = server def getServer(self): return self._server def setPort(self,Port): self._Port = Port def getPort(self): return self._Port def setCredCache(self,credcache): self._credcache = credcache def getCredCache(self): return self._credcache def transferList(self): if ( not self._needupdate and self._cachedlist is not None ): if ( debug ): sys.stderr.write("using cached version of transfer list\n") return self._cachedlist if ( debug ): sys.stderr.write("generating new version of transfer list\n") list = None try: fd = open(self._file,'r') if ( self._desctype=="urlpair" or self._desctype=="dmover" ): list = self._readSimpleXferList(fd) elif ( self._desctype=="rft" ): list = self._readRFTXferList(fd) elif ( self._desctype=="stork" ): list = self._readStorkXferList(fd) fd.close() except IOError,msg: raise IOError,msg self._needupdate = False self._cachedlist = list return list def _readSimpleXferList(self,fd): list = TransferList() lines = fd.readlines() for line in lines: if ( not line.startswith("#") ): token = line.split() if ( len(token)>1 ): list.addURLpair(token[0],token[1]) return list def _readRFTXferList(self,fd): # RFT format is basically a bunch of options in a specific order # (values only, no keywords for some unknown reason) followed by # a simple URL pair list (on separate lines, again for some unknown # reason). list = TransferList() line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("binary",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("block_size",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("tcp_buffer",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("no_tpt",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("streams",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("dcau",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("concurrency",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("src_subject",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("dest_subject",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("cleanup",line.rstrip('\n')) line = fd.readline() while ( line.startswith("#") ): line = fd.readline() list.setOption("retries",line.rstrip('\n')) while ( True ): line = fd.readline() while ( line.startswith("#") ): line = fd.readline() if ( len(line)==0 ): break src = line.rstrip('\n') line = fd.readline() while ( line.startswith("#") ): line = fd.readline() if ( len(line)==0 ): break dest = line.rstrip('\n') list.addURLpair(src,dest) return list def _readStorkXferList(self,fd): list = TransferList() src = None dest = None lines = fd.readlines() for line in lines: if ( not line.startswith("//") ): token = line.split(); if ( token[0]=="src_url" ): src = token[2].rstrip(';').rstrip('"').lstrip('"') elif ( token[0]=="dest_url" ): dest = token[2].rstrip(';').rstrip('"').lstrip('"') elif ( token[0]=="x509proxy" ): list.setOption("x509proxy",token[2].rstrip(';').rstrip('"').lstrip('"')) elif ( token[0]=="alt_protocols" ): list.setOption("alt_protocols",token[2].rstrip(';').rstrip('"').lstrip('"')) list.addURLpair(src,dest) return list def writeDescription(self,filename=None,desctype="urlpair"): list = self.transferList() if ( filename is None ): fd = open(self.getName()+"."+desctype) else: fd = open(filename,'w') if ( desctype=="urlpair" or desctype=="dmover" ): fd.write("# automatically generated by dmsub\n") fd.write("# urlpair/dmover format\n") for src in list.getAllSrcURLs(): fd.write(src+" "+list.getDestURL(src)+"\n") elif ( desctype=="rft" ): # RFT format is basically a bunch of options in a specific order # (values only, no keywords for some unknown reason) followed by # a simple URL pair list (on separate lines, again for some unknown # reason). fd.write("# automatically generated by dmsub\n") fd.write("# RFT format\n") fd.write("# binary?\n") fd.write(str(list.getOption("binary"))+"\n") fd.write("# block size in bytes\n") fd.write(str(list.getOption("block_size"))+"\n") fd.write("# TCP buffer size in bytes\n") fd.write(str(list.getOption("tcp_buffer"))+"\n") fd.write("# No 3rd-party transfers?\n") fd.write(str(not list.getOption("no_tpt"))+"\n") fd.write("# parallel streams\n") fd.write(str(list.getOption("streams"))+"\n") fd.write("# Data Channel Authentication (DCAU)\n") fd.write(str(list.getOption("dcau"))+"\n") fd.write("# concurrency\n") fd.write(str(list.getOption("concurrency"))+"\n") fd.write("# grid subject of src server\n") fd.write(str(list.getOption("src_subject"))+"\n") fd.write("# grid subject of dest server\n") fd.write(str(list.getOption("dest_subject"))+"\n") fd.write("# clean up all transfers on failure of any\n") fd.write(str(list.getOption("cleanup"))+"\n") fd.write("# max retries\n") fd.write(str(list.getOption("retries"))+"\n") fd.write("# src/dest URL pairs\n") for src in list.getAllSrcURLs(): fd.write(src+"\n") fd.write(list.getDestURL(src)+"\n") elif ( desctype=="stork" ): fd.write("// automatically generated by dmsub\n") fd.write("// stork format\n") for src in list.getAllSrcURLs(): fd.write("[\n") fd.write(" dap_type = transfer;\n") fd.write(" src_url = \""+str(src)+"\";\n") fd.write(" dest_url = \""+str(list.getDestURL(src))+"\";\n") if ( list.getOption("x509proxy")!="default" ): fd.write(" x509proxy = \""+str(list.getOption("x509proxy"))+"\";\n") elif ( "X509_USER_PROXY" in os.environ ): fd.write(" x509proxy = \""+os.environ["X509_USER_PROXY"]+"\";\n") if ( list.getOption("alt_protocols") is not None ): fd.write(" alt_protocols = \""+str(list.getOption("alt_protocols"))+"\";\n") fd.write("]\n") else: raise RuntimeError,"invalid description type \""+type+"\"" fd.close() def _doCredCache(self,proxy): if ( self.getCredCache() ): # the following assumes that $HOME is shared with the # data mover nodes on which the job will run cachedir = os.environ["HOME"]+"/.credcache" if ( not os.path.exists(cachedir) ): os.mkdir(cachedir,0700) newcred = cachedir+"/"+str(uuid.uuid1()) shutil.copyfile(os.environ["X509_USER_PROXY"],newcred) os.chmod(newcred,0600) if ( debug ): sys.stderr.write("caching credential "+proxy+" as "+newcred+"\n") return newcred else: return proxy def submit(self,dosubmit): jobid = None if ( self.getBatchType()=="pbs" or self.getBatchType()=="torque" ): jobid = self._submitPBS(dosubmit) elif ( self.getBatchType()=="dmover" ): jobid = self._submitDMover(dosubmit) elif ( self.getBatchType()=="rft" ): jobid = self._submitRFT(dosubmit) elif ( self.getBatchType()=="stork" or self.getBatchType()=="condor" ): jobid = self._submitStork(dosubmit) if ( jobid is not None ): sys.stdout.write(jobid+"\n") return jobid def _submitPBS(self,dosubmit): jobid = None epilogue = "" if ( self.getMoverType()=="stork" or self.getMoverType()=="rft" ): raise RuntimeError, "mover type \""+self.getMoverType()+"\" is incompatible with PBS" if ( debug ): sys.stderr.write("mover="+self.getMoverType()+"\n") if ( dosubmit ): cmd = "qsub" #pin,pout,perr = os.popen3(cmd) p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) pin = p.stdin perr = p.stderr pout = p.stdout else: pin = sys.stdout pout = open("/dev/null") perr = open("/dev/null") pin.write("#PBS -N "+self.getName()+"\n") pin.write("#PBS -S /bin/bash\n") if ( self.getErrLog() is not None ): pin.write("#PBS -e "+self.getErrLog()+"\n") if ( self.getOutLog() is not None ): pin.write("#PBS -o "+self.getOutLog()+"\n") if ( self.getJoinLogs() is not None ): pin.write("#PBS -j "+self.getJoinLogs()+"\n") pin.write("#PBS -l walltime="+self.getWalltime()+"\n") if ( self.getQueue() is not None ): pin.write("#PBS -q "+self.getQueue()+"\n") if ( self.getDependency() is not None ): pin.write("#PBS -W depend=afterok:"+(",afterok:".join(self.getDependency().split(":")))+"\n") if ( hold ): pin.write("#PBS -h\n") if ( self.getVars() is not None ): myvars = self.getVars() pin.write("#PBS -v "+myvars[0]) if ( len(myvars)>1 ): for var in myvars[1:]: pin.write(","+var) pin.write("\n") if ( pbs_prologue is not None ): lines = pbs_prologue.split("\\n") for line in lines: pin.write(line+"\n") list = self.transferList() if ( list.getOption("x509proxy")!="default" ): if ( debug ): sys.stderr.write("using x509 proxy "+list.getOption("x509proxy")+"\n") if ( self.getCredCache() ): newcred = self._doCredCache(list.getOption("x509proxy")) pin.write("export X509_USER_PROXY="+newcred+"\n") epilogue = epilogue+"rm -f "+newcred+"\n" else: pin.write("export X509_USER_PROXY="+list.getOption("x509proxy")+"\n") elif ( "X509_USER_PROXY" in os.environ and os.path.exists(os.environ["X509_USER_PROXY"]) ): if ( debug ): sys.stderr.write("using x509 proxy "+os.environ["X509_USER_PROXY"]+"\n") if ( self.getCredCache() ): newcred = self._doCredCache(os.environ["X509_USER_PROXY"]) pin.write("export X509_USER_PROXY="+newcred+"\n") epilogue = epilogue+"rm -f "+newcred+"\n" if ( self.getMoverType()=="guc" or self.getMoverType()=="globus-url-copy" ): pin.write("export GUC_OPTIONS=\"-q -cd -r -stripe -fast -tcp-bs "+str(list.getOption("tcp_buffer"))+" -bs "+str(list.getOption("block_size"))) if ( not list.getOption("binary") ): pin.write(" -ascii") if ( not list.getOption("dcau") ): pin.write(" -nodcau") if ( list.getOption("src_subject") is not None ): pin.write(" -ss '"+list.getOption("src_subject")+"'") if ( list.getOption("dest_subject") is not None ): pin.write(" -ds '"+list.getOption("src_subject")+"'") if ( not list.getOption("allow3rdparty") ): pin.write(" -notpt") if ( list.getOption("streams")>1 ): pin.write(" -parallel "+str(list.getOption("streams"))+" -sbs "+str(list.getOption("stripe_size"))) if ( not list.getOption("cleanup") ): pin.write(" -continue-on-error") if ( list.getOption("retries")>0 ): pin.write(" -restart -rst-retries "+str(list.getOption("retries"))) pin.write("\"\n") pin.flush() for src in list.getAllSrcURLs(): if ( debug ): sys.stderr.write("src_url="+src+"\n") if ( self.getMoverType()=="supermover" ): pin.write("supermover "+src+" "+list.getDestURL(src)+"\n") elif ( self.getMoverType()=="guc" or self.getMoverType()=="globus-url-copy" ): pin.write("globus-url-copy $GUC_OPTIONS \""+src+"\" \""+list.getDestURL(src)+"\"\n") pin.flush() if ( debug ): sys.stderr.write("done writing url list\n") if ( epilogue!="" ): if ( debug ): sys.stderr.write("appending epilogue to script\n") pin.write(epilogue) pin.close() try: if ( debug ): sys.stderr.write("parsing qsub stdout\n") stdout = pout.readlines() if ( len(stdout)>0 ): jobid = stdout[0].rstrip('\n') if ( debug ): sys.stderr.write("parsing qsub stderr\n") stderr = perr.readlines() if ( len(stderr)>0 ): for line in stderr: sys.stderr.write(line) sys.stderr.flush() except IndexError, err: sys.stderr.write(str(err) + "\n") sys.stderr.flush() pout.close() perr.close() if ( jobid is None or jobid=="" ): sys.stderr.write("No job id, assuming failure\n") sys.exit(-1) return jobid def _tmpDir(self): tmpdir = "/tmp" if ( ("TMPDIR" in os.environ) and os.path.exists(os.environ["TMPDIR"]) ): tmpdir = os.environ["TMPDIR"] elif ( ("TG_NODE_SCRATCH" in os.environ) and os.path.exists(os.environ["TG_NODE_SCRATCH"]) ): tmpdir = os.environ["TG_NODE_SCRATCH"] if ( debug ): sys.stderr.write("using tmp dir "+tmpdir+"\n") return tmpdir def _submitDMover(self,dosubmit): if ( self.getMoverType()!="dmover" ): raise RuntimeError("mover type \""+self.getMoverType()+"\" is incompatible with dmover") jobid = None list = self.transferList() proxy = list.getOption("x509proxy") if ( proxy is not None and proxy!="default" and os.path.exists(proxy) ): if ( self.getCredCache() ): newcred = self._doCredCache(proxy) os.environ["X509_PROXY"] = newcred else: os.environ["X509_PROXY"] = proxy if ( debug ): sys.stderr.write("using x5090 proxy "+os.environ["X509_PROXY"]+"\n") tmpfile = self._tmpDir()+"/u"+str(os.getuid())+"_p"+str(os.getpid())+".dmover" self.writeDescription(tmpfile,"dmover") cmd = "dsub" if ( os.environ["X509_PROXY"] is not None and os.environ["X509_PROXY"]!="default" and os.path.exists(os.environ["X509_PROXY"]) ): cmd += " -c "+os.environ["X509_PROXY"] if ( self.getQueue() is not None ): cmd += " -q "+self.getQueue() if ( self.getDependency() is not None ): cmd += " -D "+self.getDependency() if ( self.getOutLog() is not None ): cmd += " -o "+self.getOutLog() if ( self.getWalltime() is not None ): cmd += " -w "+self.getWalltime() if ( list.getOption("retries")>0 ): cmd += " -R "+str(list.getOption("retries")) if ( list.getOption("streams")>1 ): cmd += " -s "+str(list.getOption("streams")) if ( hold ): cmd += " -H" cmd += " -f "+tmpfile if ( debug ): sys.stderr.write("executing command \""+cmd+"\"\n") if ( dosubmit ): #pin,pout,perr = os.popen3(cmd) p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) pin = p.stdin perr = p.stderr pout = p.stdout pin.close() stdout = pout.readlines() for line in stdout: if ( line.startswith("Executing qsub:") ): token = line.split() if ( len(token)>2 ): jobid = token[2].rstrip('\n') sys.stderr.write(line) pout.close() stderr = perr.readlines() for line in stderr: sys.stderr.write(line) perr.close() else: sys.stderr.write(cmd+"\n") os.system("cat "+tmpfile) os.unlink(tmpfile) return jobid def _submitRFT(self,dosubmit): if ( self.getMoverType()!="rft" ): raise RuntimeError("mover type \""+self.getMoverType()+"\" is incompatible with RFT") if ( self.getServer() is None ): raise RuntimeError("RFT server not set") jobid = None proxy = self.transferList().getOption("x509proxy") if ( proxy is not None and proxy!="default" and os.path.exists(proxy) ): if ( self.getCredCache() ): newcred = self._doCredCache(proxy) os.environ["X509_PROXY"] = newcred else: os.environ["X509_PROXY"] = proxy tmpfile = self._tmpDir()+"/u"+str(os.getuid())+"_p"+str(os.getpid())+".rft" self.writeDescription(tmpfile,"rft") cmd = "rft -h "+self.getServer() if ( self.getPort() is not None ): cmd += " -r "+self.getPort() cmd += " -f "+tmpfile if ( debug ): sys.stderr.write("executing command \""+cmd+"\"\n") if ( dosubmit ): #pin,pout,perr = os.popen3(cmd) p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) pin = p.stdin perr = p.stderr pout = p.stdout pin.close() stdout = pout.readlines() for line in stdout: sys.stderr.write(line) pout.close() stderr = perr.readlines() for line in stderr: sys.stderr.write(line) perr.close() else: sys.stderr.write(cmd+"\n") os.system("cat "+tmpfile) os.unlink(tmpfile) return jobid def _submitStork(self,dosubmit): if ( self.getMoverType()!="stork" ): raise RuntimeError("mover type \""+self.getMoverType()+"\" is incompatible with Stork") proxy = self.transferList().getOption("x509proxy") if ( proxy is not None and proxy!="default" and os.path.exists(proxy) ): if ( self.getCredCache() ): newcred = self._doCredCache(proxy) os.environ["X509_PROXY"] = newcred else: os.environ["X509_PROXY"] = proxy elif ( ("X509_PROXY" in os.environ.keys()) and self.getCredCache() ): newcred = self._doCredCache(os.environ["X509_PROXY"]) os.environ["X509_PROXY"] = newcred jobid = None tmpfile = self._tmpDir()+"/u"+str(os.getuid())+"_p"+str(os.getpid())+".stork" self.writeDescription(tmpfile,"stork") cmd = "stork_submit" if ( self.getServer() is not None ): cmd += " -name "+self.getServer() if ( self.getPort() is not None ): cmd += ":"+str(self.getPort()) cmd += " "+tmpfile if ( debug ): sys.stderr.write("executing command \""+cmd+"\"\n") if ( dosubmit ): #pin,pout,perr = os.popen3(cmd) p = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) pin = p.stdin perr = p.stderr pout = p.stdout pin.close() if ( len(stdout)>0 ): jobid = stdout[0].rstrip('\n') pout.close() for line in stderr: sys.stderr.write(line) perr.close() else: sys.stderr.write(cmd+"\n") os.system("cat "+tmpfile) os.unlink(tmpfile) return jobid informat = "urlpair" mover = "globus-url-copy" rm = "pbs" cfgfile = "/usr/local/etc/dmsub.cfg" if ( os.path.exists("/etc/dmsub.cfg") ): cfgfile = "/etc/dmsub.cfg" if ( "DMSUB_CFG" in os.environ ): cfgfile = os.environ["DMSUB_CFG"] dosubmit = True queue = None jobname = None walltime = None outlog = None errlog = None joinlogs = None server = None port = None dependency = None vars = None pbs_prologue = None def usage(): ## global informat ## global rm ## global mover ## global queue ## global walltime ## global debug ## global server ## global port ## global streams ## global hold sys.stderr.write("Usage: dmsub [arguments] [-f] xfer_file\n\n") sys.stderr.write("Arguments:\n") sys.stderr.write(" -b rm, --batch=rm, --rm=rm\n\tUse \"rm\" batch environment (default is "+str(rm)+")\n") sys.stderr.write(" -C cfgfile, --config=cfgfile\n\tRead config file \"cfgfile\"\n") sys.stderr.write(" -c credfile, --credential=credfile\n\tUse file \"credfile\" as GSI credential\n") sys.stderr.write(" -D jobid[:jobid], --dependency=jobid[:jobid]\n\tDo not run data transfer job till after jobid completes\n") sys.stderr.write(" -d, --debug\n\tEnable debugging output\n") sys.stderr.write(" -e errlog, --errlog=errlog\n\tWrite stderr from data transfer to \"errlog\"\n") sys.stderr.write(" -F fmt, --format=fmt\n\tUse \"fmt\" as data transfer description format (default is "+str(informat)+")\n") sys.stderr.write(" -H, --hold\n\tSubmit job in held state if possible (default is "+str(hold)+")\n") sys.stderr.write(" -h, --help\n\tPrint this help message\n") sys.stderr.write(" -j join, --joinlogs=join\n\tJoin stdout and stderr logs (default is "+str(joinlogs)+")\n") sys.stderr.write(" -l rsrc=limit[,rsrc=limit], --limit=rsrc=limit[,rsrc=limit]\n\tSet batch resource limit (e.g. walltime)\n") sys.stderr.write(" -M mvr, --mover=mvr\n\tUse \"mvr\" as the data movement mechanism (default is "+mover+")\n") sys.stderr.write(" -N jobname, --name=jobname\n\tUse \"jobnname\" as name of data movement job\n") sys.stderr.write(" -n, --nosubmit\n\tDo a dry run without submitting any jobs\n") sys.stderr.write(" -o log, --outlog=log\n\tWrite stdout from data transfer to \"log\"\n") sys.stderr.write(" -P pt, --port=pt\n\tConnect to data transfer service on port \"pt\"\n") sys.stderr.write(" -q dest, --queue=dest\n\tSubmit data transfer job to batch queue \"dest\" (default is "+str(queue)+")\n") sys.stderr.write(" -s n, --streams=n\n\tUse up to n striped data streams whenever possible (default is "+str(streams)+")\n") sys.stderr.write(" -S svr, --server=svr\n\tConnect to data transfer service on server \"svr\"\n") sys.stderr.write(" -X, --cache-credential\n\tCache GSI credential if found\n (default)") sys.stderr.write(" -x, --no-cache-credential\n\tDo not cache GSI credential if found\n") sys.stderr.write("\nSupported batch environments: pbs, torque, condor, stork, rft\n") sys.stderr.write("Supported transfer descriptions: urlpair, dmover, rft, stork\n") sys.stderr.write("Supported data movers: globus-url-copy, supermover, stork, rft\n") sys.exit(0) def read_cfg(file): global informat global rm global mover global queue global walltime global debug global server global port global streams global stripesize global blksize global vars global pbs_prologue global credcache if ( not os.path.exists(file) ): raise IOError,file+" does not exist" elif ( not os.path.isfile(file) ): raise IOError("Config \"file\" "+file+" is not actually a file") else: if ( debug ): sys.stderr.write("reading config file "+cfgfile+"\n") fd = open(file,"r") lines = fd.readlines() for line in lines: token = line.split("=",2) if ( token!=[] and len(token)==2 and not token[0].startswith("#") ): keyword = token[0].strip() value = token[1].strip() if ( debug ): sys.stderr.write(file+": "+keyword+"="+value+"\n") if ( keyword=="batch" or keyword=="rm" ): rm = value elif ( keyword=="mover" ): mover = value elif ( keyword=="queue" ): queue = value elif ( keyword=="walltime" ): walltime = value elif ( keyword=="server" ): server = value elif ( keyword=="port" ): port = value elif ( keyword=="format" ): informat = value elif ( keyword=="streams" ): streams = value elif ( keyword=="stripe_size" ): stripesize = value elif ( keyword=="block_size" ): stripesize = value elif ( keyword=="debug" ): if ( value=="true" or value=="True" or value=="TRUE" ): debug = True else: debug = False elif ( keyword=="pbs_prologue" ): pbs_prologue = value try: if ( debug ): sys.stderr.write("invoking getopt\n") opts, args = getopt.getopt(sys.argv[1:], "b:C:c:D:de:F:f:Hhj:l:M:no:N:P:q:S:s:v:w:Xx", ["batch=", "credential=", "cache-credential", "config=s", "debug", "dependency=", "errlog=", "format=", "help", "hold", "joinlogs=", "limit=", "mover=", "name=", "no-cache-credential", "nosubmit", "outlog=", "port=", "queue=", "rm=", "server=", "streams=", "vars="]) except getopt.GetoptError, err: sys.stderr.write(str(err)+"\n") usage() if ( os.path.exists(cfgfile) ): if ( debug ): sys.stderr.write("reading global config file\n") read_cfg(cfgfile) else: if ( debug ): sys.stderr.write("global config file "+cfgfile+" not found\n") if ( debug ): sys.stderr.write("parsing command line options\n") for opt in opts: if ( opt[0]=="-b" or opt[0]=="--batch " or opt[0]=="--rm" ): if ( debug ): sys.stderr.write("setting rm to "+opt[1]+"\n") rm = opt[1] if ( opt[0]=="-C" or opt[0]=="--config" ): read_cfg(opt[1]) if ( opt[0]=="-c" or opt[0]=="--credential" ): if ( os.path.exists(opt[1]) and os.access(opt[1],os.R_OK) ): os.environ["X509_USER_PROXY"] = opt[1] elif ( not os.access(opt[1],os.R_OK) ): raise IOError(opt[1]+" is not readable") else: raise IOError(opt[1]+" does not exist") if ( opt[0]=="-D" or opt[0]=="--dependency" ): if ( debug ): sys.stderr.write("setting dependency to "+opt[1]+"\n") dependency = opt[1] if ( opt[0]=="-d" or opt[0]=="--debug" ): debug = True if ( opt[0]=="-e" or opt[0]=="--errlog" ): if ( debug ): sys.stderr.write("setting errlog to "+opt[1]+"\n") errlog = opt[1] if ( opt[0]=="-F" or opt[0]=="--format" ): if ( debug ): sys.stderr.write("setting format to "+opt[1]+"\n") informat = opt[1] if ( opt[0]=="-f" ): args.append(opt[1]) if ( opt[0]=="-H" or opt[0]=="--hold" ): hold = True if ( opt[0]=="-h" or opt[0]=="--help" ): usage() if ( opt[0]=="-j" or opt[0]=="--joinlogs" ): if ( debug ): sys.stderr.write("setting joinlogs to "+opt[1]+"\n") joinlogs = opt[1] if ( opt[0]=="-l" or opt[0]=="--limit" ): if ( debug ): sys.stderr.write("setting limits to "+opt[1]+"\n") limits = opt[1].split(",") for limit in limits: (rsrc,value) = limit.split("=",1) if ( rsrc=="walltime" ): walltime = value if ( opt[0]=="-M" or opt[0]=="--mover" ): if ( debug ): sys.stderr.write("setting mover to "+opt[1]+"\n") mover = opt[1] if ( opt[0]=="-N" or opt[0]=="--name" ): if ( debug ): sys.stderr.write("setting jobname to "+opt[1]+"\n") jobname = opt[1] if ( opt[0]=="-n" or opt[0]=="--nosubmit" ): dosubmit = False if ( opt[0]=="-o" or opt[0]=="--outlog" ): if ( debug ): sys.stderr.write("setting outlog to "+opt[1]+"\n") outlog = opt[1] if ( opt[0]=="-P" or opt[0]=="--port" ): if ( debug ): sys.stderr.write("setting port to "+opt[1]+"\n") port = opt[1] if ( opt[0]=="-q" or opt[0]=="--queue" ): if ( debug ): sys.stderr.write("setting queue to "+opt[1]+"\n") queue = opt[1] if ( opt[0]=="-S" or opt[0]=="--server" ): if ( debug ): sys.stderr.write("setting server to "+opt[1]+"\n") server = opt[1] if ( opt[0]=="-s" or opt[0]=="--streams" ): if ( debug ): sys.stderr.write("setting streams to "+opt[1]+"\n") streams = opt[1] if ( opt[0]=="-v" ): vars = opt[1] if ( opt[0]=="-w" ): walltime = opt[1]+":00" if ( opt[0]=="-X" or opt[0]=="--cache-credential" ): credcache = True if ( opt[0]=="-x" or opt[0]=="--no-cache-credential" ): credcache = False # What to do if credcache is set but no cred is specified? # Look for default and use that... if ( credcache and not ( "X509_USER_PROXY" in os.environ.keys() ) ): defproxy = "/tmp/x509up_u"+str(os.getuid()) if ( os.path.exists(defproxy) and os.access(defproxy,os.R_OK) ): os.environ["X509_USER_PROXY"] = defproxy for file in args: if ( debug ): sys.stderr.write("creating job from file "+file+"\n") job = ScheduledDataTransfer(file,informat,mover,rm,server,port,credcache) if ( jobname is not None ): job.setName(jobname) if ( queue is not None ): job.setQueue(queue) if ( walltime is not None ): job.setWalltime(walltime) if ( outlog is not None ): job.setOutLog(outlog) if ( errlog is not None ): job.setErrLog(errlog) if ( joinlogs is not None ): job.setJoinLogs(joinlogs) if ( dependency is not None ): job.setDependency(dependency) if ( vars is not None ): job.setVars(vars) if ( debug ): sys.stderr.write("submitting job from file "+file+" as job "+job.getName()+"\n") jobid = job.submit(dosubmit)