概要：ELK部署成功后，需要kibana图形展示某应用的性能。初步通过统计分析日志的形式来模拟。日志中有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 recv :{"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 查看到的字段 类型

ruby：插件确实不错,可以随意实现日志解析,相比与正则和直接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\"}\n"  >>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的如何配置 的内容请参考：1、ELK部署（filebeat、logstash、elasticsearch、kibana）_baidu_41614347的博客-CSDN博客

 达到了我的目的 kibana 中可以实时展示 消息的响应时间。