Fluent Bit完全ガイド：軽量ログプロセッサのアーキテクチャ、設定、Kubernetes連携を総まとめ

+------------------------------------------------------------------+
|                        Fluent Bit Engine                         |
|                                                                  |
|  +--------+   +--------+   +--------+   +--------+   +--------+ |
|  | Input  |-->| Parser |-->| Filter |-->| Buffer |-->| Output | |
|  +--------+   +--------+   +--------+   +--------+   +--------+ |
|  | tail   |   | json   |   | k8s    |   | memory |   | es     | |
|  | systemd|   | regex  |   | grep   |   | filesys|   | loki   | |
|  | forward|   | logfmt |   | modify |   |        |   | s3     | |
|  | http   |   | cri    |   | lua    |   |        |   | kafka  | |
|  | tcp    |   | docker |   | nest   |   |        |   | stdout | |
|  +--------+   +--------+   +--------+   +--------+   +--------+ |
|                                                                  |
|  [Scheduler]  [Router / Tag Matching]  [HTTP Server / Monitoring]|
+------------------------------------------------------------------+

  [Data Source]
       |
       v
  +---------+     +---------+     +---------+     +---------+     +---------+
  |  INPUT  | --> | PARSER  | --> | FILTER  | --> | BUFFER  | --> | OUTPUT  |
  |         |     |         |     |         |     |         |     |         |
  | データ   |     | 非構造化 |     | データ   |     | メモリ/  |     | 最終    |
  | 収集     |     | -> 構造化|     | 加工     |     | ディスク |     | 配信    |
  +---------+     +---------+     +---------+     +---------+     +---------+
       |               |               |               |               |
    Tag付与        構造化変換       enrichment      安定性確保     宛先送信
                                   フィルタリング                  (Tag照合)

Pipeline A: [tail: app-*.log] --tag:app--> [filter:k8s] --> [output:elasticsearch]
Pipeline B: [tail: sys-*.log] --tag:sys--> [filter:grep] --> [output:loki]
Pipeline C: [forward:24224]   --tag:fwd--> [filter:lua]  --> [output:s3]

# GPGキーおよびリポジトリ追加
curl https://raw.githubusercontent.com/fluent/fluent-bit/master/install.sh | sh

# または手動インストール
wget -qO - https://packages.fluentbit.io/fluentbit.key | sudo apt-key add -
echo "deb https://packages.fluentbit.io/ubuntu/$(lsb_release -cs) $(lsb_release -cs) main" | \
  sudo tee /etc/apt/sources.list.d/fluent-bit.list

sudo apt-get update
sudo apt-get install -y fluent-bit

# サービス起動
sudo systemctl start fluent-bit
sudo systemctl enable fluent-bit

cat > /etc/yum.repos.d/fluent-bit.repo << 'EOF'
[fluent-bit]
name=Fluent Bit
baseurl=https://packages.fluentbit.io/centos/$releasever/
gpgcheck=1
gpgkey=https://packages.fluentbit.io/fluentbit.key
enabled=1
EOF

sudo yum install -y fluent-bit
sudo systemctl start fluent-bit
sudo systemctl enable fluent-bit

# 最新バージョンの実行
docker run -ti cr.fluentbit.io/fluent/fluent-bit:latest

# 設定ファイルのマウント
docker run -ti \
  -v /path/to/fluent-bit.yaml:/fluent-bit/etc/fluent-bit.yaml \
  -v /var/log:/var/log \
  cr.fluentbit.io/fluent/fluent-bit:latest \
  /fluent-bit/bin/fluent-bit -c /fluent-bit/etc/fluent-bit.yaml

# Docker Composeの例
cat > docker-compose.yaml << 'EOF'
version: '3.8'
services:
  fluent-bit:
    image: cr.fluentbit.io/fluent/fluent-bit:latest
    volumes:
      - ./fluent-bit.yaml:/fluent-bit/etc/fluent-bit.yaml
      - /var/log:/var/log:ro
    ports:
      - "2020:2020"   # HTTPモニタリング
      - "24224:24224"  # Forwardプロトコル
EOF

brew install fluent-bit

# 実行
fluent-bit -c /opt/homebrew/etc/fluent-bit/fluent-bit.conf

# またはYAML設定ファイルで実行
fluent-bit -c /path/to/fluent-bit.yaml

# Helmリポジトリ追加
helm repo add fluent https://fluent.github.io/helm-charts
helm repo update

# 基本インストール
helm install fluent-bit fluent/fluent-bit \
  --namespace logging \
  --create-namespace

# values.yamlカスタマイズでインストール
helm install fluent-bit fluent/fluent-bit \
  --namespace logging \
  --create-namespace \
  -f custom-values.yaml

# アップグレード
helm upgrade fluent-bit fluent/fluent-bit \
  --namespace logging \
  -f custom-values.yaml

# GitHub Releasesからバイナリをダウンロード
FLUENT_BIT_VERSION=3.2.2
wget https://github.com/fluent/fluent-bit/releases/download/v${FLUENT_BIT_VERSION}/fluent-bit-${FLUENT_BIT_VERSION}-linux-x86_64.tar.gz

tar xzf fluent-bit-${FLUENT_BIT_VERSION}-linux-x86_64.tar.gz
cd fluent-bit-${FLUENT_BIT_VERSION}-linux-x86_64/

# 実行
./bin/fluent-bit -c conf/fluent-bit.yaml

# バージョン確認
./bin/fluent-bit --version

# fluent-bit.conf - Classic INI形式

[SERVICE]
    Flush         5
    Daemon        Off
    Log_Level     info
    Parsers_File  parsers.conf
    HTTP_Server   On
    HTTP_Listen   0.0.0.0
    HTTP_Port     2020
    Hot_Reload    On

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

[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
    K8S-Logging.Parser On
    K8S-Logging.Exclude On

[OUTPUT]
    Name          es
    Match         kube.*
    Host          elasticsearch.logging.svc.cluster.local
    Port          9200
    Logstash_Format On
    Logstash_Prefix kube
    Retry_Limit   False

# fluent-bit.yaml - YAML形式（v3.2+標準）
service:
  flush: 5
  daemon: off
  log_level: info
  parsers_file: parsers.conf
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020
  hot_reload: on

pipeline:
  inputs:
    - name: tail
      path: /var/log/containers/*.log
      parser: cri
      tag: kube.*
      mem_buf_limit: 5MB
      skip_long_lines: on
      refresh_interval: 10
      db: /var/log/flb_kube.db

  filters:
    - 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
      k8s-logging.parser: on
      k8s-logging.exclude: on

  outputs:
    - name: es
      match: kube.*
      host: elasticsearch.logging.svc.cluster.local
      port: 9200
      logstash_format: on
      logstash_prefix: kube
      retry_limit: false

# YAML環境変数の例
pipeline:
  outputs:
    - name: es
      match: '*'
      host: ${ELASTICSEARCH_HOST}
      port: ${ELASTICSEARCH_PORT}
      http_user: ${ES_USER}
      http_passwd: ${ES_PASSWORD}
      tls: ${ES_TLS_ENABLED}

# Classic環境変数の例
[OUTPUT]
    Name          es
    Match         *
    Host          ${ELASTICSEARCH_HOST}
    Port          ${ELASTICSEARCH_PORT}
    HTTP_User     ${ES_USER}
    HTTP_Passwd   ${ES_PASSWORD}

pipeline:
  outputs:
    - name: es
      http_passwd: file:///run/secrets/es-password

# fluent-bit.conf（Classic）
[SERVICE]
    Flush 5

@INCLUDE inputs.conf
@INCLUDE filters.conf
@INCLUDE outputs.conf

# fluent-bit.yaml
includes:
  - inputs.yaml
  - filters.yaml
  - outputs.yaml

service:
  flush: 5

pipeline:
  inputs:
    - name: tail
      tag: app.logs
      path: /var/log/app/*.log
      path_key: filename # ファイルパスをレコードに含める
      exclude_path: /var/log/app/debug.log # 特定ファイルを除外
      parser: json # デフォルトパーサー
      db: /var/log/flb_app.db # offset保存DB（再起動時に続きから収集）
      db.sync: normal # DB同期モード
      refresh_interval: 10 # ファイルリスト更新周期（秒）
      read_from_head: false # true: ファイル先頭から読み取り
      skip_long_lines: on # 非常に長い行をスキップ
      mem_buf_limit: 5MB # メモリバッファ上限
      rotate_wait: 5 # ローテーション待ち時間（秒）
      multiline.parser: docker, cri # マルチラインパーサー

pipeline:
  inputs:
    - name: systemd
      tag: host.systemd
      systemd_filter: _SYSTEMD_UNIT=docker.service
      systemd_filter: _SYSTEMD_UNIT=kubelet.service
      read_from_tail: on
      strip_underscores: on
      db: /var/log/flb_systemd.db

pipeline:
  inputs:
    - name: forward
      tag: forward.incoming
      listen: 0.0.0.0
      port: 24224
      buffer_chunk_size: 1M
      buffer_max_size: 6M

pipeline:
  inputs:
    # HTTP受信
    - name: http
      tag: http.logs
      listen: 0.0.0.0
      port: 9880
      successful_response_code: 201

    # TCP受信
    - name: tcp
      tag: tcp.logs
      listen: 0.0.0.0
      port: 5170
      format: json

    # UDP受信
    - name: udp
      tag: udp.logs
      listen: 0.0.0.0
      port: 5170
      format: json

pipeline:
  inputs:
    - name: kubernetes_events
      tag: kube.events
      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
      interval_sec: 1
      retention_time: 1h

pipeline:
  inputs:
    # ノードメトリクス収集
    - name: node_exporter_metrics
      tag: node.metrics
      scrape_interval: 30

    # Prometheusエンドポイントスクレイピング
    - name: prometheus_scrape
      tag: prom.metrics
      host: 127.0.0.1
      port: 9090
      metrics_path: /metrics
      scrape_interval: 10s

pipeline:
  inputs:
    - name: fluentbit_metrics
      tag: fb.metrics
      scrape_interval: 30
      scrape_on_start: true

# parsers.conf

# Nginxエラーログパーサー
[PARSER]
    Name        nginx_error
    Format      regex
    Regex       ^(?<time>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[(?<level>\w+)\] (?<pid>\d+).(?<tid>\d+): (?<message>.*)$
    Time_Key    time
    Time_Format %Y/%m/%d %H:%M:%S

# Spring Bootログパーサー
[PARSER]
    Name        spring_boot
    Format      regex
    Regex       ^(?<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+)\s+(?<level>\w+)\s+(?<pid>\d+)\s+---\s+\[(?<thread>[^\]]+)\]\s+(?<logger>\S+)\s+:\s+(?<message>.*)$
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%L

# Apache Combinedログパーサー
[PARSER]
    Name        apache_combined
    Format      regex
    Regex       ^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

# parsers.yaml
parsers:
  - name: nginx_error
    format: regex
    regex: '^(?<time>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[(?<level>\w+)\] (?<pid>\d+).(?<tid>\d+): (?<message>.*)$'
    time_key: time
    time_format: '%Y/%m/%d %H:%M:%S'

  - name: spring_boot
    format: regex
    regex: '^(?<time>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+)\s+(?<level>\w+)\s+(?<pid>\d+)\s+---\s+\[(?<thread>[^\]]+)\]\s+(?<logger>\S+)\s+:\s+(?<message>.*)$'
    time_key: time
    time_format: '%Y-%m-%dT%H:%M:%S.%L'

# マルチラインパーサー定義
[MULTILINE_PARSER]
    Name          java_stacktrace
    Type          regex
    Flush_Timeout 1000
    # 先頭行パターン：タイムスタンプで開始
    Rule          "start_state"  "/^\d{4}-\d{2}-\d{2}/"  "cont"
    # 継続行パターン：空白またはCaused byで開始
    Rule          "cont"         "/^\s+|^Caused by:/"      "cont"

[MULTILINE_PARSER]
    Name          python_traceback
    Type          regex
    Flush_Timeout 1000
    Rule          "start_state"  "/^Traceback/"            "python_tb"
    Rule          "python_tb"    "/^\s+/"                   "python_tb"
    Rule          "python_tb"    "/^\w+Error/"              "end"

pipeline:
  inputs:
    - name: tail
      tag: app.java
      path: /var/log/app/application.log
      multiline.parser: java_stacktrace
      read_from_head: true

[PARSER]
    Name        custom_time
    Format      regex
    Regex       ^(?<time>[^ ]+) (?<message>.*)$
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%LZ
    Time_Keep   On    # 変換後もtimeフィールドを保持
    Time_Offset +0900  # JSTタイムゾーン

# パーサーテスト用設定
service:
  flush: 1
  log_level: debug

pipeline:
  inputs:
    - name: tail
      tag: test
      path: /tmp/test.log
      parser: my_custom_parser

  outputs:
    - name: stdout
      match: test
      format: json_lines

# テストログの作成と確認
echo '2026-03-01T12:00:00.000Z ERROR [main] App - Connection failed' >> /tmp/test.log
fluent-bit -c test.yaml

pipeline:
  filters:
    - 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 # JSONログをトップレベルにマージ
      merge_log_key: log_parsed # マージキー名
      keep_log: off # 元のlogフィールドを削除
      k8s-logging.parser: on # Podアノテーションのパーサー設定を使用
      k8s-logging.exclude: on # Podアノテーションでログ除外
      labels: on # Pod Labelsを含める
      annotations: off # Pod Annotationsを含めるかどうか
      buffer_size: 0 # APIレスポンスバッファ（0=無制限）
      kube_meta_cache_ttl: 300 # メタデータキャッシュTTL（秒）
      use_kubelet: false # kubelet APIを使用するかどうか

{
  "log": "Connection established to database",
  "kubernetes": {
    "pod_name": "api-server-7d4b8c6f5-x2k9j",
    "namespace_name": "production",
    "pod_id": "abc-123-def",
    "container_name": "api-server",
    "container_image": "myregistry/api-server:v2.1.0",
    "labels": {
      "app": "api-server",
      "version": "v2.1.0",
      "team": "backend"
    },
    "host": "node-01"
  }
}

pipeline:
  filters:
    - name: modify
      match: "*"
      # フィールド追加
      add: environment production
      add: cluster_name main-cluster
      # フィールドリネーム
      rename: log message
      rename: stream source
      # フィールド削除
      remove: unwanted_field
      # 条件付きフィールド追加（キーが存在しない場合のみ）
      set: default_level INFO
      # ハードコピー
      copy: source source_backup

pipeline:
  filters:
    # ERRORまたはWARNレベルのみ通過
    - name: grep
      match: app.*
      regex: level (ERROR|WARN)

    # healthcheckパスを除外
    - name: grep
      match: access.*
      exclude: path /health

    # 特定のNamespaceのみ通過
    - name: grep
      match: kube.*
      regex: $kubernetes['namespace_name'] ^(production|staging)$

    # 複数条件の組み合わせ（AND）
    - name: grep
      match: app.*
      regex: level ERROR
      regex: message .*timeout.*

pipeline:
  filters:
    - name: record_modifier
      match: "*"
      record: hostname ${HOSTNAME}
      record: service_name my-application
      remove_key: unnecessary_field
      allowlist_key: timestamp
      allowlist_key: level
      allowlist_key: message

pipeline:
  filters:
    # ネスト（Nest）：フラット -> ネスト
    - name: nest
      match: '*'
      operation: nest
      wildcard: 'app_*'
      nest_under: application
      # app_name, app_version -> application: { name, version }

    # 解除（Lift）：ネスト -> フラット
    - name: nest
      match: '*'
      operation: lift
      nested_under: kubernetes
      # kubernetes: { pod_name, namespace } -> pod_name, namespace

pipeline:
  filters:
    - name: lua
      match: app.*
      script: /fluent-bit/scripts/transform.lua
      call: process_log

-- /fluent-bit/scripts/transform.lua

function process_log(tag, timestamp, record)
    -- ログレベルの正規化
    if record["level"] then
        record["level"] = string.upper(record["level"])
    end

    -- 機密情報のマスキング
    if record["message"] then
        record["message"] = string.gsub(
            record["message"],
            "%d%d%d%d%-%d%d%d%d%-%d%d%d%d%-%d%d%d%d",
            "****-****-****-****"
        )
    end

    -- レスポンス時間に基づくグレード追加
    if record["response_time"] then
        local rt = tonumber(record["response_time"])
        if rt > 5000 then
            record["performance"] = "critical"
        elseif rt > 1000 then
            record["performance"] = "slow"
        else
            record["performance"] = "normal"
        end
    end

    -- タイムスタンプフィールド追加
    record["processed_at"] = os.date("!%Y-%m-%dT%H:%M:%SZ")

    -- 2 = MODIFIED
    return 2, timestamp, record
end

pipeline:
  filters:
    - name: rewrite_tag
      match: kube.*
      rule: $kubernetes['namespace_name'] ^(production)$ prod.$TAG false
      rule: $kubernetes['namespace_name'] ^(staging)$    stg.$TAG  false
      rule: $level                        ^(ERROR)$      alert.$TAG false

pipeline:
  filters:
    - name: throttle
      match: app.*
      rate: 1000 # ウィンドウあたりの許容量
      window: 5 # ウィンドウサイズ（秒）
      interval: 1s # 評価周期
      print_status: true # ステータス出力

pipeline:
  filters:
    - name: multiline
      match: app.*
      multiline.parser: java_stacktrace
      multiline.key_content: log

pipeline:
  outputs:
    # Elasticsearch
    - name: es
      match: kube.*
      host: ${ES_HOST}
      port: 9200
      index: logs
      type: _doc
      http_user: ${ES_USER}
      http_passwd: ${ES_PASSWORD}
      logstash_format: on
      logstash_prefix: kube-logs
      logstash_dateformat: %Y.%m.%d
      time_key: '@timestamp'
      include_tag_key: true
      tag_key: fluentbit_tag
      generate_id: on # 重複防止ID生成
      buffer_size: 512KB
      tls: on
      tls.verify: on
      tls.ca_file: /certs/ca.pem
      retry_limit: 5
      workers: 2
      suppress_type_name: on # ES 8.x互換

    # OpenSearch
    - name: opensearch
      match: app.*
      host: opensearch.logging.svc
      port: 9200
      index: app-logs
      http_user: admin
      http_passwd: ${OPENSEARCH_PASSWORD}
      tls: on
      suppress_type_name: on
      trace_output: off

pipeline:
  outputs:
    - name: loki
      match: kube.*
      host: loki-gateway.logging.svc
      port: 3100
      uri: /loki/api/v1/push
      tenant_id: my-tenant
      labels: job=fluent-bit
      label_keys: $kubernetes['namespace_name'],$kubernetes['pod_name'],$kubernetes['container_name']
      label_map_path: /fluent-bit/etc/loki-labelmap.json
      remove_keys: kubernetes,stream
      auto_kubernetes_labels: on
      line_format: json
      drop_single_key: on
      http_user: ${LOKI_USER}
      http_passwd: ${LOKI_PASSWORD}
      tls: on
      tls.verify: on
      workers: 2

{
  "kubernetes": {
    "namespace_name": "namespace",
    "pod_name": "pod",
    "container_name": "container",
    "labels": {
      "app": "app"
    }
  },
  "stream": "stream"
}

pipeline:
  outputs:
    - name: s3
      match: archive.*
      region: ap-northeast-2
      bucket: my-log-bucket
      s3_key_format: /logs/$TAG/%Y/%m/%d/%H/$UUID.gz
      s3_key_format_tag_delimiters: .
      total_file_size: 50M # このサイズに到達したらアップロード
      upload_timeout: 10m # サイズ未達でもこの時間後にアップロード
      use_put_object: on
      compression: gzip
      content_type: application/gzip
      store_dir: /tmp/fluent-bit-s3 # ローカルバッファディレクトリ
      store_dir_limit_size: 512M
      retry_limit: 5
      # IAMロール使用時（IRSA / EKS Pod Identity）
      role_arn: arn:aws:iam::123456789012:role/fluent-bit-s3
      # 静的認証情報（非推奨）
      # endpoint:
      # sts_endpoint:

pipeline:
  outputs:
    - name: kafka
      match: app.*
      brokers: kafka-0:9092,kafka-1:9092,kafka-2:9092
      topics: application-logs
      timestamp_key: '@timestamp'
      timestamp_format: iso8601
      format: json
      message_key: log
      queue_full_retries: 10
      rdkafka.request.required.acks: 1
      rdkafka.log.connection.close: false
      rdkafka.compression.codec: snappy

pipeline:
  outputs:
    - name: cloudwatch_logs
      match: kube.*
      region: ap-northeast-2
      log_group_name: /eks/my-cluster/containers
      log_group_template: /eks/my-cluster/$kubernetes['namespace_name']
      log_stream_prefix: from-fluent-bit-
      log_stream_template: $kubernetes['pod_name']
      auto_create_group: on
      extra_user_agent: fluent-bit
      retry_limit: 5
      workers: 1

pipeline:
  outputs:
    - name: stdout
      match: '*'
      format: json_lines # json_lines | msgpack

pipeline:
  outputs:
    - name: forward
      match: '*'
      host: fluentd-aggregator.logging.svc
      port: 24224
      time_as_integer: off
      send_options: true
      require_ack_response: true
      # TLS
      tls: on
      tls.verify: on
      tls.ca_file: /certs/ca.pem
      # 共有キー認証
      shared_key: my-shared-secret
      self_hostname: fluent-bit-node01

pipeline:
  outputs:
    - name: prometheus_exporter
      match: metrics.*
      host: 0.0.0.0
      port: 2021
      add_label: app fluent-bit

service:
  flush: 1
  log_level: info
  # ファイルシステムバッファの有効化
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal # normal | full
  storage.checksum: off # データ整合性検証
  storage.max_chunks_up: 128 # メモリに載せる最大チャンク数
  storage.backlog.mem_limit: 5M # バックログメモリ上限
  storage.metrics: on # ストレージメトリクスの有効化

pipeline:
  inputs:
    # メモリバッファ使用（デフォルト）
    - name: tail
      tag: app.mem
      path: /var/log/app/*.log
      mem_buf_limit: 10MB # メモリバッファ上限

    # ファイルシステムバッファ使用
    - name: tail
      tag: app.fs
      path: /var/log/app/*.log
      storage.type: filesystem # filesystem | memory
      storage.pause_on_chunks_overlimit: off # 上限超過時もディスクに書き込み続行

pipeline:
  outputs:
    - name: es
      match: '*'
      host: elasticsearch
      port: 9200
      storage.total_limit_size: 1G # Output別ファイルシステムバッファ総量制限
      retry_limit: 10 # リトライ回数
      # retry_limit: false          # 無限リトライ

  [Input: データ収集]
         |
         v
  [メモリバッファに格納]
         |
    メモリ上限到達？
    /              \
   No              Yes
   |                |
   v                v
 [収集続行]   storage.type=filesystem?
                /              \
              No               Yes
              |                 |
              v                 v
       [Input一時停止]     [ディスクに書き込み]
       (データ損失リスク)        |
                          ディスク上限到達？
                          /              \
                        No              Yes
                        |                |
                        v                v
                   [書き込み続行]   [Input一時停止]

service:
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal
  storage.max_chunks_up: 128
  storage.backlog.mem_limit: 10M

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      storage.type: filesystem
      storage.pause_on_chunks_overlimit: off

  outputs:
    - name: es
      match: kube.*
      host: elasticsearch
      storage.total_limit_size: 5G
      retry_limit: false
      net.keepalive: on
      net.keepalive_idle_timeout: 30

+---------------------------------------------------------------------+
|                      Kubernetes Cluster                             |
|                                                                     |
|  +-------------------+  +-------------------+  +-------------------+|
|  |     Node 1        |  |     Node 2        |  |     Node 3        ||
|  |                    |  |                    |  |                    ||
|  | +------+ +------+ |  | +------+ +------+ |  | +------+ +------+ ||
|  | |Pod A | |Pod B | |  | |Pod C | |Pod D | |  | |Pod E | |Pod F | ||
|  | +--+---+ +--+---+ |  | +--+---+ +--+---+ |  | +--+---+ +--+---+ ||
|  |    |        |      |  |    |        |      |  |    |        |      ||
|  |    v        v      |  |    v        v      |  |    v        v      ||
|  | /var/log/containers|  | /var/log/containers|  | /var/log/containers||
|  |         |          |  |         |          |  |         |          ||
|  |    +----+-----+    |  |    +----+-----+    |  |    +----+-----+    ||
|  |    |Fluent Bit|    |  |    |Fluent Bit|    |  |    |Fluent Bit|    ||
|  |    |(DaemonSet)|   |  |    |(DaemonSet)|   |  |    |(DaemonSet)|   ||
|  |    +----+-----+    |  |    +----+-----+    |  |    +----+-----+    ||
|  +---------|----------+  +---------|----------+  +---------|----------+|
|            |                       |                       |           |
|            +----------+------------+----------+------------+           |
|                       |                       |                        |
|               +-------v--------+     +--------v--------+              |
|               | Elasticsearch  |     |  Grafana Loki   |              |
|               +----------------+     +-----------------+              |
+---------------------------------------------------------------------+

# リポジトリ追加
helm repo add fluent https://fluent.github.io/helm-charts
helm repo update

# 基本インストール
helm install fluent-bit fluent/fluent-bit \
  --namespace logging \
  --create-namespace

# インストール確認
kubectl get pods -n logging -l app.kubernetes.io/name=fluent-bit
kubectl get ds -n logging

# custom-values.yaml
kind: DaemonSet

image:
  repository: cr.fluentbit.io/fluent/fluent-bit
  tag: '3.2.2'
  pullPolicy: IfNotPresent

resources:
  requests:
    cpu: 100m
    memory: 128Mi
  limits:
    cpu: 500m
    memory: 256Mi

tolerations:
  - operator: Exists # 全ノードにデプロイ（master含む）

serviceAccount:
  create: true
  annotations:
    # IRSA（EKS）
    eks.amazonaws.com/role-arn: arn:aws:iam::123456789012:role/fluent-bit

# ボリュームマウント
volumeMounts:
  - name: varlog
    mountPath: /var/log
  - name: varlibdockercontainers
    mountPath: /var/lib/docker/containers
    readOnly: true
  - name: etcmachineid
    mountPath: /etc/machine-id
    readOnly: true

volumes:
  - name: varlog
    hostPath:
      path: /var/log
  - name: varlibdockercontainers
    hostPath:
      path: /var/lib/docker/containers
  - name: etcmachineid
    hostPath:
      path: /etc/machine-id

# 環境変数
env:
  - name: ELASTICSEARCH_HOST
    value: 'elasticsearch-master.logging.svc.cluster.local'
  - name: ELASTICSEARCH_PORT
    value: '9200'

# Fluent Bit設定
config:
  service: |
    [SERVICE]
        Flush         5
        Log_Level     info
        Daemon        off
        Parsers_File  /fluent-bit/etc/parsers.conf
        HTTP_Server   On
        HTTP_Listen   0.0.0.0
        HTTP_Port     2020
        storage.path  /var/log/fluent-bit/buffer/
        storage.sync  normal
        storage.max_chunks_up 128
        Hot_Reload    On

  inputs: |
    [INPUT]
        Name              tail
        Tag               kube.*
        Path              /var/log/containers/*.log
        multiline.parser  docker, cri
        DB                /var/log/flb_kube.db
        Mem_Buf_Limit     5MB
        Skip_Long_Lines   On
        Refresh_Interval  10
        storage.type      filesystem

  filters: |
    [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
        Kube_Tag_Prefix     kube.var.log.containers.
        Merge_Log           On
        Keep_Log            Off
        K8S-Logging.Parser  On
        K8S-Logging.Exclude On
        Labels              On
        Annotations         Off

  outputs: |
    [OUTPUT]
        Name            es
        Match           kube.*
        Host            ${ELASTICSEARCH_HOST}
        Port            ${ELASTICSEARCH_PORT}
        Logstash_Format On
        Logstash_Prefix kube
        Retry_Limit     False
        Suppress_Type_Name On

# Prometheusメトリクス収集のためのServiceMonitor
serviceMonitor:
  enabled: true
  interval: 30s
  scrapeTimeout: 10s

# Liveness/Readiness Probe
livenessProbe:
  httpGet:
    path: /
    port: http
readinessProbe:
  httpGet:
    path: /api/v1/health
    port: http

/var/log/containers/<pod-name>_<namespace>_<container-name>-<container-id>.log
  -> symlink -> /var/log/pods/<namespace>_<pod-name>_<pod-uid>/<container-name>/0.log

2026-03-01T12:00:00.123456789Z stdout F This is a log message
2026-03-01T12:00:00.123456789Z stderr P This is a partial log line

# Namespaceに応じて異なるOutputにルーティング
pipeline:
  filters:
    - name: kubernetes
      match: kube.*
      merge_log: on
      labels: on

    # プロダクションNamespaceのタグ変更
    - name: rewrite_tag
      match: kube.*
      rule: $kubernetes['namespace_name'] ^(production)$ prod.$TAG false
      rule: $kubernetes['namespace_name'] ^(staging)$    stg.$TAG  false
      rule: $kubernetes['namespace_name'] ^(monitoring)$ mon.$TAG  false

  outputs:
    # プロダクションログ -> Elasticsearch
    - name: es
      match: prod.*
      host: es-production.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: prod-logs

    # ステージングログ -> Loki
    - name: loki
      match: stg.*
      host: loki.logging.svc
      port: 3100
      labels:
        env=staging

    # モニタリングログ -> S3アーカイブ
    - name: s3
      match: mon.*
      bucket: monitoring-logs-archive
      region: ap-northeast-2
      total_file_size: 100M
      upload_timeout: 10m
      compression: gzip

    # その他のログ -> デフォルトElasticsearch
    - name: es
      match: kube.*
      host: es-default.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: default-logs

pipeline:
  filters:
    - name: kubernetes
      match: kube.*
      merge_log: on
      labels: on

    # チームLabelベースのタグ書き換え
    - name: rewrite_tag
      match: kube.*
      rule: $kubernetes['labels']['team'] ^(backend)$  team.backend.$TAG  false
      rule: $kubernetes['labels']['team'] ^(frontend)$ team.frontend.$TAG false
      rule: $kubernetes['labels']['team'] ^(data)$     team.data.$TAG     false

  outputs:
    # バックエンドチーム -> 専用Elasticsearchインデックス
    - name: es
      match: team.backend.*
      host: elasticsearch.logging.svc
      logstash_format: on
      logstash_prefix: team-backend

    # フロントエンドチーム -> 専用Loki tenant
    - name: loki
      match: team.frontend.*
      host: loki.logging.svc
      tenant_id: frontend-team
      labels:
        team=frontend

    # データチーム -> S3 + Elasticsearchデュアル
    - name: es
      match: team.data.*
      host: elasticsearch.logging.svc
      logstash_format: on
      logstash_prefix: team-data

    - name: s3
      match: team.data.*
      bucket: data-team-logs
      region: ap-northeast-2
      compression: gzip

# fluent-bit-rbac.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluent-bit
  namespace: logging
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluent-bit
rules:
  - apiGroups: ['']
    resources:
      - namespaces
      - pods
      - pods/logs
      - nodes
      - nodes/proxy
    verbs: ['get', 'list', 'watch']
  - apiGroups: ['']
    resources:
      - events
    verbs: ['get', 'list', 'watch']
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluent-bit
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluent-bit
subjects:
  - kind: ServiceAccount
    name: fluent-bit
    namespace: logging

# fluent-bit-efk.yaml
service:
  flush: 5
  log_level: info
  parsers_file: /fluent-bit/etc/parsers.conf
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal
  storage.max_chunks_up: 128

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      multiline.parser: docker, cri
      db: /var/log/flb_kube.db
      mem_buf_limit: 5MB
      skip_long_lines: on
      refresh_interval: 10
      storage.type: filesystem

  filters:
    - 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
      keep_log: off
      k8s-logging.parser: on
      k8s-logging.exclude: on
      labels: on

    # kube-system、logging Namespaceのログを除外
    - name: grep
      match: kube.*
      exclude: $kubernetes['namespace_name'] ^(kube-system|logging)$

    # 環境情報の追加
    - name: modify
      match: kube.*
      add: cluster_name my-eks-cluster
      add: environment production

  outputs:
    - name: es
      match: kube.*
      host: elasticsearch-master.logging.svc.cluster.local
      port: 9200
      http_user: elastic
      http_passwd: ${ES_PASSWORD}
      logstash_format: on
      logstash_prefix: kube-logs
      logstash_dateformat: "%Y.%m.%d"
      time_key: "@timestamp"
      include_tag_key: true
      tag_key: fluentbit_tag
      suppress_type_name: on
      generate_id: on
      retry_limit: false
      workers: 2
      tls: on
      tls.verify: on
      tls.ca_file: /certs/es-ca.pem

# fluent-bit-loki.yaml
service:
  flush: 5
  log_level: info
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      multiline.parser: docker, cri
      db: /var/log/flb_kube.db
      mem_buf_limit: 5MB
      skip_long_lines: on

  filters:
    - name: kubernetes
      match: kube.*
      merge_log: on
      keep_log: off
      labels: on
      annotations: off

  outputs:
    - name: loki
      match: kube.*
      host: loki-gateway.logging.svc.cluster.local
      port: 3100
      uri: /loki/api/v1/push
      labels: job=fluent-bit,cluster=my-cluster
      label_keys: $kubernetes['namespace_name'],$kubernetes['pod_name'],$kubernetes['container_name']
      auto_kubernetes_labels: on
      line_format: json
      workers: 2
      retry_limit: false

# fluent-bit-dual.yaml
service:
  flush: 5
  log_level: info
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal
  storage.max_chunks_up: 128

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      multiline.parser: docker, cri
      db: /var/log/flb_kube.db
      storage.type: filesystem

  filters:
    - name: kubernetes
      match: kube.*
      merge_log: on
      keep_log: off
      labels: on

  outputs:
    # リアルタイム検索/分析用Elasticsearch
    - name: es
      match: kube.*
      host: elasticsearch.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: kube-logs
      suppress_type_name: on
      retry_limit: false
      workers: 2

    # 長期保管用S3アーカイブ
    - name: s3
      match: kube.*
      region: ap-northeast-2
      bucket: log-archive-bucket
      s3_key_format: /kubernetes/$TAG/%Y/%m/%d/%H/$UUID.gz
      s3_key_format_tag_delimiters: .
      total_file_size: 100M
      upload_timeout: 10m
      compression: gzip
      content_type: application/gzip
      store_dir: /tmp/fluent-bit-s3
      store_dir_limit_size: 1G
      use_put_object: on
      retry_limit: 5
      workers: 1

# fluent-bit-routing.yaml
service:
  flush: 5
  log_level: info
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      multiline.parser: docker, cri
      db: /var/log/flb_kube.db
      storage.type: filesystem

  filters:
    - name: kubernetes
      match: kube.*
      merge_log: on
      labels: on

    # Namespace別タグ書き換え
    - name: rewrite_tag
      match: kube.*
      rule: $kubernetes['namespace_name'] ^(production)$  route.prod.$TAG  false
      rule: $kubernetes['namespace_name'] ^(staging)$     route.stg.$TAG   false
      rule: $kubernetes['namespace_name'] ^(kube-system)$ route.sys.$TAG   false

    # プロダクションからERRORのみ抽出
    - name: rewrite_tag
      match: route.prod.*
      rule: $log ^.*ERROR.*$ alert.prod.$TAG true

  outputs:
    # プロダクション全ログ -> Elasticsearch
    - name: es
      match: route.prod.*
      host: es-prod.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: prod-all
      suppress_type_name: on
      workers: 2

    # プロダクションERRORアラート -> 別インデックス
    - name: es
      match: alert.prod.*
      host: es-prod.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: prod-alerts
      suppress_type_name: on

    # ステージングログ -> Loki（コスト効率）
    - name: loki
      match: route.stg.*
      host: loki.logging.svc
      port: 3100
      labels:
        env=staging
      auto_kubernetes_labels: on
      line_format: json

    # システムログ -> S3アーカイブ（長期保管）
    - name: s3
      match: route.sys.*
      bucket: system-logs-archive
      region: ap-northeast-2
      total_file_size: 50M
      upload_timeout: 15m
      compression: gzip
      s3_key_format: /kube-system/%Y/%m/%d/$UUID.gz

    # 未分類ログ -> デフォルトElasticsearch
    - name: es
      match: kube.*
      host: es-default.logging.svc
      port: 9200
      logstash_format: on
      logstash_prefix: default-logs
      suppress_type_name: on

pipeline:
  outputs:
    - name: es
      match: '*'
      host: elasticsearch
      port: 9200
      workers: 4 # 4つの並列Worker
      net.keepalive: on
      net.keepalive_idle_timeout: 30

service:
  flush: 1 # 1秒ごとにフラッシュ（リアルタイム性重視）
  # flush: 5  # 5秒ごとにフラッシュ（スループット重視）

pipeline:
  inputs:
    - name: tail
      path: /var/log/containers/*.log
      # 高スループット環境
      buffer_chunk_size: 512KB # チャンク単位サイズ（デフォルト32KB）
      buffer_max_size: 5MB # 最大バッファサイズ（デフォルト32KB）
      mem_buf_limit: 50MB # メモリバッファ上限

  outputs:
    - name: es
      match: '*'
      buffer_size: 512KB # HTTPバッファサイズ

pipeline:
  inputs:
    # アプリケーションログパイプライン
    - name: tail
      tag: app.*
      path: /var/log/containers/app-*.log
      multiline.parser: docker, cri
      threaded: on # 別スレッドで実行

    # システムログパイプライン
    - name: tail
      tag: sys.*
      path: /var/log/containers/kube-*.log
      multiline.parser: docker, cri
      threaded: on

  outputs:
    - name: es
      match: app.*
      host: elasticsearch
      workers: 4

    - name: es
      match: sys.*
      host: elasticsearch
      workers: 2

# Hot Reloadの有効化
service:
  hot_reload: on

# 方法1：SIGHUPシグナル
kill -SIGHUP $(pidof fluent-bit)

# 方法2：HTTP API（v4.0+）
curl -X POST http://localhost:2020/api/v2/reload

# 方法3：コマンドラインオプション
fluent-bit -c fluent-bit.yaml -Y   # --enable-hot-reload

# Fluent Bitプロセスのメモリ確認
ps aux | grep fluent-bit

# HTTP APIで内部メトリクス確認
curl -s http://localhost:2020/api/v1/storage | jq .

# Prometheusメトリクスで確認
curl -s http://localhost:2020/api/v2/metrics/prometheus | grep fluentbit_input

service:
  http_server: on
  http_listen: 0.0.0.0
  http_port: 2020
  health_check: on
  hc_errors_count: 5 # この数以上のエラー発生時にunhealthy
  hc_retry_failure_count: 5 # この数以上のリトライ失敗時にunhealthy
  hc_period: 60 # ヘルスチェック周期（秒）

# ヘルスチェック
curl -s http://localhost:2020/api/v1/health
# レスポンス: ok (HTTP 200) または error (HTTP 500)

# ストレージ状態確認
curl -s http://localhost:2020/api/v1/storage | jq .
# {
#   "storage_layer": {
#     "chunks": {
#       "total_chunks": 15,
#       "mem_chunks": 10,
#       "fs_chunks": 5,
#       "fs_chunks_up": 3,
#       "fs_chunks_down": 2
#     }
#   }
# }

# Prometheusメトリクス確認
curl -s http://localhost:2020/api/v2/metrics/prometheus

# Fluent Bit自体のメトリクスをPrometheusに公開
pipeline:
  inputs:
    - name: fluentbit_metrics
      tag: fb.metrics
      scrape_interval: 30

  outputs:
    - name: prometheus_exporter
      match: fb.metrics
      host: 0.0.0.0
      port: 2021

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: fluent-bit
  namespace: logging
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app.kubernetes.io/name: fluent-bit
  endpoints:
    - port: http
      path: /api/v2/metrics/prometheus
      interval: 30s
      scrapeTimeout: 10s
  namespaceSelector:
    matchNames:
      - logging

# 1. Fluent Bitプロセス状態確認
kubectl get pods -n logging -l app.kubernetes.io/name=fluent-bit
kubectl logs -n logging <fluent-bit-pod> --tail=50

# 2. 設定ファイルの構文検証
fluent-bit -c fluent-bit.yaml --dry-run

# 3. ログファイルパス確認
kubectl exec -n logging <fluent-bit-pod> -- ls -la /var/log/containers/

# 4. DBファイル（offset）確認
kubectl exec -n logging <fluent-bit-pod> -- ls -la /var/log/flb_kube.db

# 5. 権限確認
kubectl exec -n logging <fluent-bit-pod> -- cat /var/log/containers/<target-log>

# メモリ制限設定
pipeline:
  inputs:
    - name: tail
      path: /var/log/containers/*.log
      mem_buf_limit: 5MB # Inputメモリ上限
      skip_long_lines: on # 長い行をスキップ
      buffer_chunk_size: 32KB # チャンクサイズ制限
      buffer_max_size: 32KB # 最大バッファ

service:
  storage.path: /var/log/fluent-bit/buffer/
  storage.max_chunks_up: 64 # メモリチャンク数制限（デフォルト128）

# Backpressure発生確認
curl -s http://localhost:2020/api/v1/storage | jq .

# メトリクスで確認
curl -s http://localhost:2020/api/v2/metrics/prometheus | grep -E "retries|errors|dropped"

# Backpressure対応設定例
service:
  flush: 1
  storage.path: /var/log/fluent-bit/buffer/
  storage.sync: normal
  storage.max_chunks_up: 256

pipeline:
  inputs:
    - name: tail
      tag: kube.*
      path: /var/log/containers/*.log
      storage.type: filesystem
      storage.pause_on_chunks_overlimit: off

  outputs:
    - name: es
      match: kube.*
      host: elasticsearch
      port: 9200
      workers: 4
      storage.total_limit_size: 10G
      retry_limit: false
      net.keepalive: on
      net.keepalive_idle_timeout: 15

# TLS接続テスト
kubectl exec -n logging <fluent-bit-pod> -- \
  curl -v --cacert /certs/ca.pem https://elasticsearch:9200

# 証明書の有効期限確認
kubectl exec -n logging <fluent-bit-pod> -- \
  openssl x509 -in /certs/ca.pem -noout -enddate

service:
  log_level: debug # error, warn, info, debug, trace

pipeline:
  outputs:
    # デバッグ用：すべてのレコードを標準出力へ
    - name: stdout
      match: '*'
      format: json_lines

    # 実際のOutput
    - name: es
      match: '*'
      host: elasticsearch

# Inputのみテスト
fluent-bit -i tail -p path=/var/log/test.log -o stdout

# Parserテスト
fluent-bit -i tail -p path=/var/log/test.log -p parser=json -o stdout

# 全設定テスト（dry-run）
fluent-bit -c fluent-bit.yaml --dry-run

# Fluent Bitログストリーミング
kubectl logs -n logging -l app.kubernetes.io/name=fluent-bit -f --tail=100

# 特定Podのログ
kubectl logs -n logging fluent-bit-xxxxx -f

# 前のコンテナのログ（crash時）
kubectl logs -n logging fluent-bit-xxxxx --previous

# Kubernetes Pod Security
securityContext:
  runAsNonRoot: true
  runAsUser: 1000
  readOnlyRootFilesystem: true
  capabilities:
    drop: ['ALL']

# 必要なボリュームのみマウント
volumes:
  - name: varlog
    hostPath:
      path: /var/log
      type: ''
  - name: buffer
    emptyDir:
      sizeLimit: 2Gi

# TLS強制（Output）
pipeline:
  outputs:
    - name: es
      tls: on
      tls.verify: on
      tls.ca_file: /certs/ca.pem

pipeline:
  inputs:
    - name: tail
      path: /var/log/app/*.log
      db: /var/log/flb_app.db # 必須：offset追跡
      rotate_wait: 5 # ローテーション後の待ち時間（秒）
      refresh_interval: 10 # ファイルリスト更新周期