100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""Rotating file logging for the tracker CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import sys
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
|
|
|
|
DEFAULT_LOG_DIR = "logs/tracker"
|
|
DEFAULT_LOG_MAX_BYTES = 10 * 1024 * 1024
|
|
DEFAULT_LOG_BACKUP_COUNT = 5
|
|
TRACKER_LOGGER_NAME = "meshnet.tracker"
|
|
|
|
|
|
class _ExactLevelFilter(logging.Filter):
|
|
def __init__(self, level: int) -> None:
|
|
super().__init__()
|
|
self._level = level
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
return record.levelno == self._level
|
|
|
|
|
|
class _TeeStream:
|
|
def __init__(self, stream: TextIO, logger: logging.Logger, level: int) -> None:
|
|
self._stream = stream
|
|
self._logger = logger
|
|
self._level = level
|
|
self._buffer = ""
|
|
|
|
def write(self, text: str) -> int:
|
|
self._stream.write(text)
|
|
self._stream.flush()
|
|
self._buffer += text
|
|
while "\n" in self._buffer:
|
|
line, self._buffer = self._buffer.split("\n", 1)
|
|
line = line.rstrip()
|
|
if line:
|
|
self._logger.log(self._level, line)
|
|
return len(text)
|
|
|
|
def flush(self) -> None:
|
|
self._stream.flush()
|
|
line = self._buffer.rstrip()
|
|
if line:
|
|
self._logger.log(self._level, line)
|
|
self._buffer = ""
|
|
|
|
def isatty(self) -> bool:
|
|
return self._stream.isatty()
|
|
|
|
|
|
def _make_handler(path: Path, level: int, *, max_bytes: int, backup_count: int) -> RotatingFileHandler:
|
|
handler = RotatingFileHandler(
|
|
path,
|
|
maxBytes=max_bytes,
|
|
backupCount=backup_count,
|
|
encoding="utf-8",
|
|
)
|
|
handler.setLevel(level)
|
|
handler.addFilter(_ExactLevelFilter(level))
|
|
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
return handler
|
|
|
|
|
|
def configure_tracker_file_logging(
|
|
log_dir: str | Path = DEFAULT_LOG_DIR,
|
|
*,
|
|
max_bytes: int = DEFAULT_LOG_MAX_BYTES,
|
|
backup_count: int = DEFAULT_LOG_BACKUP_COUNT,
|
|
tee_stdio: bool = True,
|
|
) -> Path:
|
|
"""Configure rotatable info/warning/error log files and return the directory."""
|
|
|
|
path = Path(log_dir).expanduser()
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
logger = logging.getLogger(TRACKER_LOGGER_NAME)
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
logger.handlers.clear()
|
|
logger.addHandler(_make_handler(path / "info.log", logging.INFO, max_bytes=max_bytes, backup_count=backup_count))
|
|
logger.addHandler(_make_handler(path / "warning.log", logging.WARNING, max_bytes=max_bytes, backup_count=backup_count))
|
|
logger.addHandler(_make_handler(path / "error.log", logging.ERROR, max_bytes=max_bytes, backup_count=backup_count))
|
|
|
|
if tee_stdio:
|
|
if not isinstance(sys.stdout, _TeeStream):
|
|
sys.stdout = _TeeStream(sys.stdout, logger, logging.INFO) # type: ignore[assignment]
|
|
if not isinstance(sys.stderr, _TeeStream):
|
|
sys.stderr = _TeeStream(sys.stderr, logger, logging.ERROR) # type: ignore[assignment]
|
|
|
|
return path
|
|
|
|
|
|
def tracker_logger() -> logging.Logger:
|
|
return logging.getLogger(TRACKER_LOGGER_NAME)
|