Filebeat 轻量级日志采集实践：安装、配置、多行合并、JSON 解析与字段处理-json-CSS教程网

Filebeat 轻量级日志采集器实战指南

✅ 适用版本：Filebeat 8.2.2 + Elasticsearch 8.2.2 / Logstash 8.2.2
✅ 部署目标：从服务器采集日志并发送至 Elasticsearch 或 Logstash

一、Beats 简介

Beats 是 Elastic 开源的轻量级数据采集平台，专为高效、低资源消耗地收集各类数据而设计。它由多个专用采集器组成，适用于大规模分布式环境下的日志、指标、网络流量等数据采集。

官方 Beats 组件一览

Beats 组件	用途
Filebeat	采集日志文件（如 Nginx、系统日志、应用日志）
Metricbeat	采集系统/服务指标（CPU、内存、MySQL、Redis 等）
Packetbeat	采集网络流量数据（抓包分析）
Winlogbeat	采集 Windows 事件日志（Event Log）
Auditbeat	采集 Linux 审计日志和文件完整性监控
Heartbeat	监测服务可用性（Ping、HTTP、TCP）

🔗 官网地址：https://www.elastic.co/***/beats

二、Filebeat 概述

Filebeat 是 Beats 家族中最常用的组件，专为日志文件采集设计。它具有以下特点：

✅ 轻量级：资源占用极低，适合部署在大量边缘服务器。
✅ 可靠：通过文件状态记录（registry）确保日志不丢失。
✅ 灵活输出：支持直接发送到 Elasticsearch、Logstash，或通过 Kafka、Redis 中转。
✅ 模块化：内置 Nginx、Apache、MySQL、System 等常见日志模块，开箱即用。
✅ 安全支持：支持 TLS 加密、身份认证，适配 ES 8.x 安全机制。

📌 典型应用场景：

收集 Nginx/Apache 访问日志、错误日志
采集 Java 应用日志（如 Spring Boot）
收集系统日志（/var/log/messages, secure 等）
多服务器日志集中到 ELK/EFK 平台

🔗 Filebeat 官网：https://www.elastic.co/***/beats/filebeat

三、部署 Filebeat 8.2.2（二进制方式）

⚠️ 注意：本文使用 Filebeat 8.2.2，与你的 Elasticsearch 8.2.2 版本严格匹配。

1. 下载 Filebeat 8.2.2

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.2.2-linux-x86_64.tar.gz

🔗 下载地址：https://www.elastic.co/downloads/beats/filebeat

2. 解压安装包

tar -xzf filebeat-8.2.2-linux-x86_64.tar.gz -C /data

建议创建软链接便于管理：

ln -s /data/filebeat-8.2.2-linux-x86_64 /data/filebeat

3. 创建配置目录

mkdir -p /data/filebeat/config

四、实战案例：典型配置场景

📁 所有配置文件建议存放于 /data/filebeat/config 目录

案例 1：标准输入 → 控制台输出（调试用）

用途：快速验证 Filebeat 是否正常工作。

# config/input-stdin.yaml
# 指定filebeat的数据源为stdin
filebeat.inputs:
  # 定义一个输入源列表（支持多个输入）
  - type: stdin        # 输入类型：标准输入（即键盘输入）
    enabled: true      # 是否启用此输入源：true 表示启用，false 表示禁用

# 指定filebeat的输出源console
output.console:
  pretty: true         # 是否格式化输出：true 表示以美观、易读的 JSON 格式打印日志
                      # 包含换行和缩进，便于调试查看结构
                      # 若为 false，则输出紧凑的单行 JSON

启动命令：

/data/filebeat/filebeat -e -c config/input-stdin.yaml

💡 输入任意文本，观察输出的 JSON 结构，重点关注 message 字段。

案例 2：采集普通日志文件input插件之log

用途：采集单行日志，如应用日志、系统日志。

# config/input-log.yaml
# 指定filebeat的数据源为log
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/myapp/app.log

# 指定filebeat的输出源console
output.console:
  pretty: true

🔍 工作机制：

Filebeat 按行读取日志

通过 registry 文件记录采集位置（inode + offset）

默认路径：/data/filebeat/data/registry/filebeat/log.json,该文件中记录了：采集的文件名称，inode，偏移量(offset)以及采集的时间戳等。

重启后自动从中断处继续采集

📌 示例数据

# 创建目录
sudo mkdir -p /var/log/myapp

# 写入示例日志
tee /var/log/myapp/app.log > /dev/null << 'EOF'
2025-08-04 10:00:00 INFO  [web-server] User login attempt from IP=192.168.1.100, UID=usr-78321
2025-08-04 10:00:02 DEBUG [auth-service] Validating token for user: alice@example.***
2025-08-04 10:00:03 INFO  [web-server] Login su***essful for user=alice@example.***, duration=125ms
2025-08-04 10:01:10 WARN  [payment-gateway] Payment timeout for order=ord-99213, retrying...
2025-08-04 10:01:15 ERROR [db-connector] Failed to query user profile: connection refused, host=db-primary, error_code=5003
2025-08-04 10:02:00 INFO  [cache-service] Redis connection restored, resuming operations
2025-08-04 10:02:30 DEBUG [api-gateway] Request received: method=GET, path=/api/v1/user/profile, from=192.168.1.105
2025-08-04 10:02:31 INFO  [api-gateway] Response sent: status=200, duration=15ms
2025-08-04 10:03:00 WARN  [file-uploader] Upload size exceeds limit: file=report.pdf, size=52MB, limit=50MB
2025-08-04 10:03:45 ERROR [email-service] SMTP server unreachable: host=mail.***pany.***, port=587, retry=2
EOF

案例 3：多行日志合并（按行数）count类型案例

用途：适用于固定格式的多行日志（如每条日志占 4 行）。

# config/multiline-count.yaml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /tmp/data.log
    # 指定多行匹配模式
  multiline:
    # 指定多行匹配的类型
    type: count
    count_lines: 4  # 每 4 行合并为一个事件

# 指定filebeat的输出源console
output.console:
  pretty: true

📌 示例数据 /tmp/data.log：

[INFO] User login attempt
User: alice
IP: 192.168.1.100
Time: 2025-08-04T10:00:00Z

[ERROR] Database connection failed
Service: db-service
Error: timeout
Retry: 3

案例 4：多行日志合并（正则匹配）pattern类型案例

用途：识别日志起始行，适用于 JSON、异常栈等结构化日志。

# config/multiline-pattern.yaml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /tmp/app-logs.json
  multiline:
    type: pattern
    pattern: ^\{                # 匹配以 "{" 开头的行
    negate: true                # 非起始行视为延续
    match: after                # 将后续行附加到上一行
    
# 指定filebeat的输出源console
output.console:
  pretty: true

📌 示例数据 /tmp/app-logs.json：

{
  "level": "INFO",
  "msg": "Application started",
  "timestamp": "2025-08-04T10:00:00Z"
}
{
  "level": "ERROR",
  "msg": "Failed to connect",
  "error": "connection timeout",
  "service": "auth-service"
}

案例 5：添加标签与自定义字段

用途：为日志添加上下文信息，便于后续过滤和分析。

# config/enrich-fields.yaml
# Filebeat 配置：演示如何为采集的日志添加标签（tags）和自定义字段（fields）
# 并使用处理器（processors）过滤不需要的日志

# 定义日志输入源列表
filebeat.inputs:
  - type: log                # 输入类型：从日志文件读取
    enabled: true            # 是否启用此输入源（true=启用，false=禁用）

    # 为该输入源添加标签，用于后续分类或路由
    tags: ["web", "production"]

    # 自定义字段：添加业务上下文信息
    fields:
      app: "user-service"    # 应用名称
      env: "prod"            # 环境：生产环境
      team: "backend"        # 负责团队

    # 控制自定义字段是否提升到 JSON 根层级
    # false：字段放在 fields 对象下，如：{"fields": {"app": "user-service", ...}}
    # true：字段直接放在根层级，如：{"app": "user-service", ...}
    fields_under_root: false

    # 指定要采集的日志文件路径（支持通配符）
    paths:
      - /var/log/user-service/*.log

  # 第二个输入源：采集数据库错误日志
  - type: log
    enabled: true
    tags: ["db", "critical"]           # 标签：数据库、关键级别

    fields:
      database: "mysql"                # 数据库类型
      role: "master"                   # 角色：主库

    # 将自定义字段提升到根层级，便于在 Kibana 中直接查询
    fields_under_root: true

    paths:
      - /var/log/mysql/error.log      # 采集 MySQL 错误日志

# 处理器（processors）：在发送前对日志事件进行处理
processors:

  # drop_event.when：条件性丢弃事件
  - drop_event.when:
      # 使用正则表达式匹配 message 字段
      regexp:
        # 如果 message 字段以 "DEBUG:" 开头，则丢弃该日志
        message: "^DEBUG:"

    # 注：此处理器会作用于所有输入源的日志
    # 即：无论是 user-service 还是 mysql 的日志，只要以 DEBUG: 开头都会被过滤掉
# 指定filebeat的输出源console
output.console:
  pretty: true

📄 日志数据示例

1. `/var/log/user-service/a***ess.log` 示例内容

2025-08-04 10:00:00 INFO  User login: id=usr-1001, ip=192.168.1.100
2025-08-04 10:00:05 DEBUG Request processed: path=/api/profile, duration=12ms
2025-08-04 10:00:10 WARN  Session expired for user: usr-1001
2025-08-04 10:00:15 ERROR Failed to update profile: DB connection timeout
DEBUG: Entering function validate_token()
2025-08-04 10:00:20 INFO  Password changed su***essfully for usr-1001

⚠️ 注意：以 DEBUG: 开头的日志将被 drop_event 处理器过滤掉，不会发送出去。

2. `/var/log/mysql/error.log` 示例内容

2025-08-04 10:00:01 [ERROR] Failed to connect to primary: host=db-master, error=Connection refused
2025-08-04 10:00:30 [Warning] Slow query detected: SELECT * FROM users WHERE active=1, duration=2.3s
DEBUG: Executing query plan for JOIN operation
2025-08-04 10:01:00 [ERROR] InnoDB: Unable to allocate memory for buffer pool
2025-08-04 10:01:15 [Note] Slave I/O thread: Connected to master

⚠️ 同样，DEBUG: 开头的日志会被自动丢弃。

🧪 Filebeat 输出示例（控制台）

假设你配置了 output.console: { pretty: true }，运行后你会看到类似以下结构：

示例 1：来自 user-service 的日志（fields 在子对象中）

{
  "@timestamp": "2025-08-04T10:00:00.000Z",
  "message": "2025-08-04 10:00:00 INFO  User login: id=usr-1001, ip=192.168.1.100",
  "tags": ["web", "production"],
  "fields": {
    "app": "user-service",
    "env": "prod",
    "team": "backend"
  },
  "log": { "file": { "path": "/var/log/user-service/a***ess.log" } },
  "input": { "type": "log" }
}

示例 2：来自 MySQL 的日志（fields 提升到根层级）

{
  "@timestamp": "2025-08-04T10:00:01.000Z",
  "message": "2025-08-04 10:00:01 [ERROR] Failed to connect to primary: host=db-master, error=Connection refused",
  "tags": ["db", "critical"],
  "database": "mysql",
  "role": "master",
  "log": { "file": { "path": "/var/log/mysql/error.log" } },
  "input": { "type": "log" }
}

✅ 总结：这个配置的作用

功能	说明
多输入源	同时采集应用日志和数据库日志
打标签（tags）	便于在 Kibana 中按 `web`、`db`、`critical` 等分类
加字段（fields）	添加 `app`、`env`、`database` 等元数据
字段层级控制	`fields_under_root` 决定字段是否在根层级
日志过滤	使用 `drop_event` 删除调试日志，减少无效数据传输

参考案例:

https://www.elastic.co/guide/en/beats/filebeat/8.2/filebeat-input-log.html#filebeat-input-log-***mon-options
https://www.elastic.co/guide/en/beats/filebeat/8.2/filtering-and-enhancing-data.html

案例 6：解析嵌套 JSON 日志（字符串内 JSON）

用途：将日志行中某个字段（如 payload）包含的 JSON 字符串 解析为结构化字段。

🔍 常见场景：API 网关、消息队列日志、审计日志中常出现 "data": "{\"key\": \"value\"}" 这类嵌套结构。

# config/json-parse.yaml
# Filebeat 配置：解析日志中嵌套的 JSON 字符串
# 作者：You are Qwen, created by Alibaba Cloud. You are a helpful assistant.

filebeat.inputs:
  # 定义日志输入源
  - type: log                # 输入类型：从文件读取日志
    enabled: true            # 启用此输入

    # 指定要采集的日志文件路径
    paths:
      - /tmp/nested-data.log  # 示例路径，需确保文件存在

    # 配置 JSON 解析规则（适用于整个日志行是 JSON 的情况）
    json.keys_under_root: true    # 将解析出的字段提升到 JSON 根层级
                                 # 例如：payload 中的 method、path 直接成为顶级字段
    json.add_error_key: true      # 如果 JSON 解析失败，添加 "error.message" 字段标记错误
                                 # 便于排查问题，不会导致日志丢失

# 指定输出目标为控制台（调试用）
output.console:
  pretty: true                  # 格式化输出 JSON，便于阅读

📄 示例日志文件：`/tmp/nested-data.log`

{ "source": "api-gateway", "timestamp": "2025-08-04T10:00:00Z", "payload": "{\"method\": \"POST\", \"path\": \"/login\", \"status\": 200, \"user_id\": \"usr-1001\"}" }
{ "source": "message-queue", "timestamp": "2025-08-04T10:00:05Z", "payload": "{\"event\": \"order_created\", \"order_id\": \"ord-88123\", \"amount\": 99.99}" }
{ "source": "audit-log", "timestamp": "2025-08-04T10:00:10Z", "payload": "{\"action\": \"delete\", \"target\": \"file-77654\", \"by\": \"admin\"}" }
{ "source": "api-gateway", "timestamp": "2025-08-04T10:00:15Z", "payload": "INVALID_JSON_HERE" }

⚠️ 最后一行是故意构造的解析失败日志，用于演示 add_error_key: true 的作用。

🧪 预期输出（控制台）

正常解析的示例：

{
  "@timestamp": "2025-08-04T10:00:00.000Z",
  "source": "api-gateway",
  "timestamp": "2025-08-04T10:00:00Z",
  "payload": "{\"method\": \"POST\", \"path\": \"/login\", \"status\": 200, \"user_id\": \"usr-1001\"}",
  "method": "POST",
  "path": "/login",
  "status": 200,
  "user_id": "usr-1001",
  "log": { "file": { "path": "/tmp/nested-data.log" } },
  "input": { "type": "log" }
}

✅ method, path, status, user_id 原本在 payload 字符串中，现在被提取为顶级字段！

解析失败的示例（自动添加 error）：

{
  "@timestamp": "2025-08-04T10:00:15.000Z",
  "source": "api-gateway",
  "timestamp": "2025-08-04T10:00:15Z",
  "payload": "INVALID_JSON_HERE",
  "error": {
    "message": "Failed to decode JSON: invalid character 'I' looking for beginning of value"
  },
  "log": { "file": { "path": "/tmp/nested-data.log" } }
}

✅ 因为 add_error_key: true，即使解析失败，日志也不会丢，而是带错误信息继续传输。

案例 7：多行 JSON 日志解析（换行分隔的 JSON 对象）

用途：采集每行一个 JSON 对象但被系统换行截断的日志，先合并多行，再解析为结构化数据。

🔍 常见场景：Java 应用打印 JSON 日志时自动换行、Docker 容器日志截断等。

# config/multiline-json.yaml
# Filebeat 配置：先合并多行 JSON，再使用处理器解析
# 适用于日志被截断、跨行的 JSON 日志
# 作者：You are Qwen, created by Alibaba Cloud. You are a helpful assistant.

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /tmp/service-logs.json   # 日志路径

    # 多行合并配置：识别 JSON 起始行
    multiline:
      type: pattern              # 使用正则模式匹配
      pattern: ^\{               # 匹配以左大括号 "{" 开头的行（即 JSON 起始）
      negate: true               # 否定逻辑：如果某行**不匹配** pattern，则视为前一行的延续
      match: after               # 将不匹配的行附加到上一行末尾

    # 使用 processors 进行 JSON 解析（更灵活，推荐方式）
    processors:
      - decode_json_fields:     # 处理器：解析 JSON 字段
          fields: ["message"]    # 指定要解析的字段名（Filebeat 默认把整行日志放入 message）
          process_array: false   # 不处理数组类型（本例为单个对象）
          max_depth: 3           # 最大嵌套深度为 3 层
          target: ""             # 解析结果放入根层级（等同于 keys_under_root）
          overwrite_keys: false  # 如果字段已存在，不覆盖（避免冲突）

# 输出到控制台，便于调试
output.console:
  pretty: true

📄 示例日志文件：`/tmp/service-logs.json`

{
  "level": "INFO",
  "service": "auth-service",
  "event": "user_login",
  "user": "alice",
  "ip": "192.168.1.100",
  "timestamp": "2025-08-04T10:00:00Z"
}
{
  "level": "ERROR",
  "service": "payment-service",
  "event": "transaction_failed",
  "order_id": "ord-99213",
  "error": "timeout",
  "duration_ms": 5000,
  "retry_count": 3,
  "timestamp": "2025-08-04T10:01:15Z"
}
{
  "level": "WARN",
  "service": "cache-service",
  "event": "connection_restored",
  "details": {
    "host": "redis-primary",
    "reconnect_time_ms": 120
  },
  "timestamp": "2025-08-04T10:02:00Z"
}

💡 注意：这个文件在实际环境中可能被日志系统（如 Docker）按行截断，导致每个 JSON 被拆成多行。

🧪 预期输出（控制台）

{
  "@timestamp": "2025-08-04T10:00:00.000Z",
  "level": "INFO",
  "service": "auth-service",
  "event": "user_login",
  "user": "alice",
  "ip": "192.168.1.100",
  "timestamp": "2025-08-04T10:00:00Z",
  "log": { "file": { "path": "/tmp/service-logs.json" } },
  "input": { "type": "log" }
}

✅ 完整的 JSON 对象被成功还原并结构化！

当然可以！以下是 案例 8：使用 filestream 的完整增强版本，包含：

✅ 逐行中文注释
✅ 补充日志示例（NDJSON 格式）
✅ 输出效果说明
✅ 与传统 log 类型的对比
✅ 安全脱敏，适合用于文档、培训或生产参考

案例 8：使用 `filestream`（推荐方式）

用途：filestream 是 log 输入类型的现代化替代，专为高效、稳定地读取日志文件而设计，支持内置解析器（如 NDJSON、CSV），是 Filebeat 7.13+ 推荐的日志采集方式。

# config/input-filestream.yaml
# Filebeat 配置：使用 filestream 输入类型采集结构化日志（如 NDJSON）
# 作者：You are Qwen, created by Alibaba Cloud. You are a helpful assistant.

filebeat.inputs:
  # 定义 filestream 输入源（推荐替代 type: log）
  - type: filestream
    enabled: true                    # 是否启用此输入

    # 指定要监控的日志文件路径（支持通配符）
    paths:
      - /var/log/nginx/a***ess.log    # 示例：Nginx 访问日志（NDJSON 格式）
      # - /var/log/app/*.ndjson     # 也可匹配多个文件

    # parsers：filestream 内置的解析器，用于预处理日志行
    parsers:
      - ndjson:                      # 解析换行分隔的 JSON（Newline-delimited JSON）
          overwrite_keys: true       # 如果解析出的字段与现有字段冲突，允许覆盖
          target: ""                 # 将解析结果提升到根层级（等同于 keys_under_root: true）
          add_error_key: true        # 解析失败时添加 error.message 字段，不丢日志
          message_key: message       # 指定哪个字段作为日志消息体（可选，默认为 message）

# 可选：额外的处理器（processors），在 parsers 之后执行
processors:
  # 如果 parsers.ndjson 未完全解析，可再用 decode_json_fields 增强
  - decode_json_fields:
      fields: ["message"]            # 解析 message 字段中的 JSON
      max_depth: 2                   # 最大嵌套深度为 2 层
      target: ""                     # 提升到根层级
      overwrite_keys: false          # 避免覆盖已存在的字段
      process_array: false           # 不处理数组类型

  # 示例：添加时间戳字段（如果日志中无时间）
  # - add_fields:
  #     target: ''
  #     fields:
  #       event_source: "nginx-a***ess"

# 输出到控制台（调试用）
output.console:
  pretty: true                       # 格式化输出 JSON，便于阅读

📄 示例日志文件：`/var/log/nginx/a***ess.log`（NDJSON 格式）

📌 NDJSON（Newline-Delimited JSON）是一种常见结构化日志格式，每行一个独立 JSON 对象。

{"time":"2025-08-04T10:00:00Z","client_ip":"192.168.1.100","method":"GET","path":"/index.html","status":200,"duration_ms":45,"user_agent":"Mozilla/5.0"}
{"time":"2025-08-04T10:00:02Z","client_ip":"192.168.1.105","method":"POST","path":"/api/login","status":401,"duration_ms":120,"user_agent":"PostmanRuntime/7.29"}
{"time":"2025-08-04T10:00:05Z","client_ip":"192.168.1.110","method":"GET","path":"/static/app.js","status":200,"duration_ms":10,"user_agent":"Safari/15.4"}
{"time":"2025-08-04T10:00:08Z","client_ip":"malicious-ip","method":"GET","path":"/admin","status":403,"duration_ms":5,"user_agent":"Nmap"}
{"invalid_json": "missing_brace"    # 故意构造的错误行，测试 add_error_key

💡 提示：Nginx 可通过 log_format 配置输出 JSON 或 NDJSON 格式日志。

🧪 预期输出（控制台）

✅ 正常解析的示例：

{
  "@timestamp": "2025-08-04T10:00:00.000Z",
  "time": "2025-08-04T10:00:00Z",
  "client_ip": "192.168.1.100",
  "method": "GET",
  "path": "/index.html",
  "status": 200,
  "duration_ms": 45,
  "user_agent": "Mozilla/5.0",
  "log": {
    "file": {
      "path": "/var/log/nginx/a***ess.log"
    },
    "offset": 0
  },
  "input": {
    "type": "filestream"
  },
  "agent": { ... },
  "host": { ... }
}

✅ 所有字段已结构化，可直接在 Kibana 中查询、过滤、可视化！

❌ 解析失败的示例（add_error_key: true 生效）：

{
  "@timestamp": "2025-08-04T10:00:08.000Z",
  "message": "{\"invalid_json\": \"missing_brace\"",
  "error": {
    "message": "Failed to decode NDJSON: unexpected end of JSON input"
  },
  "log": { "file": { "path": "/var/log/nginx/a***ess.log" } }
}

✅ 日志未丢失，仅标记错误，便于后续排查。

🔍 `filestream` 优势详解（vs `log`）

特性	`type: log`（旧）	`type: filestream`（新，推荐）
性能	一般	更高（优化的文件句柄管理）
稳定性	较好	更好（支持文件旋转、重命名更可靠）
内置解析器	无	✅ 支持 `ndjson`, `csv`, `multiline` 等
配置方式	使用 `json.*` 或 `processors`	使用 `parsers` + `processors`，逻辑更清晰
生命周期管理	基础	支持 `close_*` 规则（如按时间、大小关闭）
推荐程度	❌ 不推荐新项目使用	✅ 官方推荐，未来发展方向

🛠️ 可选高级配置（按需添加）

  # 文件关闭策略（可选）
  close:
    inactivity: 5m        # 文件 5 分钟无变化则关闭
    reader:               # 控制读取行为
      max_bytes: 1048576  # 每次最多读取 1MB

  # 排除临时文件
  exclude_files: ['\.tmp$', '\.log\.part$']

案例 9：Filebeat 采集 Nginx 日志

一、安装并配置 Nginx

1. 安装 Nginx

yum -y install nginx

✅ 说明：安装最新版 Nginx（CentOS/RHEL 系统），用于生成访问日志。

2. 修改 Nginx 配置，启用 JSON 格式日志

编辑配置文件：

vim /etc/nginx/nginx.conf

在 http { } 块中添加 JSON 日志格式定义：

log_format nginx_json
  '{'
    '"@timestamp":"$time_iso8601",'
    '"host":"$server_addr",'
    '"clientip":"$remote_addr",'
    '"SendBytes":$body_bytes_sent,'
    '"responsetime":$request_time,'
    '"upstreamtime":"$upstream_response_time",'
    '"upstreamhost":"$upstream_addr",'
    '"http_host":"$host",'
    '"uri":"$uri",'
    '"domain":"$host",'
    '"xff":"$http_x_forwarded_for",'
    '"referer":"$http_referer",'
    '"tcp_xff":"$proxy_protocol_addr",'
    '"http_user_agent":"$http_user_agent",'
    '"status":"$status"'
  '}';

🔍 注解：

log_format：定义名为 nginx_json 的日志格式

使用双引号和转义，确保输出为合法 JSON 字符串

$variable 是 Nginx 内置变量，代表请求的各个字段

每个请求将输出一行 JSON，便于后续解析

设置访问日志使用该格式

在 server { } 块中修改或添加：

a***ess_log /var/log/nginx/a***ess.log nginx_json;

✅ 说明：指定日志路径和格式为 nginx_json

3. 检查 Nginx 配置语法

nginx -t

✅ 正常输出应为：

nginx: the configuration file /etc/nginx/nginx.conf syntax is ok
nginx: configuration file /etc/nginx/nginx.conf test is su***essful

4. 启动 Nginx 服务

systemctl enable --now nginx

✅ 说明：开机自启并立即启动 Nginx

5. 生成测试日志

curl http://localhost/
curl http://localhost/404

查看日志是否生成：

tail -f /var/log/nginx/a***ess.log

示例输出（一行 JSON）：

{"@timestamp":"2025-08-05T14:19:31+08:00","host":"127.0.0.1","clientip":"127.0.0.1","SendBytes":615,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"localhost","uri":"/index.html","domain":"localhost","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"curl/7.29.0","status":"200"}

二、编写 Filebeat 配置文件采集 Nginx 日志

创建配置文件

vim config/input-log-to-console.yaml

配置内容（带逐行注释）

# config/input-log-to-console.yaml
# Filebeat 配置：采集 Nginx JSON 格式访问日志，并输出到控制台
# 定义输入源
filebeat.inputs:
  - type: log                    # 输入类型：从日志文件读取
    enabled: true                # 启用此输入
    paths:
      - /var/log/nginx/a***ess.log  # 采集 a***ess.log 及轮转文件（如 a***ess.log.1）

    # 配置 JSON 解析规则
    json.keys_under_root: true   # 将 JSON 字段提升到根层级（如 status、clientip 直接可用）
    json.add_error_key: true     # 解析失败时添加 error.message 字段，不丢日志
    json.overwrite_keys: false   # 如果字段已存在（如 @timestamp），不覆盖原值

# 输出到控制台（调试用）
output.console:
  pretty: true                   # 格式化输出 JSON，便于阅读和调试

✅ 说明：

json.keys_under_root: true 是关键，否则所有字段会嵌套在 json.* 下

支持日志轮转（a***ess.log*）

使用 add_error_key 可避免因单条日志格式错误导致整个采集失败

三、启动 Filebeat 实例

./filebeat -e -c config/input-log-to-console.yaml

参数说明：

-e：将 Filebeat 自身日志输出到 stderr，便于调试

-c：指定配置文件路径

四、使用 `filestream` 采集（推荐方式）

⚠️ type: log 已逐步被 filestream 替代，推荐新项目使用。

# config/filestream-nginx-to-console.yaml
filebeat.inputs:
  - type: filestream
    enabled: true
    paths:
      - /var/log/nginx/a***ess.log

    # 使用 NDJSON 解析器（每行一个 JSON 对象）
    parsers:
      - ndjson:
          overwrite_keys: true
          target: ""
          add_error_key: true

output.console:
  pretty: true

✅ filestream + ndjson 是更现代、更稳定的组合。

案例 10：使用 Filebeat 采集 Tomcat 结构化日志

一、下载并安装 Tomcat

1. 下载二进制包（推荐 Apache 官方源）

wget https://dlcdn.apache.org/tomcat/tomcat-9/v9.0.107/bin/apache-tomcat-9.0.107.tar.gz

🔍 提示：确保服务器已安装 wget 和网络可达。

2. 解压到统一软件目录

# 创建目录（建议使用 /data 或 /opt）
mkdir -p /data/softwares

# 解压
tar xf apache-tomcat-9.0.107.tar.gz -C /data/softwares/

# 创建软链接便于管理
ln -s /data/softwares/apache-tomcat-9.0.107 /data/tomcat

✅ 好处：升级时只需修改软链接指向新版本。

二、配置 JDK 环境（Tomcat 依赖）

1. 确保已安装 JDK 17（Tomcat 9+ 推荐）

# 示例路径，请根据实际 JDK 安装位置调整
export JAVA_HOME=/usr/lib/jvm/jdk-17.0.0.1
export PATH=$JAVA_HOME/bin:$PATH

2. 写入环境变量文件（永久生效）

cat > /etc/profile.d/java.sh << 'EOF'
export JAVA_HOME=/usr/lib/jvm/jdk-17.0.0.1
export PATH=$JAVA_HOME/bin:$PATH
EOF

3. 加载环境变量

source /etc/profile.d/java.sh

4. 验证 Java 是否可用

java -version

✅ 正常输出应包含 openjdk version "17" 或类似信息。

三、配置 Tomcat 输出 JSON 格式访问日志

🔍 默认日志为文本格式，不利于结构化分析。我们通过 A***essLogValve 配置为 NDJSON（Newline-delimited JSON） 格式。

编辑 server.xml

vim /data/tomcat/conf/server.xml

在 <Host name="localhost"...> 标签内添加以下 Valve 配置：

💡 通常位于文件约 133–149 行之间，可搜索 <Host 定位。

<Valve className="org.apache.catalina.valves.A***essLogValve"
       directory="logs"
       prefix="a***ess_log"
       suffix=".json"
       fileDateFormat=".yyyy-MM-dd"
       pattern="{&quot;time&quot;:&quot;%t&quot;,&quot;ip&quot;:&quot;%a&quot;,&quot;method&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;user_agent&quot;:&quot;%{User-Agent}i&quot;,&quot;bytes_sent&quot;:&quot;%B&quot;,&quot;request_time&quot;:&quot;%D&quot;}"
       resolveHosts="false"
       renameOnRotate="true" />

🔍 参数说明：

directory="logs"：日志输出到 logs/ 目录

prefix="a***ess_log."：文件前缀，如 a***ess_log.2025-08-05.json

suffix=".json"：后缀为 .json，便于识别

pattern：定义 JSON 结构，字段需对双引号转义为 "

%t：时间戳

%a：客户端 IP

%r：请求方法 + URI + 协议

%s：HTTP 状态码

%{User-Agent}i：User-Agent 头

%B：发送字节数

%D：请求处理时间（毫秒）

renameOnRotate="true"：每日轮转时重命名旧文件

四、启动 Tomcat 服务

1. 启动 Tomcat

/data/tomcat/bin/startup.sh

✅ 输出：Tomcat started.

2. 验证是否监听 8080 端口

ss -tlnp | grep 8080

3. 测试访问并生成日志

curl http://127.0.0.1:8080/
curl http://127.0.0.1:8080/404

4. 查看 JSON 日志是否生成

tail -f /data/tomcat/logs/a***ess_log.*.json

示例输出：

{"time":"[05/Aug/2025:14:59:11 +0800]","ip":"0:0:0:0:0:0:0:1","method":"GET / HTTP/1.1","status":"200","user_agent":"curl/7.29.0","bytes_sent":"11232","request_time":"235"}
{"time":"[05/Aug/2025:14:59:12 +0800]","ip":"0:0:0:0:0:0:0:1","method":"GET /404 HTTP/1.1","status":"404","user_agent":"curl/7.29.0","bytes_sent":"755","request_time":"8"}

✅ 成功！每行一个 JSON 对象（NDJSON 格式）。

五、编写 Filebeat 配置文件采集 Tomcat 日志

创建配置文件

vim config/tomcat-filestream-console.yaml

配置内容

# config/tomcat-filestream-console.yaml
# Filebeat 配置：采集 Tomcat JSON 格式访问日志。

filebeat.inputs:
  - type: filestream
    enabled: true
    # 采集所有 a***ess_log 开头的 JSON 文件
    paths:
      - /data/tomcat/logs/a***ess_log*.json
      # 可选：采集应用日志（如 catalina.out）
      # - /data/tomcat/logs/catalina.out

    # 内置解析器：自动解析每行为一个 JSON 对象（NDJSON）
    parsers:
      - ndjson:
          # 将解析后的字段提升到根层级
          target: ""
          # 允许覆盖已有字段（如 message）
          overwrite_keys: true
          # 解析失败时添加 error.message 字段
          add_error_key: true

    # 添加自定义字段，标识日志来源
    fields:
      app: "tomcat"
      log_type: "a***ess"
      service: "web-application"
    # false的话 不将 fields 写入根层级，避免污染
    fields_under_root: true

# 处理器链：进一步处理 message 字段（如果 parsers 未完全解析）
processors:
  - decode_json_fields:
      fields: ["message"]        # 解析 message 中的 JSON
      process_array: false
      max_depth: 2
      target: ""                 # 提升到根
      overwrite_keys: false      # 避免重复覆盖

  # 示例：移除原始 message 字段（可选，减少冗余）
  # - drop_fields:
  #     fields: ["message"]

# 输出到控制台（调试用）
output.console:
  pretty: true                   # 格式化输出，便于阅读

六、启动 Filebeat 实例

./filebeat -e -c config/tomcat-filestream-console.yaml

参数说明：

-e：输出 Filebeat 自身日志到 stderr

-c：指定配置文件

案例11：Filebeat 采集 containerd 容器类型日志使用 `dissect` 处理器

1. 背景说明

在使用 containerd 作为容器运行时的环境中，容器日志通常以 JSON 格式存储在宿主机的特定目录下。Filebeat 支持通过 container 输入类型直接采集这些日志文件，适用于 Kuber***es 或直接使用 ctr 命令运行容器的场景。

2. 环境准备

Filebeat 版本：8.2.2
容器运行时：containerd（Docker 使用 containerd 作为底层运行时）
日志路径：/var/lib/docker/containers/*/*.log

⚠️ 确保 Filebeat 有权限读取容器日志目录。

3. 配置 Filebeat

创建配置文件：

vim config/container-console.yaml

内容如下：

filebeat.inputs:
  - type: container
    paths:
      - '/var/lib/docker/containers/*/*.log'
    # 可选：只采集标准输出
    stream: stdout
    # 可选：排除某些容器
    # exclude_containers: ['.*redis.*', '.*nginx.*']
processors:
  - dissect:
      tokenizer: "%{clientip} - %{ident} [%{timestamp}] \"%{verb} %{request} HTTP/%{httpversion}\" %{response} %{bytes} \"%{referrer}\" \"%{agent}\""
      field: "message"
      target_prefix: "nginx.a***ess"
      ignore_failure: true
output.console:
  pretty: true

4. 启动 Filebeat

./filebeat -e -c config/container-console.yaml

案例12：Filebeat 采集 Redis 慢查询日志（实验性功能）

1. 背景说明

Filebeat 提供实验性 redis 输入类型，用于从 Redis 实例的慢查询日志（slow log）中提取性能数据，适用于监控高延迟命令。

⚠️ 此功能为实验性（experimental），不建议用于生产环境。

2. 启动 Redis 实例

docker run --name redis-server -d -p 6379:6379 redis:7.2.5-alpine3.20

3. 配置 Redis 慢查询日志

进入容器，设置慢查询阈值（单位：微秒）：

docker exec -it redis-server redis-cli
127.0.0.1:6379> CONFIG SET slowlog-log-slower-than 1000
OK

4. 写入测试数据并触发慢日志

127.0.0.1:6379> SET app_config '{"version": "v1.0", "timeout": 30}'
OK
127.0.0.1:6379> LPUSH request_queue "task_001" "task_002" "task_003"
(integer) 3

5. 配置 Filebeat

创建配置文件：

vim config/redis-console.yaml

filebeat.inputs:
  - type: redis
    hosts: ["172.17.0.3:6379"]  # Redis 容器 IP
    # 可选：设置密码
    # password: "your_password"
    # 可选：设置数据库
    # db: 0

output.console:
  pretty: true

查看容器IP

docker inspect 58c32cfd9c27 | grep -i "ipaddress"
            "SecondaryIPAddresses": null,
            "IPAddress": "172.17.0.3",
                    "IPAddress": "172.17.0.3",

6. 启动 Filebeat

./filebeat -e -c config/12-redis-to-console.yaml

7. 验证输出

控制台将输出 Redis 慢查询日志条目，例如：

{
  "@timestamp": "2025-08-05T10:05:00.000Z",
  "redis": {
    "slowlog": {
      "id": 1,
      "duration": 1500,
      "client_ip": "172.17.0.1",
      "***mand": "LPUSH request_queue task_001 task_002 task_003"
    }
  },
  "agent": {
    "type": "filebeat",
    "hostname": "monitor-host"
  }
}

案例13：Filebeat 采集 TCP 日志并输出到 Redis

1. 背景说明

在分布式系统中，某些应用可能通过 TCP 协议直接发送日志。Filebeat 可作为 TCP 服务器接收日志，并将结构化数据写入 Redis 缓冲区，供后续 Logstash 或消费者处理。

2. 启动 Redis 服务

docker run -d --name redis-server --***work host redis:7.2.5-alpine3.20

验证端口监听：

ss -ntl | grep 6379

3. 配置 Filebeat

创建配置文件：

vim config/tcp-to-redis.yaml

filebeat.inputs:
  - type: tcp
    host: "0.0.0.0:8888"
    # 可选：设置最大连接数
    # max_connections: 1024
    # 可选：设置超时
    # timeout: 30s

output.redis:
  hosts: ["192.168.130.62:6379"]   # ✅ 端口写在 host 后面
  key: "app-logs-ingest"
  db: 3
  timeout: 5
  # 可选：启用 SSL
  # ssl.enabled: true

4. 启动 Filebeat

./filebeat -e -c config/tcp-to-redis.yaml

5. 发送测试日志

从另一台机器发送日志：

echo "Application event: user_login su***ess from 192.168.130.62" | nc 192.168.130.62 8888

6. Redis 验证数据

进入 Redis 查看数据：

docker exec -it redis-server redis-cli -n 3
127.0.0.1:6379[3]> KEYS *
1) "app-logs-ingest"

127.0.0.1:6379[3]> TYPE app-logs-ingest
list

127.0.0.1:6379[3]> LRANGE app-logs-ingest 0 -1
1) "{\"@timestamp\":\"2025-08-06T09:44:40.032Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"8.2.2\"},\"host\":{\"name\":\"es-node-02\"},\"agent\":{\"version\":\"8.2.2\",\"ephemeral_id\":\"6f41ee94-a3f4-493b-ab5b-cad7498394ce\",\"id\":\"f11283a2-e3df-4***f-99cb-ef1fac593cfa\",\"name\":\"es-node-02\",\"type\":\"filebeat\"},\"message\":\"Application event: user_login su***ess from 192.168.130.62\",\"log\":{\"source\":{\"address\":\"192.168.130.61:37942\"}},\"input\":{\"type\":\"tcp\"},\"ecs\":{\"version\":\"8.0.0\"}}"

好的！以下是根据你的 实际环境（IP: 192.168.130.61/62/63） 重新整理、优化并标准化的 三大经典 Filebeat 日志采集案例，适用于生产级 EFK 架构。

案例14：采集 TCP 日志 → 输出到本地文件

用途：调试、备份、审计等场景

📁 配置文件：`config/tcp-to-file.yaml`

filebeat.inputs:
  - type: tcp
    host: "0.0.0.0:8888"
    max_connections: 1024
    timeout: 30s

output.file:
  path: "/data/logs/filebeat-output"         # 建议使用独立目录
  filename: app-logs.txt
  rotate_every_kb: 10240                     # 每 10MB 滚动一次
  number_of_files: 7                         # 最多保留 7 个文件
  permissions: 0644                          # 定义文件的权限

# 启用日志输出方便调试
logging.level: info

🧪 创建输出目录并启动

mkdir -p /data/logs/filebeat-output
./filebeat -e -c config/tcp-to-file.yaml

📦 发送测试数据（从任意节点）

echo "用户登录成功: zhangsan @ $(date)" | nc 192.168.130.62 8888

🔍 查看结果

cat /data/logs/filebeat-output/* |jq

输出示例：

{
  "@timestamp": "2025-08-06T09:57:13.723Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.2.2"
  },
  "agent": {
    "ephemeral_id": "888abfa0-c230-4a6c-92f5-52a8472ac87c",
    "id": "f11283a2-e3df-4***f-99cb-ef1fac593cfa",
    "name": "es-node-02",
    "type": "filebeat",
    "version": "8.2.2"
  },
  "message": "用户登录成功: zhangsan @ Wed Aug  6 17:57:13 CST 2025",
  "log": {
    "source": {
      "address": "192.168.130.62:42754"
    }
  },
  "input": {
    "type": "tcp"
  },
  "ecs": {
    "version": "8.0.0"
  },
  "host": {
    "name": "es-node-02"
  }
}

案例15：采集 TCP 日志 → 输出到 Elasticsearch 集群（自定义索引模板）

用途：结构化日志入 ES，用于 Kibana 展示

📁 配置文件：`config/tcp-to-es.yaml`

# 定义 Filebeat 的输入源（日志来源）
filebeat.inputs:
  # 配置一个 TCP 类型的输入
  - type: tcp
    # 监听本机所有 IP 的 8888 端口，接收外部通过 TCP 发送的日志
    host: "0.0.0.0:8888"

# 配置输出目标为 Elasticsearch
output.elasticsearch:
  # 指定 ES 集群的多个节点地址，实现高可用和负载均衡
  hosts: ["http://192.168.130.61:9200", "http://192.168.130.62:9200", "http://192.168.130.65:9200"]
  # 指定日志写入的索引名称，按天分割，例如：app-tcp-logs-2025.08.07
  index: "app-tcp-logs-%{+yyyy.MM.dd}"
  data_stream.enabled: false   # 强制关闭 Data Stream

# 显式关闭模板自动加载，避免报错
setup.template.enabled: false

手动创建模版

curl -X PUT "http://192.168.130.61:9200/_template/app-tcp-logs" -H "Content-Type: application/json" -d'
{
  "index_patterns": ["app-tcp-logs-*"],
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  }
}'

启动命令

./filebeat -e -c config/tcp-to-es.yaml

发送测试日志

echo "订单创建成功: order_20250807_001" | nc 192.168.130.62 8888

验证 ES 中数据

curl -s 'http://192.168.130.61:9200/app-tcp-logs-*/_search?pretty' | grep -A 5 -B 5 "order_20250807"

✅ 成功写入 ES，可用于 Kibana 可视化。

案例15：Nginx JSON 日志采集 → EFK 架构（推荐生产用法）

架构：Nginx → Filebeat → Elasticsearch → Kibana

第一步：配置 Nginx 输出 JSON 格式日志

📄 修改 Nginx 配置：`/etc/nginx/nginx.conf`

http {
    # 定义 JSON 格式日志
    log_format json_a***ess_log escape=json
      '{'
        '"@timestamp":"$time_iso8601",'
        '"clientip":"$remote_addr",'
        '"remote_user":"$remote_user",'
        '"request":"$request",'
        '"status": "$status",'
        '"body_bytes_sent": "$body_bytes_sent",'
        '"http_referer":"$http_referer",'
        '"http_user_agent":"$http_user_agent",'
        '"http_x_forwarded_for":"$http_x_forwarded_for",'
        '"upstream_response_time":"$upstream_response_time",'
        '"request_time":"$request_time",'
        '"host":"$host"'
      '}';

    # 使用 JSON 格式记录访问日志
    a***ess_log /var/log/nginx/a***ess.log json_a***ess_log;
    error_log /var/log/nginx/error.log;

    server {
        listen 80;
        server_name localhost;
        root /usr/share/nginx/html;
        index index.html;
    }
}

✅ 重载 Nginx

nginx -t && systemctl reload nginx

🧪 访问测试

curl "http://192.168.130.62?from=filebeat"

🔍 检查日志是否为 JSON

tail -1 /var/log/nginx/a***ess.log

输出应为：

{"@timestamp":"2025-08-06T10:10:20+08:00","clientip":"192.168.130.61","remote_user":"-","request":"GET /?from=filebeat HTTP/1.1","status": "200","body_bytes_sent": "612","http_referer":"-","http_user_agent":"curl/7.68.0","http_x_forwarded_for":"-","upstream_response_time":"-","request_time":"0.000","host":"192.168.130.62"}

第二步：Filebeat 采集 Nginx 日志 → ES

📁 配置文件：`config/nginx-to-es.yaml`

filebeat.inputs:
  - type: filestream
    id: nginx-a***ess
    paths:
      - /var/log/nginx/a***ess.log
    # 忽略 24 小时前的日志
    ignore_older: 24h

    # ✅ 正确方式：使用 ndjson 解析整行 JSON 日志
    parsers:
      - ndjson:
          # 整行就是 JSON，不需要嵌套在 message 字段里
          target: ""
          # 因为日志不是 { "message": "{...}" } 这种结构，所以不需要 message_key
          # 删除这一行 → 否则解析失败
          # message_key: message
          overwrite_keys: true  # 允许覆盖 @timestamp 等字段
          add_error_key: false  # 可选：不添加解析错误信息

# 添加主机和 agent 元数据
processors:
  - add_host_metadata: ~
  - add_agent_metadata: ~

# ❌ 删除 decode_json_fields 和 drop_fields
# 因为 ndjson 已经完成了解析，不需要二次处理

output.elasticsearch:
  hosts:
    - "http://192.168.130.61:9200"
    - "http://192.168.130.62:9200"
    - "http://192.168.130.65:9200"
  index: "nginx-a***ess-logs-%{+yyyy.MM.dd}"

  # 可选：设置用户名密码（如果你启用了安全）
  # username: "filebeat_writer"
  # password: "your_password"

# 🔥 关键：关闭自动模板和 ILM
setup.ilm.enabled: false
setup.template.enabled: false   # ✅ 禁用自动模板创建

# ❌ 不要再用 setup.template.* 配置
# 因为 ES 8.x 容易报错，建议手动管理

手动创建模板

PUT http://192.168.130.62:9200/_index_template/nginx-a***ess-logs-template
{
  "index_patterns": ["nginx-a***ess-logs-*"],
  "priority": 100,
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "mapping.total_fields.limit": 2000,
      "refresh_interval": "30s"
    },
    "mappings": {
      "dynamic": true,
      "properties": {
        "@timestamp": { "type": "date" },
        "clientip": { "type": "ip" },
        "remote_user": { "type": "keyword" },
        "request": { "type": "text" },
        "status": { "type": "short" },
        "body_bytes_sent": { "type": "long" },
        "http_referer": { "type": "keyword" },
        "http_user_agent": { "type": "text" },
        "http_x_forwarded_for": { "type": "ip" },
        "upstream_response_time": { "type": "float" },
        "request_time": { "type": "float" },

        "host": {
          "type": "object",
          "properties": {
            "hostname": { "type": "keyword" },
            "ip": { "type": "ip" },
            "name": { "type": "keyword" },
            "architecture": { "type": "keyword" },
            "containerized": { "type": "boolean" },
            "id": { "type": "keyword" }
          }
        },

        "hostname": { "type": "keyword" }
      }
    }
  },
  "_meta": {
    "description": "Nginx JSON log template for ES 8.x"
  }
}

▶️ 启动 Filebeat

./filebeat -e -c config/nginx-to-es.yaml

第三步：Kibana 验证（可选）

登录 Kibana：http://192.168.130.61:5601
进入 Stack Management > Index Patterns
创建索引模式：nginx-a***ess-logs-*
进入 Discover，查看实时日志

案例16：Filebeat 配置：将 JSON 字段提升到顶级

在使用 Filebeat 采集日志时，如果你的日志内容是 JSON 格式，并且你希望将这些 JSON 内容的字段 提升到顶级字段（Top-level Fields），而不是嵌套在 message 或某个子字段中，可以通过 json.keys_under_root 配置项来实现。

✅ 目标

将原始日志中 JSON 内容的字段，直接提升为 Elasticsearch 中的顶级字段。

📦 示例场景

原始日志内容（`/var/log/app.log`）：

{"status":"su***ess","user":"alice","ip":"192.168.1.1","@timestamp":"2025-04-05T12:34:56Z"}
{"status":"fail","user":"bob","ip":"192.168.1.2","@timestamp":"2025-04-05T12:35:01Z"}

✅ Filebeat 配置：将 JSON 字段提升到顶级

在你的 filebeat.yml 配置文件中，添加如下配置：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app.log

  # 启用 JSON 解析
  json.keys_under_root: true
  json.add_error_key: true
  json.message_key: "log"  # 可选，如果你的消息字段名不是默认的 message

output.elasticsearch:
  hosts:
    - "http://192.168.130.61:9200"
    - "http://192.168.130.62:9200"
    - "http://192.168.130.65:9200"
  index: "nginx-%{+yyyy.MM.dd}"

📌 配置说明

配置项	说明
`json.keys_under_root: true`	将 JSON 的键提升到顶级字段
`json.add_error_key: true`	如果 JSON 解析失败，添加 `error.message` 字段说明错误
`json.message_key`	如果你希望保留原始 JSON 内容，可以指定字段名，默认是 `message`

📈 输出效果（Elasticsearch 文档）

{
  "@timestamp": "2025-04-05T12:34:56Z",
  "status": "su***ess",
  "user": "alice",
  "ip": "192.168.1.1"
}

而不是：

{
  "@timestamp": "...",
  "message": "{\"status\":\"su***ess\",...}"
}

转载请说明出处内容投诉
CSS教程网 » Filebeat 轻量级日志采集实践：安装、配置、多行合并、JSON 解析与字段处理

2320030001

分享到：

Filebeat 轻量级日志采集器实战指南

一、Beats 简介

官方 Beats 组件一览

二、Filebeat 概述

三、部署 Filebeat 8.2.2（二进制方式）

1. 下载 Filebeat 8.2.2

2. 解压安装包

3. 创建配置目录

四、实战案例：典型配置场景

案例 1：标准输入 → 控制台输出（调试用）

案例 2：采集普通日志文件input插件之log

案例 3：多行日志合并（按行数）count类型案例

案例 4：多行日志合并（正则匹配）pattern类型案例

案例 5：添加标签与自定义字段

📄 日志数据示例

1. /var/log/user-service/a***ess.log 示例内容

2. /var/log/mysql/error.log 示例内容

🧪 Filebeat 输出示例（控制台）

示例 1：来自 user-service 的日志（fields 在子对象中）

示例 2：来自 MySQL 的日志（fields 提升到根层级）

✅ 总结：这个配置的作用

案例 6：解析嵌套 JSON 日志（字符串内 JSON）

📄 示例日志文件：/tmp/nested-data.log

🧪 预期输出（控制台）

正常解析的示例：

解析失败的示例（自动添加 error）：

案例 7：多行 JSON 日志解析（换行分隔的 JSON 对象）

📄 示例日志文件：/tmp/service-logs.json

🧪 预期输出（控制台）

案例 8：使用 filestream（推荐方式）

📄 示例日志文件：/var/log/nginx/a***ess.log（NDJSON 格式）

🧪 预期输出（控制台）

✅ 正常解析的示例：

🔍 filestream 优势详解（vs log）

🛠️ 可选高级配置（按需添加）

案例 9：Filebeat 采集 Nginx 日志

一、安装并配置 Nginx

1. 安装 Nginx

2. 修改 Nginx 配置，启用 JSON 格式日志

设置访问日志使用该格式

3. 检查 Nginx 配置语法

4. 启动 Nginx 服务

5. 生成测试日志

二、编写 Filebeat 配置文件采集 Nginx 日志

创建配置文件

配置内容（带逐行注释）

三、启动 Filebeat 实例

四、使用 filestream 采集（推荐方式）

案例 10：使用 Filebeat 采集 Tomcat 结构化日志

一、下载并安装 Tomcat

1. 下载二进制包（推荐 Apache 官方源）

2. 解压到统一软件目录

二、配置 JDK 环境（Tomcat 依赖）

1. 确保已安装 JDK 17（Tomcat 9+ 推荐）

2. 写入环境变量文件（永久生效）

3. 加载环境变量

4. 验证 Java 是否可用

三、配置 Tomcat 输出 JSON 格式访问日志

四、启动 Tomcat 服务

1. 启动 Tomcat

2. 验证是否监听 8080 端口

3. 测试访问并生成日志

4. 查看 JSON 日志是否生成

五、编写 Filebeat 配置文件采集 Tomcat 日志

创建配置文件

配置内容

六、启动 Filebeat 实例

案例11：Filebeat 采集 containerd 容器类型日志使用 dissect 处理器

1. 背景说明

2. 环境准备

3. 配置 Filebeat

4. 启动 Filebeat

案例12：Filebeat 采集 Redis 慢查询日志（实验性功能）

1. 背景说明

2. 启动 Redis 实例

3. 配置 Redis 慢查询日志

4. 写入测试数据并触发慢日志

5. 配置 Filebeat

6. 启动 Filebeat

7. 验证输出

1. `/var/log/user-service/a***ess.log` 示例内容

2. `/var/log/mysql/error.log` 示例内容

📄 示例日志文件：`/tmp/nested-data.log`

📄 示例日志文件：`/tmp/service-logs.json`

案例 8：使用 `filestream`（推荐方式）

📄 示例日志文件：`/var/log/nginx/a***ess.log`（NDJSON 格式）

🔍 `filestream` 优势详解（vs `log`）

四、使用 `filestream` 采集（推荐方式）

案例11：Filebeat 采集 containerd 容器类型日志使用 `dissect` 处理器

📁 配置文件：`config/tcp-to-file.yaml`

📁 配置文件：`config/tcp-to-es.yaml`

📄 修改 Nginx 配置：`/etc/nginx/nginx.conf`

📁 配置文件：`config/nginx-to-es.yaml`

原始日志内容（`/var/log/app.log`）：