Source: Input plugins

プラットフォーム	コマンド
Ubuntu/Debian (td-agent)	`curl -fsSL https://toolbelt.treasuredata.com/sh/install-ubuntu-jammy-td-agent4.sh \
RHEL/CentOS	`curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent4.sh \
macOS	`brew install fluentd`
Ruby Gem	`gem install fluentd`
Docker	`docker pull fluent/fluentd:latest`
Kubernetes	DaemonSet としてデプロイ（設定セクションを参照）

コマンド	説明
`fluentd -c fluent.conf`	指定された設定ファイルで Fluentd を起動する
`fluentd -c fluent.conf -vv`	詳細なデバッグ出力で実行する
`fluentd -c fluent.conf --dry-run`	設定を開始せずに検証する
`fluentd --setup ./fluent`	デフォルトの設定ディレクトリ構造を作成する
`fluentd --version`	Fluentdのバージョン情報を表示
`sudo systemctl start td-agent`	td-agentサービスを起動 (Linux)
`sudo systemctl stop td-agent`	td-agentサービスを停止
`sudo systemctl restart td-agent`	td-agentサービスを再起動
`sudo systemctl status td-agent`	td-agentサービスのステータスを確認
`sudo systemctl reload td-agent`	再起動せずに設定を再読み込み
`sudo systemctl enable td-agent`	td-agentをブート時に起動できるようにする
`sudo journalctl -u td-agent -f`	td-agentサービスのログをリアルタイムで追跡する
`echo ’{“msg”:“test”}’ \	fluent-cat debug.test`
`curl -X POST -d 'json={"event":"test"}' http://localhost:8888/test.cycle`	HTTPテストログを送信
`td-agent-gem list \	grep fluent-plugin`

コマンド	説明
`fluentd -c fluent.conf -d /var/run/fluentd.pid`	PIDファイル付きでデーモンモードでFluentdを実行
`fluentd -c fluent.conf -o /var/log/fluentd.log`	特定のログファイルに出力して実行する
`fluentd -c fluent.conf --workers 4`	複数のワーカープロセスで実行
`fluentd -c fluent.conf -vvv`	デバッグのためにトレースレベルのロギングで実行する
`fluentd --show-plugin-config=input:tail`	特定のプラグインの設定オプションを表示
`td-agent-gem install fluent-plugin-elasticsearch`	Elasticsearch出力プラグインをインストール
`td-agent-gem install fluent-plugin-kafka -v 0.17.5`	Kafkaプラグインの特定のバージョンをインストール
`td-agent-gem update fluent-plugin-s3`	S3プラグインを最新バージョンに更新
`td-agent-gem uninstall fluent-plugin-mongo`	MongoDB プラグインを削除
`td-agent-gem search -r fluent-plugin`	リポジトリで利用可能なプラグインを検索
`fluent-cat --host 192.168.1.100 --port 24224 app.logs`	リモートの Fluentd インスタンスにログを送信
`fluent-cat app.logs < /path/to/logfile.json`	Fluentdにログファイルの内容を送信
`docker run -d -p 24224:24224 -v /data/fluentd:/fluentd/etc fluent/fluentd`	Dockerでマウントされた設定で Fluentd を実行する
`sudo kill -USR1 $(cat /var/run/td-agent/td-agent.pid)`	Fluentdを優雅に再読み込み（ログファイルの再オープン）
`sudo kill -USR2 $(cat /var/run/td-agent/td-agent.pid)`	Fluentdのログファイルをリロードせずに再オープンする
`/etc/td-agent/td-agent.conf`### メイン設定ファイルの場所

td-agent (Linux): ./fluent/fluent.conf
Gemインストール: /fluentd/etc/fluent.conf
Docker:

# Source: Input plugins
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

# Filter: Process/transform logs
<filter app.**>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
    tag ${tag}
  </record>
</filter>

# Match: Output plugins
<match app.**>
  @type elasticsearch
  host elasticsearch.local
  port 9200
  index_name fluentd
  type_name fluentd
</match>
```### 基本的な設定構造
```ruby
# Forward input (receive from other Fluentd instances)
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

# Tail log files
<source>
  @type tail
  path /var/log/nginx/access.log
  pos_file /var/log/td-agent/nginx-access.pos
  tag nginx.access
  <parse>
    @type nginx
  </parse>
</source>

# HTTP input
<source>
  @type http
  port 8888
  bind 0.0.0.0
  body_size_limit 32m
  keepalive_timeout 10s
</source>

# Syslog input
<source>
  @type syslog
  port 5140
  bind 0.0.0.0
  tag system.syslog
</source>
```### ソースプラグイン（入力）
```ruby
# Add/modify record fields
<filter app.**>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
    environment production
    timestamp ${time}
  </record>
</filter>

# Parse unstructured logs
<filter app.logs>
  @type parser
  key_name message
  <parse>
    @type json
  </parse>
</filter>

# Grep filter (include/exclude)
<filter app.**>
  @type grep
  <regexp>
    key level
    pattern /^(ERROR|FATAL)$/
  </regexp>
</filter>

# Modify tag
<match app.raw.**>
  @type rewrite_tag_filter
  <rule>
    key level
    pattern /^ERROR$/
    tag app.error.${tag}
  </rule>
</match>
```### フィルタープラグイン（処理）
```ruby
# Elasticsearch output
<match app.**>
  @type elasticsearch
  host elasticsearch.local
  port 9200
  logstash_format true
  logstash_prefix fluentd
  <buffer>
    @type file
    path /var/log/fluentd/buffer/elasticsearch
    flush_interval 10s
    retry_max_interval 300s
  </buffer>
</match>

# S3 output
<match logs.**>
  @type s3
  aws_key_id YOUR_AWS_KEY_ID
  aws_sec_key YOUR_AWS_SECRET_KEY
  s3_bucket your-bucket-name
  s3_region us-east-1
  path logs/
  time_slice_format %Y%m%d%H
  <buffer time>
    timekey 3600
    timekey_wait 10m
  </buffer>
</match>

# File output
<match debug.**>
  @type file
  path /var/log/fluentd/output
  <buffer>
    timekey 1d
    timekey_use_utc true
  </buffer>
</match>

# Forward to another Fluentd
<match forward.**>
  @type forward
  <server>
    host 192.168.1.100
    port 24224
  </server>
  <buffer>
    @type file
    path /var/log/fluentd/buffer/forward
  </buffer>
</match>

# Stdout (debugging)
<match debug.**>
  @type stdout
</match>
```### マッチプラグイン（出力）
```ruby
<match pattern.**>
  @type elasticsearch
  
  # File buffer with advanced settings
  <buffer>
    @type file
    path /var/log/fluentd/buffer
    
    # Flush settings
    flush_mode interval
    flush_interval 10s
    flush_at_shutdown true
    
    # Retry settings
    retry_type exponential_backoff
    retry_wait 10s
    retry_max_interval 300s
    retry_timeout 72h
    retry_max_times 17
    
    # Chunk settings
    chunk_limit_size 5M
    queue_limit_length 32
    overflow_action drop_oldest_chunk
    
    # Compression
    compress gzip
  </buffer>
</match>

# Memory buffer for high-performance
<match fast.**>
  @type forward
  <buffer>
    @type memory
    flush_interval 5s
    chunk_limit_size 1M
    queue_limit_length 64
  </buffer>
</match>
```### バッファ設定
```ruby
<system>
  workers 4
  root_dir /var/log/fluentd
</system>

# Worker-specific sources
<worker 0>
  <source>
    @type forward
    port 24224
  </source>
</worker>

<worker 1-3>
  <source>
    @type tail
    path /var/log/app/*.log
    tag app.logs
  </source>
</worker>
```### マルチワーカー設定
```ruby
# Route to different pipelines using labels
<source>
  @type forward
  @label @mainstream
</source>

<source>
  @type tail
  path /var/log/secure.log
  @label @security
</source>

<label @mainstream>
  <filter **>
    @type record_transformer
    <record>
      pipeline mainstream
    </record>
  </filter>
  
  <match **>
    @type elasticsearch
    host es-main
  </match>
</label>

<label @security>
  <filter **>
    @type grep
    <regexp>
      key message
      pattern /authentication failure/
    </regexp>
  </filter>
  
  <match **>
    @type s3
    s3_bucket security-logs
  </match>
</label>
```### ラベルベースのルーティング
```bash
# Install Elasticsearch plugin
sudo td-agent-gem install fluent-plugin-elasticsearch

# Configure Fluentd
sudo tee /etc/td-agent/td-agent.conf > /dev/null <<'EOF'
<source>
  @type tail
  path /var/log/nginx/access.log
  pos_file /var/log/td-agent/nginx-access.pos
  tag nginx.access
  <parse>
    @type nginx
  </parse>
</source>

<match nginx.access>
  @type elasticsearch
  host localhost
  port 9200
  logstash_format true
  logstash_prefix nginx
  <buffer>
    flush_interval 10s
  </buffer>
</match>
EOF

# Restart td-agent
sudo systemctl restart td-agent

# Verify logs are flowing
sudo journalctl -u td-agent -f
```## 一般的なユースケース

### ユースケース1: NginxログをElasticsearchに収集
```yaml
# Deploy Fluentd DaemonSet
kubectl apply -f - <<'EOF'
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd
  namespace: kube-system
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd
rules:
- apiGroups: [""]
  resources: ["pods", "namespaces"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluentd
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluentd
subjects:
- kind: ServiceAccount
  name: fluentd
  namespace: kube-system
---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: kube-system
spec:
  selector:
    matchLabels:
      k8s-app: fluentd-logging
  template:
    metadata:
      labels:
        k8s-app: fluentd-logging
    spec:
      serviceAccountName: fluentd
      containers:
      - name: fluentd
        image: fluent/fluentd-kubernetes-daemonset:v1-debian-elasticsearch
        env:
        - name: FLUENT_ELASTICSEARCH_HOST
          value: "elasticsearch.logging.svc.cluster.local"
        - name: FLUENT_ELASTICSEARCH_PORT
          value: "9200"
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
EOF

# Check DaemonSet status
kubectl get daemonset -n kube-system fluentd
kubectl logs -n kube-system -l k8s-app=fluentd-logging --tail=50
```### ユースケース2: Kubernetesログ収集
```bash
# Install S3 plugin
sudo td-agent-gem install fluent-plugin-s3

# Configure S3 output
sudo tee /etc/td-agent/td-agent.conf > /dev/null <<'EOF'
<source>
  @type tail
  path /var/log/app/*.log
  pos_file /var/log/td-agent/app.pos
  tag app.logs
  <parse>
    @type json
  </parse>
</source>

<match app.logs>
  @type s3
  
  aws_key_id YOUR_AWS_ACCESS_KEY
  aws_sec_key YOUR_AWS_SECRET_KEY
  s3_bucket my-application-logs
  s3_region us-east-1
  
  path logs/%Y/%m/%d/
  s3_object_key_format %{path}%{time_slice}_%{index}.%{file_extension}
  
  <buffer time>
    @type file
    path /var/log/td-agent/s3
    timekey 3600
    timekey_wait 10m
    chunk_limit_size 256m
  </buffer>
  
  <format>
    @type json
  </format>
</match>
EOF

# Restart and verify
sudo systemctl restart td-agent
sudo systemctl status td-agent
```### ユースケース3: S3へのログ転送とローテーション
```bash
# Configure routing to multiple destinations
sudo tee /etc/td-agent/td-agent.conf > /dev/null <<'EOF'
<source>
  @type tail
  path /var/log/app/application.log
  pos_file /var/log/td-agent/app.pos
  tag app.logs
  <parse>
    @type json
  </parse>
</source>

# Copy logs to multiple destinations
<match app.logs>
  @type copy
  
  # Send to Elasticsearch
  <store>
    @type elasticsearch
    host elasticsearch.local
    port 9200
    logstash_format true
  </store>
  
  # Send to S3 for archival
  <store>
    @type s3
    s3_bucket app-logs-archive
    path logs/
    <buffer time>
      timekey 86400
    </buffer>
  </store>
  
  # Send errors to Slack
  <store>
    @type grep
    <regexp>
      key level
      pattern /^ERROR$/
    </regexp>
    @type slack
    webhook_url https://hooks.slack.com/services/YOUR/WEBHOOK/URL
    channel alerts
    username fluentd
  </store>
</match>
EOF

sudo systemctl restart td-agent
```### ユースケース4: 複数の宛先へのログルーティング
```bash
# Configure APM log forwarding
sudo tee /etc/td-agent/td-agent.conf > /dev/null <<'EOF'
<source>
  @type tail
  path /var/log/app/*.log
  pos_file /var/log/td-agent/app.pos
  tag app.logs
  <parse>
    @type json
    time_key timestamp
    time_format %Y-%m-%dT%H:%M:%S.%NZ
  </parse>
</source>

# Enrich logs with metadata
<filter app.logs>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
    environment ${ENV['ENVIRONMENT'] || 'production'}
    service_name myapp
    trace_id ${record['trace_id']}
  </record>
</filter>

# Calculate response time metrics
<filter app.logs>
  @type prometheus
  <metric>
    name http_request_duration_seconds
    type histogram
    desc HTTP request duration
    key response_time
  </metric>
</filter>

# Forward to APM system
<match app.logs>
  @type http
  endpoint http://apm-server:8200/intake/v2/events
  <buffer>
    flush_interval 5s
  </buffer>
</match>
EOF

sudo systemctl restart td-agent
```### ユースケース5: アプリケーションパフォーマンス監視の統合
`pos_file`テール入力用に適切に設定し`timekey`バッファ内の値をディスクスペースの問題を防ぐために`rotate_age`および`rotate_size`をファイル出力に使用する

- **ログを階層的にタグ付け**: ドット表記のタグ（例：`app.production.web`）を使用して、柔軟なルーティングとフィルタリングを可能にします。これにより、パターンマッチングが可能になります（例：`app.**`または`app.production.*`）。

- **Fluentdのパフォーマンスを監視**: バッファキューの長さ、リトライ回数、および送信レートを追跡します。ボトルネックを検出するために、Prometheusプラグインまたは組み込みの監視を使用します。

- **機密データを保護**: `@type secure_forward`を使用して暗号化されたログ送信を行い、`record_modifier`で機密フィールドをフィルタリングし、認証情報を含む設定ファイルのファイル権限を制限します。

- **設定変更をテスト**: 展開前に設定構文を検証するために常に`--dry-run`を使用します。本番環境に適用する前に、少量のログボリュームでルーティングロジックをテストします。

- **マルチワーカーモードを慎重に使用**: CPU集中型の処理（解析、フィルタリング）にワーカーを有効にしますが、一部のプラグインがマルチワーカーモードをサポートしていないことに注意してください。2〜4のワーカーから始め、CPU使用率を監視します。

- **グレースフルな劣化を実装**: バックプレッシャーを処理するためにバッファに`overflow_action`を設定します（要件に応じて`drop_oldest_chunk`または`block`を使用）。無限のリトライを防ぐために、適切な`retry_timeout`値を設定します。

- **ラベルで関心事を分離**: `@label`ディレクティブを使用して、異なるログタイプ用の分離された処理パイプラインを作成します。これにより、保守性が向上し、意図しないルーティングを防ぎます。

## トラブルシューティング

| 問題 | ソリューション |
|-------|----------|
| **Fluentd won't start** | Check syntax: `fluentd -c fluent.conf --dry-run`. Review logs: `sudo journalctl -u td-agent -n 100`. Verify file permissions on config and buffer directories. |
| **Logs not being collected** | Verify `pos_file` exists and is writable. Check file path patterns match actual log locations. Ensure log files have read permissions. Test with `tail -f` on the log file. |
| **High memory usage** | Switch from memory buffers to file buffers. Reduce `chunk_limit_size` and `queue_limit_length`. Enable multi-worker mode to distribute load. Check for memory leaks in custom plugins. |
| **Buffer queue growing** | Increase `flush_interval` or reduce log volume. Check downstream system capacity (Elasticsearch, S3). Verify network connectivity. Review `retry_max_interval` settings. |
| **Logs being dropped** | Check buffer `overflow_action` setting. Increase `queue_limit_length` and `chunk_limit_size`. Monitor disk space for file buffers. Review `retry_timeout` configuration. |
| **Plugin installation fails** | Ensure Ruby development headers installed: `sudo apt-get install ruby-dev build-essential`. Use correct gem command: `td-agent-gem` not `gem`. Check plugin compatibility with Fluentd version. |
| **Parse errors in logs** | Validate parser configuration with sample logs. Use `@type regexp` with proper regex patterns. Add error handling: `emit_invalid_record_to_error true`. Check time format strings. |
| **Cannot connect to Elasticsearch** | Verify Elasticsearch is running: `curl http://elasticsearch:9200`. Check firewall rules. Validate credentials if using authentication. Review Elasticsearch logs for rejection reasons. |
| **Duplicate logs appearing** | Check `pos_file` location is persistent across restarts. Verify only one Fluentd instance is running. Review `read_from_head` setting (should be `false` in production). |
| **Slow log processing** | Enable multi-worker mode. Optimize regex patterns in filters. Use `@type grep` before expensive parsers. Profile with `--trace` flag to identify bottlenecks. |
| **SSL/TLS connection errors** | Verify certificate paths and permissions. Check certificate expiration dates. Ensure CA bundle is up to date. Use `verify_ssl false` for testing only (not production). || **タイムゾーンの問題** | タイムパーサーで`utc`または`localtime`を設定します。タイムゾーン：`time_format`で`%Y-%m-%dT%H:%M:%S%z`を使用します。システムのタイムゾーンを正しく設定します。 |