좌충우돌 개발 블로그

Lostash에서 input으로 kafka를 사용하고 있었는데 다음과 같은 오류가 나왔다.

[WARN ][org.apache.kafka.common.utils.AppInfoParser] Error registering AppInfo mbean
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=logstash-1
        at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436) ~[?:?]
        at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855) ~[?:?]
        at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955) ~[?:?]
        at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890) ~[?:?]
        at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320) ~[?:?]
        at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[?:?]

중복되는 설정들이 많아 합치면서 pipeline을 사용하도록 수정했는데

Logstahs가 설정 파일을 각각의 pipeline마다 하나의 파일로 합쳐 사용하여

pipeline에 설정된 client_id가 서로 중복되어 발생한 문제였다.

이전에는 pipeline을 사용하지 않고 개별로 설정하여 사용했는데 kafka도 개별로 사용하고 있었다.

이때 문제가 되지 않았던 것은 Logstash가 설정 파일을 하나의 파일로 만들어서 사용하고

client_id가 같아도 자동으로 뒤에 숫자를 붙여 다른 client_id로 동작했기 때문이었다.

다음과 같이 client_id를 설정할 때 postfix를 붙여서 pipeline마다 다른 client_id를 갖도록 수정했다.

input {
    kafka {
        bootstrap_servers => "kafka_server:9092"
        client_id => "logstash-postfix"
        ...
    }
}

더이상 해당 오류가 발생하지 않고 kafka로부터 데이터를 잘 받아오는 것을 확인할 수 있었다.

Logstash에서 pipeline으로 개별로 설정을 하여 로그를 수집할 수 있다.

그래서 input, filter, output으로 나누어 사용하기 위해

다음과 같이 디렉토리를 구분하여 pipelines.yml을 설정했다.

- pipeline.id: input
  path.config: "/etc/path/to/input.conf"
- pipeline.id: filter
  path.config: "/etc/path/to/filter.conf"
- pipeline.id: outpue
  path.config: "/etc/path/to/output.conf"

하지만 실제론 input만 동작하고 나머지는 non_running_pipelines로 되어 동작하지 않았다.

다시 문서를 읽어보니 개별 설정마다 input이 있어야 했다.

그래서 디렉토리에서 파일을 나눈 다음 순서대로 합쳐질 수 있도록 파일명을 수정했다.

conf.d/
|- 01-input.conf
|- 02-filter.conf
|- 03-output.conf

그리고 pipelines.yml은 다음과 같이 수정하여 잘 동작하는 것을 확인할 수 있었다.

- pipeline.id: input
  path.config: "/etc/path/to/*.conf"

참고 문헌

1. https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

2. https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html

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