Add support for create actions in the node client

lib/logstash/outputs/elasticsearch/protocol.rb

@@ -41,6 +41,29 @@ def bulk(actions)
         #])
       end
 
+      # Validate the action has all required args.
+      #
+      # If any action is missing a required arg, then a
+      # Logstash::ConfigurationError is raised.
+      #
+      # After validation, this will return the "real" action name (e.g.,
+      # "create_unless_exists" once validated is just "create"). If the
+      # incoming action is unrecognized or not validated, then it is returned
+      # as-is.
+      def validate_action(action, args)
+        action_name = action
+
+        # This is a specialization of "create" that requires the ID to be specified
+        if action == "create_unless_exists"
+          # create operations without an _id is pointless and almost certainly unintentional
+          raise(LogStash::ConfigurationError, "Specifying action => 'create_unless_exists' without a document '_id' is not supported.") if args[:_id].nil?
+
+          action_name = "create"
+        end
+
+        return action_name
+      end
+
       public(:initialize, :template_install)
     end
 
@@ -80,10 +103,12 @@ def build_client(options)
 
       def bulk(actions)
         @client.bulk(:body => actions.collect do |action, args, source|
+          action_name = validate_action(action, args)
+
           if source
-            next [ { action => args }, source ]
+            next [ { action_name => args }, source ]
           else
-            next { action => args }
+            next { action_name => args }
           end
         end.flatten)
       end # def bulk
@@ -194,7 +219,9 @@ def bulk(actions)
       end # def bulk
 
       def build_request(action, args, source)
-        case action
+        action_name = validate_action(action, args)
+
+        case action_name
           when "index"
             request = org.elasticsearch.action.index.IndexRequest.new(args[:_index])
             request.id(args[:_id]) if args[:_id]
@@ -203,7 +230,13 @@ def build_request(action, args, source)
             request = org.elasticsearch.action.delete.DeleteRequest.new(args[:_index])
             request.id(args[:_id])
           #when "update"
-          #when "create"
+          when "create"
+            request = org.elasticsearch.action.index.IndexRequest.new(args[:_index])
+            request.id(args[:_id]) if args[:_id]
+            request.source(source)
+            request.opType(org.elasticsearch.action.index.IndexRequest.OpType.CREATE)
+          else
+            raise(LogStash::ConfigurationError, "action => '#{action_name}' is not currently supported.")
         end # case action
 
         request.type(args[:_type]) if args[:_type]

spec/outputs/elasticsearch_spec.rb

@@ -12,6 +12,84 @@
     expect {output.register}.to_not raise_error
   end
 
+  it "create works with _id", :elasticsearch => true do
+    # Generate a random index name
+    index = 10.times.collect { rand(10).to_s }.join("")
+    type = 10.times.collect { rand(10).to_s }.join("")
+    id = 10.times.collect { rand(10).to_s }.join("")
+
+    config <<-CONFIG
+    input {
+      generator {
+        message => "hello world"
+        count => 1
+        type => "#{type}"
+        _id => "#{id}"
+      }
+    }
+    output {
+      elasticsearch {
+        host => "127.0.0.1"
+        index => "#{index}"
+        action => "create_unless_exists"
+        flush_size => 1
+      }
+    }
+    CONFIG
+
+    agent do
+      # No need for a refresh since we can just do a get
+      ftw = FTW::Agent.new
+
+      response = ftw.get!("http://127.0.0.1:9200/#{index}/#{type}/#{id}")
+      data = ""
+      response.read_body { |chunk| data << chunk }
+      result = LogStash::Json.load(data)
+
+      # With no 'index_type' set, the document type should be the type
+      # set on the input
+      insist { result["_type"] } == type
+      insist { result["_index"] } == index
+      insist { result["found"] }
+      insist { result["_source"]["message"] } == "hello world"
+    end
+  end
+
+  describe "create action fails without _id", :elasticsearch => true do
+    # Generate a random index name
+    index = 10.times.collect { rand(10).to_s }.join("")
+
+    ["node", "transport", "http"].each do |protocol|
+      context "with protocol => #{protocol}" do
+
+        config <<-CONFIG
+        input {
+          generator {
+            message => "hello world"
+            count => 1
+          }
+        }
+        output {
+          elasticsearch {
+            host => "127.0.0.1"
+            index => "#{index}"
+            action => "create_unless_exists"
+          }
+        }
+        CONFIG
+
+        # ^^^ SHOULD FAIL AND THE FAILURE SHOULD CAUSE THE TEST TO PASS
+
+        # TODO: pickypg -- I have no idea how to check for raise_error
+        agent do
+          it "should fail in bulk" do
+            expect {context}.to raise_error
+          end
+        end
+      end
+    end
+  end
+
   describe "ship lots of events w/ default index_type", :elasticsearch => true do
     # Generate a random index name
     index = 10.times.collect { rand(10).to_s }.join("")