如該文所述，集中式日志分析平臺 - ELK Stack - 部署篇，引入消息隊列機制的 Filebeat + Logstash 分布式架構(gòu)較為靠譜，有極低的客戶端采集開銷，引入消息隊列，均衡了網(wǎng)絡(luò)傳輸，從而降低了網(wǎng)絡(luò)堵塞，尤其是丟失數(shù)據(jù)的可能性，并且可以結(jié)合 grok 做到非常靈活的日志數(shù)據(jù)格式化。部署篇中也談了 ELK + Filebeat 的部署，本文講講如何在該架構(gòu)下整合 Kafka。

多集群情況下的簡明架構(gòu)如下：

node_cluster1 -> filebeat_cluster1 -> topic1 ->                   -> index1 
                                                logstash_cluster            -> Kibana
node2_cluster -> filebeat_cluster2 -> topic2 ->                   -> index2 

Kafka 部署

測試環(huán)境我們部署一個單節(jié)點的 Kafka 集群，Broker 假設(shè)部署在 172.16.134.3，測試環(huán)境我們使用 Kafka 自己管理的 Zookeeper。

Step1. 安裝 OpenJDK Runtim

sudo yum install java-1.8.0-openjdk.x86_64

驗證安裝是否成功：

java -version

修改環(huán)境變量 ~/.bashrc 以加載 JAVA_HOME 和 JRE_HOME

export JAVA_HOME=/usr/lib/jvm/jre-1.8.0-openjdk
export JRE_HOME=/usr/lib/jvm/jre

通過 source 加載：

source ~/.bashrc

Step2. 下載 Kafka

wget http://www-us.apache.org/dist/kafka/0.10.1.1/kafka_2.11-0.10.1.1.tgz
tar -xvf kafka_2.11-0.10.1.1.tgz -C /home/admin/soft

Step3. 修改配置

因為是虛擬機環(huán)境，單個 broker ，我們只需要修改運行腳本 bin/kafka-server-start.sh，適當(dāng)縮小 heap：

KAFKA_HEAP_OPTS="-Xmx256M -Xms256M"

生產(chǎn)環(huán)境，需要修改的建議配置如下：

broker.id # 節(jié)點編號
log.dirs # 數(shù)據(jù)目錄
zookeeper.connect # ZK 連接字符串

如果需要修改 kafka 的運行日志目錄，可以修改 bin/kafka-run-class.sh：

# 在 base_dir=$(dirname $0)/.. 這行之后添加
LOG_DIR=/path/to/logs

Step4. 啟動服務(wù)

啟動 ZooKeeper：

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

啟動 Kafka：

bin/kafka-server-start.sh config/server.properties 2>&1 &

啟動成功的腳本輸出應(yīng)該如下：

[2017-02-22 16:23:27,175] INFO [Kafka Server 0], started (kafka.server.KafkaServer)

Step5.創(chuàng)建測試用的 Topic

因為 Filebeat 和 Logstash 都只有1個節(jié)點，所以我們就建立一個 partition 的 topic。只有1臺 Broker，topic 的 replication-factor 為 1：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic filebeat_test
Created topic "filebeat_test".

使用以下命令查看創(chuàng)建的 Topic：

bin/kafka-topics.sh --list --zookeeper localhost:2181

正常情況下可以看到輸出：

filebeat_test

Filebeat as Kafka Producer

在這個架構(gòu)下 Filebeat 是 Kafka 的生產(chǎn)者。修改 172.16.134.3 和 172.16.134.8 的 Filebeat 配置文件 filebeat.yml：

修改前：

output.logstash:
  hosts: ["172.16.134.2:5044"]

修改后：

output.kafka:
  # initial brokers for reading cluster metadata
  hosts: ["172.16.134.3:9092"]

  # message topic selection + partitioning
  topic: filebeat_test
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

max_message_bytes 配置項表示超過1000000B的 Event 會被直接丟棄。重啟 Filbeat：

pid=`ps aux | grep filebeat | grep -v grep | awk '{print$2}'`
sudo kill -9 $pid
./filebeat start 2>&1 &

看到如下日志表示連接 Kafka 成功：

2017-02-22T16:43:53+08:00 WARN Connected to broker at v134003.yn1.domain.com:9092 (registered as #0)

Logstash as Kafka Consumer

Logstash Kafka Consumer 和 Kafka 的兼容性列表在這里，支持 SSL 和 Kerberos。修改配置logstash.conf：

修改前：

input {
  beats {
    port => 5044
  }
}

修改后：

input {
  kafka {
    bootstrap_servers => "172.16.134.3:9092"
    consumer_threads => 1
    topics => ["filebeat_test"]
  }
}   

重啟：

pid=`ps aux | grep logstash | grep -v grep | awk '{print$2}'`
sudo kill -9 $pid
./bin/logstash -f config/logstash.conf 2>&1 &

問題排查

在配置過程完成后，遇到了一些問題，記錄下。

首先是重啟后發(fā)現(xiàn) Logstash 無法連接 ES：

[2017-02-22T17:55:01,157][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://logstash_system:xxxxxx@localhost:9200/, :path=>"/"}
[2017-02-22T17:55:01,160][WARN ][logstash.outputs.elasticsearch] Attempted to resurrect connection to dead ES instance, but got an error. {:url=>#<URI::HTTP:0x100f9cb3 URL:http://logstash_system:xxxxxx@localhost:9200/_xpack/monitoring/?system_id=logstash&system_api_version=2&interval=1s>, :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://logstash_system:xxxxxx@localhost:9200/][Manticore::SocketException] Connection refused"}

嘗試確認(rèn) ES 是否可用：

curl -u elastic:changeme http://172.16.134.2:9200

發(fā)現(xiàn) OK，確認(rèn)是沒有通過 Logstash 的 X-Pack 驗證 Monitoring Logstash，所以需要修改配置 logstash.yml：

xpack.monitoring.elasticsearch.url: "http://172.16.134.2:9200" 
xpack.monitoring.elasticsearch.username: "logstash_system" 
xpack.monitoring.elasticsearch.password: "changeme"

重啟 Logstash 后問題解決。

第二個問題是數(shù)據(jù)無法正常在 ES 中讀取。自上而下的鏈路排查。首先看 Filebeat 日志是否正常，確認(rèn)日志也正常：

2017-02-22T18:45:42+08:00 INFO Non-zero metrics in the last 30s: publish.events=1 libbeat.kafka.call_count.PublishEvents=1 libbeat.publisher.published_events=1 libbeat.kafka.published_and_acked_events=1 registrar.writes=1 registrar.s
tates.update=1
2017-02-22T18:46:12+08:00 INFO Non-zero metrics in the last 30s: libbeat.kafka.published_and_acked_events=1 libbeat.kafka.call_count.PublishEvents=1 publish.events=1 registrar.states.update=1 libbeat.publisher.published_events=1 regi
strar.writes=1
2017-02-22T18:46:42+08:00 INFO Non-zero metrics in the last 30s: libbeat.kafka.published_and_acked_events=1 libbeat.kafka.call_count.PublishEvents=1 registrar.writes=1 libbeat.publisher.published_events=1 publish.events=1 registrar.s
tates.update=1
2017-02-22T18:47:12+08:00 INFO Non-zero metrics in the last 30s: registrar.states.update=1 libbeat.kafka.published_and_acked_events=1 libbeat.publisher.published_events=1 registrar.writes=1 libbeat.kafka.call_count.PublishEvents=1 pu
blish.events=1

再看 Kafka 內(nèi)是否有數(shù)據(jù)流入，確認(rèn)是有正常數(shù)據(jù)寫入的：

./kafka-console-consumer.sh --bootstrap-server 172.16.134.3:9092 --topic filebeat_test --from-beginning

再看 Logstash 的 Events Received 是否增加，以及 ES 的 Documents 是否增加，也是增加的：

問題排查 圖1

然后進(jìn)入 Indices 中查看，發(fā)現(xiàn) %{[@metadata\][beat]}-2017.02.22 這個 Index 的 Document 一直在漲，而不是我們定義的 Filebeat-*，╮(╯Д╰)╭ ，如下圖所示：

問題排查 圖2

修改 Logstash 配置文件 logstash.conf ：

output {  
  elasticsearch {
    hosts => ["172.16.134.2:9200"]
    user => elastic 
    password => changeme
    manage_template => false
    index => "filebeat-%{+YYYY.MM.dd}"
    document_type => "filebeat"
  }
}

重啟后問題解決。

小結(jié)

本文介紹了如何把 Kafka 整合至 ELK Stack + Filbeat，并且對其中遇到的問題進(jìn)行了結(jié)合排查過程式的闡述。