| 1 |
spiga |
1.1 |
from Actor import *
|
| 2 |
|
|
from crab_util import *
|
| 3 |
|
|
import common
|
| 4 |
|
|
from ApmonIf import ApmonIf
|
| 5 |
|
|
import time
|
| 6 |
|
|
from ProgressBar import ProgressBar
|
| 7 |
|
|
from TerminalController import TerminalController
|
| 8 |
|
|
|
| 9 |
mcinquil |
1.2 |
import xml.dom.minidom
|
| 10 |
|
|
|
| 11 |
spiga |
1.1 |
class GetOutputServer(Actor):
|
| 12 |
|
|
|
| 13 |
|
|
def __init__(self, cfg_params,):
|
| 14 |
|
|
self.cfg_params = cfg_params
|
| 15 |
spiga |
1.3 |
try:
|
| 16 |
|
|
self.server_name = self.cfg_params['CRAB.server_name'] # gsiftp://pcpg01.cern.ch/data/SEDir/
|
| 17 |
mcinquil |
1.11 |
if not self.server_name.endswith("/"):
|
| 18 |
|
|
self.server_name = self.server_name + "/"
|
| 19 |
spiga |
1.3 |
except KeyError:
|
| 20 |
|
|
msg = 'No server selected ...'
|
| 21 |
|
|
msg = msg + 'Please specify a server in the crab cfg file'
|
| 22 |
|
|
raise CrabException(msg)
|
| 23 |
spiga |
1.1 |
return
|
| 24 |
|
|
|
| 25 |
|
|
def run(self):
|
| 26 |
|
|
"""
|
| 27 |
|
|
The main method of the class: retrieve the output from server
|
| 28 |
|
|
"""
|
| 29 |
|
|
common.logger.debug(5, "GetOutput server::run() called")
|
| 30 |
|
|
|
| 31 |
|
|
start = time.time()
|
| 32 |
|
|
|
| 33 |
|
|
common.scheduler.checkProxy()
|
| 34 |
|
|
|
| 35 |
|
|
common.taskDB.load()
|
| 36 |
|
|
WorkDirName =os.path.basename(os.path.split(common.work_space.topDir())[0])
|
| 37 |
|
|
projectUniqName = 'crab_'+str(WorkDirName)+'_'+common.taskDB.dict("TasKUUID")
|
| 38 |
|
|
#Need to add a check on the treashold level
|
| 39 |
|
|
# and on the task readness TODO
|
| 40 |
|
|
try:
|
| 41 |
mcinquil |
1.7 |
### retrieving project from the server
|
| 42 |
|
|
common.logger.message("Retrieving the project from the server...\n")
|
| 43 |
mcinquil |
1.2 |
|
| 44 |
|
|
copyHere = common.work_space.resDir() # MATT
|
| 45 |
farinafa |
1.8 |
if "USER.outputdir" in self.cfg_params.keys() and os.path.isdir(self.cfg_params["USER.outputdir"]):
|
| 46 |
|
|
copyHere = self.cfg_params["USER.outputdir"] + "/" # MATT
|
| 47 |
mcinquil |
1.2 |
|
| 48 |
mcinquil |
1.6 |
cmd = 'lcg-cp --vo cms --verbose gsiftp://' + str(self.server_name) + str(projectUniqName)+'/res/done.tar.gz file://'+copyHere+'done.tar.gz'# MATT
|
| 49 |
mcinquil |
1.2 |
common.logger.debug(5, cmd)
|
| 50 |
spiga |
1.1 |
copyOut = os.system(cmd +' >& /dev/null')
|
| 51 |
|
|
except:
|
| 52 |
|
|
msg = ("Output not yet available")
|
| 53 |
|
|
raise CrabException(msg)
|
| 54 |
|
|
|
| 55 |
mcinquil |
1.6 |
zipOut = "done.tar.gz"
|
| 56 |
mcinquil |
1.7 |
if os.path.exists( copyHere + zipOut ):
|
| 57 |
spiga |
1.1 |
cwd = os.getcwd()
|
| 58 |
mcinquil |
1.7 |
os.chdir( copyHere )
|
| 59 |
spiga |
1.1 |
common.logger.debug( 5, 'tar -zxvf ' + zipOut )
|
| 60 |
spiga |
1.3 |
cmd = 'tar -zxvf ' + zipOut
|
| 61 |
mcinquil |
1.13 |
cmd_out = runCommand(cmd)
|
| 62 |
|
|
toMove = os.listdir(".tmpDone/")
|
| 63 |
|
|
common.logger.debug( 5, "file returned: \n" +str(toMove)+ "\n")
|
| 64 |
|
|
for fileToMove in toMove:
|
| 65 |
|
|
common.logger.debug( 6, "moving file: "+str(fileToMove) )
|
| 66 |
|
|
cmd_out = runCommand("mv .tmpDone/"+str(fileToMove)+" .")
|
| 67 |
|
|
common.logger.debug( 6, "deleting all temporary files...")
|
| 68 |
|
|
cmd = 'rm -drf .tmpDone/'
|
| 69 |
spiga |
1.1 |
cmd_out = runCommand(cmd)
|
| 70 |
|
|
os.chdir(cwd)
|
| 71 |
mcinquil |
1.7 |
common.logger.debug( 5, 'rm -f '+copyHere+zipOut )
|
| 72 |
|
|
cmd = 'rm -f '+copyHere+zipOut
|
| 73 |
spiga |
1.1 |
cmd_out = runCommand(cmd)
|
| 74 |
|
|
|
| 75 |
mcinquil |
1.2 |
try:
|
| 76 |
farinafa |
1.10 |
# file = open(common.work_space.resDir()+"xmlReportFile.xml", "r")
|
| 77 |
mcinquil |
1.2 |
doc = xml.dom.minidom.parse(common.work_space.resDir()+ "xmlReportFile.xml" )
|
| 78 |
|
|
|
| 79 |
farinafa |
1.10 |
task = doc.childNodes[0].childNodes[1].getAttribute("taskName")
|
| 80 |
mcinquil |
1.2 |
self.countToTjob = int(doc.childNodes[0].childNodes[1].getAttribute("totJob") )
|
| 81 |
|
|
|
| 82 |
mcinquil |
1.7 |
ended = doc.childNodes[0].childNodes[1].getAttribute("ended")
|
| 83 |
|
|
|
| 84 |
mcinquil |
1.2 |
addTree = 3
|
| 85 |
|
|
if doc.childNodes[0].childNodes[3].getAttribute("id") != "all":
|
| 86 |
|
|
common.jobDB.load()
|
| 87 |
|
|
for job in range( self.countToTjob ):
|
| 88 |
|
|
idJob = doc.childNodes[0].childNodes[job+addTree].getAttribute("id")
|
| 89 |
|
|
status = doc.childNodes[0].childNodes[job+addTree].getAttribute("status")
|
| 90 |
|
|
cleared = doc.childNodes[0].childNodes[job+addTree].getAttribute("cleared")
|
| 91 |
mcinquil |
1.4 |
if int(cleared) == 1 and (status == "Done" or status == "Done (Failed)"):
|
| 92 |
mcinquil |
1.2 |
common.jobDB.setStatus( str(int(idJob)-1), "Y" )
|
| 93 |
|
|
addTree += 1
|
| 94 |
|
|
common.jobDB.save()
|
| 95 |
|
|
except Exception, ex:
|
| 96 |
farinafa |
1.9 |
msg = ("Problems accessing the report file: " + str(ex))
|
| 97 |
mcinquil |
1.2 |
raise CrabException(msg)
|
| 98 |
|
|
|
| 99 |
mcinquil |
1.7 |
common.logger.message('Task Completed at '+str(ended)+' %\n')
|
| 100 |
farinafa |
1.9 |
msg='Data for project '+str(WorkDirName)+' successfully retrieved from server \n'
|
| 101 |
mcinquil |
1.7 |
msg+='and copied in '+copyHere+' \n'
|
| 102 |
|
|
common.logger.message(msg)
|
| 103 |
spiga |
1.1 |
else:
|
| 104 |
farinafa |
1.9 |
common.logger.message("Problems have been encoutered during project transfer. Please check with [status] option if jobs have finished .\n")
|
| 105 |
spiga |
1.1 |
|
| 106 |
|
|
return
|
| 107 |
|
|
|
