#!/usr/bin/env python3 import os import re from shutil import copyfile if 'tethysdata' in os.uname()[1]: # Force matplotlib to not use any Xwindows backend. import matplotlib matplotlib.use('Agg') import pandas as pd from scripts.dlist_tools import DList from scripts.syslog_tools_io import SysIO class SysTools(DList, SysIO): """Toolbox for processing LRAUV syslogs. Notes: - SysTools eliminates redundant critical failure messages by setting time bounds on repeating errors. Filtering settings are defined in MESSAGE FILTERING SETTINGS section below. - If working from remote (i.e., not on tethysdata) SysTools will make a temporary local copy of the syslog it's processing and remove it when done (determined to be faster than reading over network). - SysTools cashes a json summary of the processed data for each vehicle. Upon execution SysTools loads the json files and will only process new deployments. You can force a fresh scan by executing lrauv-log-critical.py with the --force flag. - By default, all data products are exported to ../data/ . """ def __init__(self): self.host = os.uname()[1] # hostname self.sys_local = './data/syslog.tmp' # path for local copy of syslog (only used when working from remote) self.work_logname = '' # name of log being processed # initialize inherited objects super().__init__() # initialize runtime variables self.log_start_time = None self.log_end_time = None self.log_runtime = None self.total_runtime = pd.Timedelta(0) # initialize DataFrame variables self.logs = None # record runtime and MTBCF of processed logs self.critical = None # record non-redundant critical messages found in syslogs self.critical_all = None # record ALL critical messages found in syslogs # MESSAGE FILTERING SETTINGS # ------------------------------------------------------------------------------------------- self.ignore = ['SBIT', 'TODO'] # components to ignore self.cbit_fail = ['WATCHDOG', 'Environmental', 'ESP', # failures in these components are reported under CBIT 'STOP DEPTH REACHED', 'ABORT DEPTH REACHED', 'Backplane'] # scan CBIT msgs for these key-words and replace the component name self.min_tbcf = 17 # min time allowed between failure msgs of regular components self.min_tbcf_long = 6 * 60 # min time allowed between failure msgs of repeating components self.lookup_min_tbcf = { # defines min time allowed for specific components 'DROPWEIGHT': self.min_tbcf_long, 'ENVIRONMENTAL': self.min_tbcf_long, 'IRIDIUM:A_TIMEOUT:B': self.min_tbcf_long, 'NAL9602': 3 * 60, 'VERTICALCONTROL': 3 * 60, } # ------------------------------------------------------------------------------------------------- # Utility methods # ------------------------------------------------------------------------------------------------- def already_processed(self, log): return self.logs['log'].str.contains(log, case=True).any() def clear_dataframes(self): self.log_start_time = None self.log_end_time = None self.log_runtime = None self.reset_total_runtime() def reset_total_runtime(self): self.total_runtime = pd.Timedelta(0) def clear_runtime(self): self.log_start_time = None self.log_end_time = None self.log_runtime = None self.work_logname = '' def rm_localsys(self): if os.path.isfile(self.sys_local): os.remove(self.sys_local) # I/O (calls methods defined in SysIO) # ------------------------------------------------------------------------------------------------- def load_dataframes(self, vehicle, force_fresh_scan=False): """Load pandas DataFrames form json files. Try to load the pandas DataFrames form json files. If load fails or if a fresh scan is forced, initialize empty pandas DataFrames and start from scratch. :param vehicle: string containing vehicle name :param force_fresh_scan: override load and force a fresh scan of syslogs (bool) :return: update state of class attributes logs, critical, critical_all (pandas DataFrame) """ # clear pre-existing data self.clear_dataframes() try: # force fresh scan by raising exception and executing except block if force_fresh_scan: raise ValueError() # load the pandas DataFrames form json files self.logs, self.critical, self.critical_all = self.load_json(vehicle) # update total runtime to account for vehicle history self.total_runtime = self.logs['runtime'].sum() except ValueError: # load failed... initialize the pandas DataFrames and start from scratch self.logs, self.critical, self.critical_all = self.init_dataframes() def export_dataframes_json(self, vehicle): self.export_json(self.logs, self.critical, self.critical_all, vehicle) def export_dataframes_csv(self, vehicle, yr=''): self.export_csv(self.logs, self.critical, self.critical_all, vehicle, yr) # Update self # ------------------------------------------------------------------------------------------------- def update_critical(self, log, comp, timestamp, runtime, msg, redundant=True): """Update critical DataFrames - keep track of processed critical messages. Log all critical messages in critical_all, and non-redundant messages in critical. Compute time between critical failures (tbcf) for non-redundant messages. :param log: working log name (string) :param comp: name of failed component (string) :param timestamp: critical message timestamp (datetime64[ns]) :param runtime: critical message runtime (dimedelta64[ns]) :param msg: message associated with critical failure (string) :param redundant: True if message is redundant (bool) :return: update class attributes critical and critical_all """ # add row to critical_all DataFrame self.critical_all = self.critical_all.append( { 'vehicle': self.dlist['vehicle'], 'year': self.dlist['year'], 'deployment_name': self.dlist['name'], 'deployment': self.dlist['dep'], 'log': log, 'comp': comp, 'timestamp': timestamp, 'runtime': runtime, 'msg': msg }, ignore_index=True) if not redundant: if self.critical.empty: last_cr_runtime = pd.Timedelta(0) else: # retrieve runtime of last critical message for tbcf last_cr_runtime = self.critical.iloc[-1]['runtime'] # add row to critical_all DataFrame self.critical = self.critical.append( { 'vehicle': self.dlist['vehicle'], 'year': self.dlist['year'], 'deployment_name': self.dlist['name'], 'deployment': self.dlist['dep'], 'log': log, 'comp': comp, 'timestamp': timestamp, 'runtime': runtime, 'msg': msg, 'tbcf': runtime - last_cr_runtime }, ignore_index=True) def update_logs(self): """Update logs DataFrame - keep track of processed logs. The logs DataFrame keeps track of the following: - runtime: log runtime (dimedelta64[ns]) - fail_count: number of critical failures found in log (int) - mtbcf: mean time between critical failures for current log (dimedelta64[ns] or NaN if 0) :return: update class attribute logs """ # index critical failures found in current log log_index = self.critical['log'].str.contains(self.work_logname) # add row to logs DataFrame self.logs = self.logs.append( { 'vehicle': self.dlist['vehicle'], 'year': self.dlist['year'], 'deployment_name': self.dlist['name'], 'deployment': self.dlist['dep'], 'log': self.work_logname, 'runtime': self.log_runtime, 'fail_count': log_index.sum(), 'mtbcf': self.critical.loc[log_index]['tbcf'].mean() }, ignore_index=True) # Syslog runtime # ------------------------------------------------------------------------------------------------- def syslog_start_time(self, line): """Retrieve timestamp from syslog line. :param line: byte string :return: update class attribute log_start_time: timedelta64[ns] if succeed """ t0 = None if len(line.strip()) >= 23: try: t0 = pd.to_datetime(line[:24].decode()) except ValueError: pass if t0 is not None: self.log_start_time = t0 def syslog_end_time(self, f, buf_size=512): """Retrieve last timestamp from syslog. Read file from end to beginning until timestamp is located. :param f: open file ID :param buf_size: read buffer size (int) :return: update class attribute log_end_time: timedelta64[ns] if succeed """ block_number = -1 f.seek(0, 2) # move read pointer to end of file f_size_left = f.tell() # current position of the read pointer t1 = None # read blocks of size buf_size, in reverse order starting from the end of the file while t1 is None and f_size_left > 0: try: # move read pointer to next block (offset from end of file) f.seek(block_number * buf_size, os.SEEK_END) except IOError: # file too small, start from beginning f.seek(0) lines = f.readlines(buf_size) for l in reversed(lines): if len(l.strip()) >= 23: try: t1 = pd.to_datetime(l[:24].decode()) break except ValueError: continue block_number -= 1 f_size_left -= buf_size if t1 is not None: self.log_end_time = t1 def syslog_runtime(self): """Compute log runtime and accumulate total vehicle runtime. :return: update class attributes log_runtime and total_runtime (timedelta64[ns]) """ self.log_runtime = (self.log_end_time - self.log_start_time) self.total_runtime += self.log_runtime # Parse syslog # ------------------------------------------------------------------------------------------------- def redundant_check(self, comp, timestamp): """Determine if a critical message is redundant. A message is deemed redundant if the time between the last critical message (of same component 'comp') is under the threshold for that component (defined in min_tbcf and lookup_min_tbcf) . :param comp: failed component (string) :param timestamp: timestamp of critical message (datetime64[ns]) :return: True if message is redundant """ redundant = False # retrieve last critical message of component 'comp' from DataFrame last_cr = self.critical_all[self.critical_all['comp'].str.contains(comp, case=False)].tail(1) if not last_cr.empty: # time interval between last message to current one delta_t = (timestamp - last_cr.iloc[0]['timestamp']).total_seconds() / 60 # retrieve time threshold for component min_delta_t = next((val for key, val in self.lookup_min_tbcf.items() if key in comp.upper()), self.min_tbcf) if delta_t < min_delta_t: redundant = True return redundant def parse_critical_messages(self, lines): """Parse and record syslog lines containing critical failure messages. From each line extract the following: - timestamp (datetime64[ns]). - runtime: total runtime + time-span from beginning of log to time of critical message (timedelta64[ns]). - msg: error message associated with critical failure (string). - comp: name of failing component (string). After message is parsed, implement logic to replace failing component name if required and determine if message is redundant. Update appropriate DataFrames based on logic. :param lines: list of syslog lines containing critical failure messages. :return: update class state via update_critical method """ for line in lines: # exclude SBIT and other critical_all messages in ignore list if all(chk not in line for chk in self.ignore): log = self.work_logname timestamp = pd.to_datetime(line[:24]) # timestamp: parse line timestamp runtime = self.total_runtime + (timestamp - self.log_start_time) msg = re.search('\): (.*?)$', line).group(1) # message: grab text between brackets+colon and endl comp = re.search('\[(.*?)\]', line).group(1) # component: grab text between square brackets # replace component w/ drop-weight component if in message text drop_weight = ['drop weight', 'dropped weight'] if any(dw for dw in drop_weight if dw in msg.lower()): msg = ''.join(['[', comp, '] ', msg]) comp = 'DropWeight' # replace component CBIT w/ actual faulting component from message text if 'CBIT' in comp: # try to find component name in list of known CBIT failures cbit_rep = next((rep for rep in self.cbit_fail if rep.lower() in line.lower()), '') if cbit_rep: comp = cbit_rep else: try: # try to find component name in msg body comp = re.search(': (.*?)$', msg).group(1) except AttributeError: # if it doesn't work out, just keep component name CBIT pass # check redundancy and update appropriate DataFrames redundant = self.redundant_check(comp, timestamp) self.update_critical(log, comp, timestamp, runtime, msg, redundant) def scan_syslog(self, syslog_path, search_str=b'(CRITICAL)'): """Search syslog for critical failure messages. - search syslog lines for search_str and pass matches for parsing - extract syslog runtime data - keep track of processed logs via calls to class update methods :param syslog_path: path to working syslog :param search_str: key-word to search for :return: update class state """ # get working log name self.work_logname = syslog_path.split('/')[-2] # sync syslog to local host if working from remote (determined to be faster than reading over network) if 'tethysdata' not in self.host: copyfile(syslog_path, self.sys_local) syslog_path = self.sys_local with open(syslog_path, 'br') as s: search_match = [] for line in s: # get syslog start time if self.log_start_time is None: self.syslog_start_time(line) # search lines and extract critical failure messages if search_str in line: # TODO: allow search_str to be an array search_match.append(line.decode("utf-8", "ignore").strip()) # parse critical failure messages if any if search_match: self.parse_critical_messages(search_match) # get syslog end time self.syslog_end_time(s) # compute log run time self.syslog_runtime() # update list of processed logs self.update_logs()