flagsmith/api/util/logging.py at main · Flagsmith/flagsmith · GitHub

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
import json
import logging
import sys
from datetime import datetime
from typing import Any

from django.conf import settings
from gunicorn.config import Config  # type: ignore[import-untyped]
from gunicorn.instrument.statsd import (  # type: ignore[import-untyped]
    Statsd as GunicornLogger,
)


class JsonFormatter(logging.Formatter):
    """Custom formatter for json logs."""

    def get_json_record(self, record: logging.LogRecord) -> dict[str, Any]:
        formatted_message = record.getMessage()
        json_record = {
            "levelname": record.levelname,
            "message": formatted_message,
            "timestamp": self.formatTime(record, self.datefmt),
            "logger_name": record.name,
            "process_id": record.process,
            "thread_name": record.threadName,
        }
        if record.exc_info:
            json_record["exc_info"] = self.formatException(record.exc_info)
        return json_record

    def format(self, record: logging.LogRecord) -> str:
        return json.dumps(self.get_json_record(record))


class GunicornAccessLogJsonFormatter(JsonFormatter):
    def get_json_record(self, record: logging.LogRecord) -> dict[str, Any]:
        args = record.args
        url = args["U"]  # type: ignore[call-overload,index]
        if q := args["q"]:  # type: ignore[call-overload,index]
            url += f"?{q}"  # type: ignore[operator]

        return {
            **super().get_json_record(record),
            "time": datetime.strptime(args["t"], "[%d/%b/%Y:%H:%M:%S %z]").isoformat(),  # type: ignore[arg-type,call-overload,index]  # noqa: E501  # noqa: E501
            "path": url,
            "remote_ip": args["h"],  # type: ignore[call-overload,index]
            "method": args["m"],  # type: ignore[call-overload,index]
            "status": str(args["s"]),  # type: ignore[call-overload,index]
            "user_agent": args["a"],  # type: ignore[call-overload,index]
            "referer": args["f"],  # type: ignore[call-overload,index]
            "duration_in_ms": args["M"],  # type: ignore[call-overload,index]
            "pid": args["p"],  # type: ignore[call-overload,index]
        }


class GunicornJsonCapableLogger(GunicornLogger):  # type: ignore[misc]
    def setup(self, cfg: Config) -> None:
        super().setup(cfg)
        if getattr(settings, "LOG_FORMAT", None) == "json":
            self._set_handler(
                self.error_log,
                cfg.errorlog,
                JsonFormatter(),
            )
            self._set_handler(
                self.access_log,
                cfg.accesslog,
                GunicornAccessLogJsonFormatter(),
                stream=sys.stdout,
            )