Filebeat - 获取 Docker 容器日志并输出至 Logstash

背景简介

使用 Filebeat 收集 Docker 容器日志并推送至 Logstash。

前置信息

1. ELK Stack 8.17.0
2. Filebeat 8.17.0
3. Docker 28.5

需求信息

• 使用 Filebeat 获取 Docker 容器日志
• 将日志推送至 Logstash
• 使用 Kibana 查看日志

详细信息

文件夹架构

filebeat/
  |-- docker-compose.yaml
  |-- data/
       |-- filebeat/
            |-- filebeat.yaml

准备配置文件

• 新建 filebeat.yaml 配置文件

# ============================== Autodiscover ===============================
# 使用 templates 机制，它比 hints 更直接，更不容易出错
# 这是我们解决问题的关键配置
filebeat.autodiscover:
  providers:
    - type: docker
      templates:
        - config:
            - type: log
              paths:
                - /var/lib/docker/containers/${data.docker.container.id}/*.log

# ======================== Filebeat processors =========================
# 全局处理器，对所有输入生效
# 我们将所有处理器合并到这里，这是最佳实践
processors:
  # 【关键】添加主机元数据，用于区分不同的服务器
  - add_host_metadata:
      netinfo.enabled: false
      cache.ttl: 5m
      # geo.name: "Asia/Shanghai" # 可选，添加地理位置信息

  # 【关键】添加 Docker 元数据，如容器名、镜像名、标签等
  - add_docker_metadata: ~

  # 如果你的应用日志是 JSON 格式，强烈建议启用这个处理器
  # 它会自动解析 JSON 字段，使日志在 Kibana 中结构化
  - decode_json_fields:
      fields: ["message"]
      target: "" # 解析到根级别
      overwrite_keys: true

  # 如果你的应用是多行日志（如 Java 堆栈跟踪），请取消注释并配置
  # multiline.pattern: '^\s'
  # multiline.negate: false
  # multiline.match: after

# ======================== Filebeat outputs =========================
output.logstash:
  # 替换为你的 Logstash 服务器地址
  hosts: ["192.168.122.1:30070"]

  # 【安全注意】
  # 为了生产环境安全，强烈建议你为 Filebeat -> Logstash 的通信启用 SSL。
  # 如果启用了，你需要配置如下：
  # ssl.certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
  # ssl.certificate: "/etc/filebeat/certs/filebeat.crt"
  # ssl.key: "/etc/filebeat/certs/filebeat.key"

# ======================== Logging =========================
# 在问题解决前，建议保持 debug 级别
logging.level: debug
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

修改 Logstash 配置文件

input {
  beats {
    port => 5044
  }
}
filter {
  # 检查 message 字段是否以 { 开头，以避免解析非JSON格式的日志
  if [message] =~ /^\{.*\}$/ {
    # 1. 将 message 字段中的 JSON 字符串解析到一个新的临时字段 "docker_log_data"
    #    这样做可以避免覆盖现有的顶级字段，如 Filebeat 创建的 'log' 对象
    json {
      source => "message"
      target => "docker_log_data"
    }

    # 2. 从解析出的数据中，提取真正的日志消息，并用它覆盖原始的 'message' 字段
    #    'message' 字段是存储主要日志内容的标准字段
    mutate {
      rename => { "[docker_log_data][log]" => "message" }
    }

    # 3. （可选）如果你也需要其他字段，比如 'stream'，可以重命名它们
    # mutate {
    #   rename => { "[docker_log_data][stream]" => "[docker][stream]" }
    # }

    # 4. 清理临时字段，保持事件结构整洁
    mutate {
      remove_field => ["docker_log_data"]
    }
  }
}
output {
  elasticsearch {
    hosts => ["https://elasticsearch-node1:9200"]
    # 启用数据流
    data_stream => true
    # 为数据流设置一个名称，它会自动生成索引名
    # 例如，数据流名为 "logs-logstash.generic"，索引名会像 ".ds-logs-logstash.generic-2023.10.21-000001"
    data_stream_type => "logs"
    data_stream_dataset => "logstash"
    data_stream_namespace => "generic"
    
    user => "logstash_internal"
    password => "${LOGSTASH_INTERNAL_PASSWORD}"

    ssl => true
    cacert => "/usr/share/logstash/config/certs/ca/ca.crt"
  }
}

• 重启 Logstash

创建 Filebeat 容器

• 准备 docker-compose.yaml 配置文件

services:
  filebeat:
    image: elastic/filebeat:8.17.0
    container_name: filebeat
    restart: always
    user: root
    network_mode: host
    # 明确使用 run 指令启动
    command: ["filebeat", "run", "-e", "-c", "/usr/share/filebeat/filebeat.yml"]
    volumes:
      - ./data/filebeat/filebeat.yaml:/usr/share/filebeat/filebeat.yaml:ro
      - /var/lib/docker/containers/:/var/lib/docker/containers/:ro
      - /var/run/docker.sock:/var/run/docker.sock:ro
      - /:/hostfs:ro

• 查看 filebeat 推送logstash状态

docker exec filebeat filebeat test output

logstash: your-logstash-server-ip:5044...
  connection...
    parse host... OK
    dns lookup... OK
    addresses: your-logstash-server-ip
    dial up... OK
  TLS... WARN secure connection disabled
  talk to server... OK

以上便是本文的全部内容，感谢您的阅读，如遇到任何问题，欢迎在评论区留言讨论。