Added support for "update" action

Including doc and doc_as_upsert support, scripted updated and upsert.

Author: staaam

lib/logstash/codecs/es_bulk.rb

@@ -11,6 +11,12 @@
 class LogStash::Codecs::ESBulk < LogStash::Codecs::Base
   config_name "es_bulk"
 
+  # Parse update action source line (doc, doc_as_upsert)
+  config :parse_update, :validate => :boolean, :default => false
+
+  # Parse scripted_upsert update action source line (script, params, upsert). Relevant only when parse_update=true
+  config :parse_scripted_upsert, :validate => :boolean, :default => true
+
   public
   def initialize(params={})
     super(params)
@@ -27,7 +33,35 @@ def decode(data)
         line = LogStash::Json.load(bulk.get("message"))
         case state
         when :metadata
-          event = LogStash::Event.new(line)
+          if metadata["action"] == 'update' and @parse_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") and @parse_scripted_upsert
+              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
+            else
+              # fallback to default
+              event = LogStash::Event.new(line)
+            end
+          else
+            event = LogStash::Event.new(line)
+          end
           event.set("@metadata", metadata)
           yield event
           state = :initial

spec/codecs/es_bulk_spec.rb

@@ -3,11 +3,10 @@
 require "insist"
 
 describe LogStash::Codecs::ESBulk do
-  subject do
-    next LogStash::Codecs::ESBulk.new
-  end
-
   context "#decode" do
+    let(:config) { {} }
+    let(:subject) { LogStash::Codecs::ESBulk.new(config) }
+
     it "should return 4 events from json data" do
       data = <<-HERE
       { "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } }
@@ -42,16 +41,78 @@
       end
       insist { count } == 4
     end
-  end
 
-  context "fail to process non-bulk event then continue" do
-    it "continues after a fail" do
+    it "fail to process non-bulk event then continue" do
       decoded = false
       subject.decode("something that isn't a bulk event\n") do |event|
         decoded = true
       end
       insist { decoded } == false
     end
+
+    it "when parse_update=true, should return 7 events from json data" do
+      config.update("parse_update" => true)
+
+      data = <<-HERE
+      { "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } }
+      { "field1" : "value1" }
+      { "delete" : { "_index" : "test", "_type" : "type1", "_id" : "2" } }
+      { "create" : { "_index" : "test", "_type" : "type1", "_id" : "3" } }
+      { "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
+      subject.decode(data) do |event|
+        case count
+        when 0
+          insist { event.get("[@metadata][_id]") } == "1"
+          insist { event.get("[@metadata][action]") } == "index"
+          insist { event.get("field1") } == "value1"
+        when 1
+          insist { event.get("[@metadata][_id]") } == "2"
+          insist { event.get("[@metadata][action]") } == "delete"
+        when 2
+          insist { event.get("[@metadata][_id]") } == "3"
+          insist { event.get("[@metadata][action]") } == "create"
+          insist { event.get("field1") } == "value3"
+        when 3
+          insist { event.get("[@metadata][_id]") } == "1"
+          insist { event.get("[@metadata][action]") } == "update"
+          insist { event.get("[field2]") } == "value2"
+        when 4
+          insist { event.get("[@metadata][doc_as_upsert]") } == true
+          insist { event.get("[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("[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("[field1]") } == "value6"
+        end
+        count += 1
+      end
+      insist { count } == 7
+    end
+
   end
 
 end