Only use the type field for the event for ES versions < 6.0 (logstash-plugins#712)

Only use the type field for the event for ES versions < 6.0.

Since 6.0 only supports a single type this behavior can be problematic as
multiple types will break ES.

Fixes logstash-plugins#708

* fix tests to support new maximum_node_version feature and type computation
* fix integration tests for new type decider

Author: jsvd

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 9.0.2
+  - Ignore event's type field for the purpose of setting document `_type` if cluster is es 6.x or above
+
 ## 9.0.1
   - Update gemspec summary
 

docs/index.asciidoc

@@ -242,8 +242,10 @@ Note: This option is deprecated due to the https://www.elastic.co/guide/en/elast
 It will be removed in the next major version of Logstash.
 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.
-Unless you set 'document_type', the event 'type' will be used if it exists
-otherwise the document type will be assigned the value of 'doc'.
+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 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"]
 ===== `failure_type_logging_whitelist` 

lib/logstash/outputs/elasticsearch/common.rb

@@ -22,6 +22,7 @@ def register
 
       setup_hosts # properly sets @hosts
       build_client
+
       install_template
       check_action_validity
 
@@ -48,6 +49,10 @@ def setup_hosts
       end
     end
 
+    def maximum_seen_major_version
+      client.maximum_seen_major_version
+    end
+
     def install_template
       TemplateManager.install_template(self)
     end
@@ -199,12 +204,17 @@ def event_action_params(event)
     end
 
     # Determine the correct value for the 'type' field for the given event
+    DEFAULT_EVENT_TYPE="doc".freeze
     def get_event_type(event)
       # Set the 'type' value for the index.
       type = if @document_type
                event.sprintf(@document_type)
              else
-               event.get("type") || "doc"
+               if client.maximum_seen_major_version < 6
+                 event.get("type") || DEFAULT_EVENT_TYPE
+               else
+                 DEFAULT_EVENT_TYPE
+               end
              end
 
       if !(type.is_a?(String) || type.is_a?(Numeric))

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -82,8 +82,8 @@ def template_install(name, template, force=false)
       template_put(name, template)
     end
 
-    def connected_es_versions
-      @pool.connected_es_versions
+    def maximum_seen_major_version
+      @pool.maximum_seen_major_version
     end
 
     def bulk(actions)

lib/logstash/outputs/elasticsearch/http_client/pool.rb

@@ -107,9 +107,9 @@ def url_info
       @state_mutex.synchronize { @url_info }
     end
 
-    def connected_es_versions
+    def maximum_seen_major_version
       @state_mutex.synchronize do
-        @url_info.values.select {|v| v[:state] == :alive }.map {|v| v[:version] }
+        @maximum_seen_major_version
       end
     end
 
@@ -245,6 +245,14 @@ def healthcheck!
           es_version = get_es_version(url)
           @state_mutex.synchronize do
             meta[:version] = es_version
+            major = major_version(es_version)
+            if !@maximum_seen_major_version
+              @logger.info("ES Output version determined", :es_version => @maximum_seen_major_version)
+              set_new_major_version(major)
+            elsif major > @maximum_seen_major_version
+              @logger.warn("Detected a node with a higher major version than previously observed. This could be the result of an elasticsearch cluster upgrade.", :previous_major => @maximum_seen_major_version, :new_major => major, :node_url => url)
+              set_new_major_version(major)
+            end
             meta[:state] = :alive
           end
         rescue HostUnreachableError, BadResponseCodeError => e
@@ -434,5 +442,12 @@ def get_es_version(url)
       request = perform_request_to_url(url, :get, ROOT_URI_PATH)
       LogStash::Json.load(request.body)["version"]["number"]
     end
+
+    def set_new_major_version(version)
+      @maximum_seen_major_version = version
+      if @maximum_seen_major_version >= 6
+        @logger.warn("Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type", :es_version => @maximum_seen_major_version)
+      end
+    end
   end
 end; end; end; end;

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -4,20 +4,14 @@ class TemplateManager
     def self.install_template(plugin)
       return unless plugin.manage_template
       plugin.logger.info("Using mapping template from", :path => plugin.template)
-      template = get_template(plugin.template, get_es_major_version(plugin.client))
+      template = get_template(plugin.template, plugin.maximum_seen_major_version)
       plugin.logger.info("Attempting to install template", :manage_template => template)
       install(plugin.client, plugin.template_name, template, plugin.template_overwrite)
     rescue => e
       plugin.logger.error("Failed to install template.", :message => e.message, :class => e.class.name, :backtrace => e.backtrace)
     end
 
     private
-    def self.get_es_major_version(client)
-      # get the elasticsearch version of each node in the pool and
-      # pick the biggest major version
-      client.connected_es_versions.uniq.map {|version| version.split(".").first.to_i}.max
-    end
-
     def self.get_template(path, es_major_version)
       template_path = path || default_template_path(es_major_version)
       read_template_file(template_path)

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '9.0.1'
+  s.version         = '9.0.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/integration/outputs/compressed_indexing_spec.rb

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

spec/integration/outputs/index_spec.rb

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

spec/unit/outputs/elasticsearch/http_client/pool_spec.rb

@@ -7,11 +7,12 @@
   let(:adapter) { LogStash::Outputs::ElasticSearch::HttpClient::ManticoreAdapter.new(logger) }
   let(:initial_urls) { [::LogStash::Util::SafeURI.new("http://localhost:9200")] }
   let(:options) { {:resurrect_delay => 2, :url_normalizer => proc {|u| u}} } # Shorten the delay a bit to speed up tests
+  let(:es_node_versions) { [ "0.0.0" ] }
 
   subject { described_class.new(logger, adapter, initial_urls, options) }
 
   let(:manticore_double) { double("manticore a") }
-  before do
+  before(:each) do
 
     response_double = double("manticore response").as_null_object
     # Allow healtchecks
@@ -21,10 +22,7 @@
 
     allow(::Manticore::Client).to receive(:new).and_return(manticore_double)
 
-    allow(subject).to receive(:connected_es_versions).with(any_args).and_return(["0.0.0"])
-    allow(subject).to receive(:get_es_version).with(any_args).and_return("0.0.0")
-
-    subject.start
+    allow(subject).to receive(:get_es_version).with(any_args).and_return(*es_node_versions)
   end
 
   after do
@@ -34,10 +32,12 @@
   describe "initialization" do
     it "should be successful" do
       expect { subject }.not_to raise_error
+      subject.start
     end
   end
 
   describe "the resurrectionist" do
+    before(:each) { subject.start }
     it "should start the resurrectionist when created" do
       expect(subject.resurrectionist_alive?).to eql(true)
     end
@@ -77,6 +77,7 @@
   end
 
   describe "the sniffer" do
+    before(:each) { subject.start }
     it "should not start the sniffer by default" do
       expect(subject.sniffer_alive?).to eql(nil)
     end
@@ -92,6 +93,7 @@
 
   describe "closing" do
     before do
+      subject.start
       # Simulate a single in use connection on the first check of this
       allow(adapter).to receive(:close).and_call_original
       allow(subject).to receive(:wait_for_in_use_connections).and_call_original
@@ -120,6 +122,7 @@
   end
 
   describe "connection management" do
+    before(:each) { subject.start }
     context "with only one URL in the list" do
       it "should use the only URL in 'with_connection'" do
         subject.with_connection do |c|
@@ -167,4 +170,27 @@
       end
     end
   end
+
+  describe "version tracking" do
+    let(:initial_urls) { [
+      ::LogStash::Util::SafeURI.new("http://somehost:9200"),
+      ::LogStash::Util::SafeURI.new("http://otherhost:9201")
+    ] }
+
+    before(:each) do
+      allow(subject).to receive(:perform_request_to_url).and_return(nil)
+      subject.start
+    end
+
+    it "picks the largest major version" do
+      expect(subject.maximum_seen_major_version).to eq(0)
+    end
+
+    context "if there are nodes with multiple major versions" do
+      let(:es_node_versions) { [ "0.0.0", "6.0.0" ] }
+      it "picks the largest major version" do
+        expect(subject.maximum_seen_major_version).to eq(6)
+      end
+    end
+  end
 end

spec/unit/outputs/elasticsearch/template_manager_spec.rb

@@ -5,26 +5,6 @@
 
 describe LogStash::Outputs::ElasticSearch::TemplateManager do
 
-  describe ".get_es_major_version" do
-    let(:client) { double("client") }
-
-    before(:each) do
-      allow(client).to receive(:connected_es_versions).and_return(["5.3.0"])
-    end
-
-    it "picks the largest major version" do
-      expect(described_class.get_es_major_version(client)).to eq(5)
-    end
-    context "if there are nodes with multiple major versions" do
-      before(:each) do
-        allow(client).to receive(:connected_es_versions).and_return(["5.3.0", "6.0.0"])
-      end
-      it "picks the largest major version" do
-        expect(described_class.get_es_major_version(client)).to eq(6)
-      end
-    end
-  end
-
   describe ".default_template_path" do
     context "elasticsearch 1.x" do
       it "chooses the 2x template" do