4. Logstash与Ingest Pipeline安全数据处理

安全日志进入 Elasticsearch 之前，往往处于“原始文本”状态：防火墙日志是一行 syslog，IDS 告警是 JSON 里嵌套的 signature 字段，EDR 事件是键值对拼接，云审计日志则是多层嵌套 JSON。如果直接写入索引，字段类型混乱、无法聚合、无法做可视化。本节的核心任务，就是利用 Logstash 的 filter 插件与 Elasticsearch Ingest Pipeline 的处理器，把异构日志统一成可检索、可聚合的标准化字段。

一、两条处理路径：Logstash 与 Ingest Pipeline 的分工

Logstash 与 Ingest Pipeline 都能做字段解析，但定位不同。Logstash 运行在 Elasticsearch 之外，适合做重加工：多行合并、复杂条件分支、外部数据富化、写入前路由。Ingest Pipeline 运行在 Elasticsearch 节点内，适合做轻量、贴近索引的标准化：字段重命名、类型转换、GeoIP 富化、时间戳解析。工程上常见的组合是：Logstash 负责采集与粗解析，Ingest Pipeline 负责写入前的字段标准化与索引路由。

选择依据可以概括为：需要外部查询（如查威胁情报）、需要多行合并、需要复杂条件判断时用 Logstash；只需要字段改名、类型转换、GeoIP、User-Agent 解析时用 Ingest Pipeline，因为它省去额外进程，且能通过 _simulate API 快速验证。

二、Grok 与 Dissect：两种解析思路

Grok 基于正则，适合字段数量不固定、格式有变化的日志。例如防火墙 syslog：

filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:event_time} %{HOSTNAME:host.name} %%{WORD:vendor}/%{WORD:product}: %{GREEDYDATA:event.original}" }
  }
  date {
    match => [ "event_time", "MMM dd HH:mm:ss" ]
    target => "@timestamp"
  }
}

Grok 的代价是正则回溯，日志量大时 CPU 消耗明显。Dissect 则按分隔符逐段切分，不回溯，性能更好，但要求格式严格固定。例如 EDR 事件：

dissect {
  mapping => {
    "message" => "%{ts} %{host.name} %{user.name} %{event.action} %{process.name} %{source.ip}:%{source.port}->%{dest.ip}:%{dest.port}"
  }
}

选择原则：格式稳定用 Dissect，格式多变用 Grok；两者可组合，先用 Dissect 切出主干，再用 Grok 处理剩余字段。

三、GeoIP 与 User-Agent：富化与语义还原

GeoIP 处理器把 IP 映射为地理位置。在 Ingest Pipeline 中：

{
  "geoip": {
    "field": "source.ip",
    "target_field": "source.geo",
    "properties": ["country_name","city_name","location"]
  }
}

注意：GeoIP 只对公网 IP 有效，内网 IP 会返回空值，需用 if 条件或 ignore_missing 避免报错。User-Agent 处理器把 UA 字符串还原为 user_agent.name、user_agent.os.name、user_agent.device.name，对 Web 攻击溯源尤其有用。

四、字段标准化与索引路由

解析完成后，字段名往往五花八门：src_ip、client_ip、remote_addr 都指来源 IP。标准化就是统一到 ECS 字段：source.ip、destination.ip、event.action、event.outcome。常用处理器：

• rename：把 src_ip 改为 source.ip；
• convert：把字符串端口转为 integer，便于范围查询；
• set：补充 event.dataset: firewall，用于索引路由；
• remove：删除原始冗余字段，控制索引体积。

索引路由通过 index 处理器或 Logstash 的 elasticsearch 输出按 event.dataset 分流，例如 firewall-%{+YYYY.MM}。这样既保证字段一致，又让不同日志落在各自索引，便于生命周期管理。

五、验证与排障

Pipeline 写好后不要直接上生产，先用 _simulate 验证：

POST _ingest/pipeline/firewall-parse/_simulate
{
  "docs": [
    { "_source": { "message": "Jun 12 08:31:02 fw01 vendor/fw: src=10.1.2.3 dst=8.8.8.8 dport=443 action=allow" } }
  ]
}

观察返回的 docs[0].doc._source，确认 source.ip、destination.port、event.action 是否正确。常见问题：Grok 不匹配导致 _grokparsefailure 标签，需检查正则；时间戳解析失败导致 @timestamp 为写入时间而非事件时间；GeoIP 对私网 IP 返回空，需加条件判断。排障时优先看 tags 字段和 error.message，再对照 _mapping 确认字段类型是否符合预期。