Last active
December 17, 2015 02:38
-
-
Save sergiobuj/5536818 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| usage: submitbatchjob.sh jobWorkingDirectory jobScratchDirectory jobDescriptionFile |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| tail -f /var/log/submit/submit-server.log /var/log/submit/distributor/* /var/log/submit/monitors/monitor* | |
| ==> /var/log/submit/submit-server.log <== | |
| [Tue May 7 18:47:24 2013] (1832) ==================================================== | |
| [Tue May 7 18:47:24 2013] (1832) Connection to tcp://:830 from ('10.15.0.1', 51704) | |
| [Tue May 7 18:47:24 2013] 0: Server will time out in 60 seconds. | |
| [Tue May 7 18:47:25 2013] 0: Cumulative job load is 2.97. (Max: 510.00) | |
| [Tue May 7 18:47:25 2013] 119: Args are:['/usr/bin/submit', '-v', 'pbsvenue', 'echo', 'works'] | |
| ==> /var/log/submit/distributor/distributor.log <== | |
| [Tue May 7 18:47:25 2013] 0: quotaCommand = quota -w | tail -n 1 | |
| [Tue May 7 18:47:25 2013] 119: command = tar vchf 00000119_01_input.tar --exclude='*.svn*' -C /home/agomezl .__local_jobid.00000119_01 | |
| [Tue May 7 18:47:25 2013] 119: remoteCommand echo works | |
| [Tue May 7 18:47:25 2013] 119: command = genuserid | |
| [Tue May 7 18:47:25 2013] 119: IDENTITY = /tmp/id.bbmRuobyPq | |
| [Tue May 7 18:47:25 2013] 119: command = update-known-hosts 192.168.56.101 | |
| [Tue May 7 18:47:25 2013] 119: workingDirectory /home/torqueman/scratch/1367981245_00000119_01 | |
| [Tue May 7 18:47:25 2013] 119: command = tar vrhf 00000119_01_input.tar --exclude='*.svn*' -C /home/agomezl/00000119/01 00000119_01.pbs 00000119_01.sh | |
| [Tue May 7 18:47:25 2013] 119: command = nice -n 19 gzip 00000119_01_input.tar | |
| [Tue May 7 18:47:25 2013] 119: command = cat /home/agomezl/00000119/01/00000119_01_input.tar.gz | ssh -T -x -a -i /tmp/id.bbmRuobyPq torqueman@192.168.56.101 "/home/torqueman/receiveinput.sh /home/torqueman/scratch/1367981245_00000119_01 .__timestamp_transferred.00000119_01" | |
| [Tue May 7 18:47:56 2013] 119: cat /home/agomezl/00000119/01/00000119_01_input.tar.gz | ssh -T -x -a -i /tmp/id.bbmRuobyPq torqueman@192.168.56.101 "/home/torqueman/receiveinput.sh /home/torqueman/scratch/1367981245_00000119_01 .__timestamp_transferred.00000119_01" failed w/ exit code 64 | |
| [Tue May 7 18:47:56 2013] 119: usage: /home/torqueman/receiveinput.sh jobWorkingDirectory jobScratchDirectory transferredTimestampFile | |
| [Tue May 7 18:47:56 2013] 119: venue=1:sshPBS::192.168.56.101 status=11 cputime=0.000000 realtime=0.000000 waittime=0.000000 ncpus=1 | |
| ==> /var/log/submit/submit-server.log <== | |
| [Tue May 7 18:47:56 2013] 119: Job Status: venue=1:sshPBS::192.168.56.101 status=11 cpu=0.000000 real=0.000000 wait=0.000000 | |
| [Tue May 7 18:47:56 2013] 119: Server(JobExecuter) exiting(11). | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| [pbsvenue] | |
| venues = 192.168.56.101 | |
| remoteBatchSystem = PBS | |
| remoteUser = torqueman | |
| remotePpn = 1 | |
| venueMechanism = ssh | |
| siteMonitorDesignator = pbsmonitor | |
| remoteScratchDirectory = /home/torqueman/scratch | |
| remoteBinDirectory = /home/torqueman |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # @package hubzero-submit-common | |
| # @file VenueMechanismSsh.py | |
| # @author Steven Clark <clarks@purdue.edu> | |
| # @copyright Copyright (c) 2012 HUBzero Foundation, LLC. | |
| # @license http://www.gnu.org/licenses/lgpl-3.0.html LGPLv3 | |
| # | |
| # Copyright (c) 2012 HUBzero Foundation, LLC. | |
| # | |
| # This file is part of: The HUBzero(R) Platform for Scientific Collaboration | |
| # | |
| # The HUBzero(R) Platform for Scientific Collaboration (HUBzero) is free | |
| # software: you can redistribute it and/or modify it under the terms of | |
| # the GNU Lesser General Public License as published by the Free Software | |
| # Foundation, either version 3 of the License, or (at your option) any | |
| # later version. | |
| # | |
| # HUBzero is distributed in the hope that it will be useful, | |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
| # GNU Lesser General Public License for more details. | |
| # | |
| # You should have received a copy of the GNU Lesser General Public License | |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
| # | |
| # HUBzero is a registered trademark of HUBzero Foundation, LLC. | |
| # | |
| import sys | |
| import os | |
| import re | |
| import stat | |
| import time | |
| import datetime | |
| import shutil | |
| from hubzero.submit.LogMessage import logID as log | |
| from hubzero.submit.JobStatistic import JobStatistic | |
| from hubzero.submit.JobOutput import * | |
| from hubzero.submit.VenueMechanismCore import VenueMechanismCore | |
| from hubzero.submit.RemoteBatchCONDOR import RemoteBatchCONDOR | |
| from hubzero.submit.RemoteBatchRUNJOB import RemoteBatchRUNJOB | |
| from hubzero.submit.RemoteBatchFACTORY import RemoteBatchFACTORY | |
| from hubzero.submit.RemoteBatchPEGASUS import RemoteBatchPEGASUS | |
| from hubzero.submit.RemoteBatchLL import RemoteBatchLL | |
| from hubzero.submit.RemoteBatchLSF import RemoteBatchLSF | |
| from hubzero.submit.RemoteBatchPBS import RemoteBatchPBS | |
| from hubzero.submit.RemoteBatchPBS8 import RemoteBatchPBS8 | |
| from hubzero.submit.RemoteBatchSLURM import RemoteBatchSLURM | |
| from hubzero.submit.RemoteBatchAppScript import RemoteBatchAppScript | |
| from hubzero.submit.RemoteInstantSCRIPT import RemoteInstantSCRIPT | |
| from hubzero.submit.RemoteInstantAppScript import RemoteInstantAppScript | |
| from hubzero.submit.RemoteRunjobAppScript import RemoteRunjobAppScript | |
| from hubzero.submit.RemoteWorkflowPEGASUS import RemoteWorkflowPEGASUS | |
| class VenueMechanismSsh(VenueMechanismCore): | |
| def __init__(self, | |
| remoteMonitors, | |
| hubUserName, | |
| hubUserId, | |
| batchCommands, | |
| isParametric, | |
| localJobId, | |
| instanceId, | |
| destination, | |
| enteredCommand, | |
| gridsite, | |
| tunnelsInfo, | |
| cloudInstanceId, | |
| stageInTarFile, | |
| transferExecutable, | |
| executable, | |
| stdinput, | |
| arguments, | |
| useEnvironment, | |
| environment, | |
| isMultiCoreRequest, | |
| siteInfo, | |
| managerInfo, | |
| nCpus, | |
| nNodes, | |
| ppn, | |
| wallTime, | |
| x509SubmitProxy, | |
| disableProbeCheck, | |
| disableStateCheck, | |
| quotaLimit, | |
| timeHistoryLogs, | |
| pegasusVersion, | |
| pegasusHome): | |
| VenueMechanismCore.__init__(self,timeHistoryLogs,siteInfo,managerInfo,remoteMonitors, | |
| isMultiCoreRequest,nCpus,nNodes,ppn) | |
| self.venueMechanism = 'ssh' | |
| self.hubUserName = hubUserName | |
| self.hubUserId = hubUserId | |
| remoteScriptsDirectory = siteInfo['remoteBinDirectory'].replace("$","\$") | |
| self.receiveInput = os.path.join(remoteScriptsDirectory,batchCommands['receiveInput']) | |
| self.submitBatchJob = os.path.join(remoteScriptsDirectory,batchCommands['submitBatchJob']) | |
| self.transmitResults = os.path.join(remoteScriptsDirectory,batchCommands['transmitResults']) | |
| self.cleanupJob = os.path.join(remoteScriptsDirectory,batchCommands['cleanupJob']) | |
| self.killBatchJob = os.path.join(remoteScriptsDirectory,batchCommands['killBatchJob']) | |
| self.isParametric = isParametric | |
| self.localJobId = localJobId | |
| if instanceId.startswith('WF;'): | |
| self.instanceId = "" | |
| self.nInstances = int(instanceId.split(';')[1]) | |
| else: | |
| self.instanceId = instanceId | |
| self.nInstances = 0 | |
| self.remoteBatchSystem = siteInfo['remoteBatchSystem'] | |
| if self.remoteBatchSystem == 'RUNJOB': | |
| self.destination = 'grid' | |
| else: | |
| self.destination = destination | |
| self.venue = siteInfo['venue'] | |
| self.enteredCommand = enteredCommand | |
| self.gridsite = gridsite | |
| self.siteMonitorDesignator = siteInfo['siteMonitorDesignator'] | |
| self.tunnelDesignator = siteInfo['tunnelDesignator'] | |
| if self.tunnelDesignator != "": | |
| gatewayHost,localHost = tunnelsInfo.getSSHTunnelHosts(self.tunnelDesignator) | |
| self.gatewayHost = gatewayHost | |
| self.localHost = localHost | |
| else: | |
| self.gatewayHost = "" | |
| self.localHost = "" | |
| self.cloudDesignator = siteInfo['cloudDesignator'] | |
| self.cloudInstanceId = cloudInstanceId | |
| self.remoteUser = siteInfo['remoteUser'] | |
| self.sharedUserSpace = siteInfo['sharedUserSpace'] | |
| self.logUserRemotely = siteInfo['logUserRemotely'] | |
| self.stageFiles = siteInfo['stageFiles'] | |
| if stageInTarFile.endswith('.gz'): | |
| self.stageInTarFile = stageInTarFile | |
| else: | |
| self.stageInTarFile = stageInTarFile + '.gz' | |
| self.transferExecutable = transferExecutable | |
| self.executable = executable | |
| self.stdinput = stdinput | |
| self.arguments = arguments | |
| self.useEnvironment = useEnvironment | |
| self.environment = environment | |
| self.remoteBatchQueue = siteInfo['remoteBatchQueue'] | |
| self.pegasusTemplates = siteInfo['pegasusTemplates'] | |
| self.wallTime = wallTime | |
| self.x509SubmitProxy = x509SubmitProxy | |
| self.disableProbeCheck = disableProbeCheck | |
| self.disableStateCheck = disableStateCheck | |
| self.quotaLimit = quotaLimit | |
| self.pegasusVersion = pegasusVersion | |
| self.pegasusHome = pegasusHome | |
| self.batchLogPath = "" | |
| self.appScriptPath = "" | |
| self.batchScriptPath = "" | |
| self.nodeFilePath = "" | |
| self.remoteJobIdNumber = "" | |
| self.sshBaseCommand = "" | |
| self.sshIdentityPath = "" | |
| self.removeIdentity = False | |
| if self.remoteBatchSystem == 'SCRIPT': | |
| self.isBatchJob = False | |
| self.instanceDirectory = os.getcwd() | |
| else: | |
| self.isBatchJob = True | |
| self.instanceDirectory = os.path.join(os.getcwd(),self.localJobId,self.instanceId) | |
| if not os.path.isdir(self.instanceDirectory): | |
| os.makedirs(self.instanceDirectory) | |
| self.createSSHIdentityPath() | |
| if not self.sharedUserSpace and self.stageFiles: | |
| epoch = int(time.mktime(datetime.datetime.utcnow().timetuple())) | |
| self.workingDirectory = os.path.join(siteInfo['remoteScratchDirectory'], \ | |
| "%s_%s_%s" % (str(epoch),localJobId,self.instanceId)) | |
| else: | |
| self.workingDirectory = os.getcwd() | |
| def createSSHIdentityPath(self): | |
| exitStatus = 0 | |
| if not self.sshIdentityPath: | |
| updateKnownHosts = False | |
| if self.remoteUser.startswith('USER:'): | |
| keyfamily = self.remoteUser.split(':')[1] | |
| self.remoteUser = self.hubUserName | |
| command = 'genuserpki ' + keyfamily | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| if not exitStatus: | |
| self.sshIdentityPath = stdOutput.strip() | |
| log("IDENTITY = " + self.sshIdentityPath) | |
| updateKnownHosts = True | |
| elif self.remoteUser.startswith('USER'): | |
| self.remoteUser = self.hubUserName | |
| else: | |
| command = 'genuserid' | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| if not exitStatus: | |
| self.sshIdentityPath = stdOutput.strip() | |
| log("IDENTITY = " + self.sshIdentityPath) | |
| updateKnownHosts = True | |
| if self.sshIdentityPath: | |
| self.removeIdentity = True | |
| if not exitStatus: | |
| if updateKnownHosts: | |
| if self.tunnelDesignator == "": | |
| command = 'update-known-hosts' + " " + self.venue | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| else: | |
| if self.gatewayHost != "" and self.localHost != "": | |
| command = 'update-known-hosts' + " \'" + self.gatewayHost + "\'" | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| if not exitStatus: | |
| command = 'update-known-hosts' + " \'" + self.localHost + "\'" | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| if not exitStatus: | |
| if self.tunnelDesignator == "": | |
| self.sshBaseCommand = "ssh -T -x -a -i " + self.sshIdentityPath + " " + \ | |
| self.remoteUser + "@" + self.venue | |
| else: | |
| tunnelAddress,tunnelPort = self.remoteTunnelMonitor.getTunnelAddressPort(self.tunnelDesignator) | |
| self.sshBaseCommand = "ssh -T -x -a -i " + self.sshIdentityPath + " -p " + tunnelPort + " " + \ | |
| self.remoteUser + "@" + tunnelAddress | |
| else: | |
| self.sshBaseCommand = "ssh -T -x -a " + self.remoteUser + "@" + self.venue | |
| def createScripts(self): | |
| exitCode = 0 | |
| scriptFiles = [] | |
| if self.sshBaseCommand != "": | |
| log("workingDirectory " + self.workingDirectory) | |
| if self.remoteBatchSystem == 'PBS': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchPBS(self.localJobId,self.instanceId,appScriptName,self.environment, | |
| self.transferExecutable, | |
| self.executable,self.arguments,self.stdinput, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.wallTime,self.quotaLimit, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'PBS8': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchPBS8(self.localJobId,self.instanceId,appScriptName,self.environment, | |
| self.transferExecutable, | |
| self.executable,self.arguments,self.stdinput, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.wallTime,self.quotaLimit, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'LSF': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchLSF(self.localJobId,self.instanceId,appScriptName,self.environment, | |
| self.transferExecutable, | |
| self.executable,self.arguments,self.stdinput, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn,self.wallTime, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'LL': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchLL(self.localJobId,self.instanceId,appScriptName,self.environment, | |
| self.transferExecutable, | |
| self.executable,self.arguments,self.stdinput, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.wallTime,self.quotaLimit, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'SLURM': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchSLURM(self.localJobId,self.instanceId,appScriptName,self.environment, | |
| self.transferExecutable, | |
| self.executable,self.arguments,self.stdinput, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.wallTime,self.quotaLimit, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'CONDOR': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchCONDOR(self.localJobId,self.instanceId,self.instanceDirectory, | |
| appScriptName,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.wallTime) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'RUNJOB': | |
| remoteAppScript = RemoteRunjobAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable,self.stdinput, | |
| self.arguments,self.environment, | |
| self.isMultiCoreRequest,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchRUNJOB(self.localJobId,self.instanceId,self.instanceDirectory, | |
| appScriptName,self.environment,self.transferExecutable,self.executable,self.arguments, | |
| self.isMultiCoreRequest,self.managerInfo, | |
| self.nCpus,self.ppn,self.wallTime, | |
| self.gridsite, | |
| self.x509SubmitProxy,self.disableProbeCheck,self.disableStateCheck, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'FACTORY': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchFACTORY(self.hubUserName, | |
| self.localJobId,self.instanceId,self.instanceDirectory, | |
| appScriptName,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.wallTime) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'PEGASUS': | |
| if self.isParametric: | |
| if self.instanceId != "": | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| batchLogName = "" | |
| batchScriptName = "" | |
| batchScript = "" | |
| nodeFileName = "" | |
| nodeList = [] | |
| else: | |
| appScriptName = "" | |
| appScript = "" | |
| workflow = RemoteWorkflowPEGASUS(self.hubUserName,self.localJobId, | |
| self.pegasusVersion,self.pegasusHome, | |
| self.pegasusTemplates,self.x509SubmitProxy,self.gridsite, | |
| self.timeHistoryLogs) | |
| batchScriptName,batchScript = workflow.buildWorkflowScript() | |
| batchLogName = "" | |
| nodeFileName = "" | |
| nodeList = [] | |
| else: | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteBatchPEGASUS(self.hubUserName, | |
| self.localJobId,self.instanceId,self.instanceDirectory, | |
| appScriptName, | |
| self.executable,self.arguments, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.x509SubmitProxy,self.gridsite,self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'SCRIPT': | |
| remoteAppScript = RemoteInstantAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.workingDirectory, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteInstantSCRIPT(self.localJobId,self.instanceId,self.workingDirectory,appScriptName, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| elif self.remoteBatchSystem == 'BGSCRIPT': | |
| remoteAppScript = RemoteBatchAppScript(self.localJobId,self.instanceId,self.isParametric, | |
| self.transferExecutable,self.executable, | |
| self.stdinput,self.arguments, | |
| self.useEnvironment,self.environment, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.timeHistoryLogs) | |
| appScriptName,appScript = remoteAppScript.buildAppScript() | |
| remoteBatch = RemoteInstantSCRIPT(self.localJobId,self.instanceId,self.workingDirectory,appScriptName, | |
| self.isMultiCoreRequest,self.siteInfo,self.managerInfo, | |
| self.nNodes,self.ppn, | |
| self.timeHistoryLogs) | |
| batchLogName,batchScriptName,batchScript = remoteBatch.buildBatchScript() | |
| nodeFileName,nodeList = remoteBatch.getBatchNodeList() | |
| if batchLogName != "": | |
| self.batchLogPath = os.path.join(self.instanceDirectory,batchLogName) | |
| if appScriptName != "": | |
| self.appScriptPath = os.path.join(self.instanceDirectory,appScriptName) | |
| if batchScriptName != "": | |
| self.batchScriptPath = os.path.join(self.instanceDirectory,batchScriptName) | |
| if nodeFileName != "": | |
| self.nodeFilePath = os.path.join(self.instanceDirectory,nodeFileName) | |
| jobSubmissionMechanism = self.venueMechanism + self.remoteBatchSystem | |
| self.jobIndex = 1 | |
| if not self.jobIndex in self.jobStatistics: | |
| self.jobStatistics[self.jobIndex] = JobStatistic(self.nCpus) | |
| self.jobStatistics[self.jobIndex]['jobSubmissionMechanism'] = jobSubmissionMechanism | |
| if batchScript != "": | |
| fpBatchScript = open(self.batchScriptPath,'w') | |
| if fpBatchScript: | |
| fpBatchScript.write(batchScript) | |
| fpBatchScript.close() | |
| scriptFiles.append(self.batchScriptPath) | |
| if self.remoteBatchSystem == 'SCRIPT': | |
| os.chmod(self.batchScriptPath,stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH) | |
| elif self.remoteBatchSystem == 'BGSCRIPT': | |
| os.chmod(self.batchScriptPath,stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH) | |
| elif self.remoteBatchSystem == 'RUNJOB': | |
| os.chmod(self.batchScriptPath,stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH) | |
| elif self.remoteBatchSystem == 'PEGASUS': | |
| os.chmod(self.batchScriptPath,stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH) | |
| else: | |
| log("could not open %s for writing" % (self.batchScriptPath)) | |
| sys.stderr.write("could not open %s for writing\n" % (self.batchScriptPath)) | |
| sys.stderr.flush() | |
| self.jobStatistics[0]['exitCode'] = 1 | |
| exitCode = 1 | |
| if appScript != "": | |
| fpAppScript = open(self.appScriptPath,'w') | |
| if fpAppScript: | |
| fpAppScript.write(appScript) | |
| fpAppScript.close() | |
| os.chmod(self.appScriptPath,stat.S_IRWXU|stat.S_IRGRP|stat.S_IXGRP|stat.S_IROTH|stat.S_IXOTH) | |
| scriptFiles.append(self.appScriptPath) | |
| else: | |
| log("could not open %s for writing" % (self.appScriptPath)) | |
| sys.stderr.write("could not open %s for writing\n" % (self.appScriptPath)) | |
| sys.stderr.flush() | |
| self.jobStatistics[0]['exitCode'] = 1 | |
| exitCode = 1 | |
| if len(nodeList) > 0: | |
| fpNodes = open(self.nodeFilePath,'w') | |
| if fpNodes: | |
| fpNodes.write('\n'.join(nodeList)+'\n') | |
| fpNodes.close() | |
| else: | |
| log("could not open %s for writing" % (self.nodeFilePath)) | |
| sys.stderr.write("could not open %s for writing\n" % (self.nodeFilePath)) | |
| sys.stderr.flush() | |
| self.jobStatistics[0]['exitCode'] = 1 | |
| exitCode = 1 | |
| else: | |
| exitCode = 1 | |
| if not exitCode: | |
| self.scriptsCreated = True | |
| return(exitCode,scriptFiles) | |
| def sendFiles(self): | |
| self.filesSent = True | |
| exitStatus = 0 | |
| if self.remoteBatchSystem != 'RUNJOB': | |
| if self.remoteBatchQueue != "": | |
| self.jobStatistics[self.jobIndex]['venue'] = self.remoteBatchQueue + '@' + self.venue | |
| else: | |
| self.jobStatistics[self.jobIndex]['venue'] = self.venue | |
| if not self.sharedUserSpace and self.stageFiles: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| if self.sshBaseCommand != "": | |
| stageInTarPath = os.path.join(self.instanceDirectory,self.stageInTarFile) | |
| command = "cat " + stageInTarPath + " | " + self.sshBaseCommand + \ | |
| " \"" + self.receiveInput + " " + remoteWorkingDirectory + " " + self.timeHistoryLogs['timestampTransferred'] + "\"" | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| log(stdOutput) | |
| else: | |
| exitStatus = 1 | |
| stdOutput = "" | |
| stdError = "" | |
| if exitStatus: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = 11 | |
| elif self.sharedUserSpace and self.stageFiles: | |
| remoteWorkingDirectory = self.instanceDirectory | |
| if self.sshBaseCommand != "": | |
| command = self.sshBaseCommand + " \"" + self.receiveInput + " " + \ | |
| remoteWorkingDirectory + " " + \ | |
| self.timeHistoryLogs['timestampTransferred'] + "\"" | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| log(stdOutput) | |
| else: | |
| exitStatus = 1 | |
| stdOutput = "" | |
| stdError = "" | |
| if exitStatus: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = 11 | |
| if exitStatus: | |
| self.filesSent = False | |
| # if stdOutput != "": | |
| # stdFile = os.path.join(self.instanceDirectory,"%s_%s.stdout" % (self.localJobId,self.instanceId)) | |
| # fpStd = open(stdFile,'a') | |
| # if fpStd: | |
| # fpStd.write(stdOutput) | |
| # fpStd.close() | |
| if stdError != "": | |
| stdFile = os.path.join(self.instanceDirectory,"%s_%s.stderr" % (self.localJobId,self.instanceId)) | |
| fpStd = open(stdFile,'a') | |
| if fpStd: | |
| fpStd.write(stdError) | |
| fpStd.close() | |
| return(self.filesSent) | |
| def executeJob(self): | |
| exitStatus = 0 | |
| if self.remoteBatchSystem == 'SCRIPT': | |
| if self.stageFiles: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| commandArgs = [] | |
| commandArgs.append(self.submitBatchJob) | |
| commandArgs.append(remoteWorkingDirectory) | |
| if self.batchScriptPath != "": | |
| commandArgs.append(os.path.join('.',os.path.basename(self.batchScriptPath))) | |
| else: | |
| commandArgs.append(os.path.join('.',os.path.basename(self.appScriptPath))) | |
| if self.logUserRemotely: | |
| commandArgs.append(self.hubUserName) | |
| commandArgs.append(str(self.hubUserId)) | |
| commandArgs.append(self.localJobId.lstrip('0') + '_' + self.instanceId) | |
| commandArgs.append(self.executable) | |
| command = self.sshBaseCommand + " \"" + ' '.join(commandArgs) + "\"" | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeSSHCommand(command,self.tunnelDesignator,True) | |
| self.jobSubmitted = True | |
| if self.cloudDesignator != "": | |
| self.remoteCloudMonitor.decrementCloudInstance(self.cloudDesignator,self.cloudInstanceId, \ | |
| "%s_%s" % (self.localJobId,self.instanceId)) | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| else: | |
| if self.batchScriptPath != "": | |
| command = self.sshBaseCommand + " " + self.batchScriptPath | |
| else: | |
| command = self.sshBaseCommand + " " + self.appScriptPath | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command,True) | |
| self.jobSubmitted = True | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| elif self.remoteBatchSystem == 'BGSCRIPT': | |
| if self.stageFiles: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| commandArgs = [] | |
| commandArgs.append(self.submitBatchJob) | |
| commandArgs.append(remoteWorkingDirectory) | |
| if self.batchScriptPath != "": | |
| commandArgs.append(os.path.join('.',os.path.basename(self.batchScriptPath))) | |
| else: | |
| commandArgs.append(os.path.join('.',os.path.basename(self.appScriptPath))) | |
| if self.logUserRemotely: | |
| commandArgs.append(self.hubUserName) | |
| commandArgs.append(str(self.hubUserId)) | |
| commandArgs.append(self.localJobId.lstrip('0') + '_' + self.instanceId) | |
| commandArgs.append(self.executable) | |
| command = self.sshBaseCommand + " \"" + ' '.join(commandArgs) + "\"" | |
| log("command = " + command) | |
| exitStatus,remoteJobId,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| if not exitStatus: | |
| remoteJobId = remoteJobId.strip() | |
| log("remoteJobId = " + remoteJobId) | |
| self.jobSubmitted = True | |
| self.remoteJobIdNumber = remoteJobId | |
| self.jobStatistics[self.jobIndex]['remoteJobIdNumber'] = self.remoteJobIdNumber | |
| self.remoteJobMonitor.postNewJobSubmission(self.siteMonitorDesignator,self.remoteJobIdNumber, | |
| str(self.hubUserId), | |
| "%s_%s" % (self.localJobId,self.instanceId),self.destination) | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| else: | |
| if self.stageFiles: | |
| if self.sharedUserSpace: | |
| remoteWorkingDirectory = self.instanceDirectory | |
| else: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| commandArgs = [] | |
| commandArgs.append(self.submitBatchJob) | |
| commandArgs.append(remoteWorkingDirectory) | |
| commandArgs.append(os.path.join('.',os.path.basename(self.batchScriptPath))) | |
| if self.logUserRemotely: | |
| commandArgs.append(self.hubUserName) | |
| commandArgs.append(str(self.hubUserId)) | |
| commandArgs.append(self.localJobId.lstrip('0') + '_' + self.instanceId) | |
| commandArgs.append(self.executable) | |
| command = self.sshBaseCommand + " \"" + ' '.join(commandArgs) + "\"" | |
| log("command = " + command) | |
| exitStatus,remoteJobId,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| if not exitStatus: | |
| remoteJobId = remoteJobId.strip() | |
| log("remoteJobId = " + remoteJobId) | |
| try: | |
| if self.remoteBatchSystem == 'PBS': | |
| self.remoteJobIdNumber = remoteJobId.split('.')[0] | |
| elif self.remoteBatchSystem == 'PBS8': | |
| self.remoteJobIdNumber = remoteJobId.split('.')[0] | |
| elif self.remoteBatchSystem == 'LSF': | |
| # Job <851126> is submitted to default queue <normal>. | |
| self.remoteJobIdNumber = remoteJobId.split('<')[1].split('>')[0] | |
| elif self.remoteBatchSystem == 'LL': | |
| self.remoteJobIdNumber = remoteJobId | |
| elif self.remoteBatchSystem == 'SLURM': | |
| self.remoteJobIdNumber = remoteJobId.split()[-1] | |
| elif self.remoteBatchSystem == 'CONDOR': | |
| # Submitting job(s). Logging submit event(s). 1 job(s) submitted to cluster 469609. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| elif self.remoteBatchSystem == 'RUNJOB': | |
| # Your job has been submitted with id 19.0 | |
| self.remoteJobIdNumber = re.search('with id [0-9]+',remoteJobId).group().split()[2] + ".0" | |
| elif self.remoteBatchSystem == 'FACTORY': | |
| # Submitting job(s). Logging submit event(s). 1 job(s) submitted to cluster 105. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| elif self.remoteBatchSystem == 'PEGASUS': | |
| # 1 job(s) submitted to cluster 105. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| self.jobStatistics[self.jobIndex]['remoteJobIdNumber'] = self.remoteJobIdNumber | |
| except: | |
| exitStatus = 248 | |
| if not exitStatus: | |
| self.jobSubmitted = True | |
| if self.remoteBatchSystem == 'PEGASUS': | |
| self.remoteJobMonitor.postNewWorkflowSubmission(self.siteMonitorDesignator,self.remoteJobIdNumber, | |
| str(self.hubUserId),self.enteredCommand, | |
| self.localJobId,self.nInstances,self.destination) | |
| else: | |
| self.remoteJobMonitor.postNewJobSubmission(self.siteMonitorDesignator,self.remoteJobIdNumber, | |
| str(self.hubUserId), | |
| "%s_%s" % (self.localJobId,self.instanceId),self.destination) | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| else: | |
| if self.sharedUserSpace: | |
| remoteWorkingDirectory = self.instanceDirectory | |
| else: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| commandArgs = [] | |
| commandArgs.append(self.submitBatchJob) | |
| commandArgs.append(remoteWorkingDirectory) | |
| commandArgs.append(os.path.join('.',os.path.basename(self.batchScriptPath))) | |
| command = self.sshBaseCommand + " \"" + ' '.join(commandArgs) + "\"" | |
| log("command = " + command) | |
| exitStatus,remoteJobId,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| if not exitStatus: | |
| remoteJobId = remoteJobId.strip() | |
| log("remoteJobId = " + remoteJobId) | |
| try: | |
| if self.remoteBatchSystem == 'PBS': | |
| self.remoteJobIdNumber = remoteJobId.split('.')[0] | |
| elif self.remoteBatchSystem == 'PBS8': | |
| self.remoteJobIdNumber = remoteJobId.split('.')[0] | |
| elif self.remoteBatchSystem == 'LSF': | |
| # Job <851126> is submitted to default queue <normal>. | |
| self.remoteJobIdNumber = remoteJobId.split('<')[1].split('>')[0] | |
| elif self.remoteBatchSystem == 'LL': | |
| self.remoteJobIdNumber = remoteJobId | |
| elif self.remoteBatchSystem == 'SLURM': | |
| self.remoteJobIdNumber = remoteJobId.split()[-1] | |
| elif self.remoteBatchSystem == 'CONDOR': | |
| # Submitting job(s). Logging submit event(s). 1 job(s) submitted to cluster 469609. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| elif self.remoteBatchSystem == 'RUNJOB': | |
| # Your job has been submitted with id 19.0 | |
| self.remoteJobIdNumber = re.search('with id [0-9]+',remoteJobId).group().split()[2] + ".0" | |
| elif self.remoteBatchSystem == 'FACTORY': | |
| # Submitting job(s). Logging submit event(s). 1 job(s) submitted to cluster 105. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| elif self.remoteBatchSystem == 'PEGASUS': | |
| # 1 job(s) submitted to cluster 105. | |
| self.remoteJobIdNumber = re.search('cluster [0-9]+',remoteJobId).group().split()[1] + ".0" | |
| self.jobStatistics[self.jobIndex]['remoteJobIdNumber'] = self.remoteJobIdNumber | |
| except: | |
| exitStatus = 248 | |
| if not exitStatus: | |
| self.jobSubmitted = True | |
| if self.remoteBatchSystem == 'PEGASUS': | |
| self.remoteJobMonitor.postNewWorkflowSubmission(self.siteMonitorDesignator,self.remoteJobIdNumber, | |
| str(self.hubUserId),self.enteredCommand, | |
| self.localJobId,self.nInstances,self.destination) | |
| else: | |
| self.remoteJobMonitor.postNewJobSubmission(self.siteMonitorDesignator,self.remoteJobIdNumber, | |
| str(self.hubUserId), \ | |
| "%s_%s" % (self.localJobId,self.instanceId),self.destination) | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| else: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = exitStatus | |
| if not self.jobSubmitted: | |
| if remoteJobId != "": | |
| stdFile = os.path.join(self.instanceDirectory,"%s_%s.stdout" % (self.localJobId,self.instanceId)) | |
| fpStd = open(stdFile,'a') | |
| if fpStd: | |
| fpStd.write(remoteJobId) | |
| fpStd.close() | |
| if stdError != "": | |
| stdFile = os.path.join(self.instanceDirectory,"%s_%s.stderr" % (self.localJobId,self.instanceId)) | |
| fpStd = open(stdFile,'a') | |
| if fpStd: | |
| fpStd.write(stdError) | |
| fpStd.close() | |
| return(self.jobSubmitted) | |
| def getWaitForJobInfo(self): | |
| waitForJobInfo = {} | |
| waitForJobInfo['isBatchJob'] = self.isBatchJob | |
| waitForJobInfo['siteMonitorDesignator'] = self.siteMonitorDesignator | |
| waitForJobInfo['remoteJobId'] = self.remoteJobIdNumber | |
| waitForJobInfo['knownSite'] = self.destination | |
| return(waitForJobInfo) | |
| def waitForJob(self): | |
| if (self.siteMonitorDesignator != "") and \ | |
| (self.remoteJobIdNumber != ""): | |
| if self.remoteBatchSystem == 'RUNJOB': | |
| self.remoteJobMonitor.waitForBatchJob(self.siteMonitorDesignator,self.remoteJobIdNumber) | |
| else: | |
| if self.destination != "": | |
| self.remoteJobMonitor.waitForBatchJob(self.siteMonitorDesignator,self.remoteJobIdNumber,self.destination) | |
| def retrieveFiles(self): | |
| exitStatus = 0 | |
| if self.remoteBatchSystem != 'RUNJOB': | |
| if not self.sharedUserSpace and self.stageFiles: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| if self.jobSubmitted: | |
| command = self.sshBaseCommand + " \"" + self.transmitResults + " " + remoteWorkingDirectory + "\"" + \ | |
| " | tar xzmf - --ignore-case --exclude '*hub-proxy.*' -C " + self.instanceDirectory | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeSSHCommand(command,self.tunnelDesignator) | |
| log(stdOutput) | |
| if exitStatus: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = 12 | |
| # if stdOutput != "": | |
| # stdFile = os.path.join(self.instanceDirectory,"%s_%s.stdout" % (self.localJobId,self.instanceId)) | |
| # fpStd = open(stdFile,'a') | |
| # if fpStd: | |
| # fpStd.write(stdOutput) | |
| # fpStd.close() | |
| if stdError != "": | |
| stdFile = os.path.join(self.instanceDirectory,"%s_%s.stderr" % (self.localJobId,self.instanceId)) | |
| fpStd = open(stdFile,'a') | |
| if fpStd: | |
| fpStd.write(stdError) | |
| fpStd.close() | |
| elif self.sharedUserSpace and self.stageFiles: | |
| outTarFile = "%s_%s_output.tar" % (self.localJobId,self.instanceId) | |
| stageOutTarFiles = [] | |
| stageOutTarFiles.append(os.path.join(self.instanceDirectory,"%s.gz" % (outTarFile))) | |
| stageOutTarFiles.append(os.path.join(self.instanceDirectory,outTarFile)) | |
| stageOutTarFiles.append("%s.gz" % (outTarFile)) | |
| stageOutTarFiles.append(outTarFile) | |
| stageOutTarFiles.append(os.path.join(os.path.dirname(self.instanceDirectory),"%s.gz" % (outTarFile))) | |
| stageOutTarFiles.append(os.path.join(os.path.dirname(self.instanceDirectory),outTarFile)) | |
| for stageOutTarFile in stageOutTarFiles: | |
| if os.path.isfile(stageOutTarFile): | |
| if stageOutTarFile.endswith('.gz'): | |
| command = "tar xzmf " + stageOutTarFile + \ | |
| " --ignore-case --exclude '*hub-proxy.*' -C " + self.instanceDirectory | |
| else: | |
| command = "tar xmf " + stageOutTarFile + \ | |
| " --ignore-case --exclude '*hub-proxy.*' -C " + self.instanceDirectory | |
| log("command = " + command) | |
| exitStatus,stdOutput,stdError = self.executeCommand(command) | |
| if exitStatus == 0: | |
| os.remove(stageOutTarFile) | |
| else: | |
| log(stdOutput) | |
| break | |
| if exitStatus: | |
| self.jobStatistics[self.jobIndex]['exitCode'] = 12 | |
| # if stdOutput != "": | |
| # stdFile = os.path.join(self.instanceDirectory,"%s_%s.stdout" % (self.localJobId,self.instanceId)) | |
| # fpStd = open(stdFile,'a') | |
| # if fpStd: | |
| # fpStd.write(stdOutput) | |
| # fpStd.close() | |
| if stdError != "": | |
| stdFile = os.path.join(self.instanceDirectory,"%s_%s.stderr" % (self.localJobId,self.instanceId)) | |
| fpStd = open(stdFile,'a') | |
| if fpStd: | |
| fpStd.write(stdError) | |
| fpStd.close() | |
| if not exitStatus: | |
| self.filesRetrieved = True | |
| def cleanupFiles(self): | |
| if not self.filesCleanedup: | |
| if self.remoteBatchSystem == 'RUNJOB': | |
| workDirectoryName = "runjob-workdir-%s_%s" % (self.localJobId,self.instanceId) | |
| workDirectoryName = os.path.join(self.instanceDirectory,workDirectoryName) | |
| if os.path.isdir(workDirectoryName): | |
| command = "cleanupkilled.sh " + workDirectoryName + " " + self.timeHistoryLogs['jobGridJobId'] | |
| log("command = " + command) | |
| log(self.executeCommand(command)[1]) | |
| if self.jobSubmitted: | |
| self.recordGridResourceUse() | |
| self.logGridJobId() | |
| self.logGridHistories() | |
| if os.path.isdir(workDirectoryName): | |
| shutil.rmtree(workDirectoryName,True) | |
| elif self.remoteBatchSystem == 'PEGASUS': | |
| workDirectoryName = os.path.join(self.instanceDirectory,'work') | |
| if self.jobSubmitted and os.path.isdir(workDirectoryName): | |
| if not self.sharedUserSpace: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| else: | |
| remoteWorkingDirectory = self.instanceDirectory | |
| command = self.sshBaseCommand + " \"" + self.cleanupJob + " " + remoteWorkingDirectory + "\"" | |
| log("command = " + command) | |
| log(self.executeSSHCommand(command,self.tunnelDesignator)[1]) | |
| if self.isParametric: | |
| getPegasusStdTimeFiles(self.instanceDirectory,self.timeHistoryLogs['timeResults']) | |
| else: | |
| processPegasusFiles(self.instanceDirectory,self.timeHistoryLogs['timeResults']) | |
| # if os.path.isdir(workDirectoryName): | |
| # shutil.rmtree(workDirectoryName,True) | |
| inProcessDirectoryName = os.path.join(self.instanceDirectory,'InProcessResults') | |
| if os.path.isdir(inProcessDirectoryName): | |
| shutil.rmtree(inProcessDirectoryName,True) | |
| for ftype in '.dax','_sites.xml','.pegasusrc','_tc.txt': | |
| pegasusFile = os.path.join(self.instanceDirectory,"%s_%s%s" % (self.localJobId,self.instanceId,ftype)) | |
| if os.path.isfile(pegasusFile): | |
| os.remove(pegasusFile) | |
| else: | |
| pegasusFile = os.path.join(self.instanceDirectory,"%s%s" % (self.localJobId,ftype)) | |
| if os.path.isfile(pegasusFile): | |
| os.remove(pegasusFile) | |
| pegasusFile = os.path.join(self.instanceDirectory,"%s_%s.stderr" % (self.localJobId,self.instanceId)) | |
| if(os.path.exists(pegasusFile)): | |
| if(os.path.getsize(pegasusFile) == 0): | |
| os.remove(pegasusFile) | |
| stageInTarFile = os.path.join(self.instanceDirectory,self.stageInTarFile) | |
| if os.path.isfile(stageInTarFile): | |
| os.remove(stageInTarFile) | |
| else: | |
| if not self.sharedUserSpace and self.stageFiles: | |
| remoteWorkingDirectory = self.workingDirectory.replace("$","\$") | |
| if self.jobSubmitted: | |
| command = self.sshBaseCommand + " \"" + self.cleanupJob + " " + remoteWorkingDirectory + "\"" | |
| log("command = " + command) | |
| log(self.executeSSHCommand(command,self.tunnelDesignator)[1]) | |
| elif self.sharedUserSpace and self.stageFiles: | |
| stageOutTarFile = "%s_%s_output.tar.gz" % (self.localJobId,self.instanceId) | |
| if os.path.isfile(stageOutTarFile): | |
| os.remove(stageOutTarFile) | |
| else: | |
| stageOutTarFile = "%s_%s_output.tar" % (self.localJobId,self.instanceId) | |
| if os.path.isfile(stageOutTarFile): | |
| os.remove(stageOutTarFile) | |
| if self.removeIdentity: | |
| if os.path.isfile(self.sshIdentityPath): | |
| os.remove(self.sshIdentityPath) | |
| if self.batchScriptPath: | |
| if os.path.isfile(self.batchScriptPath): | |
| os.remove(self.batchScriptPath) | |
| if self.appScriptPath: | |
| if os.path.isfile(self.appScriptPath): | |
| os.remove(self.appScriptPath) | |
| if self.nodeFilePath: | |
| if os.path.isfile(self.nodeFilePath): | |
| os.remove(self.nodeFilePath) | |
| if self.batchLogPath: | |
| if os.path.isfile(self.batchLogPath): | |
| os.remove(self.batchLogPath) | |
| if self.cloudDesignator != "": | |
| self.remoteCloudMonitor.decrementCloudInstance(self.cloudDesignator,self.cloudInstanceId, \ | |
| "%s_%s" % (self.localJobId,self.instanceId)) | |
| self.filesCleanedup = True | |
| def killScripts(self, | |
| signalNumber): | |
| if self.remoteJobIdNumber: | |
| if self.remoteBatchSystem == 'RUNJOB': | |
| self.logGridHistories() | |
| command = self.sshBaseCommand + \ | |
| " \"" + self.killBatchJob + " " + self.remoteJobIdNumber + " " + self.remoteBatchSystem + "\"" | |
| log("command = " + command) | |
| log(self.executeSSHCommand(command,self.tunnelDesignator)[1]) | |
| elif self.remoteBatchSystem == 'SCRIPT' and self.childPid != 0: | |
| # log("Send TERM to child script process %d" % (self.childPid)) | |
| # os.kill(childPid,signalNumber.SIGTERM) | |
| if self.batchScriptPath != "": | |
| command = self.sshBaseCommand + " pkill -TERM -f " + os.path.basename(self.batchScriptPath) | |
| else: | |
| command = self.sshBaseCommand + " pkill -TERM -f " + os.path.basename(self.appScriptPath) | |
| log("command = " + command) | |
| log(self.executeSSHCommand(command,self.tunnelDesignator)[1]) | |
| if self.batchScriptPath != "": | |
| command = self.sshBaseCommand + " pkill -0 -f " + os.path.basename(self.batchScriptPath) | |
| else: | |
| command = self.sshBaseCommand + " pkill -0 -f " + os.path.basename(self.appScriptPath) | |
| log("command = " + command) | |
| statusCode = self.executeSSHCommand(command,self.tunnelDesignator)[0] | |
| while statusCode == 0: | |
| time.sleep(2) | |
| statusCode = self.executeSSHCommand(command,self.tunnelDesignator)[0] | |
| maximumJobIndex = max(self.jobStatistics.keys()) | |
| self.jobStatistics[maximumJobIndex]['exitCode'] = 1 << 8 | signalNumber | |
| self.scriptsKilled = True | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment