Only add _type if ES version < 8

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 10.2.2
+  - Fixed 8.x type removal compatibility issue [#892](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/892) 
+
 ## 10.2.1
   - Fixed wording and corrected option in documentation [#881](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/881) [#883](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/883)
 

docs/index.asciidoc

@@ -298,13 +298,18 @@ Elasticsearch with the same ID.
   * There is no default value for this setting.
   * This option is deprecated
 
-Note: This option is deprecated due to the https://www.elastic.co/guide/en/elasticsearch/reference/6.0/removal-of-types.html[removal of types in Elasticsearch 6.0].
+NOTE: This option is deprecated due to the https://www.elastic.co/guide/en/elasticsearch/reference/6.0/removal-of-types.html[removal of types in Elasticsearch 6.0].
 It will be removed in the next major version of Logstash.
+
+NOTE: This value is ignored and has no effect for Elasticsearch clusters `8.x`.
+
 This sets the document type to write events to. Generally you should try to write only
 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,11 @@ 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
       }
 
+      params[:_type] = get_event_type(event) if client.maximum_seen_major_version < 8
+
       if @pipeline
         params[:pipeline] = event.sprintf(@pipeline)
       end
@@ -276,8 +277,10 @@ def get_event_type(event)
                  event.get("type") || DEFAULT_EVENT_TYPE_ES6
                elsif client.maximum_seen_major_version == 6
                  DEFAULT_EVENT_TYPE_ES6
-               else
+               elsif client.maximum_seen_major_version == 7
                  DEFAULT_EVENT_TYPE_ES7
+               else
+                 nil
                end
              end
 

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '10.2.1'
+  s.version         = '10.2.2'
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

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
@@ -82,7 +93,6 @@ def self.es_version
     end
   end
 
-
   def self.es_version_satisfies?(*requirement)
     es_version = RSpec.configuration.filter[:es_version] || ENV['ES_VERSION'] || ENV['ELASTIC_STACK_VERSION']
     if es_version.nil?

spec/integration/outputs/compressed_indexing_spec.rb

@@ -50,9 +50,8 @@
         response = http_client.get("#{index_url}/_search?q=*&size=1000")
         result = LogStash::Json.load(response.body)
         result["hits"]["hits"].each do |doc|
-          if ESHelper.es_version_satisfies?(">= 6")
-            expect(doc["_type"]).to eq(type)
-          end
+          expect(doc["_type"]).to eq(type) if ESHelper.es_version_satisfies?(">= 6", "< 8")
+          expect(doc).not_to include("_type") if ESHelper.es_version_satisfies?(">= 8")
           expect(doc["_index"]).to eq(index)
         end
       end

spec/integration/outputs/index_spec.rb

@@ -80,9 +80,8 @@
       response = http_client.get("#{index_url}/_search?q=*&size=1000")
       result = LogStash::Json.load(response.body)
       result["hits"]["hits"].each do |doc|
-        if ESHelper.es_version_satisfies?(">= 6")
-          expect(doc["_type"]).to eq(type)
-        end
+        expect(doc["_type"]).to eq(type) if ESHelper.es_version_satisfies?(">= 6", "< 8")
+        expect(doc).not_to include("_type") if ESHelper.es_version_satisfies?(">= 8")
         expect(doc["_index"]).to eq(index)
       end
     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,44 @@
       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
+
+        context "with 'document type set'" do
+          let(:options) { super.merge("document_type" => "bar")}
+          it "should get the event type from the 'document_type' setting" do
+            action_tuple = subject.send(:event_action_tuple, LogStash::Event.new("type" => "foo"))
+            action_params = action_tuple[1]
+            expect(action_params).to include(:_type => "bar")
+          end
+        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
+
+        context "with 'document type set'" do
+          let(:options) { super.merge("document_type" => "bar")}
+          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
+    end
+
     describe "with auth" do
       let(:user) { "myuser" }
       let(:password) { ::LogStash::Util::Password.new("mypassword") }