좌충우돌 개발 블로그

다음과 같이 각 서버에 Fluentd를 설치해 수집하는 서버로 로그를 전송하고

Kafka를 일종의 버퍼로 하여 ELK 스택이나 다른 것들을 이용할 수 있도록 구성했다.

각 서버에서 tail로 로그 파일을 읽어서 forward로 송/수신하고 있었는데

이 과정에서 다음과 같은 에러가 발생하며 누락 혹은 지연되는 경우가 발생했다.

[warn]: #0 failed to flush the buffer. retry_time=0 next_retry_seconds=... chunk="..." error_class=Fluent::Plugin::ForwardOutput::NoNodesAvailable error="no nodes are available"

worker를 늘려 보기도 하고 버퍼 설정을 바꿔보기도 했지만 위의 에러는 계속 나왔다.

원인이 무엇인지 한동안 못찾다가 UDP 방화벽을 열어줘야한다는 글을 봤다.

그래서 문서를 보니 forward 통신을 할 때 heartbeat을 보내는 설정이 있었다.

transport가 기본값이라 tcp로 바꿔보니 해당 에러가 더이상 나지 않았다.

(아마도 기본값인 경우 udp로도 전송하는데 막혀있어 에러가 난 것으로 생각된다.)

하지만 1초(기본값)마다 heartbeat을 보내서 로그 전송 속도가 눈에 띄게 줄었다.

heartbeat을 굳이 필요없어 none으로 하니 속도가 빨라진 것을 확인할 수 있었다.

참고 문헌

1. https://qiita.com/kentokento/items/60b6654d4eea1dad5f42

2. https://docs.fluentd.org/output/forward#heartbeat_type

Fluentd에서 로그를 전송하는 부분을 수정한 이후로 잘 동작하고 있었지만

flush_interval을 5분으로 하여 실시간으로 로그를 볼 수가 없었다.

그리고 Fluentd 서버를 확인해 보니 로그 전송 버퍼가 많이 쌓여 있었다.

그래서 버퍼를 빠르게 비우고 로그도 최대한 실시간으로 볼 수 있도록

Fluentd의 설정 문서를 읽어보면서 여러 가지들을 적용해보았다.

우선은 버퍼 사이즈를 줄여보면서 쌓여있는 버퍼가 빠르게 전송되는지 확인해보았다.

처음에는 buffer_queue_limit를 32로 buffer_chunk_limit를 2m로

그 다음에는 buffer_queue_limit를 64, buffer_chunk_limit를 1m로

테스트해보았지만 두번 다 버퍼는 계속 쌓였다.

그래서 num_threads를 8로 적용해보니 버퍼가 빠르게 없어지는 것을 확인할 수 있었다.

그리고 flush_interval을 1s로 줄여도 버퍼가 많이 쌓이지 않았고 Kibana를 통해 로그를 보니

거의 실시간(30초 정도 지연)으로 처리되는 것을 확인할 수 있었다.

최종적으로 설정한 값은 다음과 같다.

buffer_queue_limit 64
buffer_chunk_limit 1m
flush_interval 1s
num_threads 8

참고 문헌

1. https://docs.fluentd.org/v/0.12/output/kafka#buffer_queue_limit-buffer_chunk_limit

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