Add version support

lib/logstash/outputs/elasticsearch/common.rb

@@ -159,6 +159,14 @@ def event_action_params(event)
         params[:_retry_on_conflict] = @retry_on_conflict
       end
 
+      if @version
+        params[:_version] = event.sprintf(@version)
+      end
+
+      if @version_type
+        params[:_version_type] = event.sprintf(@version_type)
+      end
+
       params
     end
 

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -59,6 +59,15 @@ def self.included(mod)
       # Elasticsearch with the same ID.
       mod.config :document_id, :validate => :string
 
+      # The document's version to use. Overrides Elasticsearch's internal version scheme.
+      mod.config :version, :validate => :string
+
+      # Allows for using different versioning system by using your own.
+      # Note: 'external_gte' and are for special cases. They may result in loss of data
+      # if used incorrectly.
+      # See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-index_.html#_version_types[here] for more on using version_type.
+      mod.config :version_type, :validate => ["internal", 'external', "external_gt", "external_gte"]
+
       # A routing override to be applied to all processed events.
       # This can be dynamic using the `%{foo}` syntax.
       mod.config :routing, :validate => :string

spec/unit/outputs/elasticsearch_spec.rb

@@ -281,6 +281,33 @@
     end
   end
 
+  describe "the version options" do
+    let(:event) { LogStash::Event.new("message" => "blah") }
+    subject(:eso) { LogStash::Outputs::ElasticSearch.new(options) }
+
+    context "when providing a version" do
+      let(:options) { {"action" => "update", "version" => "5", "version_type" => "external"} }
+
+      it "should add the given version" do
+        action, params, event_data = eso.event_action_tuple(event)
+        expect(params).to include({:_version => "5"})
+      end
+
+      it "should add the given version_type" do
+        action, params, event_data = eso.event_action_tuple(event)
+        expect(params).to include({:_version_type => "external"})
+      end
+    end
+
+    context "with an invalid version_type" do
+      let(:options) { {"action" => "index", "version" => "6", "version_type" => "myversiontype"} }
+
+      it "should raise a configuration error" do
+        expect { subject.register }.to raise_error(LogStash::ConfigurationError)
+      end
+    end
+  end
+
   describe "stale connection check" do
     let(:validate_after_inactivity) { 123 }
     subject(:eso) { LogStash::Outputs::ElasticSearch.new("validate_after_inactivity" => validate_after_inactivity) }