Elasticsearch와 Kafka

Kafka

Apache 프로젝트이며 open source stream 처리 플랫폼이기 때문에 Logstash 혹은 Beats와 동일한 목적을 수행하며 AWS Kinesis와도 여러 공통점이 있다. 높은 처리량과 짧은 지연시간을 갖고 있는 Pub/Sub 매커니즘이며, 수많은 서버의 로그와 같이 여러 입력 데이터 소스를 입력으로 받을 수 있고, 안정적으로 Target에 Publish 할 수 있다. 또한 처리중인 stream을 저장할 수 있는 기능을 가진 프레임 워크이다.
따라서 Logstash와 많은 공통점이 있지만 이미 데이터가 Kafka를 통해 유입되고 있는 환경에서 데이터를 Elasticsearch와 같은 다른 Target으로 가져와야 하는 경우 혹은 Logstash를 활용하여 Kafka에서 출력되는 데이터를 자동 구문 분석을 해야하는 경우가 있을 수 있다.

Logstash + Kafka

Logstash와 Kafka의 연결은 매우 간단하다. 이전에 했던 것 처럼 config 파일의 input 부분에 Kafka 입력을 설정해주기만 하면 된다.

input {
    kafka {
        bootstrap_servers => "localhost:9092"
        topics => ["kafka-logs"]
    }
}

이렇게 모든 부트스트랩 서버의 목록과 모니터할 topics 목록을 제공해주면 Kafka에서 데이터를 Publish할 때 topic에 게시된다.

Kafka 설치

Kafka는 종속성이 필요하다.

• Zookeeper
  • 서비스가 사용하는 구성을 관리하고 보관한다.
  • sudo apt-get install zookeeperd
• Kafka
  • kafka download에서 최신버전의 Binary download를 받는다. 링크를 우클릭으로 주소 복사한 후 리눅스 터미널에 wget 복사한 주소 를 입력해주면 된다.
  • tar -xvf kafka_2.13-3.3.1.tgz
    • 압축 해제 후 생성된 폴더로 진입한다.
  • sudo bin/kafka-server-start.sh config/server.properties
    • 서버 실행

Kafka 실행

데이터를 발행할 토픽을 만들고 Logstash에 저장해보자.

Topic 생성

• Kafka 서버를 실행시켜놓은 상태로 새로운 세션을 열어 Kakfa 디렉토리의 topic 스크립트를 실행하여 로그 데이터를 수신하는데 사용할 새로운 topic을 생성할 수 있다.
  • sudo bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafka-logs

Logstash 실행

• 이제 해당 stream을 실제로 수신 대기 할 수 있도록 Logstash를 설정해야한다.
  • sudo vim /etc/logstash/conf.d/logstash.conf
• 다음과 같이 설정하고 실행해주자.

input {
        kafka {
                bootstrap_servers => "localhost:9092"
                topics => ["kafka-logs"]
        }
}

filter {
        grok {
                match => {"message" => "%{COMBINEDAPACHELOG}"}
        }
        date {
                match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
        }
}

output {
        elasticsearch {
                hosts => ["localhost:9200"]
                index => "kafka-logs"
        }
        stdout {
                codec => rubydebug
        }
}

• Logstash는 Kakfa에 대해 listen 상태가 된다. 이제 Kafka의 'kafka-logs' topic에 데이터를 publish하면 Logstash가 이를 다시 Elasticsearch 서버에 전달한다. 따라서 데이터를 publish 하기 위해 새로운 세션을 열어주자.

Data Publish

Kakfa 콘솔 스크립트를 사용하여 쉽게 데이터를 넣을 수 있다.

• sudo bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-logs < ../access_log
  • 홈 디렉토리에 있는 access_log 파일을 발행하는 스크립트를 실행시키는 명령어이다.
  • Logstash의 콘솔 로그를 통해 정상적으로 실행되고 있는지 확인할 수 있다.
    • data publishing -> topic -> logstash -> elasticsearch 로 데이터가 이동하고 있는 것이다.
• 모든 작업이 완료되면 elasticsearch에서 정상적으로 index가 생성되었는지 확인해주면 된다.
  • curl -XGET localhost:9200/kakfa-logs/_search?pretty
• result 일부
  • 사용한 로그는 교육을 위해 제공되는 로그를 사용하였다.

    {
      "_index" : "kafka-logs",
      "_type" : "_doc",
      "_id" : "GO3Bp4QBhGqw408HXtDh",
      "_score" : 1.0,
      "_source" : {
        "httpversion" : "1.1",
        "request" : "/wp-cron.php?doing_wp_cron=1493537093.4810650348663330078125",
        "timestamp" : "30/Apr/2017:07:24:53 +0000",
        "referrer" : "\"http://sundog-soft.com/wp-cron.php?doing_wp_cron=1493537093.4810650348663330078125\"",
        "@version" : "1",
        "@timestamp" : "2017-04-30T07:24:53.000Z",
        "message" : "54.210.20.202 - - [30/Apr/2017:07:24:53 +0000] \"POST /wp-cron.php?doing_wp_cron=1493537093.4810650348663330078125 HTTP/1.1\" 200 - \"http://sundog-soft.com/wp-cron.php?doing_wp_cron=1493537093.4810650348663330078125\" \"WordPress/4.7.4; http://sundog-soft.com\"",
        "verb" : "POST",
        "ident" : "-",
        "agent" : "\"WordPress/4.7.4; http://sundog-soft.com\"",
        "auth" : "-",
        "response" : "200",
        "clientip" : "54.210.20.202"
      }
    }