如何通过Python Logging向ELK Stack发送带自定义字段的结构化日志？

你遇到的问题本质是Logstash未配置解析JSON格式的输入，导致Python发送的JSON字符串被整体识别为message字段内容。由于无法修改Logstash配置，可通过以下两种方式从应用端解决：

方案1：生成键值对格式日志（适配Logstash默认解析） #

修改日志格式化器，输出key=value格式的字符串，Logstash的默认kv过滤器会自动将这些键值对解析为独立字段：

import logging
from logstash_async.handler import AsynchronousLogstashHandler
from datetime import datetime

class CustomKVFormatter(logging.Formatter):
    def format(self, record):
        # 生成UTC时间戳，符合ELK标准格式
        timestamp = datetime.utcfromtimestamp(record.created).isoformat() + "Z"
        # 构造键值对日志片段
        log_parts = [
            f"timestamp={timestamp}",
            f"level={record.levelname}",
            f"logger={record.name}",
            f"message={repr(record.getMessage())}",
            f"filename={record.pathname}",
            f"funcName={record.funcName}",
            f"appName=MyPythonApp"
        ]
        # 处理异常信息
        if record.exc_info:
            exc_text = self.formatException(record.exc_info)
            log_parts.append(f"exception={repr(exc_text)}")
        return " ".join(log_parts)

def activate_logging() -> None:
    logstash_handler = AsynchronousLogstashHandler("0.0.0.0", 5000, database_path=None)
    logstash_handler.setFormatter(CustomKVFormatter())
    logging.basicConfig(
        level=logging.INFO,
        force=True,
        handlers=[logstash_handler]
    )

使用时直接调用日志方法即可，额外字段也会被自动解析：

logger = logging.getLogger(__name__)
logger.info("用户登录成功", extra={"user_id": 123})

方案2：强制Logstash识别JSON格式（需确认Logstash输入支持） #

如果Logstash的TCP/UDP输入实际配置了codec => json（只是你未察觉），可调整格式化器生成Logstash兼容的JSON结构，确保自定义字段处于顶级层级：

import logging
from logstash_async.formatter import LogstashFormatter
try:
    import json
except ImportError:
    import simplejson as json
from logstash_async.handler import AsynchronousLogstashHandler

class CustomLogstashFormatter(LogstashFormatter):
    def format(self, record):
        # 先基于父类生成Logstash标准JSON结构
        base_log = json.loads(super().format(record))
        # 注入自定义顶级字段
        base_log.update({
            "filename": record.pathname,
            "funcName": record.funcName,
            "appName": "MyPythonApp"
        })
        return json.dumps(base_log)

def activate_logging() -> None:
    logstash_handler = AsynchronousLogstashHandler("0.0.0.0", 5000, database_path=None)
    # 使用默认参数生成Logstash兼容格式
    formatter = CustomLogstashFormatter(message_type='python')
    logstash_handler.setFormatter(formatter)
    logging.basicConfig(
        level=logging.INFO,
        force=True,
        handlers=[logstash_handler]
    )

关键说明 #

• 方案1无需依赖Logstash的任何特殊配置，兼容性最强，适合无法修改Logstash的场景。
• 方案2需要Logstash输入配置了codec => json，否则依然会将JSON字符串放入message字段。

内容的提问来源于stack exchange，提问作者Luigi Loria