Wazuh Engine 的 CMStore事件处理内容仓库的命名空间、UUID 缓存与并发模型设计【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuhCMStore 是 Wazuh Engine 的事件处理内容仓库负责集中管理解码器decoders、过滤器filters、输出outputs、集成integrations、键值库KVDBs与策略policy这些定义事件处理管线的全部资源。本文基于 CMStore 模块文档 及其源码实现展开帮助读者理解 CMStore 的三层架构ICMStore→CMStoreNS→CacheNS、UUID 双向缓存机制、命名空间隔离策略与读写并发模型掌握后你能够在 Engine 的 builder、router、kvdbstore、cmcrud 等消费方中正确读写内容资产并定位内容同步与缓存一致性问题。一、CMStore 在 Engine 中的定位CMStore 是 Wazuh Engine 的单一事实来源single source of truthbuilder 从它读取内容并编译出事件处理管线backend 最终执行该管线。除了管线构建CMStore 还对外提供 CRUD 层被三类外部使用者依赖消费方依赖目标用途buildercmstore::icmstore读取 policy、integrations、decoders、KVDBs 和 outputs编译出事件处理管线routercmstore::icmstore使用ICMStoreNSReader与类型定义进行环境构建和路由配置kvdbstorecmstore::icmstore从 CMStore 读取 KVDB 定义填充内存中的只读 KVDB 层cmcrudcmstore::icmstore将 CMStore 的 CRUD 操作通过管理 API 暴露出去api/testercmstore::icmstore测试 API 使用 CMStore reader 校验与测试管线配置从源码结构看这套职责对应 CMakeLists.txt 中的目标划分cmstore_icmstoreINTERFACE公共接口与数据类型、cmstore_cmstoreSTATIC具体实现链接 yml、base、cmstore_mocksGMock 模拟、cmstore_utest与cmstore_ctest单元/组件测试。接口层与实现层分离意味着消费方只需依赖接口头文件即可做 mock 测试这也是该模块可测试性的基础。二、总体架构三层结构与命名空间隔离CMStore 的架构可以概括为顶层命名空间管理器 每命名空间存储 每命名空间缓存三层┌─────────────────────────────────────────────┐ │ ICMStore │ │ (namespace management) │ │ createNamespace / deleteNamespace / getNS │ └────────────────┬────────────────────────────┘ │ ┌──────────────────────┼──────────────────────┐ ▼ ▼ ▼ ┌────────────────┐ ┌────────────────┐ ┌────────────────┐ │ CMStoreNS │ │ CMStoreNS │ │ CMStoreNS │ │ ns: wazuh │ │ ns: custom │ │ ns: ... │ │ ┌──────────┐ │ │ ┌──────────┐ │ │ │ │ │ CacheNS │ │ │ │ CacheNS │ │ │ │ │ │ UUID↔Name│ │ │ │ UUID↔Name│ │ │ │ │ └──────────┘ │ │ └──────────┘ │ │ │ │ Filesystem: │ │ Filesystem: │ │ decoders/ │ │ decoders/ │ │ decoders/ │ │ filters/ │ │ filters/ │ │ filters/ │ │ outputs/ │ │ outputs/ │ │ outputs/ │ │ integrations/ │ │ integrations/ │ │ integrations/ │ │ kvdbs/ │ │ kvdbs/ │ │ kvdbs/ │ │ policy.json │ │ policy.json │ │ policy.json │ │ cache_ns.json │ │ cache_ns.json │ │ cache_ns.json │ │ │ └────────────────┘ └────────────────┘ └────────────────┘ Consumers: ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ builder │ │ router │ │ kvdbstore│ │ cmcrud │ └──────────┘ └──────────┘ └──────────┘ └──────────┘命名空间Namespace是磁盘上彼此隔离的内容分区每个命名空间拥有独立目录内部按资源类型划分子目录另加一个 policy 文件和一个 UUID 缓存文件。命名空间 ID 由 types.hpp 中的NamespaceId类型约束构造时立即校验合法性isValidName非空、且仅允许字母数字与下划线std::isalnum(c) || c _非法名称直接抛出std::runtime_error禁用的命名空间名称有三个output、system、default源码中定义为FORBIDDEN_NAMESPACES向量见 cmstore.cpp 第 18-19 行在创建、删除、重命名时均会被拒绝std::hashNamespaceId特化提供哈希支持使命名空间 map 可做到 O(1) 查找。顶层CMStore内部持有unordered_mapNamespaceId, shared_ptrICMstoreNS与一个shared_mutex见 cmstore.hpp并在注释中明确要求只应存在一个 CMStore 实例以避免对命名空间的竞态条件。三、并发模型两级 shared_mutexCMStore 采用两级读写锁读多写少场景下互不阻塞作用域锁粒度命名空间 mapCMStoreshared_mutex读shared / 写unique每命名空间的文件 缓存CMStoreNSshared_mutex读shared / 写unique从 cmstore.cpp 的实现可以确认这一模型getNS、getNSReader、existsNamespace、getNamespaces均取std::shared_lock而createNamespace、deleteNamespace、renameNamespace均取std::unique_lock。每命名空间的CMStoreNS同样持有mutable std::shared_mutex m_mutex保护文件与缓存访问见 storens.hpp 第 46 行所有资源 CRUD 操作的流程统一为校验 → 加锁 → 更新缓存 → 写文件 → 落盘缓存。此外命名空间删除还有一个引用保护机制deleteNamespace()会检查use_count() 1map 本身持有一个引用若检测到仍有活动的shared_ptr读者/写者则中止删除renameNamespace()采用相同策略且重命名时若文件系统 rename 失败会把 map 中的旧条目回滚保证内存与磁盘状态一致。四、资源类型与资源分类资源类型枚举定义在 types.hpp 中实际源码包含 6 个值比 README 多出一个UNDEFINED 0兜底值enum class ResourceType : uint8_t { UNDEFINED 0, DECODER 1, OUTPUT 2, FILTER 3, INTEGRATION 4, KVDB 5 };配套提供resourceTypeFromString()/resourceTypeToString()在decoder、output、filter、integration、kvdb字符串与枚举之间做 constexpr 转换。各资源类型的语义如下Decoders解码器— 解析原始日志事件可组成父子层级parent-child hierarchiesFilters过滤器— 应用于事件的预处理pre-filter/后处理post-filter过滤器Outputs输出— 定义处理后的事件发往何处indexer、alerts 等Integrations集成— 在一个类别category下组织 decoders 和 KVDBs 的 UUID 清单manifestKVDBs键值库— 用于事件富化的键值查找表内容以 JSON 对象形式存储。需要说明的是getResourceTypeFromAssetName()将DECODER、OUTPUT、FILTER视为资产asset资产名采用type/name/version三段式结构base::Name类型parts().size()必须等于 3首段必须匹配类型前缀见 detail.hpp 中的checkAssetName()。INTEGRATION 与 KVDB 不属于 asset走独立的读取接口。集成类别CategoriesIntegration 必须归属于 8 个预定义类别之一定义于 categories.hppaccess-management、applications、cloud-services、network-activity、other、security、system-activity、unclassifiedIntegration构造时会对 category 做存在性校验exists()非法类别直接抛错。五、UUID 系统资源的规范化标识每个资源都有一个UUIDv4标识符。当资源通过 YAML/JSON 内容创建时CMStore 会走upsertUUID()流程见 storens.hpp 第 78 行声明将内容解析为 JSON若存在/id字段pathns::JSON_ID_PATH校验其是否为合法 UUIDv4合法则原样使用若不存在生成新 UUIDv4 并注入回内容返回最终 UUID。UUID 是贯穿 policy、integrations 与所有交叉引用的规范化标识。Policy构造函数会通过cm::store::detail::findDuplicateOrInvalidUUID()detail.hpp 第 58-85 行对 integrations、outputs、filters 三个 UUID 数组统一校验每个 UUID 必须是合法 UUIDv4且大小写不敏感地检查重复违规即抛异常。六、双向缓存 CacheNS 与 cache_ns.json每个命名空间维护一个内存中的双向缓存cachens.hpp提供两类 O(1) 查找UUID → (Name, ResourceType)m_uuidToEntryMapunordered_mapstring, EntryData(Name, ResourceType) → UUIDm_nameTypeToUUIDMap键为std::tuplestring, ResourceType配自定义NameTypeHash。关键设计点非线程安全CacheNS类注释明确标注 This class is not thread-safe. External synchronization is required其并发正确性完全由外层CMStoreNS的shared_mutex保证序列化缓存序列化为 JSON 数组落盘文件为cache_ns.json每次写操作flush都会刷新createNamespace()时初始化为空数组[]见 cmstore.cpp 第 144-151 行加载与重建CMStoreNS构造时调用loadCacheFromDisk()若缓存文件损坏则降级为rebuildCacheFromStorage()——扫描该命名空间下所有资源目录从每个文件中重新提取 UUID 与名称重建双向映射重建也失败才抛异常条目操作addEntry()重复的 UUID 或 (name, type) 抛错、removeEntryByUUID()、removeEntryByNameType()、getCollection(type)按类型导出 (UUID, Name) 元组列表等。这套磁盘为准 缓存加速 损坏可重建的设计使得即使cache_ns.json丢失或损坏命名空间仍可自恢复代价只是一次全目录扫描。七、Policy事件处理管线的完整定义Policydatapolicy.hpp定义命名空间内完整的事件处理管线其期望的 JSON 格式摘自源码注释如下{ type: policy, metadata: { title: Development 0.0.1 }, enabled: true, root_decoder: 5c1df6b6-1458-4b2e-9001-96f67a8b12c8, origin_space: space1, index_unclassified_events: true, index_discarded_events: true, cleanup_decoder_variables: true, filters: [], enrichments: [file, domain-name, ip, url, geo], integrations: [ 42e28392-4f5e-473d-89e8-c9030e6fedc2, a7fe64a2-0a03-414f-8692-8441bdfe6f69, 5c1df6b6-1458-4b2e-9001-96f67a8b12c8, f61133f5-90b9-49ed-b1d5-0b88cb04355e, 369c3128-9715-4a30-9ff9-22fcac87688b ], outputs: [], hash: 7ab287...5180, id: eb5c2519-feff-4789-8542-9a0453cc8690 }各字段语义与默认值结合fromJson()解析逻辑字段必选说明 / 默认值metadata/title否策略标题缺失或为空时取Untitled Policyenabled是布尔缺失则抛错root_decoder是可为 null入口解码器 UUIDnull 视为空字符串integrations是集成 UUID 有序数组缺失则抛错filters是过滤链 UUID 数组每个 filter 带 pre-filter/post-filter 类型缺失则抛错enrichments是富化插件数组file、ip、domain-name、url、geo缺失则抛错outputs否输出 UUID 数组可省略origin_space否用于输出解析的空间键默认UNDEFINED仅允许字母数字与下划线正则^[a-zA-Z0-9_]$非法字符抛错index_unclassified_events是是否索引未分类事件缺失则抛错index_discarded_events是是否索引被丢弃事件缺失则抛错cleanup_decoder_variables否是否清理解码器临时变量默认 true源码中该字段缺失时返回 truehash否完整性校验哈希可省略id—策略自身的 UUID策略文件持久化为命名空间目录下的policy.jsonpathns::POLICY_FILE通过upsertPolicy()/deletePolicy()维护getPolicy()在策略不存在或读取失败时抛std::runtime_error。Integration 与 KVDB 数据结构DataIntegration 的期望 JSON 格式摘自源码注释{ id: 5c1df6b6-1458-4b2e-9001-96f67a8b12c8, title: windows, enabled: true, category: security, default_parent: 85853f26-5779-469b-86c4-c47ee7d400b4, decoders: [ 85853f26-5779-469b-86c4-c47ee7d400b4, 4aa06596-5ba9-488c-8354-2475705e1257, 4da71af3-fff5-4b67-90d6-51db9e15bc47, 6f8bd7d2-8516-4b2b-a6f1-cc924513c404 ], kvdbs: [] }其中default_parent为可选字段name取自/metadata/title路径。KVDBdatakvdb.hpp则包含 uuid、name、contentJSON 对象与 enabled 字段。三种数据类型的字段汇总类型命名空间关键字段Policycm::store::dataTypetitle、enabled、root_decoder、integrations[]、filters[]、enrichments[]、outputs[]、origin_space、hash、index_unclassified_events、index_discarded_events、cleanup_decoder_variablesIntegrationcm::store::dataTypeuuid、name、enabled、category、default_parent?、decoders[]、kvdbs[]KVDBcm::store::dataTypeuuid、name、contentJSON 对象、enabled八、资产适配adaptDecoder / adaptFilterdetail.hpp 还提供adaptDecoder()与adaptFilter()两个函数在资源交给 builder 消费之前将 YAML/JSON 文档归一化为规范的键顺序。以解码器为例规范化顺序为name → parents → definitions → check → parse|* → normalize → enabled → id源码中adaptDecoder()逐字段处理先提取并校验/name必须是decoder/...三段式合法名称再依次透传parents等字段。适配层保证了无论上游文档字段顺序如何builder 看到的都是同一规范结构——这是内容仓库与管线编译器解耦的关键一环。九、公共接口ICMStore / ICMStoreNSReader / ICMstoreNS三个接口全部定义在 icmstore.hpp按只读 → 读写两层切分消费方可按最小权限原则选择9.1 ICMStore顶层命名空间管理namespace cm::store { class ICMStore { virtual std::shared_ptrICMStoreNSReader getNSReader(const NamespaceId nsId) const 0; virtual std::shared_ptrICMstoreNS getNS(const NamespaceId nsId) 0; virtual std::shared_ptrICMstoreNS createNamespace(const NamespaceId nsId) 0; virtual void deleteNamespace(const NamespaceId nsId) 0; virtual void renameNamespace(const NamespaceId from, const NamespaceId to) 0; virtual bool existsNamespace(const NamespaceId nsId) const 0; virtual std::vectorNamespaceId getNamespaces() const 0; }; }9.2 ICMStoreNSReader命名空间只读视图builder 与 router 通过它读取管线定义。核心方法分五组class ICMStoreNSReader { // 通用 virtual const NamespaceId getNamespaceId() const 0; virtual std::vectorstd::tuplestd::string, std::string getCollection(ResourceType type) const 0; virtual std::tuplestd::string, ResourceType resolveNameFromUUID(const std::string uuid) const 0; virtual std::string resolveUUIDFromName(const std::string name, ResourceType type) const 0; virtual bool assetExistsByName(const base::Name name) const 0; virtual bool assetExistsByUUID(const std::string uuid) const 0; // Policy virtual dataType::Policy getPolicy() const 0; // Integrations virtual dataType::Integration getIntegrationByName(const std::string name) const 0; virtual dataType::Integration getIntegrationByUUID(const std::string uuid) const 0; // KVDBs virtual dataType::KVDB getKVDBByName(const std::string name) const 0; virtual dataType::KVDB getKVDBByUUID(const std::string uuid) const 0; // Assets (decoders, filters, outputs) virtual json::Json getAssetByName(const base::Name name) const 0; virtual json::Json getAssetByUUID(const std::string uuid) const 0; virtual const std::vectorjson::Json getOutputsForSpace(std::string_view spaceKey) const 0; // 模板辅助 templatetypename T auto getResourceByName(const std::string name) const; templatetypename T auto getResourceByUUID(const std::string uuid) const; };两个模板辅助函数用if constexpr在编译期把返回类型分派到Integration/KVDB/json::Jsonasset对不支持的 T 触发static_assert让调用方一行代码即可按类型取资源。getOutputsForSpace()的解析规则在头文件注释中写明若 outputs 路径下存在以给定 space key 命名的目录则从该目录加载 outputs否则回退到default/目录。9.3 ICMstoreNS读写扩展class ICMstoreNS : public ICMStoreNSReader { virtual std::string createResource(const std::string name, ResourceType type, const json::Json content) 0; virtual void updateResourceByName(const std::string name, ResourceType type, const json::Json content) 0; virtual void updateResourceByUUID(const std::string uuid, const json::Json content) 0; virtual void deleteResourceByName(const std::string name, ResourceType type) 0; virtual void deleteResourceByUUID(const std::string uuid) 0; virtual void upsertPolicy(const dataType::Policy policy) 0; virtual void deletePolicy() 0; };需要注意一个实现细节README 文档中写的是const std::string ymlContent而当前源码签名实际为const json::Json content——即 CRUD 层接收的是已解析的 JSON 对象而非原始 YAML 字符串YAML→JSON 的解析发生在调用方如 cmcrud 的 API 层。接口语法还特别提醒createResource/updateResource*不校验内容是否匹配资源类型的 schemaschema 校验由上层负责。十、实现细节磁盘布局、路径映射与权限具体实现分布在 src/cmstore.cpp、src/storens.cpp、src/cachens.cpp 与文件工具 src/fileutils.hpp。要点如下CMStore 构造函数CMStore(path, outputsPath)两个路径都必须绝对路径且已存在的目录否则抛std::runtime_error写权限探测在 basePath 下尝试创建临时文件.wazuh_test_write_permission与临时目录.wazuh_test_dir_permission成功则删除并继续失败则带errno/error_code信息抛错源码刻意avoiding check mode_t用实际写测试代替权限位检查调用loadAllNamespacesFromDisk()遍历 basePath 下所有子目录以目录名构造NamespaceId自动触发命名合法性校验跳过禁用命名空间为每个命名空间构造CMStoreNS实例装入 map。CMStoreNS 磁盘布局与 CRUD命名空间目录下按资源类型划分子目录常量集中定义在 storens.hpp 第 17-32 行的pathns命名空间decoders/、filters/、outputs/、integrations/、kvdbs/以及文件常量policy.json策略、cache_ns.json缓存、.json资产扩展名路径安全资源名中的/在落盘文件名中被替换为_资源类型决定所在子目录getResourcePaths()存储格式资产统一存为.json策略存为policy.json文件权限文件 0640、目录 0750由fileutils::setDirectoryPermissions()等保证输出解析getOutputsForSpace()优先使用 space key 专属目录缺失时回退default/对应pathns::DEFAULT_OUTPUTS_DIR。缓存一致性任何写操作create/update/delete resource、upsertPolicy 等都遵循更新内存缓存 → 写资源文件 → flush 缓存到磁盘的顺序配合两级 shared_mutex保证读侧看到的缓存与磁盘最终一致。十一、目录结构与构建目标模块的完整目录布局以当前仓库为准src/engine/source/cmstore/ ├── CMakeLists.txt ├── interface/cmstore/ # 公共接口 │ ├── icmstore.hpp # ICMStore, ICMstoreNS, ICMStoreNSReader │ ├── types.hpp # ResourceType 枚举, NamespaceId, 数据类型导入 │ ├── categories.hpp # AVAILABLE_CATEGORIES, exists() │ ├── detail.hpp # adaptDecoder(), adaptFilter(), UUID 校验 │ ├── datapolicy.hpp # dataType::Policy — 管线定义 │ ├── dataintegration.hpp # dataType::Integration — 集成清单 │ └── datakvdb.hpp # dataType::KVDB — 键值库定义 ├── include/cmstore/ │ └── cmstore.hpp # CMStore — 顶层具体实现 ├── src/ │ ├── cmstore.cpp # CMStore命名空间生命周期、磁盘加载 │ ├── storens.hpp # CMStoreNS — 每命名空间实现 │ ├── storens.cpp # CRUD、policy、integration、KVDB、assets │ ├── cachens.hpp # CacheNS — UUID↔Name 双向缓存 │ ├── cachens.cpp # 缓存序列化、增删查 │ └── fileutils.hpp # 文件 I/O 辅助upsert、read、delete、权限 ├── test/ │ ├── mocks/cmstore/ │ │ └── mockcmstore.hpp # GMock: MockICMStoreNSReader, MockICMstoreNS, MockICMstore │ └── src/ │ ├── unit/ │ │ ├── cachens_test.cpp # CacheNS 单元测试 │ │ ├── cmstore_test.cpp # CMStore/CMStoreNS 单元测试 │ │ └── detail_test.cpp # detail 辅助函数测试 │ └── component/ │ └── cmstore_test.cpp # 基于真实文件系统的组件测试 └── benchmark/src/ └── cmsync_bench.cpp # 同步基准测试CMake 目标一览与 CMakeLists.txt 一致目标类型别名说明cmstore_icmstoreINTERFACEcmstore::icmstore公共接口 数据类型链接 basecmstore_cmstoreSTATICcmstore::cmstore具体实现链接 yml、basecmstore_mocksINTERFACEcmstore::mocks所有接口的 GMock 模拟链接 GTest::gmockcmstore_utestExecutable—单元测试cache、store、detailcmstore_ctestExecutable—组件测试真实文件系统十二、测试策略模块的测试分三层单元测试test/src/unit/cachens_test.cpp覆盖 CacheNS 的双向查找、序列化/反序列化、冲突检测cmstore_test.cpp覆盖 CMStore 命名空间操作与 CMStoreNS 行为detail_test.cpp覆盖adaptDecoder/adaptFilter/UUID 校验等辅助函数组件测试test/src/component/cmstore_test.cpp使用真实文件系统执行完整的资源 CRUD 周期验证文件 缓存 策略三者的端到端一致性Mock 层test/mocks/cmstore/mockcmstore.hpp为ICMStore、ICMstoreNS、ICMStoreNSReader三个接口提供 GMock 派生类供 builder、cmcrud 等下游模块的测试注入。此外benchmark/src/cmsync_bench.cpp提供内容同步路径的基准测试用于度量命名空间同步在大内容量下的开销。小结CMStore 用顶层 map 每命名空间目录 每命名空间双向缓存的最小化设计支撑了 Wazuh Engine 内容管理的全部需求命名空间提供内容隔离与独立生命周期UUIDv4 双向缓存让跨资源交叉引用policy ↔ integrations ↔ decoders/KVDBs保持 O(1) 解析两级shared_mutex让 builder/router 的高频读与 cmcrud 的低频写互不阻塞而cache_ns.json的可重建性则为磁盘损坏场景提供了自恢复能力。对需要理解 Engine 管线如何从内容编译为执行的读者CMStore 是必须先看的一块基石继续深入时建议从 icmstore.hpp 的三个接口读起再对照 cmstore.cpp 与 storens.cpp 的实现最后用 组件测试 验证你对 CRUD 周期的理解。【免费下载链接】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），仅供参考