Added support for "update" action

Including doc and doc_as_upsert support, scripted updated and upsert.

Author: Michael Gelfand

lib/logstash/codecs/es_bulk.rb

@@ -27,7 +27,32 @@ def decode(data)
         line = LogStash::Json.load(bulk.get("message"))
         case state
         when :metadata
-          event = LogStash::Event.new(line)
+          if metadata["action"] == 'update'
+            if line.has_key?("doc")
+              event = LogStash::Event.new(line["doc"])
+              if line.has_key?("doc_as_upsert")
+                metadata["doc_as_upsert"] = line["doc_as_upsert"]
+              end
+            elsif line.has_key?("params")
+              event = LogStash::Event.new(line["params"])
+              metadata["scripted_upsert"] = true
+              if line.has_key?("script")
+                metadata["script"] = line["script"]
+                metadata["script_type"] = "inline"
+              elsif line.has_key?("script_id")
+                metadata["script"] = line["script_id"]
+                metadata["script_type"] = "indexed"
+              end
+              if line.has_key?("lang")
+                metadata["script_lang"] = line["lang"]
+              end
+              if line.has_key?("upsert")
+                metadata["upsert"] = LogStash::Json.dump(line["upsert"])
+              end
+            end
+          else
+            event = LogStash::Event.new(line)
+          end
           event.set("@metadata", metadata)
           yield event
           state = :initial

spec/codecs/es_bulk_spec.rb

@@ -8,7 +8,7 @@
   end
 
   context "#decode" do
-    it "should return 4 events from json data" do
+    it "should return 7 events from json data" do
       data = <<-HERE
       { "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } }
       { "field1" : "value1" }
@@ -17,6 +17,12 @@
       { "field1" : "value3" }
       { "update" : {"_id" : "1", "_type" : "type1", "_index" : "index1"} }
       { "doc" : {"field2" : "value2"} }
+      { "update" : {"_id" : "1", "_type" : "type1", "_index" : "index1"} }
+      { "doc" : {"field2" : "value2"}, "doc_as_upsert": true }
+      { "update" : {"_id" : "5", "_type" : "type1", "_index" : "index1"} }
+      { "params" : {"field1" : "value5"}, "script": "some script", "upsert": {} }
+      { "update" : {"_id" : "6", "_type" : "type1", "_index" : "index1"} }
+      { "params" : {"field1" : "value6"}, "script_id": "some_script_id", "lang": "js", "upsert": { "field2": "value7"} }
       HERE
 
       count = 0
@@ -37,10 +43,30 @@
           insist { event.get("[@metadata][_id]") } == "1"
           insist { event.get("[@metadata][action]") } == "update"
           insist { event.get("[doc][field2]") } == "value2"
+        when 4
+          insist { event.get("[@metadata][doc_as_upsert]") } == true
+          insist { event.get("[doc][field2]") } == "value2"
+        when 5
+          insist { event.get("[@metadata][_id]") } == "5"
+          insist { event.get("[@metadata][action]") } == "update"
+          insist { event.get("[@metadata][script]") } == "some script"
+          insist { event.get("[@metadata][script_type]") } == "inline"
+          insist { event.get("[@metadata][scripted_upsert]") } == true
+          insist { event.get("[@metadata][upsert]") } == "{}"
+          insist { event.get("[doc][field1]") } == "value5"
+        when 6
+          insist { event.get("[@metadata][_id]") } == "6"
+          insist { event.get("[@metadata][action]") } == "update"
+          insist { event.get("[@metadata][script]") } == "some_script_id"
+          insist { event.get("[@metadata][script_type]") } == "indexed"
+          insist { event.get("[@metadata][script_lang]") } == "js"
+          insist { event.get("[@metadata][scripted_upsert]") } == true
+          insist { event.get("[@metadata][upsert]") } == "{\"field2\":\"value7\"}"
+          insist { event.get("[doc][field1]") } == "value6"
         end
         count += 1
       end
-      insist { count } == 4
+      insist { count } == 7
     end
   end