将Python日志记录为字典或JSON格式的文件

41

我正在尝试设置日志记录,使得可以同时记录在stdout和文件中。 我已经成功地使用以下代码完成了此操作:

logging.basicConfig(
        level=logging.DEBUG, format='%(asctime)-15s %(levelname)-8s %(message)s',
        datefmt='%a, %d %b %Y %H:%M:%S', handlers=[logging.FileHandler(path), logging.StreamHandler()])
这将输出类似于以下内容:
2018-05-02 18:43:33,295 DEBUG    Starting new HTTPS connection (1): google.com
2018-05-02 18:43:33,385 DEBUG    https://google.com:443 "GET / HTTP/1.1" 301 220
2018-05-02 18:43:33,389 DEBUG    Starting new HTTPS connection (1): www.google.com
2018-05-02 18:43:33,490 DEBUG    https://www.google.com:443 "GET / HTTP/1.1" 200 None
我希望实现的是将输出记录到文件中,而不是打印到标准输出,但仍然保持其作为类似于字典或JSON对象的形式(同时保留当前stdout)。例如: [{'time': '2018-05-02 18:43:33,295','level': 'DEBUG','message': 'Starting new HTTPS connection (1): google.com'},{...},{...}] 。这可行吗?我知道可以在进程完成后对日志文件进行后处理,但我正在寻找更优雅的解决方案,因为我记录的某些内容本身就很大。

5
你需要做的是创建一个自定义的Formatter,将一个LogRecord进行JSON编码(当然,可能需要先预处理一下)。然后,你需要创建一个标准输出处理器,使用默认的formatter,和一个文件处理器,使用你的自定义formatter。这并不完全容易,但是高级日志教程可以帮助你入手,并且食谱中的一些部分有一些相关的示例代码。 - abarnert
1
最后一件事:你不能只是“将JSON写入文件”。嗯,你_可以_(只要你的顶级文本始终是对象或数组),但大多数JSON解析代码无法处理单个文件中任意流的JSON文本。你需要使用类似JSON Lines的东西,或者两种几乎相同的格式之一,它们稍微限制了JSON编码中允许的内容,以便您可以保证每行文本文件上只有一个JSON文本。 - abarnert
2
@abarnert 我目前正在探索食谱的高级部分。它似乎是可行的。也许吧。我不确定它如何处理我已经在记录的大型JSON对象,但我尝试一下。谢谢! - securisec
@securisec 同时,如果您成功解决了问题,并且能够将其压缩成一个示例以帮助其他人,那么您应该编写并接受自己问题的答案。 - abarnert
会做的@abarnert。我正在探索您的第一个答案,关于设置两个格式化程序,但是您在第三个答案中是正确的;使用JSON格式的输出确实很麻烦。 - securisec
显示剩余4条评论
9个回答

34

我也遇到了这个问题,但个人认为对于这样的问题,使用外部库可能有些过度。

我研究了一下 logging.Formatter 的代码,并编写了一个子类,在我的案例中实现了目标(我的目标是生成一个JSON文件,由Filebeat读取并进一步记录到ElasticSearch中)。

类:

import logging
import json


class JsonFormatter(logging.Formatter):
    """
    Formatter that outputs JSON strings after parsing the LogRecord.

    @param dict fmt_dict: Key: logging format attribute pairs. Defaults to {"message": "message"}.
    @param str time_format: time.strftime() format string. Default: "%Y-%m-%dT%H:%M:%S"
    @param str msec_format: Microsecond formatting. Appended at the end. Default: "%s.%03dZ"
    """
    def __init__(self, fmt_dict: dict = None, time_format: str = "%Y-%m-%dT%H:%M:%S", msec_format: str = "%s.%03dZ"):
        self.fmt_dict = fmt_dict if fmt_dict is not None else {"message": "message"}
        self.default_time_format = time_format
        self.default_msec_format = msec_format
        self.datefmt = None

    def usesTime(self) -> bool:
        """
        Overwritten to look for the attribute in the format dict values instead of the fmt string.
        """
        return "asctime" in self.fmt_dict.values()

    def formatMessage(self, record) -> dict:
        """
        Overwritten to return a dictionary of the relevant LogRecord attributes instead of a string. 
        KeyError is raised if an unknown attribute is provided in the fmt_dict. 
        """
        return {fmt_key: record.__dict__[fmt_val] for fmt_key, fmt_val in self.fmt_dict.items()}

    def format(self, record) -> str:
        """
        Mostly the same as the parent's class method, the difference being that a dict is manipulated and dumped as JSON
        instead of a string.
        """
        record.message = record.getMessage()
        
        if self.usesTime():
            record.asctime = self.formatTime(record, self.datefmt)

        message_dict = self.formatMessage(record)

        if record.exc_info:
            # Cache the traceback text to avoid converting it multiple times
            # (it's constant anyway)
            if not record.exc_text:
                record.exc_text = self.formatException(record.exc_info)

        if record.exc_text:
            message_dict["exc_info"] = record.exc_text

        if record.stack_info:
            message_dict["stack_info"] = self.formatStack(record.stack_info)

        return json.dumps(message_dict, default=str)

用法:

只需将格式化程序传递给日志处理程序即可。

    json_handler = FileHandler("foo.json")
    json_formatter = JsonFormatter({"level": "levelname", 
                                    "message": "message", 
                                    "loggerName": "name", 
                                    "processName": "processName",
                                    "processID": "process", 
                                    "threadName": "threadName", 
                                    "threadID": "thread",
                                    "timestamp": "asctime"})
    json_handler.setFormatter(json_formatter)

说明:

logging.Formatter 接收一个字符串,通过插值输出格式化的日志记录;而 JsonFormatter 接收一个字典,其中键会成为 JSON 字符串中记录值的键,而值则是可以被记录的 LogRecord 的属性对应的字符串。(在文档中提供了可用的列表 这里)。

主要的“问题”是解析日期和时间戳,而默认的格式化程序实现具有这些类属性:default_time_formatdefault_msec_format

default_msec_format 传递给 time.strftime(),并且 default_msec_format 被插值以添加毫秒,因为 time.strftime() 不提供这些的格式化选项。

原则是这些现在是实例属性,可以以 time_formatmsec_format 的形式提供,以自定义父类(未更改,因为它没有被重写)formatTime() 方法的行为方式。

如果您想要自定义时间格式,则可以技术上覆盖它,但我个人发现使用其他东西要么冗余,要么限制了实际的格式化选项。但是根据您的需要随意调整。

输出:

使用上述格式化选项记录的示例 JSON 记录,其中类中设置了默认的时间格式选项,如下所示:

{"level": "INFO", "message": "Starting service...", "loggerName": "root", "processName": "MainProcess", "processID": 25103, "threadName": "MainThread", "threadID": 4721200640, "timestamp": "2021-12-04T08:25:07.610Z"}

作为扩展(因为主机名不在日志规范中),我改变了 formatMessage 方法的返回值,使其返回 {fmt_key: host if fmt_key == "host" else record.__dict__[fmt_val] for fmt_key, fmt_val in self.fmt_dict.items()}。同时,在文件开头添加了 import platformhost = platform.node()。当从多个主机发送日志并集中消费时,比如将日志发送到 elastisearch 或类似的工具中,我发现这种方法非常有用。(我不是 Python 专家,可能有更好的实现方式)。(在实例化 jsonFormatter 时添加 "host":"")。 - r2evans
1
@r2evans 我肯定计划按照你所写的方式来做,只是为了为logstash准备数据。我不想在logstash中编写格式化程序/解析器,我的主要后端是Python,我希望在(发送至)logstash之前预先结构化日志数据。因此,我正在添加自定义参数,如主机、IP以及前端应用在调用API时发送的一些浏览器活动。 - Mehmet Burak Sayıcı

14
所以根据@abarnert的建议,我找到了这个链接,它为这个概念的大部分工作提供了一个良好的路径。目前的代码如下:
logger=logging.getLogger()
logger.setLevel(logging.DEBUG)

file_handler=logging.FileHandler('foo.log')
stream_handler=logging.StreamHandler()

stream_formatter=logging.Formatter(
    '%(asctime)-15s %(levelname)-8s %(message)s')
file_formatter=logging.Formatter(
    "{'time':'%(asctime)s', 'name': '%(name)s', \
    'level': '%(levelname)s', 'message': '%(message)s'}"
)

file_handler.setFormatter(file_formatter)
stream_handler.setFormatter(stream_formatter)

logger.addHandler(file_handler)
logger.addHandler(stream_handler)

虽然它不能完全满足要求,但它不需要任何预处理,并且允许我创建两个日志处理程序。
之后,我可以使用类似以下的方式:
from ast import literal_eval

with open('foo.log') as f:
    for line in f:
        for key, value in literal_eval(line):
            do something ...

使用dict对象而不是与格式不正确的JSON斗争,以实现我所设定的目标。
仍然希望有一个更优雅的解决方案。

1
很遗憾,输出数据不是有效的JSON数据。json.load("{'time':'2020-11-04 15:42:30,193', 'name': 'Jack\"s'}") - ahuigo
9
注意,如果你的消息中有单引号,你的解决方案将会出错。 - overgroove
1
@overgroove,为了解决这个问题,我只是使用了json.dumps并删除了引号。import jsonfile_formatter=logging.Formatter(json.dumps({'time':'%(asctime)s', 'name': '%(name)s', 'level': '%(levelname)s', 'message': '%(message)s'})) - deesolie
@deesolie ... 或者只需创建自己的格式化程序来执行 json.dumps 这个操作:https://stackoverflow.com/a/77064730/1603480无论如何,这个解决方案并不理想,可能不应该被标记为接受的答案。 - undefined

4

我希望能够得到JSON输出,这样我就可以在Promtail和Loki中更好地处理它。我刚刚更新了我的logging.json中的格式化程序,我将其用作dictConfig。

"formatters": {
    "normalFormatter": {
        "format": "{\"time\": \"%(asctime)s\", \"name\": \"[%(name)s]\", \"levelname\": \"%(levelname)s\", \"message\": \"%(message)s\"}"
    }
}

加载配置并获取根记录器,如下:

import json
import logging


# setup logger
with open("logging.json") as f:
    config_dict = json.load(f)
    logging.config.dictConfig(config_dict)

# get root logger
logger = logging.getLogger(__name__)

如果你想使用dictConfig,请确保你提供了所有必要的字段。

https://docs.python.org/3/library/logging.config.html

你可能需要定义一个格式化程序,一个处理程序(使用格式化程序),以及一个日志记录器(使用处理程序)。
例如,logging.json。
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
    "normalFormatter": {
        "format": "{\"time\": \"%(asctime)s\", \"name\": \"[%(name)s]\", \"levelname\": \"%(levelname)s\", \"message\": \"%(message)s\"}"
    }
},
"handlers": {
    "demohandler": {
        "level": "INFO",
        "formatter": "normalFormatter",
        "class": "logging.handlers.TimedRotatingFileHandler",
        "filename": "./files/logs/YourLogFile.log",
        "when": "d",
        "interval": 30,
        "backupCount": 4,
        "utc": true
    }
},
"loggers": {
    "root": {
        "handlers": ["demohandler"],
        "level": "INFO"
    },
    "someModule": {
        "handlers": ["demohandler"],
        "level": "INFO",            
        "propagate": 0
    }        
}
}

3
我可以使用这个自定义格式化程序来实现这个结果:
import json
import logging


class CustomJsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        super(CustomJsonFormatter, self).format(record)
        output = {k: str(v) for k, v in record.__dict__.items()}
        return json.dumps(output)


cf = CustomJsonFormatter()
sh = logging.StreamHandler()
sh.setFormatter(cf)

logger = logging.getLogger("my.module")
logger.addHandler(sh)

# simple json output
logger.warning("This is a great %s!", "log")
# enrich json output
logger.warning("This is an even greater %s!", "log", extra={'foo': 'bar'})

输出:

{"name": "my.module", "msg": "This is a great %s!", "args": "('log',)", "levelname": "WARNING", "levelno": "30", "pathname": "/Users/olivier/test.py", "filename": "test.py", "module": "test", "exc_info": "None", "exc_text": "None", "stack_info": "None", "lineno": "20", "funcName": "<module>", "created": "1661868378.5048351", "msecs": "504.8351287841797", "relativeCreated": "1.3060569763183594", "thread": "4640826880", "threadName": "MainThread", "processName": "MainProcess", "process": "81360", "message": "This is a great log!"}
{"name": "my.module", "msg": "This is an even greater %s!", "args": "('log',)", "levelname": "WARNING", "levelno": "30", "pathname": "/Users/olivier/test.py", "filename": "test.py", "module": "test", "exc_info": "None", "exc_text": "None", "stack_info": "None", "lineno": "22", "funcName": "<module>", "created": "1661868378.504962", "msecs": "504.9619674682617", "relativeCreated": "1.4328956604003906", "thread": "4640826880", "threadName": "MainThread", "processName": "MainProcess", "process": "81360", "foo": "bar", "message": "This is an even greater log!"}

2

使用此代码,您可以将完整的回溯、时间戳和级别添加到所选的json文件中。

import json
import traceback
from datetime import datetime

def addLogging(logDict:dict):
    loggingsFile = 'loggings.json'

    with open(loggingsFile) as f:
        data = json.load(f)

    data.append(logDict)

    with open(loggingsFile, 'w') as f:
        json.dump(data, f)

def currentTimeUTC():
    return datetime.now().strftime('%d/%m/%Y %H:%M:%S')

try:
    print(5 / 0)
except ZeroDivisionError:
    fullTraceback = str(traceback.format_exc())
    addLogging({'timestamp': currentTimeUTC(), 'level': 'error', 'traceback': fullTraceback})

输出:

[
    {
        "timestamp": "09/06/2020 17:38:00",
        "level": "error",
        "traceback": "Traceback (most recent call last):\n  File \"d:testFullTraceback.py\", line 19, in <module>\n    print(5/0)\nZeroDivisionError: division by zero\n"
    }
]

那么在每个异常处理程序中都必须显式调用addLogging()吗? - user2399453

1

我能够使用python-json-logger库创建它,这很简单且易于使用。

Django


from pythonjsonlogger import jsonlogger

##This is to add custom keys
class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
        log_record['level'] = record.levelname


# Logging
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'json': {
            '()': CustomJsonFormatter, # if you want to use custom logs class defined above
            # '()': jsonlogger.JsonFormatter, # without custom logs
            'format': '%(level)s %(levelname)s %(asctime)s %(module)s %(process)d %(thread)d %(message)s %(threadName)s %(special)s %(run)s\
                %(name)s %(created)s %(processName)s %(relativeCreated)d %(funcName)s %(levelno)d %(msecs)d %(pathname)s %(lineno)d %(filename)s'
        },
    },
    'handlers': {
        'null': {
            'class': 'logging.NullHandler',
        },
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'json' 
        },
    },
    .....
}

Flask

from logging.config import dictConfig
from pythonjsonlogger import jsonlogger
import os

# This will set global root logging config across all the modules using in the app

##This is to add custom keys
class CustomJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
        log_record['level'] = record.levelname


def setup():
    LOG_FILE = '/tmp/app/app.json'
    if not os.path.exists(os.path.dirname(LOG_FILE)):  # if LOG_FILE dir doesn't exist, creates it.
        os.makedirs(os.path.dirname(LOG_FILE))
    dictConfig({
        'version': 1,
        'formatters': {
            'default': {
                'format': '%(asctime)s - %(module)s - %(levelname)s - %(message)s',
            },
            'json': {
            '()': CustomJsonFormatter, # if you want to use custom logs class defined above
            # '()': jsonlogger.JsonFormatter, # without custom logs
            'format': '%(level)s %(levelname)s %(asctime)s %(module)s %(process)d %(thread)d %(message)s %(threadName)s %(special)s %(run)s\
                %(name)s %(created)s %(processName)s %(relativeCreated)d %(funcName)s %(levelno)d %(msecs)d %(pathname)s %(lineno)d %(filename)s'
        },
        },
        'handlers': {'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': LOG_FILE,
            'maxBytes': 10485760,
            'backupCount': 5,
            'formatter': 'json'
        },
        'console':{
            'class':'logging.StreamHandler',
            'formatter': 'json'
        }},
        'root': {
            'level': 'INFO',
            'handlers': ['file', 'console']
        }
    })


希望这能帮助你轻松设置。

1

如果您不介意安装一个模块来帮助,可以使用json_log_formatter模块。JSON输出具有比请求的更多属性。repo提到了自定义输出的属性,但我还没有将其集成到一个工作示例中。

json_log_formatter

import logging
import json_log_formatter

# Set Basic Logging 
self.loggers = logging.getLogger(__name__)
self.loggers.setLevel(logging.DEBUG)
self.formatter = logging.Formatter(fmt='%(asctime)-15s %(levelname)-8s %(message)s', datefmt = '%a, %d %b %Y %H:%M:%S')


# Config for JSON File Handler
self.logFileHandler = logging.FileHandler(SOME-PATH, mode='a')
self.fileFormatter = json_log_formatter.VerboseJSONFormatter()
self.logFileHandler.setFormatter(self.fileFormatter)
self.logFileHandler.setLevel(logging.INFO)
self.loggers.addHandler(self.logFileHandler)


# Config for Stream Handler
self.logStreamHandler = logging.StreamHandler()
self.logStreamHandler.setFormatter(self.formatter)
self.logStreamHandler.setLevel(logging.INFO)
self.loggers.addHandler(self.logStreamHandler)

2
或者,只需使用@securisec的答案并使用json.dumps解决任何引号问题。这对我来说效果最好,但我将保留此帖子供其他人使用(如果他们选择)。 import jsonfile_formatter = logging.Formatter(json.dumps({'time':'%(asctime)s','name':'%(name)s','level':'%(levelname)s','message':'%(message)s'})) - deesolie

0
我想保存完整的LogRecord对象,以便稍后在模块的最大集成下检查我的日志。因此,我像这样检查了对象:
class Handler_json(Handler):

def emit(self, record: LogRecord) -> None:
    json_data = {}
    for attr in filter(lambda attr: not attr.endswith("__"), dir(record)):
        json_data[attr] = record.__getattribute__(attr)
    del json_data["getMessage"]
    print(json_data)

这是Handler的子类,emit是被重写的方法,每个LogRecord都会调用它。Dir返回对象的所有属性和方法。我排除了特殊方法,并删除了getMessage方法,因为它对于JSON对象表示不需要。

这可以很好地集成到日志记录中,如下所示:

logger = getLogger(__name__)
logger.setLevel(DEBUG)
handle_json = Handler_json()
logger.addHandler(handle_json)
logger.info("my info")

结果看起来像这样:

{
'args': (),
'created': 1639925351.0648422,
'exc_info': None,
'exc_text': None,
'filename': 'my_logging.py',
'funcName': 'restore_log_from_disk',
'levelname': 'INFO',
'levelno': 20,
'lineno': 142,
'module': 'my_logging',
'msecs': 64.84222412109375,
'msg': 'my info',
'name': '__main__',
'pathname': 
'/home/jindrich/PycharmProjects/my_logging.py',
'process': 146331,
'processName': 'MainProcess',
'relativeCreated': 1.6417503356933594,
'stack_info': None,
'thread': 140347192436544,
'threadName': 'MainThread'
}

然后你可以从磁盘加载数据,在文档中进行一些挖掘后重新创建对象。


不必通过循环和过滤器创建一个字典并填充,使用字典推导会更简洁:json_data = {a: getattr(record, a) for a in dir(record) if not a.endswith("__")}。而且从全局角度来看,创建一个格式化器比创建一个处理器更容易。 - undefined

0
你还可以设计一个简单的格式化程序,就像这个例子一样(基本上创建一个字典并将其序列化为JSON格式)。
然后添加一个使用这个格式化程序将数据转储到文件的处理程序,以及一个常规处理程序用于在控制台打印,这样你就可以开始了。
下面是一个示例,我想在执行后将所有数据加载到Pandas DataFrame中:
import json
import logging
from logging import LogRecord

import pandas as pd


class JsonFormatter(logging.Formatter):
    """Formatter to dump error message into JSON"""

    def format(self, record: LogRecord) -> str:
        record_dict = {
            "level": record.levelname,
            "date": self.formatTime(record),
            "message": record.getMessage(),
            "module": record.module,
            "lineno": record.lineno,
        }
        return json.dumps(record_dict)



logger = logging.getLogger()
logger.setLevel(logging.INFO)

json_h = logging.FileHandler("errors.ndjson", mode="w")
formatter = JsonFormatter()
json_h.setFormatter(formatter)
logger.addHandler(json_h)

stream_h = logging.StreamHandler()
stream_h.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(stream_h)


logger.error("""some error with "double quotes" and 'single quotes'""")
logger.critical("some critical text")
logger.warning("some warning")
logger.info("some info")

df = pd.read_json("errors.ndjson", lines=True)
print(df.dtypes)  # type: ignore
print(df)

你还可以使用专用模块Python JSON Logger
...
from pythonjsonlogger import jsonlogger


log_format = "%(asctime)s - %(module)s#%(lineno)d - %(levelname)s - %(message)s"
formatter = jsonlogger.JsonFormatter(
    fmt=log_format, rename_fields={"levelname": "level", "asctime": "date"}
)

logger = logging.getLogger()

json_h= logging.FileHandler("errors.ndjson", mode="w")
json_h.setFormatter(formatter)
logger.addHandler(json_h)
...


请注意,与python-json-logger相比,简单实现具有以下限制:
  • 硬编码的属性选择(但可以通过Formatter类的参数进行参数化)
  • 不支持Traceback / Exception(即exc_info
  • 不显示extra信息(例如logger.info("message", extra={"field": "value"}

输出:

{"level": "ERROR", "date": "2023-09-08 15:05:26,816", "message": "some error with \"double quotes\" and 'single quotes'", "module": "try_json_logger", "lineno": 50}
{"level": "CRITICAL", "date": "2023-09-08 15:05:26,816", "message": "some critical text", "module": "try_json_logger", "lineno": 51}
{"level": "WARNING", "date": "2023-09-08 15:05:26,816", "message": "some warning", "module": "try_json_logger", "lineno": 52}
{"level": "INFO", "date": "2023-09-08 15:05:26,816", "message": "some info", "module": "try_json_logger", "lineno": 53}

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接