如何在Structlog日志中添加Message字段及ElasticSearch集成最佳实践

一、确保日志包含"message"字段 #

structlog默认会将logger.info()的第一个参数存入event字段，而非message字段。要实现需求，可通过以下两种方式修改配置：

方法1：将"event"字段重命名为"message" #

在structlog.configure()的处理器列表中添加structlog.processors.EventRenamer("message")，直接将默认的event字段替换为message：

修改后的structlog.configure配置：

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.filter_by_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.EventRenamer("message"),  # 添加该处理器
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

方法2：保留"event"字段同时添加"message"字段 #

如果需要同时保留event和message字段，可以自定义一个处理器，将event的值复制到message：

def add_message_field(logger, method_name, event_dict):
    # 将event字段的值复制到message
    event_dict["message"] = event_dict.get("event")
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.filter_by_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        add_message_field,  # 添加自定义处理器
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

修改后，调用logger.info("push notification subscribed successfully!")时，日志会按配置生成对应的message字段；对于django-structlog中间件生成的日志（如request_finished），上述配置同样会将event字段转换或复制为message字段，匹配你的期望格式。

二、结构化日志推送至ElasticSearch的最佳实践 #

1. 选择可靠的推送方式 #

• Filebeat + ElasticSearch：推荐使用Filebeat收集本地JSON日志文件（如你配置的json_file），再批量推送到ES。这种方式解耦应用与ES，避免ES故障影响业务，且Filebeat自带重试、本地缓存机制，可靠性更高。
• 直接ES Handler：若需实时推送，可使用elasticsearch.handlers.LoggingHandler，但必须配置批量提交参数，避免频繁请求ES导致性能损耗。

2. 规范字段与索引映射 #

• 提前定义ES索引的映射模板，指定字段类型：例如将timestamp设为date类型，code设为integer，request_id、user_id设为keyword类型，避免ES自动映射导致的类型混乱。
• 统一字段命名规则，避免大小写、格式不一致，方便后续查询与分析。

3. 索引生命周期管理（ILM） #

• 按时间分片创建索引（如app-logs-YYYY-MM-DD），配合ES的索引生命周期策略，自动完成索引的滚动、归档、删除，减少存储成本，提升查询性能。

4. 批量推送优化 #

• 开启批量推送功能：Filebeat可配置bulk_max_size，ES Handler可设置bulk_size，减少网络请求次数，提升推送效率。

5. 故障处理与可靠性 #

• 配置本地缓存：当ES不可用时，将日志临时缓存到本地文件或队列，待恢复后再推送，避免日志丢失。
• 添加监控告警：监控ES的推送成功率、延迟等指标，及时发现故障。

6. 安全配置 #

• 使用HTTPS连接ES集群，开启身份验证（用户名/密码或API密钥），确保日志传输与存储的安全性。
• 限制ES用户权限，仅赋予日志推送所需的写入权限，避免过度授权。

7. 性能隔离 #

• 避免在业务请求线程中同步推送日志，可使用异步线程或队列（如Celery）处理日志推送，防止ES性能波动影响业务响应速度。

内容的提问来源于stack exchange，提问作者Keshav Khanal