EFK Stack 面试题

ELK: 数据源 → Logstash（统一采集处理）→ Elasticsearch → Kibana
EFK: 数据源 → Fluentd/Fluent Bit（采集）→ Elasticsearch → Kibana
                 可选项：Fluent Bit → Fluentd（聚合层）→ ES

事件 = { Tag, Time, Record }

Tag:   "nginx.access"      → 路由标识，用于匹配 Input/Output
Time:  2026-05-26T10:00:00Z → 事件时间戳
Record: {"message":"GET /api 200", "status":200} → JSON 数据本体

Input Plugin → Event Router → Output Plugin
                    │
              Tag 匹配 Engine
              Input/Output/Buffer/Filter 插件

# nginx 日志 → ES
<match nginx.**>
  @type elasticsearch
  host es-cluster
  index_name nginx-logs
</match>

# app 日志 → Kafka
<match app.**>
  @type kafka2
  brokers kafka:9092
  topic app-logs
</match>

# 未匹配 → 丢弃（防止内存泄漏）
<match **>
  @type null
</match>

graph TD
    Input --> Filter --> OutputBuffer[Output Buffer] --> OutputSink[Output Sink]
    OutputBuffer --> BufferChunk["Buffer Chunk<br/>(File/Memory)"]

<match nginx.**>
  @type elasticsearch

  # Buffer 配置
  <buffer>
    @type file
    path /var/log/fluentd/buffer/nginx

    # Chunk 大小限制（默认 8MB）
    chunk_limit_size 8MB

    # Chunk 数量上限（默认 512）
    chunk_limit_records 10000

    # 刷新间隔
    flush_interval 5s
    flush_at_shutdown true

    # 重试配置
    retry_timeout 72h              # 最大重试时间
    retry_max_times 60             # 最大重试次数
    retry_forever false            # 是否无限重试
    retry_secondary_threshold 0.8  # 触发二次输出阈值

    # 排空队列
    overflow_action block          # block / throw_exception / drop_oldest_chunk
  </buffer>

  # 二次输出（失败降级）
  <secondary>
    @type file
    path /var/log/fluentd/failed/nginx
  </secondary>
</match>

Stage (buffer) → Queued → 输出成功 → 删除
                     ↓
               输出失败 → Retry
                     ↓
            Retry 耗尽 → Secondary 输出 / 丢弃

<source>
  @type tail
  path /var/log/nginx/access.log
  tag nginx.access
  pos_file /var/log/fluentd/pos/nginx.access.pos

  # 读取位置
  read_from_head true

  # 轮转策略
  rotate_wait 5s

  # 编码
  encoding UTF-8
  from_encoding ASCII-8BIT

  # 日志格式解析（用正则）
  format /^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$/

  # 或使用自定义 parser
  <parse>
    @type nginx
  </parse>
</source>

# 源端（Fluent Bit → Fluentd 聚合层）
<source>
  @type forward
  port 24224
  bind 0.0.0.0

  # 安全
  <transport tls>
    cert_path /etc/fluentd/certs/fluentd.crt
    private_key_path /etc/fluentd/certs/fluentd.key
  </transport>

  # 客户端认证
  <security>
    self_hostname aggregator-1
    shared_key secret_key
  </security>
</source>

Input → Filter1 → Filter2 → Filter3 → Output
         Record   Record     Record
        修改      过滤       增强

# 1. Grep Filter：仅保留 ERROR
<filter app.**>
  @type grep
  <regexp>
    key level
    pattern ^(ERROR|CRITICAL)$
  </regexp>
</filter>

# 2. Record Transformer：添加元数据
<filter app.**>
  @type record_transformer
  enable_ruby true
  <record>
    hostname "#{Socket.gethostname}"
    env ${record["env"] || "production"}
    log_level ${record["level"].downcase}
  </record>
</filter>

# 3. Parser Filter：解析 JSON message
<filter app.**>
  @type parser
  key_name message
  reserve_data true
  <parse>
    @type json
  </parse>
</filter>

# 4. GeoIP 增强
<filter nginx.**>
  @type geoip
  geoip_database /etc/fluentd/GeoLite2-City.mmdb
  geoip3_keys ["country_name", "city", "location"]
  <record>
    city         ${"geoip.city"}
    country      ${"geoip.country_name"}
    coordinates  ${"geoip.location"}
  </record>
  skip_adding_null_record true
</filter>

<filter nginx.**>   ：处理所有 nginx 前缀事件（不输出）
<match nginx.**>    ：匹配并输出（匹配后不再继续匹配）

1. Input 收到事件
   ↓
2. 写入 Buffer Chunk（文件/内存）
   ↓
3. Chunk 达到条件（大小/时间/记录数）
   ↓
4. 发送到目标（ES/Kafka/S3）
   ↓
5. 等待确认（Acknowledge）
   ↓
成功：删除 Chunk
失败：保留 Chunk → Retry → 重试达到上限 → Secondary

<match nginx.**>
  @type elasticsearch

  <buffer>
    @type file                    # 文件缓冲，进程崩溃可恢复
    path /var/log/fluentd/buffer/nginx

    chunk_limit_size 8MB
    flush_interval 5s
    flush_at_shutdown true        # 关闭时确保 flush

    retry_timeout 72h             # 最多重试 72 小时
    retry_max_times 60
    retry_forever false

    # 指数退避
    retry_exponential_backoff_base 2
    retry_max_interval 60         # 最大重试间隔 60s
  </buffer>

  # 重试耗尽后的二次输出（降级到文件）
  <secondary>
    @type file
    path /var/log/fluentd/failed/nginx
  </secondary>
</match>

<match nginx.**>
  @type elasticsearch
  id_key request_id      # 指定 ID 字段，ES 按 ID 幂等
  <buffer>
    @type file
  </buffer>
</match>

graph TD
    subgraph 采集层["采集层（每个节点）"]
        FB1["Fluent Bit<br/>(DaemonSet)"]
        FB2["Fluent Bit<br/>(DaemonSet)"]
        FB3["Fluent Bit<br/>(DaemonSet)"]
    end
    FB1 --> FA["Fluentd<br/>(Aggregator)<br/>聚合层（集群）"]
    FB2 --> FA
    FB3 --> FA
    FA --> ES["Elasticsearch / Kafka / S3"]

# Fluent Bit DaemonSet（每个节点采集）
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
spec:
  template:
    spec:
      containers:
        - name: fluent-bit
          image: fluent/fluent-bit:2.2
          resources:
            requests:
              memory: 10Mi
              cpu: 10m
            limits:
              memory: 50Mi
              cpu: 200m
# Fluentd Deployment（聚合层）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fluentd-aggregator
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: fluentd
          image: fluent/fluentd:v1.16
          resources:
            requests:
              memory: 256Mi
              cpu: 200m
            limits:
              memory: 512Mi
              cpu: 500m

POS 文件内容（/var/log/fluentd/pos/nginx.access.pos）：
/var/log/nginx/access.log  167890  2026-05-26T10:30:00+08:00
/var/log/nginx/error.log   45231   2026-05-26T10:30:00+08:00
           ↑                  ↑                ↑
        文件路径          字节偏移量        最后修改时间

场景：Nginx logrotate 轮转
1. access.log → access.log.1（rename）
2. 新 access.log 创建
3. Fluentd in_tail 检测：
   inode 变化 + 文件大小归零
   → 继续读取新的 access.log
   → POS 文件记录新 inode

配置选项：
  read_from_head true   → 重启后从头读取（否则从尾读取）
  rotate_wait 5s        → 轮转等待时间

<source>
  @type tail

  # 文件路径（支持通配符）
  path /var/log/nginx/*.log,/var/log/app/*.log
  exclude_path ["*.gz", "*.tmp"]

  # POS 文件路径
  pos_file /var/log/fluentd/pos/all.pos
  pos_file_compaction_interval 72h  # POS 文件压缩间隔

  # 读取策略
  read_from_head true               # 从头读取
  refresh_interval 5s               # 扫描间隔

  # 多行合并
  multiline_flush_interval 2s

  # 编码
  encoding UTF-8

  # 格式解析
  <parse>
    @type json
  </parse>
</source>

<source>
  @type tail
  path /var/log/app/error.log
  tag app.error

  # 正则匹配起始行
  multiline_flush_interval 2s

  <parse>
    @type multiline
    format_firstline /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/
    format1 /^(?<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+(?<level>\w+)\s+(?<message>.*)/
  </parse>
</source>

K8s Cluster
│
├── Node 1
│   └── Fluent Bit Pod（DaemonSet）
│       ├── /var/log/containers/*.log → 容器标准输出日志
│       ├── /var/log/pods/*/...       → Pod 日志
│       └── /var/lib/docker/containers/*/... → Docker 日志（可选）
│
├── Node 2
│   └── Fluent Bit Pod（DaemonSet）
│
├── Node N
│   └── Fluent Bit Pod（DaemonSet）
│
└── Fluentd Aggregator (Deployment)
    ├── Forward input ← Fluent Bit
    ├── Kubernetes Metadata Filter
    ├── Buffer (File)
    └── Elasticsearch Output

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluent-bit-config
data:
  fluent-bit.conf: |
    [SERVICE]
        Flush         5
        Log_Level     info
        Parsers_File  parsers.conf

    [INPUT]
        Name              tail
        Tag               kube.*
        Path              /var/log/containers/*.log
        Parser            docker
        DB                /var/log/flb_kube.db
        Mem_Buf_Limit     50MB
        Skip_Long_Lines   On
        Refresh_Interval  10

    [FILTER]
        Name                kubernetes
        Match               kube.*
        Kube_URL            https://kubernetes.default.svc:443
        Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
        Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
        Merge_Log           On
        Merge_Log_Key       log_parsed
        K8S-Logging.Parser  On
        K8S-Logging.Exclude Off

    [OUTPUT]
        Name            forward
        Match           *
        Host            fluentd-aggregator
        Port            24224    

# 接收 Fluent Bit 转发
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

# K8s 元数据清洗
<filter kube.**>
  @type record_transformer
  <record>
    # 清理 Kubernetes 多余字段
    docker_id ${record["docker"]["container_id"] rescue nil}
    pod_name ${record["kubernetes"]["pod_name"]}
    namespace ${record["kubernetes"]["namespace_name"]}
    container_name ${record["kubernetes"]["container_name"]}
    host ${record["host"]}
  </record>
  remove_keys docker,kubernetes,stream
</filter>

# 按命名空间分流
<match kube.production.**>
  @type elasticsearch
  hosts es-cluster:9200
  index_name production-logs-%Y.%m.%d
  <buffer>
    @type file
    path /var/log/fluentd/buffer/production
  </buffer>
</match>

<match kube.staging.**>
  @type elasticsearch
  hosts es-cluster:9200
  index_name staging-logs-%Y.%m.%d
  <buffer>
    @type file
    path /var/log/fluentd/buffer/staging
  </buffer>
</match>

日志事件：{ "log": "GET /api 200", "kubernetes": {"pod_name": "nginx-7d9f8c-abc12"} }
                    │
           kubernetes_metadata Filter
                    │
         调用 K8s API 查询 Pod 详细信息
                    │
         缓存结果（减少 API 调用）
                    │
日志事件：{
           "log": "GET /api 200",
           "kubernetes": {
             "pod_name": "nginx-7d9f8c-abc12",
             "namespace_name": "production",
             "container_name": "nginx",
             "labels": { "app": "nginx", "env": "prod" },
             "annotations": { "logging.fluentd.io/parser": "nginx" },
             "host": "node-1"
           }
         }

<filter kube.**>
  @type kubernetes_metadata

  # K8s API 连接
  kubernetes_url https://kubernetes.default.svc:443
  ca_file /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
  bearer_token_file /var/run/secrets/kubernetes.io/serviceaccount/token

  # 缓存配置
  cache_size 1000              # Pod 缓存数量
  cache_ttl 3600               # 缓存过期时间（秒）
  watch_mode true              # Watch API（实时更新）

  # 标签注入
  labels true                  # 注入 Pod Labels
  annotations false             # 注入 Pod Annotations

  # 自定义字段
  container_name_to_delete container_name  # 删除原始 container_name
  use_journal true             # 使用 journald 字段映射
</filter>

# Watch 模式：通过 Watch API 实时同步 Pod 变化
<filter kube.**>
  @type kubernetes_metadata
  watch_mode true
  watch_interval 15            # Watch 重连间隔（秒）
  watch_buffer_size 10000      # Watch 事件缓冲区
end

# 非 Watch 模式：定期轮询（默认 5 分钟）
<filter kube.**>
  @type kubernetes_metadata
  watch_mode false
  cache_ttl 300                # 5 分钟 TTL
end

# 无标签（默认管道）
<source>
  @type tail
  tag nginx.access
</source>

# 带标签
<source>
  @type tail
  tag app.error
  @label @ERROR_STREAM      # 路由到 @ERROR_STREAM
</source>

# 标签定义
<label @ERROR_STREAM>
  <filter app.**>
    @type grep
    <regexp>
      key level
      pattern ERROR
    </regexp>
  </filter>

  <match app.**>
    @type elasticsearch
    host es-cluster
    index_name errors-%Y.%m.%d
  </match>
</label>

# 默认处理
<match **>
  @type elasticsearch
  host es-cluster
  index_name all-logs-%Y.%m.%d
</match>

graph LR
    S1["Source<br/>nginx.access"] --> NORMAL["@label @NORMAL<br/>Filter → Output"] --> ES1["ES (normal-logs)"]
    S2["Source<br/>app.error"] --> ERROR["@label @ERROR<br/>Filter → Output"] --> ES2["ES (error-logs)"]
    S3["Source<br/>audit.log"] --> AUDIT["@label @AUDIT<br/>Filter → Output"] --> S3out["S3 (archive)"]

# 每个 Label 可使用独立的 worker
<system>
  workers 4
</system>

<label @NORMAL>
  <match **>
    @type elasticsearch
    # 此 Label 使用 worker 0-1
  </match>
</label>

<label @ERROR>
  <match **>
    @type elasticsearch
    # 此 Label 使用 worker 2-3
  </match>
</label>

Fluentd 主进程
    │
    ├── Worker 0 (main)
    │       ├── Input → Filter → Output
    │       └── Buffer
    │
    ├── Worker 1
    │       ├── Input → Filter → Output
    │       └── Buffer
    │
    └── Worker 2
            ├── Input → Filter → Output
            └── Buffer

# fluentd.conf
<system>
  workers 4                     # Worker 线程数
  root_dir /var/log/fluentd     # 工作目录
  log_level info
</system>

# 各 Worker 独立 Source
<source>
  @type tail
  @id tail_nginx_worker0
  path /var/log/nginx/access.log
  tag nginx.access
  <parse>
    @type nginx
  </parse>
</source>

# 将特定 Source 绑定到特定 Worker
<source>
  @type tail
  @id tail_app_log_worker1
  path /var/log/app/*.log
  @label @APP

  # 绑定 Worker 1
  <worker>
    workers 1
  </worker>
</source>

<source>
  @type tail
  @id tail_syslog_worker2
  path /var/log/syslog
  @label @SYSLOG

  # 绑定 Worker 2
  <worker>
    workers 2
  </worker>
</source>

# 利用 in_forward 原生多线程
<source>
  @type forward
  @id forward_input
  port 24224

  # 绑定多个 Worker
  <worker>
    workers 0-3
  </worker>
</source>

<label @NORMAL>
  # 自动负载均衡到 4 个 Worker
  <match **>
    @type elasticsearch
    hosts es-1:9200,es-2:9200,es-3:9200
    <buffer>
      flush_thread_count 4       # 每个 Worker 4 个 Flush 线程
    </buffer>
  </match>
</label>

跨 Worker 转发需使用 in_forward + out_forward
同一 Worker 内可使用 @label 路由
不同 Worker 间不能直接 @label 跳转

<match nginx.**>
  @type elasticsearch

  # ES 连接
  hosts es-node-1:9200,es-node-2:9200
  scheme https
  port 9200

  # 认证
  user fluentd_writer
  password ${ES_PASSWORD_ENV}

  # 索引名称（支持时间变量）
  index_name nginx-logs-%Y.%m.%d

  # 文档 ID（用于幂等写入）
  id_key request_id

  # 批量写入
  bulk_message_request_threshold 10MB
  flush_interval 5s

  # SSL
  ssl_verify true
  ca_file /etc/fluentd/certs/ca.crt
  client_cert /etc/fluentd/certs/client.crt
  client_key /etc/fluentd/certs/client.key

  # 重试策略
  <secondary>
    @type file
    path /var/log/fluentd/failed/nginx
  </secondary>
</match>

<match nginx.**>
  @type elasticsearch

  # 模板名称
  template_name nginx-logs-template

  # 模板文件
  template_file /etc/fluentd/templates/nginx-logs.json

  # 自动创建索引
  create_index true

  # 模板覆盖
  template_overwrite true

  # 滚动索引
  rollover_index true
  index_date_pattern "now/d{yyyy.MM.dd}"
</match>

{
  "index_patterns": ["nginx-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.refresh_interval": "30s",
      "index.lifecycle.name": "logs_retention"
    },
    "mappings": {
      "dynamic": true,
      "properties": {
        "@timestamp":  { "type": "date" },
        "message":     { "type": "text" },
        "method":      { "type": "keyword" },
        "path":        { "type": "text", "index": false },
        "status":      { "type": "integer" },
        "latency_ms":  { "type": "float" }
      }
    }
  }
}

<match nginx.**>
  @type elasticsearch

  # 批量大小（按记录数）
  bulk_message_request_threshold 5000

  # 批量大小（按字节）
  chunk_limit_size 10MB

  # 刷新间隔
  flush_interval 5s

  # 并发
  flush_thread_count 4

  # 流式写入（禁用批量确认等待）
  reload_after 100000          # 每 10 万条重新连接
  reload_forever true

  # 压缩
  compress_request gzip

  # 超时
  request_timeout 30s
  slow_flush_log_threshold 10s
</match>

<match archive.**>
  @type s3

  # S3 连接
  s3_bucket logs-archive
  s3_region ap-southeast-1

  # 路径模板
  path logs/%Y/%m/%d/${tag[1]}/%H

  # Access Key（推荐使用 IAM Role 或环境变量）
  access_key_id ${AWS_ACCESS_KEY}
  secret_access_key ${AWS_SECRET_KEY}

  # S3 对象命名
  s3_object_key_format "%{path}/%{time_slice}_%{index}.%{file_extension}"

  # Buffer 配置
  <buffer tag,time>
    @type file
    path /var/log/fluentd/buffer/s3
    timekey 600                 # 10 分钟切片
    timekey_wait 60
    timekey_use_utc true
    chunk_limit_size 50MB
  </buffer>
</match>

<match archive.**>
  @type s3

  # 压缩格式
  store_as gzip                # gzip / lzo / snappy / json / text

  # 压缩级别
  compression_level 6          # 1-9，默认 6

  # 文件扩展名
  s3_object_key_format "%{path}/%{time_slice}_%{index}.%{file_extension}"
  # 自动根据 store_as 添加扩展名（.gz）
</match>

<match archive.**>
  @type s3

  # MinIO 兼容 S3 API
  s3_bucket logs-archive
  s3_region us-east-1

  # MinIO 端点
  endpoint http://minio:9000
  force_path_style true         # 使用路径样式（非虚拟主机样式）

  # 认证
  access_key_id ${MINIO_ACCESS_KEY}
  secret_access_key ${MINIO_SECRET_KEY}

  <buffer tag,time>
    @type file
    path /var/log/fluentd/buffer/s3
    timekey 3600
    timekey_wait 60
  </buffer>

  # 自定义路径
  path production/logs/${tag[1]}/%Y/%m/%d/
</match>

<match kafka.**>
  @type kafka2

  # Kafka 连接
  brokers kafka-1:9092,kafka-2:9092,kafka-3:9092

  # 主题
  default_topic app-logs

  # 消息键（用于分区决策）
  default_message_key log_key        # 字段名

  # 分区策略
  partition_key "${tag}"             # 按 Tag 分区
  # partition_hash true              # 一致性哈希

  # 压缩
  compression_codec gzip             # gzip/snappy/lz4/zstd

  # Producer 配置
  max_send_retries 3
  required_acks -1                   # all（等待所有副本确认）
  ack_timeout 30000                  # ms

  <buffer>
    @type file
    path /var/log/fluentd/buffer/kafka
    flush_interval 5s
    chunk_limit_size 1MB
  </buffer>
</match>

<match kafka.**>
  @type kafka2

  # 1. 按 log_level 分区
  partition_key "${record['log_level']}"

  # 2. 按哈希分区
  # partition_key "random"
  # 一致性哈希（相同 Key 进入相同分区）
  # partition_hash true

  # 3. 随机分区（默认）
  # 无需配置 partition_key

  # 4. 自定义 Key
  <format>
    @type json
  </format>
</match>

<match kafka.nginx.**>
  @type kafka2
  default_topic nginx-logs
  <buffer>
    @type file
    path /var/log/fluentd/buffer/kafka/nginx
  </buffer>
</match>

<match kafka.app.**>
  @type kafka2
  default_topic app-logs
  <buffer>
    @type file
    path /var/log/fluentd/buffer/kafka/app
  </buffer>
</match>

<match kafka.audit.**>
  @type kafka2
  default_topic audit-logs
  <buffer>
    @type file
    path /var/log/fluentd/buffer/kafka/audit
  </buffer>
</match>

<match kafka.**>
  @type kafka2

  # 自定义 Headers
  headers {
    "source": "fluentd",
    "env": "${record['env']}"
  }

  # 自动注入 Fluentd 元数据
  headers_from_record true
  headers_record_key headers
</match>

# 类型声明
<source>
  @type tail
  tag app.log
</source>

# 参数赋值
path /var/log/app/*.log
port 24224

# 嵌套块
<match **>
  @type elasticsearch
  <buffer>
    @type file
  </buffer>
</match>

# 环境变量
password ${ENV['ES_PASSWORD']}
host ${ENV['HOSTNAME']}

# Ruby 表达式（在 record_transformer 中使用）
<filter app.**>
  @type record_transformer
  enable_ruby true
  <record>
    # 字符串操作
    env "${record['env'].downcase rescue 'unknown'}"
    hostname "#{Socket.gethostname}"
    timestamp "#{Time.now.iso8601}"

    # 条件赋值
    severity "${record['level'] == 'ERROR' ? 'critical' : 'normal'}"

    # 数值计算
    latency_ms "${record['latency_s'] * 1000}"
  </record>
</filter>

# 根据 Tag 条件分流
<match nginx.**>
  # Nginx 日志处理
</match>

<match app.**>
  # 应用日志处理
</match>

# 根据记录字段条件
<match **>
  @type forward
  <match>
    key level
    pattern ^ERROR$
    @type stdout
  </match>
  <match>
    key response
    pattern ^5\d\d$
    @type elasticsearch
    host es-alert
    index_name errors-5xx
  </match>
</match>

# 使用 include 拆分配置文件
@include /etc/fluentd/conf.d/*.conf

# 条件 include
@include if File.exist?('/etc/fluentd/conf.d/extra.conf')
  /etc/fluentd/conf.d/extra.conf
@include

# 测试环境配置覆盖
@include "#{ENV['FLUENTD_ENV'] || 'production'}.conf"

# /etc/fluentd/plugins/filter_sanitize.rb
module Fluent::Plugin
  class SanitizeFilter < Filter
    Fluent::Plugin.register_filter('sanitize', self)

    # 配置参数
    config_param :sensitive_fields, :array, default: ['password', 'token', 'ssn']
    config_param :replacement, :string, default: '***REDACTED***'

    def configure(conf)
      super
    end

    def filter(tag, time, record)
      # 递归清洗敏感字段
      sanitize_record(record)
      record
    end

    private

    def sanitize_record(record)
      record.each do |key, value|
        if @sensitive_fields.any? { |f| key.to_s.downcase.include?(f) }
          record[key] = @replacement
        elsif value.is_a?(Hash)
          sanitize_record(value)
        end
      end
    end
  end
end

# /etc/fluentd/plugins/out_webhook.rb
module Fluent::Plugin
  class WebhookOutput < Output
    Fluent::Plugin.register_output('webhook', self)

    config_param :endpoint, :string
    config_param :http_method, :string, default: 'POST'
    config_param :headers, :hash, default: {}

    def configure(conf)
      super
      @http = Net::HTTP
    end

    def multi_write(chunks)
      chunks.each do |chunk|
        send_batch(chunk)
      end
    end

    private

    def send_batch(chunk)
      events = []
      chunk.each do |time, record|
        events << record
      end

      request = Net::HTTP::Post.new(
        URI(@endpoint),
        'Content-Type' => 'application/json'
      )

      @headers.each { |k, v| request[k] = v }
      request.body = JSON.generate(events)

      response = Net::HTTP.start(URI(@endpoint).host, URI(@endpoint).port, use_ssl: true) do |http|
        http.request(request)
      end

      unless response.is_a?(Net::HTTPSuccess)
        raise "Webhook failed: #{response.code} #{response.message}"
      end
    end
  end
end

# 方式1：插件目录（推荐）
# 目录: /etc/fluentd/plugins/
# 文件: filter_sanitize.rb
# 配置中使用 type sanitize

# 方式2：Gem 安装
# gem install fluent-plugin-elasticsearch
# 配置中使用 type elasticsearch

# 方式3：源码加载
# /etc/fluentd/plugin_libs/
# 在 conf 中 require

graph LR
    subgraph 边缘A["边缘设备 A"]
        FBA["Fluent Bit<br/>(500KB)"]
    end
    subgraph 边缘B["边缘设备 B"]
        FBB["Fluent Bit<br/>(500KB)"]
    end
    边缘A -->|4G/WiFi| 中心["中心 Fluentd<br/>(聚合层)"]
    边缘B --> 中心
    中心 --> ESS3["ES / S3"]

[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    Name         tail
    Path         /var/log/sensor/*.log
    Tag          iot.sensor
    DB           /tmp/flb_sensor.db
    Mem_Buf_Limit 5MB

[FILTER]
    Name         modify
    Match        iot.*
    Add          device_id ${DEVICE_ID}
    Add          location factory-a

[OUTPUT]
    Name         forward
    Match        *
    Host         central-fluentd.example.com
    Port         24224
    # 网络不稳定时的本地缓存
    Retry_Limit  10
    Retry_Wait   30
    # 压缩传输
    Compress     gzip

# Fluentd 中心端配置：处理边缘离线
<match iot.**>
  @type elasticsearch

  <buffer>
    @type file
    path /var/log/fluentd/buffer/iot

    # 大容量缓存（边缘离线可达数小时）
    chunk_limit_size 50MB
    total_limit_size 10GB

    # 长时间重试
    retry_timeout 168h          # 7 天
    retry_forever true

    # 背压处理
    overflow_action block       # 阻塞而非丢弃
  </buffer>
</match>

# 内置 Monitor Agent
<source>
  @type monitor_agent
  bind 0.0.0.0
  port 24220
</source>

# 接口信息
# GET /api/plugins.json → 所有插件状态
# GET /api/config.json → 配置信息

# 示例输出
curl http://localhost:24220/api/plugins.json
{
  "plugins": [
    {
      "plugin_id": "tail_nginx",
      "type": "tail",
      "output_plugin": false,
      "retry_count": 0,
      "buffer_queue_length": 0,
      "buffer_total_queued_size": 0
    }
  ]
}

# Prometheus Input（暴露 Metrics）
<source>
  @type prometheus
  bind 0.0.0.0
  port 24221
  metrics_path /metrics
</source>

# 内置 Metrics
<source>
  @type prometheus_monitor
</source>

<source>
  @type prometheus_output_monitor
</source>

# 自定义 Metrics
<filter **>
  @type prometheus
  <metric>
    name fluentd_event_count
    type counter
    desc "Total number of events received"
    key event_count
  </metric>
</filter>

groups:
  - name: fluentd
    rules:
      - alert: FluentdBufferQueueGrowing
        expr: rate(fluentd_status_buffer_queue_length[5m]) > 0
        for: 10m
        annotations:
          summary: "Fluentd buffer queue is growing"

      - alert: FluentdRetryCountHigh
        expr: fluentd_status_retry_count > 100
        for: 5m

      - alert: FluentdProcessDown
        expr: up{job="fluentd"} == 0
        for: 1m

      - alert: FluentdFlushTimeHigh
        expr: fluentd_status_flush_time > 60
        for: 5m

graph TD
    LB["Load Balancer"] --> FD1["Fluentd<br/>Node 1"]
    LB --> FD2["Fluentd<br/>Node 2"]
    LB --> FD3["Fluentd<br/>Node 3"]
    FD1 --> ES["Elasticsearch<br/>Cluster"]
    FD2 --> ES
    FD3 --> ES

# 所有节点配置相同
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

<match **>
  @type elasticsearch
  hosts es-1:9200,es-2:9200,es-3:9200

  <buffer>
    @type file
    path /var/log/fluentd/buffer
  </buffer>
</match>

graph TD
    Input --> Active["Fluentd Active<br/>Forward + Processing"]
    Active --> ES1[ES]
    Active -->|心跳监测| Standby["Fluentd Standby<br/>(备用)"]
    Standby -->|故障切换| ES2[ES]

# Fluentd 配置保持不变
# Keepalived 管理 VIP

# Primary
vrrp_instance VI_1 {
    state MASTER
    interface eth0
    virtual_router_id 51
    priority 100
    virtual_ipaddress {
        192.168.1.100/24    # Fluentd VIP
    }
}

# Backup
vrrp_instance VI_1 {
    state BACKUP
    interface eth0
    virtual_router_id 51
    priority 50
    virtual_ipaddress {
        192.168.1.100/24
    }
}

# 1. 查看 Buffer 积压
curl http://localhost:24220/api/plugins.json | jq '.plugins[] | select(.buffer_queue_length>0)'

# 2. 查看 Flush 耗时
curl http://localhost:24220/api/plugins.json | jq '.plugins[] | .flush_time_count'

# 3. 系统资源
top -p $(pgrep -f fluentd)

# 4. 线程状态
grep "fluentd" /proc/$(pgrep -f fluentd)/status

# 5. Ruby GC 分析
# 在 fluentd.yml 添加
--trace-gc
--gc-stat

<system>
  workers 4                     # Multi-Worker
  root_dir /var/log/fluentd
</system>

<source>
  @type tail
  @id tail_nginx
  path /var/log/nginx/access.log
  tag nginx.access

  # 关键优化
  read_bytes_limit 5MB          # 单次读取大小
  refresh_interval 5s
</source>

<filter nginx.**>
  @type record_transformer
  enable_ruby false             # 禁用 Ruby（性能提升显著）
  <record>
    hostname ${hostname}
  </record>
</filter>

<match nginx.**>
  @type elasticsearch
  hosts es-1:9200,es-2:9200

  <buffer>
    @type file
    path /data/fluentd/buffer/nginx    # 独立 SSD 磁盘
    chunk_limit_size 32MB              # 增大 Chunk
    queue_limit_length 8192            # 队列上限

    flush_thread_count 8               # 并发 Flush
    flush_interval 3s
    flush_at_shutdown true

    # 禁用 retry 的指数退避上限
    retry_max_interval 5

    # 性能模式（减少 checkpoint 频率）
    checkpoint_interval 60
  </buffer>

  # 批量写入优化
  bulk_message_request_threshold 32MB
  reload_after 100000
  compress_request gzip
</match>

[INPUT]
    Name   tail
    Path   /var/log/app/*.log
    Tag    app

[FILTER]
    Name   grep
    Match  app.*
    Regex  log_level ^ERROR$

[FILTER]
    Name   modify
    Match  app.*
    Add    env production

[OUTPUT]
    Name   es
    Match  app.*
    Host   es-cluster

[INPUT]
    Name        tail
    Path        /var/log/app/*.log
    Tag         app
    # 绑定 Processor
    processors  app-processor

[PROCESSOR]
    Name        app-processor

    # Phase 1: 过滤
    [[PROCESSOR.phases]]
        [[PROCESSOR.phases.rules]]
            action      grep
            key         log_level
            pattern     ^(ERROR|CRITICAL)$

    # Phase 2: 字段增强
    [[PROCESSOR.phases]]
        [[PROCESSOR.phases.rules]]
            action      modify
            add         {"env": "production", "source": "fluent-bit"}

    # Phase 3: 指标聚合（批处理）
    [[PROCESSOR.phases]]
        [[PROCESSOR.phases.rules]]
            action      aggregate
            key         error_type
            operation   count
            window      60

[PROCESSOR]
    Name     metrics-processor

    # 10 秒窗口内按 status 聚合
    [[PROCESSOR.phases]]
        [[PROCESSOR.phases.rules]]
            action      aggregate
            key         status
            operation   count
            window      10
            output_key  request_count

        [[PROCESSOR.phases.rules]]
            action      aggregate
            key         latency_ms
            operation   avg
            window      10
            output_key  avg_latency

graph LR
    subgraph Agent["Fluentd (Agent)"]
        OF["out_forward<br/>Load Balance<br/>Node 1 / Node 2 / Node 3"]
    end
    subgraph Aggregator["Fluentd (Aggregator)"]
        IF["in_forward"]
    end
    Agent -->|TCP 24224| Aggregator

<match **>
  @type forward

  # 多个目标节点（负载均衡 + 故障转移）
  <server>
    host aggregator-1.example.com
    port 24224
    weight 60                   # 权重
  </server>
  <server>
    host aggregator-2.example.com
    port 24224
    weight 40
  </server>
  <server>
    host aggregator-3.example.com
    port 24224
    standby true                # 备用节点
  </server>

  # 心跳检测
  <heartbeat>
    interval 10s                # 心跳间隔
    type tcp                    # TCP 心跳
  </heartbeat>

  # Ack 确认
  ack_response_timeout 30s      # 等待 Ack 超时
  require_ack_response true     # 需要 Ack 确认

  # 连接池
  keepalive true
  keepalive_timeout 120s        # 长连接超时

  # 压缩
  compress gzip

  # 序列化
  <format>
    @type msgpack               # 高效的二进制格式
  </format>

  # Buffer
  <buffer>
    @type file
    path /var/log/fluentd/buffer/forward
    flush_interval 5s
    retry_timeout 72h
  </buffer>
</match>

<source>
  @type forward

  # 监听配置
  bind 0.0.0.0
  port 24224

  # 连接配置
  linger_timeout 10000          # 连接超时（ms）
  resolve_interval 10s          # DNS 解析间隔

  # 安全认证
  <security>
    self_hostname aggregator-1
    shared_key ${FLUENTD_SHARED_KEY}
  </security>

  # 传输层安全
  <transport tls>
    cert_path /etc/fluentd/certs/fluentd.crt
    private_key_path /etc/fluentd/certs/fluentd.key
    ca_path /etc/fluentd/certs/ca.crt
    client_cert_auth true
  </transport>
</source>

发送端                       接收端
   │                          │
   │────── Event Data ───────→│
   │                          │ 写入 Buffer
   │                          │────── ACK ──────→ 确认
   │ 删除 Buffer Chunk        │
   │                          │
   │─── 无响应 / 超时 ──────→│ 故障
   │ 保留 Chunk               │
   │ 重试 / 切换节点          │

边缘节点（资源受限）    Fluent Bit
       ↓ Forward
聚合层（Filter/Buffer）  Fluentd / Vector
       ↓
目标存储               ES / S3 / Kafka

<source>
  @type kafka

  # Kafka 集群
  brokers kafka-1:9092,kafka-2:9092,kafka-3:9092

  # 消费者组
  consumer_group fluentd-consumer-group

  # 主题
  topics app-logs

  # 分区分配策略
  partition_assignment_strategy round_robin   # round_robin / range / sticky

  # Offset 策略
  offset_committed true          # 提交 Offset
  offset_commit_interval 10      # Offset 提交间隔（秒）
  offset_commit_threshold 1000   # Offset 提交阈值

  # 初始偏移量
  # start_from_beginning true    # 从最早开始
  # default_offset oldest        # oldest / latest

  # 并发
  max_bytes 1048576              # 单次最大拉取（1MB）
  max_wait_time 5000             # 最大等待（ms）
  min_bytes 1                    # 最小字节数

  # 解析
  <parse>
    @type json
  </parse>
</source>

<match kafka.**>
  @type kafka2

  brokers kafka-1:9092,kafka-2:9092

  # 主题
  default_topic app-logs

  # 分区策略
  partition_by key                          # 按 Key 分区
  partition_key "${record['host']}"          # 分区键

  # 或按一致性哈希
  # partition_hash true
  # partition_key "${record['app_name']}"

  # 确认级别
  required_acks -1                           # all / 1 / 0

  # Producer 配置
  max_send_retries 5
  ack_timeout 30000

  # 压缩
  compression_codec gzip

  # 消息格式
  <format>
    @type json
  </format>

  # Buffer
  <buffer>
    @type file
    path /var/log/fluentd/buffer/kafka
    flush_interval 5s
  </buffer>
</match>

<source>
  @type syslog

  # 监听配置
  port 5140
  bind 0.0.0.0

  # 协议支持
  protocol_type udp                # udp / tcp

  # RFC 协议选择
  rfc rfc5424                      # rfc3164 / rfc5424 / auto

  # 解析配置
  message_format auto               # auto / rfc3164 / rfc5424 / ceesyslog

  # 兼容模式
  <parse>
    @type syslog
    with_priority true             # 包含 PRI 头
    support_octet_counted_frames true  # TCP octet counting
  </parse>

  # Tag 生成
  tag syslog.${facility}.${severity}
</source>

原始消息（RFC 5424）：
<14>1 2026-05-26T10:30:00Z hostname app 1234 msgid - [meta@0 key="val"] This is a test

解析后：
{
  "facility": 1,
  "severity": 6,
  "pri": 14,
  "time": "2026-05-26T10:30:00Z",
  "host": "hostname",
  "ident": "app",
  "pid": "1234",
  "msgid": "msgid",
  "message": "This is a test",
  "structured_data": {
    "meta@0": { "key": "val" }
  }
}

<source>
  @type syslog
  port 514
  bind 0.0.0.0
  protocol_type udp

  # 接收缓冲区
  receive_buffer_size 16384       # 16KB 缓冲区

  # 多线程
  <worker>
    workers 2
  </worker>

  # 丢包防护
  <parse>
    @type syslog
  </parse>
</source>

<source>
  @type tail
  path /var/log/app/error.log
  tag app.error

  <parse>
    @type multiline

    # 第一行匹配正则
    format_firstline /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/

    # 后续行格式
    format1 /^(?<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+(?<level>\w+)\s+(?<message>.*)/
    format2 /^\s+(?<stack_trace>.*)/
  </parse>
</source>

# Fluent Bit v2.0+ Multiline Filter
[INPUT]
    Name   tail
    Path   /var/log/app/error.log
    Tag    app.error

[FILTER]
    Name   multiline
    Match  app.*

    # 多行规则
    multiline.parser   java          # 内置 Java 堆栈
    # multiline.parser go            # 内置 Go
    # multiline.parser python        # 内置 Python

    # 或自定义
    # multiline.parser custom
    # multiline.rule   start_state   /^\d{4}-\d{2}-\d{2}/  cont
    # multiline.rule   cont          /^\s+at/             cont
    # multiline.rule   cont          /^Caused by:/        cont
    # multiline.rule   any           /.*/                 end

    # Buffer 类型
    multiline.flush_timeout  5      # 5 秒未匹配新行则强制 flush

# 输入（多行）
2026-05-26 10:30:00 ERROR - NullPointerException
    at com.example.service.UserService.getUser(UserService.java:25)
    at com.example.controller.UserController.handle(UserController.java:10)
Caused by: java.lang.RuntimeException: Database connection failed
    at com.example.dao.UserDao.findById(UserDao.java:45)
    ... 10 more

# 输出（单条事件）
{
  "time": "2026-05-26T10:30:00",
  "level": "ERROR",
  "message": "NullPointerException\n    at com.example.service.UserService.getUser(UserService.java:25)\n    at com.example.controller.UserController...",
  "stack_trace": "    at com.example.service.UserService.getUser(UserService.java:25)\n    ..."
}

[FILTER]
    Name   multiline
    Match  app.*

    # 自定义规则：systemd journal 格式
    multiline.rule   start_state   /^\d{4}-\d{2}-\d{2}/    cont
    multiline.rule   cont          /^\s+/                  cont
    multiline.rule   cont          /./                     end

<source>
  @type http
  port 9880
  bind 0.0.0.0

  # 格式
  <parse>
    @type json
  </parse>

  # 跨域
  cors_allow_origins ["*"]

  # 路径（可选）
  # 访问 http://localhost:9880/app.logs 时 tag = app.logs
  use_default_path true
</source>

# POST JSON
curl -X POST http://localhost:9880/app.logs \
  -H "Content-Type: application/json" \
  -d '{"message": "test log", "level": "ERROR", "timestamp": "2026-05-26T10:00:00Z"}'

# 批量发送
curl -X POST http://localhost:9880/app.logs \
  -H "Content-Type: application/json" \
  -d '[
    {"message": "log1", "level": "INFO"},
    {"message": "log2", "level": "ERROR"}
  ]'

# MessagePack 格式
curl -X POST http://localhost:9880/app.logs \
  -H "Content-Type: application/msgpack" \
  --data-binary @data.msgpack

<source>
  @type http
  port 9880
  bind 0.0.0.0

  # Webhook 路径路由
  # POST /github-webhook → tag = github-webhook
  # POST /datadog-webhook → tag = datadog-webhook
  use_default_path true

  # 认证
  <auth>
    username ${HTTP_AUTH_USER}
    password ${HTTP_AUTH_PASS}
  </auth>
</source>

# 处理 GitHub Webhook
<match github-webhook>
  @type stdout
</match>

# 调试：输出到控制台（Rubydebug 格式）
<match debug.**>
  @type stdout
  <format>
    @type stdout
  </format>
</match>

# 输出示例
# 2026-05-26 10:30:00.000000000 +0800 nginx.access: {"method":"GET","status":200,"path":"/api"}

# 使用 JSON 格式输出
<match debug.**>
  @type stdout
  <format>
    @type json
  </format>
</match>

<match archive.**>
  @type file

  # 输出路径
  path /var/log/fluentd/output/${tag[1]}

  # 文件命名
  <buffer tag,time>
    @type file
    path /var/log/fluentd/buffer/file
    timekey 3600
    timekey_wait 10
  </buffer>

  # 格式
  <format>
    @type json
  </format>

  # 压缩
  compress gzip
  append true                     # 追加模式
  symlink_path /var/log/fluentd/current  # 符号链接指向最新文件
</match>

# 测试：模拟生产管道，输出到文件便于比对
<match test.**>
  @type file
  path /tmp/fluentd-test/${tag}
  <buffer>
    @type memory
  </buffer>
</match>

# 验证命令
# diff /tmp/fluentd-test/expected.log /tmp/fluentd-test/actual.log

# 1. 命令行单行模式
fluentd -e '
  <source>
    @type tail
    path /var/log/test.log
    tag test.input
    <parse>
      @type json
    </parse>
  </source>
  <match test.*>
    @type stdout
  </match>
'

# 2. 检查事件格式
fluentd --dry-run -c /etc/fluentd/fluentd.conf

# 3. 输出 Buffer 内容
ruby -e '
  require "msgpack"
  data = File.read("/var/log/fluentd/buffer/nginx/buffer.b513d")
  puts MessagePack.unpack(data)
'

# 1. Dry-run 模式（验证配置但不启动）
fluentd --dry-run -c /etc/fluentd/fluentd.conf

# 2. 检查配置文件语法
ruby -c /etc/fluentd/fluentd.conf

# 3. 使用 `--show-plugin-config` 查看插件参数
fluentd --show-plugin-config output:elasticsearch

# 测试 Pipeline
# test_pipeline.rb
require 'fluent/test'
require 'fluent/plugin/out_stdout'

class TestOutput < Test::Unit::TestCase
  def setup
    Fluent::Test.setup
  end

  def test_filter_and_output
    # 创建测试 Driver
    d = Fluent::Test::Driver::Output.new(Fluent::Plugin::StdoutOutput)
    d.configure(<<-CONFIG)
      <format>
        @type stdout
      </format>
    CONFIG

    # 模拟事件
    time = event_time("2026-05-26 10:00:00 UTC")
    d.run(default_tag: "test") do
      d.feed(time, {"message" => "test log", "level" => "ERROR"})
    end

    # 验证输出
    assert_equal(1, d.events.length)
    assert_equal("ERROR", d.events[0][2]["level"])
  end
end