正在计算事件间隔时间

Question

filebeat --> logstash --> elastic search --> kibana

@timestamp                      messageId                               event 
May 19th 2016, 02:55:29.003     00e02f2f-32d5-9509-870a-f80e54dc8775    system1Enter
May 19th 2016, 02:55:29.200     00e02f2f-32d5-9509-870a-f80e54dc8775    system1Exit
May 19th 2016, 02:55:29.205     00e02f2f-32d5-9509-870a-f80e54dc8775    system2Enter
May 19th 2016, 02:55:29.453     00e02f2f-32d5-9509-870a-f80e54dc8775    system2Exit

messageId                               in1:1->2:in2
00e02f2f-32d5-9509-870a-f80e54dc8775    197:5:248

Answer 1

您可以仅通过Logstashaggregate筛选器来实现这一点，但是，您必须基本上重新实现elapsed筛选器已经完成的工作，所以这将是一个遗憾，对吗？

然后，让我们混合使用Logstashaggregate过滤器和elapsed过滤器。后者用于测量每个阶段的时间，前者用于将所有时间信息聚合到最后一个事件中。

附注：您可能需要重新考虑您的时间戳格式，使其更符合解析的标准。我已经将它们转换为ISO 8601以使其更易于解析，但请随意滚动您自己的正则表达式。

2016-05-19T02:55:29.003 00e02f2f-32d5-9509-870a-f80e54dc8775 system1Enter
2016-05-19T02:55:29.200 00e02f2f-32d5-9509-870a-f80e54dc8775 system1Exit
2016-05-19T02:55:29.205 00e02f2f-32d5-9509-870a-f80e54dc8775 system2Enter
2016-05-19T02:55:29.453 00e02f2f-32d5-9509-870a-f80e54dc8775 system2Exit

filter {
  grok {
    match => ["message", "%{TIMESTAMP_ISO8601:timestamp} %{UUID:messageId} %{WORD:event}"]
    add_tag => [ "%{event}" ]
  }
  date {
    match => [ "timestamp", "ISO8601"]
  }
  # Measures the execution time of system1
  elapsed {
    unique_id_field => "messageId"
    start_tag => "system1Enter"
    end_tag => "system1Exit"
    new_event_on_match => true
    add_tag => ["in1"]
  }
  # Measures the execution time of system2
  elapsed {
    unique_id_field => "messageId"
    start_tag => "system2Enter"
    end_tag => "system2Exit"
    new_event_on_match => true
    add_tag => ["in2"]
  }
  # Measures the time between system1 and system2
  elapsed {
    unique_id_field => "messageId"
    start_tag => "system1Exit"
    end_tag => "system2Enter"
    new_event_on_match => true
    add_tag => ["1->2"]
  }
  # Records the execution time of system1
  if "in1" in [tags] and "elapsed" in [tags] {
    aggregate {
      task_id => "%{messageId}"
      code => "map['report'] = [(event['elapsed_time']*1000).to_i]"
      map_action => "create"
    }
  }
  # Records the time between system1 and system2
  if "1->2" in [tags] and "elapsed" in [tags] {
    aggregate {
      task_id => "%{messageId}"
      code => "map['report'] << (event['elapsed_time']*1000).to_i"
      map_action => "update"
    }
  }
  # Records the execution time of system2
  if "in2" in [tags] and "elapsed" in [tags] {
    aggregate {
      task_id => "%{messageId}"
      code => "map['report'] << (event['elapsed_time']*1000).to_i; event['report'] = map['report'].join(':')"
      map_action => "update"
      end_of_task => true
    }
  }
}

{
                 "@timestamp" => "2016-05-21T04:20:51.731Z",
                       "tags" => [ "elapsed", "elapsed_match", "in1" ],
               "elapsed_time" => 0.197,
                  "messageId" => "00e02f2f-32d5-9509-870a-f80e54dc8775",
    "elapsed_timestamp_start" => "2016-05-19T00:55:29.003Z"
}

{
                 "@timestamp" => "2016-05-21T04:20:51.734Z",
                       "tags" => [ "elapsed", "elapsed_match", "1->2" ],
               "elapsed_time" => 0.005,
                  "messageId" => "00e02f2f-32d5-9509-870a-f80e54dc8775",
    "elapsed_timestamp_start" => "2016-05-19T00:55:29.200Z"
}

{
                 "@timestamp" => "2016-05-21T04:20:51.736Z",
                       "tags" => [ "elapsed", "elapsed_match", "in2" ],
               "elapsed_time" => 0.248,
                  "messageId" => "00e02f2f-32d5-9509-870a-f80e54dc8775",
    "elapsed_timestamp_start" => "2016-05-19T00:55:29.205Z"
                     "report" => "197:5:248"
}