Wazuh 基准测试工具 Engine Event Streams从文本日志到引擎事件帧的压测负载实现【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh导读本文围绕 Wazuh 仓库中inventory_sync基准测试工具tool_simulator的第二类负载——engine event streams展开讲解它如何以纯文本日志文件为输入逐行生成1:location:line形式的引擎事件帧并通过与 inventory_sync 完全相同的 TCP/AES/zlib 传输栈送入wazuh-engine事件摄入链路。读完本文你将掌握 engine 步骤的场景 JSON 配置、四种终止语义、并发与限速模型、相关度量列以及从文本行到线上字节的完整帧封装链路。背景为什么基准工具需要第二类负载Wazuh 管理器的事件处理链路wazuh-engine即原analysisd负责对代理上报的日志进行解码与规则匹配。要对其进行压测传统做法是拉起大量真实代理成本高且难以控制负载形态。inventory_sync基准工具Go 实现入口为 cmd/benchmark_sender/main.go默认的负载是 inventory_sync 会话把 FlatBuffer 消息以s:module_id:前缀封装后发送。而engine event stream是发送端支持的第二种负载类型逐行读取一个文本文件每行产生一帧帧的标识符块identifier blob格式为1:location:line。两种负载共享完全相同的传输层同一个到 remoted 的 TCP 套接字、相同的 AES zlib 长度前缀封装、相同的按代理注册authd 1515 端口与 AES 密钥派生。区别只在于内部标识符块的形态——engine 帧不含s:前缀也没有 FlatBuffer 字节整帧就是一行明文。详细协议见 docu/05-wire-protocol.md。典型使用场景engine event stream 的设计目标集中在三类场景对管理器事件摄入路径产生负载在不拉起真实代理的前提下向wazuh-engine灌入事件观察解码/规则流水线的吞吐与稳定性。用已知语料压测解码器/规则管线将 apache、syslog、sshd 等真实日志行收集成文件作为语料输入结果可复现、可对比。混合负载同一个代理可以同时跑 inventory_sync 通道和 engine 流通道两条 lane 共享同一个 socket帧在 TCP 流上交错发送模拟真实代理既上报资产清单又上报日志的行为。线上格式Wire Format与帧链engine 事件的标识符块定义为identifier_blob 1: location : line其中line是文件中去掉末尾\n后的一行。随后整个 blob 交给wire.EncodeText(aesKey, agentID, identifier_blob)后续的 MD5 zlib Wazuh pad PKCS#7 AES 长度前缀封装与 inventory_sync 完全一致。开头的1是队列字节queue byte对应管理器端 engine/source/base/src/eventParser.cpp 中提取的/wazuh/protocol/queue字段ASCII 49 即1。该字节目前是硬编码的文档中说明将其暴露为按步骤可配置项属于后续工作。端到端帧链示例假设输入文件某行为Jun 1 00:00:01 host01 sshd[1234]: Accepted password for root当location syslog、使用标准测试夹具name bench-0001-aaaaaaaaaaaa、id 001、manager_key deadbeef…时完整的封装链为identifier_blob : 1:syslog:Jun 1 00:00:01 host01 sshd[1234]: Accepted password for root inner_event : MD5_hex(routing_prefix || identifier_blob) || routing_prefix || identifier_blob (其中 routing_prefix 55555 1234567891 : 5555 :) zlib_compressed : zlib.compress(inner_event, level6) wazuh_padded : (! × N) || zlib_compressed (N 8 - len%8, 或已对齐时为 8) aes_padded : PKCS#7 pad 到 16 字节对齐 encrypted : AES_256_CBC(aesKey, IV FEDCBA0987654321, aes_padded) frame : !001!#AES: || encrypted wire : uint32_le(len(frame)) || frame其中的routing_prefix555551234567891:5555:是管理器期望的代理路由魔数字符串IV固定为FEDCBA0987654321发送端与管理端os_crypto_aes_op.c两侧硬编码一致。这一链路的语义等价性由 internal/wire/parity_test.go 中的测试守护它用相同的(name, id, key)校验 AES 密钥派生结果、帧头!001!#AES:前缀以及EncodeText→DecodeFrame的往返一致。一个容易被忽略的实现细节行长度并非无上限。internal/engine/source.go中定义了wazuhMaxInnerEvent 65536对应 remoted 端OS_MAXSTR的静态解压缓冲区上限与wazuhInnerEventOverhead 5632 字节 MD5 21 字节路由前缀 2 字节1: 1 字节 location 与 line 之间的冒号。超过65536 - 56 - len(location)的行会被静默跳过并计入CEngineLinesTooLong计数器——因为一旦超出remoted 的ReadSecMSG会返回KS_CORRUPT并直接关闭 TCP 连接破坏整个压测会话。场景 Schemaengine 步骤的配置一个 engine 步骤在场景 JSON 中以engine键声明且与kind、dump三者互斥加载器强制只设置其一见 internal/scenario/loader.go 中resolveStep的 XOR 校验。{ engine: sample_payloads/engine/syslog.log, location: syslog, max_eps: 500, loop: true, duration: 5.0, run_while_siblings_active: true }字段说明字段必填默认值说明engine是—输入文本文件路径。先相对于场景文件解析再回退到 benchmark 目录。max_eps是—每流速率上限。engine 流拒绝0无界没有意义。location否engine文件去扩展名的 basename作为帧location字段发送的逻辑位置字符串。loop否true到达 EOF 时true回卷文件继续false结束本次迭代。duration否0不限engine 步骤墙钟运行时间上限秒。0 不限。run_while_siblings_active否false为true时同一代理上所有非 engine 通道一结束engine 源即终止。要求车队中至少有一条非 engine 通道见 Validation。initial_delay与repeat_delay与 inventory_sync 步骤语义一致。repeat_count对 engine 步骤被限制为1——engine 通过loopduration自行控制迭代若再嵌套repeat_count会产生歧义的截止时间。终止语义Termination Semanticsengine 步骤的Run()在下列条件任一先触发时返回whichever-first 组合并在步骤结束时输出一行 info 日志engine step terminated: filebasename locationloc reasonreason events_sentN elapsedTsreason触发条件eof文件读到 EOF 且loopfalse。durationduration截止时间到仅当duration 0。siblings该代理的非 engine 兄弟通道计数降到 0仅当run_while_siblings_active。ctx外层上下文被取消SIGINT /repeat_until/ 编排器取消。两个新终止器被设计为可组合步骤可以同时设置duration和run_while_siblings_active此时duration作为兄弟通道迟迟不结束时的安全上限。当两者都未设置且looptrue时步骤一直运行到ctx被取消即传统行为。兄弟计数是按代理、按迭代的一条通道只要包含至少一个非 engine 步骤即被计为兄弟。runner 在启动任何通道 goroutine 之前预先递增计数每条非 engine 通道退出时通过defer递减选择加入的 engine 源每发送 20 个事件轮询一次该计数廉价原子读。该实现细节与文档描述一致可对照 internal/runner/runner.go 的runIteration与internal/engine/source.go的siblingsPollEvery 20常量。禁用字段按步骤以下字段仅属于 inventory-sync当步骤自身显式设置了engine时加载器会拒绝session_type、sync_mode、data_size、use_databatch、retransmit、payload_size、pad_field、modulecheck_checksum、auto_resync、module、index、option。默认值继承Defaults Inheritance场景级的defaults块在加载前会合并进每个步骤。为了让 engine inventory 混合场景更顺手加载器在把defaults合并进 engine 步骤前会剥离其中仅属于 inventory 的字段。被剥离的集合 上述禁用字段并集 inventory 专属的重试/超时旋钮offline_retry、offline_retry_delay、start_ack_timeout、end_ack_processing_timeout、end_ack_timeout、ack_timeout_retry、ack_timeout_retry_delay、post_data_delay。这样带来的结果场景可以保留一份适用于 inventory 通道的丰富defaults块engine 通道会静默忽略不适用键而 engine 步骤上的局部覆盖仍然优先于 defaults因为它们位于步骤自身 map 中而非被过滤后的 defaults。剥离逻辑在loader.go的filterDefaultsForEngine与inventoryOnlyDefaultKeys中实现且刻意保持清单精简——对两者都适用的旋钮initial_delay、repeat_delay、max_eps不在此列。校验规则Validation除上述字段级规则外加载器还会拒绝engine 步骤上repeat_count 1改用loopduration。duration 0。车队中某 engine 步骤设置了run_while_siblings_activetrue但没有任何包含非 engine 步骤的通道。没有兄弟时计数要么从 0 开始engine 在发出任何事件前就退出要么永不递减两种情况都没有意义。校验实现位于loader.go的validateEngineSiblings会给出包含run_while_siblings_active的明确报错。并发模型engine 事件流遵循一套清晰的并发约定与 docu/07-concurrency-and-pacing.md 描述的整体模型一致每个 agent 的每个engine步骤对应一个 goroutine同一代理上的多个通道可以各自并发运行自己的 engine 源。代理的sendMu串行化 socket 写入——engine 事件与 inventory_sync 写入可以任意顺序交错AES 帧封装层对每次WriteFrame调用是原子的。EPS 限速按流进行每个源实例一个golang.org/x/time/rate.Limiterburst1与 Python 发送端不突发的行为一致见 internal/pacing/limiter.go。SIGINT /ctx.Done()会在每个迭代边界中断读循环、限速器的Wait与写入。每个代理的runIteration拥有一个atomic.Int32兄弟计数包含非 engine 步骤的通道 goroutine 在启动前递增、退出时递减设置了run_while_siblings_active的 engine 源持有该计数指针每 20 个事件读取一次。度量bench.csv 中的 engine 列三列新度量被加入bench.csv通过--report-engine开关控制默认true。它们出现在标准 Python 列集合之后因此 result_summary.py 无需改动即可继续工作。计数器定义可对照 internal/metrics/counters.go 中的CEngineEventsSent、CEngineFilesEOFWrap、CEngineSendErrors。列语义engine_events_sent成功写成一帧的行数同时递增messages_sent。engine_files_eof_wrap设置了looptrue的流在 EOF 处回卷的次数。engine_send_errorsSendText期间发生的写错误socket 失效、封装失败。此外sender_summary.json中的messages.engine_*反映相同的累计总量与之配套的还有keepalives_sent、shutdowns_sent、merged_sum_updates、end_retries、engine_lines_too_long等 Go 侧新增列。命令行入口 cmd/benchmark_sender/main.go 中--report-engine默认开启运行时日志会打印engine_colstrue/false便于核对。验收标准在 11-acceptance-criteria 基础上的新增项文档定义了从 AC-K 到 AC-R 共八条验收标准覆盖冒烟、混合、终止语义、负载与校验五个维度全部可在 scenarios/ 目录中找到对应场景文件AC-K冒烟engine_smoke.json对本地管理器运行约 10 秒 500 EPS 产生engine_events_sent ≥ 4000且管理器日志中针对bench-*代理 ID 无Decoding error。AC-L混合engine_burst_mixed.json用 5 个代理跑 30 秒同时满足sessions_completed 0与engine_events_sent 200000无解码器错误tcpdump 抓包解码后可见 engine 帧与 inventory_sync 帧在同一 TCP 流上交错且无损坏。AC-Mdurationengine_duration_only.json在5 ± 0.5 s内完成engine_events_sent ≈ 2500500 EPS × 5 s且每个代理恰好一条reasonduration的终止日志。AC-Neofengine_eof_short_file.json在 1 s内完成engine_events_sent 50与short.log行数一致engine_files_eof_wrap 0loopfalsereasoneof。AC-Osiblingsengine_while_siblings_inventory.json在 FIM 通道完成时结束约 3–4 秒engine 源记录reasonsiblingsengine_events_sent与 500 EPS × 实际耗时一致。AC-Pwhichever-firstengine_duration_AND_siblings.json以reasonsiblings结束兄弟先于 30 秒duration安全网engine_duration_safety_net.json则在大约 20 秒处以reasonduration结束而 FIM 通道此后继续流式传输整体运行时长 ≈ FIM 时长。AC-Q负载engine_fleet_load.json50 个代理每个代理 1 条 engine 3 条 inventory 通道完成全部 150 个 inventory 会话无sessions_failedengine_events_sent与各代理 inventory 实际耗时 × 200 EPS 成正比且每个 engine 步骤都记录一条终止日志。AC-R校验engine_invalid_all_engine.json在加载期即被拒绝错误信息包含子串run_while_siblings_active同理engine 步骤repeat_count 1的场景被拒绝错误信息包含repeat_count must be 1。源码印证engine 源的核心循环internal/engine/source.go 是 engine 源的规范实现其Run/runOnce逻辑与文档完全对应New在location未设置时用文件 basename 去扩展名兜底并据wazuhMaxInnerEvent - wazuhInnerEventOverhead - len(loc)计算maxLineBytes。Run优先安装duration截止时间context.WithTimeout并在defer中打印终止日志四种terminationReasoneof/duration/siblings/ctx的判定顺序与文档表一一对应。runOnce以 64 KB 缓冲的bufio.Reader逐行读取行尾仅剥离\n保留\r兄弟计数每siblingsPollEvery20行检查一次构造1:location:line后调用conn.SendText发送并递增CEngineEventsSent与CMessagesSent。测试侧internal/engine/engine_test.go 用假 remoted 监听器验证了非循环模式恰好发出 5 帧且内容为1:syslog:line以及循环模式在 EOF 处回卷并推进engine_files_eof_wrap。这些测试与 AC-K/AC-N 互为印证。深入阅读与本文主题强相关的仓库文件场景 Schema 总览docu/03-scenario-schema.md线协议与帧栈docu/05-wire-protocol.md并发与限速模型docu/07-concurrency-and-pacing.md度量与输出docu/08-metrics-and-output.md原有验收标准docu/11-acceptance-criteria.md引擎源实现internal/engine/source.go负载语料sample_payloads/engine/syslog.log千行夹具与short.log五十行夹具场景文件集scenarios/ 下的engine_*.json系列需要说明的是engine event stream 负载的用途是对本地管理器或测试环境产生可控压测流量以验证性能与稳定性实际部署压测前请确认目标管理器已正确配置 authd1515与 remoted1514端口并准备好已知格式的日志语料文件。【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考