企业官网建设流程全解析

Wazuh Engine 源码架构解析：wazuh-manager-analysisd 的事件管线、模块分层与启动依赖注入

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh

Wazuh Manager 的安全事件处理核心wazuh-engine（随发行包以wazuh-manager-analysisd守护进程形式交付）负责把 agent 上报的原始安全日志解码为 Wazuh Common Schema（WCS）、做 GeoIP/IOC/KVDB 富化、按策略（policy）编排处理并转发到 Wazuh Indexer。本文以 src/engine/source/README.md 为骨架，结合 main.cpp 等真实源码，完整梳理source/源码树的模块分层、事件数据流、启动阶段的依赖注入顺序与领域术语体系，帮助开发者在动手阅读或修改引擎代码前建立一张准确的全局地图。

引擎的定位：从 remoted 到 Indexer 的中间处理层

wazuh-engine是 Wazuh manager 的解码、富化与路由引擎：它接收来自remoted及外部生产者的原始安全事件，使用用户定义的decoder（解码器）将其解析为Wazuh Common Schema (WCS)，进行富化（GeoIP、IOC、KVDB 查询），让事件流经一个或多个policy（策略），最终把归一化后的 JSON 文档转发到 Wazuh Indexer（以及可选的文件输出）。

引擎同时支持**独立模式（standalone）**运行：设置环境变量WAZUH_ENGINE_STANDALONE=true即可用于开发、测试，或作为 wazuh-indexer 内的独立内容处理器/校验器。从源码可以确认这一开关的定义位置——process.hpp 中声明了constexpr auto ENV_ENGINE_STANDALONE = "WAZUH_ENGINE_STANDALONE";，而独立运行的启动脚本 run_engine.sh 正是通过export WAZUH_ENGINE_STANDALONE="true"开启该模式。两种运行方式的差异会体现在日志初始化、索引器连接配置来源等启动分支上（见下文“启动与依赖注入”一节）。

面向操作员的文档（配置编写、规则集作者指南、CLI 用法）位于 引擎用户手册；本文只覆盖开发者/源码树视角：每个目录做什么、模块间如何依赖、进程启动时如何组装。

事件管线（Event Pipeline）数据流

原文档给出的管线图完整呈现了事件从入口到出口的走向，这里原样保留并做逐段解读：

+----------------------+ remoted / VD / Others ─►│ httpsrv (events) │ UDS HTTP ingestion +---------┬------------+ │ JSON event ▼ +----------------------+ │ router/Orchestrator │ fan-out per active policy +----------┬-----------+ │ ┌──────────────────────┼──────────────────────┐ ▼ ▼ ▼ Policy A (std) Policy B (custom) Tester session │ │ │ └──────────┬───────────┴──────────┬───────────┘ ▼ ▼ ┌────────────────────────────────────────────────┐ │ bk::IController (Rx or Taskflow) │ │ │ │ pre-filter → decoders → pre-enrichment │ │ → enrichment (geo, ioc, kvdb) │ │ → post-filter → outputs │ └──────────────────────┬─────────────────────────┘ │ ▼ wazuh-indexer / file outputs (streamlog)

关键语义有三点：

• 一条入站事件 ⇒ 每个激活策略一次独立遍历。策略之间互不阻塞；标准空间（standard）与自定义空间（custom）的策略可以并发处理同一条事件。
• 策略内部，解码器按层级组织：根解码器（root decoder）向子解码器分发，解码器再被归组到integration（集成）中。每个解码器恰好属于一个 integration。
• builder模块负责编译：把策略资产编译成表达式树；bk后端再把这棵树物化为可执行管线（RxCpp 可观察图或 Taskflow DAG 二选一）。

这条链路在源码中的锚点可以一一对应：事件经 UDS HTTP 进入后，main.cpp中创建的第二台httpsrv::Server（"Event services"）注册了POST /events/enriched路由，其处理函数api::event::handlers::pushEvent(orchestrator, dumper, agentMetadataCache)直接把事件推入router::Orchestrator（见 main.cpp）；管线末端则统一收敛到wiconnector（Indexer 出口）与streamlog（文件输出）。

source/模块分层地图

source/目录下的模块按角色分组，依赖方向为“下层依赖上层”（箭头指向被依赖方）。各模块若自带 README，深度说明在其自身文档中，此处只做索引。当前仓库中这些目录均已确认存在：agentcache、api、base、bk、builder、cmcrud、cmstore、cmsync、conf、confremote、defs、dumper、fastmetrics、fastqueue、geo、hlp、httpsrv、iockvdb、iocsync、kvdbstore、logicexpr、logpar、parsec、proto、rawevtindexer、router、scheduler、schemf、store、streamlog、wiconnector、yml，外加 main.cpp 与 stackExecutor.hpp。

Foundation（基础层）

几乎被所有其他模块使用：

Storage（存储层）

• store/ — 可插拔驱动的 JSON 文档存储（内置FileDriver），即引擎的持久化 KV。
• cmstore/ — 内容仓库：持有 decoder、filter、output、integration、KVDB 与 policy 的命名空间，并维护双向 UUID↔名称缓存。
• kvdbstore/ — 从cmstore物化的内存 KVDB 缓存，供 decoder/filter 查询；无 handler 持有即过期。
• iockvdb/ — 基于 RocksDB 的 IOC 数据库，支持整体数据库实例的 RCU 风格原子热切换。

Compilation & execution backend（编译与执行后端）

• builder/ — 编译中枢：从cmstore读取资产，产出可执行的IPolicy表达式树；通过BuilderDeps结构体拉入logpar、schemf、kvdbstore、iockvdb、geo、streamlog与wiconnector。无 README，公共接口在 builder/include/builder/。
• bk/ — 两种可互换的执行后端（RxCpp 可观察图或 Taskflow DAG），支持节点追踪与热加载。

Enrichment & I/O services（富化与 I/O 服务）

• geo/ — MaxMind GeoIP/ASN 查询，基于哈希的数据库热加载。
• streamlog/ — 异步滚动日志通道（按大小+时间、gzip、保留策略），供文件输出与dumper使用。
• dumper/ — 可开关的原始事件转储器，激活时经streamlog落盘。
• scheduler/ — 优先级线程池任务调度器，负责周期性同步与指标刷新。
• wiconnector/ — Wazuh Indexer 的线程安全客户端：事件、策略资源、IOC 与远端配置。这是通往 indexer 的唯一出口。

Synchronization & remote configuration（同步与远端配置）

• confremote/ — 从 indexer 拉取远端运行时配置，被拒绝时回滚。
• cmcrud/ — API 与cmstore变更之间的校验/适配层，强制规范化的变更顺序与命名空间导入的原子性。
• cmsync/ — 周期性从 indexer 同步内容；策略变化时热替换 router 路由。
• iocsync/ — 周期性把 IOC 同步进iockvdb，原子热切换。
• rawevtindexer/ — 可开关的原始（预处理前）事件取证式索引。

Routing & runtime（路由与运行时）

• router/ — 生产用Router（worker 池）+ 同步式Tester+Orchestrator门面。持有事件队列、策略Environment与路由热替换。

API gateway（API 网关）

• httpsrv/ — 基于 cpp-httplib 的 UDS HTTP 服务器。main中创建两个实例：管理 API 与可选的远端事件接收器。
• api/ — 按域划分的 handler 工厂：router、tester、cmcrud、geo、ioccrud、dumper、rawevtindexer、metrics、event。负责 JSON↔protobuf 互转并委托给对应域接口。

Entry point（入口）

• main.cpp — 进程入口：信号/守护进程处理、依赖注入装配、用于 LIFO 关闭的StackExecutor。
• stackExecutor.hpp — 按构造顺序记录关闭回调，执行时倒序（LIFO）运行。

高层模块依赖图

原文档的依赖图省略了base、conf、proto等基础库（它们到处被使用），突出运行时中枢：

┌──────────────┐ │ api │ (handlers per domain) └──────┬───────┘ │ ┌──────▼───────┐ │ httpsrv │ └──────────────┘ ┌──────────────┐ ┌─────────────────┐ ┌──────────────┐ │ cmsync │───►│ router │◄───│ fastqueue │ └──────┬───────┘ │ (Orchestrator) │ └──────────────┘ │ └────────┬────────┘ │ │ ▼ ▼ ┌──────────────┐ ┌──────────────┐ │ cmcrud │ │ builder │ ─── compilation hub └──────┬───────┘ └──┬───┬───┬───┘ │ │ │ │ ▼ │ │ └────────────► geo, streamlog ┌──────────────┐ │ │ │ cmstore │◄───────┘ └────► logpar ──► hlp ──► parsec └──────┬───────┘ schemf │ kvdbstore ▼ iockvdb ◄── iocsync ┌────────┐ │ store │◄── confremote, rawevtindexer └────────┘ ┌────────────────┐ │ wiconnector │ ──► wazuh-indexer (sole egress) └────────────────┘ ▲ cmsync, iocsync, confremote, rawevtindexer, streamlog, builder

四个关键关系值得牢记：

• router是运行时中枢：拥有事件队列、worker 线程与路由生命周期；cmsync负责热替换其路由。
• builder是编译中枢：所有参与事件处理的依赖都经BuilderDeps汇聚于此。
• store与cmstore是数据中枢：持久化状态（schema、允许字段、规则集、同步状态）经由它们流动。
• wiconnector是通往 indexer 的唯一出口：一切出站 OpenSearch 流量都经过它。

启动与依赖注入：main.cpp 的八个装配阶段

引擎模块在 main.cpp 中通过std::shared_ptr与StackExecutor装配，后者记录拆除回调并按 LIFO 执行关机。构造分阶段进行，每个阶段只依赖它之上的阶段；部分阶段受conf::key::SERVER_ENABLE_EVENT_PROCESSING门控。逐段对照源码验证如下：

1. 进程引导。解析命令行选项（main.cpp 支持-f前台运行、-t测试配置、-d调试级别可重复、-h帮助）；随后按模式初始化日志：standalone 走logging::getStandaloneLoggingConfig()（支持按日/按大小滚动），manager 模式则经base::libwazuhshared::init()复用 wazuh-shared 日志并chdir到 Wazuh home；非 standalone 模式下若未加-f则goDaemon()守护化。信号处理上，SIGINT/SIGTERM仅置位g_shutdown_requested（main.cpp），SIGPIPE直接忽略（SIG_IGN，main.cpp）。

2. 配置加载。conf::Conf从etc/wazuh-manager-internal-options.conf加载（main.cpp），之后所有模块经confManager.get<T>(key::…)读取。

3. 核心数据层。store::Store（FileDriver）→cmstore::CMStore→kvdbstore::KVDBManager→iockvdb::KVDBManager(store)→geo::Manager(store, downloader)→fastmetrics::registerManager()→schemf::Schema。其中 schema 从 store 读取schema/engine-schema/0；读取失败只告警而不终止——引擎会以“无 schema”降级运行，日志提示与 indexer mapping 的一致性不再受保证（main.cpp）。

4. 解析层。hlp::initTZDB(...)初始化时区数据库后构建logpar::Logpar——它需要 store 中的schema/wazuh-logpar-overrides/0文档（该文档读取失败会直接抛异常终止启动，与 schema 的降级策略不同，main.cpp）——最后hlp::registerParsers(logpar)。

5. 调度与 I/O。scheduler::Scheduler始终创建，并且第一个注册进退出栈（注释明确说明：它必须在所有模块之前终止，以确保已调度的任务在关机前停止，main.cpp）。此后读取enableProcessing开关；开启时创建wiconnector::WIndexerConnector并注册队列指标拉取回调（INDEXER_QUEUE_SIZE、INDEXER_EVENTS_DROPPED、INDEXER_QUEUE_USAGE_PERCENT），再创建streamlog::LogManager(store, scheduler)。

6. 编译中枢。组装builder::BuilderDeps（携带logpar、kvdbManager、IOCkvdb、geoManager、streamLogger、indexerConnector以及文件输出的streamlog::RotationConfig——基础路径、命名模式、最大大小、缓冲、是否压缩、压缩级别、最大文件数、累积上限，均取自conf::key::STREAMLOG_*系列键），随后构建builder::Builder(cmStore, schemaValidator, defs, allowedFields, builderDeps, store)与cmcrud::CrudService(cmStore, builder)（main.cpp）。allowedFields同样从 store 的schema/allowed-fields/0加载，缺失时仅告警并退化为“不限制字段”。

7. 后台服务（受enableProcessing门控）。依次创建confremote::ConfRemoteManager、rawevtindexer::RawEventIndexer（可经confremote的index_raw_events触发器热重载开关）、router::Orchestrator（立即启动并注册关闭回调）、cmsync::CMSync、iocsync::IocSync（经 scheduler 按IOC_SYNC_INTERVAL周期调度，interval 为 0 时禁用）、Geo 同步任务（GEO_SYNC_INTERVAL，从 manifest 拉取 GeoLite2-City/ASN 数据库）与dumper::Dumper(streamLogger)。

8. API 面。创建httpsrv::Server（"API services"，payload 上限由SERVER_API_PAYLOAD_MAX_BYTES控制且对负值有防回绕校验），按域注册 handler：metrics、geo、router、tester、dumper、rawevtindexer、cmcrud、ioccrud、status，最后apiServer->start(SERVER_API_SOCKET)。若enableProcessing开启，还会创建第二台httpsrv::Server（"Event services"）作为远端事件接收器，监听SERVER_ENRICHED_EVENTS_SOCKET（main.cpp）。

主循环与关机的一个细节：进入运行态后，首次内容同步任务cm-sync-task在synchronize()之前调用orchestrator->expandWorkerPool()——目的是让首次同步修改完整的 worker 池而不仅是主 worker（main.cpp）。此外还有“无可用路由”的状态监控：内容同步完成前入站事件会被丢弃，首次启动时该情况记 INFO、后续记 WARNING。

关机是严格逆序：StackExecutor按 LIFO 执行回调——API 服务器先停（并 join 客户端连接）、后台服务请求 shutdown 并 join、orchestrator 排空队列、streamlog与wiconnector冲刷缓冲、scheduler 停止，日志最后拆除。StackExecutor的实现本身极简：std::deque存回调，execute()从栈顶弹出执行，单个回调抛异常不会中断后续回调（stackExecutor.hpp）。一个值得注意的注册顺序细节：wiconnector先注册shutdown()再注册requestShutdown()，借助 LIFO 保证破坏性关闭先于协作式关闭执行，确保进行中的分页循环中止并释放共享锁（main.cpp）。

从源码可验证的关键参数约束

启动过程中对 indexer 连接器参数有硬性范围校验，超出即抛异常终止（main.cpp），这些是调参时的硬边界：

事件队列侧，Orchestrator的入站队列是带字节上限的fastqueue::CQueue<router::IngestEvent>（EVENT_QUEUE_SIZE/EVENT_QUEUE_EPS/EVENT_QUEUE_MAX_BYTES，默认队列长度 131072，见 conf.cpp）；测试事件走独立的StdQueue。内容同步周期CM_SYNC_INTERVAL默认 120 秒（conf.cpp）。

领域术语表

引擎使用一套小型但高度专有的词汇，贯穿代码、API 与用户手册，阅读源码前务必对齐：

• Event（事件）— 代表一条安全日志行的 JSON 文档，携带 agent/cluster 元数据，是流经引擎的工作单元。
• Wazuh Common Schema (WCS)— 所有输出事件必须遵循的权威类型化字段 schema。由 indexer 拥有；引擎启动时从 store 获取schema/engine-schema/0。
• Asset（资产）— 最小内容单元（decoder、filter、output、integration、KVDB、schema），以<type>/<name>/<version>寻址，例如decoder/aws-cloudtrail/0。
• Decoder（解码器）— 解析并归一化事件到 WCS 字段的资产。层级组织（root → children），归组到 integration 中。
• Integration（集成）— 属于同一产品或日志源的解码器 + KVDB 的有序组。每个解码器恰好属于一个 integration。
• Filter（过滤器）— 对事件的布尔谓词。pre-filter 在解码前丢弃事件；post-filter 在事件到达输出前丢弃事件。
• Output（输出）— 处理完事件的目的地（Wazuh Indexer、文件）。随 manager 打包分发，不从内容源同步。
• Policy（策略）— 命名的处理管线：pre-filter → decoders → pre-enrichment → enrichment → post-filter → outputs。多个策略并发运行。
• Namespace / Space（命名空间/空间）— 逻辑内容分区。出厂两个空间：standard（Wazuh 维护）与custom（用户）。indexer 是事实来源，引擎在本地镜像它。
• KVDB— 处理期间供 decoder/filter 查询的轻量键值存储。普通 KVDB 按空间隔离；IOC 与 Geo 数据库是全局的。
• Helper（辅助函数）— 可从 decoder/filter 阶段调用的可复用函数：条件类（用于check）与映射类（用于map）。
• Stage（阶段）— decoder 内部的操作块：check（布尔）、parse|<field>（抽取）、normalize（含嵌套map）。
• Route / Environment（路由/环境）— orchestrator 持有的已编译策略的运行时实例。路由可热替换，无需重启。

代码约定与继续深入的路径

• 全引擎使用C++17；clang-format与clang-tidy配置在engine/下。
• 每个模块在include/<module>/暴露I<Module>接口，实现在src/，GoogleTest 单元测试在test/src/unit/，供其他模块测试使用的 mock 在test/mocks/。
• 依赖一律以std::shared_ptr注入，且只在 main.cpp 中装配——模块自身从不实例化依赖。
• API handler 遵循工厂模式（api::xxx::handlers::registerHandlers(...)的统一签名在 main.cpp 中一目了然）。

继续深入的入口：

• 引擎用户手册 — 面向操作员的文档与快速上手，附数据流 mermaid 图；同目录还有 架构、配置、API 参考 等。
• router/README.md — 运行时编排、路由生命周期与 tester 语义。
• builder/include/builder/builder.hpp — 策略编译的公共接口（builder 无 README，从这里入手）。
• bk/README.md — Rx 与 Taskflow 两种执行后端。
• CMakeLists.txt —add_executable(wazuh-engine .../main.cpp)与 RPATH 配置说明了 manager 安装（$ORIGIN/../lib）与独立包（$ORIGIN/lib）两种运行布局。

小结

wazuh-engine的源码树组织体现了清晰的“编译—执行—同步”三分法：builder把内容资产编译为表达式树，bk把树物化为可执行管线，router持有运行时路由并可被cmsync热替换，而wiconnector把一切出站流量收敛为单一出口。所有装配集中在main.cpp的八个阶段中完成，StackExecutor保证 LIFO 安全关机；WAZUH_ENGINE_STANDALONE=true则让同一份代码能以独立进程形态服务于开发、测试与 indexer 内的内容校验场景。掌握这份地图后，无论是排查事件丢包、调整队列与批量参数，还是新增一个富化维度，都可以按图索骥地定位到具体模块。

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh