3、日志采集实战：使用 Filebeat 采集 Nginx 日志、Logstash 管道配置与过滤规则

好，咱们进入实战环节。前面聊了那么多理论，说白了，日志采集是整个可观测性体系的「水源」。水源不干净，后面分析得再漂亮也是白搭。我个人习惯，在生产环境里，日志采集首选 Filebeat + Logstash 这套组合拳。为什么？轻量、稳定、好维护。

今天咱们就手把手过一遍：怎么用 Filebeat 把 Nginx 日志老老实实采上来，再通过 Logstash 管道给它「洗个澡」，变成结构化的、能直接扔进 Elasticsearch 分析的数据。

3.1 为什么是 Filebeat 而不是 Logstash 直接采？

你可能会问：Logstash 也能采集日志啊，为什么非要中间加一层 Filebeat？

嗯，这里有个坑。Logstash 是用 Java 写的，吃内存比较厉害。如果你在每台 Nginx 服务器上都部署一个 Logstash，那资源开销会让你肉疼。我在项目中遇到过，一台 4C8G 的机器，光 Logstash 就占了 2G 内存，业务方直接来找我喝茶了。

Filebeat 是用 Go 写的，内存占用通常只有几十 MB。它的职责很纯粹：读文件、发数据。至于解析、过滤、转换这些重活，交给后面的 Logstash 集群去做。这叫「各司其职」。

3.2 Filebeat 采集 Nginx 日志实战

先看 Filebeat 的配置。假设你的 Nginx 日志在 /var/log/nginx/access.log，格式是默认的 combined 格式。

3.2.1 基础配置

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log*
  fields:
    log_type: nginx_access
  fields_under_root: true

output.logstash:
  hosts: ["192.168.1.100:5044"]

这段配置很简单：告诉 Filebeat 去读 /var/log/nginx/ 下所有以 access.log 开头的文件（包括轮转的 access.log.1 等）。然后加了一个自定义字段 log_type: nginx_access，方便 Logstash 后面做路由判断。

3.2.2 多行日志处理

Nginx 的 access log 一般是一行一条，但 error log 可能有多行。比如 Java 的堆栈信息。这时候需要配置 multiline：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/error.log
  multiline:
    pattern: '^\d{4}/\d{2}/\d{2}'
    negate: true
    match: after
  fields:
    log_type: nginx_error

这个配置的意思是：如果一行不以日期开头（^\d{4}/\d{2}/\d{2}），就把它合并到上一行的后面。这样堆栈信息就不会被拆成好几条记录了。

3.3 Logstash 管道配置与过滤规则

数据到了 Logstash，才是真正「变魔术」的地方。Logstash 的管道由三部分组成：input → filter → output。咱们重点讲 filter，因为这是数据清洗的核心。

3.3.1 接收 Filebeat 数据

# logstash.conf
input {
  beats {
    port => 5044
  }
}

嗯，就这么简单。Logstash 会监听 5044 端口，接收 Filebeat 发来的数据。

3.3.2 用 grok 解析 Nginx 日志

Nginx 的 combined 格式长这样：

192.168.1.1 - - [10/Oct/2023:13:55:36 +0000] "GET /api/users HTTP/1.1" 200 2326 "https://example.com" "Mozilla/5.0"

我们需要把它拆成：客户端 IP、时间戳、请求方法、URL、状态码、响应大小、Referer、User-Agent 等字段。用 grok 搞定：

filter {
  if [log_type] == "nginx_access" {
    grok {
      match => {
        "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:url} HTTP/%{NUMBER:http_version}\" %{NUMBER:status_code} %{NUMBER:body_bytes} \"%{DATA:referrer}\" \"%{DATA:user_agent}\""
      }
    }
  }
}

这里用了一个条件判断 if [log_type] == "nginx_access"，确保只处理 Nginx access 日志。grok 模式里，%{IPORHOST:client_ip} 这种写法，冒号前面是预定义模式，冒号后面是你自定义的字段名。

3.3.3 日期转换与字段处理

解析出来的 timestamp 是字符串，我们需要把它转成 Logstash 的 @timestamp 字段，方便后续按时间排序和聚合：

filter {
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
  mutate {
    remove_field => ["timestamp", "message"]
    convert => {
      "body_bytes" => "integer"
      "status_code" => "integer"
    }
  }
}

这里做了三件事：

• 把 timestamp 字符串解析成 @timestamp 时间戳
• 删掉原始的 timestamp 和 message 字段（省空间）
• 把 body_bytes 和 status_code 转成整数类型，方便后面做数值计算

3.3.4 User-Agent 解析

User-Agent 字符串又长又乱，直接存起来分析效率很低。用 useragent 插件把它拆开：

filter {
  useragent {
    source => "user_agent"
    target => "ua"
  }
}

这样会生成 ua.os_name、ua.browser_name、ua.device 等字段。比如你可以直接查「有多少请求来自 Chrome 浏览器」，不用再写正则去匹配了。

3.4 完整管道示例

把上面这些串起来，一个完整的 Logstash 管道配置大概长这样：

input {
  beats {
    port => 5044
  }
}

filter {
  if [log_type] == "nginx_access" {
    grok {
      match => {
        "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:url} HTTP/%{NUMBER:http_version}\" %{NUMBER:status_code} %{NUMBER:body_bytes} \"%{DATA:referrer}\" \"%{DATA:user_agent}\""
      }
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
    useragent {
      source => "user_agent"
      target => "ua"
    }
    mutate {
      remove_field => ["timestamp", "message"]
      convert => {
        "body_bytes" => "integer"
        "status_code" => "integer"
      }
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "nginx-access-%{+YYYY.MM.dd}"
  }
}

3.5 知识体系与核心逻辑

下面这张图，帮你理清整个采集链路的脉络：

3.6 性能调优建议

最后，分享几个我在生产环境里踩过的坑和优化经验：

组件	常见问题	优化方案
Filebeat	采集速度跟不上日志写入速度	调大 harvester_buffer_size（默认 16384），或增加 spool_size
Logstash	grok 解析成为性能瓶颈	使用 grok 的 break_on_match 参数，或把不常用的字段用 drop 丢弃
Logstash	内存占用过高	设置 pipeline.workers 为 CPU 核心数，pipeline.batch.size 从 125 调小到 50
整体	日志丢失	Filebeat 开启 close_inactive 和 clean_inactive，Logstash 开启持久队列

好了，这一章的内容就到这。日志采集是地基，地基打牢了，后面的性能分析才能站得稳。下一章咱们会深入 Elasticsearch，看看怎么把这些清洗好的数据用起来。

---