使用 Logstash 计算特定日志事件的持续时间

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 就可以满足您的需求。诀窍是在收到PURCHASE 事件后立即搜索BEGIN_CUST 事件。该插件的文档包括可以满足您的大部分需求的示例，但这里是一个摘要：

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"
  }
}

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

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