Add support to Serverless Elasticsearch (#1145)

The pool stores build flavor in health check to determine if the
Elasticsearch is serverless.
In serverless, ILM, legacy template API and license checker are disabled,
and the plugin throws ConfigurationError when ilm_enabled => true or
template_api => legacy

Co-authored-by: João Duarte <[email protected]>

Author: kaisecheng

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.16.0
+  - Added support to Serverless Elasticsearch [#1445](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1145)
+
 ## 11.15.9
   - allow dlq_ settings when using data streams [#1144](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1144)
 

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -93,6 +93,10 @@ def maximum_seen_major_version
       @pool.maximum_seen_major_version
     end
 
+    def serverless?
+      @pool.serverless?
+    end
+
     def alive_urls_count
       @pool.alive_urls_count
     end

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

@@ -48,6 +48,8 @@ def initialize(original_error, url)
       :sniffer_delay => 10,
     }.freeze
 
+    BUILD_FLAVOUR_SERVERLESS = 'serverless'.freeze
+
     def initialize(logger, adapter, initial_urls=[], options={})
       @logger = logger
       @adapter = adapter
@@ -75,6 +77,7 @@ def initialize(logger, adapter, initial_urls=[], options={})
       @license_checker = options[:license_checker] || LogStash::PluginMixins::ElasticSearch::NoopLicenseChecker::INSTANCE
 
       @last_es_version = Concurrent::AtomicReference.new
+      @build_flavour = Concurrent::AtomicReference.new
     end
 
     def start
@@ -250,14 +253,18 @@ def healthcheck!(register_phase = true)
           # 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
-          es_version = get_es_version(url)
+          version_info = get_es_version(url)
+          es_version = version_info.fetch('number', nil)
+          build_flavour = version_info.fetch('build_flavor', nil)
+
           if es_version.nil?
             logger.warn("Failed to retrieve Elasticsearch version data from connected endpoint, connection aborted", :url => url.sanitized.to_s)
             next
           end
           @state_mutex.synchronize do
             meta[:version] = es_version
             set_last_es_version(es_version, url)
+            set_build_flavour(build_flavour)
 
             alive = @license_checker.appropriate_license?(self, url)
             meta[:state] = alive ? :alive : :dead
@@ -475,7 +482,7 @@ def get_es_version(url)
 
       response = LogStash::Json.load(response.body)
 
-      response.fetch('version', {}).fetch('number', nil)
+      response.fetch('version', {})
     end
 
     def last_es_version
@@ -486,6 +493,10 @@ def maximum_seen_major_version
       @state_mutex.synchronize { @maximum_seen_major_version }
     end
 
+    def serverless?
+      @build_flavour.get == BUILD_FLAVOUR_SERVERLESS
+    end
+
     private
 
     # @private executing within @state_mutex
@@ -515,5 +526,9 @@ def warn_on_higher_major_version(major, url)
                    previous_major: @maximum_seen_major_version, new_major: major, node_url: url.sanitized.to_s)
     end
 
+    def set_build_flavour(flavour)
+      @build_flavour.set(flavour)
+    end
+
   end
 end; end; end; end;

lib/logstash/outputs/elasticsearch/ilm.rb

@@ -14,11 +14,18 @@ def ilm_in_use?
       return @ilm_actually_enabled if defined?(@ilm_actually_enabled)
       @ilm_actually_enabled =
         begin
-          if @ilm_enabled == 'auto'
+          if serverless?
+            raise LogStash::ConfigurationError, "Invalid ILM configuration `ilm_enabled => true`. " +
+              "Serverless Elasticsearch cluster does not support Index Lifecycle Management." if @ilm_enabled.to_s == 'true'
+            @logger.info("ILM auto configuration (`ilm_enabled => auto` or unset) resolved to `false`. "\
+              "Serverless Elasticsearch cluster does not support Index Lifecycle Management.") if @ilm_enabled == 'auto'
+            false
+          elsif @ilm_enabled == 'auto'
             if ilm_on_by_default?
               ilm_alias_set?
             else
-              @logger.info("Index Lifecycle Management is set to 'auto', but will be disabled - Your Elasticsearch cluster is before 7.0.0, which is the minimum version required to automatically run Index Lifecycle Management")
+              @logger.info("ILM auto configuration (`ilm_enabled => auto` or unset) resolved to `false`."\
+                " Elasticsearch cluster is before 7.0.0, which is the minimum version required to automatically run Index Lifecycle Management")
               false
             end
           elsif @ilm_enabled.to_s == 'true'

lib/logstash/outputs/elasticsearch/license_checker.rb

@@ -11,6 +11,8 @@ def initialize(logger)
     # @param url [LogStash::Util::SafeURI] ES node URL
     # @return [Boolean] true if provided license is deemed appropriate
     def appropriate_license?(pool, url)
+      return true if pool.serverless?
+
       license = extract_license(pool.get_license(url))
       case license_status(license)
       when 'active'

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -13,8 +13,11 @@ def self.install_template(plugin)
                            "We recommend either setting `template_api => legacy` to continue providing legacy-style templates, " +
                            "or migrating your template to the composable style and setting `template_api => composable`. " +
                            "The legacy template API is slated for removal in Elasticsearch 9.")
+      elsif plugin.template_api == 'legacy' && plugin.serverless?
+        raise LogStash::ConfigurationError, "Invalid template configuration `template_api => legacy`. Serverless Elasticsearch does not support legacy template API."
       end
 
+
       if plugin.template
         plugin.logger.info("Using mapping template from", :path => plugin.template)
         template = read_template_file(plugin.template)
@@ -61,11 +64,13 @@ def self.resolve_template_settings(plugin, template)
         plugin.logger.trace("Resolving ILM template settings: under 'settings' key", :template => template, :template_api => plugin.template_api, :es_version => plugin.maximum_seen_major_version)
         legacy_index_template_settings(template)
       else
-        template_endpoint = template_endpoint(plugin)
-        plugin.logger.trace("Resolving ILM template settings: template doesn't have 'settings' or 'template' fields, falling back to auto detection", :template => template, :template_api => plugin.template_api, :es_version => plugin.maximum_seen_major_version, :template_endpoint => template_endpoint)
-        template_endpoint == INDEX_TEMPLATE_ENDPOINT ?
-          composable_index_template_settings(template) :
+        use_index_template_api = index_template_api?(plugin)
+        plugin.logger.trace("Resolving ILM template settings: template doesn't have 'settings' or 'template' fields, falling back to auto detection", :template => template, :template_api => plugin.template_api, :es_version => plugin.maximum_seen_major_version, :index_template_api => use_index_template_api)
+        if use_index_template_api
+          composable_index_template_settings(template)
+        else
           legacy_index_template_settings(template)
+        end
       end
     end
 
@@ -100,12 +105,25 @@ def self.read_template_file(template_path)
     end
 
     def self.template_endpoint(plugin)
-      if plugin.template_api == 'auto'
-        plugin.maximum_seen_major_version < 8 ? LEGACY_TEMPLATE_ENDPOINT : INDEX_TEMPLATE_ENDPOINT
-      elsif plugin.template_api.to_s == 'legacy'
-        LEGACY_TEMPLATE_ENDPOINT
+      index_template_api?(plugin) ? INDEX_TEMPLATE_ENDPOINT : LEGACY_TEMPLATE_ENDPOINT
+    end
+
+    def self.index_template_api?(plugin)
+      case plugin.serverless?
+      when true
+        true
       else
-        INDEX_TEMPLATE_ENDPOINT
+        case plugin.template_api
+        when 'auto'
+          plugin.maximum_seen_major_version >= 8
+        when 'composable'
+          true
+        when 'legacy'
+          false
+        else
+          plugin.logger.warn("Invalid template_api value #{plugin.template_api}")
+          true
+        end
       end
     end
 

lib/logstash/plugin_mixins/elasticsearch/common.rb

@@ -145,6 +145,10 @@ def maximum_seen_major_version
       client.maximum_seen_major_version
     end
 
+    def serverless?
+      client.serverless?
+    end
+
     def alive_urls_count
       client.alive_urls_count
     end

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '11.15.9'
+  s.version         = '11.16.0'
   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

@@ -59,11 +59,14 @@ def routing_field_name
   end
 
   def self.es_version
-    [
-      nilify(RSpec.configuration.filter[:es_version]),
-      nilify(ENV['ES_VERSION']),
-      nilify(ENV['ELASTIC_STACK_VERSION']),
-    ].compact.first
+    {
+      "number" => [
+        nilify(RSpec.configuration.filter[:es_version]),
+        nilify(ENV['ES_VERSION']),
+        nilify(ENV['ELASTIC_STACK_VERSION']),
+      ].compact.first,
+      "build_flavor" => 'default'
+    }
   end
 
   RSpec::Matchers.define :have_hits do |expected|

spec/fixtures/license_check/active.json

@@ -0,0 +1,16 @@
+{
+  "license": {
+    "status": "active",
+    "uid": "d85d2c6a-b96d-3cc6-96db-5571a789b156",
+    "type": "enterprise",
+    "issue_date": "1970-01-01T00:00:00.000Z",
+    "issue_date_in_millis": 0,
+    "expiry_date": "2100-01-01T00:00:00.000Z",
+    "expiry_date_in_millis": 4102444800000,
+    "max_nodes": null,
+    "max_resource_units": 100000,
+    "issued_to": "Elastic Cloud",
+    "issuer": "API",
+    "start_date_in_millis": 0
+  }
+}