Spaces:
Configuration error
Configuration error
File size: 3,878 Bytes
3818b61 | 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 | """
Structured JSON logging for Workforce Microservice.
"""
import json
import logging
import logging.handlers
import os
import sys
import traceback
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict
SERVICE_NAME = "workforce-ms"
LOG_DIR = Path(os.getenv("LOG_DIR", "logs"))
LOG_MAX_BYTES = int(os.getenv("LOG_MAX_BYTES", 50 * 1024 * 1024))
LOG_BACKUP_COUNT = int(os.getenv("LOG_BACKUP_COUNT", "10"))
class JSONFormatter(logging.Formatter):
RESERVED = frozenset({
"args", "created", "exc_info", "exc_text", "filename",
"funcName", "levelname", "levelno", "lineno", "message",
"module", "msecs", "msg", "name", "pathname", "process",
"processName", "relativeCreated", "stack_info", "thread", "threadName",
})
def format(self, record: logging.LogRecord) -> str:
payload: Dict[str, Any] = {
"timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"service": SERVICE_NAME,
"pid": record.process,
}
if record.levelno >= logging.WARNING:
payload["caller"] = f"{record.pathname}:{record.lineno}"
for key, val in record.__dict__.items():
if key not in self.RESERVED and not key.startswith("_"):
payload[key] = val
if record.exc_info and record.exc_info[0] is not None:
exc_type, exc_value, exc_tb = record.exc_info
payload["exception"] = {
"type": exc_type.__name__,
"message": str(exc_value),
"stacktrace": traceback.format_exception(exc_type, exc_value, exc_tb),
}
try:
return json.dumps(payload, default=str)
except Exception:
return json.dumps({"message": str(record.getMessage()), "service": SERVICE_NAME}, default=str)
class ConsoleFormatter(logging.Formatter):
CYAN = "\x1b[36m"
YELLOW = "\x1b[33m"
RED = "\x1b[31m"
GREY = "\x1b[38;5;240m"
BOLD_RED = "\x1b[1;31m"
RESET = "\x1b[0m"
LEVEL_COLOURS = {
logging.DEBUG: "\x1b[38;5;240m",
logging.INFO: "\x1b[36m",
logging.WARNING: "\x1b[33m",
logging.ERROR: "\x1b[31m",
logging.CRITICAL: "\x1b[1;31m",
}
FMT = "%(asctime)s %(levelname)-8s %(name)s — %(message)s"
def format(self, record: logging.LogRecord) -> str:
colour = self.LEVEL_COLOURS.get(record.levelno, self.RESET)
formatter = logging.Formatter(
f"{colour}{self.FMT}{self.RESET}", datefmt="%Y-%m-%d %H:%M:%S"
)
return formatter.format(record)
def setup_logging(level: str = "INFO") -> None:
numeric_level = getattr(logging, level.upper(), logging.INFO)
root = logging.getLogger()
root.setLevel(numeric_level)
root.handlers.clear()
# Console handler
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(numeric_level)
ch.setFormatter(ConsoleFormatter())
root.addHandler(ch)
# File handlers (JSON)
try:
LOG_DIR.mkdir(parents=True, exist_ok=True)
fh = logging.handlers.RotatingFileHandler(
LOG_DIR / "app.log", maxBytes=LOG_MAX_BYTES, backupCount=LOG_BACKUP_COUNT
)
fh.setLevel(numeric_level)
fh.setFormatter(JSONFormatter())
root.addHandler(fh)
eh = logging.handlers.RotatingFileHandler(
LOG_DIR / "app_errors.log", maxBytes=LOG_MAX_BYTES, backupCount=LOG_BACKUP_COUNT
)
eh.setLevel(logging.ERROR)
eh.setFormatter(JSONFormatter())
root.addHandler(eh)
except Exception:
pass # Non-critical if log dir can't be created
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)
|