MODULES: Fixes for bad module specified and elasticsearch client logging silence (#7367)

* Fixes for bad module specified and elasticsearch client logging silence & iterate over the intersection of specified and available modules

logstash-core/lib/logstash/config/modules_common.rb

@@ -13,31 +13,57 @@ module LogStash module Config
       plugin_modules = LogStash::PLUGIN_REGISTRY.plugins_with_type(:modules)
 
       modules_array = settings.get("modules.cli").empty? ? settings.get("modules") : settings.get("modules.cli")
-      logger.debug("Configured modules", :modules_array => modules_array.to_s)
-      module_names = []
+      if modules_array.empty?
+        # no specifed modules
+        return pipelines
+      end
+      logger.debug("Specified modules", :modules_array => modules_array.to_s)
+
       module_names = modules_array.collect {|module_hash| module_hash["name"]}
       if module_names.length > module_names.uniq.length
         duplicate_modules = module_names.group_by(&:to_s).select { |_,v| v.size > 1 }.keys
         raise LogStash::ConfigLoadingError, I18n.t("logstash.modules.configuration.modules-must-be-unique", :duplicate_modules => duplicate_modules)
       end
-      ### Here is where we can force the modules_array to use only [0] for 5.5, and leave
-      ### a warning/error message to that effect.
-      modules_array.each do |module_hash|
-        begin
-          import_engine = LogStash::Modules::Importer.new(LogStash::ElasticsearchClient.build(module_hash))
 
-          current_module = plugin_modules.find { |allmodules| allmodules.module_name == module_hash["name"] }
-          alt_name = "module-#{module_hash["name"]}"
+      available_module_names = plugin_modules.map(&:module_name)
+      specified_and_available_names = module_names & available_module_names
+
+      if (specified_and_available_names).empty?
+        i18n_opts = {:specified_modules => module_names, :available_modules => available_module_names}
+        raise LogStash::ConfigLoadingError, I18n.t("logstash.modules.configuration.modules-unavailable", i18n_opts)
+      end
+
+      specified_and_available_names.each do |module_name|
+        connect_fail_args = {}
+        begin
+          module_hash = modules_array.find {|m| m["name"] == module_name}
+          current_module = plugin_modules.find { |allmodules| allmodules.module_name == module_name }
+
+          alt_name = "module-#{module_name}"
           pipeline_id = alt_name
 
           current_module.with_settings(module_hash)
-          current_module.import(import_engine)
-          config_string = current_module.config_string
+          esclient = LogStash::ElasticsearchClient.build(module_hash)
+          config_test = settings.get("config.test_and_exit")
+          if esclient.can_connect? || config_test
+            if !config_test
+              current_module.import(LogStash::Modules::Importer.new(esclient))
+            end
 
-          pipelines << {"pipeline_id" => pipeline_id, "alt_name" => alt_name, "config_string" => config_string, "settings" => settings}
+            config_string = current_module.config_string
+
+            pipelines << {"pipeline_id" => pipeline_id, "alt_name" => alt_name, "config_string" => config_string, "settings" => settings}
+          else
+            connect_fail_args[:module_name] = module_name
+            connect_fail_args[:hosts] = esclient.host_settings
+          end
         rescue => e
           raise LogStash::ConfigLoadingError, I18n.t("logstash.modules.configuration.parse-failed", :error => e.message)
         end
+
+        if !connect_fail_args.empty?
+          raise LogStash::ConfigLoadingError, I18n.t("logstash.modules.configuration.elasticsearch_connection_failed", connect_fail_args)
+        end
       end
       pipelines
     end

logstash-core/lib/logstash/elasticsearch_client.rb

@@ -25,7 +25,22 @@ module LogStash class ElasticsearchClient
     def initialize(settings, logger)
       @settings = settings
       @logger = logger
-      @client = Elasticsearch::Client.new(client_args)
+      @client_args = client_args
+      @client = Elasticsearch::Client.new(@client_args)
+    end
+
+    def can_connect?
+      begin
+        head(SecureRandom.hex(32).prepend('_'))
+      rescue Elasticsearch::Transport::Transport::Errors::BadRequest
+        true
+      rescue Manticore::SocketException
+        false
+      end
+    end
+
+    def host_settings
+      @client_args[:hosts]
     end
 
     def delete(path)
@@ -70,7 +85,7 @@ module LogStash class ElasticsearchClient
       {
         :transport_class => Elasticsearch::Transport::Transport::HTTP::Manticore,
         :hosts => [*unpack_hosts],
-        :logger => @logger,
+        # :logger => @logger, # silence the client logging
       }
     end
 
@@ -94,4 +109,12 @@ module LogStash class ElasticsearchClient
   def head(path)
     @client.head(path)
   end
+
+  def can_connect?
+    @client.can_connect?
+  end
+
+  def host_settings
+    @client.host_settings
+  end
 end end # class LogStash::ModulesImporter

logstash-core/locales/en.yml

@@ -97,6 +97,14 @@ en:
         modules-variables-malformed: >-
           Failed to parse module variable %{rawvar}.  Must be in -M
           "MODULE_NAME.KEY.SUBKEY=VALUE" format
+        modules-unavailable: >-
+          The modules specified are not available yet.
+          Specified modules: %{specified_modules}
+          Available modules: %{available_modules}
+        elasticsearch_connection_failed: >-
+          Failed to import module configurations to Elasticsearch.
+          Module: %{module_name} has hosts: %{hosts}
+
     runner:
       short-help: |-
         usage:
@@ -117,7 +125,7 @@ en:
         Settings 'path.config' (-f) or 'config.string' (-e) can't be used in conjunction with
         (--modules) or the "modules:" block in the logstash.yml file.
       cli-module-override: >-
-        Both command-line and logstash.yml modules configurations detected. 
+        Both command-line and logstash.yml modules configurations detected.
         Using command-line module configuration and ignoring logstash.yml module
         configuration.
       reload-without-config-path: >-
@@ -209,21 +217,21 @@ en:
           the empty string for the '-e' flag.
         modules: |+
           Load Logstash modules.
-          Modules can be defined using multiple instances 
-          '--modules module1 --modules module2', 
-             or comma-separated syntax 
-          '--modules=module1,module2' 
+          Modules can be defined using multiple instances
+          '--modules module1 --modules module2',
+             or comma-separated syntax
+          '--modules=module1,module2'
           Cannot be used in conjunction with '-e' or '-f'
           Use of '--modules' will override modules declared
           in the 'logstash.yml' file.
         modules_variable: |+
           Load variables for module template.
-          Multiple instances of '-M' or 
+          Multiple instances of '-M' or
           '--modules.variable' are supported.
           Ignored if '--modules' flag is not used.
-          Should be in the format of 
+          Should be in the format of
           '-M "MODULE_NAME.var.PLUGIN_TYPE.PLUGIN_NAME.VARIABLE_NAME=VALUE"'
-          as in 
+          as in
           '-M "example.var.filter.mutate.fieldname=fieldvalue"'
         configtest: |+
           Check configuration for valid syntax and then exit.

logstash-core/spec/logstash/runner_spec.rb

@@ -7,8 +7,11 @@ require "stud/temporary"
 require "logstash/util/java_version"
 require "logstash/logging/json"
 require "logstash/config/source_loader"
+require "logstash/config/modules_common"
+require "logstash/elasticsearch_client"
 require "json"
 require_relative "../support/helpers"
+require_relative "../support/matchers"
 
 class NullRunner
   def run(args); end
@@ -126,6 +129,7 @@ describe LogStash::Runner do
     context "with a good configuration" do
       let(:pipeline_string) { "input { } filter { } output { }" }
       it "should exit successfully" do
+        expect(logger).not_to receive(:fatal)
         expect(subject.run(args)).to eq(0)
       end
     end
@@ -305,6 +309,102 @@ describe LogStash::Runner do
     end
   end
 
+  describe "logstash modules" do
+    describe "--config.test_and_exit" do
+      subject { LogStash::Runner.new("") }
+      let(:args) { ["-t", "--modules", module_string] }
+
+      context "with a good configuration" do
+        let(:module_string) { "cef" }
+        it "should exit successfully" do
+          expect(logger).not_to receive(:fatal)
+          expect(subject.run(args)).to eq(0)
+        end
+      end
+
+      context "with a bad configuration" do
+        let(:module_string) { "rlwekjhrewlqrkjh" }
+        it "should fail by returning a bad exit code" do
+          expect(logger).to receive(:fatal)
+          expect(subject.run(args)).to eq(1)
+        end
+      end
+    end
+
+    describe "--modules" do
+      let(:args) { ["--modules", module_string] }
+
+
+      context "with an available module specified but no connection to elasticsearch" do
+        let(:module_string) { "cef" }
+        before do
+          expect(logger).to receive(:fatal) do |msg, hash|
+            expect(msg).to eq("An unexpected error occurred!")
+            expect(hash).to be_a_config_loading_error_hash(
+              /Failed to import module configurations to Elasticsearch. Module: cef/)
+          end
+          expect(LogStash::Agent).to receive(:new) do |settings, source_loader|
+            pipelines = LogStash::Config::ModulesCommon.pipeline_configs(settings)
+            expect(pipelines).to eq([])
+            agent
+          end
+        end
+        it "should log fatally and return a bad exit code" do
+          expect(subject.run("bin/logstash", args)).to eq(1)
+        end
+      end
+
+      context "with an available module specified and a mocked connection to elasticsearch" do
+        let(:module_string) { "cef" }
+        let(:client) { double(:client) }
+        let(:response) { double(:response) }
+        before do
+          allow(response).to receive(:status).and_return(404)
+          allow(client).to receive(:head).and_return(response)
+          allow(client).to receive(:can_connect?).and_return(true)
+          allow(LogStash::ElasticsearchClient).to receive(:build).and_return(client)
+
+          expect(client).to receive(:put).at_least(15).times do |path, content|
+            LogStash::ElasticsearchClient::Response.new(201, "", {})
+          end
+          expect(LogStash::Agent).to receive(:new) do |settings, source_loader|
+            pipelines = LogStash::Config::ModulesCommon.pipeline_configs(settings)
+            expect(pipelines).not_to be_empty
+            cef_pipeline = pipelines.first
+            expect(cef_pipeline).to include("pipeline_id", "config_string")
+            expect(cef_pipeline["pipeline_id"]).to include('cef')
+            expect(cef_pipeline["config_string"]).to include('index => "cef-')
+            agent
+          end
+          expect(logger).not_to receive(:fatal)
+          expect(logger).not_to receive(:error)
+        end
+        it "should not terminate logstash" do
+          expect(subject.run("bin/logstash", args)).to be_nil
+        end
+      end
+
+      context "with an unavailable module specified" do
+        let(:module_string) { "fancypants" }
+        before do
+          expect(logger).to receive(:fatal) do |msg, hash|
+            expect(msg).to eq("An unexpected error occurred!")
+            expect(hash).to be_a_config_loading_error_hash(
+              /The modules specified are not available yet. Specified modules: \["fancypants"\] Available modules:/)
+          end
+          expect(LogStash::Agent).to receive(:new) do |settings, source_loader|
+            pipelines = LogStash::Config::ModulesCommon.pipeline_configs(settings)
+            expect(pipelines).to eq([])
+            agent
+          end
+        end
+        it "should log fatally and return a bad exit code" do
+          expect(subject.run("bin/logstash", args)).to eq(1)
+        end
+      end
+    end
+  end
+
   describe "--log.level" do
     before :each do
       allow_any_instance_of(subject).to receive(:show_version)

logstash-core/spec/support/matchers.rb

@@ -131,3 +131,17 @@ RSpec::Matchers.define :be_a_source_with_metadata do |protocol, id, text = nil|
    expect(actual.text).to match(text) unless text.nil?
   end
 end
+
+RSpec::Matchers.define :be_a_config_loading_error_hash do |regex|
+  match do |hash|
+    expect(hash).to include(:error)
+    error = hash[:error]
+    expect(error).to be_a(LogStash::ConfigLoadingError)
+    expect(error.message).to match(regex)
+  end
+
+  match_when_negated do
+    raise "Not implemented"
+  end
+end
+