filebeats+logstash将日志数据实时同步导入至ES
一、主要流程在存储日志文件的客户端机器filebeats,用来获取日志数据,通过指定5044端口,同步传输至logstash,logstash进行相关解析,同步导入至ES。二、安装filebeats、logstash官网下载对应版本filebeats、logstash的tar压缩包,解压至指定目录(/usr/local)即可完成安装。三、编写logstash配置文件,指定从5044端口获取beat
一、主要流程
在存储日志文件的客户端机器安装filebeats,用来获取日志数据,通过指定5044端口,同步传输至logstash,logstash进行相关解析,同步导入至ES。
二、安装filebeats、logstash
官网下载对应版本filebeats、logstash的tar压缩包,解压至指定目录(/usr/local)即可完成安装。
三、编写logstash配置文件,指定从5044端口获取beats传输的数据
创建文件夹/usr/local/logstash-7.13.3/myconf/,存放配置文件。
编辑配置文件:vim apilog.conf,内容如下:
input {
beats {
port => 5044
}
}
filter {
grok {
#patterns_dir => ["./patterns"]
match => { "message" => "%{TIMESTAMP_ISO8601:time} %{LOGLEVEL:level} %{GREEDYDATA:data}" }
}
mutate {
remove_field => ["ecs","agent","host","tags","input"]
}
}
output {
elasticsearch {
hosts => ["http://192.168.163.132:9200"]
index => "apilog"
}
}
各插件解释:
input:
从5044端口获取beats传输的数据
filter:
对日志数据进行解析,使用grok插件,进行正则匹配。
日志文件中,默认每行的数据为一条message,使用正则表达式,解析message,获取指定内容。
2021-07-28 09:06:56,204 DEBUG org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager:234 - Connection manager shut down
2021-07-28 09:06:56,234 DEBUG io.lettuce.core.RedisClient:466 - Initiate shutdown (100, 100, MILLISECONDS)
2021-07-28 09:06:56,324 DEBUG io.lettuce.core.resource.DefaultClientResources:599 - Initiate shutdown (0, 2, SECONDS)
2021-07-28 09:06:56,348 DEBUG io.lettuce.core.resource.DefaultEventLoopGroupProvider:245 - Initiate shutdown (0, 2, SECONDS)
2021-07-28 09:06:56,410 INFO com.baomidou.dynamic.datasource.DynamicRoutingDataSource:184 - dynamic-datasource start closing ....
2021-07-28 09:06:56,411 INFO com.zaxxer.hikari.HikariDataSource:350 - oauth - Shutdown initiated...
2021-07-28 09:06:56,496 INFO com.zaxxer.hikari.HikariDataSource:352 - oauth - Shutdown completed.
使用grok插件解析获取message中的【时间】【日志等级】【日志内容】。
对应的正则表达式为:
match => { "message" => "%{TIMESTAMP_ISO8601:time} %{LOGLEVEL:level} %{GREEDYDATA:data}" }
%{TIMESTAMP_ISO8601:time},获取该条message中的时间,新增索引字段time并赋值
%{LOGLEVEL:level},获取该条message中的日志等级,新增索引字段level并赋值
%{GREEDYDATA:data},获取该条message中的日志内容,新增索引字段data并赋值
使用mutate插件,去除对应es索引中不需要的字段
mutate {
remove_field => ["ecs","agent","host","tags","input"]
}
output:
指定数据导入目标为ES,配置对应信息:
hosts =>【主机:端口】
index =>【索引名称】。
四、编写filebeats配置文件,配置文件名可自定义,启动时指定对应配置文件即可
vim myfilebeat.yml,编辑添加如下配置代码,注意严格遵守缩进规则,每个层级两个空格缩进。
#log4j log
filebeat.inputs:
- type: log
paths:
- /var/logtest/*.log
multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
output.logstash:
hosts: ["192.168.163.132:5044"]
配置解释:
filebeat.inputs:
- type: log
paths:
- /var/logtest/*.log
指定读取内容类型为log,指定log文件所在路径
multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
以时间格式xxxx-xx-xx来判断每行的开头,如果不是时间开头,则判断为与上一个时间开头的行为同一行,用来处理日志内容跨行的情况↓:
output.logstash:
output.logstash:
hosts: ["192.168.163.132:5044"]
指定数据同步至5044端口(之前logstash配置的beats数据接收端口)
配置完成。
五、启动logstash、filebeats
启动logstash:
/usr/local/logstash-7.13.3/bin/logstash -f /usr/local/logstash-7.13.3/myconf/apilog.conf
启动filebeats:
/usr/local/filebeat-7.13.3-linux-x86_64/filebeat -e -c /usr/local/filebeat-7.13.3-linux-x86_64/myfilebeat.yml
有新日志生成,filebeats则会发现并触发同步:
六、检查ES对应日志索引同步情况
索引概览:
指定时间查询:
查询结果:
更多推荐
所有评论(0)