ELK通过Logstash收集日志并写入Kafak

一、Kafak 部署

zookeeper 集群部署参考csdn连接&＃xff1a;https://blog.csdn.net/qq_42606357/category_11241804.html
kafka 集群部署参考csdn连接&＃xff1a;https://linck.blog.csdn.net/article/details/119257228

二、配置WEB服务器的日志写入kafka

root&＃64;web1:/etc/logstash/conf.d# vim nginx-log-to-kafka.conf
input {
file {
path &＃61;> "/var/log/nginx/access.log"
type &＃61;> "nginx-accesslog"
start_position &＃61;> "beginning"
stat_interval &＃61;> "3 second"
codec &＃61;> "json"
}
file {
path &＃61;> "/apps/nginx/logs/error.log"
type &＃61;> "nginx-errorlog"
start_position &＃61;> "beginning"
stat_interval &＃61;> "3 second"
}
}
output {
if [type] &＃61;&＃61; "nginx-accesslog" {
kafka {
bootstrap_servers &＃61;> "10.0.0.40:9092,10.0.0.41:9092,10.0.0.42:9092"
topic_id &＃61;> "lck-nginx-accesslog"
codec &＃61;> "json"
}
}
if [type] &＃61;&＃61; "nginx-errorlog" {
kafka {
bootstrap_servers &＃61;> "10.0.0.40:9092,10.0.0.41:9092,10.0.0.42:9092"
topic_id &＃61;> "lck-nginx-errorlog"
codec &＃61;> "json"
}
}
}


三、检测Logstash配置文件语法是否正确

root&＃64;web1:~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/nginx-log-to-kafka.conf -t


四、启动服务并验证

root&＃64;web1:~# systemctl restart logstash.service


尝试访问页面&＃xff0c;生成日志&＃xff0c;到kafak客户端查看是否生成日志

五、配置其他 logstash 服务器从 kafak 读取数据并传到 elasticsearch

10.0.0.35 安装 logstash

apt install -y openjdk-8-jdk
# 将 logstash-7.12.1-amd64.deb 软件包传到 /usr/local/src 目录下&＃xff0c;并进行安装
dpkg -i /usr/local/src/logstash-7.12.1-amd64.deb


修改配置文件

root&＃64;ubuntu1804:~# vim /etc/logstash/conf.d/kafka-to-es.conf
input {
kafka {
bootstrap_servers &＃61;> "10.0.0.40:9092,10.0.0.41:9092,10.0.0.42:9092"
topics &＃61;> ["lck-nginx-accesslog","lck-nginx-errorlog"]
codec &＃61;> "json"
}
}
output {
if [type] &＃61;&＃61; "nginx-accesslog" {
elasticsearch {
hosts &＃61;> ["10.0.0.31:9200"]
index &＃61;> "kafka-nginx-newindex-accesslog-%{&＃43;YYYY.MM.dd}"
}
}
if [type] &＃61;&＃61; "nginx-errorlog" {
elasticsearch {
hosts &＃61;> ["10.0.0.31:9200"]
index &＃61;> "kafka-nginx-newindex-errorlog-%{&＃43;YYYY.MM.dd}"
}
}
}


六、检测Logstash配置文件语法是否正确

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis-to-es.conf -t


七、启动服务并验证

systemctl restart logstash.service


八、创建索引方便查询日志

8.1 创建访问日志索引

8.2 创建错误日志索引