在 Logstash 中使用 Winlogbeat 模块：加载并接入 Ingest Pipeline 解析 Windows 日志-开发者社区

数据工程
后端

【免费下载链接】logstash

Logstash - transport and process your logs, events, or other data

项目地址：https://gitcode.com/gh_mirrors/lo/logstash

Logstash 与 Winlogbeat 模块的集成并不需要复杂的 Logstash 过滤器重写：Winlogbeat 自带的每个模块都内置了完整的解析逻辑（ingest node pipelines、Elasticsearch 模板、输入配置与 Kibana 仪表盘），你只需要把 pipeline 加载进 Elasticsearch，再让 Logstash 的 elasticsearch 输出按事件元数据%{[@metadata][pipeline]}动态选择对应 pipeline 即可。读完本文，你将掌握从winlogbeat setup --pipelines到 Logstash 管线配置的完整接入流程，并能处理数据流（data streams）与 Kafka 中间链路等实际场景。

Winlogbeat 模块与 fileset 结构

Winlogbeat 随发行版预置了一批模块（modules），例如 security、sysmon、powershell 等，它们封装了从各类 Windows 日志提供方采集、解析、丰富和可视化数据所需的全部配置。每个 Winlogbeat 模块由一个或多个 fileset 组成，而每个 fileset 包含：

ingest node pipelines：在 Elasticsearch 侧完成日志解析与字段丰富；
Elasticsearch 模板（index templates）：定义索引映射；
Winlogbeat 输入配置：指定采集的 Windows 事件日志通道（如 Security、System）；
Kibana 仪表盘：提供开箱即用的可视化。

这种"模块即配置包"的设计意味着：解析逻辑默认由 Elastic Stack 的 ingest node 承担，而不是在 Logstash 中通过 grok 过滤器手工实现。你可以把 Winlogbeat 模块与 Logstash 结合使用，但需要额外的几步设置，最简单的方式就是使用 Winlogbeat 提供的 ingest pipelines 完成解析。

两种解析路径：ingest pipeline 还是 Logstash 侧解析

把 Winlogbeat（或 Filebeat）模块数据接入 Logstash 时，存在两条可行的解析路径：

使用 ingest pipelines（推荐，本文主题）：在 Elasticsearch 侧加载模块自带的 ingest pipeline，Logstash 只负责传输，不做 grok/date 等解析。Logstash 配置极其轻量，且字段结构与官方模块保持一致。
Logstash 侧自行解析：跳过 pipeline 加载步骤，在 Logstash 配置中用 grok、date、geoip 等过滤器完成等价解析。这种方式不需要 ingest node，但需要手工维护解析规则。

仓库中 docs/static/filebeat_modules/system/pipeline.conf 与 docs/static/filebeat_modules/nginx/pipeline.conf 就是后一种路径的参考实现：它们用 grok 匹配 sshd/sudo/useradd 或 nginx access/error 日志，用 date 过滤纠正时间戳，用 geoip 丰富 IP 信息，并输出到elasticsearch。作为对比，如果采用 ingest pipeline 方案，这些解析工作全部下沉到 Elasticsearch 的 ingest node，Logstash 管线只需下面介绍的两段式配置。

第一步：把 ingest pipelines 加载进 Elasticsearch

在安装了 Winlogbeat 的系统上，运行带--pipelines选项的setup命令，为指定模块加载 ingest pipelines。例如，以下命令加载 security 与 sysmon 两个模块的 ingest pipelines：

winlogbeat setup --pipelines --modules security,sysmon

这一步必须能连上 Elasticsearch，因为 Winlogbeat 需要把 ingest pipelines 真正写入 Elasticsearch。如果你的 Winlogbeat 配置了其他输出（比如直接输出到 Logstash），可以临时禁用当前 output、临时启用 elasticsearch output，执行完 setup 命令后再恢复原配置。

补充说明：setup 加载的 pipelines 会以winlogbeat-<module>之类的命名注册到 Elasticsearch 的_ingest/pipelineAPI 中；后续 Winlogbeat 采集事件时，会在事件的[@metadata][pipeline]字段里写入应当使用的 pipeline 名（如winlogbeat-security），Logstash 正是依赖这个元数据做动态路由。

第二步：配置 Logstash 按元数据选择 pipeline

在安装了 Logstash 的系统上，创建一个 Logstash pipeline 配置：从某个 Logstash 输入（例如 beats 或 Kafka）读取数据，发送到 elasticsearch 输出；在 elasticsearch 输出中把pipeline选项设为%{[@metadata][pipeline]}，即可使用此前加载的 ingest pipelines。

下面是一个从 Beats 输入读取数据、按 Winlogbeat 模块 ingest pipelines 解析的完整示例：

input { beats { port => 5044 } } output { if [@metadata][pipeline] { elasticsearch { hosts => "https://061ab24010a2482e9d64729fdb0fd93a.us-east-1.aws.found.io:9243" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}" <1> action => "create" <2> pipeline => "%{[@metadata][pipeline]}" <3> user => "elastic" password => "secret" } } else { elasticsearch { hosts => "https://061ab24010a2482e9d64729fdb0fd93a.us-east-1.aws.found.io:9243" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}" <1> action => "create" user => "elastic" password => "secret" } } }

<1> 如果你在配置中禁用了数据流（data streams），请把index设为%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}。数据流默认启用。
<2> 如果禁用了数据流，可以移除该设置，或按需改为其他值。
<3> 该设置让 Logstash 根据事件中携带的元数据自动选择正确的 ingest pipeline。

关键点拆解：

pipeline => "%{[@metadata][pipeline]}"：这是整套方案的核心。Winlogbeat 模块采集时会在元数据中标记应使用的 ingest pipeline（如 security 模块对应winlogbeat-security），Logstash 在 elasticsearch 输出时把这个字段值作为 pipeline 名传给 Elasticsearch 的_ingest执行。
manage_template => false：索引模板由 Winlogbeat setup 步骤负责加载，Logstash 不应再覆盖管理模板，避免与模块自带的模板冲突。
if [@metadata][pipeline]条件分支：对未携带 pipeline 元数据的事件（如自定义输入），走不带pipeline设置的输出分支，防止向 Elasticsearch 传递空 pipeline 名。

变体：中间加一层 Kafka

原文档明确指出输入可以是 beats 或 Kafka 等。当 Winlogbeat 与 Logstash 之间隔着 Kafka 时，配置方式同理，只是把输入换成 Kafka input。仓库中 docs/reference/use-filebeat-modules-kafka.md 给出了 Filebeat 场景的完整示例，Winlogbeat 可完全照搬：

input { kafka { bootstrap_servers => "myhost:9092" topics => ["filebeat"] codec => json } } output { if [@metadata][pipeline] { elasticsearch { hosts => "https://myEShost:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" pipeline => "%{[@metadata][pipeline]}" user => "elastic" password => "secret" } } else { elasticsearch { hosts => "https://myEShost:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" user => "elastic" password => "secret" } } }

Kafka 链路下，Winlogbeat 侧需要把输出从 elasticsearch 切换为 Kafka output（在winlogbeat.yml中注释掉 elasticsearch output、启用 Kafka output，topic 固定为与 Logstash Kafka input 的topics一致），并确保 JSON codec 输出以便 Logstash 侧还原字段。Logstash 侧启动方式不变：

bin/logstash -f mypipeline.conf

数据写入 Elasticsearch 后，可在 Kibana 中打开 Winlogbeat 模块对应的仪表盘查看 Windows 安全事件、Sysmon 事件等可视化结果。

小结

把 Winlogbeat 模块接入 Logstash 的本质是职责分工：Winlogbeat 负责采集并在元数据中标注 pipeline，Elasticsearch ingest node 负责解析，Logstash 只做动态路由与写入。只要完成"setup 加载 pipelines + elasticsearch 输出配置pipeline => "%{[@metadata][pipeline]}"+manage_template => false"这三件事，就能用最少的 Logstash 配置获得与官方模块完全一致的字段结构和可视化体验。