File size: 7,000 Bytes
526cf2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """Logging entrypoint for the ad-creative env.
Standard-library pattern: library code only does `logging.getLogger(__name__)` and
emits — it attaches NO handlers except a NullHandler on the package logger (so a
"no handler" warning never fires and records still PROPAGATE to whatever the
application configures on the ROOT logger). An application entrypoint (our scripts,
or a future FE) calls configure_logging() ONCE to install a root handler; because our
loggers propagate, a FE that configures its own root logging captures our logs without
calling us, and there is no double-logging because the library owns no real handler.
AD_LOG_FORMAT=json gives hosting-ready structured output.
"""
from __future__ import annotations
import json
import logging
import os
import sys
from pathlib import Path
_PKG = "ad_creative_env"
_PLAIN_FMT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
_FORMATS = ("plain", "json")
# Dev default is DEBUG so a plain run surfaces every boundary log without extra env.
# Flip to "INFO" (or set AD_LOG_LEVEL) when quieting down for production.
_DEFAULT_LEVEL = "DEBUG"
# Chatty third-party loggers pinned to WARNING so a DEBUG run surfaces OUR records,
# not their per-chunk trace (e.g. PIL's PngImagePlugin STREAM lines, urllib3 retries).
_NOISY_LIBS = ("PIL", "urllib3")
# Library null handler: silences "no handler" and lets records propagate to whatever
# handler the application installs on root. Added once, at import.
logging.getLogger(_PKG).addHandler(logging.NullHandler())
# Standard LogRecord attributes + our JSON schema keys — everything else on a record
# is a structured `extra=` field we surface in JSON. (makeLogRecord baseline is
# version-robust: 3.10/3.11 lack `taskName`, 3.12 adds it; the explicit union covers both.)
_RESERVED = set(vars(logging.makeLogRecord({}))) | {"message", "asctime", "taskName"}
_SCHEMA_KEYS = ("ts", "level", "logger", "msg", "exc", "stack")
def _safe_str(value) -> str:
"""str(value) that never raises (a hostile __str__/__repr__ can't crash logging)."""
try:
return str(value)
except Exception:
return "<unstringable>"
def _json_safe(value):
"""Return value if JSON-serializable, else a best-effort safe string, never raising."""
try:
json.dumps(value)
return value
except (TypeError, ValueError):
return _safe_str(value)
class JsonFormatter(logging.Formatter):
"""One-line JSON per record: {ts, level, logger, msg, +extra}. Canonical keys win.
Every value path is guarded so a hostile __str__/__repr__, malformed %-args, or a
non-serializable exception argument degrades to a safe string instead of raising —
a logging call must never crash the caller.
"""
def format(self, record: logging.LogRecord) -> str:
try:
msg = record.getMessage()
except Exception: # malformed %-args
msg = "<unformattable msg " + _safe_str(record.msg) + ">"
# extras FIRST so the canonical schema keys assigned below always win.
payload = {
key: _json_safe(val)
for key, val in record.__dict__.items()
if key not in _RESERVED and key not in _SCHEMA_KEYS and not key.startswith("_")
}
payload["ts"] = _safe_str(self.formatTime(record))
payload["level"] = record.levelname
payload["logger"] = record.name
payload["msg"] = msg
if record.exc_info:
try:
payload["exc"] = self.formatException(record.exc_info)
except Exception:
payload["exc"] = "<unformattable exception>"
if record.stack_info:
try:
payload["stack"] = self.formatStack(record.stack_info)
except Exception:
payload["stack"] = "<unformattable stack>"
try:
return json.dumps(payload, default=_safe_str)
except Exception: # last-resort: never let logging raise
return json.dumps({"level": record.levelname, "logger": record.name,
"msg": "<unserializable log record>"})
def _valid_level(name: str) -> bool:
# getLevelName(name) -> int for a known name, else a "Level X" string (3.10-safe).
return isinstance(logging.getLevelName(name), int)
def configure_logging(level=None, fmt=None, *, log_file=None, force=False) -> logging.Logger:
"""Install managed handlers on the ROOT logger, basicConfig-style. Idempotent.
level: arg > AD_LOG_LEVEL env > "DEBUG" (dev default). fmt: arg > AD_LOG_FORMAT env
> "plain". log_file: arg > AD_LOG_FILE env > None. Raises ValueError on an unknown
level or format (fail-fast for both env vars).
Always installs a stderr StreamHandler (terminal output). When a log_file is
resolved, ALSO installs a FileHandler opened in mode "w", so each configure call
OVERWRITES the file — the file holds only the latest run. Both handlers are tagged
managed, so a reconfigure drops+closes them and re-installs cleanly.
Like logging.basicConfig, this is a NO-OP when the root logger already has FOREIGN
(non-managed) handlers — i.e. a host app configured logging — unless force=True. So
calling it from our scripts installs output, but if a host already owns root we do
not duplicate its output or override its level (our loggers just propagate into it).
"""
level = (level or os.environ.get("AD_LOG_LEVEL") or _DEFAULT_LEVEL).upper()
fmt = (fmt or os.environ.get("AD_LOG_FORMAT") or "plain").lower()
log_file = log_file or os.environ.get("AD_LOG_FILE")
if fmt not in _FORMATS:
raise ValueError(f"AD_LOG_FORMAT must be one of {_FORMATS}, got {fmt!r}")
if not _valid_level(level):
raise ValueError(f"AD_LOG_LEVEL must be a valid level name, got {level!r}")
root = logging.getLogger()
foreign = [h for h in root.handlers if not getattr(h, "_ad_managed", False)]
if foreign and not force:
return root # a host owns root logging; our loggers propagate into it
root.setLevel(level)
for handler in list(root.handlers): # idempotent: drop AND close our prior handlers
if getattr(handler, "_ad_managed", False):
root.removeHandler(handler)
handler.close()
formatter = JsonFormatter() if fmt == "json" else logging.Formatter(_PLAIN_FMT)
handlers = [logging.StreamHandler(sys.stderr)] # terminal, always
if log_file:
path = Path(log_file)
path.parent.mkdir(parents=True, exist_ok=True)
handlers.append(logging.FileHandler(path, mode="w", encoding="utf-8")) # overwrite
for handler in handlers:
setattr(handler, "_ad_managed", True) # sentinel: marks OUR handler vs a host's
handler.setFormatter(formatter)
root.addHandler(handler)
for name in _NOISY_LIBS: # only reached when WE own root (host-owned returns early)
logging.getLogger(name).setLevel(logging.WARNING)
return root
|