概要:ELKapp开发定制公司部署成功后,需要kibanaapp开发定制公司图形展示某应用的性能。app开发定制公司初步通过统计分析日志的形式来模拟。日志中有sendTime :消息发出时间,recvTime:处理完毕后打印的日志时间。通过logstash 对日志进行拆分并计算recvTime和sendTime的差值即处理时间(本文标记为responseTime)。并将responseTime展示在kibana中
1、logstash 配置文件
logstash 的配置文件 input 是来自filebeat 端口5044 (filebeat 用于收集out.log的日志)
filter 是对日志内容进行匹配拆分和转换的操作。
logstash-beat.conf 配置文件内容
- input {
- beats {
- port => "5044"
- }
- }
-
-
-
- filter{
- 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)"}
-
- }
-
-
- output {
- stdout {
- }
- }
-
- #output {
- # elasticsearch {
- # hosts => ["http://localhost:9200"]
- # index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
- # }
- #}
grok:是一种采用组合多个预定义的正则表达式,用来匹配分割文本并映射到关键字的工具。通常用来对日志数据进行预处理。logstash的filter模块中grok插件是其实现之一。
eg:日志格式
<2021-10-27 13:02:05,201> INFO :{"ID":"000007","Bond":"100001","BondName":"XN0001","sendTime":"2021-10-27 13:02:00,301"}
grok匹配 :
"message" , "\<%{TIMESTAMP_ISO8601:recvTime}\> INFO recv :{\"ID\":\"%{NUMBER:recvID}\",.*\"sendTime\":\"%{TIMESTAMP_ISO8601:sendTime}\""
(说明:message 是logstash 会把收到的日志内容存储在message字段中;
标红的部分是需要将一些值存储到属性中 比如:%{TIMESTAMP_ISO8601:recvTime} ,recvTime 会匹配到 2021-10-27 13:02:05,201;
可以用到正则表达式,日志中原有的< " 需要转义字符前面加\ )
Date:日期拆件
可以把存放在recvTime 和sendTime 中匹配到text 类型的日期转换为 date 类型
date {
match => ["recvTime", "yyyy-MM-dd HH:mm:ss,SSS"]
target => "recvTimeD"
}
以上配置 是将recvTime 中的值转换为 “yyyy-MM-dd HH:mm:ss,SSS”格式 并存储到target
下面是通过kibana 查看到的字段 类型
:插件确实不错,可以随意实现日志解析,相比与正则和直接logstash中的json插件要方便灵活的多。 event.get 可以获取到logstash 中属性的值 event.set 是为属性赋值。 下面是分别获取recvTimeD和sendTimeD的值并转换成int类型 然后相减之后赋值给 responseTime。
ruby {code => "event.set('responseTime', event.get('recvTimeD').to_i - event.get('sendTimeD').to_i)"}
2、 现象验证:
启动logstash 标准输出 :
在out.log中输入日志信息:
-
-
- 日志:<2021-10-27 13:02:05,201> INFO recv :{"ID":"000007","Bond":"100001","BondName":"XN0001","sendTime":"2021-10-27 13:02:00,301"}
-
- 用命令的方式追加日志:
- printf "<2021-10-27 13:02:05,201> INFO recv :{\"ID\":\"000007\",\"Bond\":\"100001\",\"BondName\":\"XN0001\",\"sendTime\":\"2021-10-27 13:02:00,301\"}" >>out.log
logstash 输出:可以看到responseTime 得到的值是5
- {
- "responseTime" => 5,
- "@version" => "1",
- "sendTime" => "2021-10-27 13:02:00,301",
- "ecs" => {
- "version" => "1.11.0"
- },
- "recvID" => "000007",
- "message" => "<2021-10-27 13:02:05,201> INFO recv :{\"ID\":\"000007\",\"Bond\":\"100001\",\"BondName\":\"XN0001\",\"sendTime\":\"2021-10-27 13:02:00,301\"}",
- "tags" => [
- [0] "beats_input_codec_plain_applied"
- ],
- "@timestamp" => 2021-10-27T05:48:09.913Z,
- "log" => {
- "offset" => 1638,
- "file" => {
- "path" => "/home/chen/tool/log/out.log"
- }
- },
- "input" => {
- "type" => "log"
- },
- "host" => {
- "os" => {
- "version" => "7 (Core)",
- "platform" => "centos",
- "type" => "linux",
- "name" => "CentOS Linux",
- "codename" => "Core",
- "family" => "redhat",
- "kernel" => "3.10.0-1160.31.1.el7.x86_64"
- },
- "mac" => [
- [0] "52:54:00:aa:64:82"
- ],
- "architecture" => "x86_64",
- "ip" => [
- [0] "10.0.4.10",
- [1] "fe80::5054:ff:feaa:6482"
- ],
- "name" => "VM-4-10-centos",
- "hostname" => "VM-4-10-centos",
- "containerized" => false,
- "id" => "2eaabde29fb446cb8202d9ac64d99f37"
- },
- "recvTime" => "2021-10-27 13:02:05,201",
- "sendTimeD" => 2021-10-27T05:02:00.301Z,
- "agent" => {
- "version" => "7.15.0",
- "type" => "filebeat",
- "name" => "VM-4-10-centos",
- "hostname" => "VM-4-10-centos",
- "ephemeral_id" => "b3fd97d2-c8ce-45d2-8e52-6a609c366270",
- "id" => "1d86f210-382a-4889-8d0a-d99f642ae0e3"
- },
- "recvTimeD" => 2021-10-27T05:02:05.201Z
- }
重新停掉logstash 并修改配置 输出到es 启动logstash 在kibana查看现象如下图:
关于ELK的如何配置 的内容请参考:
达到了我的目的 kibana 中可以实时展示 消息的响应时间。