3. 日志采集

日志采集是整个可观测性链条的第一公里。说白了，你后面分析能力再强，数据进不来或者进来的是脏数据，全都白搭。这一章我带你过三个主流工具：Filebeat、Logstash、Fluentd。它们各有各的脾气，也各有各的适用场景。

核心观点：日志采集不是简单的“读文件→发出去”。你要考虑采集性能、断点续传、数据清洗、背压处理。这三个工具，我建议你至少熟练掌握两个。

日志采集三大工具架构对比 API 应用 API 应用 API 应用 Filebeat Logstash Fluentd Elasticsearch Elasticsearch 多种后端 Filebeat：轻量、低资源、适合边车模式 Logstash：管道丰富、适合复杂清洗、资源消耗大 Fluentd：插件生态好、统一日志层、社区活跃

3.1 Filebeat：轻量级采集器配置

Filebeat 是我个人最常用的采集器。为什么？因为它够轻。一个二进制文件扔上去就能跑，内存占用通常不到 20MB。你想想看，在容器环境里，每个 Pod 塞一个 Filebeat 边车，几乎感觉不到资源开销。

3.1.1 核心配置结构

Filebeat 的配置核心就两块：输入（input）和输出（output）。中间可以加一些简单的处理模块，但别指望它做复杂的数据清洗——那不是它的活。

# filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/api-gateway/*.log
    - /var/log/nginx/access.log
  fields:
    service: api-gateway
    env: production
  fields_under_root: true
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2}'
    negate: true
    match: after

output.elasticsearch:
  hosts: ["http://elasticsearch:9200"]
  index: "api-logs-%{+yyyy.MM.dd}"
  username: "filebeat"
  password: "${ES_PWD}"

# 别忘了加这个
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat.log
  keepfiles: 7

我的经验：fields_under_root: true 这个配置很多人会漏掉。不加的话，你自定义的字段会被嵌套在 fields 对象里，查询时多一层路径，很烦人。

3.1.2 多行日志处理

API 调用审计日志经常跨多行——比如 Java 的堆栈跟踪。Filebeat 的 multiline 配置就是干这个的。

我曾经遇到过一个坑：某个微服务的日志时间戳格式是 yyyy-MM-dd HH:mm:ss，但有个别日志行开头是空格。结果 Filebeat 把堆栈跟踪的每一行都当成独立事件发了出去。后来我加了个 negate: true，问题解决。

3.1.3 断点续传与注册表

Filebeat 用 registry 文件记录每个文件的读取位置。就算进程重启，它也能从断点处继续读，不会丢数据也不会重复发。

# 查看注册表状态
filebeat -e -d "publish"

注意：千万别手贱删了 data/registry 目录。我有个同事在生产环境上干过这事，结果 Filebeat 把过去一周的日志重新发了一遍，ES 集群直接被打满。

3.2 Logstash：管道处理

Logstash 是 ELK 栈里的“老大哥”。它不负责采集——那是 Filebeat 的事——它负责管道处理。说白了，数据进来之后怎么洗、怎么拆、怎么补字段，都是 Logstash 的活。

3.2.1 管道生命周期

Logstash 的管道分三个阶段：输入 → 过滤 → 输出。每个阶段都可以有多个插件串起来。

阶段 常用插件 作用
输入 beats, http, tcp, file 接收数据源
过滤 grok, mutate, date, json 解析、清洗、转换
输出 elasticsearch, stdout, kafka 发送到目标

3.2.2 Grok 解析实战

API 审计日志最常见的格式是 JSON。但有时候你会遇到非结构化的文本日志，比如 Nginx 的 access log。这时候 Grok 就派上用场了。

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

filter {
  if [service] == "api-gateway" {
    grok {
      match => {
        "message" => "%{IP:client_ip} - %{USER:ident} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status_code} %{NUMBER:bytes}"
      }
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
    mutate {
      remove_field => ["message", "ident"]
    }
  }
}

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

关键点：Grok 的正则表达式很吃性能。我建议你在 if 条件里先做分流，只对需要的日志做 Grok 解析，别一股脑全塞进去。

3.2.3 性能调优

Logstash 是 Java 写的，内存消耗不小。我见过有人一台 4G 的机器跑 Logstash，结果 OOM 了。几个调优参数你得记住：

  • pipeline.workers：默认是 CPU 核数，可以适当调大
  • pipeline.batch.size：默认 125，处理大日志时可以调到 500
  • pipeline.batch.delay：默认 50ms，适当调小可以降低延迟
# jvm.options
-Xms2g
-Xmx2g

# logstash.yml
pipeline.workers: 4
pipeline.batch.size: 500
pipeline.batch.delay: 10

3.3 Fluentd：部署实战

Fluentd 是 CNCF 毕业项目，在 Kubernetes 生态里用得特别多。它的设计哲学和 Logstash 不太一样——Fluentd 强调统一日志层，所有日志先到 Fluentd，再由它路由到不同后端。

3.3.1 架构与部署

Fluentd 的部署方式有两种：DaemonSet（每个节点一个）和边车模式（每个 Pod 一个）。我个人更推荐 DaemonSet，管理成本低。

# fluentd-daemonset.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: logging
spec:
  selector:
    matchLabels:
      name: fluentd
  template:
    metadata:
      labels:
        name: fluentd
    spec:
      containers:
      - name: fluentd
        image: fluent/fluentd:v1.16
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: config
          mountPath: /fluentd/etc
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: config
        configMap:
          name: fluentd-config

3.3.2 配置文件结构

Fluentd 的配置用 <source>、<filter>、<match> 三个指令来组织。和 Logstash 的 input/filter/output 是对应的。

# fluentd.conf
<source>
  @type tail
  path /var/log/containers/*.log
  pos_file /var/log/fluentd-containers.log.pos
  tag kubernetes.*
  format json
  read_from_head true
</source>

<filter kubernetes.**>
  @type kubernetes_metadata
  annotation_match ["prometheus.io/scrape"]
</filter>

<match kubernetes.**>
  @type elasticsearch
  host elasticsearch.logging.svc.cluster.local
  port 9200
  logstash_format true
  logstash_prefix api-audit
  flush_interval 5s
</match>

避坑指南：pos_file 一定要配置。我曾经在测试环境忘了加，结果每次 Fluentd 重启都重新读一遍所有日志，ES 里全是重复数据。

3.3.3 插件生态

Fluentd 的插件超过 1000 个。常用的几个：

  • fluent-plugin-elasticsearch：输出到 ES
  • fluent-plugin-kafka：输出到 Kafka
  • fluent-plugin-rewrite-tag-filter：动态修改标签
  • fluent-plugin-grok-parser：类似 Logstash 的 Grok

3.4 工具选型对比

三个工具怎么选？我列个表给你参考：

维度 Filebeat Logstash Fluentd
资源消耗 极低（~20MB） 较高（~1GB+） 中等（~200MB）
处理能力 简单过滤 复杂管道 中等复杂度
插件生态 有限 丰富 非常丰富
K8s 集成 好（Operator） 一般 优秀（CNCF）
适用场景 边车采集 集中清洗 统一日志层

我的建议：小团队、轻量场景用 Filebeat + ES 就够了。日志量大、清洗逻辑复杂，加一层 Logstash。如果你们已经在用 K8s，Fluentd 是更自然的选择。

3.5 实战：搭建一个完整的采集管道

最后，我带你走一遍完整的搭建流程。假设我们要采集 API 网关的审计日志，经过 Logstash 清洗后存入 ES。

  1. 部署 Filebeat：在每个 API 网关节点上部署 Filebeat，配置输入为日志文件，输出指向 Logstash 的 5044 端口。
  2. 配置 Logstash：接收 Beats 输入，用 Grok 解析日志，用 Mutate 清理字段，输出到 ES。
  3. 验证管道：用 stdout { codec => rubydebug } 先输出到控制台，确认解析正确后再切到 ES。
  4. 监控采集状态：Filebeat 的 filebeat status 命令可以查看采集进度，Logstash 的监控 API 在 http://localhost:9600。
# 验证 Logstash 管道
curl http://localhost:9600/_node/stats/pipeline?pretty

重要：生产环境一定要加 pipeline.ecs_compatibility: disabled（如果你用旧版 ES）。否则 Logstash 默认的 ECS 模式会把字段名改掉，你的 Kibana 仪表盘可能全崩。

← 上一章 📖 返回目录 下一章 →