1. logstash설치 및 실행

kafka message -> logstash -> ES에 넣어만 보기

input {
  kafka {
    bootstrap_servers => "182.193.17.35:9092"
    topics => "EfssAction"
    group_id => "log-to-es-group"
    consumer_threads => 1
  }
}

filter {

}

output { 
  elasticsearch {
    hosts => ["182.193.17.33:9220"]
    index => "log-insert"
  }
  stdout { codec => rubydebug }
}

kafka-client 버전확인.

kafka-input-plugin과 output-plugin이 integration으로 합쳐졌다고한다.

아래 들어가면 kafka-client버전을 확인 할 수 있음.. 호환되나 확인해봄

D:\ELK\logstash-7.7.0\vendor\bundle\jruby\2.5.0\gems\logstash-integration-kafka-10.0.1-java\vendor\jar-dependencies\org\apache\kafka\kafka-clients

logstash 실행

{logstash설치경로}로 이동 후 실행

bin\logstash.bat -f ./config/logstash.conf

검색

  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "log-insert",
        "_type" : "doc",
        "_id" : "EKtJmHIB5PmJ0m_JzGP1",
        "_score" : 1.0,
        "_source" : {
          "message" : "com.***.***.kafka.model.KafkaMessage...

메세지가 잘 indexing되지만 보내는 쪽에서 특정 java class로 wrapping해서 보내고 있음.

이거를 적절히 filter하고 가공하려면 java filter plugin을 개발해야 함