[로그 수집] logstash 장애시 이슈 없을까?

'filebeat > logstash > es' 로 현재 구성되어 있어 서비스 로그를 계속적으로 es 에 적재하고 있다.

그러나 logstash 설정 변경이 필요한 상황이라 logstash 를 재기동 해야 하는 상황으로 이때 이슈가 없을까? 에 대한 테스트를 진행해본다.

테스트는 아래와 같이 진행해본다.

1) 수동 restart

1. logstash 다운

$ kill -9 [logstash PID]

2. filebeat 에서 읽는 로그 파일에 데이터 추가

$ cat service.log
{'name':'ParkHyunJun'}
{'name':'LeeHoSeong'}
{'name':'thewayhj'}
{'name':'LeeNow'}
{'name':'hongYooLee'}
{'name':'test'}
$ echo "{'name':'failtest'}" >> service.log
$ cat service.log
{'name':'ParkHyunJun'}
{'name':'LeeHoSeong'}
{'name':'thewayhj'}
{'name':'LeeNow'}
{'name':'hongYooLee'}
{'name':'test'}
{'name':'failtest'}

3. logstash 기동 

$ start_es

4. 추가한 로그 정상적으로 읽어 오는지 

// kafka (s3의 데이터로 확인)

// es (kibana 로 확인) 

filebeat 는 at-least-once(최소 한번 전송) 을 보장하기에 중복은 발생할 수 있으나 데이터 누락은 없어 logstash 가 종료 될시 재 전송을 진행한다.

logstash 의 경우 아래와 같이 처리하며 큐에 있는 데이터를 기반으로 처리하기에

 'input → "memory/Persistent Queue" → filter → output'

queue.type 가 memory 일시 비정상적인 종료될때 queue.max_bytes 만큼(default : 1GB) 데이터 누락이 발생한다.

그러나 persisted(영구 저장)으로 되어 있을시 비정상적인 종료에 의해 종료될때 at-least-once(최소 한번 전송) 을 보장하여 누락 방지 한다.

추가적으로 queue.max_bytes 만큼 큐가 찼을시 더이상 새 연결을 허용 하지 않으며 file beat 는 연결이 허용될때까지 연결을 시도한뒤 파일을 전송하기에 queue size 에 대한 데이터 누락은 없다.

- 관련 문서 : https://www.elastic.co/guide/en/beats/filebeat/current/how-filebeat-works.html

- 큐 사이즈 도달시 처리 관련 문서 : https://www.elastic.co/guide/en/logstash/current/persistent-queues.html#backpressure-persistent-queue

2) auto reload 모드

// config (기존)

input {
	beats {
		port => 5044
		host => "0.0.0.0"
	}
}
output {
	stdout{
		codec => dots
	}
	elasticsearch{
		hosts => "genie-hong-06"
		index => "servicelog-%{+yyyy.MM.dd}"
	}
	#kafka {
    #    	bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
    #    	codec => json
    #    	topic_id => "logstash-test"
    #  }
}

// 변경

input {
	beats {
		port => 5044
		host => "0.0.0.0"
	}
}
output {
	stdout{
		codec => dots
	}
	elasticsearch{
		hosts => "genie-hong-06"
		index => "servicelog-%{+yyyy.MM.dd}"
	}
	kafka {
        	bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
        	codec => json
        	topic_id => "logstash-test"
      }
}

- config 수정하자마자 reload 되었음

[2022-02-14T13:46:55,741][INFO ][logstash.pipelineaction.reload] Reloading pipeline {"pipeline.id"=>:main}

- 이전 데이터는(수정전 데이터) 파일비트로 부터 넘어오지 않는다.

- 수정 이후 추가된 데이터는 반영된다.

$ cat service.log
{'name':'ParkHyunJun'}
{'name':'LeeHoSeong'}
{'name':'thewayhj'}
{'name':'LeeNow'}
{'name':'hongYooLee'}
{'name':'test'}
{'name':'failtest'}
{'name':'adddata'}
$ echo "{'name':'foofoo'}" >> service.log

// es (kibana)

- 8개 > 9개로 변경 되었다.

// kafka (s3)

만약,

syntax error 발생시키면 어떻게 될까?

// 설정 수정 (' }' 를 제거 하여 오류 발생 시킴)

input {
	beats {
		port => 5044
		host => "0.0.0.0"
	}
}
output {
	stdout{
		codec => dots
	}
	elasticsearch{
		hosts => "genie-hong-06"
		index => "servicelog-%{+yyyy.MM.dd}"
	}
	kafka {
        	bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
        	codec => json
        	topic_id => "logstash-test"
      
}

// 로그 

3초마다 오류 발생

.[2022-02-14T14:00:20,587][ERROR][logstash.agent           ] Failed to execute action {

// 데이터 추가

echo "{'name':'rooroo'}" >> service.log
$ cat service.log
{'name':'ParkHyunJun'}
{'name':'LeeHoSeong'}
{'name':'thewayhj'}
{'name':'LeeNow'}
{'name':'hongYooLee'}
{'name':'test'}
{'name':'failtest'}
{'name':'adddata'}
{'name':'foofoo'}
{'name':'rooroo'}

// kafka (s3)

// es (kibana)

9개 > 10개

// 설정

kafka-10, 11, 12는 없는 호스트명 이다.

input {
	beats {
		port => 5044
		host => "0.0.0.0"
	}
}
output {
	stdout{
		codec => dots
	}
	elasticsearch{
		hosts => "genie-hong-06"
		index => "servicelog-%{+yyyy.MM.dd}"
	}
	kafka {
        	bootstrap_servers => "kafka-10:9092,kafka-11:9092,kafka-12:9092"
        	codec => json
        	topic_id => "logstash-test"
     }
}

// 로그 

reload 가 실패 했다는 에러 로그 이다.

[2022-02-14T14:17:55,220][INFO ][logstash.pipelineaction.reload] Reloading pipeline {"pipeline.id"=>:main}
[2022-02-14T14:18:00,243][WARN ][org.logstash.execution.ShutdownWatcherExt] {"inflight_count"=>0, "stalling_threads_info"=>{"other"=>[{"thread_id"=>320, "name"=>"[main]<beats", "current_call"=>"[...]/vendor/bundle/jruby/2.5.0/gems/logstash-input-beats-6.0.4-java/lib/logstash/inputs/beats.rb:204:in `run'"}, {"thread_id"=>318, "name"=>"[main]>worker0", "current_call"=>"[...]/logstash-core/lib/logstash/java_pipeline.rb:262:in `block in start_workers'"}, {"thread_id"=>319, "name"=>"[main]>worker1", "current_call"=>"[...]/logstash-core/lib/logstash/java_pipeline.rb:262:in `block in start_workers'"}]}}
[2022-02-14T14:18:00,258][ERROR][org.logstash.execution.ShutdownWatcherExt] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information.
[2022-02-14T14:18:02,929][WARN ][org.apache.kafka.clients.ClientUtils][main] Couldn't resolve server kafka-10:9092 from bootstrap.servers as DNS resolution failed for kafka-10
[2022-02-14T14:18:02,963][WARN ][org.apache.kafka.clients.ClientUtils][main] Couldn't resolve server kafka-11:9092 from bootstrap.servers as DNS resolution failed for kafka-11
[2022-02-14T14:18:02,985][WARN ][org.apache.kafka.clients.ClientUtils][main] Couldn't resolve server kafka-12:9092 from bootstrap.servers as DNS resolution failed for kafka-12
[2022-02-14T14:18:03,005][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Reload<main>, action_result: false", :backtrace=>nil}

// 이상태에서 데이터를 추가한뒤 설정을 정상적으로 만들시 오류 일시 동안에 추가된 데이터가 들어온다.

만약 kafka 가 주석된 상황에서 데이터(a)를 넣고 잘못된 kafka 호스트를 넣고 데이터(b)를 넣고 정상 호스트명을 넣으면 ?

// 데이터(a)를 넣고

// 잘못된 host

[2022-02-14T14:27:08,179][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Reload<main>, action_result: false", :backtrace=>nil}

//데이터(b)를 넣고 

정상적인 설정도 데이터가 반영 되지 않음

// 정상 호스트명을 넣으면

kafka, es 모두 b 반영

reload 옵션 사용시 설정을 잘못 하였다면(syntax error, 잘못된 설정 옵션) 이전 설정대로 적용 된다.

만약 kafka 설정 자체가 잘못 되었다면 전체적으로 동작하지 않고, 설정이 정상적으로 변경 됐을시 전체적으로 동작 안한 시점 부터 반영된다.

# 기존에 사용하던 설정

1) filebeat
#=========================== Filebeat inputs =============================
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /logs/app_logs/service.json
    - /logs/app_logs/test.json

  fields:
    type: "standard_log"
    copy_to: "kafka"
    index_name: "servicename"

#============================= Filebeat modules ===============================
filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: true

  # Period on which files under path should be checked for changes
  #reload.period: 10s

#================================ Outputs =====================================
#----------------------------- Logstash output --------------------------------
output.logstash:
  hosts: [ "x.x.x.x:5044", "x.x.x.x:5044", "x.x.x.x:5044", "x.x.x.x:5044", "x.x.x.x:5044", "x.x.x.x:5044", "x.x.x.x:5044"]

2) logstash
- Logstash는 클러스터링되지 않고 독립 실행형
- 기본 설정을 사용하여 inmemory queue 사용 
- 기본 설정을 사용하여 queue size는 최대 1GB 저장
- restart 없이 반영가능한 reload 시용 (--config.reload.automatic=true)
output {

    elasticsearch {
      hosts => ["x.x.x.x:9200", "x.x.x.x:9200", "x.x.x.x:9200"]
      index => "%{[@metadata][index_name]}-test.log.%{[@metadata][date]}"
    }

 webhdfs {
      host => "hadoop-01"
      port => 50070
	single_file_per_thread => true
      path => "/logs/%{[@metadata][_date]}/%{[@metadata][index_name]}.%{[@metadata][host_name]}.%{[@metadata][thread_id]}"
      user => "hadoop"
      codec => "json"
    }
}

최종 결론,

현재 구조상 filebeat 는 공식 문서로 보았을시 데이터 누락은 없어 (중복은 있을수 있다.) logstash 가 죽은 이후 데이터에 대한 부분은 누락이 없으나 logstash가 죽기 전에 내부 인메모리 큐에 저장된 데이터(현재 설정은 1GB)에 대해서는 누락이 발생한다.

logstash가 down 됐을시에 대한 데이터 누락 이슈 해결을 위해서는 logstash 내부 큐 타입을 메모리 > 영구 저장 타입으로 변경하거나 다른 대응 방안을 검토 해봐야 할 것 같다.

logstash 설정 변경에 대해서는 reload 옵션에 의해 자동 반영되고 만약 설정이 잘못 되더라도 이전 설정대로 유지 또는 설정 정상화 후 이전 데이터 반영 하니 설정 변경에 대해서는 이전 데이터 파이프 라인에 대한 영향은 없을것으로 보인다.