Only add _type if ES version < 8

docs/index.asciidoc

@@ -304,7 +304,9 @@ This sets the document type to write events to. Generally you should try to writ
 similar events to the same 'type'. String expansion `%{foo}` works here.
 If you don't set a value for this option:
 
-- for elasticsearch clusters 6.x and above: the value of 'doc' will be used;
+- for elasticsearch clusters 8.x: no value will be used;
+- for elasticsearch clusters 7.x: the value of '_doc' will be used;
+- for elasticsearch clusters 6.x: the value of 'doc' will be used;
 - for elasticsearch clusters 5.x and below: the event's 'type' field will be used, if the field is not present the value of 'doc' will be used.
 
 [id="plugins-{type}s-{plugin}-failure_type_logging_whitelist"]

lib/logstash/outputs/elasticsearch/common.rb

@@ -70,10 +70,13 @@ def event_action_tuple(event)
       params = {
         :_id => @document_id ? event.sprintf(@document_id) : nil,
         :_index => event.sprintf(@index),
-        :_type => get_event_type(event),
         routing_field_name => @routing ? event.sprintf(@routing) : nil
       }
 
+      if client.maximum_seen_major_version < 8
+        params[:_type] = get_event_type(event)
+      end
+
       if @pipeline
         params[:pipeline] = event.sprintf(@pipeline)
       end

spec/es_spec_helper.rb

@@ -27,13 +27,24 @@ def get_client
   end
 
   def doc_type
-    if ESHelper.es_version_satisfies?(">=7")
+    if ESHelper.es_version_satisfies?(">=8")
+      nil
+    elsif ESHelper.es_version_satisfies?(">=7")
       "_doc"
     else
       "doc"
     end
   end
 
+  def self.action_for_version(action)
+    action_params = action[1]
+    if ESHelper.es_version_satisfies?(">=8")
+      action_params.delete(:_type)
+    end
+    action[1] = action_params
+    action
+  end
+
   def todays_date
     Time.now.strftime("%Y.%m.%d")
   end

spec/integration/outputs/retry_spec.rb

@@ -4,9 +4,9 @@
 describe "failures in bulk class expected behavior", :integration => true do
   let(:template) { '{"template" : "not important, will be updated by :index"}' }
   let(:event1) { LogStash::Event.new("somevalue" => 100, "@timestamp" => "2014-11-17T20:37:17.223Z", "@metadata" => {"retry_count" => 0}) }
-  let(:action1) { ["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event1] }
+  let(:action1) { ESHelper.action_for_version(["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event1]) }
   let(:event2) { LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0] }, "@timestamp" => "2014-11-17T20:37:17.223Z", "@metadata" => {"retry_count" => 0}) }
-  let(:action2) { ["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event2] }
+  let(:action2) { ESHelper.action_for_version(["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event2]) }
   let(:invalid_event) { LogStash::Event.new("geoip" => { "location" => "notlatlon" }, "@timestamp" => "2014-11-17T20:37:17.223Z") }
 
   def mock_actions_with_response(*resp)

spec/unit/outputs/elasticsearch_spec.rb

@@ -76,6 +76,26 @@
       end
     end
 
+    describe "building an event action tuple" do
+      context "for 7.x elasticsearch clusters" do
+        let(:maximum_seen_major_version) { 7 }
+        it "should include '_type'" do
+          action_tuple = subject.send(:event_action_tuple, LogStash::Event.new("type" => "foo"))
+          action_params = action_tuple[1]
+          expect(action_params).to include(:_type => "_doc")
+        end
+      end
+
+      context "for 8.x elasticsearch clusters" do
+        let(:maximum_seen_major_version) { 8 }
+        it "should not include '_type'" do
+          action_tuple = subject.send(:event_action_tuple, LogStash::Event.new("type" => "foo"))
+          action_params = action_tuple[1]
+          expect(action_params).not_to include(:_type)
+        end
+      end
+    end
+
     describe "with auth" do
       let(:user) { "myuser" }
       let(:password) { ::LogStash::Util::Password.new("mypassword") }