Source: Input plugins

📋 Copy All Fluentd Commands 📄 Generate Fluentd PDF Guide

플랫폼	명령어
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## 고급 사용법
./fluent/fluent.conf## 구성

주요 구성 파일 위치

• td-agent (Linux): /fluentd/etc/fluent.conf
• Gem 설치: ```ruby

Source: Input plugins

@type forward port 24224 bind 0.0.0.0

Filter: Process/transform logs

<filter app.**> @type record_transformer hostname ”#{Socket.gethostname}” tag ${tag}

Match: Output plugins

<match app.**> @type elasticsearch host elasticsearch.local port 9200 index_name fluentd type_name fluentd

- **Docker**: ```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>

기본 구성 구조

# 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>

소스 플러그인 (입력)

# 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>

필터 플러그인 (처리)

<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>

매치 플러그인 (출력)

<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>

버퍼 구성

# 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>

다중 워커 구성

# 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

레이블 기반 라우팅

# 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

일반적인 사용 사례

사용 사례 1: Nginx 로그를 Elasticsearch로 수집

# 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

사용 사례 2: Kubernetes 로그 수집

# 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

사용 사례 3: S3로 로그 회전 및 전달

# 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

사용 사례 4: 다중 대상 로그 라우팅

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 및 플러그인을 정기적으로 업데이트하세요. 일관성을 보장하기 위해 프로덕션에서 플러그인 버전을 고정하세요.

문제 해결

문제	솔루션
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).