如何将Python应用日志推送至Splunk？前置条件及实现方法咨询

前置要求 #

• Splunk端必须先启用HTTP Event Collector（HEC）服务，生成有效HEC Token，确认Splunk实例的访问地址、端口在Python应用网络环境下可通
• 提前确认Splunk侧分配的索引名称、源类型（sourcetype），推送时的参数需和Splunk侧配置一致
• 若使用HTTPS协议推送，公网环境建议配置SSL证书校验，内网测试环境可关闭校验减少配置步骤

日志格式要求 #

不需要强制提前转换为键值对结构，Splunk本身支持非结构化日志的接入和后续字段提取，但更推荐提前转换为JSON或键值对的结构化格式：

• 不需要在Splunk侧额外编写字段提取规则，降低后续日志检索、统计分析的使用成本
• 避免非结构化日志解析时出现字段匹配错误、数据丢失的问题
  可直接配合Python的JSON日志格式化工具实现结构化转换，不需要额外开发复杂逻辑。

splunk_handler实操示例 #

依赖安装 #

执行命令安装所需包：
pip install splunk-handler python-json-logger

可运行示例代码 #

import logging
from splunk_handler import SplunkHandler
from pythonjsonlogger import jsonlogger

# 替换为自身环境的实际配置
SPLUNK_CONFIG = {
    "host": "你的Splunk实例IP/域名",
    "port": 8088, # HEC服务默认端口，修改过则替换为实际端口
    "token": "你的HEC令牌",
    "index": "分配的日志索引名",
    "sourcetype": "自定义的日志源类型",
    "verify": False, # 公网环境建议设置为证书路径开启校验，内网可关闭
    "protocol": "https"
}

# 初始化Splunk日志处理器
splunk_handler = SplunkHandler(**SPLUNK_CONFIG)
# 配置JSON结构化日志格式
formatter = jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s %(module)s %(funcName)s %(lineno)d"
)
splunk_handler.setFormatter(formatter)
splunk_handler.setLevel(logging.INFO)

# 绑定到日志实例
logger = logging.getLogger("python_app_log")
logger.addHandler(splunk_handler)
logger.setLevel(logging.INFO)

# 测试推送
if __name__ == "__main__":
    # 普通日志
    logger.info("普通日志推送测试")
    # 带自定义业务字段的结构化日志
    logger.info(
        "用户操作日志",
        extra={
            "user_id": 10086,
            "operate": "submit_order",
            "status": "success",
            "cost_ms": 156
        }
    )

常见排查点 #

如果运行后Splunk侧无法查询到日志，优先检查：

• HEC是否开启了对应索引的写入权限，HEC Token是否有效
• Python应用所在服务器到Splunk实例的对应端口是否放行，无防火墙拦截
• Splunk侧的时间范围选择是否覆盖日志生成的时间

内容的提问来源于stack exchange，提问作者Sarvendra Singh