log_trawl.py

Go to the documentation of this file.

#! /usr/bin/env python

import sys, os, stat
import argparse
from collections import OrderedDict
import code
import glob

# Permission mask: user r,w; group r
PERMISSION_RW_R = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
# Permission mask: user r,w; group r; others r
PERMISSION_RW_R_R = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH

def checkIsFile(path):
  if os.path.isfile(path):
    return 1
  print ("Warning, not a file: ", path)
  return 0

def setPermission(path, permission=PERMISSION_RW_R_R):
  if checkIsFile(path):
    os.chmod(path, permission)

def tail(path, window=20):
    f = open(path, 'r')
    BUFSIZ = 1024
    f.seek(0, 2)
    bytes = f.tell()
    size = window
    block = -1
    data = []
    while size > 0 and bytes > 0:
        if (bytes - BUFSIZ > 0):
            # Seek back one whole BUFSIZ
            f.seek(block*BUFSIZ, 2)
            # read BUFFER
            data.append(f.read(BUFSIZ))
        else:
            # file too small, start from begining
            f.seek(0,0)
            # only read what was not read
            data.append(f.read(bytes))
        linesFound = data[-1].count('\n')
        size -= linesFound
        bytes -= BUFSIZ
        block -= 1
    return '\n'.join(''.join(data).splitlines()[-window:])



class Process:
  # Constructor
  def __init__(self, num, proj):
    self.num     = num
    self.msgs    = []
    self.proj    = proj
    self.err     = FileViewer(self.errPath())
    self.out     = FileViewer(self.outPath())

  # Implementation of str()
  def __str__(self):
    return "Process " + str(self.num) + ", last message: \n" + self.lastMsg()

  # How the thing represents it self, notably for tostring and the "in" operator
  def __repr__(self):
    return str(self)

  # How the thing represents it self, notably for tostring and the "in" operator
  def __contains__(self, test):
    return test in self.lastMsg()

  # Implementation of ==
  def __eq__(self,other):
    return self.num == other.num and self.proj.logPath == other.proj.logPath

  # Get the number of messages in the .log file
  def nMsgs(self):
    return len(self.msgs)

  # Get the last message, usually the most interesting one
  def lastMsg(self):
    return self.msgs[len(self.msgs) - 1]

  # Add a message to the process, intended for use in the Project class
  # when parsing .log files
  def addMsg(self, msg):
    self.msgs.append(msg)

  # Print all messages for this process
  def printMsgs(self):
    for msg in self.msgs:
      print (msg)

  ## Get the path to the std err file
  def errPath(self):
    wildcard = self.proj.logPath.replace(".log", "*.%s.err" %(str(self.num)))
    files = glob.glob(wildcard)
    if len(files) != 1:
      raise Exception("Too many (%s) files found matching path %s"
                      % (len(files), wildcard))
    path = files[0]
    checkIsFile(path)
    return path

  ## Get the path to the std out file
  def outPath(self):
    wildcard = self.proj.logPath.replace(".log", "*.%s.out" %(str(self.num)))
    files = glob.glob(wildcard)
    if len(files) != 1:
      raise Exception("Too many (%s) files found matching path %s"
                      % (len(files), wildcard))
    path = files[0]
    checkIsFile(path)
    return path

  ## Tail the std err file
  def errTail(self, window=20):
    path = self.errPath()
    if not checkIsFile(path):
      return
    return tail(path, window)

  ## Tail the std out file
  def outTail(self, window=20):
    path = self.outPath()
    if not checkIsFile(path):
      return
    return tail(path, window)

  ## Open the std out file with less
  def errLess(self):
    path = self.errPath()
    if not checkIsFile(path):
      return
    print (os.system("less " + path))
    return

  ## Open the std err file with less
  def outLess(self):
    path = self.outPath()
    if not checkIsFile(path):
      return
    print (os.system("less " + path))
    return

  ## Find the return code from the last message, -1 if last message is not
  ## normal termination
  def retCode(self):
    if not "return value" in self.lastMsg():
      return -1
    start = self.lastMsg().find("return value")

    end  = self.lastMsg().find(")", start)
    valStr = self.lastMsg()[start:end]
    values = valStr.split()
    try:
      return int(values[2])
    except:
      raise Exception("Can't find return code, improperly fomatted message: %s"
                       % (self.lastMsg()))


states = ["normal", "abnormal", "incomplete"]

class Project:
  def __init__(self, logPath):
    # Path to log file
    self.logPath = logPath
    # List of all messages from .log file, not sorted by process
    self.msgs    = []
    # Main dictionary of processes, indexed by condor job number.
    self.procs   = OrderedDict()
    #
    self.decomp  = dict()
    # List of process objects that ended normally  (return value == 0 )
    self.normal = []
    # List of process objects that ended abnormally
    # (return value != 0 or any other last message)
    self.abnormal = []
    # List of process objects that are incomplete
    # (no return value, in progress or abnormally terminated)
    self.incomplete = []

    self.stateMap = OrderedDict()
    global states

    for state in states:
      self.stateMap[state] = []

    if self.logPath.endswith(".cmd"):
      print ("Condor .cmd file supplied, swapping extension for .log")
      self.logPath = logPath.replace(".cmd", ".log")
    self.separateMessages()
    self.processRecon()
    self.inspectProcs()
    self.decomposeModes()

  ## separateMessages() turns all of the separate messages in the file into
  ## entries in a list.
  ### Need to break the log into chunks.  Many messages from many jobs
  ### Example:
  ###  ...
  ###  009 (17365926.880.000) 08/06 21:16:09 Job was aborted by the user.
  ###        The system macro SYSTEM_PERIODIC_REMOVE expression ...
  ###         atus) > 86400*3)))' evaluated to TRUE
  ###  ...
  ### Seems we can rely on those dots, here we go
  def separateMessages(self):
    log = open(self.logPath, 'r')
    msg = ""
    for line in log:
      if line.strip() == "...":
        self.msgs.append(msg) # append the current message to the list
        msg = "" # Reset the current message
      else:
        msg += line


  ### processRecon() loops over all the messages to make Process objects, fills
  ### a dict.
  ### Relies on separateMessages() to have been called first, but that happens
  ### properly in the constructor
  def processRecon(self):
    for msg in self.msgs:
      smsg = msg.split()

      if len(smsg) > 2:
        if "028" in smsg[0]:
          continue
        if smsg[1][0] == "(":
          numStr = smsg[1]
          num = int(numStr.split(".")[1])
          if num in self.procs.keys():
            self.procs[num].addMsg(msg)
          else:
            self.procs[num] = Process(num, self)
            self.procs[num].addMsg(msg)

  ### Inspect the processes, make sure the numbers jive and decompose them into
  ### "normal" and "abnormal"
  def inspectProcs(self):
    print ("Found",  self.nProcs(), "processes.")
    lastKey = None
    sequential = True
    firstKey = None
    for  key in sorted(self.procs.keys()):
      if lastKey == None:
        lastKey = key
        firstKey = key
        continue
      diff = key - lastKey
      if not diff == 1:
        sequential = False
        break
      lastKey = key

    if sequential and firstKey == 0:
      print ("Process numbers are sequential starting with %s, things seem ok."\
            % (str(firstKey)))
    elif sequential:
      print ("Process numbers are sequential, but do not start with 0.  " + \
            "Be careful with your indexing.")
      print ("First process number is %s." % (str(firstKey)))
    else:
      print ("Process numbers are not sequential, be careful with your indexing.")


    for proc in self.procs.values():
      if proc.retCode() == 0:
        self.stateMap["normal"].append(proc)
      elif proc.retCode() > 0:
        self.stateMap["abnormal"].append(proc)
      else:
        self.stateMap["incomplete"].append(proc)

    self.normal     = self.stateMap["normal"]
    self.abnormal   = self.stateMap["abnormal"]
    self.incomplete = self.stateMap["incomplete"]
    print (len(self.stateMap["normal"]), \
          "of those processes ended normally with status 0")
    print (len(self.stateMap["abnormal"]), \
          "of those processes ended normally but with non-zero status")
    print (len(self.stateMap["incomplete"]), \
          "of those processes are either still in progress ended abnormally")

  ##
  def decomposeModes(self):
    for proc in self.procs.values():
      splitMsg = proc.lastMsg().split()
      try:
        mode =  int(splitMsg[0])
      except:
        raise Exception("Could not find mode (three digits) at the beginning" +\
                        "of this message: " + str(proc))

      if not mode in self.decomp.keys():
        self.decomp[mode] = []

      self.decomp[mode].append(proc)
    sum = 0
    for mode in self.decomp.values():
      sum += len(mode)
    if not sum == self.nProcs():
      raise Exception("Sum of decomposed processes not equal to total number "+\
                      "of processes.")

  def printModes(self):
    print ("Jobs have", len(self.decomp), "different modes.  "+\
          "Here is an example of each.")
    for mode in self.decomp.keys():
      print ("* Mode", mode, "-- there are", len(self.decomp[mode]), \
            "of these, one example is:")
      for line in str(self.decomp[mode][0]).splitlines():
        print ("\t", line)
      print ("")

  ## Number of process objects in dict
  def nProcs(self):
    return len(self.procs)

  ## Implementation of [] operator for indexing
  def __getitem__(self, n):
    return self.procs[n]

  # Implementation of len(), returns number of items in dictionary
  def __len__(self):
    return len(self.procs)

  # Implementation of iter, use as "for thing in proj"
  def __iter__(self):
    for key in self.procs.keys():
      yield self.procs[key]

  ## Make .log and all .err and .out group readable and user read/writable
  def makeReadable(self, permission=PERMISSION_RW_R_R):
    setPermission(self.logPath, permission)
    for proc in self.procs.values():
      setPermission(proc.outPath(), permission)
      setPermission(proc.errPath(), permission)



def findLogInDirectory(logDir):
  logDirContents = os.listdir(logDir)
  for fileName in logDirContents:
    if fileName.endswith(".log"):
      return os.path.join(logDir,fileName)
  return None



def retreiveLogFromJobSubClient(jobId):
  print ("JobSub Client Job ID supplied, finding logs.")
  if "CONDOR_TMP" in os.environ and os.path.isdir(os.environ["CONDOR_TMP"]):
    print ("Found $CONDOR_TMP, logs will be stored there. ")
    fetchLocation = os.environ["CONDOR_TMP"]
  else:
    print ("Failed to find $CONDOR_TMP, fetching logs to current directory.")
    fetchLocation = "./"

  logDir = os.path.join(fetchLocation, jobId)

  if not os.path.isdir(logDir):
    os.mkdir(logDir)
  print ("Directory:", logDir)
  extantLogFile = findLogInDirectory(logDir)
  if extantLogFile:
    print ("Log file previously fetched, location:", extantLogFile)
    return extantLogFile

  else:
    print ("Fetching log files...")
    os.system("jobsub_fetchlog --jobid " + jobId + " --unzipdir " + logDir)

  fetchedLogFile = findLogInDirectory(logDir)
  if fetchedLogFile:
    print ("Log file location:", fetchedLogFile)
    return fetchedLogFile
  else:
    raise Exception("Failed to filnd log file after fetching, " + \
                    "something went wrong.")

class FileViewer:
  def __init__(self, path):
    self.path = path
  def __repr__(self):
    print (self.path)
    if not checkIsFile(self.path):
      return
    os.system("less " + self.path)
    return ""







def main(argv):
  parser = argparse.ArgumentParser(description="""
            Loop over the ".log" file and find the final state of each
            process, then find the .err and .out""")
  parser.add_argument('logFile', help="""
    The input file to run over, or a jobsub_client jobid.
    In the jobid case, the files are fetched to CONDOR_TMP if it is defined,
    or falls back to the current working directory if not defined.  If the log
    has already been fetcheded, it just finds it there.""", type=str)

  args = parser.parse_args()
  logPath = args.logFile

  if not os.path.isfile(logPath) and "@" in logPath and ".fnal.gov" in logPath:
    logPath = retreiveLogFromJobSubClient(logPath)


  proj = Project(logPath)
  return proj



if __name__ == "__main__":
  proj = main(sys.argv)
  code.interact(local=locals(), banner="")