Improve logging and error handling

Author: robbavey

lib/logstash/outputs/elasticsearch/common.rb

@@ -403,10 +403,9 @@ def verify_ilm
     end
 
     def verify_ilm_config
-      # Default index name is logstash-XXXX and the default write alias is logstash - should we just change overwrite the default index name here?
-      @index = @ilm_write_alias if @index == LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
-      raise LogStash::ConfigurationError, "ILM configuration error: The index name #{@index} does not match the write alias #{@ilm_write_alias}" if @ilm_write_alias != @index
-      raise LogStash::ConfigurationError, "ILM configuration error: The template name #{@template_name} does not match the write alias #{@ilm_write_alias}" if @ilm_write_alias != @template_name
+      # Overwrite the index with the write alias.
+      @logger.warn "Overwriting index name with rollover alias #{@ilm_write_alias}" if @index != LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
+      @index = @ilm_write_alias
       verify_ilm_policy unless ilm_policy_default?
     end
 
@@ -446,8 +445,6 @@ def write_alias_payload
       }
     end
 
-
-
     def policy_payload
       policy_path = ::File.expand_path(ILM_POLICY_PATH, ::File.dirname(__FILE__))
       LogStash::Json.load(::IO.read(policy_path))

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -362,7 +362,21 @@ def write_alias_exists?(name)
     # Create a new write alias
     def write_alias_put(alias_name, alias_definition)
       logger.info("Creating write alias #{alias_name}")
-      @pool.put(CGI::escape(alias_name), nil, LogStash::Json.dump(alias_definition))
+      begin
+        @pool.put(CGI::escape(alias_name), nil, LogStash::Json.dump(alias_definition))
+      # If the rollover alias already exists, ignore the error that comes back from Elasticsearch
+      #
+      rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
+        if e.response_code == 400 && e.response_body
+          error_body = LogStash::Json.load(e.response_bod)
+          if error_body['error'] && error_body['error']['root_cause'] &&
+            error_body['error']['root_cause'][0] && error_body['error']['root_cause'][0]['type'] == 'resource_already_exists_exception'
+            logger.info("Write Alias #{alias_name} already exists. Skipping")
+            return
+          end
+        end
+      raise e
+      end
     end
 
     def get_xpack_info

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -24,7 +24,11 @@ def self.install(client, template_name, template, template_overwrite)
 
     def self.add_ilm_settings_to_template(plugin, template)
       # Include ilm settings in template:
+			plugin.logger.info("Overwriting template name and pattern as ILM is enabled.")
       template['template'] = "#{plugin.ilm_write_alias}-*"
+      if template['settings'] && (template['settings']['index.lifecycle.name'] || template['settings']['index.lifecycle.rollover_alias'])
+        plugin.logger.info("Overwriting index lifecycle name and rollover alias as ILM is enabled.")
+      end
       template['settings'].update({ 'index.lifecycle.name' => plugin.ilm_policy, 'index.lifecycle.rollover_alias' => plugin.ilm_write_alias})
     end
 

spec/es_spec_helper.rb

@@ -3,6 +3,7 @@
 require 'elasticsearch'
 require_relative "support/elasticsearch/api/actions/delete_ilm_policy"
 require_relative "support/elasticsearch/api/actions/get_alias"
+require_relative "support/elasticsearch/api/actions/put_alias"
 require_relative "support/elasticsearch/api/actions/get_ilm_policy"
 require_relative "support/elasticsearch/api/actions/put_ilm_policy"
 
@@ -84,6 +85,17 @@ def put_policy(client, policy_name, policy)
     client.put_ilm_policy({:name => policy_name, :body=> policy})
   end
 
+  def put_alias(client, the_alias, index)
+    body = {
+        "aliases" => {
+            index => {
+                "is_write_index"=>  true
+            }
+        }
+    }
+    client.put_alias({name: the_alias, body: body})
+  end
+
   def clean_ilm(client)
     client.get_ilm_policy.each_key {|key| client.delete_ilm_policy(name: key)}
   end

spec/fixtures/template-with-policy-es7x.json

@@ -3,9 +3,7 @@
   "version" : 60001,
   "settings" : {
     "index.refresh_interval" : "1s",
-    "number_of_shards": 1,
-    "index.lifecycle.name": "custom-policy",
-    "index.lifecycle.rollover_alias": "custom"
+    "number_of_shards": 1
   },
   "mappings" : {
     "_doc" : {

spec/integration/outputs/ilm_spec.rb

@@ -192,6 +192,7 @@
     }
     let (:policy) { small_max_doc_policy }
 
+
     let (:small_max_doc_policy) {
       {"policy" => {
           "phases"=> {
@@ -416,6 +417,7 @@
         before :each do
           put_policy(@es,ilm_policy_name, policy)
         end
+
         it 'should create the write alias' do
           expect(@es.indices.exists_alias(index: ilm_write_alias)).to be_falsey
           subject.register
@@ -424,6 +426,18 @@
           expect(@es.get_alias(name: ilm_write_alias)).to include("#{ilm_write_alias}-#{todays_date}-000001")
         end
 
+        context 'when the custom write alias already exists' do
+          it 'should ignore the already exists error' do
+            expect(@es.indices.exists_alias(index: ilm_write_alias)).to be_falsey
+            put_alias(@es, "#{ilm_write_alias}-#{todays_date}-000001", ilm_write_alias)
+            expect(@es.indices.exists_alias(index: ilm_write_alias)).to be_truthy
+            subject.register
+            sleep(1)
+            expect(@es.get_alias(name: ilm_write_alias)).to include("#{ilm_write_alias}-#{todays_date}-000001")
+          end
+
+        end
+
         it 'should write the ILM settings into the template' do
           subject.register
           sleep(1)

spec/support/elasticsearch/api/actions/put_alias.rb

@@ -0,0 +1,24 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License;
+# you may not use this file except in compliance with the Elastic License.
+
+module Elasticsearch
+  module API
+    module Actions
+
+      # @option arguments [String] :name The name of the alias (*Required*)
+      # @option arguments [Hash]   :The alias definition(*Required*)
+
+      def put_alias(arguments={})
+        raise ArgumentError, "Required argument 'name' missing" unless arguments[:name]
+        raise ArgumentError, "Required argument 'body' missing" unless arguments[:body]
+        method = HTTP_PUT
+        path   = Utils.__pathify Utils.__escape(arguments[:name])
+
+        params = Utils.__validate_and_extract_params arguments
+        body   = arguments[:body]
+        perform_request(method, path, params, body.to_json).body
+      end
+    end
+  end
+end

spec/support/elasticsearch/api/actions/put_ilm_policy.rb

@@ -6,8 +6,8 @@ module Elasticsearch
   module API
     module Actions
 
-      # @option arguments [String] :name The name of the template (*Required*)
-      # @option arguments [Hash] :body The template definition (*Required*)
+      # @option arguments [String] :name The name of the policy (*Required*)
+      # @option arguments [Hash] :body The policy definition (*Required*)
 
       def put_ilm_policy(arguments={})
         raise ArgumentError, "Required argument 'name' missing" unless arguments[:name]