General tidy up

Author: robbavey

ci/run.sh

@@ -179,7 +179,6 @@ else
       start_es
       bundle exec rspec -fd $extra_tag_args --tag update_tests:painless --tag update_tests:groovy --tag es_version:$es_distribution_version $spec_path
       ;;
-
     5.*)
       setup_es https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-${ES_VERSION}.tar.gz
       es_distribution_version=$(get_es_distribution_version)

lib/logstash/outputs/elasticsearch/common.rb

@@ -1,10 +1,8 @@
 require "logstash/outputs/elasticsearch/template_manager"
-require 'logstash/environment'
-
 
 module LogStash; module Outputs; class ElasticSearch;
   module Common
-    attr_reader :client, :hosts, :ilm_manager
+    attr_reader :client, :hosts
 
     # These codes apply to documents, not at the request level
     DOC_DLQ_CODES = [400, 404]
@@ -28,7 +26,7 @@ def register
 
       setup_hosts # properly sets @hosts
       build_client
-      install_template_after_successful_connection
+      setup_after_successful_connection
       check_action_validity
       @bulk_request_metrics = metric.namespace(:bulk_requests)
       @document_level_metrics = metric.namespace(:documents)
@@ -43,7 +41,7 @@ def multi_receive(events)
       retrying_submit(events.map {|e| event_action_tuple(e)})
     end
 
-    def install_template_after_successful_connection
+    def setup_after_successful_connection
       @template_installer ||= Thread.new do
         sleep_interval = @retry_initial_interval
         until successful_connection? || @stopping.true?
@@ -52,9 +50,9 @@ def install_template_after_successful_connection
           sleep_interval = next_sleep_interval(sleep_interval)
         end
         if successful_connection?
-          verify_ilm
+          verify_ilm_readiness if ilm_enabled?
           install_template
-          setup_ilm
+          setup_ilm if ilm_enabled?
         end
       end
     end
@@ -123,7 +121,6 @@ def maximum_seen_major_version
       client.maximum_seen_major_version
     end
 
-
     def routing_field_name
       maximum_seen_major_version >= 6 ? :routing : :_routing
     end
@@ -374,61 +371,48 @@ def setup_ilm
       maybe_create_ilm_policy
     end
 
-    def verify_ilm
-      return true unless ilm_enabled?
-      begin
-        verify_ilm_config
+    def verify_ilm_readiness
+      return unless ilm_enabled?
 
-        # For elasticsearch versions without ilm enablement, the _ilm endpoint will return a 400 bad request - the
-        # endpoint is interpreted as an index, and will return a bad request error as that is an illegal format for
-        # an index.
+      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
+
+      # 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"]
         ilm = features["ilm"] unless features.nil?
         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']
-        true
       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
-          false
           raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not installed on your Elasticsearch cluster"
         elsif e.response_code == 400
-          false
           raise LogStash::ConfigurationError, "Index Lifecycle management is enabled in logstash, but not installed on your Elasticsearch cluster"
         else
           raise e
         end
       end
     end
 
-    def verify_ilm_config
-      # Overwrite the index with the rollover alias.
-      @logger.warn "Overwriting index name with rollover alias #{@ilm_rollover_alias}" if @index != LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
-      @index = @ilm_rollover_alias
-      verify_ilm_policy unless ilm_policy_default?
-    end
-
-    def ilm_policy_ok?
-
-    end
-
     def ilm_policy_default?
       ilm_policy == DEFAULT_POLICY
     end
 
-    def verify_ilm_policy
-      raise LogStash::ConfigurationError, "The specified ILM policy does not exist" unless client.ilm_policy_exists?(ilm_policy)
-    end
-
     def maybe_create_ilm_policy
       if ilm_policy_default? && !client.ilm_policy_exists?(ilm_policy)
         client.ilm_policy_put(ilm_policy, policy_payload)
       end
     end
 
     def maybe_create_rollover_alias
+      @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
+
       client.rollover_alias_put(rollover_alias_target, rollover_alias_payload) unless client.rollover_alias_exists?(ilm_rollover_alias)
     end
 
@@ -438,9 +422,9 @@ def rollover_alias_target
 
     def rollover_alias_payload
       {
-          "aliases" => {
+          'aliases' => {
               ilm_rollover_alias =>{
-                  "is_write_index" =>  true
+                  'is_write_index' =>  true
               }
           }
       }

lib/logstash/outputs/elasticsearch/elasticsearch-template-es6x.json

@@ -2,8 +2,7 @@
   "template" : "logstash-*",
   "version" : 60001,
   "settings" : {
-    "index.refresh_interval" : "5s",
-    "number_of_shards": 1
+    "index.refresh_interval" : "5s"
   },
   "mappings" : {
     "_default_" : {

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -364,8 +364,7 @@ def rollover_alias_put(alias_name, alias_definition)
       logger.info("Creating rollover alias #{alias_name}")
       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
-      #
+        # 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
             logger.info("Rollover Alias #{alias_name} already exists. Skipping")

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -23,7 +23,6 @@ def self.install(client, template_name, template, template_overwrite)
     end
 
     def self.add_ilm_settings_to_template(plugin, template)
-      # Include ilm settings in 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'

spec/integration/outputs/ilm_spec.rb

@@ -442,7 +442,6 @@
         it 'should write the ILM settings into the template' do
           subject.register
           sleep(1)
-          puts @es.indices.get_template(name: template_name)[template_name]
           expect(@es.indices.get_template(name: template_name)[template_name]["index_patterns"]).to eq(["#{ilm_rollover_alias}-*"])
           expect(@es.indices.get_template(name: template_name)[template_name]["settings"]['index']['lifecycle']['name']).to eq(ilm_policy_name)
           expect(@es.indices.get_template(name: template_name)[template_name]["settings"]['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias)

spec/integration/outputs/parent_spec.rb

@@ -2,7 +2,7 @@
 require "logstash/outputs/elasticsearch"
 
 if ESHelper.es_version_satisfies?("<= 5.x")
-  context "when using elasticsearch 5.x and before", :integration => "changethis" do
+  context "when using elasticsearch 5.x and before", :integration => true do
     shared_examples "a type based parent indexer" do
       let(:index) { 10.times.collect { rand(10).to_s }.join("") }
       let(:type) { 10.times.collect { rand(10).to_s }.join("") }

spec/integration/outputs/routing_spec.rb

@@ -34,28 +34,28 @@
     end
 end
 
-# describe "(http protocol) index events with static routing", :integration => true do
-#   it_behaves_like 'a routing indexer' do
-#     let(:routing) { "test" }
-#     let(:config) {
-#       {
-#         "hosts" => get_host_port,
-#         "index" => index,
-#         "routing" => routing
-#       }
-#     }
-#   end
-# end
-# 
-# describe "(http_protocol) index events with fieldref in routing value", :integration => true do
-#   it_behaves_like 'a routing indexer' do
-#     let(:routing) { "test" }
-#     let(:config) {
-#       {
-#         "hosts" => get_host_port,
-#         "index" => index,
-#         "routing" => "%{message}"
-#       }
-#     }
-#   end
-# end
+describe "(http protocol) index events with static routing", :integration => true do
+  it_behaves_like 'a routing indexer' do
+    let(:routing) { "test" }
+    let(:config) {
+      {
+        "hosts" => get_host_port,
+        "index" => index,
+        "routing" => routing
+      }
+    }
+  end
+end
+
+describe "(http_protocol) index events with fieldref in routing value", :integration => true do
+  it_behaves_like 'a routing indexer' do
+    let(:routing) { "test" }
+    let(:config) {
+      {
+        "hosts" => get_host_port,
+        "index" => index,
+        "routing" => "%{message}"
+      }
+    }
+  end
+end