diff --git a/src/endpoint_monitor/logging.py b/src/endpoint_monitor/logging.py new file mode 100644 index 0000000..819a8f1 --- /dev/null +++ b/src/endpoint_monitor/logging.py @@ -0,0 +1,36 @@ +import json +import logging +from datetime import UTC, datetime +from urllib.parse import urlsplit, urlunsplit + + +_RESERVED = {"name", "msg", "args", "levelname", "levelno", "pathname", "filename", "module", "exc_info", "exc_text", "stack_info", "lineno", "funcName", "created", "msecs", "relativeCreated", "thread", "threadName", "processName", "process", "taskName"} + + +def redact_url(value: str) -> str: + try: + parsed = urlsplit(value) + host = parsed.hostname or "" + if ":" in host and not host.startswith("["): + host = f"[{host}]" + port = f":{parsed.port}" if parsed.port else "" + return urlunsplit((parsed.scheme, host + port, parsed.path or "/", "", "")) + except (TypeError, ValueError): + return "" + + +class JsonFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + data = {"timestamp": datetime.now(UTC).isoformat(), "level": record.levelname, "event": record.getMessage()} + for key, value in record.__dict__.items(): + if key not in _RESERVED and not key.startswith("_"): + data[key] = value + return json.dumps(data, default=str, separators=(",", ":")) + + +def configure_logging(level: str) -> None: + handler = logging.StreamHandler() + handler.setFormatter(JsonFormatter()) + root = logging.getLogger() + root.handlers[:] = [handler] + root.setLevel(level)