Python如何通过requests.post向Fluentd端点批量发送大量数据

问题根因 #

你遇到的414状态码（URI Too Long）和请求体承载的日志条数无关，是代码逻辑错误导致：你在遍历每条日志的循环中重复给URL拼接?time=参数，提交的日志越多，URL长度就越长，超过Fluentd或前置代理的URI长度限制就会触发报错。

代码修复 #

你需要把time参数从URL移到请求体的日志结构体中，修复后的代码示例如下：

import sys
import json
import requests

# 用with语法自动管理文件句柄，避免泄漏
with open('/etc/td-agent/data2.json', 'r') as f:
    data = json.load(f)

actions = []
url = ''
u_name = ''
p_word = ''

with open(sys.argv[-1], 'r') as input_file:
    for line in input_file:
        temp = json.loads(line)
        tenantid = temp['HTTP_FLUENT_TAG']
        message = temp['message']
        message_json = json.loads(message)
        h_name = data['account_details'][tenantid]['hostname']
        u_name = data['account_details'][tenantid]['username']
        p_word = data['account_details'][tenantid]['password']
        url = f'https://{h_name}'
        for element in message_json:
            action = {
                "msg": element['log'],
                "id": element['ID'],
                "time": str(element['date'])
            }
            actions.append(action)

r = requests.post(url, auth=(u_name, p_word), json=actions)
print(r.status_code)

Fluentd批量提交实现方案 #

Fluentd原生支持批量日志提交，常用方案有两种：

1. HTTP接口批量提交 #

就是你当前使用的方式，修复URL问题后即可直接使用：

• 提交格式：将多条日志组装为JSON数组放到POST请求体即可，不需要调用额外的批量API
• 配置调整：如果遇到请求体大小限制，修改Fluentd http输入插件的body_size_limit配置，比如设置为body_size_limit 100m即可支持单次100MB的请求体；如果Fluentd前有Nginx等反向代理，也需要同步调整代理的client_max_body_size配置
• 性能建议：单次批量大小控制在5~20MB范围时传输效率最优

2. Forward协议批量提交 #

如果对传输性能、可靠性要求更高，推荐使用Fluentd原生Forward协议：

• 优势：原生支持批量压缩、断点重试、传输保序，性能比HTTP接口高30%以上
• Python实现：可以直接使用fluent-logger库实现批量提交，示例如下：

from fluent import sender

# 初始化Fluentd连接
sender.setup(
    tag='your_log_tag',
    host='你的Fluentd服务地址',
    port=24224, # Forward协议默认端口，需要Fluentd配置开启forward输入
    buffer_overflow_handler=lambda x: print(f'缓存溢出，丢弃日志: {x}')
)

# 批量发送日志
for log_item in actions:
    sender.emit('log', log_item)

# 关闭连接，强制刷入所有缓存日志
sender.close()

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