Checking the httpd error log for errors and warnings, including limiting checks from a recent known position forward.
| 7 | |
| 8 | |
| 9 | class HttpdErrorLog: |
| 10 | """Checking the httpd error log for errors and warnings, including |
| 11 | limiting checks from a recent known position forward. |
| 12 | """ |
| 13 | |
| 14 | RE_ERRLOG_WARN = re.compile(r'.*\[[^:]+:warn].*') |
| 15 | RE_ERRLOG_ERROR = re.compile(r'.*\[[^:]+:error].*') |
| 16 | RE_APLOGNO = re.compile(r'.*\[[^:]+:(error|warn)].* (?P<aplogno>AH\d+): .+') |
| 17 | |
| 18 | def __init__(self, path: str): |
| 19 | self._path = path |
| 20 | self._ignored_matches = [] |
| 21 | self._ignored_lognos = set() |
| 22 | # remember the file position we started with |
| 23 | self._start_pos = 0 |
| 24 | if os.path.isfile(self._path): |
| 25 | with open(self._path) as fd: |
| 26 | self._start_pos = fd.seek(0, SEEK_END) |
| 27 | self._recent_pos = self._start_pos |
| 28 | self._recent_errors = [] |
| 29 | self._recent_warnings = [] |
| 30 | self._caught_errors = set() |
| 31 | self._caught_warnings = set() |
| 32 | self._caught_matches = set() |
| 33 | |
| 34 | def __repr__(self): |
| 35 | return f"HttpdErrorLog[{self._path}, errors: {' '.join(self._recent_errors)}, " \ |
| 36 | f"warnings: {' '.join(self._recent_warnings)}]" |
| 37 | |
| 38 | @property |
| 39 | def path(self) -> str: |
| 40 | return self._path |
| 41 | |
| 42 | def clear_log(self): |
| 43 | if os.path.isfile(self.path): |
| 44 | os.truncate(self.path, 0) |
| 45 | self._start_pos = self._recent_pos = 0 |
| 46 | self._recent_errors = [] |
| 47 | self._recent_warnings = [] |
| 48 | self._caught_errors = set() |
| 49 | self._caught_warnings = set() |
| 50 | self._caught_matches = set() |
| 51 | |
| 52 | def _lookup_matches(self, line: str, matches: List[str]) -> bool: |
| 53 | for m in matches: |
| 54 | if re.match(m, line): |
| 55 | return True |
| 56 | return False |
| 57 | |
| 58 | def _lookup_lognos(self, line: str, lognos: set) -> bool: |
| 59 | if len(lognos) > 0: |
| 60 | m = self.RE_APLOGNO.match(line) |
| 61 | if m and m.group('aplogno') in lognos: |
| 62 | return True |
| 63 | return False |
| 64 | |
| 65 | def clear_ignored_matches(self): |
| 66 | self._ignored_matches = [] |