nested_objects implemented

lib/logstash/inputs/gelf.rb

@@ -51,6 +51,9 @@ class LogStash::Inputs::Gelf < LogStash::Inputs::Base
   #
   config :strip_leading_underscore, :validate => :boolean, :default => true
 
+  # Whether or not to process dots in fields or leave them in place
+  config :nested_objects, :validate => :boolean, :default => false
+
   RECONNECT_BACKOFF_SLEEP = 5
   TIMESTAMP_GELF_FIELD = "timestamp".freeze
   SOURCE_HOST_FIELD = "source_host".freeze
@@ -115,6 +118,7 @@ def udp_listener(output_queue)
 
       remap_gelf(event) if @remap
       strip_leading_underscore(event) if @strip_leading_underscore
+      nested_objects(event) if @nested_objects
       decorate(event)
 
       output_queue << event
@@ -196,4 +200,87 @@ def strip_leading_underscore(event)
        event.remove(key)
      end
   end # deef removing_leading_underscores
+
+  private
+  def nested_objects(event)
+    # process nested, create objects as needed, when key is 0, create an array. if object already exists and is an array push it.
+    base_target = event.to_hash
+    base_target.keys.each do |key|
+      next unless key.include? "."
+      value = event.get(key)
+      previous_key = nil
+      first_key = nil
+      target = base_target
+
+      keys = key.split(".")
+      if key =~ /\.$/
+        keys.push("");
+      end
+
+      keys.each do |subKey|
+        if previous_key.nil?
+          first_key = subKey
+        else#skip first subKey
+          if !container_has_element?(target, previous_key)
+            if subKey =~ /^\d+$/
+              target = set_container_element(target, previous_key, Array.new)
+            else
+              target = set_container_element(target, previous_key, Hash.new)
+            end
+          end
+          target = get_container_element(target, previous_key)
+        end
+        previous_key = subKey
+      end
+      target = set_container_element(target, previous_key, value)
+      event.remove(key)
+      event.set(first_key, base_target[first_key])
+    end
+  end
+
+  private
+  def get_container_element(container, key)
+    if container.is_a?(Array)
+      container[Integer(key)]
+    elsif container.is_a?(Hash)
+      container[key]
+    else #Event
+      raise "not an array or hash"
+    end
+  end
+
+  private
+  def set_container_element(container, key, value)
+    if container.is_a?(Array)
+      if !/\A\d+\z/.match(key)
+        #key is not an integer, so we need to convert array to hash
+        container = Hash[container.map.with_index { |x, i| [i, x] }]
+        container[key] = value
+        return container
+      else
+        container[Integer(key)] = value
+      end
+    elsif container.is_a?(Hash)
+      container[key] = value
+    else #Event
+      raise "not an array or hash"
+    end
+
+    return container
+  end
+
+  private
+  def container_has_element?(container, key)
+    if container.is_a?(Array)
+      if !/\A\d+\z/.match(key)
+        return false
+      else
+        !container[Integer(key)].nil?
+      end
+    elsif container.is_a?(Hash)
+      container.key?(key)
+    else
+      raise "not an array or hash"
+    end
+  end
 end # class LogStash::Inputs::Gelf

spec/inputs/gelf_spec.rb

@@ -70,6 +70,63 @@
     end
   end
 
+  it "reads nested gelf messages " do
+    port = 12210
+    host = "127.0.0.1"
+    chunksize = 1420
+    gelfclient = GELF::Notifier.new(host, port, chunksize)
+
+    conf = <<-CONFIG
+      input {
+        gelf {
+          port => "#{port}"
+          host => "#{host}"
+          nested_objects => true
+        }
+      }
+    CONFIG
+
+    result = input(conf) do |pipeline, queue|
+      # send a first message until plugin is up and receives it
+      while queue.size <= 0
+        gelfclient.notify!("short_message" => "prime")
+        sleep(0.1)
+      end
+      gelfclient.notify!("short_message" => "start")
+
+      e = queue.pop
+      while (e.get("message") != "start")
+        e = queue.pop
+      end
+
+      gelfclient.notify!({
+        "short_message" => "test nested",
+        "_toto.titi" => "objectValue",
+        "_foo.0" => "first",
+        "_foo.1" => "second",
+        "_ca.0.titi" => "1",
+        "_ca.1.titi" => "2",
+        "_empty." => "pouet",
+        "_not_an_array.0" => "bob",
+        "_not_an_array.1" => "alice",
+        "_not_an_array.length" => "carol",
+      })
+
+      queue.pop
+    end
+
+    insist { result.get("message") } == "test nested"
+    insist { result.get("toto")["titi"] } == "objectValue"
+    insist { result.get("foo") } == ["first", "second"]
+    insist { result.get("ca")[0]["titi"] } == "1"
+    insist { result.get("ca")[1]["titi"] } == "2"
+    insist { result.get("empty")[""]} == "pouet"
+    insist { result.get("not_an_array")["0"]} == "bob"
+    insist { result.get("not_an_array")["1"]} == "alice"
+    insist { result.get("not_an_array")["length"]} == "carol"
+    insist { result.get("host") } == Socket.gethostname
+  end
+
   context "timestamp coercion" do
     # these test private methods, this is advisable for now until we roll out this coercion in the Timestamp class
     # and remove this