1. ELK + Kafka + Filebeat架构

ELK构架

• Filebeat：

日志采集。

• Logstash：

官网描述：Logstash：Collect，Enrich and Transport。收集，丰富和传输，Logstash是用于管理事件和日志的工具。可以使用它来收集日志，解析它们，然后存储它们以备以后使用（例如用来搜索）。

• Shipper：

在每个目标系统上运行的客户端，负责收集日志消息。

• Kafka：

作为消息队列解耦了处理过程，同时提高了可扩展性。具有峰值处理能力，使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

• Kerberos：

基于共享秘钥对称加密的安全网络认证系统，它避免了将密码在网络上传输，而是将密码作为对称加密的秘钥，通过是否能够解密来验证用户的身份。

• Indexer：

负责按管理员指定的方式，汇集日志并转发日志给elasticsearch。

• Elasticsearch：

官网描述：Elasticsearch：Store，Search and Analyze。存储，搜索和分析，用于存储所有的日志。

• Search Guard：

Search Guard是Elasticsearch的安全插件。它为后端系统（如Kerberos）提供身份验证和授权，并向Elasticsearch添加审核日志记录和文档/字段级安全性。

• Kibana：

官网描述：Kibana：Explore，Visualize and Share。浏览，可视化和分享，为Logstash和ElasticSearch提供日志分析友好的Web界面，帮助汇总、分析和搜索重要数据。

2 为什么要使用ELK

• 开发可以通过登录到Kibana拿到日志，而不再需要通过运维，减轻运维的工作，同时运维也方便进行管理；

• 可以解决单个日志文件巨大，常用的文本工具难以分析，检索困难；

• 日志种类多，分散在不同的位置难以查找，通过ELK可以进行统一管理；

3 验证流程

Filebeat->Logstash->Kafka->Logstash->Elasticsearch->Kibana

软件版本说明：

• Filebeat版本：5.2.1
• Logstash版本：5.2.2
• Kafka版本：0.10.1
• Elasticsearch版本：5.2.2
• Kibana版本：5.2.2
• jdk版本：1.8.0_112

1. filebeat是通过yum进行安装的，因此修改filebeat配置文件/etc/filebeat目录下filebeat.yml文件对应的output.logstash中logstash的host位置，以及配置监控的日志文件的路径，这里监控的是在/root/remoa目录下的dengqy.log日志文件。

elk1
elk2

2. 查看kafka中broker服务器的hosts位置：

   
   elk3

3. 查看Elasticsearch中服务器的host位置：在/opt/package/elasticsearch-5.2.2/config目录下查看elasticsearch.yml文件中的host的ip地址及运行HTTP服务的端口，端口注释即使用默认端口9200。

elk4

4. 查看用于Elasticsearch身份验证的keys文件位置：

elk5

5. 配置remoatest2.conf文件作为filebeat input logstash以及logstash output kafka。

input {
        beats{
                port = 5044
        }
}
output {
    stdout{codec = rubydebug}
    kafka{
                topic_id ="dengqytopic"
                bootstrap_servers = "hdp1.example.com:9092"
                security_protocol ="SASL_PLAINTEXT"
                sasl_kerberos_service_name = "kafka"
                jaas_path = "/tmp/kafka_jaas.conf.demouser"
                kerberos_config = "/etc/krb5.conf"
                compression_type = "none"
                acks = "1"
        }
}

（6）配置remoatest3.conf文件作为kafka input logstash以及logstash output elasticsearch。

input{
        kafka{
                bootstrap_servers ="hdp1.example.com:9092"
                security_protocol = "SASL_PLAINTEXT"
                sasl_kerberos_service_name = "kafka"
                jaas_path = "/tmp/kafka_jaas.conf.demouser"
                kerberos_config ="/etc/krb5.conf"
                topics =["dengqytopic"]
        }
}
output{
        stdout{
                codec = rubydebug
        }
        elasticsearch{
                hosts =
 ["kdc1.example.com:9200","kdc2.example.com:9200"]
                user =logstash
                password = logstash
                action = "index"
                index = "logstash-dengqy-%{+YYYY.MM.dd}"
                truststore =  "/opt/package/logstash-5.2.2/config/keys/truststore.jks"
                truststore_password = whoami
                ssl = true
                ssl_certificate_verification =true
                codec ="json"
        }
}

7. 启动消费者监控：

KAFKA_HEAP_OPTS="-Djava.security.auth.login.config=/usr/hdp/current/kafka-broker/config/kafka_client_jaas.conf -Xmx512M"  /opt/package/kafka_2.10-0.10.1.0/bin/kafka-console-consumer.sh --topic dengqytopic --bootstrap-server hdp1.example.com:9092  --from-beginning --consumer.config /opt/package/kafka_2.10-0.10.1.0/config/consumer.properties

elk6

8. 启动remoatest2.conf脚本：

bash ../bin/logstash -f remoatest2.conf

elk7

9. 启动remoatest3.conf脚本：

bash bin/logstash -f config/remoatest3.conf

elk8

10. 启动filebeat收集：

service filebeat start

elk9

11. 查看到remoatest3.conf中配置的stdout{codec => rubydebug}标准输出：

    
    elk10

12. 查看到remoatest2.conf中配置的stdout{codec => rubydebug}标准输出：

elk12

13. 查看到消费者中处理日志：

    
    elk13

14. 在Kibana中查看到对应的index：

GET _cat/indices

elk14

15. 查看该index详细内容，数据总量及数据内容一致，即存入elasticsearch中成功：

GET logstash-dengqy-2017.09.08/_search

elk15

3. 转载

blog.csdn.net/remoa_dengqinyi