一、logstash结合kafka收集系统日志和nginx日志
架构图：
环境准备：
A主机：kibana、elasticsearch，有条件可以将两个服务器分开：192.168.7.100
B主机：logstash主机:192.168.7.101/nginx服务器也在此主机上
C主机：logstash主机：192.168.7.102
D主机：kafka/zookeeper：192.168.7.104
E主机：kafka/zookeeper：192.168.7.105
1、使用logstash-B主机收集系统日志和nginx日志写入kafka
1、我们在搭建完kafka及zookeeper的基础上，选择提取zookeeper的leader的主机进行存储
1[root@tomcat-web2 ~]# /usr/local/zookeeper/bin/zkServer.sh status 2ZooKeeper JMX enabled by default 3Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg 4Mode: leader # 查看到192.168.7.105主机是leader模式
2、在logstash主机/etc/logstash/conf.d/目录下创建收集系统日志和nginx日志的文件
1[root@logstash-1 conf.d]# cat log-to-kafka.conf 2input { 3 file { 4 path => "/var/log/messages" 5 type => "kafka-syslog-log-7-101" # 定义系统日志类型 6 start_position => "beginning" 7 stat_interval => "2" 8 } 9 10 file { 11 path => "/var/log/nginx/access.log" 12 type => "kafka-nginx-access-log-7-101" # 定义nginx的日志类型 13 start_position => "beginning" 14 stat_interval => "2" 15 codec => "json" # 如果是json格式的日志，必须写入json格式 16 } 17} 18 19output { 20 if [type] == "kafka-syslog-log-7-101" { 21 kafka { 22 topic_id => "kafka-syslog-log-7-101" 23 bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka的leader主机的IP地址上 24 codec => "json" # 写入的时候使用json编码，因为logstash收集后会转换为json格式 25 } 26 file { 27 path => "/tmp/syslog.log" # 将收集到的访问系统的日志临时存在此文件下，方便排错，解决问题后，可以删除此两项 28 }} 29 30 31 if [type] == "kafka-nginx-access-log-7-101" { 32 kafka { 33 topic_id => "kafka-nginx-access-log-7-101" 34 bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka主机的IP地址 35 codec => "json" # 写入的时候使用json编码，因为Logstash收集后会转换为json格式 36 } 37 file { 38 path => "/tmp/nginx-access.log" # 临时先存放在此文件下，方便排错，如果没有错误，可以删除此项 39 }} 40}
2、验证logstash配置文件的正确性
[root@logstash-1 conf.d]# logstash -f log-to-kafka.conf
3、开始访问nginx网页，（192.168.7.101:80）再查看访问的临时log信息
4、在logstash主机上测试系统日志访问情况
1[root@logstash-1 ~]# echo 1 >> /var/log/messages 2[root@logstash-1 ~]# echo 2 >> /var/log/messages 3[root@logstash-1 ~]# echo 3 >> /var/log/messages
5、查看此时的临时文件信息
此时测试logstash收集到日志写入到kafka中正常，可以将测试的file部分脚本删除掉，然后启动logstash服务。
1file { 2 path => "/tmp/syslog.log" # 删除掉两个测试部分 3 } 4 5 file { 6 path => "/tmp/nginx-access.log" 7 }
启动D主机的logstash服务。
# systemctl start logstash
2、配置logstash-C主机从kafka读取系统日志和nginx日志
1、在logstash-A主机的/etc/logstash/conf.d/目录下创建一个收集kafka配置文件
1input { 2 kafka { 3 topics => "kafka-syslog-log-7-101" 4 bootstrap_servers => "192.168.7.105:9092" # 从kafka的leader主机上提取缓存 5 codec => "json" # 在提取kafka主机的日志时，需要写成json格式 6 } 7 8[root@logstash conf.d]# vim kafka-to-es_logstash.conf 9 kafka { 10 topics => "kafka-nginx-access-log-7-101" 11 bootstrap_servers => "192.168.7.105:9092" # 从kafka的leader主机提取数据 12 codec => "json" # 提取kafka日志时，需要写成json格式 13 } 14 15} 16 17 18 19output { 20 if [type] == "kafka-syslog-log-7-101" { # 转到elasticsearch主机上 21 elasticsearch { 22 hosts => ["192.168.7.100:9200"] 23 index => "kafka-syslog-7-101-%{+YYYY.ww}" # YYYY代表年，ww代表周 24 }} 25 26 if [type] == "kafka-nginx-access-log-7-101" { # 转到elasticsearch主机上 27 elasticsearch { 28 hosts => ["192.168.7.100:9200"] 29 index => "kafka-nginx-access-log-7-101-%{+YYYY.MM.dd}" 30 }} 31}
2、启动logstash-B主机的服务器
# systemctl start logstash
3、在head插件上查看收集到的信息，已经收集到了syslog和nginx的log
3、在kibana网页上创建索引
1、在kibana创建索引
2、在discover查看nginx日志信息
3、在logstash-D主机的系统日志中输入信息，在discover查看系统日志信息
1[root@logstash-1 conf.d]# echo 1 >> /var/log/messages 2[root@logstash-1 conf.d]# echo 2 >> /var/log/messages 3[root@logstash-1 conf.d]# echo 3 >> /var/log/messages
