LogAnalyser.py

Go to the documentation of this file.

import datetime
import glob
import os
import re
import sys


from GBSTimeStamp      import TimeStampToUnixTime
from JobAttemptSummary import JobAttemptSummary, ParseLogEntry

def LastMonthArgs():
    """Return tuple (start,end,html_file_name) that are suitable to feed into
    LogAnalyser to produce last month's statistics."""

    today      = datetime.date.today()
    end_date   = today - datetime.timedelta(today.day)
    start_date = end_date - datetime.timedelta(end_date.day-1)
    return(start_date.isoformat(),end_date.isoformat(),start_date.strftime("%b_%Y").lower()+".html")

class LogAnalyser:
    
    """An object to analyse a series of directories holding GBS global
    log files within a date window and produce a report on overall
    performance, including numbers MINOS UK has to supply to GridPP as
    their 'GRID Metrics'.

    Analysis contains all the jobs submitted during the date window including
    those that end after the end of the window.  So to fully analyse all jobs,
    it may be necessary to include log files that postdate the date window.
    A warning will be issued for any job that is not complete unless a log
    file is processed that is at least 7 days later that the job start time. 

    Internally the analyser records information about each individual
    job submissions into a

        JobAttemptSummary (JAS)

    objects. A new object is created in the self.ActiveJAS pool each time a
    "submitted" line in the log file is encountered and recieves
    further log file lines for the same job attempt until it receives
    on that terminates it at which time it tells the LogAnalyser to move
    it to the self.CompletedJAS pool"""

    def __init__(self,start_date,end_date,directories,html_file=None,title="MINOS Grid Production"):
        """Initialise analyser:-
        
        start_date    Start date string e.g. "2008-01-01"
        end_date      End date string   e.g. "2008-04-01"
        directories   List of directory names e.g. ['/home/west/log_files/set1','/home/west/log_files/set2']
        title         Optional title for summary.
        html_file     Name of HTML file (if None will write plain text to STDOUT)
        """

        self.debug = 0    #Switch on/off by hand.

        # The maximum number of seconds between submission and job completion.  This
        # number is used to:-
        # o  Flag up possibly missing start records if only see end record more
        #    than limit after start record.
        # o  Terminate any JAS that has run that long without seeing an end record.

        self.max_job_duration = 10*24*60*60

        if start_date > end_date: (start_date,end_date) = (end_date,start_date)
        self.start_ts = start_date
        if not re.search(r'\d\d\d\d-\d\d-\d\d',start_date):
            print "Bad start date " + start_date
            return
        self.start_ut = TimeStampToUnixTime(start_date)
        if not self.start_ut: return
        
        self.end_ts = end_date
        if not re.search(r'\d\d\d\d-\d\d-\d\d',end_date):
            print "Bad end date " + end_date
            return
        self.end_ut = TimeStampToUnixTime(end_date)
        if not self.end_ut: return

        self.directories = directories
        for dir in self.directories:
            if not os.path.isdir(dir):
                print 'Cannot find directory ' + dir
                return

        self.html_file      = html_file
        self.title          = title
        self.non_jas_errors = {} # non-jas-error-description -> number of errors
        

        # Prepare 2 pools of JobAttemptSummary (JAS) objects:-
        #
        #  ActiveJAS:    hash JAS-name -> obj
        #  CompletedJAS: list of completed JAS (not a hash; JAS-name may be duplicated)

        self.ActiveJAS    = {}
        self.CompletedJAS = []

        # Perform analysis
        self.Run()
        return
#_______________________________________________________________________________________________

    def AnalyseCollectedJAS(self):
        """Analyse the collection of JobAttemptSummary (JAS) objects."""


        # Prepare to write to file or stdout.
        if self.html_file:
            html = 1
            file_html = open(self.html_file,"w")
            tr = "<tr><td>"
            td = "<td>"
            br = "<br>"
            pg = "<p>"
        else:
            html = 0
            file_html = open("/dev/tty","w")
            tr = ""
            td = "  "
            br = "\n"
            pg = "\n\n"
            
        pr = file_html.write
            
        num_days = int( 1.5 + (self.end_ut - self.start_ut)/(24*60*60)) # Time window is inclusive

        title = self.title
        if not title: title = "GBS Global Log File Analysis"
        title += " for period %s to %s (%d days)\n" % (self.start_ts,self.end_ts,num_days)
        if html: pr("<html>\n<title>%s</title>\n<body>\n<h1>%s</h1>\n" % (title,title))
        else:    pr("\n\n\n%s\n" % title)

        pr("%sBased on log files analysed from:-%s" % (pg,br))
        if html: pr("<pre>\n")
        for dir in self.directories: pr("  %s\n" % dir)
        if html: pr("</pre>\n")

        num_tot = len(self.CompletedJAS)
        num_success   = 0
        num_p_success = 0
        num_fail      = 0
        num_new_seed  = 0

        num_latency   = 0
        sum_latency   = 0

        sum_cpu       = 0
        sum_good_cpu  = 0
        sum_bad_cpu   = 0

        sum_fail_lost = 0
        sum_fail_grid = 0
        sum_fail_user = 0

        for jas in self.CompletedJAS:
            if jas.IsSuccess():          num_success   += 1
            elif jas.IsPartialSuccess(): num_p_success += 1
            elif jas.IsFailure():        num_fail      +=1
            else:
                print "debug job not classified"
                self.IncrementNonJasErrorCount("Log analysis bug - investigate!")
            if jas.RequestNewSeed():     num_new_seed += 1
            cpu = jas.GetCpuKSI2K()
            if cpu:
                sum_cpu  += cpu
                if jas.IsFailure(): sum_bad_cpu  += cpu
                else:               sum_good_cpu += cpu
            if jas.IsFailure():
                if   jas.fail == jas.FAIL_LOST: sum_fail_lost += 1
                elif jas.fail == jas.FAIL_GRID: sum_fail_grid += 1
                elif jas.fail == jas.FAIL_USER: sum_fail_user += 1
                else:
                    print "debug %s " % str(jas.fail)
                    self.IncrementNonJasErrorCount("Log analysis bug - investigate!")
             
            submit_ut = jas.GetSubmitDate()
            start_ut  = jas.GetStartDate()
            if submit_ut and start_ut:
                num_latency += 1
                sum_latency += start_ut - submit_ut

            num_not_success   = num_fail + num_p_success
            num_tot_grid_fail = sum_fail_lost + sum_fail_grid
            
        pr("\nNumber of job submissions analysed: %d (%5.1f/day)\n\n" % (num_tot,num_tot/num_days))
        if not num_tot: return

        if html: pr("<h2>GRID Metrics</h2>\n")
        else:    pr("\nGRID Metrics\n")
        if html: pr("<p><table border=1><tr><th>Metric<th>Value\n")
            
        pr("%sThroughput (1) %s%5.1f KSI2K \n"     % (tr,td,sum_good_cpu/(60.*24*num_days)))
        pr("%sRobustness (2) %s%5.1f%%\n"          % (tr,td,(100. - 100.*num_tot_grid_fail/num_tot)))
        latency = "No data"
        if num_latency: latency = "%5.1f mins" % (sum_latency/(num_latency*60))
        pr("%sLatency (3)    %s%s\n"               % (tr,td,latency))
        pr("%sWastage (4)    %s%5.1f%%\n"          % (tr,td,100.*sum_bad_cpu/(max(sum_cpu,.001))))


        if html:
            pr("</table>\n<p><font size=-1>Where:-\n<ol>\n")
            pr("<li><b>Throughput</b> The total number of useful KSI2K CPU days delivered per day\n")
            pr("<li><b>Robustness</b> The fraction of submitted jobs that were not GRID related failures\n")
            pr("<li><b>Latency</b> The mean time between submission and start of execution\n")
            pr("<li><b>Wastage</b> The fraction of total CPU known to be wasted on jobs that failed.\n")
            pr("</ol></font>\n")

        if num_not_success:
            if html: pr("<h2>Job Failure Analysis</h2>\n")
            else:    pr("\nJob Failure Analysis\n\n")
            pr("Of the %5.1f%% that were not successful (including the %5.1f%% that were GRID failures)%s" %\
               ( 100.*num_not_success/num_tot,100.*num_tot_grid_fail/num_tot,pg ))
            if html: pr("<p><table border=1><tr><th>Error<th>Rate\n")

            pr("%sJob failed to run or no output returned%s%5.1f%%\n" % (tr,td,100.*sum_fail_lost/num_not_success))
            pr("%sJob aborted (failed to exit wrapper normally)%s%5.1f%%\n" % (tr,td,100.*sum_fail_grid/num_not_success))
            pr("%sApplication returned non-zero exit code%s%5.1f%%\n" % (tr,td,100.*sum_fail_user/num_not_success))
            pr("%sApplication only partially successful%s%5.1f%%\n" % (tr,td,100.*num_p_success/num_not_success))
            pr("%sApplication requested new seed%s%5.1f%%\n" % (tr,td,100.*num_new_seed/num_not_success))
            if html: pr("</table><p>\n")
        
        if html: pr("<h2>Non-Job Failure Analysis</h2>\n<p><table border=1><tr><th>Error<th>Total<th>Rate\n")
        else:    pr("\nNon-Job Failure Analysis\n\n")

        errors = self.non_jas_errors.keys()
        errors.sort()
        for error in errors:
            num_errors = self.non_jas_errors[error]
            pr("%s%s%s%d%s%5.1f%%\n" % (tr,error,td,num_errors,td,(100.*num_errors/num_tot)))
        if html: pr("</table><p>\n")

        if not errors: pr("%s - No errors recorded%s" % (br,br))
 
        file_html.close()
        if html: print '\nResults written to %s' % self.html_file
        

#_______________________________________________________________________________________________

    def AnalyseNonJasLine(self,line):
        """Analyse non-JAS line and count errors.

        Look at any line that is not part of a job attempt summary that is an error or fatal"""
        mo = re.search(r' (-\w-)\s+\d+\s+(.+)',line)
        if not mo:
            print "Cannot analyse non-JAS line %s " % line
            self.IncrementNonJasErrorCount("Cannot analyse non-job-attempt line")
            return
        (severity,error) = mo.groups()
        if severity == "-I-": return
        if   re.search("Removing orphaned Ganga job",line):
            self.IncrementNonJasErrorCount("Removing orphaned Ganga job")
        elif re.search("Failed to complete Ganga job submit",line):
            self.IncrementNonJasErrorCount("Failed to complete Ganga job submit")
        elif re.search("Ignoring malformed line:",line):
            self.IncrementNonJasErrorCount("Ignoring malformed line:")
        elif re.search("Timeout reached before asynchronous call ended",line):
            self.IncrementNonJasErrorCount("Timeout reached before asynchronous call ended")
        # The "Timeout reached ..." is always followed by "...Attempting to delete .." so ignore that
        elif re.search("\.\.\.Attempting to delete child processes ",line):
            pass
        elif re.search("Unknown Ganga status",line):
            self.IncrementNonJasErrorCount("Unknown Ganga status")
        elif re.search("GBSIOHelper: EOF on <open file",line):
            self.IncrementNonJasErrorCount("GBSIOHelper: EOF on <open file")
        else:
            print "Cannot handle non-JAS error: %s" % error
            self.IncrementNonJasErrorCount("Cannot handle non-JAS error")
 
#_______________________________________________________________________________________________

    def IncrementNonJasErrorCount(self,description):
        """Increment non-JAS error count."""
        if not self.non_jas_errors.has_key(description): self.non_jas_errors[description] = 0
        self.non_jas_errors[description] += 1

#_______________________________________________________________________________________________

    def ProcessDirectory(self,dir):
        """Process all gbs_global_*.log files in 'dir' and records results in set of JAS objects."""

        print "Processing directory %s ... " % dir

        log_files = glob.glob("%s/gbs_global_*.log" % dir)
        log_files.sort()

        for log_file in log_files:

            log_file_name = os.path.basename(log_file)
            mo = re.search(r'_(\d\d\d\d-\d\d-\d\d).log',log_file_name)
            if not mo:
                print "  skipping %s (does not end YYYY-MM-DD.log)" % log_file_name
                continue
            log_ts = mo.group(1)

            # Skip log files that predate the start date and quit once
            # the end date has been past when the pool of active JAS
            # are exhausted.
            
            if log_ts < self.start_ts:
                print "  skipping %s (predates start date %s)" % (log_file_name,self.start_ts)
                continue
            if log_ts > self.end_ts and not self.ActiveJAS:
                print "  skipping remaining log files as they postdate end date %s and there are no incomplete jobs" %\
                      self.end_ts
                return

            self.ProcessLogFile(log_file,log_ts <= self.end_ts)

            # Terminate any JAS that should have completed by now
            expire_ut = TimeStampToUnixTime(log_ts) - self.max_job_duration
            jas_list = self.ActiveJAS.keys()
            for jas_id in jas_list:
                jas = self.ActiveJAS[jas_id]
                if jas.GetSubmitDate() < expire_ut: jas.Terminate()

        # If there are any active JAS they will have to be terminated.
        if self.ActiveJAS:
            print "  Terminating any remaining active JAS ..."
            jas_list = self.ActiveJAS.keys()
            num_term = 0
            for jas_id in jas_list:
                
                if self.debug: print "    Terminating " + jas_id
                self.ActiveJAS[jas_id].Terminate()
                self.IncrementNonJasErrorCount("Missing job end line")
                num_term += 1
            print "  ... %d terminated." % num_term


#_______________________________________________________________________________________________

    def ProcessLogFile(self,log_file,start_new_jas):
        """Process log file 'log_file' and record results in set of JAS objects.

        Create new JAS only if 'start_new_jas' True."""
        
        log_file_name = os.path.basename(log_file)
        print "  processing " + log_file_name
        file_log = open(log_file)
        for log_line in file_log:

            parse_results = ParseLogEntry(log_line)
            if not parse_results:
                self.AnalyseNonJasLine(log_line)
                continue

            (jas_id,event) = parse_results[1:3]

            #HACK: There was a bug (fixed at the end of May 2008) recording the try_id with try number 0
            #HACK: when reporting "No output files".  See if any active JAS matches except for try number
            #HACK: and fix up line if so.
            if jas_id[-1] == '0':
                for ji in self.ActiveJAS.keys():
                    if jas_id[:-1] == ji[:-1]:
                        log_line = re.compile('\.0 ').sub('.%s ' % ji[-1],log_line)
                        parse_results = ParseLogEntry(log_line)
                        jas_id = ji
                        break

            # "submitted" entries always start a new JAS.  Jobs that get stuck
            # with Ganga state "submitted" will eventually get killed by GBS
            # and we want to record the failed attempt.
            if event == "submitted" and self.ActiveJAS.has_key(jas_id):
                jas = self.ActiveJAS[jas_id]
                jas.Terminate()

            # Ignore non "submitted" entries for non-existant JAS if within maximum
            # job lifetime of the start window - they are probably jobs that were already
            # running before the window opened.
            if         event != "submitted" \
                   and not self.ActiveJAS.has_key(jas_id) \
                   and (parse_results[0] - self.max_job_duration) < self.start_ut: continue
                
            if self.ActiveJAS.has_key(jas_id): self.ActiveJAS[jas_id].Update(log_line)
            elif start_new_jas: self.ActiveJAS[jas_id] = JobAttemptSummary(self,log_line)
        
#_______________________________________________________________________________________________

    def Run(self):
        """Perform analysis."""

        for dir in self.directories:
            self.ProcessDirectory(dir)
            print "Collected %d job summaries so far" % len(self.CompletedJAS)

        self.AnalyseCollectedJAS()
        
        

        

        

        

---