Added ability to specify timestamp in message

[consulthys]

This is a quick fix for #124 in order to be able to add a specific timestamp to messages sunk to Kafka topics.

[talevy]

since this is a new feature and not just a bug fix. I'd be happy to bump this to 6.2.0

[consulthys]

No prob, makes sense, I will do.

[charsyam]

+1

[thinker0]

+1

[henridf]

Looks good to me, any chance this can be reviewed by the core team? Sorely needing it (and Kafka support is surprisingly weak across the board in log shipping tools).

[jinleileiking]

+1

[jordansissel]

The code looks OK, but this needs to be rebased.

Also needs docs added in docs/index.asciidoc

[jordansissel]

@consulthys I think @talevy's original comment applies here. This should be a minor version bump (not patch).

Regarding the docs, having the docs in docs/index.asciidoc is somewhat new for plugins, so let me know if you run into troubles with it.

[consulthys]

@jordansissel I've resolved the conflicts, amended the docs and bumped the minor version. Let me know if you want me to do anything else.

[jordansissel]

LogStash::Timestamp#to_i returns 'seconds since epoch':

[6] pry(#<LogStash::Runner>)> LogStash::Event.new.timestamp.to_i
=> 1505207654

But the Kafka RFC indicates in a few places that the unit for the Long timestamp is milliseconds, not seconds.

I think you need (ts.to_f * 1000).to_i to convert to milliseconds.

[consulthys]

Note that I'm not calling to_i on a Logstash::Timestamp, but on the wrapped Joda DateTime object and I was under the impression that back then it returned milliseconds, but agreed, not anymore. I'll change that.

[jordansissel]

Yeah, I checked timestamp.time and in Logstash 5.x, at least, it is a ruby Time object:

[6] pry(#<LogStash::Runner>)> date = LogStash::Filters::Date.new("match" => [ "message", "ISO8601" ])
=> <LogStash::Filters::Date match=>["message", "ISO8601"], id=>"date_e40418e5-e0ba-4b0f-a3a5-a04b62920d0e", enable_metric=>true, periodic_flush=>false, target=>"@timestamp", tag_on_failure=>["_dateparsefailure"]>
[7] pry(#<LogStash::Runner>)> e = LogStash::Event.new
=> 2017-09-13T04:11:30.230Z %{host} %{message}
[8] pry(#<LogStash::Runner>)> e.timestamp.time.to_i
=> 1505275890
[9] pry(#<LogStash::Runner>)> e.timestamp.time.class
=> Time

# Now try applying the date filter:
[11] pry(#<LogStash::Runner>)> e.set("message", "2017-01-01T00:00:00.000Z")
=> "2017-01-01T00:00:00.000Z"
[12] pry(#<LogStash::Runner>)> date.filter(e)
=> 2017-01-01T00:00:00.000Z %{host} 2017-01-01T00:00:00.000Z
[13] pry(#<LogStash::Runner>)> e.timestamp.time.to_i
=> 1483228800
[14] pry(#<LogStash::Runner>)> e.timestamp.time.class
=> Time

^^^ Still a Time object.

Internally, a LogStash::Event is a wrapper of the internal org.logstash.Event, and both have a Timestamp class. In Jruby, this timestamp type is provided by JrubyTimestampExtLibrary.java which has a ruby time method that returns a ruby Time object.

So, from Ruby, due to our Jruby wrappers, you don't get to see the org.logstash.Timestamp object (which is backed by Joda DateTime type), you see a Ruby Time type.

[consulthys]

True, I made the same exploration and came to the same conclusion.

Still, I'm a bit surprised that one has to do such a transformation in order to get a millisecond value out of a LogStash::Timestamp. Even more surprising is the fact that nowhere else in the whole Logstash codebase something like this is needed. And if my previous statement is wrong (which I hope), then why isn't there a utility method somewhere that gets me that value?

[jordansissel]

I don't know anywhere else we need to output milliseconds-since-epoch. I checked a few places I know we use time that isn't ISO8601 formatted:

• Nagios output uses seconds-since-epoch (not milliseconds)
• Graphite output uses seconds-since-epoch
• Influxdb output supports custom time precision (microseconds up to hours)

[consulthys]

Ok, thanks for checking. I was under the impression that getting a milliseconds representation out of a date/timestamp was a common feature in most mainstream languages.

I actually found one place in the Logstash core lib that does the same transformation and could benefit from having that factored out into the TimeStamp class: https://github.com/elastic/logstash/blob/master/logstash-core/lib/logstash/agent.rb#L182

If not feasible, I'll just apply your suggestion above so we can let it be.

[jordansissel]

what is the goal here? I can see this very easily causing mistakes for users. What would an expected sprintf be, here?

• "%{+%s}%{+SSS}" (same result as if you call time.to_f * 1000 basically)
• Something else?

[consulthys]

The idea of using sprintf here was described in my original issue (#124), but quick, it was helpful in cases where a field containing a value in seconds could also be used as the message timestamp.

[jordansissel]

I must have missed this comment; sorry for the delays.

If the use case is taking a numeric value as the kafka message timestamp, then I would prefer this be event.get(@message_timestamp) and not event.sprintf.

[jinleileiking]

waiting for merged

[jordansissel]

@consulthys I left some inline comments in review with the main question being if the timestamp used was correct. Based on my reading of the kafka docs, the timestamp is milliseconds and this PR uses seconds (Time#to_i gives seconds-since-epoch). It's unclear to me from the Kafka docs what origin the ProducerMessage timestamp has (unix epoch? process start time? system boot time? something else?)

After some digging, I think I finally found the answer. In KAFKA-3025 it says this:

Timestamp is the number of milliseconds since Unix Epoch

[jinleileiking]

No this patch. No elk for ksql

[consulthys]

Thanks @jordansissel, it's been months since I developed that so I'll have another look in details shortly :-)

[jordansissel]

Confession: I found the Time(Ruby) vs DateTime(Joda) to be confusing. I think we kept Time (Ruby) for backwards compatibility having event.timestamp have a predictable and backwards-compatible interface as we were building the new Java internals.

[jinleileiking]

@jordansissel can it be merged ?

[jinleileiking]

Not merged?

[jinleileiking]

@jordansissel

[jordansissel]

@jinleileiking Looking at the recent comments, there is still review going on in this issue. If you wish invest in speeding this up, help with testing, reviewing, or with code is always appreciated.

[consulthys]

@jordansissel I commented your change request but didn't get any feedback. Once you do, I'll make sure to wrap this up. Thanks much!

[jinleileiking]

I want to test it in my env, seems not work with logstash 2.4.0

[guyboertje]

@consulthys
It looks like another rebase is needed.

@jordansissel
Is Valentin's replies to your questions satisfactory?

[neerav-salaria-guavus]

Any update on this change to be merged @consulthys @jordansissel .

[daggerjames]

still waiting .