如该文所述，集中式日志分析平台 - 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，并且对其中遇到的问题进行了结合排查过程式的阐述。
网友评论