logging, routing

packages/tracker/meshnet_tracker/logging_setup.py

@@ -0,0 +1,99 @@
+"""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)