标签：sys tar root 定义 提取数据 oca put 信息 ble
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、我们在搭建完kafka及zookeeper的基础上，选择提取zookeeper的leader的主机进行存储
[root@tomcat-web2 ~]# /usr/local/zookeeper/bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg Mode: leader # 查看到192.168.7.105主机是leader模式
2、在logstash主机/etc/logstash/conf.d/目录下创建收集系统日志和nginx日志的文件
[root@logstash-1 conf.d]# cat log-to-kafka.conf
input {
file {
path => "/var/log/messages"
type => "kafka-syslog-log-7-101" # 定义系统日志类型
start_position => "beginning"
stat_interval => "2"
}
file {
path => "/var/log/nginx/access.log"
type => "kafka-nginx-access-log-7-101" # 定义nginx的日志类型
start_position => "beginning"
stat_interval => "2"
codec => "json" # 如果是json格式的日志，必须写入json格式
}
}
output {
if [type] == "kafka-syslog-log-7-101" {
kafka {
topic_id => "kafka-syslog-log-7-101"
bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka的leader主机的IP地址上
codec => "json" # 写入的时候使用json编码，因为logstash收集后会转换为json格式
}
file {
path => "/tmp/syslog.log" # 将收集到的访问系统的日志临时存在此文件下，方便排错，解决问题后，可以删除此两项
}}
if [type] == "kafka-nginx-access-log-7-101" {
kafka {
topic_id => "kafka-nginx-access-log-7-101"
bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka主机的IP地址
codec => "json" # 写入的时候使用json编码，因为Logstash收集后会转换为json格式
}
file {
path => "/tmp/nginx-access.log" # 临时先存放在此文件下，方便排错，如果没有错误，可以删除此项
}}
}
2、验证logstash配置文件的正确性
[root@logstash-1 conf.d]# logstash -f log-to-kafka.conf
3、开始访问nginx网页，（192.168.7.101:80）再查看访问的临时log信息
4、在logstash主机上测试系统日志访问情况
[root@logstash-1 ~]# echo 1 >> /var/log/messages [root@logstash-1 ~]# echo 2 >> /var/log/messages [root@logstash-1 ~]# echo 3 >> /var/log/messages
5、查看此时的临时文件信息
此时测试logstash收集到日志写入到kafka中正常，可以将测试的file部分脚本删除掉，然后启动logstash服务。
file {
path => "/tmp/syslog.log" # 删除掉两个测试部分
}
file {
path => "/tmp/nginx-access.log"
}
启动D主机的logstash服务。
# systemctl start logstash
1、在logstash-A主机的/etc/logstash/conf.d/目录下创建一个收集kafka配置文件
input {
kafka {
topics => "kafka-syslog-log-7-101"
bootstrap_servers => "192.168.7.105:9092" # 从kafka的leader主机上提取缓存
codec => "json" # 在提取kafka主机的日志时，需要写成json格式
}
[root@logstash conf.d]# vim kafka-to-es_logstash.conf
kafka {
topics => "kafka-nginx-access-log-7-101"
bootstrap_servers => "192.168.7.105:9092" # 从kafka的leader主机提取数据
codec => "json" # 提取kafka日志时，需要写成json格式
}
}
output {
if [type] == "kafka-syslog-log-7-101" { # 转到elasticsearch主机上
elasticsearch {
hosts => ["192.168.7.100:9200"]
index => "kafka-syslog-7-101-%{+YYYY.ww}" # YYYY代表年，ww代表周
}}
if [type] == "kafka-nginx-access-log-7-101" { # 转到elasticsearch主机上
elasticsearch {
hosts => ["192.168.7.100:9200"]
index => "kafka-nginx-access-log-7-101-%{+YYYY.MM.dd}"
}}
}
2、启动logstash-B主机的服务器
# systemctl start logstash
3、在head插件上查看收集到的信息，已经收集到了syslog和nginx的log
1、在kibana创建索引
2、在discover查看nginx日志信息
3、在logstash-D主机的系统日志中输入信息，在discover查看系统日志信息
[root@logstash-1 conf.d]# echo 1 >> /var/log/messages [root@logstash-1 conf.d]# echo 2 >> /var/log/messages [root@logstash-1 conf.d]# echo 3 >> /var/log/messages
ELK之八----Logstash结合kafka收集系统日志和nginx日志
标签：sys tar root 定义 提取数据 oca put 信息 ble
原文地址：https://www.cnblogs.com/struggle-1216/p/12499293.html
