如何配置Splunk HEC实现每秒批量发送最多1000条应用事件

一、Splunk HEC端前置调优 #

先把接收端配置改对，不然发送端压再高也会被限流丢数：

• 调整对应HEC令牌的限流阈值：默认单令牌限流为1MB/s，按单条日志平均1KB计算，1000条/秒刚好打满默认阈值，需要把对应令牌的「每秒最大接收数据量」调到2MB/s以上留冗余，同时关闭单令牌的请求频率限制
• 优先使用你示例中的/services/collector/raw端点，相比结构化事件端点性能高30%以上，不需要为每条日志单独封装JSON结构，直接传换行分隔的纯文本即可
• 调整HEC接收队列长度为默认值的2倍，确认HEC所在节点CPU、内存预留充足，避免突发流量导致队列溢出丢事件

注意：你给出的curl示例里用空格分隔多条日志的写法是错误的，/services/collector/raw端点要求多条事件之间必须用换行符\n分隔，否则HEC会将整段请求体识别为单条日志，导致解析失败、字段提取错误。正确的单次批量请求格式参考：

curl "https://splunk-example.com:8088/services/collector/raw?channel=093DCD-BC98-8UET-8AFE-8413C3825C4C&sourcetype=test_type&index=test_index" \
-H "Authorization: Splunk ******-****-****-****-*********" \
-d $'<log line 1>\n<log line 2>\n<log line 3>\n<log line 4>'

二、发送端核心实现逻辑 #

不要用单条循环curl的方式发送，这种方式受TCP连接重建开销限制，每秒最多发送200条左右，完全达不到1000条/秒的要求，必须实现攒批发送+精确流控+连接复用：

• 内存攒批：本地维护固定长度的内存日志队列，每攒够100-200条日志，或者攒批等待时间达到100ms（两个条件满足任意一个），就将批次内的所有日志用换行符拼接成一次请求发送，单次请求体大小控制在512KB-1MB之间，避免包过大导致超时重传
• 精确流控：用令牌桶算法实现速率控制，按1000条/秒的阈值匀速释放令牌，比如每10ms释放10个令牌，拿到令牌的日志才能进入发送队列，不要靠sleep()硬等，速率误差会非常大
• 连接复用：开启HTTP长连接（Keep-Alive）和TLS会话复用，避免每次请求都重新建立TCP连接，能减少90%以上的网络开销
• 异常重试：对返回429（限流）、503（服务不可用）、网络超时的请求，用100ms起步的指数退避策略重试，不要无间隔死循环重试避免打崩链路

以下是可直接复用的最小实现示例（Python）：

import time
import requests
from queue import Queue
from threading import Thread

# 基础配置
HEC_ENDPOINT = "https://splunk-example.com:8088/services/collector/raw?channel=093DCD-BC98-8UET-8AFE-8413C3825C4C&sourcetype=test_type&index=test_index"
HEC_AUTH_HEADER = {"Authorization": "Splunk ******-****-****-****-*********"}
TARGET_SEND_RATE = 1000  # 目标每秒发送条数
BATCH_MAX_SIZE = 100     # 单批次最大日志条数
BATCH_MAX_WAIT = 0.1     # 单批次最大等待时间，单位秒

log_queue = Queue(maxsize=2000)
# 初始化会话复用连接
http_session = requests.Session()
http_session.headers.update(HEC_AUTH_HEADER)
http_session.keep_alive = True

def send_worker():
    batch_buffer = []
    last_send_time = time.time()
    while True:
        # 从队列取日志，超时则触发批次发送
        try:
            log_line = log_queue.get(timeout=BATCH_MAX_WAIT)
            batch_buffer.append(log_line)
        except:
            pass

        # 满足批次条件则发送
        if len(batch_buffer) >= BATCH_MAX_SIZE or (time.time() - last_send_time >= BATCH_MAX_WAIT and batch_buffer):
            payload = "\n".join(batch_buffer).encode("utf-8")
            try:
                resp = http_session.post(HEC_ENDPOINT, data=payload, timeout=5)
                # 发送失败则将日志塞回队列重试
                if resp.status_code != 200 or resp.json().get("code") != 0:
                    for line in reversed(batch_buffer):
                        log_queue.put(line, block=False)
                    time.sleep(0.1)
            except Exception:
                for line in reversed(batch_buffer):
                    log_queue.put(line, block=False)
                time.sleep(0.1)
            # 重置批次缓存
            batch_buffer.clear()
            last_send_time = time.time()

def rate_limit_producer():
    """令牌桶流控+日志生产，实际使用时替换为你的日志采集逻辑即可"""
    token_bucket = 0
    last_tick_time = time.time()
    while True:
        current_time = time.time()
        time_pass = current_time - last_tick_time
        # 补充令牌
        token_bucket = min(token_bucket + time_pass * TARGET_SEND_RATE, TARGET_SEND_RATE)
        # 有令牌则生产一条日志
        if token_bucket >= 1:
            # 此处替换为实际读取到的业务日志
            log_queue.put(f"business log generated at {time.time()}")
            token_bucket -= 1
        last_tick_time = current_time
        time.sleep(0.01) # 10ms调度精度足够支撑1000条/秒的速率要求

if __name__ == "__main__":
    # 启动发送工作线程
    Thread(target=send_worker, daemon=True).start()
    rate_limit_producer()

三、效果校验 #

• 发送端本地计数：每10秒统计一次实际发送成功的日志条数，校准速率误差，稳定在950-1050条/秒区间即可，不需要追求绝对零误差
• Splunk端校验：执行搜索语句index=test_index sourcetype=test_type | timechart span=1s count，查看实际入索引的日志速率是否符合预期，有没有丢数
• 监控HEC节点CPU使用率，如果稳定超过70%，需要扩容HEC节点或者调整节点资源配额，避免性能瓶颈

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