PERFORMANCE: filter_func can optionally take an array of events to make batched filters much faster

logstash-core/lib/logstash/config/config_ast.rb

@@ -77,7 +77,7 @@ def compile
         # of the output/filter function
         definitions << "define_singleton_method :#{type}_func do |event|"
         definitions << "  targeted_outputs = []" if type == "output"
-        definitions << "  events = [event]" if type == "filter"
+        definitions << "  events = event" if type == "filter"
         definitions << "  @logger.debug? && @logger.debug(\"#{type} received\", \"event\" => event.to_hash)"
 
         sections.select { |s| s.plugin_type.text_value == type }.each do |s|

logstash-core/lib/logstash/pipeline.rb

@@ -498,13 +498,9 @@ def worker_loop(batch_size, batch_delay)
   end
 
   def filter_batch(batch)
-    batch.each do |event|
-      return if @force_shutdown.true?
-
-      filter_func(event).each do |e|
-        #these are both original and generated events
-        batch.merge(e) unless e.cancelled?
-      end
+    filter_func(batch.to_a).each do |e|
+      #these are both original and generated events
+      batch.merge(e) unless e.cancelled?
     end
     @filter_queue_client.add_filtered_metrics(batch)
     @events_filtered.increment(batch.size)
@@ -662,7 +658,7 @@ def shutdown_workers
   def filter(event, &block)
     maybe_setup_out_plugins
     # filter_func returns all filtered events, including cancelled ones
-    filter_func(event).each {|e| block.call(e)}
+    filter_func([event]).each {|e| block.call(e)}
   end
 
   # perform filters flush and yield flushed event to the passed block

logstash-core/lib/logstash/util/wrapped_acked_queue.rb

@@ -287,6 +287,12 @@ def cancel(event)
         # @cancelled[event] = true
       end
 
+      def to_a
+        events = []
+        each {|e| events << e}
+        events
+      end
+
       def each(&blk)
         # take care not to cause @originals or @generated to change during iteration
 

logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb

@@ -218,6 +218,12 @@ def cancel(event)
         # @cancelled[event] = true
       end
 
+      def to_a
+        events = []
+        each {|e| events << e}
+        events
+      end
+
       def each(&blk)
         # take care not to cause @originals or @generated to change during iteration
         @iterating = true

logstash-core/spec/logstash/pipeline_spec.rb

@@ -618,9 +618,9 @@ class TestPipeline < LogStash::Pipeline
 
     it "should handle evaluating different config" do
       expect(pipeline1.output_func(LogStash::Event.new)).not_to include(nil)
-      expect(pipeline1.filter_func(LogStash::Event.new)).not_to include(nil)
+      expect(pipeline1.filter_func([LogStash::Event.new])).not_to include(nil)
       expect(pipeline2.output_func(LogStash::Event.new)).not_to include(nil)
-      expect(pipeline1.filter_func(LogStash::Event.new)).not_to include(nil)
+      expect(pipeline1.filter_func([LogStash::Event.new])).not_to include(nil)
     end
   end
 
@@ -700,9 +700,9 @@ class TestPipeline < LogStash::Pipeline
       # in the current instance and was returning an array containing nil values for
       # the match.
       expect(pipeline1.output_func(LogStash::Event.new)).not_to include(nil)
-      expect(pipeline1.filter_func(LogStash::Event.new)).not_to include(nil)
+      expect(pipeline1.filter_func([LogStash::Event.new])).not_to include(nil)
       expect(pipeline2.output_func(LogStash::Event.new)).not_to include(nil)
-      expect(pipeline1.filter_func(LogStash::Event.new)).not_to include(nil)
+      expect(pipeline1.filter_func([LogStash::Event.new])).not_to include(nil)
     end
   end