Spaces:
Configuration error
Configuration error
File size: 4,161 Bytes
b143975 | 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 | """
Production-grade logging configuration for Analytics 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, Optional
SERVICE_NAME = "analytics-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):
"""Emit log records as single-line JSON objects."""
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:
payload["message"] = str(record.getMessage())
return json.dumps(payload, default=str)
class ConsoleFormatter(logging.Formatter):
GREY = "\x1b[38;5;240m"
CYAN = "\x1b[36m"
YELLOW = "\x1b[33m"
RED = "\x1b[31m"
BOLD_RED = "\x1b[1;31m"
RESET = "\x1b[0m"
LEVEL_COLORS = {
logging.DEBUG: "\x1b[38;5;240m",
logging.INFO: "\x1b[36m",
logging.WARNING: "\x1b[33m",
logging.ERROR: "\x1b[31m",
logging.CRITICAL: "\x1b[1;31m",
}
def format(self, record: logging.LogRecord) -> str:
color = self.LEVEL_COLORS.get(record.levelno, self.RESET)
ts = datetime.fromtimestamp(record.created, tz=timezone.utc).strftime("%H:%M:%S")
msg = record.getMessage()
base = f"{self.GREY}{ts}{self.RESET} {color}{record.levelname:<8}{self.RESET} {record.name} - {msg}"
if record.exc_info:
base += "\n" + self.formatException(record.exc_info)
return base
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)
is_json = os.getenv("LOG_FORMAT", "json").lower() == "json"
ch.setFormatter(JSONFormatter() if is_json else ConsoleFormatter())
root.addHandler(ch)
# File handlers
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 file logging fails
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)
|