Revert "Add distribution check"

This reverts commit 87847af

Signed-off-by: Vijayan Balasubramanian <balasvij@amazon.com>

Author: VijayanB

lib/logstash/outputs/opensearch.rb

@@ -68,7 +68,6 @@
 class LogStash::Outputs::OpenSearch < LogStash::Outputs::Base
   declare_threadsafe!
 
-  require "logstash/outputs/opensearch/distribution_checker"
   require "logstash/outputs/opensearch/http_client"
   require "logstash/outputs/opensearch/http_client_builder"
   require "logstash/plugin_mixins/opensearch/api_configs"
@@ -218,7 +217,7 @@ def register
 
     @logger.info("New OpenSearch output", :class => self.class.name, :hosts => @hosts.map(&:sanitized).map(&:to_s))
 
-    @client = build_client(DistributionChecker.new(@logger))
+    @client = build_client
 
     @after_successful_connection_thread = after_successful_connection do
       begin

lib/logstash/outputs/opensearch/distribution_checker.rb

@@ -332,7 +332,6 @@ def build_pool(options)
       adapter = build_adapter(options)
 
       pool_options = {
-        :distribution_checker => options[:distribution_checker],
         :sniffing => sniffing,
         :sniffer_delay => options[:sniffer_delay],
         :sniffing_path => options[:sniffing_path],

lib/logstash/outputs/opensearch/http_client.rb

@@ -8,7 +8,6 @@
 #  GitHub history for details.
 
 require "concurrent/atomic/atomic_reference"
-require "logstash/plugin_mixins/opensearch/noop_distribution_checker"
 
 module LogStash; module Outputs; class OpenSearch; class HttpClient;
   class Pool
@@ -41,7 +40,6 @@ def message
     end
 
     attr_reader :logger, :adapter, :sniffing, :sniffer_delay, :resurrect_delay, :healthcheck_path, :sniffing_path, :bulk_path
-    attr_reader :distribution_checker
 
     ROOT_URI_PATH = '/'.freeze
 
@@ -80,7 +78,6 @@ def initialize(logger, adapter, initial_urls=[], options={})
       @stopping = false
 
       @last_version = Concurrent::AtomicReference.new
-      @distribution_checker = options[:distribution_checker] || LogStash::PluginMixins::OpenSearch::NoopDistributionChecker::INSTANCE
     end
 
     def start
@@ -239,8 +236,7 @@ def healthcheck!
           @state_mutex.synchronize do
             meta[:version] = version
             set_last_version(version, url)
-            alive = @distribution_checker.is_supported?(self, url, @maximum_seen_major_version)
-            meta[:state] = alive ? :alive : :dead
+            meta[:state] = :alive
           end
         rescue HostUnreachableError, BadResponseCodeError => e
           logger.warn("Attempted to resurrect connection to dead OpenSearch instance, but got an error", url: url.sanitized.to_s, exception: e.class, message: e.message)
@@ -415,18 +411,9 @@ def return_connection(url)
       end
     end
 
-    def get_version_map(url)
-      request = perform_request_to_url(url, :get, ROOT_URI_PATH)
-      LogStash::Json.load(request.body)['version']
-    end
-
     def get_version(url)
-      get_version_map(url)['number'] # e.g. "7.10.0"
-    end
-
-    def get_distribution(url)
-      version_map = get_version_map(url)
-      version_map['distribution']
+      request = perform_request_to_url(url, :get, ROOT_URI_PATH)
+      LogStash::Json.load(request.body)["version"]["number"] # e.g. "7.10.0"
     end
 
     def last_version

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

@@ -24,7 +24,6 @@ def self.build(logger, hosts, params)
       client_settings[:proxy] = params["proxy"] if params["proxy"]
 
       common_options = {
-        :distribution_checker => params["distribution_checker"],
         :client_settings => client_settings,
         :metric => params["metric"],
         :resurrect_delay => params["resurrect_delay"]

lib/logstash/outputs/opensearch/http_client_builder.rb

@@ -24,8 +24,8 @@ module Common
     # Perform some OpenSearch options validations and Build the HttpClient.
     # Note that this methods may sets the @user, @password, @hosts and @client ivars as a side effect.
     # @return [HttpClient] the new http client
-    def build_client(distribution_checker=nil)
-      params["distribution_checker"] = distribution_checker
+    def build_client
+
       # the following 3 options validation & setup methods are called inside build_client
       # because they must be executed prior to building the client and logstash
       # monitoring and management rely on directly calling build_client

lib/logstash/plugin_mixins/opensearch/common.rb

@@ -16,8 +16,7 @@
   let(:adapter) { LogStash::Outputs::OpenSearch::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(:node_versions) { [ "7.0.0" ] }
-  let(:get_distribution) { "opensearch" }
+  let(:node_versions) { [ "0.0.0" ] }
 
   subject { described_class.new(logger, adapter, initial_urls, options) }
 
@@ -32,7 +31,6 @@
     allow(::Manticore::Client).to receive(:new).and_return(manticore_double)
 
     allow(subject).to receive(:get_version).with(any_args).and_return(*node_versions)
-    allow(subject.distribution_checker).to receive(:get_distribution).and_return(get_distribution)
   end
 
   after do
@@ -212,96 +210,13 @@
     end
 
     it "picks the largest major version" do
-      expect(subject.maximum_seen_major_version).to eq(7)
+      expect(subject.maximum_seen_major_version).to eq(0)
     end
 
     context "if there are nodes with multiple major versions" do
-      let(:node_versions) { [ "0.0.0", "7.0.0" ] }
+      let(:node_versions) { [ "0.0.0", "6.0.0" ] }
       it "picks the largest major version" do
-        expect(subject.maximum_seen_major_version).to eq(7)
-      end
-    end
-  end
-  describe "distribution checking" do
-    before(:each) do
-      allow(subject).to receive(:health_check_request)
-    end
-
-    let(:options) do
-      super().merge(:distribution_checker => distribution_checker)
-    end
-
-    context 'when DistributionChecker#is_supported? returns false' do
-      let(:distribution_checker) { double('DistributionChecker', :is_supported? => false) }
-
-      it 'does not mark the URL as active' do
-        subject.update_initial_urls
-        expect(subject.alive_urls_count).to eq(0)
-      end
-    end
-
-    context 'when DistributionChecker#is_supported? returns true' do
-      let(:distribution_checker) { double('DistributionChecker', :is_supported? => true) }
-
-      it 'marks the URL as active' do
-        subject.update_initial_urls
-        expect(subject.alive_urls_count).to eq(1)
-      end
-    end
-  end
-  describe 'distribution checking with cluster output' do
-    let(:options) do
-      super().merge(:distribution_checker => LogStash::Outputs::OpenSearch::DistributionChecker.new(logger))
-    end
-
-    before(:each) do
-      allow(subject).to receive(:health_check_request)
-    end
-
-    context 'when using opensearch' do
-
-      context "cluster doesn't return a valid distribution" do
-        let(:get_distribution) { nil }
-        context "major version is not 7" do
-          let(:node_versions) { [ "6.0.0" ] }
-
-          it 'marks the url as dead' do
-            subject.update_initial_urls
-            expect(subject.alive_urls_count).to eq(0)
-          end
-
-          it 'logs message' do
-            expect(subject.distribution_checker).to receive(:log_not_supported).once.and_call_original
-            subject.update_initial_urls
-          end
-        end
-        context "major version is  7" do
-          let(:node_versions) { [ "7.10.2" ] }
-
-          it "marks the url as active" do
-            subject.update_initial_urls
-            expect(subject.alive_urls_count).to eq(1)
-          end
-
-          it 'does not log message' do
-            expect(subject.distribution_checker).to_not receive(:log_not_supported)
-            subject.update_initial_urls
-          end
-
-        end
-      end
-      context 'cluster returns valid distribution' do
-        let(:get_distribution) { 'opensearch' }
-
-        it "marks the url as active" do
-          subject.update_initial_urls
-          expect(subject.alive_urls_count).to eq(1)
-        end
-
-        it 'does not log message' do
-          expect(subject.distribution_checker).to_not receive(:log_not_supported)
-          subject.update_initial_urls
-        end
+        expect(subject.maximum_seen_major_version).to eq(6)
       end
     end
   end