如该文所述，集中式日志分析平台 - ELK Stack - 部署篇，引入消息队列机制的 Filebeat + Logstash 分布式架构较为靠谱，有极低的客户端采集开销，引入消息队列，均衡了网络传输，从而降低了网络堵塞，尤其是丢失数据的可能性，并且可以结合 grok 做到非常灵活的日志数据格式化。部署篇中也谈了 ELK + Filebeat 的部署，本文讲讲如何在该架构下整合 Kafka。

多集群情况下的简明架构如下：

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

Kafka 部署

测试环境我们部署一个单节点的 Kafka 集群，Broker 假设部署在 172.16.134.3，测试环境我们使用 Kafka 自己管理的 Zookeeper。

Step1. 安装 OpenJDK Runtim

sudo yum install java-1.8.0-openjdk.x86_64

验证安装是否成功：

java -version

修改环境变量 ~/.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. 修改配置

因为是虚拟机环境，单个 broker ，我们只需要修改运行脚本 bin/kafka-server-start.sh，适当缩小 heap：

KAFKA_HEAP_OPTS="-Xmx256M -Xms256M"

生产环境，需要修改的建议配置如下：

broker.id # 节点编号
log.dirs # 数据目录
zookeeper.connect # ZK 连接字符串

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

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

Step4. 启动服务

启动 ZooKeeper：

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

启动 Kafka：

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

启动成功的脚本输出应该如下：

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

Step5.创建测试用的 Topic

因为 Filebeat 和 Logstash 都只有1个节点，所以我们就建立一个 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".

使用以下命令查看创建的 Topic：

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

正常情况下可以看到输出：

filebeat_test

Filebeat as Kafka Producer

在这个架构下 Filebeat 是 Kafka 的生产者。修改 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 &

问题排查

在配置过程完成后，遇到了一些问题，记录下。

首先是重启后发现 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"}

尝试确认 ES 是否可用：

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

发现 OK，确认是没有通过 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 后问题解决。

第二个问题是数据无法正常在 ES 中读取。自上而下的链路排查。首先看 Filebeat 日志是否正常，确认日志也正常：

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 内是否有数据流入，确认是有正常数据写入的：

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

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

问题排查 图1

然后进入 Indices 中查看，发现 %{[@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"
  }
}

重启后问题解决。

小结

本文介绍了如何把 Kafka 整合至 ELK Stack + Filbeat，并且对其中遇到的问题进行了结合排查过程式的阐述。