| import os |
| import re |
| import time |
| from datetime import datetime, timedelta |
| from io import SEEK_END |
| from typing import List, Tuple, Any |
| |
| |
| class HttpdErrorLog: |
| """Checking the httpd error log for errors and warnings, including |
| limiting checks from a last known position forward. |
| """ |
| |
| RE_ERRLOG_ERROR = re.compile(r'.*\[(?P<module>[^:]+):error].*') |
| RE_ERRLOG_WARN = re.compile(r'.*\[(?P<module>[^:]+):warn].*') |
| RE_APLOGNO = re.compile(r'.*\[(?P<module>[^:]+):(error|warn)].* (?P<aplogno>AH\d+): .+') |
| RE_SSL_LIB_ERR = re.compile(r'.*\[ssl:error].* SSL Library Error: error:(?P<errno>\S+):.+') |
| |
| def __init__(self, path: str): |
| self._path = path |
| self._ignored_modules = [] |
| self._ignored_lognos = set() |
| self._ignored_patterns = [] |
| # remember the file position we started with |
| self._start_pos = 0 |
| if os.path.isfile(self._path): |
| with open(self._path) as fd: |
| self._start_pos = fd.seek(0, SEEK_END) |
| self._last_pos = self._start_pos |
| self._last_errors = [] |
| self._last_warnings = [] |
| self._observed_erros = set() |
| self._observed_warnings = set() |
| |
| def __repr__(self): |
| return f"HttpdErrorLog[{self._path}, errors: {' '.join(self._last_errors)}, " \ |
| f"warnings: {' '.join(self._last_warnings)}]" |
| |
| @property |
| def path(self) -> str: |
| return self._path |
| |
| def clear_log(self): |
| if os.path.isfile(self.path): |
| os.remove(self.path) |
| self._start_pos = 0 |
| self._last_pos = self._start_pos |
| self._last_errors = [] |
| self._last_warnings = [] |
| self._observed_erros = set() |
| self._observed_warnings = set() |
| |
| def set_ignored_modules(self, modules: List[str]): |
| self._ignored_modules = modules.copy() if modules else [] |
| |
| def set_ignored_lognos(self, lognos: List[str]): |
| if lognos: |
| for l in lognos: |
| self._ignored_lognos.add(l) |
| |
| def add_ignored_patterns(self, patterns: List[Any]): |
| self._ignored_patterns.extend(patterns) |
| |
| def _is_ignored(self, line: str) -> bool: |
| for p in self._ignored_patterns: |
| if p.match(line): |
| return True |
| m = self.RE_APLOGNO.match(line) |
| if m and m.group('aplogno') in self._ignored_lognos: |
| return True |
| return False |
| |
| def get_recent(self, advance=True) -> Tuple[List[str], List[str]]: |
| """Collect error and warning from the log since the last remembered position |
| :param advance: advance the position to the end of the log afterwards |
| :return: list of error and list of warnings as tuple |
| """ |
| self._last_errors = [] |
| self._last_warnings = [] |
| if os.path.isfile(self._path): |
| with open(self._path) as fd: |
| fd.seek(self._last_pos, os.SEEK_SET) |
| for line in fd: |
| if self._is_ignored(line): |
| continue |
| m = self.RE_ERRLOG_ERROR.match(line) |
| if m and m.group('module') not in self._ignored_modules: |
| self._last_errors.append(line) |
| continue |
| m = self.RE_ERRLOG_WARN.match(line) |
| if m: |
| if m and m.group('module') not in self._ignored_modules: |
| self._last_warnings.append(line) |
| continue |
| if advance: |
| self._last_pos = fd.tell() |
| self._observed_erros.update(set(self._last_errors)) |
| self._observed_warnings.update(set(self._last_warnings)) |
| return self._last_errors, self._last_warnings |
| |
| def get_recent_count(self, advance=True): |
| errors, warnings = self.get_recent(advance=advance) |
| return len(errors), len(warnings) |
| |
| def ignore_recent(self): |
| """After a test case triggered errors/warnings on purpose, add |
| those to our 'observed' list so the do not get reported as 'missed'. |
| """ |
| self._last_errors = [] |
| self._last_warnings = [] |
| if os.path.isfile(self._path): |
| with open(self._path) as fd: |
| fd.seek(self._last_pos, os.SEEK_SET) |
| for line in fd: |
| if self._is_ignored(line): |
| continue |
| m = self.RE_ERRLOG_ERROR.match(line) |
| if m and m.group('module') not in self._ignored_modules: |
| self._observed_erros.add(line) |
| continue |
| m = self.RE_ERRLOG_WARN.match(line) |
| if m: |
| if m and m.group('module') not in self._ignored_modules: |
| self._observed_warnings.add(line) |
| continue |
| self._last_pos = fd.tell() |
| |
| def get_missed(self) -> Tuple[List[str], List[str]]: |
| errors = [] |
| warnings = [] |
| if os.path.isfile(self._path): |
| with open(self._path) as fd: |
| fd.seek(self._start_pos, os.SEEK_SET) |
| for line in fd: |
| if self._is_ignored(line): |
| continue |
| m = self.RE_ERRLOG_ERROR.match(line) |
| if m and m.group('module') not in self._ignored_modules \ |
| and line not in self._observed_erros: |
| errors.append(line) |
| continue |
| m = self.RE_ERRLOG_WARN.match(line) |
| if m: |
| if m and m.group('module') not in self._ignored_modules \ |
| and line not in self._observed_warnings: |
| warnings.append(line) |
| continue |
| return errors, warnings |
| |
| def scan_recent(self, pattern: re, timeout=10): |
| if not os.path.isfile(self.path): |
| return False |
| with open(self.path) as fd: |
| end = datetime.now() + timedelta(seconds=timeout) |
| while True: |
| fd.seek(self._last_pos, os.SEEK_SET) |
| for line in fd: |
| if pattern.match(line): |
| return True |
| if datetime.now() > end: |
| raise TimeoutError(f"pattern not found in error log after {timeout} seconds") |
| time.sleep(.1) |
| return False |
