Logstash event @timestamp adjustment

elklogstashrubytimestamp

I have standard Windows IIS log files with event date/time stamp information and timetaken (in milliseconds).

I would like to be able to adjust the event time (@timestamp) by subtracting the "timetaken" to be able to record when the event started rather than when it was completed.

I've looked at using the ruby plug in but my ruby knowledge is zero.

My first attempt is this:

ruby {
code => "event['@timestamp_adj'] = (event['@timestamp'].to_f - (event['timetaken'].to_f/1000)))"
}

However, this casts the timestamp to a numeric, how can get it back to a date?

Some sample date: (with redactions and mangles for sensitive data)

{
"@timestamp" => "2015-10-22T22:59:49.000Z",
 "timestamp" => "2015-10-22 23:59:49",
    "method" => "GET",
      "page" => "/spacer.gif",
  "response" => "200",
 "timetaken" => "2120",
"@timestamp_adj" => 1445554789.0
}

In this (made up) case, the event time is 22:59:49.000 and took 2.120 seconds (2,120 milliseconds) to complete. What I want to have is @timestamp_adj to record 22:59:48.880

Either:

How do I convert the now "numeric" timestamp back to a time string that Elasticsearch will recognise?
or how to I do Date/time math without corrupting the date/time encoding?
Thanks in advance.

Best Answer

You can create a new time object with the timestamp and the subtract the time taken like this

Time.new(event['@timestamp_adj']) - (event['timetaken'] / 1000)

Related Solutions

Linux – timestamp +5 hours logstash

The timestamp you gave in your example appears to be correct. It was about 18 minutes before you posted this question.

It appears your server (I presume you have a good reason, but you might not) is configured with a timezone of US/Eastern or something similar. But logstash logs everything with UTC time, to prevent a wide variety of problems that occur when storing and processing local time.

ESX performance stats in Logstash

Here's an example input, you would just need rsyslog or a logging tool to send logs to the input port (1514 in this example) and then filter the data:

Input

input {
        tcp {
                type => "VMware"
                port => "1514"
        }
}

Filter

filter {
    if "VMware" in [tags] {
            multiline {
                    pattern => "-->"
                    what => "previous"
            }
            grok {
                    break_on_match => true
                    match => [
                            "message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?<message-body>(?<message_system_info>(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) \[%{DATA:message_service_info}]\ (?<syslog_message>(%{GREEDYDATA})))",
                            "message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: (?<message-body>(?<message_system_info>(?:\[%{DATA:message_thread_id} %{DATA:syslog_level} \'%{DATA:message_service}\'\ ?%{DATA:message_opID}])) (?<syslog_message>(%{GREEDYDATA})))",
                            "message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: %{GREEDYDATA:syslog_message}"
                    ]
            }
            syslog_pri { }
            date {
                    match => [ "syslog_timestamp", "YYYY-MM-ddHH:mm:ss,SSS" ]
                    timezone => "UTC"
            }
            mutate {
                    replace => [ "@source_host", "%{syslog_hostname}" ]
            }
            mutate {
                    replace => [ "@message", "%{syslog_message}" ]
            }
            if "Device naa" in [message] {
                    grok {
                            break_on_match => false
                            match => [
                                    "message", "Device naa.%{WORD:device_naa} performance has %{WORD:device_status}%{GREEDYDATA} of %{INT:datastore_latency_from}%{GREEDYDATA} to %{INT:datastore_latency_to}",
                                    "message", "Device naa.%{WORD:device_naa} performance has %{WORD:device_status}%{GREEDYDATA} from %{INT:datastore_latency_from}%{GREEDYDATA} to %{INT:datastore_latency_to}"
                            ]
                    }
            }
            if "connectivity issues" in [message] {
                    grok {
                            match => [
                                    "message", "Hostd: %{GREEDYDATA} : %{DATA:device_access} to volume %{DATA:device_id} %{DATA:datastore} (following|due to)"
                            ]
                    }
            }
            if "WARNING" in [message] {
                    grok {
                            match => [
                                    "message", "WARNING: %{GREEDYDATA:vmware_warning_msg}"
                            ]
                    }
            }
    }

}

Related Topic