Splunk - Use streamstats with transaction command to find gaps in event logs

Question

I need to capture gaps in time between logs for a specific event. I have successfully captured all the logs associated with the event (publication) using the transaction command. In a separate search, I am able to find the gaps in logs, but am struggling to combine the two into one search to find the time gaps in event logs by transaction.

Note: There can be multiple unique transactions associated with a publicationID which is why I think I need to use the transaction command to accurately capture the start and end to a publication request transaction.

This search captures the set of logs associated with a publication transaction.

index=test *publication*| transaction publicationID startswith=(message="*Request*publication*")

This search returns the gaps I'm looking for, but only when I narrow the search to a specific publicationID and time period.

index=test *publication*| where publicationID="55432556"| streamstats current=f last(_time) as last_time by publicationID | eval gap = last_time - _time | where gap > 3600 | eval gap= tostring(gap,"duration") | convert ctime(last_time) as last_time ctime(_time) as _time| table publicationID _time last_time gap

Am I going about this all wrong? How can I best capture gaps greater than 1 hour in event logs by transaction for a set of transactions instead of having to include where criteria for a specific publicationID?

Ideally, I want the results to look something like this:

publicationID _time last_time gap55432556 11/05/2022 14:41:06 11/05/2022 16:44:06 01:04:1855432556 11/05/2022 17:30:00 11/05/2022 19:30:06 02:00:0633932543 11/06/2022 08:30:00 11/06/2022 09:30:06 01:00:0688465272 11/06/2022 11:25:00 11/06/2022 14:25:55 03:00:5555432556 11/07/2022 12:30:00 11/07/2022 16:30:00 04:00:00

Best Answer

Accepted Answer

Without knowing what your data looks like, we can only guess, but something like the following should work:

index=ndx sourcetype=srctp message=* publicationID=*| fields - _raw| fields publicationID _time | sort 0 publicationID _time| eval message=if(match(message,"Request.+publication"),message+tostring(_time),message)| rex field=message "Request.+publication(?<begin>.+)"| filldown begin| stats min(_time) as start max(_time) as end by begin publicationID| eval gap=end-start| where gap>3600| eval start=strftime(start,"%c"), end=strftime(end,"%c")

What this does:

a little housekeeping (keep only the fields we need, chuck the rest)
[re]sort to ensure we're grouped by publicationID, then by _time instead of only by _time
add a unique identified to message if it matches the pattern for being a beginning value
extract a custom field (begin)
if message doesn't contain a value, begin will be null
filldown the value of begin (works just like Excel's filldown functionality)
stats-out the start and end times of every individual publicationID event set (a la transaction, but faster and more flexible)
filter every line to ensure there is more than a 1 hour (3600 second) "gap" (perhaps this would be better worded "duration"?)
format the start and end times for readability

Splunk - Use streamstats with transaction command to find gaps in event logs

Best Answer

Random Posts