좌충우돌 개발 블로그

기존에 사용 중이던 ElasticSearch 서버가 근래에 자주 다운됐었다.

그래서 새로운 방식으로 로그를 수집해보려고 했었다.

Hadoop에 JSON으로 저장해서 Zeppelin에서 보려고 하니 불편 사항이 많았다.

다른 것들이 더 있을까 하고 찾아봤지만 ElasticSearch와 Kibana 만큼 로그를 검색하고 조회하는 도구가 없었다.

결국 두 개를 사용하기로 하고 Logstash 만 다른 걸 이용해보려고 하다가 Apache Flume이 생각나서 적용해보려고 했다.

구조는 Source -> Channel -> Sink 로 돼있고 설정이 어려워 보이지 않아 시도해보았다.

하지만 java.lang.NoClassDefFoundError: org/elasticsearch/common/io/BytesStream

에러가 발생하면서 종료됐다.

찾아보니 ElasticSearch jar 파일을 별도로 추가해야한다고 해서 추가했는데도 되지 않았다.

무엇인가 이상하여 찾아보니 Flume에서 제공하는 라이브러리에서

org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder를 임포트하여 사용하는데

해당 패키지는 ElasticSearch 6.x 이상에서는 없어진 패키지였다.

Flume을 굳이 사용하고자 한다면 sink 를 직접 만들어서 사용해야 해서 Logstash로 선회했다.

(물론 fluentd를 사용해도 되지만 다른 곳에서 사용하여 다른 것을 써보고 싶었다)

참고 문헌

1. https://github.com/apache/flume/blob/trunk/flume-ng-sinks/flume-ng-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchLogStashEventSerializer.java
2. https://github.com/elastic/elasticsearch/tree/5.6/core/src/main/java/org/elasticsearch/common/xcontent

기존에 사용하던 Grafana를 신규 버전으로 업그레이드하면서 기존 데이터를 백업할 필요가 있었다.

이전에 한 번 해봤을 때는 일일이 JSON export를 해서 신규 서버에 import 하는 식으로 했었다.

그때는 양이 많지 않아 금방 가능해졌지만 어느새 많이 생겨 일일히 하기에는 문제가 있었다.

그래서 할 수 있는 방법이나 도구가 있는지 찾아보니 Grafana에서 제공하는 API가 있어 이를 활용하기로 했다.

데이터 소스와 폴더는 쉽게 추출해서 옮길 수 있었지만 대시보드의 경우 제공하는 API로는 제대로 동작하지 않았다.

그래서 Grafana에서 import 할 때 사용하는 API를 이용하여 옮길 수 있었다.

작성한 코드는 다음과 같으며 즐겨찾기를 제외한 모든 데이터가 잘 이전된 것을 확인할 수 있었다.

참고 문헌

1. grafana.com/docs/grafana/latest/http_api/

개발용으로 ElasticSearch를 설치할 필요가 있었다.

예전에 서버 몇대에 직접 설치를 했었는데 이번에는 kubernets를 활용하여 설치해보았다.

helm 버전은 3 으로 진행했으며 처음 해보는 거라 다음과 같이 저장소를 추가하고 설치해보았다.

$ helm repo add elastic https://helm.elastic.co
$ helm install elasticsearch elastic/elasticsearch

위와 같이 하면 기본 구성으로 ElasticSearch가 설치되어 올라간 것을 확인할 수 있었다.

하지만 Ingress 연동은 설정으로 추가할 수 없어 해당 저장소를 클론한 다음 elasticsearch/values.yaml 을 수정해야 했다.

# Enabling this will publically expose your Elasticsearch instance.
# Only enable this if you have security enabled on your cluster
ingress:
  enabled: true
  annotations: {}
    # kubernetes.io/ingress.class: nginx
    # kubernetes.io/tls-acme: "true"
  path: /
  hosts:
    - example.com
  tls: []
  #  - secretName: chart-example-tls
  #    hosts:
  #      - chart-example.local

ingress에 enabled가 false로 돼있는데 이것을 true로 변경하고 hosts에 연결할 도메인 주소를 넣으면 된다.

$ helm install elasticsearch --version version ./elasticsearch/

위와 같이 배포한 뒤 Ingress까지 잘 연동되는 것을 확인할 수 있었다.

끝으로 Kibana와 Logstash를 연동할 때 Ingress에 설정한 도메인만 넣으면 기본적으로 9200 포트로 설정된다.

하지만 80으로 접속을 해야하기 때문에 http://example.com:80 으로 포트를 명시하여 설정하면 잘 동작한다. 

참고 문헌

1. https://github.com/elastic/helm-charts

로그를 쌓는데 Logstash에서 간헐적으로 OOM 에러가 발생했었다.

[ERROR][logstash.pipeline ] Exception in pipelineworker, the pipeline stopped processing new events, please check your filter configuration and restart Logstash. {:pipeline_id=>"mds", "exception"=>"Java heap space", "backtrace"=>["org.jruby.util.ByteList.<init>(ByteList.java:92)", "org.jruby.util.ByteList.dup(ByteList.java:328)", "org.jruby.util.ByteList.dup(ByteList.java:300)", "org.jruby.RubyString.doClone(RubyString.java:1346)", "org.logstash.Cloner.deep(Cloner.java:25)", "org.logstash.Cloner.deepMap(Cloner.java:65)", "org.logstash.Cloner.deep(Cloner.java:21)", "org.logstash.Event.clone(Event.java:310)", "org.logstash.ext.JrubyEventExtLibrary$RubyEvent.rubyClone(JrubyEventExtLibrary.java:147)", "org.logstash.ext.JrubyEventExtLibrary$RubyEvent.rubyClone(JrubyEventExtLibrary.java:142)", "java.base/java.lang.invoke.DirectMethodHandle$Holder.invokeSpecial(DirectMethodHandle$Holder)", "java.base/java.lang.invoke.LambdaForm$MH/0x00000008406a3440.invoke(LambdaForm$MH)", "java.base/java.lang.invoke.DelegatingMethodHandle$Holder.delegate(DelegatingMethodHandle$Holder)", "java.base/java.lang.invoke.LambdaForm$MH/0x000000084069e840.guard(LambdaForm$MH)", "java.base/java.lang.invoke.DelegatingMethodHandle$Holder.delegate(DelegatingMethodHandle$Holder)", "java.base/java.lang.invoke.LambdaForm$MH/0x000000084069e840.guard(LambdaForm$MH)", "java.base/java.lang.invoke.Invokers$Holder.linkToCallSite(Invokers$Holder)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_filter_minus_split_minus_3_dot_1_dot_7.lib.logstash.filters.split.RUBY$block$filter$1(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-filter-split-3.1.7/lib/logstash/filters/split.rb:88)", "java.base/java.lang.invoke.DirectMethodHandle$Holder.invokeStatic(DirectMethodHandle$Holder)", "java.base/java.lang.invoke.LambdaForm$MH/0x0000000841005040.invoke(LambdaForm$MH)", "java.base/java.lang.invoke.Invokers$Holder.invokeExact_MT(Invokers$Holder)", "org.jruby.runtime.CompiledIRBlockBody.yieldDirect(CompiledIRBlockBody.java:146)", "org.jruby.runtime.BlockBody.yield(BlockBody.java:114)", "org.jruby.runtime.Block.yield(Block.java:165)", "org.jruby.RubyArray.each(RubyArray.java:1792)", "java.base/java.lang.invoke.LambdaForm$DMH/0x000000084069ac40.invokeVirtual(LambdaForm$DMH)", "java.base/java.lang.invoke.LambdaForm$MH/0x00000008406b3440.invoke(LambdaForm$MH)", "java.base/java.lang.invoke.LambdaForm$MH/0x00000008406c2840.guardWithCatch(LambdaForm$MH)", "java.base/java.lang.invoke.LambdaForm$MH/0x00000008406c3440.invoke(LambdaForm$MH)", "java.base/java.lang.invoke.DelegatingMethodHandle$Holder.delegate(DelegatingMethodHandle$Holder)", "java.base/java.lang.invoke.LambdaForm$MH/0x0000000840674040.guard(LambdaForm$MH)", "java.base/java.lang.invoke.DelegatingMethodHandle$Holder.delegate(DelegatingMethodHandle$Holder)"], :thread=>"#<Thread:0x33a1a707 sleep>"}
[ERROR][org.logstash.Logstash ] java.lang.OutOfMemoryError: Java heap space
[WARN ][logstash.pipeline ] CAUTION: Recommended inflight events max exceeded! Logstash will run with up to 16000 events in memory in your current configuration. If your message sizes are large this may cause instability with the default heap size. Please consider setting a non-standard heap size, changing the batch size (currently 2000), or changing the number of pipeline workers (currently 8) {:pipeline_id=>"mds", :thread=>"#<Thread:0x705f801 run>"}

처음에는 JVM 옵션을 기본 설정인 -Xms1g -Xmx1g 로 하고 있어 이를 늘려서 재시작을 했다.

하지만 곧 다시 에러가 발생하면서 로그를 더이상 받지 못했다.

그래서 Kafka Input 설정 중에 max_poll_records 를 변경하면서 모니터링했다.

몇 시간 동안 잘 되는가 싶더니 곧 에러가 나더니 다시 로그 수집을 못하고 있었다.

filter에서 사용하는 부분을 검사하다 이번에 추가된 split이 문제가 있는 것 같아 구글링을 했다.

구글링을 해보니 해당 플러그인에 메모리 누수 문제가 발생하는 것 같았다.

그래서 이를 대체할 방법을 찾다가 다음과 같이 해보라는 글을 보고 따라해봤다.

def register(params)
    @field = params['field']
    @target = params['target']
end

def filter(event)
    data = event.get(@field)
    event.remove(@field)
    a = []
    data.each { |x|
        e = event.clone
        e.set(@target, x)
        a << e
    }
    a
end

json {
    source => "message"
    remove_field => [ "message" ]
}
ruby {
    path => '/my/path/splitData.rb'
    script_params => { field => "data" target => "data" }
}

split을 변경한 이후로 메모리 누수로 인하여 로그 수집이 중단되는 현상이 나오진 않았다.

참고 문헌

1. https://github.com/logstash-plugins/logstash-filter-split/issues/35

2. https://discuss.elastic.co/t/how-to-split-into-multiple-events-dynamically-for-a-given-json-tried-from-various-question-in-forums/167745/11