diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ac30a7b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,13 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "log_setup" +version = "0.1.0" +description = "stdlib app-entry-point logging setup: live run.log, rotation, gzip, retention, consistent format" +requires-python = ">=3.10" +dependencies = [] + +[tool.hatch.build.targets.wheel] +packages = ["src/log_setup"] diff --git a/src/log_setup/__init__.py b/src/log_setup/__init__.py new file mode 100644 index 0000000..ed1dbc7 --- /dev/null +++ b/src/log_setup/__init__.py @@ -0,0 +1,22 @@ +"""log_setup — app-entry-point logging configuration (sync, stdlib only). + +call once at an application's entry point to configure the whole process: a live +run.log, rotation (daily/size/on_start), gzip of rolled files, retention, console +output, and a consistent `time | module | level | message` format. + + from log_setup import setup_logging + + setup_logging(name="run", level="INFO") # daily rotation, logs/ dir, gzip + log = logging.getLogger(__name__) + log.info("up") # -> run.log + console + +reusable libraries do NOT call this — they only `logging.getLogger(__name__)` and +emit; the application owns this setup. shipping logs to a backend is out of scope +(that's Promtail's job against the produced files). +""" + +from .setup import setup_logging + +__all__ = ["setup_logging"] + +__version__ = "0.1.0" diff --git a/src/log_setup/formats.py b/src/log_setup/formats.py new file mode 100644 index 0000000..2eba61d --- /dev/null +++ b/src/log_setup/formats.py @@ -0,0 +1,15 @@ +"""default log format + datefmt for the app-wide setup. + +one format for v0.1.0, used on both console and file. `%(name)s` is the getLogger +name the emitting module used, so each library/module shows in the line. +""" + +import logging + +DEFAULT_FORMAT = "%(asctime)s | %(name)s | %(levelname)s | %(message)s" +DEFAULT_DATEFMT = "%Y-%m-%d %H:%M:%S" + + +def build_formatter(fmt=None, datefmt=None) -> logging.Formatter: + """build a logging.Formatter from overrides, falling back to the defaults""" + return logging.Formatter(fmt or DEFAULT_FORMAT, datefmt or DEFAULT_DATEFMT) diff --git a/src/log_setup/rotation.py b/src/log_setup/rotation.py new file mode 100644 index 0000000..dfb0d6d --- /dev/null +++ b/src/log_setup/rotation.py @@ -0,0 +1,102 @@ +"""custom namer/rotator + on-start rotation + retention pruning (stdlib only). + +the stdlib rotating handlers roll a file next to the live file; these helpers +override the namer/rotator so rolled files land in `log_dir` and are gzipped when +asked, keep the live file at its stable path, and handle the on-start and prune +paths the handlers don't manage themselves. +""" + +import gzip +import os +import shutil +import time +from typing import Callable, Tuple + + +def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]: + """namer: redirect a rolled filename into log_dir, adding .gz when compressing + + the handler hands us the default rolled path (next to the live file); we keep its + basename but place it under log_dir, and append .gz so the gzipped name matches. + """ + def namer(default_name: str) -> str: + base = os.path.basename(default_name) + target = os.path.join(log_dir, base) + return target + ".gz" if compress else target + return namer + + +def make_rotator(compress: bool) -> Callable[[str, str], None]: + """rotator: move (or gzip) the source live file to the destination rolled path""" + def rotator(source: str, dest: str) -> None: + if not os.path.exists(source): + return + if compress: + with open(source, "rb") as src, gzip.open(dest, "wb") as dst: + shutil.copyfileobj(src, dst) + os.remove(source) + else: + os.replace(source, dest) + return rotator + + +def rotate_on_start(live_path: str, log_dir: str, compress: bool, clock=time.localtime) -> None: + """move an existing live file into log_dir with a timestamp, gzipped if asked + + no-op if the live file doesn't exist. used by rotate="on_start" before the fresh + handler opens a new live file. the timestamp form is run.<%Y-%m-%d_%H-%M-%S>.log. + """ + if not os.path.exists(live_path): + return + stem = os.path.splitext(os.path.basename(live_path))[0] + stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock()) + dest = os.path.join(log_dir, f"{stem}.{stamp}.log") + if compress: + dest += ".gz" + with open(live_path, "rb") as src, gzip.open(dest, "wb") as dst: + shutil.copyfileobj(src, dst) + os.remove(live_path) + else: + os.replace(live_path, dest) + + +def prune(log_dir: str, stem: str, backup_count: int) -> None: + """keep only the newest `backup_count` rolled files for a given stem in log_dir + + matches files beginning with `.` (e.g. run.*), sorted by mtime, deleting the + oldest beyond the count. used for on_start, which the handlers don't auto-prune. + """ + if backup_count <= 0: + return + try: + entries = [ + os.path.join(log_dir, name) + for name in os.listdir(log_dir) + if name.startswith(f"{stem}.") and name != f"{stem}.log" + ] + except OSError: + return + files = [(p, _safe_mtime(p)) for p in entries if os.path.isfile(p)] + files.sort(key=lambda pair: pair[1], reverse=True) + for path, _ in files[backup_count:]: + try: + os.remove(path) + except OSError: + pass + + +def _safe_mtime(path: str) -> float: + """mtime of a path, or 0 if it can't be stat'd""" + try: + return os.path.getmtime(path) + except OSError: + return 0.0 + + +def attach_rolling(handler, log_dir: str, compress: bool) -> Tuple[Callable, Callable]: + """wire the custom namer + rotator onto a rotating handler; return them""" + namer = make_namer(log_dir, compress) + rotator = make_rotator(compress) + handler.namer = namer + handler.rotator = rotator + return namer, rotator diff --git a/src/log_setup/setup.py b/src/log_setup/setup.py new file mode 100644 index 0000000..f9f922b --- /dev/null +++ b/src/log_setup/setup.py @@ -0,0 +1,158 @@ +"""app-entry-point logging setup (sync, stdlib only). + +`setup_logging` configures the root logger once for the whole process: a live +run.log at a stable path, rotation (daily/size/on_start/none) into a logs/ dir, gzip +of rolled files, retention, console output, and a consistent format. it is called by +the APPLICATION, not by reusable libraries (those stay emit-only). it is idempotent +(no duplicate handlers on repeat calls), never crashes the app over logging, and can +route through a background queue so an async event loop doesn't block on file I/O. +""" + +import atexit +import logging +import logging.handlers +import os +import queue +from typing import Optional, Union + +from .formats import build_formatter +from .rotation import attach_rolling, prune, rotate_on_start + +log = logging.getLogger(__name__) + +_MARKER = "_log_setup_owned" +_listener = None + + +def _level_value(level: Union[int, str]) -> int: + """coerce a level name or int to a logging level int (defaults to INFO)""" + if isinstance(level, int): + return level + return logging.getLevelName(str(level).upper()) if isinstance(level, str) else logging.INFO + + +def _clear_owned(root: logging.Logger) -> None: + """remove only the handlers this lib previously added; leave app handlers alone""" + global _listener + if _listener is not None: + _listener.stop() + _listener = None + for handler in list(root.handlers): + if getattr(handler, _MARKER, False): + root.removeHandler(handler) + try: + handler.close() + except Exception: + pass + + +def _tag(handler: logging.Handler) -> logging.Handler: + """mark a handler as owned by this lib so idempotent re-setup can clear it""" + setattr(handler, _MARKER, True) + return handler + + +def _file_handler( + name: str, live_path: str, log_dir: str, rotate: Optional[str], + backup_count: int, max_bytes: int, compress: bool, +) -> logging.Handler: + """build the configured file handler with custom rolling into log_dir""" + if rotate == "size": + handler = logging.handlers.RotatingFileHandler( + live_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8", + ) + attach_rolling(handler, log_dir, compress) + elif rotate == "daily": + handler = logging.handlers.TimedRotatingFileHandler( + live_path, when="midnight", backupCount=backup_count, encoding="utf-8", + ) + attach_rolling(handler, log_dir, compress) + else: + if rotate == "on_start": + rotate_on_start(live_path, log_dir, compress) + prune(log_dir, name, backup_count) + handler = logging.FileHandler(live_path, encoding="utf-8") + return handler + + +def setup_logging( + name: str = "run", + log_dir: str = "logs", + level: Union[int, str] = "INFO", + rotate: Optional[str] = "daily", + backup_count: int = 14, + max_bytes: int = 10_000_000, + compress: bool = True, + console: bool = False, + queue: bool = False, + fmt: Optional[str] = None, + datefmt: Optional[str] = None, +) -> logging.Logger: + """configure the root logger for the whole process and return it + + `name` -> .log live file at cwd; rolled/compressed copies go to `log_dir`. + `rotate` is "daily" (default), "size", "on_start", or None. `console=True` adds a + stdout handler (off by default — the file is the output). `queue=True` routes records + through a background QueueListener so file I/O never blocks the caller (the listener + is stopped at exit). idempotent: a repeat call clears only the handlers this function + added. never raises over logging — an unwritable `log_dir` falls back to console-only + with a warning even when `console` is off, so output is never silently lost. + """ + global _listener + + root = logging.getLogger() + root.setLevel(_level_value(level)) + _clear_owned(root) + + formatter = build_formatter(fmt, datefmt) + live_path = f"{name}.log" + + handlers = [] + + file_ok = True + try: + os.makedirs(log_dir, exist_ok=True) + except OSError: + file_ok = False + + if file_ok: + try: + fh = _file_handler(name, live_path, log_dir, rotate, backup_count, max_bytes, compress) + fh.setFormatter(formatter) + handlers.append(fh) + except OSError: + file_ok = False + + if console or not file_ok: + sh = logging.StreamHandler() + sh.setFormatter(formatter) + handlers.append(sh) + + if queue: + record_queue: "queue.Queue" = _make_queue() + qh = _tag(logging.handlers.QueueHandler(record_queue)) + root.addHandler(qh) + _listener = logging.handlers.QueueListener(record_queue, *handlers, respect_handler_level=True) + _listener.start() + atexit.register(_stop_listener) + else: + for handler in handlers: + root.addHandler(_tag(handler)) + + if not file_ok: + log.warning("log_setup: log_dir %r not writable; logging to console only", log_dir) + + return root + + +def _make_queue() -> "queue.Queue": + """unbounded in-memory queue for the QueueHandler -> QueueListener path""" + return queue.Queue(-1) + + +def _stop_listener() -> None: + """stop the background queue listener, flushing remaining records""" + global _listener + if _listener is not None: + _listener.stop() + _listener = None