Filebeat与Logstash对比分析：ELK日志收集架构最佳实践

情境与背景

在现代分布式系统中，日志收集与分析是保障业务稳定运行的关键能力。ELK（Elasticsearch、Logstash、Kibana）技术栈是日志处理领域最成熟的解决方案之一，而Filebeat和Logstash作为ELK的核心组件，承担着日志采集与处理的重任。

理解两者的定位差异和使用场景，是构建高效、可靠日志系统的基础。作为高级DevOps/SRE工程师，掌握Filebeat与Logstash的选型和配置，是必备的核心技能。两者是互补关系：Filebeat负责轻量级采集，Logstash负责复杂处理，联合使用才能发挥ELK的最大价值。

一、ELK技术栈概述

1.1 ELK组件架构

flowchart LR
    subgraph 数据采集["数据采集层"]
        A["应用日志"]
        B["系统日志"]
        C["容器日志"]
    end

    subgraph 数据处理["数据处理层"]
        D["Filebeat"]
        E["Kafka"]
        F["Logstash"]
    end

    subgraph 数据存储["数据存储层"]
        G["Elasticsearch"]
    end

    subgraph 数据展示["数据展示层"]
        H["Kibana"]
    end

    A --> D
    B --> D
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H

    style D fill:#c8e6c9
    style F fill:#fff3e0
    style G fill:#e3f2fd

1.2 组件职责

二、Filebeat详解

2.1 Filebeat是什么

Filebeat是Elastic官方开发的轻量级日志采集器，用Go语言编写，单进程内存占用仅约40MB。Filebeat设计目标是用最少的资源完成可靠的日志采集，通过backpressure机制避免数据丢失。

flowchart TB
    subgraph Filebeat架构
        A["Input插件"] --> B["Harvester收割机"]
        B --> C["Spooler缓冲区"]
        C --> D["Output插件"]
    end

    subgraph Input类型
        A1["Log文件"]
        A2["Redis"]
        A3["Kafka"]
        A4["Docker"]
    end

    subgraph Output类型
        D1["Elasticsearch"]
        D2["Logstash"]
        D3["Kafka"]
        D4["File"]
    end

    A1 --> A
    A2 --> A
    A3 --> A
    A4 --> A

    D --> D1
    D --> D2
    D --> D3
    D --> D4

    style A fill:#e3f2fd
    style B fill:#c8e6c9
    style C fill:#fff3e0

2.2 Filebeat核心概念

2.3 Filebeat配置示例

# filebeat.yml
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
      - /var/log/nginx/*.log
    multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
    multiline.negate: true
    multiline.match: after
    fields:
      service: myapp
      environment: production

  - type: docker
    containers.paths:
      - /var/lib/docker/containers/*/*.log
    containers.ids:
      - '*'

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_docker_metadata: ~
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
      add_error_key: true

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: 'logs-%{[fields.service]}'
  partition.round_robin:
    reachable_only: false
  required_acks: 1
  compression: gzip
  max_message_bytes: 1048576

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

三、Logstash详解

3.1 Logstash是什么

Logstash是ELK技术栈中的数据处理引擎，用JRuby编写，运行在JVM上。Logstash提供了丰富的Input/Filter/Output插件，能够完成复杂的数据采集、解析、转换和输出任务。

flowchart TB
    subgraph Logstash处理管道
        A["Input"] --> B["Filter"]
        B --> C["Output"]
    end

    subgraph Input插件
        A1["File"]
        A2["Kafka"]
        A3["Beats"]
        A4["JDBC"]
    end

    subgraph Filter插件
        B1["Grok"]
        B2["Mutate"]
        B3["Date"]
        B4["GeoIP"]
        B5["JSON"]
    end

    subgraph Output插件
        C1["Elasticsearch"]
        C2["Kafka"]
        C3["File"]
        C4["Email"]
    end

    A1 --> A
    A2 --> A
    A3 --> A
    A4 --> A

    B --> B1
    B --> B2
    B --> B3
    B --> B4
    B --> B5

    C1 --> C
    C2 --> C
    C3 --> C
    C4 --> C

    style A fill:#e3f2fd
    style B fill:#c8e6c9
    style C fill:#fff3e0

3.2 Logstash核心概念

3.3 Logstash配置示例

# pipeline.conf
input {
  kafka {
    bootstrap_servers => "kafka1:9092,kafka2:9092"
    topics => ["logs-*"]
    group_id => "logstash-consumer"
    codec => json
    decorate_events => true
  }

  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
  }
}

filter {
  if [fields][service] == "nginx" {
    grok {
      match => {
        "message" => '%{IPORHOST:[nginx][client_ip]} - %{DATA:[nginx][user]} \[%{HTTPDATE:[nginx][timestamp]}\] "%{WORD:[nginx][method]} %{URIPATHPARAM:[nginx][uri]} HTTP/%{NUMBER:[nginx][http_version]}" %{NUMBER:[nginx][status]} %{NUMBER:[nginx][body_bytes]} "%{DATA:[nginx][referrer]}" "%{DATA:[nginx][agent]}"'
      }
      remove_field => "message"
    }

    mutate {
      convert => {
        "[nginx][status]" => "integer"
        "[nginx][body_bytes]" => "integer"
      }
    }

    date {
      match => ["[nginx][timestamp]", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }

    geoip {
      source => "[nginx][client_ip]"
      target => "[nginx][geoip]"
      database => "/etc/logstash/GeoLite2-City.mmdb"
    }
  }

  if [fields][service] == "application" {
    json {
      source => "message"
      target => "parsed"
    }

    mutate {
      rename => {
        "[parsed][level]" => "[log][level]"
        "[parsed][message]" => "[log][message]"
        "[parsed][trace_id]" => "[trace][id]"
      }
      remove_field => ["parsed", "message"]
    }
  }

  mutate {
    add_field => {
      "[@metadata][index_prefix]" => "logs-%{+YYYY.MM.dd}"
    }
  }
}

output {
  elasticsearch {
    hosts => ["es1:9200", "es2:9200", "es3:9200"]
    manage_template => false
    index => "%{[@metadata][index_prefix]}"
    document_type => "_doc"
    action => "index"
    sniffing => true
    pool_max => 50
    pool_max_per_route => 25
  }

  kafka {
    bootstrap_servers => "kafka1:9092,kafka2:9092"
    topic_id => "logs-processed"
    codec => json
    compression => "snappy"
  }

  if "ERROR" in [log][level] {
    email {
      to => "oncall@example.com"
      from => "logstash-alert@example.com"
      subject => "Error Alert: %{[log][message]}"
      via => "smtp"
      options => {
        address => "smtp.example.com"
        port => 587
        enable_starttls_auto => true
      }
    }
  }

  stdout {
    codec => rubydebug
  }
}

四、Filebeat与Logstash对比

4.1 核心维度对比

4.2 性能对比

flowchart TB
    A["日志处理能力"] --> B["Filebeat"]
    A --> C["Logstash"]

    B --> B1["单实例吞吐量"]
    B1 --> B2["50-100 MB/s"]

    C --> C1["单实例吞吐量"]
    C1 --> C2["20-50 MB/s"]

    A --> D["资源消耗对比"]

    D --> D1["Filebeat: 40MB内存"]
    D --> D2["Logstash: 1-4GB内存"]

    style B fill:#c8e6c9
    style C fill:#fff3e0

4.3 选型决策树

flowchart TB
    A["日志处理需求"] --> B{"是否需要复杂解析?"}
    B -->|是| C["Logstash"]
    B -->|否| D{"日志量大吗?"}
    D -->|是| E{"需要多输出吗?"}
    D -->|否| F["Filebeat"]
    E -->|是| G["Filebeat + Logstash"]
    E -->|否| H["Filebeat + Kafka"]
    C --> I["是否需要横向扩展?"}
    I -->|是| J["Logstash集群"]
    I -->|否| K["单节点Logstash"]

    style C fill:#fff3e0
    style G fill:#e3f2fd
    style F fill:#c8e6c9

五、生产环境最佳实践

5.1 架构模式选择

模式一：Filebeat + Kafka + Logstash（推荐大型生产环境）

flowchart LR
    subgraph 采集层
        A1["App Server 1"]
        A2["App Server 2"]
        A3["App Server N"]
    end

    subgraph 缓冲层
        B["Kafka Cluster"]
    end

    subgraph 处理层
        C["Logstash Cluster"]
    end

    subgraph 存储层
        D["Elasticsearch"]
        E["Kibana"]
    end

    A1 --> B
    A2 --> B
    A3 --> B
    B --> C
    C --> D
    D --> E

    style B fill:#fff3e0
    style C fill:#e3f2fd

模式二：Filebeat直连Logstash（中小型环境）

flowchart LR
    A["Filebeat"] --> B["Logstash"]
    B --> C["Elasticsearch"]

    style A fill:#c8e6c9
    style B fill:#fff3e0

模式三：Filebeat直连Elasticsearch（极简场景）

flowchart LR
    A["Filebeat"] --> B["Elasticsearch"]
    B --> C["Kibana"]

    style A fill:#c8e6c9
    style B fill:#e3f2fd

5.2 Filebeat最佳实践

K8s环境部署配置

# filebeat-kubernetes.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config
  namespace: logging
data:
  filebeat.yml: |
    filebeat.inputs:
      - type: container
        paths:
          - /var/log/containers/*.log
        processors:
          - add_kubernetes_metadata:
              host: ${NODE_NAME}
              matchers:
                - logs_path:
                    logs_path: "/var/log/containers/"

    processors:
      - add_host_metadata:
          when.not.contains.tags: forwarded
      - add_cloud_metadata: ~
      - add_docker_metadata: ~

    output.elasticsearch:
      hosts: ['${ELASTICSEARCH_HOST:elasticsearch}:${ELASTICSEARCH_PORT:9200}']
      index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"

    setup.kibana:
      host: '${KIBANA_HOST:kibana}:${KIBANA_PORT:5601}'

    setup.ilm.enabled: auto
    setup.ilm.rollover_alias: "filebeat"
    setup.ilm.pattern: "{now/d}-000001"
    setup.ilm.policy_name: "filebeat-policy"
---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: filebeat
  namespace: logging
spec:
  selector:
    matchLabels:
      app: filebeat
  template:
    metadata:
      labels:
        app: filebeat
    spec:
      serviceAccountName: filebeat
      terminationGracePeriodSeconds: 30
      hostNetwork: true
      dnsPolicy: ClusterFirstWithHostNet
      containers:
        - name: filebeat
          image: docker.elastic.co/beats/filebeat:8.12.0
          args:
            - -c
            - /etc/filebeat.yml
            - -e
          env:
            - name: NODE_NAME
              valueFrom:
                fieldRef:
                  fieldPath: spec.nodeName
          securityContext:
            runAsUser: 0
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: 1000m
              memory: 512Mi
          volumeMounts:
            - name: config
              mountPath: /etc/filebeat.yml
              readOnly: true
              subPath: filebeat.yml
            - name: data
              mountPath: /usr/share/filebeat/data
            - name: varlogcontainers
              mountPath: /var/log/containers
              readOnly: true
            - name: varlogpods
              mountPath: /var/log/pods
              readOnly: true
            - name: varlibdockercontainers
              mountPath: /var/lib/docker/containers
              readOnly: true
      volumes:
        - name: config
          configMap:
            defaultMode: 0640
            name: filebeat-config
        - name: varlogcontainers
          hostPath:
            path: /var/log/containers
        - name: varlogpods
          hostPath:
            path: /var/log/pods
        - name: varlibdockercontainers
          hostPath:
            path: /var/lib/docker/containers
        - name: data
          emptyDir: {}

Filebeat Module使用

# 启用官方Module，自动解析Nginx日志
filebeat.modules:
  - module: nginx
    access:
      enabled: true
      var.paths: ["/var/log/nginx/access.log*"]
    error:
      enabled: true
      var.paths: ["/var/log/nginx/error.log*"]

  - module: system
    syslog:
      enabled: true
    auth:
      enabled: true

  - module: docker
    enabled: true
    containers:
      path: /var/lib/docker/containers
      stream: all

5.3 Logstash最佳实践

JVM调优配置

# jvm.options
-Xms4g
-Xmx4g
-XX:+UseG1GC
-XX:MaxGCPauseMillis=500
-XX:+PrintGCTimeStamps
-XX:+PrintGCDateStostamps
-XX:+PrintTenuringDistribution
-XX:NumberOfGCLogFiles=10
-XX:+UseGCLogFileRotation
-Xloggc:/var/log/logstash/gc.log

# 关闭JMX远程监控（生产环境）
-Dcom.sun.management.jmxremote=false

Pipeline并发优化

# pipelines.yml
- pipeline.id: main
  path.config: "/etc/logstash/pipelines/main/*.conf"
  pipeline.workers: 4
  pipeline.batch.size: 125
  pipeline.batch.delay: 50

- pipeline.id: slow-logs
  path.config: "/etc/logstash/pipelines/slow/*.conf"
  pipeline.workers: 2
  pipeline.batch.size: 50
  queue.type: persisted
  queue.max_bytes: 1gb

内存管道配置

# pipeline.conf
pipeline {
  workers: 4
  batch_size: 125
  batch_delay: 50
}

# 启用持久化队列，防止数据丢失
queue {
  type => persisted
  path => "/var/lib/logstash/queue"
  capacity {
    max_bytes => 10gb
    max_events => 0
  }
}

5.4 Kafka缓冲配置

Kafka主题配置

# 创建日志主题
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092,kafka2:9092 \
  --topic logs-raw \
  --partitions 12 \
  --replication-factor 2 \
  --config retention.ms=604800000 \
  --config max.message.bytes=10485760

# 配置保留策略
kafka-topics.sh --alter \
  --topic logs-raw \
  --bootstrap-server kafka1:9092,kafka2:9092 \
  --add-config cleanup.policy=delete,retention.ms=604800000

5.5 故障排查

Filebeat排查命令

# 测试配置
filebeat test config -c /etc/filebeat/filebeat.yml

# 查看采集状态
filebeat -e -d "*"  # 调试模式运行

# 查看Harvester状态
GET _nodes/<node_id>/stats/fs?pretty  # 通过Elasticsearch API

# 检查registrar状态
filebeat export registry --dir=/var/lib/filebeat/registry

# 监控日志采集
filebeat监控 API: GET /_node/stats?pretty

Logstash排查命令

# 测试配置语法
/usr/share/logstash/bin/logstash --config.test_and_exit -f /etc/logstash/pipeline.conf

# 调试模式运行
/usr/share/logstash/bin/logstash --debug -f /etc/logstash/pipeline.conf

# 查看JVM状态
jstat -gcutil <pid> 1000 10

# 线程dump
jstack <pid> > /tmp/logstash-threaddump.txt

# GC日志分析
gceasy.io 分析gc.log

常见问题与解决

5.6 监控告警

Filebeat监控指标

# filebeat监控配置
filebeat.metrics:
  enabled: true
  # 报告间隔
  period: 30s

# 通过Elasticsearch监控
GET /_nodes/<node_id>/stats/filebeat?pretty

Logstash监控指标

# Logstash API
GET /_node/stats/pipeline?pretty
GET /_node/stats/jvm?pretty

# 关键指标
{
  "pipeline": {
    "events": {
      "duration_in_millis": 1250,
      "in": 1000,
      "out": 998,
      "filtered": 998
    },
    "queue": {
      "capacity": {
        "queue_size_in_bytes": 1073741824,
        "max_queue_size_in_bytes": 10737418240
      }
    }
  },
  "jvm": {
    "mem": {
      "heap_used_in_bytes": 2143000000,
      "heap_max_in_bytes": 4290000000
    }
  }
}

六、面试精简版

6.1 一分钟版本

Filebeat和Logstash是ELK日志系统的两个核心组件，定位不同：Filebeat是轻量级日志采集器，用Go编写，资源消耗低（约40MB内存），负责从日志源采集数据、做基本过滤和多行合并，适合大规模分散部署在每台主机上；Logstash是重量级日志处理管道，用JVM运行，功能强大，支持丰富的Input/Filter/Output插件，能做复杂的数据解析、转换和富化，适合集中部署做复杂处理。生产环境最佳实践是Filebeat负责采集，通过Kafka缓冲后，Logstash做集中处理，最后写入Elasticsearch。

6.2 记忆口诀

日志采集选Filebeat，轻量低耗Go语言；
日志处理选Logstash，插件丰富JVM跑；
量大轻量Filebeat，复杂解析Logstash；
生产架构Kafka解耦，采集处理分离跑。

6.3 关键词速记

参考链接：SRE运维面试题全解析：从理论到实践（第三部分）