|
|
import logging
|
|
|
import logging.handlers
|
|
|
import json
|
|
|
import datetime
|
|
|
import os
|
|
|
import sys
|
|
|
|
|
|
try:
|
|
|
from zoneinfo import ZoneInfo
|
|
|
except ImportError:
|
|
|
from backports.zoneinfo import ZoneInfo # type: ignore[no-redef]
|
|
|
|
|
|
from src.config import LOG_LEVEL, SERVICE_NAME
|
|
|
from src.enums import OptimumOHEnum
|
|
|
|
|
|
|
|
|
_TZ_LOCAL = ZoneInfo("Asia/Jakarta")
|
|
|
|
|
|
# ANSI Color Codes
|
|
|
RESET = "\033[0m"
|
|
|
COLORS = {
|
|
|
"DEBUG": "\033[36m", # Cyan
|
|
|
"INFO": "\033[32m", # Green
|
|
|
"WARNING": "\033[33m", # Yellow
|
|
|
"WARN": "\033[33m", # Yellow
|
|
|
"ERROR": "\033[31m", # Red
|
|
|
"CRITICAL": "\033[1;31m", # Bold Red
|
|
|
}
|
|
|
|
|
|
# Fields in the log record injected by the logging middleware or extra= kwargs
|
|
|
_REQUEST_FIELDS = frozenset({
|
|
|
"method", "path", "status_code", "duration_ms",
|
|
|
"request_id", "user_id", "user_ip",
|
|
|
"error_id", "result", "error_type", "error_message",
|
|
|
})
|
|
|
|
|
|
|
|
|
class LogLevels(OptimumOHEnum):
|
|
|
info = "INFO"
|
|
|
warn = "WARN"
|
|
|
error = "ERROR"
|
|
|
debug = "DEBUG"
|
|
|
|
|
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
|
"""
|
|
|
Outputs exactly 14 fixed fields per log line in JSON format.
|
|
|
|
|
|
Fields (always present, null when not applicable):
|
|
|
timestamp, service_name, level, method, path, status_code,
|
|
|
duration_ms, request_id, user_id, user_ip, error_id,
|
|
|
result, error_type, error_message
|
|
|
"""
|
|
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
|
dt = datetime.datetime.fromtimestamp(record.created, tz=_TZ_LOCAL)
|
|
|
timestamp = dt.isoformat(timespec="microseconds")
|
|
|
|
|
|
status_code = getattr(record, "status_code", None)
|
|
|
result = getattr(record, "result", None)
|
|
|
if result is None and status_code is not None:
|
|
|
result = "Request completed" if int(status_code) < 400 else "Request failed"
|
|
|
|
|
|
log_record = {
|
|
|
"timestamp": timestamp,
|
|
|
"service_name": SERVICE_NAME,
|
|
|
"level": record.levelname,
|
|
|
"method": getattr(record, "method", None),
|
|
|
"path": getattr(record, "path", None),
|
|
|
"status_code": status_code,
|
|
|
"duration_ms": getattr(record, "duration_ms", None),
|
|
|
"request_id": getattr(record, "request_id", None),
|
|
|
"user_id": getattr(record, "user_id", None),
|
|
|
"user_ip": getattr(record, "user_ip", None),
|
|
|
"error_id": getattr(record, "error_id", None),
|
|
|
"result": result,
|
|
|
"error_type": getattr(record, "error_type", None),
|
|
|
"error_message": getattr(record, "error_message", None),
|
|
|
}
|
|
|
|
|
|
log_json = json.dumps(log_record, default=str)
|
|
|
|
|
|
if sys.stdout.isatty():
|
|
|
color = COLORS.get(record.levelname, "")
|
|
|
return f"{color}{log_json}{RESET}"
|
|
|
|
|
|
return log_json
|
|
|
|
|
|
|
|
|
def configure_logging() -> None:
|
|
|
log_level = str(LOG_LEVEL).upper()
|
|
|
if log_level not in list(LogLevels):
|
|
|
log_level = LogLevels.error
|
|
|
|
|
|
root_logger = logging.getLogger()
|
|
|
root_logger.setLevel(log_level)
|
|
|
|
|
|
if root_logger.hasHandlers():
|
|
|
root_logger.handlers.clear()
|
|
|
|
|
|
formatter = JSONFormatter()
|
|
|
|
|
|
# stdout handler
|
|
|
stream_handler = logging.StreamHandler(sys.stdout)
|
|
|
stream_handler.setFormatter(formatter)
|
|
|
root_logger.addHandler(stream_handler)
|
|
|
|
|
|
# rotating file handler — logs/app.log 10 MB × 5 backups
|
|
|
os.makedirs("logs", exist_ok=True)
|
|
|
file_handler = logging.handlers.RotatingFileHandler(
|
|
|
"logs/app.log",
|
|
|
maxBytes=10 * 1024 * 1024,
|
|
|
backupCount=5,
|
|
|
encoding="utf-8",
|
|
|
)
|
|
|
file_handler.setFormatter(formatter)
|
|
|
root_logger.addHandler(file_handler)
|
|
|
|
|
|
# Reconfigure uvicorn loggers to propagate through our formatter
|
|
|
for logger_name in ["uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"]:
|
|
|
uvicorn_logger = logging.getLogger(logger_name)
|
|
|
uvicorn_logger.handlers = []
|
|
|
uvicorn_logger.propagate = True
|
|
|
|
|
|
# Silence chatty loggers — we emit one structured log per request ourselves
|
|
|
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
|
|
|
|
|
|
slowapi_logger = logging.getLogger("slowapi")
|
|
|
slowapi_logger.setLevel(logging.ERROR)
|
|
|
slowapi_logger.propagate = False
|
|
|
|
|
|
logging.getLogger("slack_sdk.web.base_client").setLevel(logging.CRITICAL) |