当系统规模比较大，模块非常多时，或者一些重要日志需要归档时，就需要有个日志统一管理平台。本人所在公司有个需求，边缘端的日志需要统一上传管理，方便后续问题定位。方案是边缘端定时打包日志上传到fastdfs，然后把url推送到业务端，业务根据url从fastdfs上获取到日志，解压缩到指定目录，filebeat获取到日志发生变化，然后解析发送到kafka，logstash监听kafka把日志存储到es里面，然后业务系统可以在界面查询。本文主要介绍EKL搭建，对于业务系统ES操作封装后续有时间介绍。
环境：filebeat(8.0.0)+logstash(8.0.0)+kafka(2.8.0)+es(7.14.0)+kibana(7.14.2)
一、filebeat搭建
1、从官网下载Download Filebeat ? Lightweight Log Analysis | Elastic，上传到服务器，进入上传目录，执行命令：tar -zxvf filebeat-8.0.0-linux-x86_64.tar.gz
2、修改配置文件filebeat.yml，只要配置文件路径和输出kafka
?
output.kafka:
? enabled: true
? hosts: ["127.0.0.1:9092"]
? topic: 'filebeat-logs'
? partition.round_robin:
??? reachable_only: false
? required_acks: 1
? compression: gzip
? max_message_bytes: 1000000
? username: demo
? password: demo@123
? sasl.mechanism: PLAIN
? client_id: filebeat202
说明：本人配置了kafka服务端连接需要登录验证，需要配置username/password/sasl.mechanism，如kafka不需要登录验证，则不需要配置这三个选项
3、启动filebeat，执行nohup ./filebeat -e -c filebeat.yml &
二、kafka搭建?
可参照前期写的kafka搭建
kafka服务端设置用户和密码登录及springboot访问实现_guaotianxia的博客-CSDN博客_登陆kafka
三、logstash搭建?
1、下载logstash （https://www.elastic.co/cn/downloads/logstash）上传到指定目录，并执行命令：tar -zxvf?logstash-8.0.0-linux-x86_64.tar.gz
2、创建logstash.conf文件，?可以cp logstash-sample.conf logstash.conf,添加如下内容
?
input {
#连接kafka，从kafka获取日志内容
? kafka {
??? bootstrap_servers => "127.0.0.1:9092"
??? group_id => "logstash_log_202"
??? client_id => "logstash1"
??? auto_offset_reset => "latest"
??? topics => ["filebeat-logs"]
??? #add_field => {"logs_type" => "grandhonor_log"}
??? codec => json { charset => "UTF-8" }
??? sasl_mechanism => "PLAIN"
??? security_protocol => "SASL_PLAINTEXT"
?sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='hehe'? password='123456';"
? }
}
filter {
? ?#格式化@timestamp时间为年月，并赋值为indexmonth变量
???? ruby {
???????code => 'event.set("indexmonth", event.get("@timestamp").time.localtime.strftime("%Y%m"))'
???? }
??? grok {
? ? ? ?#解析日志内容，可匹配多种日志内容，获取日志时间、模块名、日志级别和内容。具体写法参照截屏图片，因为正则匹配csdn会无法显示，反斜杠都改为双反斜杠
??????? match => {
???????? "message" => "\\[%{TIMESTAMP_ISO8601:logTime}\\]\\s{0,}\\[%{DATA:moduleName}\\]\\s{0,}\\[%{LOGLEVEL:logLevel}\\]\\s{0,}%{GREEDYDATA:content}","\\[%{TIMESTAMP_ISO8601:logTime}\\]\\s{0,1}%{LOGLEVEL:logLevel}\\s{0,1}%{GREEDYDATA:content}"
??????? }
??? }
? #如果无法从日志内容解析到日志时间，则使用自带的时间戳
??? if ![logTime]{
?????? mutate {
???????? copy => { "@timestamp" => "logTime" }
????? }
??? }
? #如果获取到日志级别，格式化为小写
??? if [logLevel]{
?????? mutate {
???????? lowercase => ["logLevel"]
????? }
??? }
? #使用时间控件格式化时间，否则如es字段默认为text格式
??? date{
??????? match => ["logTime","yyyy-MM-dd HH:mm:ss.SSS","yyyy-MM-dd HH:mm:ss","ISO8601"]
??????? target => "logTime"
??????? locale => "cn"
?? }
??? grok {
? ? ? ?#解析日志路径，获取设备序列号
??????? match => {
???????? "log" => "%{GREEDYDATA}/%{WORD:deviceSn}-%{GREEDYDATA}\.log"
? ??????}
??? }
?? mutate {
? ? ? #添加临时字段，不会写入es
????? add_field => { "[@metadata][target_index]" => "logstash-log-%{+YYYY.MM}" }
? ? ?#重命名字段值
????? rename => ["@timestamp","reportTime"]
? ? #删除字段，不写入es
remove_field=> ["host","agent","ecs","tags","fields","@version","input","event","container","offset"]
?? }
}
output {
? elasticsearch {
??? hosts => ["http://192.168.103.202:9200"]
??? index => "logstash_log-%{indexmonth}"
??? #index => "%{[@metadata][target_index]}"
??? user => "elastic"
??? password => "123456"
? }
}
说明：在写grok，解析日志内容为结构化数据时，可以使用kibana的Dev tools进行测试，如下
四.es搭建
参照前期写的es搭建
linux环境skywalking搭建及项目应用_guaotianxia的博客-CSDN博客_linux安装部署skywalking
五、kibana搭建
1、从官网下载（Download Kibana Free | Get Started Now | Elastic），执行tar -zxvf ?kibana-7.14.2-linux-x86_64.tar.gz
2、进入config目录，修改kibana.yml配置文件
3、启动，执行命令：nohup ./kibana > /dev/null 2>&1 &
4、浏览器输入：http://192.168.103.202:5601，输入用户名和密码
?六、验证
使用kibana的dev tools，查询是否有日志写入
GET logstash_log-202204/_search
{?
? "size": 20,?
? "query": {
? ? "match_all": {}
? }
}
