EFK (ESB연계KAFKA)

 EFK란 Elasticsearch + Fluentd + Kibana 입니다.  쿠버네티스환경에서 수많은 컨테이너의 로그를 수집하여 통합적으로 로그를 UI로 확인할 수 있는 인터페이스입니다.

테스트 서버에 아래 같은 데모를 구축하여 테스트 완료 하였습니다.

먼저 MariaDB에 적재 되어 있는 Table의 Data를 Fluentd가 아래 소스와 같이 SQL을 사용하여 json형식으로 추출하고 Kafka에 PUB해준 후  Topic에 적재된 Data를 다시 SUB하여 ElasticSearch에 넣어 줍니다.

이후 kibana에서 해당 Index 패턴을 정의 해준 후 Dashboard형식으로 표출 합니다.

 

Mysql To Kafka 소스

<source>
  @type mysql_replicator
  host IP
  username root
  password ishark2020
  database ishark
  query SELECT id,id2,id3 FROM fluenttestsss
  primary_key id
  interval 10s
  enable_delete no
  tag replicator.livechat.chat_chennaibox.${event}.${primary_key}
</source>
<match replicator.**>
  @type kafka2
 
  brokers IP:9092
  default_topic test_heejae3
 
  <format>
    @type json
  </format>
 
  <buffer fluentd-test>
    @type memory
    flush_interval 3s
    overflow_action drop_oldest_chunk
  </buffer>
</match>

구동 화면

Kafdrop을 통한 Data 적재 확인

Kafka to Elastic 소스

<source>
  @type kafka
  
  brokers IP:9092
  topics test_heejae3
  format json
</source>
<match topic4>
  @type copy
  <store>
    @type stdout
  </store>
  <store>
    @type elasticsearch
    host IP
    port 9200
    logstash_format true                       
    logstash_prefix rocktest             
    logstash_dateformat %Y%m%d               
    include_tag_key true                     
    tag_key @logname                          
  
    buffer_type file                           
    buffer_path /var/log/td-agent/buffer/syslog/
    flush_interval 1s                          
    buffer_chunk_limit 256m                    
    buffer_queue_limit 256                     
    retry_wait 5                               
    buffer_queue_full_action drop_oldest_chunk
     <format>
      @type json
    </format> 
  </store>
</match>

Kibana 에서 Index 조회