Python 中 JSON 日志的旋转文件处理程序

Question

我正在使用 Python 保存 JSON 日志。下面是代码：

log_file = 'app_log.json'

log_json = dict()
log_json["Data"] = {}

log_json['Data']['Key1'] = "value1"
log_json['Data']['alert'] = False
log_json['Data']['Key2'] = "N/A"

log_json['Created'] = datetime.datetime.utcnow().isoformat()

with open(log_file, "a") as f:
    json.dump(log_json, f)#, ensure_ascii=False)
    f.write("\n")

现在，上面的代码正在生成日志文件。但我注意到文件大小增加了很多，将来我可能会面临磁盘空间问题。我想知道是否有任何可用于 JSON 的预构建旋转文件处理程序，其中我们可以提到固定大小，比方说 100MB，达到此大小后，它将删除并重新创建新文件。

我之前曾使用

from logging.handlers import RotatingFileHandler

来处理

.log

文件，但也想对

.json

文件执行此操作。

Answer 1

Python 不关心日志文件名。

您也可以将用于 .log 文件的 旋转处理程序 用于 .json 文件。

请参阅下面的示例

# logging_example.py

import logging
import logging.handlers
import os
import time

logfile = os.path.join("/tmp", "demo_logging.json")

logger = logging.getLogger(__name__)

fh = logging.handlers.RotatingFileHandler(logfile, mode='a', maxBytes=1000, backupCount=5)  # noqa:E501

fh.setLevel(logging.DEBUG)

formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")

fh.setFormatter(formatter)

logger.addHandler(fh)
logger.setLevel(logging.DEBUG)

while 1:
    time.sleep(1)
    logger.info("Long string to increase the file size")

如果您在 Unix 环境中工作，还可以查看 logrotate。这是一个很棒且简单的工具，具有良好的文档，可以完全满足您的需求。

Answer 2

1. 您可以使用 RotatingFileHandler

    实现

   结构化日志记录

import json
import logging
import logging.handlers
from datetime import datetime

class StructuredMessage:
    def __init__(self, message, /, **kwargs):
        self.message = message
        self.kwargs = kwargs

    def __str__(self):
        return '%s >>> %s' % (self.message, json.dumps(self.kwargs))

_ = StructuredMessage   # optional, to improve readability

log_json = {}
log_json["Data"] = {}

log_json['Data']['Key1'] = "value1"
log_json['Data']['alert'] = False
log_json['Data']['Key2'] = "N/A"

log_json['Created'] = datetime.utcnow().isoformat()

LOG_FILENAME = 'logging_rotatingfile_example.out'

# Set up a specific logger with our desired output level
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# Add the log message handler to the logger
handler = logging.handlers.RotatingFileHandler(
              LOG_FILENAME, maxBytes=20, backupCount=5)
bf = logging.Formatter('%(message)s')
handler.setFormatter(bf)

logger.addHandler(handler)
logger.info(_('INFO', **log_json))

注意：检查此处以获取有关

structured-logging-python

的更多信息

2. 您也可以使用 json-logging-python 与

   RotatingFileHandler

import logging
import json
import traceback
from datetime import datetime
import copy
import json_logging
import sys

json_logging.ENABLE_JSON_LOGGING = True


def extra(**kw):
    '''Add the required nested props layer'''
    return {'extra': {'props': kw}}


class CustomJSONLog(logging.Formatter):
    """
    Customized logger
    """

    def get_exc_fields(self, record):
        if record.exc_info:
            exc_info = self.format_exception(record.exc_info)
        else:
            exc_info = record.exc_text
        return {'python.exc_info': exc_info}

    @classmethod
    def format_exception(cls, exc_info):
        return ''.join(traceback.format_exception(*exc_info)) if exc_info else ''

    def format(self, record):
        json_log_object = {"@timestamp": datetime.utcnow().isoformat(),
                           "level": record.levelname,
                           "message": record.getMessage(),
                           "caller": record.filename + '::' + record.funcName
                           }
        json_log_object['data'] = {
            "python.logger_name": record.name,
            "python.module": record.module,
            "python.funcName": record.funcName,
            "python.filename": record.filename,
            "python.lineno": record.lineno,
            "python.thread": record.threadName,
            "python.pid": record.process
        }
        if hasattr(record, 'props'):
            json_log_object['data'].update(record.props)

        if record.exc_info or record.exc_text:
            json_log_object['data'].update(self.get_exc_fields(record))

        return json.dumps(json_log_object)


json_logging.init_non_web(custom_formatter=CustomJSONLog, enable_json=True)

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
LOG_FILENAME = 'logging_rotating_json_example.out'

handler = logging.handlers.RotatingFileHandler(
              LOG_FILENAME, maxBytes=20, backupCount=5)

logger.addHandler(handler)

log_json = {}
log_json["Data"] = {}

log_json['Data']['Key1'] = "value1"
log_json['Data']['alert'] = False
log_json['Data']['Key2'] = "N/A"

logger.info('Starting')
logger.debug('Working', extra={"props":log_json})

注意：检查此处以获取有关

json-logging-python

的更多信息

Answer 3

您可以在写入/追加到文件之前尝试此操作。这应该检查文件是否已达到最大行数，然后它会从文件开头删除一行代码，然后像往常一样附加到文件末尾。

filename = 'file.txt'
maxLines = 100

count = len(open(filename).readlines())

if(count > maxLines) {
  with open(filename, 'r') as fin:
    data = fin.read().splitlines(True)
  with open(filename, 'w') as fout:
    fout.writelines(data[1:])
}