in_forward: Make lazy evaluation for sent by Fluent Bit suspected payloads

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

Author: cosmo0920

lib/fluent/plugin/in_forward.rb

@@ -25,6 +25,43 @@ module Fluent::Plugin
   class ForwardInput < Input
     Fluent::Plugin.register_input('forward', self)
 
+    class MetadataTimeEventStream < Fluent::EventStream
+      def initialize(es)
+        @es = es
+      end
+
+      def empty?
+        @es.empty?
+      end
+
+      def dup
+        self.class.new(@es.dup)
+      end
+
+      def size
+        @es.size
+      end
+
+      def repeatable?
+        @es.repeatable?
+      end
+
+      def slice(index, num)
+        self.class.new(@es.slice(index, num))
+      end
+
+      def each(unpacker: nil, &block)
+        @es.each(unpacker: unpacker) do |time, record|
+          next if record.nil?
+
+          time = time[0] if time.is_a?(Array)
+          time = Fluent::EventTime.now if time.nil? || time.to_i == 0
+          block.call(time, record)
+        end
+        nil
+      end
+    end
+
     # See the wiki page below for protocol specification
     # https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1
 
@@ -317,7 +354,13 @@ def on_message(msg, chunk_size, conn)
         else
           es = Fluent::MessagePackEventStream.new(entries, nil, size.to_i)
         end
-        es = normalize_event_stream(tag, es, conn.remote_host)
+        es = if @skip_invalid_event
+               normalize_event_stream(tag, es, conn.remote_host)
+             elsif option['fluent_signal'] == 0
+               MetadataTimeEventStream.new(es)
+             else
+               es
+             end
         if @enable_field_injection
           es = add_source_info(es, conn)
         end
@@ -361,11 +404,8 @@ def invalid_event?(tag, time, record)
 
     def normalize_event_stream(tag, es, remote_host)
       new_es = Fluent::MultiEventStream.new
-      normalized = @skip_invalid_event || es.is_a?(Array)
       es.each { |time, record|
-        extracted_time = extract_event_time(time)
-        normalized ||= !time.equal?(extracted_time)
-        time = extracted_time
+        time = extract_event_time(time)
         if @skip_invalid_event && invalid_event?(tag, time, record)
           log.warn "skip invalid event:", host: remote_host, tag: tag, time: time, record: record
           next
@@ -374,7 +414,7 @@ def normalize_event_stream(tag, es, remote_host)
         time = Fluent::EventTime.now if time.nil? || time.to_i == 0 # `to_i == 0` for empty EventTime
         new_es.add(time, record)
       }
-      normalized ? new_es : es
+      new_es
     end
 
     def extract_event_time(time)