in_forward: Improve interoperability with metadata encapsulated timestamps

Signed-off-by: Hiroshi Hatake <hiroshi@chronosphere.io>

Author: cosmo0920

lib/fluent/plugin/in_forward.rb

@@ -332,7 +332,7 @@ def on_message(msg, chunk_size, conn)
                entries.each { |e|
                  record = e[1]
                  next if record.nil?
-                 time = e[0]
+                 time = extract_event_time(e[0])
                  time = Fluent::EventTime.now if time.nil? || time.to_i == 0 # `to_i == 0` for empty EventTime
                  es.add(time, record)
                }
@@ -367,12 +367,14 @@ def on_message(msg, chunk_size, conn)
     end
 
     def invalid_event?(tag, time, record)
+      time = extract_event_time(time)
       !((time.is_a?(Integer) || time.is_a?(::Fluent::EventTime)) && record.is_a?(Hash) && tag.is_a?(String))
     end
 
     def check_and_skip_invalid_event(tag, es, remote_host)
       new_es = Fluent::MultiEventStream.new
       es.each { |time, record|
+        time = extract_event_time(time)
         if invalid_event?(tag, time, record)
           log.warn "skip invalid event:", host: remote_host, tag: tag, time: time, record: record
           next
@@ -382,6 +384,10 @@ def check_and_skip_invalid_event(tag, es, remote_host)
       new_es
     end
 
+    def extract_event_time(time)
+      time.is_a?(Array) ? time[0] : time
+    end
+
     def add_source_info(es, conn)
       new_es = Fluent::MultiEventStream.new
       if @source_address_key && @source_hostname_key