我有一条消息流过几个系统，每个系统都记录消息的入口和出口，并带有时间戳和uuid messageId。 我通过以下方式摄取所有日志：
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做这个的最好方式是什么？ Logstash过滤器？ kibana计算字段？
I have a message that flows through several systems, each system logs message entry and exit with a timestamp and a uuid messageId. I'm ingesting all logs through:
filebeat --> logstash --> elastic search --> kibanaAs a result I now have these events:
@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 system2ExitI would like to produce a report (ideally a stacked bar or column) of time spent in each system:
messageId in1:1->2:in2 00e02f2f-32d5-9509-870a-f80e54dc8775 197:5:248What is the best way to do this? Logstash filters? kibana calculated fields?
最满意答案
您只能使用Logstash aggregate过滤器来实现这一点，但是，您必须重新实现elapsed过滤的过滤器已经完成的工作，所以这会是一种耻辱，对吧？
然后让我们使用Logstash aggregate过滤器和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首先，我使用三个elapsed过滤器（每个阶段in1 1->2和in2 ），然后使用三个聚合过滤器来收集所有的时间信息。 它看起来像这样：
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 } } }在前两个事件之后，你会得到一个像这样的新事件，这表明在system1中花费了197ms：
{ "@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" }第三次事件发生后，您会看到类似这样的事件，它显示了system1和system2之间花了多少时间，即5ms：
{ "@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" }第四次事件发生后，你会得到一个像这样的新事件，它显示了系统2花费了多少时间，即248ms。 该事件还包含report字段，其中包含消息的所有时间信息
{ "@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" }You can achieve this with the Logstash aggregate filter only, however, you'd have to substantially re-implement what the elapsed filter already does, so that'd be a shame, right?
Let's then use a mix of the Logstash aggregate filter and the elapsed filter. The latter is used to measure the time of each stage and the former is used to aggregate all the timing information into the last event.
Side note: you might want to rethink your timestamp format to make it something more standard for parsing. I've transformed them to ISO 8601 to make it easier to parse, but feel free to roll your own regex.
So I'm starting from the following logs:
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 system2ExitFirst I'm using three elapsed filters (one for each stage in1, 1->2 and in2) and then three aggregate filters in order to gather all the timing information. It looks like this:
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 } } }After the first two events, you'll get a new event like this, which shows that 197ms have been spent in system1:
{ "@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" }After the third event, you'll get an event like this, which shows how much time is spent between system1 and system2, i.e. 5ms:
{ "@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" }After the fourth event, you'll get a new event like this one, which shows how much time was spent in system2, i.e. 248ms. That event also contains a report field with all the timing information of the message
{ "@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" }更多推荐
发布评论