Split filter: ability to split arrays within a JSON structure

lib/logstash/filters/split.rb

@@ -1,6 +1,6 @@
-# encoding: utf-8
 require "logstash/filters/base"
 require "logstash/namespace"
+require "logstash/event"
 
 # The split filter is for splitting multiline messages into separate events.
 #
@@ -22,6 +22,10 @@ class LogStash::Filters::Split < LogStash::Filters::Base
   # The field which value is split by the terminator
   config :field, :validate => :string, :default => "message"
 
+  # If true, the array data will be passed forward as a single hash element with @field as the key.
+  # If false, treat the entire array element as a new event for further processing.
+  config :reuse_element, :validate => :boolean, :default => true
+
   public
   def register
     # Nothing to do
@@ -31,27 +35,31 @@ def register
   def filter(event)
     return unless filter?(event)
 
-    events = []
+    splits = []
 
     original_value = event[@field]
 
-    # If for some reason the field is an array of values, take the first only.
-    original_value = original_value.first if original_value.is_a?(Array)
-
-    # Using -1 for 'limit' on String#split makes ruby not drop trailing empty
-    # splits.
-    splits = original_value.split(@terminator, -1)
+    if original_value.is_a?(Array)
+      splits = original_value
+    else
+      # Using -1 for 'limit' on String#split makes ruby not drop trailing empty
+      # splits.
+      splits = original_value.split(@terminator, -1)
+    end
 
-    # Skip filtering if splitting this event resulted in only one thing found.
-    return if splits.length == 1
-    #or splits[1].empty?
+    # Skip filtering if splitting this event resulted in only one thing found
+    return if splits.length <= 1
 
     splits.each do |value|
       next if value.empty?
 
-      event_split = event.clone
-      @logger.debug("Split event", :value => value, :field => @field)
-      event_split[@field] = value
+      if @reuse_element
+        event_split = event.clone
+        @logger.debug("Split event", :value => value, :field => @field)
+        event_split[@field] = value
+      else
+        event_split = LogStash::Event.new(value)
+      end
       filter_matched(event_split)
 
       # Push this new event onto the stack at the LogStash::FilterWorker