79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Logging helpers: per-run log files and consistent formatting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import logging
|
|
import threading
|
|
from typing import Iterator, Optional
|
|
|
|
LOG_FORMAT = "%(asctime)s %(levelname)-7s [%(name)s] %(message)s"
|
|
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
|
|
|
|
# Loggers that describe a capture analysis end to end.
|
|
ANALYSIS_LOGGERS = ("analyzer", "IoTAnalyzer.analyze")
|
|
|
|
|
|
class _ThreadFilter(logging.Filter):
|
|
"""Only pass records emitted by the thread that opened the run log."""
|
|
|
|
def __init__(self, thread_id: int):
|
|
super().__init__()
|
|
self.thread_id = thread_id
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
return record.thread == self.thread_id
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def run_log(path: Optional[str], level: int = logging.DEBUG) -> Iterator[Optional[logging.Handler]]:
|
|
"""Write everything this thread logs about the analysis to ``path``.
|
|
|
|
Concurrent analyses in other threads do not leak into the file.
|
|
"""
|
|
if not path:
|
|
yield None
|
|
return
|
|
handler: Optional[logging.FileHandler] = None
|
|
try:
|
|
handler = logging.FileHandler(path, mode="w", encoding="utf-8")
|
|
except OSError as exc:
|
|
logging.getLogger("analyzer").warning("Run log %s could not be opened: %s", path, exc)
|
|
yield None
|
|
return
|
|
handler.setLevel(level)
|
|
handler.setFormatter(logging.Formatter(LOG_FORMAT, DATE_FORMAT))
|
|
handler.addFilter(_ThreadFilter(threading.get_ident()))
|
|
loggers = [logging.getLogger(n) for n in ANALYSIS_LOGGERS]
|
|
previous = [(lg, lg.level) for lg in loggers]
|
|
for lg in loggers:
|
|
lg.addHandler(handler)
|
|
if lg.level == logging.NOTSET or lg.level > level:
|
|
lg.setLevel(level)
|
|
try:
|
|
yield handler
|
|
finally:
|
|
for lg, lvl in previous:
|
|
lg.removeHandler(handler)
|
|
lg.setLevel(lvl)
|
|
handler.close()
|
|
|
|
|
|
def configure_console(level: str = "INFO") -> None:
|
|
"""Console logging for CLI use."""
|
|
logging.basicConfig(level=getattr(logging, level.upper(), logging.INFO),
|
|
format=LOG_FORMAT, datefmt=DATE_FORMAT)
|
|
|
|
|
|
def tail(path: str, lines: int = 400, max_bytes: int = 256 * 1024) -> str:
|
|
"""Return the last ``lines`` lines of a text file."""
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
fh.seek(0, 2)
|
|
size = fh.tell()
|
|
fh.seek(max(0, size - max_bytes))
|
|
data = fh.read().decode("utf-8", "replace")
|
|
except OSError:
|
|
return ""
|
|
return "\n".join(data.splitlines()[-lines:])
|