[Log] Fluentd Kafka Logstash 연동

ELK를 이용하여 로그 수집/분석을 하기 위해 Fluentd와 Kafka를 연동할 필요가 있었다.

 

그래서 Fluentd에서 Kafka로 로그를 전송할 수 있도록 설정을 추가했는데

 

Kafka가 죽어서(원인은 파악하지 못했다.) Fluentd가 설치된 로그 서버가 행이 걸리는 경우가 발생했다.

 

retry_limit의 기본값이 17로 돼있어서 발생한 문제로 Kafka로 전송이 실패해서 버퍼가 계속 쌓인 상태에서

 

해당 버퍼를 Kafka로 전송하려가 I/O가 가득 차 해당 서버가 행이 걸려 다른 서버로 절체된 것이었다.

 

retry_limit을 1로 주고 테스트를 해보니 Kafka 전송에 실패해도 버퍼가 더이상 쌓이지 않는 것을 확인했다.

<match>
  @type copy

  <store>
    @type kafka_buffered

    ...
    retry_limit 1
    ...
  </store>
</match>

Logstash에서 Kafka 연동할 때 토픽별로 인덱스를 설정해 Elastic Search에 저장하기 위해

 

filter와 output에 다음의 내용을 추가하면 된다.

if [@metadata][kafka][topic] == "topic" {
}

위의 설정을 추가하지 않는 경우 filter에서는 오류가 나지 않을 수도 있지만

 

output에서는 가장 먼저 설정된 인덱스만 인식하여 나머지 인덱스에는 저장이 되지 않는다.

 

참고 문헌

1. https://docs.fluentd.org/v/0.12/output/kafka#retry_limit-disable_retry_limit