Add support for 'auto' setting for ilm_enabled (#838)

* Add support for 'auto' setting for ilm_enabled

Adds a new confifuration option for `ilm_enabled` - `auto`. Using `auto` will
enable ILM if the target Elasticsearch cluster is > 7.0, and has the ILM
feature enabled and available.

* Fix code review comments, log formatting

Removes puts statement, fixes indentation, and fixes misleading
log messages when either the default template is used, or when
a template cannot be found.

* Fix doc typo

Author: robbavey

docs/index.asciidoc

@@ -94,26 +94,20 @@ happens, the problem is logged as a warning, and the event is dropped. See
 <<dead-letter-queues>> for more information about processing events in the DLQ.
 
 [id="plugins-{type}s-{plugin}-ilm"]
-==== Index Lifecycle Management (Beta)
+==== Index Lifecycle Management
 
-beta[]
 
 [NOTE]
 The Index Lifecycle Management feature requires plugin version `9.3.1` or higher.
 
 [NOTE]
 This feature requires an Elasticsearch instance of 6.6.0 or higher with at least a Basic license
 
-Logstash can use the {ref}/index-lifecycle-management.html[Index Lifecycle Management] to automate the management of indices over time.
+Logstash can use {ref}/index-lifecycle-management.html[Index Lifecycle Management] to automate the management of indices over time.
 
-To configure the Elasticsearch output to use Index Lifecycle Management, set the `ilm_enabled` flag to true in the output definition:
-
-[source,ruby]
-    output {
-      elasticsearch {
-        ilm_enabled => true
-      }
-    }
+The use of Index Lifecycle Management is controlled by the `ilm_enabled` setting. By default, this is will
+automatically detect whether the Elasticsearch instance supports ILM, and will use it if it is available. `ilm_enabled`
+can also be set to `true` or `false` to override the automatic detection, or disable ILM.
 
 This will overwrite the index settings and adjust the Logstash template to write the necessary settings for the template
 to support index lifecycle management, including the index policy and rollover alias to be used.
@@ -132,7 +126,6 @@ See config below for an example:
 [source,ruby]
     output {
       elasticsearch {
-        ilm_enabled => true
         ilm_rollover_alias: "custom"
         ilm_pattern: "000001"
         ilm_policy: "custom_policy"
@@ -192,7 +185,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-healthcheck_path>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-hosts>> |<<uri,uri>>|No
 | <<plugins-{type}s-{plugin}-http_compression>> |<<boolean,boolean>>|No
-| <<plugins-{type}s-{plugin}-ilm_enabled>> |<<boolean,boolean>>|No
+| <<plugins-{type}s-{plugin}-ilm_enabled>> |<<string,string>>, one of `["true", "false", "auto"]`|No
 | <<plugins-{type}s-{plugin}-ilm_pattern>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ilm_policy>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ilm_rollover_alias>> |<<string,string>>|No
@@ -374,10 +367,15 @@ Enable gzip compression on requests. Note that response compression is on by def
 [id="plugins-{type}s-{plugin}-ilm_enabled"]
 ===== `ilm_enabled`
 
-  * Value type is <<boolean,boolean>>
-  * Default value is `false`
+  * Value can be any of: `true`, `false`, `auto`
+  * Default value is `auto`
+
+The default setting of `auto` will automatically enable the Index Lifecycle Management feature, if the Elasticsearch
+ cluster is running Elasticsearch version `7.0.0` or higher with the ILM feature enabled, and disable it otherwise.
 
-Setting this flag to `true` will enable indices to be managed by the Index Lifecycle Management feature in Elasticsearch.
+Setting this flag to `false` will disable the Index Lifecycle Management feature, even if the Elasticsearch cluster supports ILM.
+Setting this flag to `true` will enable Index Lifecycle Management feature, if the Elasticsearch cluster supports it. This is required
+to enable Index Lifecycle Management on a version of Elasticsearch earlier than version `7.0.0`.
 
 NOTE: This feature requires a Basic License or above to be installed on an Elasticsearch cluster version 6.6.0 or later
 

lib/logstash/outputs/elasticsearch/common.rb

@@ -47,7 +47,6 @@ def setup_after_successful_connection
           sleep_interval = next_sleep_interval(sleep_interval)
         end
         if successful_connection?
-          verify_ilm_readiness if ilm_enabled?
           install_template
           setup_ilm if ilm_enabled?
         end
@@ -339,6 +338,10 @@ def safe_bulk(actions)
       end
     end
 
+    def default_index?(index)
+      @index == LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
+    end
+
     def dlq_enabled?
       # TODO there should be a better way to query if DLQ is enabled
       # See more in: https://github.com/elastic/logstash/issues/8064

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -5,6 +5,7 @@ module CommonConfigs
 
     DEFAULT_INDEX_NAME = "logstash-%{+YYYY.MM.dd}"
     DEFAULT_POLICY = "logstash-policy"
+    DEFAULT_ROLLOVER_ALIAS = 'logstash'
 
     def self.included(mod)
       # The index to write events to. This can be dynamic using the `%{foo}` syntax.
@@ -141,10 +142,10 @@ def self.included(mod)
       # ILM configurations (beta)
       # -----
       # Flag for enabling Index Lifecycle Management integration.
-      mod.config :ilm_enabled, :validate => :boolean, :default => false
+      mod.config :ilm_enabled, :validate => [true, false, 'true', 'false', 'auto'], :default => 'auto'
 
       # Rollover alias used for indexing data. If rollover alias doesn't exist, Logstash will create it and map it to the relevant index
-      mod.config :ilm_rollover_alias, :validate => :string, :default => 'logstash'
+      mod.config :ilm_rollover_alias, :validate => :string, :default => DEFAULT_ROLLOVER_ALIAS
 
       # appends “{now/d}-000001” by default for new index creation, subsequent rollover indices will increment based on this pattern i.e. “000002”
       # {now/d} is date math, and will insert the appropriate value automatically.

lib/logstash/outputs/elasticsearch/ilm.rb

@@ -5,41 +5,66 @@ module Ilm
 
     def setup_ilm
       return unless ilm_enabled?
-      @logger.info("Using Index lifecycle management - this feature is currently in beta.")
-      @logger.warn "Overwriting supplied index name with rollover alias #{@ilm_rollover_alias}" if @index != LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
-      @index = ilm_rollover_alias
+      if default_index?(@index) || !default_rollover_alias?(@ilm_rollover_alias)
+        logger.warn("Overwriting supplied index #{@index} with rollover alias #{@ilm_rollover_alias}") unless default_index?(@index)
+        @index = @ilm_rollover_alias
+        maybe_create_rollover_alias
+        maybe_create_ilm_policy
+      end
+    end
 
-      maybe_create_rollover_alias
-      maybe_create_ilm_policy
+    def default_rollover_alias?(rollover_alias)
+      rollover_alias == LogStash::Outputs::ElasticSearch::DEFAULT_ROLLOVER_ALIAS
     end
 
     def ilm_enabled?
-      @ilm_enabled
+      return @ilm_actually_enabled if defined?(@ilm_actually_enabled)
+      @ilm_actually_enabled =
+        begin
+          if @ilm_enabled == 'auto'
+            if ilm_on_by_default?
+              ilm_ready, error = ilm_ready?
+              if !ilm_ready
+                @logger.info("Index Lifecycle Management is set to 'auto', but will be disabled - #{error}")
+                false
+              else
+                true
+              end
+            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")
+              false
+            end
+          elsif @ilm_enabled.to_s == 'true'
+            ilm_ready, error = ilm_ready?
+            raise LogStash::ConfigurationError,"Index Lifecycle Management is set to enabled in Logstash, but cannot be used - #{error}"  unless ilm_ready
+            true
+          else
+            false
+          end
+        end
     end
 
-    def verify_ilm_readiness
-      return unless ilm_enabled?
+    def ilm_on_by_default?
+      maximum_seen_major_version >= 7
+    end
 
+    def ilm_ready?
       # Check the Elasticsearch instance for ILM readiness - this means that the version has to be a non-OSS release, with ILM feature
       # available and enabled.
       begin
         xpack = client.get_xpack_info
-        features = xpack["features"]
+        features = xpack.nil? || xpack.empty? ? nil : xpack["features"]
         ilm = features.nil? ? nil : features["ilm"]
-        raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not installed on your Elasticsearch cluster" if features.nil? || ilm.nil?
-        raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not available in your Elasticsearch cluster" unless ilm['available']
-        raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not enabled in your Elasticsearch cluster" unless ilm['enabled']
-
-        unless ilm_policy_default? || client.ilm_policy_exists?(ilm_policy)
-          raise LogStash::ConfigurationError, "The specified ILM policy #{ilm_policy} does not exist on your Elasticsearch instance"
-        end
-
+        return false, "Index Lifecycle management is not installed on your Elasticsearch cluster" if features.nil? || ilm.nil?
+        return false, "Index Lifecycle management is not available in your Elasticsearch cluster" unless ilm['available']
+        return false, "Index Lifecycle management is not enabled in your Elasticsearch cluster" unless ilm['enabled']
+        return true, nil
       rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
         # Check xpack endpoint: If no xpack endpoint, then this version of Elasticsearch is not compatible
         if e.response_code == 404
-          raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not installed on your Elasticsearch cluster"
+          return false, "Index Lifecycle management is not installed on your Elasticsearch cluster"
         elsif e.response_code == 400
-          raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not installed on your Elasticsearch cluster"
+          return false, "Index Lifecycle management is not installed on your Elasticsearch cluster"
         else
           raise e
         end
@@ -53,8 +78,10 @@ def ilm_policy_default?
     end
 
     def maybe_create_ilm_policy
-      if ilm_policy_default? && !client.ilm_policy_exists?(ilm_policy)
-        client.ilm_policy_put(ilm_policy, policy_payload)
+      if ilm_policy_default?
+          client.ilm_policy_put(ilm_policy, policy_payload) unless client.ilm_policy_exists?(ilm_policy)
+      else
+        raise LogStash::ConfigurationError, "The specified ILM policy #{ilm_policy} does not exist on your Elasticsearch instance" unless client.ilm_policy_exists?(ilm_policy)
       end
     end
 

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -3,7 +3,13 @@ class TemplateManager
     # To be mixed into the elasticsearch plugin base
     def self.install_template(plugin)
       return unless plugin.manage_template
-      plugin.logger.info("Using mapping template from", :path => plugin.template)
+      if plugin.template.nil?
+        plugin.logger.info("Using default mapping template")
+      else
+        plugin.logger.info("Using mapping template from", :path => plugin.template)
+      end
+
+
       template = get_template(plugin.template, plugin.maximum_seen_major_version)
       add_ilm_settings_to_template(plugin, template) if plugin.ilm_enabled?
       plugin.logger.info("Attempting to install template", :manage_template => template)
@@ -23,7 +29,6 @@ def self.install(client, template_name, template, template_overwrite)
     end
 
     def self.add_ilm_settings_to_template(plugin, template)
-			plugin.logger.info("Overwriting index patterns, as ILM is enabled.")
       # Overwrite any index patterns, and use the rollover alias. Use 'index_patterns' rather than 'template' for pattern
       # definition - remove any existing definition of 'template'
       template.delete('template') if template.include?('template')
@@ -48,7 +53,7 @@ def self.default_template_path(es_major_version)
     end
 
     def self.read_template_file(template_path)
-      raise ArgumentError, "Template file '#{@template_path}' could not be found!" unless ::File.exists?(template_path)
+      raise ArgumentError, "Template file '#{template_path}' could not be found!" unless ::File.exists?(template_path)
       template_data = ::IO.read(template_path)
       LogStash::Json.load(template_data)
     end

spec/es_spec_helper.rb

@@ -67,6 +67,13 @@ def self.es_version
     end
   end
 
+  RSpec::Matchers.define :have_index_pattern do |expected|
+    match do |actual|
+      test_against = Array(actual['index_patterns'].nil? ? actual['template'] : actual['index_patterns'])
+      test_against.include?(expected)
+    end
+  end
+
 
   def self.es_version_satisfies?(*requirement)
     es_version = RSpec.configuration.filter[:es_version] || ENV['ES_VERSION']
@@ -125,6 +132,38 @@ def supports_ilm?(client)
       false
     end
   end
+
+  def max_docs_policy(max_docs)
+  {
+    "policy" => {
+      "phases"=> {
+        "hot" => {
+          "actions" => {
+            "rollover" => {
+              "max_docs" => max_docs
+            }
+          }
+        }
+      }
+    }
+  }
+  end
+
+  def max_age_policy(max_age)
+  {
+    "policy" => {
+      "phases"=> {
+        "hot" => {
+          "actions" => {
+            "rollover" => {
+              "max_age" => max_age
+            }
+          }
+        }
+      }
+    }
+  }
+  end
 end
 
 RSpec.configure do |config|

spec/integration/outputs/compressed_indexing_spec.rb

@@ -58,7 +58,11 @@
     end
 
     it "sets the correct content-encoding header and body is compressed" do
-      allow(subject.client.pool.adapter.client).to receive(:send).with(anything, /_template/, anything).and_call_original
+      # Allow xpack endpoint to be checked\
+      allow(subject.client.pool.adapter.client).to receive(:send).at_least(:once).with(anything, /_template/, anything).and_call_original
+      allow(subject.client.pool.adapter.client).to receive(:send).with(anything, /_xpack/, anything).and_call_original
+      allow(subject.client.pool.adapter.client).to receive(:send).with(anything, /logstash/, anything).and_call_original
+      allow(subject.client.pool.adapter.client).to receive(:send).with(anything, /_ilm/, anything).and_call_original
       expect(subject.client.pool.adapter.client).to receive(:send).
         with(anything, anything, {:headers=>{"Content-Encoding"=>"gzip", "Content-Type"=>"application/json"}, :body => a_valid_gzip_encoded_string}).
         and_call_original