do not assume event field values as mutable and code cleanups and proper timestamp handling

do not assume event field values as mutable and code cleanups and proper timestamp handling

missing encoding header

remove unnecessary inject_field

specs cleanups

Author: colinsurprenant

lib/logstash/filters/json.rb

@@ -46,54 +46,64 @@ class LogStash::Filters::Json < LogStash::Filters::Base
   # NOTE: if the `target` field already exists, it will be overwritten!
   config :target, :validate => :string
 
-  public
+  JSONPARSEFAILURE_TAG = "_jsonparsefailure"
+
   def register
     # Nothing to do here
-  end # def register
+  end
 
-  public
   def filter(event)
-    
-
-    @logger.debug("Running json filter", :event => event)
-
-    return unless event.include?(@source)
-
-    # TODO(colin) this field merging stuff below should be handled in Event.
+    @logger.debug? && @logger.debug("Running json filter", :event => event)
 
     source = event[@source]
+    return unless source
 
     begin
       parsed = LogStash::Json.load(source)
-      # If your parsed JSON is an array, we can't merge, so you must specify a
-      # destination to store the JSON, so you will get an exception about
-      if parsed.kind_of?(Array) && @target.nil?
-        raise('Parsed JSON arrays must have a destination in the configuration')
-      elsif @target.nil?
-        event.to_hash.merge! parsed
-      else
-        event[@target] = parsed
+    rescue => e
+      event.tag(JSONPARSEFAILURE_TAG)
+      @logger.warn("Error parsing json", :source => @source, :raw => source, :exception => e)
+      return
+    end
+
+    if @target
+      event[@target] = parsed
+    else
+      unless parsed.is_a?(Hash)
+        event.tag(JSONPARSEFAILURE_TAG)
+        @logger.warn("Parsed JSON object/hash requires a target configuration option", :source => @source, :raw => source)
+        return
       end
 
-      # If no target, we target the root of the event object. This can allow
-      # you to overwrite @timestamp and this will typically happen for json
-      # LogStash Event deserialized here.
-      if !@target && event.timestamp.is_a?(String)
-        event.timestamp = LogStash::Timestamp.parse_iso8601(event.timestamp)
+      # TODO: (colin) the timestamp initialization should be DRY'ed but exposing the similar code
+      # in the Event#init_timestamp method. See https://github.com/elastic/logstash/issues/4293
+
+      # a) since the parsed hash will be set in the event root, first extract any @timestamp field to properly initialized it
+      parsed_timestamp = parsed.delete(LogStash::Event::TIMESTAMP)
+      begin
+        timestamp = parsed_timestamp ? LogStash::Timestamp.coerce(parsed_timestamp) : nil
+      rescue LogStash::TimestampParserError => e
+        timestamp = nil
       end
 
-      filter_matched(event)
-    rescue => e
-      tag = "_jsonparsefailure"
-      event["tags"] ||= []
-      event["tags"] << tag unless event["tags"].include?(tag)
-      @logger.warn("Trouble parsing json", :source => @source,
-                   :raw => event[@source], :exception => e)
-      return
+      # b) then set all parsed fields in the event
+      parsed.each{|k, v| event[k] = v}
+
+      # c) finally re-inject proper @timestamp
+      if parsed_timestamp
+        if timestamp
+          event.timestamp = timestamp
+        else
+          event.timestamp = LogStash::Timestamp.new
+          @logger.warn("Unrecognized #{LogStash::Event::TIMESTAMP} value, setting current time to #{LogStash::Event::TIMESTAMP}, original in #{LogStash::Event::TIMESTAMP_FAILURE_FIELD} field", :value => parsed_timestamp.inspect)
+          event.tag(LogStash::Event::TIMESTAMP_FAILURE_TAG)
+          event[LogStash::Event::TIMESTAMP_FAILURE_FIELD] = parsed_timestamp.to_s
+        end
+      end
     end
 
-    @logger.debug("Event after json filter", :event => event)
-
-  end # def filter
+    filter_matched(event)
 
-end # class LogStash::Filters::Json
+    @logger.debug? && @logger.debug("Event after json filter", :event => event)
+  end
+end

spec/filters/json_spec.rb

@@ -1,3 +1,5 @@
+# encoding: utf-8
+
 require "logstash/devutils/rspec/spec_helper"
 require "logstash/filters/json"
 require "logstash/timestamp"
@@ -103,52 +105,72 @@
     end
   end
 
-  context "when json could not be parsed" do
+  context "using message field source" do
 
     subject(:filter) {  LogStash::Filters::Json.new(config)  }
 
-    let(:message)    { "random_message" }
-    let(:config)     { {"source" => "message"} }
-    let(:event)      { LogStash::Event.new("message" => message) }
+    let(:config) { {"source" => "message"} }
+    let(:event) { LogStash::Event.new("message" => message) }
 
     before(:each) do
       filter.register
       filter.filter(event)
     end
 
-    it "add the failure tag" do
-      expect(event).to include "tags"
-    end
+    context "when json could not be parsed" do
+      let(:message) { "random_message" }
 
-    it "uses an array to store the tags" do
-      expect(event['tags']).to be_a Array
-    end
+      it "add the failure tag" do
+        expect(event).to include("tags")
+      end
 
-    it "add a json parser failure tag" do
-      expect(event['tags']).to include "_jsonparsefailure"
-    end
+      it "uses an array to store the tags" do
+        expect(event['tags']).to be_a(Array)
+      end
+
+      it "add a json parser failure tag" do
+        expect(event['tags']).to include("_jsonparsefailure")
+      end
 
-    context "there are two different errors added" do
+      context "there are two different errors added" do
 
-      let(:event)  { LogStash::Event.new("message" => message, "tags" => ["_anotherkinfoffailure"] ) }
+        let(:event)  { LogStash::Event.new("message" => message, "tags" => ["_anotherkinfoffailure"] ) }
 
-      it "pile the different error messages" do
-        expect(event['tags']).to include "_jsonparsefailure"
-      end
+        it "pile the different error messages" do
+          expect(event['tags']).to include("_jsonparsefailure")
+        end
 
-      it "keep the former error messages on the list" do
-        expect(event['tags']).to include "_anotherkinfoffailure"
+        it "keep the former error messages on the list" do
+          expect(event['tags']).to include("_anotherkinfoffailure")
+        end
       end
     end
 
     context "the JSON is an ArrayList" do
-
-      let(:message)  { "[1, 2, 3]" }
+      let(:message) { "[1, 2, 3]" }
 
       it "adds the failure tag" do
-        expect(event['tags']).to include "_jsonparsefailure"
+        expect(event['tags']).to include("_jsonparsefailure")
+      end
+    end
+
+    context "json contains valid timestamp" do
+      let(:message) { "{\"foo\":\"bar\", \"@timestamp\":\"2015-12-02T17:40:00.666Z\"}" }
+
+      it "should set json timestamp" do
+        expect(event.timestamp).to be_a(LogStash::Timestamp)
+        expect(event.timestamp.to_s).to eq("2015-12-02T17:40:00.666Z")
       end
     end
 
+    context "json contains invalid timestamp" do
+      let(:message) { "{\"foo\":\"bar\", \"@timestamp\":\"foobar\"}" }
+
+      it "should set timestamp to current time" do
+        expect(event.timestamp).to be_a(LogStash::Timestamp)
+        expect(event["tags"]).to include(LogStash::Event::TIMESTAMP_FAILURE_TAG)
+        expect(event[LogStash::Event::TIMESTAMP_FAILURE_FIELD]).to eq("foobar")
+      end
+    end
   end
 end