check if ES is licensed before pushing out events to it

* don't mark connection as not alive. only log a warn

Author: jsvd

lib/logstash/outputs/elasticsearch.rb

@@ -244,6 +244,10 @@ def close
     @client.close if @client
   end
 
+  def self.oss?
+    LogStash::OSS
+  end
+
   @@plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-elasticsearch-/ }
 
   @@plugins.each do |plugin|

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

@@ -31,6 +31,7 @@ def message
     attr_reader :logger, :adapter, :sniffing, :sniffer_delay, :resurrect_delay, :healthcheck_path, :sniffing_path, :bulk_path
 
     ROOT_URI_PATH = '/'.freeze
+    LICENSE_PATH = '/_license'.freeze
 
     DEFAULT_OPTIONS = {
       :healthcheck_path => ROOT_URI_PATH,
@@ -66,13 +67,21 @@ def initialize(logger, adapter, initial_urls=[], options={})
       @url_info = {}
       @stopping = false
     end
+
+    def oss?
+     LogStash::Outputs::ElasticSearch.oss?
+    end
 
     def start
-      update_urls(@initial_urls)
+      update_initial_urls
       start_resurrectionist
       start_sniffer if @sniffing
     end
 
+    def update_initial_urls
+      update_urls(@initial_urls)
+    end
+
     def close
       @state_mutex.synchronize { @stopping = true }
 
@@ -101,7 +110,7 @@ def in_use_connections
     end
 
     def alive_urls_count
-      @state_mutex.synchronize { @url_info.values.select {|v| !v[:state] == :alive }.count }
+      @state_mutex.synchronize { @url_info.values.select {|v| v[:state] == :alive }.count }
     end
 
     def url_info
@@ -236,13 +245,29 @@ def start_resurrectionist
       end
     end
 
+    def get_license(url)
+      response = perform_request_to_url(url, :get, LICENSE_PATH)
+      LogStash::Json.load(response.body)
+    end
+
+    def valid_es_license?(url)
+      license = get_license(url)
+      license.fetch("license", {}).fetch("status", nil) == "active"
+    rescue => e
+      false
+    end
+
+    def health_check_request(url)
+      perform_request_to_url(url, :head, @healthcheck_path)
+    end
+
     def healthcheck!
       # Try to keep locking granularity low such that we don't affect IO...
       @state_mutex.synchronize { @url_info.select {|url,meta| meta[:state] != :alive } }.each do |url,meta|
         begin
           logger.debug("Running health check to see if an Elasticsearch connection is working",
                         :healthcheck_url => url, :path => @healthcheck_path)
-          response = perform_request_to_url(url, :head, @healthcheck_path)
+          health_check_request(url)
           # If no exception was raised it must have succeeded!
           logger.warn("Restored connection to ES instance", :url => url.sanitized.to_s)
           # We reconnected to this node, check its ES version
@@ -254,10 +279,22 @@ def healthcheck!
               @logger.info("ES Output version determined", :es_version => major)
               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)
+              @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.sanitized.to_s)
               set_new_major_version(major)
             end
-            meta[:state] = :alive
+            if oss? || valid_es_license?(url)
+              meta[:state] = :alive
+            else
+              # As this version is to be shipped with Logstash 7.x we won't mark the connection as unlicensed
+              #
+              #  logger.error("Cannot connect to the Elasticsearch cluster configured in the Elasticsearch output. Logstash requires the default distribution of Elasticsearch. Please update to the default distribution of Elasticsearch for full access to all free features, or switch to the OSS distribution of Logstash.", :url => url.sanitized.to_s)
+              #  meta[:state] = :unlicensed
+              #
+              # Instead we'll log a deprecation warning and mark it as alive:
+              #
+              log_license_deprecation_warn(url)
+              meta[:state] = :alive
+            end
           end
         rescue HostUnreachableError, BadResponseCodeError => e
           logger.warn("Attempted to resurrect connection to dead ES instance, but got an error.", url: url.sanitized.to_s, error_type: e.class, error: e.message)
@@ -269,6 +306,10 @@ def stop_resurrectionist
       @resurrectionist.join if @resurrectionist
     end
 
+    def log_license_deprecation_warn(url)
+      logger.warn("DEPRECATION WARNING: Connecting to an OSS distribution of Elasticsearch using the default distribution of Logstash will stop working in Logstash 8.0.0. Please upgrade to the default distribution of Elasticsearch, or use the OSS distribution of Logstash", :url => url.sanitized.to_s)
+    end
+
     def resurrectionist_alive?
       @resurrectionist ? @resurrectionist.alive? : nil
     end

spec/es_spec_helper.rb

@@ -9,6 +9,10 @@
 
 require 'json'
 
+unless defined?(LogStash::OSS)
+  LogStash::OSS = ENV['DISTRIBUTION'] != "default"
+end
+
 module ESHelper
   def get_host_port
     if ENV["INTEGRATION"] == "true"

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

@@ -8,12 +8,15 @@
   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" ] }
+  let(:oss) { true }
+  let(:valid_license) { true }
 
   subject { described_class.new(logger, adapter, initial_urls, options) }
 
   let(:manticore_double) { double("manticore a") }
   before(:each) do
 
+    allow(::LogStash::Outputs::ElasticSearch).to receive(:oss?).and_return(oss)
     response_double = double("manticore response").as_null_object
     # Allow healtchecks
     allow(manticore_double).to receive(:head).with(any_args).and_return(response_double)
@@ -23,6 +26,8 @@
     allow(::Manticore::Client).to receive(:new).and_return(manticore_double)
 
     allow(subject).to receive(:get_es_version).with(any_args).and_return(*es_node_versions)
+    allow(subject).to receive(:oss?).and_return(oss)
+    allow(subject).to receive(:valid_es_license?).and_return(valid_license)
   end
 
   after do
@@ -235,4 +240,31 @@
       end
     end
   end
+
+  describe "license checking" do
+    before(:each) do
+      allow(subject).to receive(:health_check_request)
+    end
+    context "when using default logstash distribution" do
+      let(:oss) { false }
+      context "if ES doesn't return a valid license" do
+        let(:valid_license) { false }
+        it "marks the url as active" do
+          subject.update_initial_urls
+          expect(subject.alive_urls_count).to eq(1)
+        end
+        it "logs a warning" do
+          expect(subject).to receive(:log_license_deprecation_warn).once
+          subject.update_initial_urls
+        end
+      end
+      context "if ES returns a valid license" do
+        let(:valid_license) { true }
+        it "marks the url as active" do
+          subject.update_initial_urls
+          expect(subject.alive_urls_count).to eq(1)
+        end
+      end
+    end
+  end
 end