JobAttemptSummary.py

Go to the documentation of this file.

import re

from GBSTimeStamp      import TimeStampToUnixTime

#  This file defines:-
#
#  The function:  ParseLogEntry
#  The class:     JobAttemptSummary
#
#  both are components LogAnalyser (q.v.)

def ParseLogEntry(log_line):
    """Parse global log file line entry holding job submission information.

    For valid entries returns a tuple containing:-

      log_ut  Log file line Unix time  
      jas_id  Job submission entry: <task>.<job-try-id>
      event   Data recorded on line

    For invalid entries: return None.
    """
    
    mo = re.search(r'(\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d)\s+\S+\s+\d+\s+(\S+)\s+(\S+?\d+)\s+(.+)',log_line.strip())
    if not mo: return None
    log_ut = TimeStampToUnixTime(mo.group(1))
    task   = mo.group(2)
    try_id = mo.group(3)
    if try_id[0:5] == "JTID:": try_id = try_id[5:]
    
    jas_id = "%s.%s" % (task,try_id)
    event  = mo.group(4)
    return (log_ut,jas_id,event)

#_______________________________________________________________________________________________

 

class JobAttemptSummary:
    """An object to collect all the global log information for a single job submission
    and present a summary of it."""

    def __init__(self,parent,log_line):
        """Initialise object using data from log line."""

        self.parent = parent
        self.debug  = parent.debug  #Propagate debug level to child.

        # State. Always one of:-
        self.STATE_ACTIVE          = 1
        self.STATE_SUCCESS         = 2
        self.STATE_PARTIAL_SUCCESS = 3
        self.STATE_FAILURE         = 4

        self.state = self.STATE_ACTIVE

        # Failure type.  Only defined if self.state == self.STATE_FAILURE
        self.FAIL_LOST  = 1
        self.FAIL_GRID  = 2
        self.FAIL_USER  = 3

        self.fail = None

        self.cpu              = None   # CPU time if known
        self.submit_ut        = None   # Submit Unix time
        self.start_ut         = None   # Start Unix time
        self.id               = ""     # JAS Id (see ParseLogEntry)
        self.request_new_seed = False  # Set True if job attept ends by requesting new seed.
        
        self.ksi2k_rating = 1.23 # Based on 700 RAL T1 node values obtained in Jun 2008.
        
        parse_results = ParseLogEntry(log_line)
        if not parse_results:
            print "Creating JAS with bad line: " + log_line
            self.Terminate(self.STATE_FAILURE,self.LOST)
            return

        # Process first line - which should have event == "submitted"
        (self.submit_ut,self.id,event) = parse_results
        if self.debug: print "    creating JAS %s with %s" % (self.id,event)
        if event == "submitted": return
        print "Initialising JAS %s with wrong line: %s" % (self.id,log_line)
        self.parent.IncrementNonJasErrorCount("Initialising job summary with wong line")
        self.Update(log_line)
    
#_______________________________________________________________________________________________


    def GetCpu(self):        return self.cpu
    def GetCpuKSI2K(self):
        if self.cpu: return self.cpu*self.ksi2k_rating
        return None
    def GetFailType(self):   return self.fail
    def GetId(self):         return self.id
    def GetStartDate(self):  return self.start_ut
    def GetSubmitDate(self): return self.submit_ut
    
#_______________________________________________________________________________________________


    def IsActive(self):          return self.state == self.STATE_ACTIVE
    def IsFailure(self):         return self.state == self.STATE_FAILURE
    def IsPartialSuccess(self):  return self.state == self.STATE_PARTIAL_SUCCESS
    def IsSuccess(self):         return self.state == self.STATE_SUCCESS
    
#_______________________________________________________________________________________________

    def RequestNewSeed(self):    return self.request_new_seed
#_______________________________________________________________________________________________

    def Terminate(self,state=None,fail=None):
        """Terminate object and, if still in parent's active list, move to complete."""
        # If no arg supplied treat as forced termination due to end of log files or
        # start of replacement with same ID.
        if not state:
            state = self.STATE_FAILURE
            fail  = self.FAIL_LOST
        self.state = state
        self.fail  = fail
        if self.parent.ActiveJAS.has_key(self.id):
            del self.parent.ActiveJAS[self.id]
            self.parent.CompletedJAS.append(self)
            if self.debug: print "      moved JAS %s to completed" % self.id
            
#_______________________________________________________________________________________________

    def Update(self,log_line):
        """Add log information."""

        ##  Make sure that line refers to this object.

        parse_results = ParseLogEntry(log_line)
        if not parse_results:
            print "Updating JobAttemptSummary with bad line: " + log_line
            self.parent.IncrementNonJasErrorCount("Cannot parse job summary line")
            return

        if not self.IsActive():
            print "Updating JobAttemptSummary %s, which is no longer active, with line: %s" % (self.id,log_line)
            self.parent.IncrementNonJasErrorCount("Job summary line for inactive job")
            return

        (submit_time,id,event) = parse_results
        if id != self.id:
            print "Updating JobAttemptSummary %s with wrong line: %s" % (self.id,log_line)
            self.parent.IncrementNonJasErrorCount("Job summary line for wrong job")
            return

        if self.debug: print "      updating %s with %s" % (self.id,event)

        ## Deal with all non terminating events.
        
        if event == "submitted":
            print "Unexpectedly updating existing JobAttemptSummary %s with line: %s" % (self.id,log_line)
            self.parent.IncrementNonJasErrorCount("Missing job start line")
            self.submit_time = submit_time
            return

        if re.search("started\s+\S+",event):
            mo = re.search("started\s+(.*)",event)
            self.start_ut = TimeStampToUnixTime(mo.group(1))
            return

        if    re.search("^No output files",event) \
           or re.search("^Failed to retrieve output",event) \
           or re.search("^Failed to unpack",event) \
           :
            self.fail = self.FAIL_LOST
            return

        if event == "killed by user": return

        ## Deal with all terminating events.

        if    re.search("^lost Ganga job with ID",event) \
           or re.search("^killed Ganga job",event):
            self.Terminate(self.STATE_FAILURE,self.FAIL_LOST)

        if re.search("^returned\s+\S+",event):
            mo = re.search("CPU:\s+(\S+)\s+min",event)
            if mo: self.cpu = float(mo.group(1))
            mo = re.search("KSI2K=(\S+)",event)
            if mo:
                self.ksi2k_rating = float(mo.group(1))
            
            mo = re.search("\S+\s+(\S+)\s+(\S+)",event)
            (status,com_level) = mo.groups()

            if status == "SUCCEEDED":
                self.Terminate(self.STATE_SUCCESS)
                return

            # Treat RETRY at USER level communication which did not signal RESTART as a
            # partial success.
            if status == "RETRY" and  com_level == "USER" and not re.search("RESTART",event):
                self.Terminate(self.STATE_PARTIAL_SUCCESS)
                if re.search("NEW_SEED",event): self.request_new_seed = True
                return

            # Treat all other HELD, RETRY and FAILED as failures with type based on
            # communication level, unless already set
            if status == "HELD" or status == "RETRY" or status == "FAILED":
                if not self.fail:
                    self.fail = self.FAIL_GRID
                    if com_level == "USER" or com_level == "APPLICATION": self.fail = self.FAIL_USER
                self.Terminate(self.STATE_FAILURE,self.fail)
                return

            print "Unhandled JAS update : %s" % event
            
                

---