elasticsearch - 使用Logstash计算特定Log事件的持续时间

Question

目标:我想使用logstash计算特定事件持续了多长时间。

场景:考虑一位正在从我的页面搜索要购买产品的客户。他正在访问的每个页面以及持续时间都被记录在日志中。现在，我想查找普通客户花多长时间才能获得产品。以及我的服务器花多长时间回复他。

现在这是我的日志文件:16-09-2004 00:37:22 BEGIN_CUST
ts:16-09-2004T00:37:26+05:30
ID-XYZ456
16-09-2004 00:37:23 PAGE_1
ID-XYZ456
ts:16-09-2004T00:39:26+05:30
16-09-2004 00:37:23 PAGE_2
ID-XYZ456
ts:16-09-2004T00:41:26+05:30
16-09-2004 00:37:23 BUT_REQ
ID-XYZ456
ts:16-09-2004T00:43:26+05:30
16-09-2004 00:37:23 PURCHASE
ID-XYZ456
ts:16-09-2004T00:47:26+05:30
16-09-2004 00:51:22 BEGIN_CUST
ts:16-09-2004T00:52:26+05:30
ID-YUB98I
16-09-2004 00:53:23 PAGE_1
ID-YUB98I
16-09-2004 00:55:23 PURCHASE
ID-YUB98I
在上面的日志文件中，很明显BEGIN_CUST是事件的开始，而PURCHASE是事件的结束。
ID(充当每个客户的唯一ID)。

我已经尝试了脚本字段。但是由于以下几点，它无法给我带来正确的结果，

不需要客户购买它。

客户购买可能持续几秒钟。

还有什么更好的方法可以在Kibana的单独字段中绘制单个客户的工期，以使用Logstash对其进行可视化。

提前致谢。

Answer 1

只要您使用ElasticSearch作为商店，the elasticsearch filter就能满足您的需求。诀窍是在您获得BEGIN_CUST事件后立即搜索PURCHASE事件。该插件的文档包括和示例，这些示例可以完成您所需要的大部分工作，但是这里是一个摘要:

if [trans_type] == "PURCHASE" {
  elasticsearch {
    hosts => localhost,
    query => "trans_type:BEGIN_CUST AND cust_id:%{[cust_id]}],
    fields => { "@timestamp" => "started" }
  }
  date {
    match => [ "[started]", "ISO8601" ]
    target => "[started]"
  }
  ruby {
    code => "event['shopping_time'] = (event['@timestamp'] - event['started'] rescue nil"
  }
}

这将产生一个 shopping_time字段，以 BEGIN_CUST记录到达与第一个 PURCHASE到达之间的秒数为单位。如果客户购买两次，则每个 PURCHASE记录将基于相同的 shopping_time拥有自己的 BEGIN_CUST字段。

通过查询ElasticSearch的BEGIN_CUST记录，并在PURCHASE记录的 @timestamp字段中使用该记录上的 started数据，可以实现此目的。然后， date {}过滤器将其转换为日期时间数据类型。最后， ruby {}块计算当前 @timestamp字段与从ElasticSearch中拉出的字段之间的时间差，从而创建 shopping_time字段。