---------------- Filebeat+Kafka+ELK ----------------
1.部署 Zookeeper+Kafka 集群 (前面已经配过 20.0.0.101、20.0.0.102、20.0.0.103)
https://blog.csdn.net/m0_56509725/article/details/132908696?spm=1001.2014.3001.5501
1.1 配置ELK 在:
https://blog.csdn.net/m0_56509725/article/details/132853050?spm=1001.2014.3001.5501
2.部署 Filebeat
cd /usr/local/filebeat
vim filebeat.yml
filebeat.prospectors:
- type: log
enabled: true
paths:
- /var/log/httpd/access_log
tags: ["access"]
- type: log
enabled: true
paths:
- /var/log/httpd/error_log
tags: ["error"]
......
#添加输出到 Kafka 的配置
output.kafka:
enabled: true
hosts: ["20.0.0.101:9092","20.0.0.102:9092","20.0.0.103:9092"] #指定 Kafka 集群配置
topic: "httpd" #指定 Kafka 的 topic
#注意要把Elasticsearch output下的内容注释掉,否则会报错
#启动 filebeat
./filebeat -e -c filebeat.yml
3.部署 ELK,在 Logstash 组件所在节点上新建一个 Logstash 配置文件
cd /etc/logstash/conf.d/
vim kafka.conf
input {
kafka {
bootstrap_servers => "20.0.0.101:9092,20.0.0.102:9092,20.0.0.103:9092" #kafka集群地址
topics => "httpd" #拉取的kafka的指定topic
type => "httpd_kafka" #指定 type 字段
codec => "json" #解析json格式的日志数据
auto_offset_reset => "latest" #拉取最近数据,earliest为从头开始拉取
decorate_events => true #传递给elasticsearch的数据额外增加kafka的属性数据
}
}
output {
if "access" in [tags] {
elasticsearch {
hosts => ["20.0.0.105:9200"]
index => "httpd_access-%{+YYYY.MM.dd}"
}
}
if "error" in [tags] {
elasticsearch {
hosts => ["20.0.0.105:9200"]
index => "httpd_error-%{+YYYY.MM.dd}"
}
}
stdout { codec => rubydebug }
}
#启动 logstash
logstash -f kafka.conf
注:生产黑屏操作es时查看所有的索引:curl -X GET “localhost:9200/_cat/indices?v”
4.浏览器访问 http://20.0.0.106:5601 登录 Kibana,单击“Create Index Pattern”按钮添加索引“filebeat_test-*”,单击 “create” 按钮创建,单击 “Discover” 按钮可查看图表信息及日志信息。