目录
概要：
1.待分析日志的格式（1条日志）：
2.filebeat 设置获取日志的路径：
3.logstash 匹配日志
4.调试
5.Elasticsearch 插件（ 方便对es的查询与删除）
6.对已有日志进行分析
7、Kibana展示
概要：
使用kibana展示日志分析结果的初衷是性能分析，终于分析出了第一份日志。性能压测情况下，某个日志文件的分析出的响应时间分布。
上篇文章请参看：
logstash config filter 配置（grok、date、ruby）:日志拆分转换并展示在kibana中_baidu_41614347的博客-CSDN博客概要：ELK部署成功后，需要kibana图形展示某应用的性能。初步通过统计分析日志的形式来模拟。日志中有sendTime ：消息发出时间，recvTime:处理完毕后打印的日志时间。通过logstash对日志进行拆分并计算recvTime和sendTime的差值即处理时间（本文标记为responseTime）。并将responseTime展示在kibana中1、logstash配置文件logstash的配置文件input是来自filebeat端口5044 （filebeat用于收集ou...https://blog.csdn.net/baidu_41614347/article/details/120991440目前的kibana的展示如下图，左侧横轴是消息接收到的时间，纵轴是reponseTime的值，右侧图形是日志的具体信息，展示的字段有recvTimeD ,responseTime,recvID,message(可以根据需要自己添加)方便查看日志。
下面是详细的操作过程：
1.待分析日志的格式（1条日志）：
2021-10-08 15:00:56,684 INFO [SseMarketConsumer] c.c.c.s.m.s.c.SseMarketConsumer [78] poll msg: StdImsMsg{msgId=1,sendTime=1633676456341,msgType=SSE_SNAP_QUOT,msgSource=SSE-MDGW,comstarOrgId=null,sourceOrgId=null,comstarCategory=1,comstarCode=018009-SSEO,securityKey=018009,serializerCode=7,bodyIdx=62,byteSize=1091}
2.filebeat 设置获取日志的路径：
fields：向输出的每一条日志添加额外的信息，比如"log_source : outlog1"，方便后续对日志进行分组统计。默认情况下，会在输出信息的fields子目录下以指定的新增fields建立子目录。log_source该字段在logstash 配置文件中会用到
3.logstash 匹配日志
（双引号，<>,[]这些需要转义）
grok中要匹配的message ：
"message" , "%{TIMESTAMP_ISO8601:recvTime} INFO .* StdImsMsg{msgId=%{NUMBER:recvID},sendTime=%{NUMBER:sendTime},msgType.*}"
详细看下面的代码
input {
beats {
port => "5044"
}
}
filter{
if [fields][log_source] == "outlog" {
grok{
match => [
"message" , "<%{TIMESTAMP_ISO8601:recvTime}> INFO recv :{"ID":"%{NUMBER:recvID}",.*"sendTime":"%{TIMESTAMP_ISO8601:sendTime}"}"
]
}
date {
match => ["recvTime", "yyyy-MM-dd HH:mm:ss,SSS"]
target => "recvTimeD"
}
date {
match => ["sendTime", "yyyy-MM-dd HH:mm:ss,SSS"]
target => "sendTimeD"
}
ruby {code => "event.set('responseTime', event.get('recvTimeD').to_i - event.get('sendTimeD').to_i)"}
mutate {
lowercase => ["log_source"]}
}
else if [fields][log_source] == "outlog1" {
grok{
match => [
"message" , "%{TIMESTAMP_ISO8601:recvTime} INFO .* StdImsMsg{msgId=%{NUMBER:recvID},sendTime=%{NUMBER:sendTime},msgType.*}"
]
}
date {
match => ["recvTime", "yyyy-MM-dd HH:mm:ss,SSS"]
target => "recvTimeD"
}
ruby {
code => "
event.set('recv_time',(event.get('recvTimeD').to_f.round(3)*1000).to_i)
event.set('responseTime', event.get('recv_time') - event.get('sendTime').to_i)
"
}
}
else {
drop{}
}
}
#output {
# stdout {
# }
#}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "%{[@metadata][beat]}-%{[fields][log_source]}-%{+YYYY.MM.dd}"
}
}
4.调试
在开始匹配的时候可以output 标准输出 调试成功后在输出到elasticsearch中
./bin/logstash -f ./config/logstash-beat.conf --config.reload.automatic 可以自动加载config 不需要每次修改配置后重启logstash
用print 命令调试：
printf "2021-10-08 15:00:57,685 INFO [SseMarketConsumer] c.c.c.s.m.s.c.SseMarketConsumer [78] poll msg: StdImsMsg{msgId=1,sendTime=1633676456341,msgType=SSE_SNAP_QUOT,msgSource=SSE-MDGW,comstarOrgId=null,sourceOrgId=null,comstarCategory=1,comstarCode=018009-SSEO,securityKey=018009,serializerCode=7,bodyIdx=62,byteSize=1091}n" >>out1.log
如果是标准输出 这个时候界面上会出来日志分析结果可以初步判断一下 responseTime 是否正确。
5.Elasticsearch 插件（ 方便对es的查询与删除）
谷歌可以用es插件
1.谷歌安装插件
右侧三小点-更多工具-扩展程序 点击加载已解压的扩展程序 （选择elasticsearch-head的插件解压后的文件夹）
如果提示：此扩展程序可能已损坏 则有可能文件真的损坏了，可以更换或者高版本的插件，而且拷贝传输的过程中最好是压缩包
压缩包：链接链接：https://pan.baidu.com/s/1Cuu3mUvNPJhw83XO7i3-iw
提取码：oarw
选择扩展程序-elasticsearch-Head 打开后输入es的IP地址和端口号 然后点击链接按钮 就可以看es 中的概要信息，索引，数据浏览等，还支持es查询
6.对已有日志进行分析
在确保out1.log这个文件的日志可以正常获取和分析后 将需要分析的日传到out1.log中
cat hangqing.log < out1.log
此时查询es 可以看到导入的数据
7、Kibana展示
1.建立索引模式：
2. dashboard中新增图表
3.最后调整，可以去掉第一条的异常值（这条是测试用的测试数据）
3.调整图表展示方式。
综上 Elastic Stack 可以实时分析日志，也可以分析历史已有日志，前期投入可能会比较大，但是一次配置结束后 可以长期使用。
