Adding a config to skip_healthcheck (#170)

* Adding a config for default_server_major_version, and use it when the version can't be fetched from the root path of OpenSearch.

Signed-off-by: Deep Datta <deedatta@amazon.com>

Author: deepdatta

lib/logstash/outputs/opensearch.rb

@@ -199,11 +199,14 @@ class LogStash::Outputs::OpenSearch < LogStash::Outputs::Base
   # Set which ingest pipeline you wish to execute for an event. You can also use event dependent configuration
   # here like `pipeline => "%{INGEST_PIPELINE}"`
   config :pipeline, :validate => :string, :default => nil
-
+  
   # When set to true, use legacy templates via the _template API
   # When false, use index templates using the _index_template API
   config :legacy_template, :validate => :boolean, :default => true
 
+  # The OpenSearch server major version to use when it's not available from the Healthcheck endpoint.
+  config :default_server_major_version, :validate => :number
+
   attr_reader :client
   attr_reader :default_index
   attr_reader :default_template_name

lib/logstash/outputs/opensearch/http_client.rb

@@ -328,7 +328,8 @@ def build_pool(options)
         :healthcheck_path => options[:healthcheck_path],
         :resurrect_delay => options[:resurrect_delay],
         :url_normalizer => self.method(:host_to_url),
-        :metric => options[:metric]
+        :metric => options[:metric],
+        :default_server_major_version => options[:default_server_major_version]
       }
       pool_options[:scheme] = self.scheme if self.scheme
 

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

@@ -40,6 +40,7 @@ def message
     end
 
     attr_reader :logger, :adapter, :sniffing, :sniffer_delay, :resurrect_delay, :healthcheck_path, :sniffing_path, :bulk_path
+    attr_reader :default_server_major_version
 
     ROOT_URI_PATH = '/'.freeze
 
@@ -68,6 +69,7 @@ def initialize(logger, adapter, initial_urls=[], options={})
         @resurrect_delay = merged[:resurrect_delay]
         @sniffing = merged[:sniffing]
         @sniffer_delay = merged[:sniffer_delay]
+        @default_server_major_version = merged[:default_server_major_version]
       end
 
       # Used for all concurrent operations in this class
@@ -412,8 +414,15 @@ def return_connection(url)
     end
 
     def get_version(url)
-      request = perform_request_to_url(url, :get, ROOT_URI_PATH)
-      LogStash::Json.load(request.body)["version"]["number"] # e.g. "7.10.0"
+      response = perform_request_to_url(url, :get, ROOT_URI_PATH)
+      if response.code != 404 && !response.body.empty?
+        return LogStash::Json.load(response.body)["version"]["number"] # e.g. "7.10.0"
+      end
+      if @default_server_major_version.nil?
+        @logger.error("Failed to get version from health_check endpoint and default_server_major_version is not configured.")
+        raise "get_version failed! no default_server_major_version configured."
+      end
+      "#{default_server_major_version}.0.0"
     end
 
     def last_version

lib/logstash/outputs/opensearch/http_client_builder.rb

@@ -27,7 +27,8 @@ def self.build(logger, hosts, params)
       common_options = {
         :client_settings => client_settings,
         :metric => params["metric"],
-        :resurrect_delay => params["resurrect_delay"]
+        :resurrect_delay => params["resurrect_delay"],
+        :default_server_major_version => params["default_server_major_version"]
       }
 
       if params["sniffing"]