class JobRecovery:
    """
    Job recovery
    Specialized algorithm for recovering previously failed stage-out transfers
    The algorithm works by scanning all panda job directories on the work node, or in any provided external (remote or at least outside of
    the pilot work directory) directory as specified in schedconfig.recoverdir, for job state files.
    The algorithm only tries to recover jobs whose job state file (jobState-<jobId>.[pickle|json]) has not been modified for over two
    heartbeat periods. A stranded job that is found by the algorithm is referred to as a lost job. The last action
    performed by the pilot before the lost job failed is stored in the job state file variable "state". Only lost jobs
    that are in states "stageout", "finished", "failed" or "holding" will be recovered. A job can have the following
    job states:
       "startup"      : First update to the job state file
       "setup"        : Pilot is setting up the runtime options
       "stagein"      : Stage-in of input files
       "running"      : Payload execution; output files are created (no point in recovery if job found in this state)
       "stageout"     : Stage-out of output files
       "finished"     : Last update to the job state file (alternative)
       "failed"       : Last update to the job state file (alternative)
       "holding"      : Last update to the job state file (alternative)
    The job state file contains a dictionary of log and output files and the state at the last time of the update. The
    format of this file dictionary is: { filename1 : state1, filename2 : state2, ... }
    where the state variables have the list form "file_state", "reg_state".
    "file_state" can assume the following values:
      "not_created"   : initial value for all files at the beginning of the job
      "created"       : file was created and is waiting to be transferred
      "transferred"   : file has already been transferred (no further action)
      "missing"       : file was never created, the job failed (e.g. output file of a failed job; a log should never be missing)
    "reg_state" can assume the following values:
      "not_registered": file was not registered in the catalog (might not matter if the server does the registration)
      "registered"    : file was already registered in the catalog (no further action)
    E.g. a file with state = "created", "not_registered" should first be transferred and then registered in the LFC (assuming that the pilot
    is responsible for LFC registrations)
    """

    # private data members
    __error = PilotErrors()                # PilotErrors object
    __errorLabel = "WARNING"               # Error string label
    __errorString = "!!WARNING!!1999!! %s" # Default error string
    __heartbeatPeriod = 30*60              # Heartbeat period in seconds
    __initDir = ""                         # Job recovery init dir
    __jobId = ""                           # Current job id
    __jobSchedulerId = ""                  # Job scheduler id
    __logFileDir = ""                      # Location of the log file
    __maxJobRec = 20                       # Default maximum number of job recoveries
    __maxNumberOfRecoveryAttempts = 15     # Max recovery attempts
    __outputDir = ""                       # Special output directory (Nordugrid, CERNVM)
    __pilot_version = ""                   # Pilot version string
    __pilot_version_tag = ""               # PR/RC    
    __pilot_initdir = ""                   # Pilot init dir
    __pilotErrorDiag = ""                  # Error diagnostics string
    __pilotId = ""                         # Pilot id
    __psport = 25443                       # Default panda server port
    __pshttpurl = ""                       # Job dispatcher
    __proxycheckFlag = True                # Should the proxy be validated?
    __recovered = ""                       # Comma separated string of all recovered job id's (even failed attempted recoveries)
    __recoveryAttempt = 0                  # Recovery attempt number
    __recoveryDir = ""                     # Current recovery dir
    __recoveryDirs = []                    # List of all base dirs to be scanned for lost jobs
    __site = None                          # Site object
    __testLevel = "0"                      # Debugging info (remove)
    __uflag = None                         # 'user job flag' if this pilot is to get a user analysis job or ordinary job
    __updateServer = True                  # Should the server be updated?
    __workNode = None                      # Node object

    __PASSED_INIT = False                  # Init boolean

    # Main file dictionary definition
    fileStateDictionary = {}

    def __init__(self,
                 errorLabel="WARNING",
                 heartbeat=30*60,
                 jobSchedulerId="",
                 logFileDir="",
                 maxjobrec=20,
                 maxNumberOfRecoveryAttempts=15,
                 outputDir="",
                 pilot_version="",
                 pilot_version_tag="",
                 pilot_initdir=None,
                 pilotId="",
                 probe=False,
                 pshttpurl=None,
                 psport=None,
                 site=None,
                 testLevel="0",
                 uflag=None,
                 updateServer=True,
                 workNode=None,
                 *args, **kwrds):
        """ Default initialization """

        self.__errorLabel = errorLabel
        self.__heartbeatPeriod = heartbeat
        self.__initDir = os.getcwd()
        self.__jobSchedulerId = jobSchedulerId
        self.__logFileDir = logFileDir
        self.__maxJobRec = maxjobrec
        self.__maxNumberOfRecoveryAttempts = maxNumberOfRecoveryAttempts
        self.__outputDir = outputDir
        self.__pilot_version = pilot_version
        self.__pilot_version_tag = pilot_version_tag
        self.__pilot_initdir = pilot_initdir
        self.__pilotId = pilotId
        self.__pshttpurl = pshttpurl
        self.__psport = psport
        self.__site = site
        self.__testLevel = testLevel
        self.__uflag = uflag
        self.__updateServer = updateServer
        self.__workNode = workNode

        # set all recovery directories
        if probe:
            self.setRecoveryDirs()
            if self.__recoveryDirs == []:
                self.__pilotErrorDiag = "Job recovery initialization failed"
                tolog(self.__errorString % self.__pilotErrorDiag)
                self.__PASSED_INIT = False
            else:
                self.__PASSED_INIT = True
        else:
            self.__PASSED_INIT = True

    def setRecoveryDirs(self):
        """ Set the recovery directories """

        dirs = []
        if self.__site:
            _dir = self.__site.wntmpdir
        else:
            _dir = ""
        if _dir == "":
            if os.environ.has_key('TMPDIR'):
                _dir = os.environ['TMPDIR']
            elif os.environ.has_key('OSG_WN_TMP'):
                _dir = os.environ['OSG_WN_TMP']
            elif os.path.exists("/tmp"):
                _dir = "/tmp"
            elif os.path.exists("/scratch"):
                _dir = "/scratch"
            else:
                self.__pilotErrorDiag = "Could not locate any scratch dirs"
                tolog(self.__errorString % self.__pilotErrorDiag)
                _dir = ""
        if _dir != "":
            dirs.append(_dir)

        extradir = readpar('wntmpdir')
        if extradir != "" and extradir != "None" and extradir not in dirs:
            dirs.append(extradir)

        # check queuedata for external recovery directory
        # an empty externalRecoveryDir means that recovery should only search local WN disk for lost jobs
        # make sure the recovery directory actually exists (will not be added to dir list if empty)
        externalRecoveryDir = self.verifyRecoveryDir(readpar('recoverdir'))
        if externalRecoveryDir != "":
            dirs.append(externalRecoveryDir)

        if dirs != []:
            tolog("Job recovery will probe: %s" % str(dirs))
            self.__recoveryDirs = dirs
        else:
            self.__pilotErrorDiag = "Did not identify any base recovery directories"
            tolog(self.__errorString % self.__pilotErrorDiag)
            
    def setRecoveryDir(self, dir):
        """ Set the current recovery dir """
        self.__recoveryDir = dir

    def getRecoveryDir(self):
        """ Return the current recovery dir """
        return self.__recoveryDir

    def getInitDir(self):
        """ Return the init dir """
        return self.__initDir

    def createAtomicLockFile(self, file_path):
        """ Create an atomic lockfile while probing this dir to avoid a possible race-condition """

        lockfile_name = os.path.join(os.path.dirname(file_path), "ATOMIC_LOCKFILE")
        try:
            # acquire the lock
            fd = os.open(lockfile_name, os.O_EXCL|os.O_CREAT)
        except OSError:
            # work dir is locked, so exit
            tolog("Found lock file: %s (skip this dir)" % (lockfile_name))
            fd = None
        else:
            tolog("Created lock file: %s" % (lockfile_name))
        return fd, lockfile_name

    def releaseAtomicLockFile(self, fd, lockfile_name):
        """ Release the atomic lock file """

        try:
            os.close(fd)
            os.unlink(lockfile_name)
        except Exception, e:
            if "Bad file descriptor" in e:
                tolog("Lock file already released")
            else:
                self.__pilotErrorDiag = "Could not release lock file: %s" % (e)
                tolog(self.__errorString % self.__pilotErrorDiag)
        else:
            tolog("Released lock file: %s" % (lockfile_name))

    def verifyRecoveryDir(self, recoverdir):
        """ Make sure that the recovery directory actually exists, return empty string otherwise """

        status = False
        # does the recovery dir actually exists?
        if recoverdir and recoverdir != "":
            if os.path.exists(recoverydir):
                tolog("External recovery dir verified: %s" % (recoverdir))
                status = True
        if not status:
            recoverdir = ""

        return recoverdir

    def updatePandaServer(self, job, site, workerNode, port, xmlstr=None, spaceReport=False, log=None, ra=0, jr=False, schedulerID=None, pilotID=None, updateServer=True):
        """ Update the PanDA server """

        # create and instantiate the client object
        from PandaServerClient import PandaServerClient
        client = PandaServerClient(pilot_version=self.__pilot_version, \
                                   pilot_version_tag=self.__pilot_version_tag, \
                                   pilot_initdir=self.__pilot_initdir, \
                                   jobSchedulerId=self.__jobSchedulerId, \
                                   pilotId=self.__pilotId, \
                                   updateServer=self.__updateServer, \
                                   jobrec=True, \
                                   pshttpurl=self.__pshttpurl)

        # update the panda server
        return client.updatePandaServer(job, site, workerNode, port, xmlstr=xmlstr, spaceReport=spaceReport, log=log, ra=ra, jr=jr)

    def performRecovery(self):
        """ Run job recovery in all directories """

        status = False

        if not self.__PASSED_INIT:
            self.__pilotErrorDiag = "Aborting job recovery due to previous failure"
            tolog(self.__errorString % self.__pilotErrorDiag)
            return status

        # keep track of the starting dir
        startingDir = os.getcwd()

        # scan all recovery dirs
        JS = JobState()
        for _dir in self.__recoveryDirs:
            status = self.run(_dir, JS)

        # return to the starting dir
        chdir(startingDir)

        return status

    def run(self, _dir, JS):
        """ Run main job recovery algorithm """

        status = False
        tolog("Running job recovery in: %s" % (_dir))

        # grab all job state files in all work directories
        job_state_files_pickle = glob(os.path.join(_dir, "/Panda_Pilot_*/jobState-*.pickle"))
        job_state_files_json = glob(os.path.join(_dir, "/Panda_Pilot_*/jobState-*.json"))
        tolog("Number of found job state pickle files: %d" % (len(job_state_files_pickle)))
        tolog("Number of found job state json files: %d" % (len(job_state_files_json)))

        _rec_nr = 0
        job_state_files = job_state_files_json + job_state_files_pickle
        if not job_state_files:
            return True

        file_nr = 0
        # loop over all found job state files
        for job_state_file in job_state_files:

            # reset status for every file
            status = False

            # create an atomic lockfile while probing this dir to avoid a possible race-condition
            fd, lockfile_name = self.createAtomicLockFile(job_state_file)
            if not fd:
                # set status to True in case this was the last file
                status = True
                continue

            dirname = os.path.dirname(job_state_file)
            tolog("Probing dir: %s" % (dirname))

            # make sure the LOCKFILE is present
            lockFile = os.path.join(dirname, 'LOCKFILE')
            if not os.path.exists(lockFile):
                # release the atomic lockfile and go to the next directory
                self.releaseAtomicLockFile(fd, lockfile_name)

                # set status to True in case this was the last file
                status = True
                continue
            else:
                try:
                    tolog("Found %s created at %d" % (lockFile, os.path.getmtime(lockFile)))
                except Exception, e:
                    tolog(self.__errorString % "Could not read modification time of %s, try to continue: %s" % (lockFile, e))

            if (file_nr + 1) > self.__maxJobRec:
                tolog("Maximum number of job recoveries exceeded for this pilot: %d" % (self.__maxJobRec))
                # release the atomic lockfile and go to the next directory
                self.releaseAtomicLockFile(fd, lockfile_name)

                # set status to True in case this was the last file
                status = True
                break

            tolog("Processing job state file %d/%d: %s" % (file_nr + 1, len(job_state_files), job_state_file))
            current_time = int(time.time())

            # when was file last modified?
            try:
                file_modification_time = os.path.getmtime(job_state_file)
            except Exception, e:
                # skip this file since it was not possible to read the modification time
                tolog(self.__errorString % "Could not read modification time: %s" % (e))

                # release the atomic lockfile and go to the next directory
                self.releaseAtomicLockFile(fd, lockfile_name)

                # do not increase file counter, keep status = False
            else:
                # was the job state file updated longer than 2 heart beats ago?
                if (current_time - file_modification_time) > 2*self.__heartbeatPeriod:
                    # found lost job recovery file
                    tolog("File was last modified %d seconds ago (limit=%d, t=%d, tmod=%d)" %\
                          (current_time - file_modification_time, 2*self.__heartbeatPeriod, current_time, file_modification_time))

                    # open the job state file
                    if JS.get(job_state_file):
                        # decode the job state info
                        _job, _site, _node, self.__recoveryAttempt = JS.decode()

                        # set the jobId number
                        self.__jobId = _job.jobId

                    if not (_job and _site and _node):
                        tolog("Missing data object (job, site or node not defined)")
                        # release the atomic lockfile and go to the next directory
                        self.releaseAtomicLockFile(fd, lockfile_name)

                        # do not increase file counter, keep status = False
                        continue

                    # only continue if current pilot is of same type as lost job (to prevent potential permission problems)
                    if not self.isSameType(_job.trf.split(",")[0], self.__uflag):
                        # release the atomic lockfile and go to the next directory
                        self.releaseAtomicLockFile(fd, lockfile_name)
                        status = True
                        continue

                    # uncomment this code for recovery of certain panda ids only
                    # allowedJobIds = ['1018129684','1018129685']
                    # if _job.jobId not in allowedJobIds:
                    #     tolog("Job id %s not in allowed id list" % (_job.jobId))
                    #     continue

                    # query the job state file for job information
                    if not self.verifyJobInfo(_job, _site, JS):
                        # release the atomic lockfile and go to the next directory
                        self.releaseAtomicLockFile(fd, lockfile_name)
                        continue

                    tolog("Job %s is currently in state \'%s\' (according to job state file - recover)" %\
                          (_job.jobId, _job.result[0]))

                    # switch to the work dir of the old pilot
                    chdir(_site.workdir)

                    # abort if max number of recovery attempts has been exceeded
                    if self.__recoveryAttempt > self.__maxNumberOfRecoveryAttempts - 1:
                        self.__pilotErrorDiag = "Max number of recovery attempts exceeded: %d" % (self.__maxNumberOfRecoveryAttempts)
                        tolog(self.__errorString % self.__pilotErrorDiag)
                        _job.setState(['failed', _job.result[1], self.__error.ERR_LOSTJOBMAXEDOUT])

                        # update the server
                        rt = self.updateServer(_job, _site, JS)
                        if rt == 0:
                            number_of_recoveries += 1

                        # release the atomic lockfile and go to the next directory
                        self.releaseAtomicLockFile(fd, lockfile_name)
                        file_nr += 1
                        continue
                    else:
                        # increase recovery attempt counter
                        self.__recoveryAttempt += 1

                    tolog("Recovery attempt: %d" % (self.__recoveryAttempt))

                    # update job state file at this point to prevent a parallel pilot from doing a simultaneous recovery
                    # (note: should not really be necessary due to the atomic lock file)
                    _retjs = updateJobState(_job, _site, _node, _recoveryAttempt)

####


                # Job state file was recently modified
                else:
                    tolog("(Job state file was recently modified - skip it)")
                    # atomic lockfile will be released below

            # (end main "for job_state_file in job_state_files"-loop)
            # release the atomic lockfile and go to the next directory
            self.releaseAtomicLockFile(fd, lockfile_name)

            # increase file counter
            file_nr += 1

        return status

    def updateServer(self, _job, _site, JS):
        """ Update the Panda server """

        rt, retNode = self.updatePandaServer(_job, _site, self.__workNode, self.__psport, ra=self.__recoveryAttempt)
        if rt == 0:
            tolog("Lost job %s updated (exit code %d)" % (_job.jobId, _job.result[2]))

            # did the server send back a command?
            if "tobekilled" in _job.action:
                self.__pilotErrorDiag = "Panda server returned a \'tobekilled\' command"
                tolog(self.__errorString % self.__pilotErrorDiag)

            if _job.result[0] == "failed":
                tolog("NOTE: This job has been terminated. Will now remove workdir.")
                if not JS.cleanup():
                    self.__pilotErrorDiag = "failed to cleanup"
                    tolog(self.__errorString % self.__pilotErrorDiag)
        else:
            tolog("Panda server returned a %d" % (rt))
            tolog("(Failed to update panda server - leave for next pilot)")

        return rt

    def verifyJobInfo(self, _job, _site, JS):
        """ Query the job state file for job information """

        status = False

        # only proceed with job recovery if job is in holding or lostheartbeart state
        if _job.result[0] == 'holding' or _job.result[0] == 'lostheartbeat':
            tolog("Recovery will be attempted")
            status = True
        elif _job.result[0] == 'failed':
            tolog("Recovery will be skipped")
            tolog("Further recovery attempts will be prevented for this job (will leave work dir)")
            if not JS.rename(_site, _job):
                tolog("(Fate of job state file left for next pilot)")
        else:
            tolog("Not enough information in job state file, will query server")
            if self.recoverWithServer(_job, _site, JS):
                status = True

        return status

    def recoverWithServer(self, _job, _site, JS):
        """ query server about job and decide if it should be recovered or not """

        status = False

        # get job status from server
        jobStatus, jobAttemptNr, jobStatusCode = pUtil.getJobStatus(_job.jobId, self.__pshttpurl, self.__psport, self.__pilot_initdir)

        # recover this job?
        if jobStatusCode == 20:
            tolog("Received general error code from dispatcher call (leave job for later pilot)")
        elif not (jobStatus == 'holding' and jobStatusCode == 0):
            tolog("Job %s is currently in state \'%s\' with attemptNr = %d (according to server - will not be recovered)" % \
                  (_job.jobId, jobStatus, jobAttemptNr))

            if _job.attemptNr != jobAttemptNr or jobStatus == "transferring" or jobStatus == "failed" or \
                   jobStatus == "notfound" or jobStatus == "finished" or "tobekilled" in _job.action:
                tolog("Further recovery attempts will be prevented for this job")
                if not JS.rename(_site, _job):
                    tolog("(Fate of job state file left for next pilot)")
                else:
                    if not JS.cleanup():
                        tolog("!!WARNING!!1110!! Failed to cleanup")
        else:
            # check if the attemptNr (set during initial getJob command) is the same
            # as the current jobAttemptNr from the server (protection against failed lost
            # heartbeat jobs due to reassigned panda job id numbers)
            if _job.attemptNr != jobAttemptNr:
                tolog("WARNING: Attempt number mismatch for job %s (according to server - will not be recovered)" % (_job.jobId))
                tolog("....Initial attempt number: %d" % (_job.attemptNr))
                tolog("....Current attempt number: %d" % (jobAttemptNr))
                tolog("....Job status (server)   : %s" % (jobStatus))
                tolog("....Job status (state)    : %s" % (_job.result[0]))
                tolog("Further recovery attempts will be prevented for this job")
                if not JS.rename(_site, _job):
                    tolog("(Fate of job state file left for next pilot)")
                else:
                    if not JS.cleanup():
                        tolog("!!WARNING!!1110!! Failed to cleanup")
                pass
            else:
                tolog("Attempt numbers from server and job state file agree: %d" % (_job.attemptNr))

                # the job state as given by the dispatcher should only be different from that of
                # the job state file for 'lostheartbeat' jobs. This state is only set like this
                # in the job state file. The dispatcher will consider it as a 'holding' job.
                tolog("Job %s is currently in state \'%s\' (according to job state file: \'%s\') - recover" % (_job.jobId, jobStatus, _job.result[0]))

                status = True

        return status

    def recover(self):
        """ """

        # get file state dictionary from file

        # get object from job state file

        # check job state on server

        # loop over all files in the file state dictionary
        for filename in fileStateDictionary.keys():
            state = ""
            tolog("Processing file: \'%s\' in state \'%s\'" % (filename, state))
            transferFile(filename, state)
            registerFile(filename, state)

    def transferFile(self, filename, state):
        """ Transfer file to SE if necessary """

        # should the file be transferred?
        if "created" in state:
            # file needs to be transferred
            tolog("Preparing to transfer file")
        elif "transferred" in state:
            tolog("File was already transferred")
        elif "missing" in state:
            # file was never created, so skip it
            pass
        else:
            self.__pilotErrorDiag = "Can not recovery a file in this state: %s" % (state)
            tolog(self.__errorString % self.__pilotErrorDiag)

    def registerFile(self, filename, state):
        """ Register file in LFC if necessary """

        if "not_registered" in state:
            # file needs to be registered
            tolog("Preparing to register file")
        elif "registered" in state:
            tolog("File was already registered")
        else:
            self.__pilotErrorDiag = "Can not register a file in this state: %s" % (state)
            tolog(self.__errorString % self.__pilotErrorDiag)

    def verifyFileStatus(self):
        """ """

        pass

    def isSameType(self, trf, userflag):
        """ is the lost job of same type as the current pilot? """

        # treat userflag 'self' as 'user'
        if userflag == 'self':
            userflag = 'user'

        if (isAnalysisJob(trf) and userflag == 'user') or \
               (not isAnalysisJob(trf) and userflag != 'user'):
            sametype = True
            if userflag == 'user':
                tolog("Lost job is of same type as current pilot (analysis pilot, lost analysis job trf: %s)" % (trf))
            else:
                tolog("Lost job is of same type as current pilot (production pilot, lost production job trf: %s)" % (trf))
        else:
            sametype = False
            if userflag == 'user':
                tolog("Lost job is not of same type as current pilot (analysis pilot, lost production job trf: %s)" % (trf))
            else:
                tolog("Lost job is not of same type as current pilot (production pilot, lost analysis job trf: %s)" % (trf))

        return sametype

    def setLog(self):
        """ Redirect log messages to recovery_reports/recovery_report-<jobId>-<attemptNr>.txt """

        status = False

        # does recovery_reports/ exist?
        _path = "%s/recovery_reports" % (self.__recoveryDir)
        if not os.path.exists(_path):
            try:
                os.mkdir(_path)
            except Exception, e:
                self.__pilotErrorDiag = "Failed to create recovery reports dir: %s" % (e)
                tolog(self.__errorString % self.__pilotErrorDiag)
            else:
                tolog("Created %s" % (_path))
                status = True
        else:
            status = True

        if status:
            # proceed to set the log file
            fileName = os.path.join(_path, "recovery_report-%s-%d.txt" % (self.__jobId, self.__recoveryAttempt))
            msg = "Redirecting log messages to: " % (fileName)
            print msg
            tolog(msg)
            setPilotlogFilename(fileName)

        return status

    def updateJobStateTest(self, job, thisSite, workerNode, recoveryAttempt=0, mode=""):
        """ update the job state file """

        # NOTE: this function will eventually replace pilot::updateJobState() when new job rec is in place

        status = False

        # IGNORE TEST MODE
        mode = ""

        # create a job state object and give it the current job state information
        JS = JobState()
        if JS.put(job, thisSite, workerNode, recoveryAttempt=recoveryAttempt, mode=mode):
            if recoveryAttempt > 0:
                tolog("Successfully updated job state file (recovery attempt number: %d) with state: %s" % (recoveryAttempt, job.jobState))
            else:
                tolog("Successfully updated job state file with state: %s" % (job.jobState))
                status = True
        else:
            self.__pilotErrorDiag = "Failed to update job state file"
            tolog(self.__errorString % self.__pilotErrorDiag)

        return status def killProcesses(pid, pgrp):
    """ kill a job upon request """

    pUtil.tolog("killProcesses() called")

    # if there is a known subprocess pgrp, then it should be enough to kill the group in one go
    status = False

    _sleep = True
    if pgrp != 0:
        # kill the process gracefully
        pUtil.tolog("Killing group process %d" % (pgrp))
        try:
            os.killpg(pgrp, signal.SIGTERM)
        except Exception,e:
            pUtil.tolog("WARNING: Exception thrown when killing the child group process with SIGTERM: %s" % (e))
            _sleep = False
        else:
            pUtil.tolog("(SIGTERM sent)")

        if _sleep:
            _t = 30
            pUtil.tolog("Sleeping %d s to allow processes to exit" % (_t))
            time.sleep(_t)

        try:
            os.killpg(pgrp, signal.SIGKILL)
        except Exception,e:
            pUtil.tolog("WARNING: Exception thrown when killing the child group process with SIGKILL: %s" % (e))
        else:
            pUtil.tolog("(SIGKILL sent)")
            status = True

    if not status:
        # firstly find all the children process IDs to be killed
        children = []
        findProcessesInGroup(children, pid)

        # reverse the process order so that the athena process is killed first (otherwise the stdout will be truncated)
        children.reverse()
        pUtil.tolog("Process IDs to be killed: %s (in reverse order)" % str(children))

        # find which commands are still running
        try:
            cmds = getProcessCommands(os.geteuid(), children)
        except Exception, e:
            pUtil.tolog("getProcessCommands() threw an exception: %s" % str(e))
        else:
            if len(cmds) <= 1:
                pUtil.tolog("Found no corresponding commands to process id(s)")
            else:
                pUtil.tolog("Found commands still running:")
                for cmd in cmds:
                    pUtil.tolog(cmd)

                # loop over all child processes
                for i in children:
                    # dump the stack trace before killing it
                    dumpStackTrace(i)

                    # kill the process gracefully
                    try:
                        os.kill(i, signal.SIGTERM)
                    except Exception,e:
                        pUtil.tolog("WARNING: Exception thrown when killing the child process %d under SIGTERM, wait for kill -9 later: %s" % (i, str(e)))
                        pass
                    else:
                        pUtil.tolog("Killed pid: %d (SIGTERM)" % (i))

                    _t = 10
                    pUtil.tolog("Sleeping %d s to allow process to exit" % (_t))
                    time.sleep(_t)

                    # now do a hardkill just in case some processes haven't gone away
                    try:
                        os.kill(i, signal.SIGKILL)
                    except Exception,e:
                        pUtil.tolog("WARNING: Exception thrown when killing the child process %d under SIGKILL, ignore this if it is already killed by previous SIGTERM: %s" % (i, str(e)))
                        pass
                    else:
                        pUtil.tolog("Killed pid: %d (SIGKILL)" % (i))

    pUtil.tolog("Killing any remaining orphan processes")
    killOrphans() def checkProcesses(pid):
    """ Check the number of running processes """

    children = []
    n = 0
    try:
        findProcessesInGroup(children, pid)
    except Exception, e:
        pUtil.tolog("!!WARNING!!2888!! Caught exception in findProcessesInGroup: %s" % (e))
    else:
        n = len(children)
        pUtil.tolog("Number of running processes: %d" % (n))
    return n def killOrphans():
    """ Find and kill all orphan processes belonging to current pilot user """

    if 'BOINC' in pUtil.env['sitename']:
        pUtil.tolog("BOINC job, not looking for orphan processes")
        return

    if 'PILOT_NOKILL' in os.environ:
        return

    pUtil.tolog("Searching for orphan processes")
    cmd = "ps -o pid,ppid,args -u %s" % (commands.getoutput("whoami"))

    processes = commands.getoutput(cmd)
    pattern = re.compile('(\d+)\s+(\d+)\s+(\S+)')

    count = 0
    for line in processes.split('\n'):
        ids = pattern.search(line)
        if ids:
            pid = ids.group(1)
            ppid = ids.group(2)
            args = ids.group(3)
            if 'cvmfs2' in args:
                pUtil.tolog("Ignoring possible orphan process running cvmfs2: pid=%s, ppid=%s, args='%s'" % (pid, ppid, args))
            elif 'pilots_starter.py' in args:
                pUtil.tolog("Ignoring Pilot Launcher: pid=%s, ppid=%s, args='%s'" % (pid, ppid, args))
            elif ppid == '1':
                count += 1
                pUtil.tolog("Found orphan process: pid=%s, ppid=%s, args='%s'" % (pid, ppid, args))
                if args.endswith('bash'):
                    pUtil.tolog("Will not kill bash process")
                else:
                    try:
                        os.killpg(int(pid), signal.SIGKILL)
                    except Exception as e:
                        pUtil.tolog("!!WARNING!!2323!! Failed to send SIGKILL: %s" % e)
                        cmd = 'kill -9 %s' % (pid)
                        ec, rs = commands.getstatusoutput(cmd)
                        if ec != 0:
                            pUtil.tolog("!!WARNING!!2999!! %s" % (rs))
                        else:
                            pUtil.tolog("Killed orphaned process %s (%s)" % (pid, args))
                    pUtil.tolog("Killed orphaned process group %s (%s)" % (pid, args))

    if count == 0:
        pUtil.tolog("Did not find any orphan processes")
    else:
        pUtil.tolog("Found %d orphan process(es)" % (count)) def getMaxMemoryUsageFromCGroups():
    """ Read the max_memory from CGROUPS file memory.max_usage_in_bytes"""

    max_memory = None

    # Get the CGroups max memory using the pilot pid
    pid = os.getpid()
    path = "/proc/%d/cgroup" % (pid)
    if os.path.exists(path):
        cmd = "grep memory %s" % (path)
        pUtil.tolog("Executing command: %s" % (cmd))
        out = commands.getoutput(cmd)
        if out == "":
            pUtil.tolog("(Command did not return anything)")
        else:
            pUtil.tolog(out)
            if ":memory:" in out:
                pos = out.find('/')
                path = out[pos:]
                pUtil.tolog("Extracted path = %s" % (path))

                pre = getCGROUPSBasePath()
                if pre != "":
                    path = pre + os.path.join(path, "memory.max_usage_in_bytes")
                    pUtil.tolog("Path to CGROUPS memory info: %s" % (path))

                    try:
                        f = open(path, 'r')
                    except IOError, e:
                        pUtil.tolog("!!WARNING!!2212!! Could not open file %s: %s" % (path, e))
                    else:
                        max_memory = f.read()
                        f.close()

                else:
                    pUtil.tolog("CGROUPS base path could not be extracted - not a CGROUPS site")
            else:
                pUtil.tolog("!!WARNING!!2211!! Invalid format: %s (expected ..:memory:[path])" % (out))
    else:
        pUtil.tolog("Path %s does not exist (not a CGROUPS site)" % path)

    return max_memory def get_current_cpu_consumption_time(pid):
    # get all the child processes                                                                                                                               
    children = []
    findProcessesInGroup(children, pid)

    cpuconsumptiontime = 0
    for _pid in children:
        _cpuconsumptiontime = get_instant_cpu_consumption_time(_pid)
        if _cpuconsumptiontime:
            cpuconsumptiontime += _cpuconsumptiontime

    return cpuconsumptiontime def findProcessesInGroup(cpids, pid):
    """ recursively search for the children processes belonging to pid and return their pids
    here pid is the parent pid for all the children to be found
    cpids is a list that has to be initialized before calling this function and it contains
    the pids of the children AND the parent as well """

    cpids.append(pid)
    psout = commands.getoutput("ps -eo pid,ppid -m | grep %d" % pid)
    lines = psout.split("\n")
    if lines != ['']:
        for i in range(0, len(lines)):
            try:
                thispid = int(lines[i].split()[0])
                thisppid = int(lines[i].split()[1])
            except Exception as e:
                pUtil.tolog('exception caught: %s (lines[1]=%s)' % (e, lines[1]))
            else:
                if thisppid == pid:
                    findProcessesInGroup(cpids, thispid) class PilotErrors:
    """ Pilot error handling """

    # error codes
    ERR_UNKNOWNERROR = 0
    ERR_GENERALERROR = 1008
    ERR_DIRECTIOFILE = 1009 # harmless, just means that copy-to-scratch was skipped in favor or direct i/o access
    ERR_GETDATAEXC = 1097
    ERR_NOLOCALSPACE = 1098
    ERR_STAGEINFAILED = 1099
    ERR_REPNOTFOUND = 1100
    ERR_LRCREGCONNREF = 1101
    ERR_NOSUCHFILE = 1103
    ERR_USERDIRTOOLARGE = 1104
    ERR_LFCADDCSUMFAILED = 1105
    ERR_STDOUTTOOBIG = 1106
    ERR_MISSDBREL = 1107
    ERR_FAILEDLCGREG = 1108
    ERR_CMTCONFIG = 1109
    ERR_SETUPFAILURE = 1110
    ERR_RUNJOBEXC = 1111
    ERR_PILOTEXC = 1112
    ERR_GETLFCIMPORT = 1113
    ERR_PUTLFCIMPORT = 1114
    ERR_NFSSQLITE = 1115
    ERR_QUEUEDATA = 1116
    ERR_QUEUEDATANOTOK = 1117
    ERR_CURLSPACE = 1118
    ERR_DDMSPACE = 1119
    ERR_NOSTMATCHDEST = 1120 # not used
    ERR_NOLFCSFN = 1122
    ERR_MISSINGGUID = 1123
    ERR_OUTPUTFILETOOLARGE = 1124
    ERR_NOPFC = 1130
    ERR_PUTFUNCNOCALL = 1131
    ERR_LRCREG = 1132
    ERR_NOSTORAGE = 1133
    ERR_MKDIR = 1134
    ERR_FAILEDSIZELOCAL = 1135
    ERR_FAILEDMD5LOCAL = 1136
    ERR_STAGEOUTFAILED = 1137
    ERR_FAILEDSIZE = 1138
    ERR_PUTWRONGSIZE = 1139
    ERR_FAILEDMD5 = 1140
    ERR_PUTMD5MISMATCH = 1141
    ERR_CHMODTRF = 1143
    ERR_PANDAKILL = 1144
    ERR_GETMD5MISMATCH = 1145
    ERR_DYNTRFINST = 1146
    ERR_FAILEDRM = 1148
    ERR_TRFDOWNLOAD = 1149
    ERR_LOOPINGJOB = 1150
    ERR_GETTIMEOUT = 1151
    ERR_PUTTIMEOUT = 1152
    ERR_LOSTJOBNOTFINISHED = 1153
    ERR_LOSTJOBLOGREG = 1154
    ERR_LOSTJOBFILETRANSFER = 1155
    ERR_LOSTJOBRECOVERY = 1156
    ERR_LOSTJOBMAXEDOUT = 1158
    ERR_LOSTJOBPFC = 1159
    ERR_LRCREGSTRSIZE = 1160
    ERR_LOSTJOBXML = 1161
    ERR_LRCREGDUP = 1162
    ERR_NOPROXY = 1163
    ERR_MISSINGLOCALFILE = 1164
    ERR_MISSINGOUTPUTFILE = 1165
    ERR_SIGPIPE = 1166
    ERR_MISSFILEXML = 1167
    ERR_SIZETOOLARGE = 1168
    ERR_FAILEDLFCREG = 1169
    ERR_FAILEDADLOCAL = 1170
    ERR_GETADMISMATCH = 1171
    ERR_PUTADMISMATCH = 1172
    ERR_PANDAMOVERFILENOTCACHED = 1173
    ERR_PANDAMOVERTRANSFER = 1174
    ERR_GETWRONGSIZE = 1175
    ERR_NOCHILDPROCESSES = 1176
    ERR_NOVOMSPROXY = 1177
    ERR_NOSTAGEDFILES = 1178
    ERR_FAILEDLFCGETREPS = 1179
    ERR_GETGLOBUSSYSERR = 1180
    ERR_PUTGLOBUSSYSERR = 1181
    ERR_FAILEDLFCGETREP = 1182
    ERR_GUIDSEXISTSINLRC = 1183
    ERR_MISSINGPFC = 1184
    ERR_NOSOFTWAREDIR = 1186
    ERR_NOPAYLOADMETADATA = 1187
    ERR_LCGGETTURLS = 1188
    ERR_LCGGETTURLSTIMEOUT = 1189
    ERR_LFNTOOLONG = 1190
    ERR_ZEROFILESIZE = 1191
    ERR_DBRELNOTYETTRANSFERRED = 1192
    ERR_SEPROBLEM = 1193
    ERR_NOFILEVERIFICATION = 1194
    ERR_COMMANDTIMEOUT = 1195
    ERR_GETFAILEDTOMOUNTNFS4 = 1196
    ERR_GETPNFSSYSTEMERROR = 1197
    ERR_MKDIRWORKDIR = 1199
    ERR_KILLSIGNAL = 1200
    ERR_SIGTERM = 1201
    ERR_SIGQUIT = 1202
    ERR_SIGSEGV = 1203
    ERR_SIGXCPU = 1204
#    ERR_USERKILL = 1205 # not used by pilot
    ERR_SIGBUS = 1206
    ERR_SIGUSR1 = 1207
    ERR_NOPAYLOADOUTPUT = 1210
    ERR_MISSINGINSTALLATION = 1211
    ERR_PAYLOADOUTOFMEMORY = 1212
    ERR_REACHEDMAXTIME = 1213
    ERR_DAFSNOTALLOWED = 1214
    ERR_NOTCPCONNECTION = 1215
    ERR_NOPILOTTCPSERVER = 1216
    ERR_CORECOUNTMISMATCH = 1217
    ERR_RUNEVENTEXC = 1218
    ERR_UUIDGEN = 1219
    ERR_UNKNOWN = 1220
    ERR_FILEEXIST = 1221
    ERR_GETKEYPAIR = 1222
    ERR_BADALLOC = 1223
    ERR_ESRECOVERABLE = 1224
    ERR_ESMERGERECOVERABLE = 1225
    ERR_GLEXEC = 1226
    ERR_ESATHENAMPDIED = 1227
    ERR_ESFATAL = 1228
    ERR_TEFATAL = 1229
    ERR_TEBADURL = 1230
    ERR_TEINVALIDGUID = 1231
    ERR_TEWRONGGUID = 1232
    ERR_TEHOSTNAME = 1233
    ERR_EXECUTEDCLONEJOB = 1234
    ERR_PAYLOADEXCEEDMAXMEM = 1235
    ERR_FAILEDBYSERVER = 1236
    ERR_ESKILLEDBYSERVER = 1237
    ERR_NOEVENTS = 1238
    ERR_OVERSUBSCRIBEDEVENTS = 1239
    ERR_ESMESSAGESERVER = 1240
    ERR_ESOBJECTSTORESETUP = 1241
    ERR_CHKSUMNOTSUP = 1242
    ERR_ESPREFETCHERDIED = 1243
    ERR_NORELEASEFOUND = 1244
    ERR_TOOFEWEVENTS = 1245
    ERR_NOUSERTARBALL = 1246
    ERR_BADXML = 1247
    ERR_WRONGARCHITECTURE = 1248
    ERR_RUNGENFAILURE = 1249

    # internal error codes
    ERR_DDMREG = 1
    ERR_FILEONTAPE = 2

    pilotError = {
        ERR_UNKNOWNERROR : "",
        ERR_GENERALERROR : "General pilot error, consult batch log",
        ERR_GETDATAEXC : "Get function can not be called for staging input file",
        ERR_NOLOCALSPACE : "No space left on local disk",
        ERR_STAGEINFAILED : "Get error: Staging input file failed",
        ERR_REPNOTFOUND : "Get error: Replica not found",
        ERR_LRCREGCONNREF : "LRC registration error: Connection refused",
        # 1102 : "Expected output file does not exist", # not used, see ERR_MISSINGOUTPUTFILE below
        ERR_NOSUCHFILE : "No such file or directory",
        ERR_USERDIRTOOLARGE : "User work directory too large",
        ERR_LFCADDCSUMFAILED : "Put error: Failed to add file size and checksum to LFC",
        ERR_STDOUTTOOBIG : "Payload stdout file too big",
        ERR_MISSDBREL : "Get error: Missing DBRelease file",
        ERR_FAILEDLCGREG : "Put error: LCG registration failed",
        ERR_CMTCONFIG : "Required CMTCONFIG incompatible with WN",
        ERR_SETUPFAILURE : "Failed during setup",
        ERR_RUNJOBEXC : "Exception caught by RunJob*",
        ERR_PILOTEXC : "Exception caught by pilot",
        ERR_GETLFCIMPORT : "Get error: Failed to import LFC python module",
        ERR_PUTLFCIMPORT : "Put error: Failed to import LFC python module",
        ERR_NFSSQLITE : "NFS SQLite locking problems",
        ERR_QUEUEDATA : "Pilot could not download queuedata",
        ERR_QUEUEDATANOTOK : "Pilot found non-valid queuedata",
        ERR_CURLSPACE : "Pilot could not curl space report",
        ERR_DDMSPACE : "Pilot aborted due to DDM space shortage",
        ERR_NOSTMATCHDEST : "Space token descriptor does not match destination path",
        # 1121 : "Can not read the xml file for registering output files to dispatcher", # not used
        ERR_NOLFCSFN : "Bad replica entry returned by lfc_getreplicas(): SFN not set in LFC for this guid",
        ERR_MISSINGGUID : "Missing guid in output file list",
        ERR_OUTPUTFILETOOLARGE : "Output file too large",
        ERR_NOPFC : "Get error: Failed to get PoolFileCatalog",
        ERR_PUTFUNCNOCALL : "Put function can not be called for staging out",
        ERR_LRCREG : "LRC registration error (consult log file)",
        ERR_NOSTORAGE : "Put error: Fetching default storage URL failed",
        ERR_MKDIR : "Put error: Error in mkdir on localSE, not allowed or no available space",
        ERR_FAILEDSIZELOCAL : "Could not get file size in job workdir",
        ERR_FAILEDMD5LOCAL : "Error running md5sum on the file in job workdir",
        ERR_STAGEOUTFAILED : "Put error: Error in copying the file from job workdir to localSE",
        ERR_FAILEDSIZE : "Put error: could not get the file size on localSE",
        ERR_PUTWRONGSIZE : "Put error: Problem with copying from job workdir to local SE: size mismatch",
        ERR_FAILEDMD5 : "Put error: Error running md5sum on the file on local SE",
        ERR_PUTMD5MISMATCH : "Put error: Problem with copying from job workdir to local SE: md5sum mismatch",
        # 1142 : "Put error: failed to register the file on local SE", # not used
        ERR_CHMODTRF : "Failed to chmod trf",
        ERR_PANDAKILL : "This job was killed by panda server",
        ERR_GETMD5MISMATCH : "Get error: md5sum mismatch on input file",
        ERR_DYNTRFINST : "Trf installation dir does not exist and could not be installed",
        # 1147 : "Put error: dccp returned readOnly",  # not used
        ERR_FAILEDRM : "Put error: Failed to remove readOnly file in dCache",
        ERR_TRFDOWNLOAD : "wget command failed to download trf",
        ERR_LOOPINGJOB : "Looping job killed by pilot",
        ERR_GETTIMEOUT : "Get error: Input file staging timed out",
        ERR_PUTTIMEOUT : "Put error: File copy timed out",
        ERR_LOSTJOBNOTFINISHED : "Lost job was not finished",
        ERR_LOSTJOBLOGREG : "Failed to register log file",
        ERR_LOSTJOBFILETRANSFER : "Failed to move output files for lost job",
        ERR_LOSTJOBRECOVERY : "Pilot could not recover job",
        # 1157 : "Could not create log file", # not used
        ERR_LOSTJOBMAXEDOUT : "Reached maximum number of recovery attempts",
        ERR_LOSTJOBPFC : "Job recovery could not read PoolFileCatalog.xml file (guids lost)",
        ERR_LRCREGSTRSIZE : "LRC registration error: file name string size exceeded limit of 250",
        ERR_LOSTJOBXML : "Job recovery could not generate xml for remaining output files",
        ERR_LRCREGDUP : "LRC registration error: Non-unique LFN",
        ERR_NOPROXY : "Grid proxy not valid",
        ERR_MISSINGLOCALFILE : "Get error: Local input file missing",
        ERR_MISSINGOUTPUTFILE : "Put error: Local output file missing",
        ERR_SIGPIPE : "Put error: File copy broken by SIGPIPE",
        ERR_MISSFILEXML : "Get error: Input file missing in PoolFileCatalog.xml",
        ERR_SIZETOOLARGE : "Get error: Total file size too large",
        ERR_FAILEDLFCREG : "Put error: File registration failed",
        ERR_FAILEDADLOCAL : "Error running adler32 on the file in job workdir",
        ERR_GETADMISMATCH : "Get error: adler32 mismatch on input file",
        ERR_PUTADMISMATCH : "Put error: adler32 mismatch on output file",
        ERR_PANDAMOVERFILENOTCACHED : "PandaMover staging error: File is not cached",
        ERR_PANDAMOVERTRANSFER : "PandaMover transfer failure",
        ERR_GETWRONGSIZE : "Get error: Problem with copying from local SE to job workdir: size mismatch",
        ERR_NOCHILDPROCESSES : "Pilot has no child processes (job wrapper has either crashed or did not send final status)",
        ERR_NOVOMSPROXY : "Voms proxy not valid",
        ERR_NOSTAGEDFILES : "Get error: No input files are staged",
        ERR_FAILEDLFCGETREPS : "Get error: Failed to get replicas",
        ERR_GETGLOBUSSYSERR : "Get error: Globus system error",
        ERR_PUTGLOBUSSYSERR : "Put error: Globus system error",
        ERR_FAILEDLFCGETREP : "Get error: Failed to get replica",
        ERR_GUIDSEXISTSINLRC : "LRC registration error: Guid-metadata entry already exists",
        ERR_MISSINGPFC : "Put error: PoolFileCatalog could not be found in workdir",
        # 1185 : "Put error: Error running adler32 on the file in job workdir", # not used
        ERR_NOSOFTWAREDIR : "Software directory does not exist",
        ERR_NOPAYLOADMETADATA : "Payload metadata is not available",
        ERR_LCGGETTURLS : "lcg-getturls failed",
        ERR_LCGGETTURLSTIMEOUT : "lcg-getturls was timed-out",
        ERR_LFNTOOLONG : "LFN too long (exceeding limit of 150 characters)",
        ERR_ZEROFILESIZE : "Illegal zero file size",
        ERR_DBRELNOTYETTRANSFERRED : "DBRelease file has not been transferred yet",
        ERR_NOFILEVERIFICATION : "File verification failed",
        ERR_COMMANDTIMEOUT : "Command timed out",
        ERR_GETFAILEDTOMOUNTNFS4 : "Get error: Failed to mount NSF4",
        ERR_GETPNFSSYSTEMERROR : "Get error: PNFS system error",
        # 1198 : "Can not check the child process status from the heartbeat process", # not used
        ERR_MKDIRWORKDIR : "Could not create directory",
        ERR_KILLSIGNAL : "Job terminated by unknown kill signal",
        ERR_SIGTERM : "Job killed by signal: SIGTERM",
        ERR_SIGQUIT : "Job killed by signal: SIGQUIT",
        ERR_SIGSEGV : "Job killed by signal: SIGSEGV",
        ERR_SIGXCPU : "Job killed by signal: SIGXCPU",
        ERR_SIGUSR1 : "Job killed by signal: SIGUSR1",
        ERR_SIGBUS : "Job killed by signal: SIGBUS",
        ERR_NOPAYLOADOUTPUT : "No payload output",
        ERR_MISSINGINSTALLATION : "Missing installation",
        ERR_PAYLOADOUTOFMEMORY : "Payload ran out of memory",
        ERR_REACHEDMAXTIME : "Reached batch system time limit",
        ERR_DAFSNOTALLOWED : "Site does not allow requested direct access or file stager",
        ERR_NOTCPCONNECTION : "Failed to open TCP connection to localhost (worker node network problem)",
        ERR_NOPILOTTCPSERVER : "Pilot TCP server has died",
        ERR_CORECOUNTMISMATCH : "Mismatch between core count in job and queue definition",
        ERR_RUNEVENTEXC : "Exception caught by runEvent",
        ERR_UUIDGEN : "uuidgen failed to produce a guid",
        ERR_UNKNOWN : "Job failed due to unknown reason (consult log file)",
        ERR_FILEEXIST : "File already exist",
        ERR_GETKEYPAIR : "Failed to get security key pair",
        ERR_BADALLOC : "TRF failed due to bad_alloc",
        ERR_ESMERGERECOVERABLE : "Recoverable Event Service Merge error",
        ERR_ESRECOVERABLE: "Recoverable Event Service error",
        ERR_GLEXEC: "gLExec related error",
        ERR_ESATHENAMPDIED: "AthenaMP ended Event Service job prematurely",
        ERR_ESFATAL: "Fatal Event Service error",
        ERR_TEFATAL: "Fatal Token Extractor error",
        ERR_TEHOSTNAME: "Token Extractor error: Host name could not be resolved",
        ERR_TEBADURL: "Token Extractor error: Bad URL",
        ERR_TEINVALIDGUID: "Token Extractor error: Invalid GUID length",
        ERR_TEWRONGGUID: "Token Extractor error: No tokens for this GUID",
        ERR_EXECUTEDCLONEJOB: "Already executed clone job",
        ERR_PAYLOADEXCEEDMAXMEM: "Payload exceeded maximum allowed memory",
        ERR_FAILEDBYSERVER: "Failed by server",
        ERR_ESKILLEDBYSERVER: "Event Service job killed by server",
        ERR_NOEVENTS: "Event Service no available events",
        ERR_OVERSUBSCRIBEDEVENTS: "Event Service over subscribed events",
        ERR_ESMESSAGESERVER: "Event service message server error",
        ERR_ESOBJECTSTORESETUP: "Event service objectstore setup",
        ERR_CHKSUMNOTSUP: "Mover error: query checksum is not supported",
        ERR_ESPREFETCHERDIED: "Prefetcher ended Event Service job prematurely",
        ERR_NORELEASEFOUND: "No release candidates found",
        ERR_TOOFEWEVENTS: "Too few events, less events than minimal requirement",
        ERR_NOUSERTARBALL: "User tarball cannot be downloaded from PanDA server",
        ERR_BADXML: "Badly formed XML",
        ERR_WRONGARCHITECTURE: "Job built on wrong architecture",
        ERR_RUNGENFAILURE: "RunGen failure (consult log file)",

    }

    getErrorCodes = [1097, 1099, 1100, 1103, 1107, 1113, 1130, 1145, 1151, 1164, 1167, 1168, 1171, 1175, 1178, 1179, 1180, 1182]
    putErrorCodes = [1101, 1114, 1122, 1131, 1132, 1133, 1134, 1135, 1136, 1137, 1138, 1140, 1141, 1152, 1154, 1155, 1181]
    recoverableErrorCodes = [0] + putErrorCodes

    # Error codes that will issue a Pilot-controlled resubmission
    PilotResubmissionErrorCodes = [1008, 1098, 1099, 1110, 1113, 1114, 1115, 1116, 1117, 1137, 1139, 1151, 1152, 1171, 1172, 1177, 1179, 1180, 1181, 1182, 1188, 1189, 1195, 1196, 1197, 1219]

    # Error codes used with FAX fail-over (only an error code in this list will allow FAX fail-over)
    PilotFAXErrorCodes = [1103] + PilotResubmissionErrorCodes

    def getPilotErrorDiag(self, code=0):
        """ Return text corresponding to error code """

        pilotErrorDiag = ""
        if code in self.pilotError.keys():
            pilotErrorDiag = self.pilotError[code]
        else:
            pilotErrorDiag = "Unknown pilot error code"
        return pilotErrorDiag

    def isGetErrorCode(self, code=0):
        """ Determine whether code is in the put error list or not """

        state = False
        if code in self.getErrorCodes:
            state = True

        return state

    def isPutErrorCode(self, code=0):
        """ Determine whether code is in the put error list or not """

        state = False
        if code in self.putErrorCodes:
            state = True

        return state

    @classmethod
    def isRecoverableErrorCode(self, code=0):
        """ Determine whether code is a recoverable error code or not """

        return code in self.recoverableErrorCodes

    def isPilotResubmissionErrorCode(self, code=0):
        """ Determine whether code issues a Pilot-controlled resubmission """

        state = False
        if code in self.PilotResubmissionErrorCodes:
            state = True

        return state

    def isPilotFAXErrorCode(self, code=0):
        """ Determine whether code allows for a FAX fail-over """

        state = False
        if code in self.PilotFAXErrorCodes:
            state = True

        return state

    @classmethod
    def getErrorStr(self, code):
        """
        Avoids exception if an error is not in the dictionary.
        An empty string is returned if the error is not in the dictionary.
        """
        return self.pilotError.get(code, '')

    def getErrorName(self, code):
        """ From the error code to get the error name"""
        for k in self.__class__.__dict__.keys():
            if self.__class__.__dict__[k] == code:
                return k
        return None class WatchDog:
    
    def __init__(self):
        self.__env = Configuration()
    
    def pollChildren(self):
        """
        check children processes, collect zombie jobs, and update jobDic status
        """

        error = PilotErrors()

        # tolog("---pollChildren: %s" % str(jobDic))
        for k in self.__env['jobDic'].keys():
            try:
                _id, rc = os.waitpid(self.__env['jobDic'][k][0], os.WNOHANG)
            except OSError, e:
                try:
                    if self.__env['jobDic'][k][1].result[0] == "finished" or self.__env['jobDic'][k][1].result[0] == "failed" or self.__env['jobDic'][k][1].result[0] == "holding":
                        continue
                except:
                    pUtil.tolog("!!FAILED!!1000!! Pilot failed to check the job state: %s" % traceback.format_exc())

                pUtil.tolog("Harmless exception when checking job %s, %s" % (self.__env['jobDic'][k][1].jobId, e))
                if str(e).rstrip() == "[Errno 10] No child processes":
                    pilotErrorDiag = "Exception caught by pilot watchdog: %s" % str(e)
                    pUtil.tolog("!!FAILED!!1000!! Pilot setting state to failed since there are no child processes")
                    pUtil.tolog("!!FAILED!!1000!! %s" % (pilotErrorDiag))
                    self.__env['jobDic'][k][1].result[0] = "failed"
                    self.__env['jobDic'][k][1].currentState = self.__env['jobDic'][k][1].result[0]
                    if self.__env['jobDic'][k][1].result[2] == 0:
                        self.__env['jobDic'][k][1].result[2] = error.ERR_NOCHILDPROCESSES
                    if self.__env['jobDic'][k][1].pilotErrorDiag == "":
                        self.__env['jobDic'][k][1].pilotErrorDiag = pilotErrorDiag
                else:
                    pass
            else:
                if _id: # finished
                    rc = rc%255 # exit code
                    if k == "prod": # production job is done
                        self.__prodJobDone = True
                        pUtil.tolog("Production job is done")
                    if self.__env['jobDic'][k][1].result[0] != "finished" and self.__env['jobDic'][k][1].result[0] != "failed" and self.__env['jobDic'][k][1].result[0] != "holding":
                        if not rc: # rc=0, ok job
                            if not self.__env['jobDic'][k][1].result[1]:
                                self.__env['jobDic'][k][1].result[1] = rc # transExitCode (because pilotExitCode is reported back by child job)
                        else: # rc != 0, failed job
                            self.__env['jobDic'][k][1].result[1] = rc # transExitCode

    def collectZombieJob(self, tn=None):
        """
        collect zombie child processes, tn is the max number of loops, plus 1,
        to avoid infinite looping even if some child proceses really get wedged;
        tn=None means it will keep going till all children zombies collected.        
        """
        time.sleep(1)
        if self.__env['zombieJobList'] and tn > 1:
            pUtil.tolog("--- collectZombieJob: --- %d, %s" % (tn, str(self.__env['zombieJobList'])))
            tn -= 1
            for x in self.__env['zombieJobList']:
                try:
                    pUtil.tolog("Zombie collector trying to kill pid %s" % str(x))
                    _id, rc = os.waitpid(x, os.WNOHANG)
                except OSError,e:
                    pUtil.tolog("Harmless exception when collecting zombie jobs, %s" % str(e))
                    self.__env['zombieJobList'].remove(x)
                else:
                    if _id: # finished
                        self.__env['zombieJobList'].remove(x)
                self.collectZombieJob(tn=tn) # recursion

        if self.__env['zombieJobList'] and not tn: # for the infinite waiting case, we have to
            # use blocked waiting, otherwise it throws
            # RuntimeError: maximum recursion depth exceeded
            for x in self.__env['zombieJobList']:
                try:
                    _id, rc = os.waitpid(x, 0)
                except OSError,e:
                    pUtil.tolog("Harmless exception when collecting zombie jobs, %s" % str(e))
                    self.__env['zombieJobList'].remove(x)
                else:
                    if _id: # finished
                        self.__env['zombieJobList'].remove(x)
                self.collectZombieJob(tn=tn) # recursion class PilotTCPServer(threading.Thread):
    """ TCP server used to send TCP messages from runJob to pilot """

    def __init__(self, handler, name='PilotTCPServer'):
        """ constructor, setting initial variables """

        # find an available port to create the TCPServer
        n = 0
        while n < 20: # try 20 times before giving up, which means run 20 pilots on one machine simultaneously :)
            n += 1
            self.port = random.randrange(1, 800, 1) + 8888
            # avoid known port used for local monitoring (req. by A. de Salvo)
            if self.port == 9256:
                self.port += 1
            try:
                self.srv = TCPServer(('localhost',self.port), handler)
            except socket.error, e:
                tolog("WARNING: Can not create TCP server on port %d, re-try... : %s" % (self.port, str(e)))
            else:
                # tolog("Using port %d" % self.port)
                break

        if n >= 20: # raise some exception later ??
            self.srv = None
            self.port = None
            
        self._stopevent = threading.Event() 
        threading.Thread.__init__(self, name=name)

    def run(self):
        """ main control loop """
        tolog("%s starts" % str((self.getName( ),)))

        while not self._stopevent.isSet():
            self.srv.handle_request()
                
    def join(self, timeout=None):
        """ Stop the thread and wait for it to end. """
        tolog("join called on thread %s" % self.getName())
        self._stopevent.set() # signal thread to stop
        self.srv.socket.shutdown(2) # shutdown the connection in two-ways
        threading.Thread.join(self, timeout) # wait until the thread terminates or timeout occurs class UpdateHandler(BaseRequestHandler):
    """ update self.__env['jobDic'] status with the messages sent from child via socket, do nothing else """
    
    def __init__(self, request, client_address, server):
        self.__env = Configuration()
        BaseRequestHandler.__init__(self, request, client_address, server)

    def handle(self):
        try:
            pUtil.tolog("Connected from %s" % str(self.client_address))
            data = self.request.recv(4096)
            jobmsg = data.split(";")
            pUtil.tolog("--- TCPServer: Message received from child is : %s" % json.dumps(jobmsg))
            jobinfo = {}
            for i in jobmsg:
                if not i: continue # skip empty line
                try:
                    jobinfo[i.split("=")[0]] = i.split("=")[1]
                except Exception, e:
                    pUtil.tolog("!!WARNING!!1999!! Exception caught: %s" % (e))
    
            # update self.__env['jobDic']
            pUtil.tolog("Debug: jobdict keys: %s" % self.__env['jobDic'].keys())
            pUtil.tolog("Debug: jobinfo: %s" % jobinfo)
            found_job = False
            for k in self.__env['jobDic'].keys():
                if str(self.__env['jobDic'][k][1].jobId) == str(jobinfo["jobid"]): # job pid matches
                    found_job = True

                    # wait no more than 30 seconds if it's updating panda server.
                    # otherwise if state changed to finished/failed when it's updating panda server, an error will happen
                    if jobinfo["status"] in ['failed', 'finished']:
                        startWait = time.time()
                        while self.__env['jobDic'][k][1].updatingPandaServer and (time.time() - startWait) < 30:
                            time.sleep(1)
                        
#                if self.__env['jobDic'][k][2] == int(jobinfo["pgrp"]) and self.__env['jobDic'][k][1].jobId == int(jobinfo["jobid"]): # job pid matches
                    # protect with try statement in case the pilot server goes down (jobinfo will be corrupted)
                    try:
                        old_pilotecode = self.__env['jobDic'][k][1].result[2]
                        old_pilotErrorDiag = self.__env['jobDic'][k][1].pilotErrorDiag

                        self.__env['jobDic'][k][1].lastState = self.__env['jobDic'][k][1].currentState
                        self.__env['jobDic'][k][1].currentState = jobinfo["status"]
                        if jobinfo["status"] == "stagein":
                            self.__env['stagein'] = True
                            self.__env['stageout'] = False
                            self.__env['jobDic'][k][1].result[0] = "running"
                        elif jobinfo["status"] == "stageout":
                            self.__env['stagein'] = False
                            self.__env['stageout'] = True
                            self.__env['jobDic'][k][1].result[0] = "running"
                            self.__env['stageoutStartTime'] = int(time.time())
                        else:
                            self.__env['stagein'] = False
                            self.__env['stageout'] = False
                            self.__env['jobDic'][k][1].result[0] = jobinfo["status"]
                        self.__env['jobDic'][k][1].result[1] = int(jobinfo["transecode"]) # transExitCode
                        self.__env['jobDic'][k][1].result[2] = int(jobinfo["pilotecode"]) # pilotExitCode
                        self.__env['jobDic'][k][1].timeStageIn = jobinfo["timeStageIn"]
                        self.__env['jobDic'][k][1].timeStageOut = jobinfo["timeStageOut"]
                        self.__env['jobDic'][k][1].timeSetup = jobinfo["timeSetup"]
                        self.__env['jobDic'][k][1].timeExe = jobinfo["timeExe"]
                        self.__env['jobDic'][k][1].cpuConsumptionTime = int(float(jobinfo["cpuTime"]))
                        self.__env['jobDic'][k][1].cpuConsumptionUnit = jobinfo["cpuUnit"]
                        self.__env['jobDic'][k][1].cpuConversionFactor = jobinfo["cpuConversionFactor"]
                        self.__env['jobDic'][k][1].jobState = jobinfo["jobState"]
                        self.__env['jobDic'][k][1].vmPeakMax = int(jobinfo["vmPeakMax"])
                        self.__env['jobDic'][k][1].vmPeakMean = int(jobinfo["vmPeakMean"])
                        self.__env['jobDic'][k][1].RSSMean = int(jobinfo["RSSMean"])
                        self.__env['jobDic'][k][1].JEM = jobinfo["JEM"]
                        self.__env['jobDic'][k][1].dbTime = jobinfo["dbTime"]
                        self.__env['jobDic'][k][1].dbData = jobinfo["dbData"]
                        self.__env['jobDic'][k][1].cmtconfig = jobinfo["cmtconfig"]

                        try:
                            self.__env['jobDic'][k][1].pgrp = int(jobinfo["pgrp"])
                        except:
                            pUtil.tolog("!!WARNING!!2222!! Failed to convert pgrp value to int: %s" % (e))
                        else:
                            pUtil.tolog("Process groups: %d (pilot), %d (sub process)" % (os.getpgrp(), self.__env['jobDic'][k][1].pgrp))

                        tmp = self.__env['jobDic'][k][1].result[0]
                        if (tmp == "failed" or tmp == "holding" or tmp == "finished") and jobinfo.has_key("logfile"):
                            self.__env['jobDic'][k][1].logMsgFiles.append(jobinfo["logfile"])

                        if jobinfo.has_key("external_stageout_time"):
                            try:
                                self.__env['jobDic'][k][1].external_stageout_time = int(float(jobinfo["external_stageout_time"]))
                            except:
                                pUtil.tolog(traceback.format_exc())

                        if jobinfo.has_key("subStatus"):
                            self.__env['jobDic'][k][1].subStatus = jobinfo["subStatus"]

                        if jobinfo.has_key("pilotErrorDiag"):
                            self.__env['jobDic'][k][1].pilotErrorDiag = pUtil.decode_string(jobinfo["pilotErrorDiag"])

                        if jobinfo.has_key("exeErrorDiag"):
                            self.__env['jobDic'][k][1].exeErrorDiag = pUtil.decode_string(jobinfo["exeErrorDiag"])

                        if jobinfo.has_key("exeErrorCode"):
                            self.__env['jobDic'][k][1].exeErrorCode = int(jobinfo["exeErrorCode"])

                        if jobinfo.has_key("filesWithFAX"):
                            self.__env['jobDic'][k][1].filesWithFAX = int(jobinfo["filesWithFAX"])
    
                        if jobinfo.has_key("filesWithoutFAX"):
                            self.__env['jobDic'][k][1].filesWithoutFAX = int(jobinfo["filesWithoutFAX"])

                        if jobinfo.has_key("bytesWithFAX"):
                            self.__env['jobDic'][k][1].bytesWithFAX = int(jobinfo["bytesWithFAX"])

                        if jobinfo.has_key("bytesWithoutFAX"):
                            self.__env['jobDic'][k][1].bytesWithoutFAX = int(jobinfo["bytesWithoutFAX"])

                        if jobinfo.has_key("filesAltStageOut"):
                            self.__env['jobDic'][k][1].filesAltStageOut = int(jobinfo["filesAltStageOut"])

                        if jobinfo.has_key("filesNormalStageOut"):
                            self.__env['jobDic'][k][1].filesNormalStageOut = int(jobinfo["filesNormalStageOut"])

                        if jobinfo.has_key("nEvents"):
                            try:
                                self.__env['jobDic'][k][1].nEvents = int(jobinfo["nEvents"])
                            except Exception, e:
                                pUtil.tolog("!!WARNING!!2999!! jobinfo did not return an int as expected: %s" % str(e))
                                self.__env['jobDic'][k][1].nEvents = 0
                        if jobinfo.has_key("nEventsW"):
                            try:
                                self.__env['jobDic'][k][1].nEventsW = int(jobinfo["nEventsW"])
                            except Exception, e:
                                pUtil.tolog("!!WARNING!!2999!! jobinfo did not return an int as expected: %s" % str(e))
                                self.__env['jobDic'][k][1].nEventsW = 0
    
                        if jobinfo.has_key("finalstate"):
                            self.__env['jobDic'][k][1].finalstate = jobinfo["finalstate"]
                        if jobinfo.has_key("spsetup"):
                            self.__env['jobDic'][k][1].spsetup = jobinfo["spsetup"]
                            # restore the = and ;-signs
                            self.__env['jobDic'][k][1].spsetup = self.__env['jobDic'][k][1].spsetup.replace("^", ";").replace("!", "=")
                            pUtil.tolog("Handler received special setup command: %s" % (self.__env['jobDic'][k][1].spsetup))

                        if jobinfo.has_key("output_latereg"):
                            self.__env['jobDic'][k][1].output_latereg = jobinfo["output_latereg"]
    
                        if jobinfo.has_key("output_fields"):
                            self.__env['jobDic'][k][1].output_fields = pUtil.stringToFields(jobinfo["output_fields"])
                            pUtil.tolog("Got output_fields=%s" % str(self.__env['jobDic'][k][1].output_fields))
                            pUtil.tolog("Converted from output_fields=%s" % str(jobinfo["output_fields"]))

                        # corrupted files
                        if jobinfo.has_key("corruptedFiles"):
                            self.__env['jobDic'][k][1].corruptedFiles = jobinfo["corruptedFiles"]

                        # hpc status
                        if jobinfo.has_key("mode"):
                            self.__env['jobDic'][k][1].mode = jobinfo['mode']
                        if jobinfo.has_key("hpcStatus"):
                            self.__env['jobDic'][k][1].hpcStatus = jobinfo['hpcStatus']
                        if jobinfo.has_key("yodaJobMetrics"):
                            self.__env['jobDic'][k][1].yodaJobMetrics = json.loads(jobinfo['yodaJobMetrics'])
                        if jobinfo.has_key("coreCount"):
                            self.__env['jobDic'][k][1].coreCount = jobinfo['coreCount']
                        if jobinfo.has_key("HPCJobId"):
                            self.__env['jobDic'][k][1].HPCJobId = jobinfo['HPCJobId']

                        # zip output
                        if jobinfo.has_key("outputZipName"):
                            self.__env['jobDic'][k][1].outputZipName = jobinfo['outputZipName']
                        if jobinfo.has_key("outputZipBucketID"):
                            self.__env['jobDic'][k][1].outputZipBucketID = jobinfo['outputZipBucketID']

                        if (self.__env['jobDic'][k][1].result[2] and self.__env['jobDic'][k][1].result[2] != old_pilotecode) or\
                           (self.__env['jobDic'][k][1].pilotErrorDiag and len(self.__env['jobDic'][k][1].pilotErrorDiag) and self.__env['jobDic'][k][1].pilotErrorDiag != old_pilotErrorDiag):
                            updatePilotErrorReport(self.__env['jobDic'][k][1].result[2], self.__env['jobDic'][k][1].pilotErrorDiag, "2",  self.__env['jobDic'][k][1].jobId, self.__env['pilot_initdir'])
                    except Exception, e:
                        pUtil.tolog("!!WARNING!!1998!! Caught exception. Pilot server down? %s" % str(e))
                        try:
                            pUtil.tolog("Received jobinfo: %s" % str(jobinfo))
                        except:
                            pass

            if not found_job:
                pUtil.tolog("Debug: job not found.")
                pUtil.tolog("Debug: jobdict keys: %s" % self.__env['jobDic'].keys())
                pUtil.tolog("Debug: jobinfo: %s" % jobinfo)
                for k1 in self.__env['jobDic'].keys():
                    pUtil.tolog("Debug: jobkey %s, jobid %s" % (k1, str(self.__env['jobDic'][k1][1].jobId)))
        except Exception, e:
            pUtil.tolog("!!WARNING!!1998!! Caught exception. Pilot server down? %s" % str(e))
            
            
        # pUtil.tolog("---updateHandler : self.__env['jobDic'] is %s" % str(self.__env['jobDic']))
        self.request.send("OK") def updatePilotErrorReport(pilotErrorCode, pilotErrorDiag, priority, jobID, workdir):
    """ Write pilot error info to file """
    # Report format:
    # { jobID1: { priority1: [{ pilotErrorCode1:<nr>, pilotErrorDiag1:<str> }, .. ], .. }, .. }
    # The pilot will report only the first of the highest priority error when it reports the error at the end of the job
    # Use the following priority convention:
    # "0": highest priority [e.g. errors that originate from the main pilot module (unless otherwise necessary)]
    # "1": high priority [e.g. errors that originate from the Monitor module (-||-)]
    # "2": normal priority [errors that originate from other modules (-||-)]
    # etc

    # Convert to string if integer is sent for priority
    if type(priority) != str:
        priority = str(priority)

    filename = getPilotErrorReportFilename(workdir)
    if os.path.exists(filename):
        # The file already exists, read it back (with unicode to utf-8 conversion)
        dictionary = getJSONDictionary(filename)
    else:
        dictionary = {}

    # Sort the new error
    if dictionary.has_key(jobID):
        jobID_dictionary = dictionary[jobID]

        # Update with the latest error info
        if not jobID_dictionary.has_key(priority):
            dictionary[jobID][priority] = []

        new_dictionary = { 'pilotErrorCode':pilotErrorCode, 'pilotErrorDiag':pilotErrorDiag }

        # Update the dictionary with the new info
        dictionary[jobID][priority].append(new_dictionary)
        print dictionary

    else:
        # Create a first entry into the error report
        dictionary[jobID] = {}
        dictionary[jobID][priority] = []
        dictionary[jobID][priority].append({})
        dictionary[jobID][priority][0] = { 'pilotErrorCode':pilotErrorCode, 'pilotErrorDiag':pilotErrorDiag }

    # Finally update the file
    status = writeJSON(filename, dictionary) def getDirSize(d):
    """ Return the size of directory d using du -sk """

    tolog("Checking size of work dir: %s" % (d))
    from commands import getoutput
    size_str = getoutput("du -sk %s" % (d))
    size = 0

    # E.g., size_str = "900\t/scratch-local/nilsson/pilot3z"
    try:
        # Remove tab and path, and convert to int (and B)
        size = int(size_str.split("\t")[0])*1024
    except Exception, e:
        tolog("!!WARNING!!4343!! Failed to convert to int: %s" % (e))
    else:
        tolog("Size of directory %s: %d B" % (d, size))

    return size def storeWorkDirSize(workdir_size, pilot_initdir, job, correction=True):
    """ Store the measured remaining disk space """
    # If correction=True, then input and output file sizes will be deducated

    filename = os.path.join(pilot_initdir, getWorkDirSizeFilename(job.jobId))
    dictionary = {} # FORMAT: { 'workdir_size': [value1, value2, ..] }
    workdir_size_list = []

    if os.path.exists(filename):
        # Read back the dictionary
        dictionary = readJSON(filename)
        if dictionary != {}:
            workdir_size_list = dictionary['workdir_size']
        else:
            tolog("!!WARNING!!4555!! Failed to read back remaining disk space from file: %s" % (filename))

    # Correct for any input and output files
    if correction:

        total_size = 0L # B

        if os.path.exists(job.workdir):
            # Find out which input and output files have been transferred and add their sizes to the total size
            # (Note: output files should also be removed from the total size since outputfilesize is added in the task def)

            # First remove the log file from the output file list
            outFiles = []
            for f in job.outFiles:
                if not job.logFile in f:
                    outFiles.append(f)

            # Then update the file list in case additional output files have been produced
            # Note: don't do this deduction since it is not known by the task definition
            #outFiles, dummy, dummy = discoverAdditionalOutputFiles(outFiles, job.workdir, job.destinationDblock, job.scopeOut)

            file_list = job.inFiles + outFiles
            for f in file_list:
                if f != "":
                    total_size = addToTotalSize(os.path.join(job.workdir, f), total_size)

            tolog("Total size of present input+output files: %d B (work dir size: %d B)" % (total_size, workdir_size))
            workdir_size -= total_size
        else:
            tolog("WARNING: Can not correct for input/output files since workdir does not exist: %s" % (job.workdir))

    # Append the new value to the list and store it
    workdir_size_list.append(workdir_size)
    dictionary = {'workdir_size': workdir_size_list}
    status = writeJSON(filename, dictionary)
    if status:
        tolog("Stored %d B in file %s" % (workdir_size, filename))

    return status def getOsTimesTuple(workdir):
    """ Read os.times() from a txt file and convert it back to a proper os.times() tuple again """
    # This function is used to calculate the cpu consumption time. The t0_times.txt file is created just before the
    # payload is executed in

    times = []
    failed = False
    path = os.path.join(workdir, 't0_times.txt')

    if os.path.exists(path):
        with open(path, 'r') as f:
            for t in f.read().split():
                # remove any initial (, trailing ) or ,
                a = t.strip('(').strip(')').strip(',')
                try:
                    times.append(float(a))
                except ValueError as e:
                    tolog("!!WARNING!!1212!! Exception caught: offending value=%s (cannot convert to float)" % (e))
                    failed = True
                    break

            if not failed:
                # consistency check
                if len(times) == 5:
                    return tuple(times)
                else:
                    tolog("!!WARNING!!1222!! os.times() tuple has wrong length (not 5): %s" % str(times))
            else:
                tolog("!!WARNING!!1222!! Failed to convert os.times() txt file to tuple - CPU consumption meausurement cannot be done")
                return None
    else:
        tolog("t0 file does not exist - probably the payload was not executed")

    return None def readFile(filename):
    """ Read the contents of a file """

    contents = ""
    if os.path.exists(filename):
        try:
            f = open(filename, 'r')
        except IOError, e:
            tolog("!!WARNING!!2121!! Failed to open file %s: %s" % (filename, e))
        else:
            try:
                contents = f.read()
            except Exception, e:
                tolog("!!WARNING!!2122!! Failed to read file %s: %s" % (filename, e))
            f.close()
    else:
        tolog("!!WARNING!!2121!! File does not exist: %s" % (filename))

    return contents def get_files(pattern="*.log"):

    files = []
    stdout = getoutput("find . -name %s" % pattern)
    if stdout:
        # remove last \n if present
        if stdout.endswith('\n'):
            stdout = stdout[:-1]
        files = stdout.split('\n')

    # make sure all files exist
    return verified_files(files) def tail(filename, lines=10):

    return getoutput('tail -%d %s' % (lines, filename)) def find_latest_modified_file(list_of_files):

    # make sure all files exist
    list_of_files = verified_files(list_of_files)

    latest_file = max(list_of_files, key=os.path.getmtime)
    try:
        mtime = int(os.path.getmtime(latest_file))
    except Exception as e:
        tolog("!!WARNING!!2323!! Int conversion failed for mod time: %s" % e)
        mtime = None

    return latest_file, mtime
import atexit
import time
import os
import commands
import json
import re
import sys
import pUtil
import signal
import traceback
import Node
import Job
import inspect
            import traceback
            import traceback
from shutil import copy, copy2
from random import shuffle
from glob import glob
from JobRecovery import JobRecovery
from processes import killProcesses, checkProcesses, killOrphans, getMaxMemoryUsageFromCGroups, get_current_cpu_consumption_time, findProcessesInGroup
from PilotErrors import PilotErrors
from FileStateClient import createFileStates, dumpFileStates, getFileState
from WatchDog import WatchDog
from PilotTCPServer import PilotTCPServer
from UpdateHandler import UpdateHandler
from RunJobFactory import RunJobFactory
from FileHandling import updatePilotErrorReport, getDirSize, storeWorkDirSize, getOsTimesTuple, readFile, get_files, tail, find_latest_modified_file
                from ThreadPool import ThreadPool
globalSite = None

class Monitor:

    __skip = False
    __localsizelimit_stdout = 2*1024**2   # size limit of payload stdout size during running. unit is in kB

    def __init__(self, env):
        self.__error = PilotErrors()
        self.__env = env

        #self.__env['pilot_startup'] = None
        self.__env['create_softlink'] = True
        self.__env['return_code'] = None
        self.__env['curtime_sp'] = int(time.time())
        self.__env['curtime_pr'] = int(time.time())
        self.__env['lastTimeFilesWereModified'] = {}
        self.__wdog = WatchDog()
        self.__runJob = None # Remember the RunJob instance

        # using thread pool
        self.__threadpool = None
        self.__jobs_added_to_threadpool = []
        self.__jobs_cleaned_up = []

        # register cleanup function
        atexit.register(pUtil.cleanup, self.__wdog, self.__env['pilot_initdir'], self.__env['wrapperFlag'], self.__env['rmwkdir'])

        signal.signal(signal.SIGTERM, pUtil.sig2exc)
        signal.signal(signal.SIGQUIT, pUtil.sig2exc)
        signal.signal(signal.SIGSEGV, pUtil.sig2exc)
        signal.signal(signal.SIGXCPU, pUtil.sig2exc)
        signal.signal(signal.SIGBUS,  pUtil.sig2exc)
        signal.signal(signal.SIGUSR1, pUtil.sig2exc)

    def __allowLoopingJobKiller(self):
        """ Should the looping job killer be run? """

        # This decision is not on an Experiment level but is relevant only for the type of subprocess that is requested
        # E.g. the looping job killer should normally be run for a normal ATLAS job, but not on an HPC. Therefore, the
        # decision is made inside the subprocess class (RunJob, RunJobHPC, ..)

        allow = True

        # Which RunJob class do we need to ask?
        if not self.__runJob:
            # First get an Experiment object, which will tell us which subprocess to ask
            thisExperiment = pUtil.getExperiment(self.__env['experiment'])
            subprocessName = thisExperiment.getSubprocessName(self.__env['jobDic']["prod"][1].eventService)
            pUtil.tolog("subprocessName = %s" % (subprocessName))

            # Now get an instance of the corresponding class from the RunJobFactory
            factory = RunJobFactory()
            _rJ = factory.newRunJob(subprocessName)
            self.__runJob = _rJ()

        if self.__runJob:
            # Is the looping job killer allowed by the subprocess?
            allow = self.__runJob.allowLoopingJobKiller()
            name = os.path.splitext(self.__runJob.getRunJobFileName())[0]
            if allow:
                pUtil.tolog("Looping job killer is allowed by subprocess %s" % (name))
            else:
                pUtil.tolog("Looping job killer is not allowed by subprocess %s" % (name))
        else:
            pUtil.tolog("!!WARNING!!2121!! Could not get an instance of a RunJob* class (cannot decide about looping job killer)")
            allow = False

        return allow

    def __checkPayloadStdout(self):
        """ Check the size of the payload stdout """

        # loop over all parallel jobs
        # (after multitasking was removed from the pilot, there is actually only one job)
        for k in self.__env['jobDic'].keys():
            # get list of log files
            fileList = glob("%s/log.*" % (self.__env['jobDic'][k][1].workdir))

            # is this a multi-trf job?
            nJobs = self.__env['jobDic'][k][1].jobPars.count("\n") + 1
            for _i in range(nJobs):
                # get name of payload stdout file created by the pilot
                _stdout = self.__env['jobDic'][k][1].stdout
                if nJobs > 1:
                    _stdout = _stdout.replace(".txt", "_%d.txt" % (_i + 1))
                filename = "%s/%s" % (self.__env['jobDic'][k][1].workdir, _stdout)

                # add the primary stdout file to the fileList
                fileList.append(filename)

            # now loop over all files and check each individually (any large enough file will fail the job)
            for filename in fileList:

                if "job.log.tgz" in filename:
                    pUtil.tolog("(Skipping file size check of file (%s) since it is a special log file)" % (filename))
                    continue

                if os.path.exists(filename):
                    try:
                        # get file size in bytes
                        fsize = os.path.getsize(filename)
                    except Exception, e:
                        pUtil.tolog("!!WARNING!!1999!! Could not read file size of %s: %s" % (filename, str(e)))
                    else:
                        # is the file too big?
                        if fsize > self.__env['localsizelimit_stdout'] * 1024: # convert to bytes
                            pilotErrorDiag = "Payload stdout file too big: %d B (larger than limit %d B)" % (fsize, self.__env['localsizelimit_stdout'] * 1024)
                            pUtil.tolog("!!FAILED!!1999!! %s" % (pilotErrorDiag))
                            # kill the job
                            #pUtil.tolog("Going to kill pid %d" %lineno())
                            pUtil.createLockFile(True, self.__env['jobDic'][k][1].workdir, lockfile="JOBWILLBEKILLED")
                            killProcesses(self.__env['jobDic'][k][0], self.__env['jobDic'][k][1].pgrp)
                            self.__env['jobDic'][k][1].result[0] = "failed"
                            self.__env['jobDic'][k][1].currentState = self.__env['jobDic'][k][1].result[0]
                            self.__env['jobDic'][k][1].result[2] = self.__error.ERR_STDOUTTOOBIG
                            self.__env['jobDic'][k][1].pilotErrorDiag = pilotErrorDiag
                            self.__skip = True

                            # store the error info

                            updatePilotErrorReport(self.__env['jobDic'][k][1].result[2], pilotErrorDiag, "1",  self.__env['jobDic'][k][1].jobId, self.__env['pilot_initdir'])