normalise hosts to URI::Generics, which reliably preserve defaults

An upstream bug in the Elasticsearch Ruby Client's handling of `String` host
arguments that begin with a schema (e.g., `https://localhost`) causes it to
default to port 80 or 443, depending on the schema, instead of Elasticsearch's
port 9200.

Since the Elasticsearch Ruby Client will accept a `URI` in this case, and will
correctly handle falling through to appropriate defaults, we normalise to
`URI::Generic`, which does not have a default port.

We absorb the `ssl => true` case into this normalisation, as its previous
implementation prevented the use of non-default ports in the array provided
to `hosts`.

We also add support for IPv6 addresses, requiring a square-bracketed notation
(see: RFC2732)

Supersedes: #104
Resolves:   #110
Resolves:   #111

Author: yaauie

docs/index.asciidoc

@@ -221,6 +221,11 @@ fields => {
 
   * Value type is <<array,array>>
   * Default value is `["localhost:9200"]`
+  * Format for each entry is one of:
+    - a bare ipv4-literal, optionally followed by a colon and port number
+    - a hostname, optionally followed by a colon and port number
+    - a square-bracketed ipv6-literal, optionally followed by a colon and port number
+    - any of the above can be prefixed with an explicit schema (e.g., `http://` or `https://`)
 
 List of elasticsearch hosts to use for querying.
 
@@ -281,7 +286,9 @@ Comma-delimited list of `<field>:<direction>` pairs that define the sort order
   * Value type is <<boolean,boolean>>
   * Default value is `false`
 
-SSL
+Force SSL/TLS secured communication to Elasticsearch cluster.
+Leaving this unspecified will use whatever scheme is specified in the URLs listed in <<plugins-{type}s-{plugin}-hosts>>, where mixed schemes are supported.
+If SSL is set to `true`, the plugin will refuse to start if any of the hosts specifies an `http://` scheme.
 
 [id="plugins-{type}s-{plugin}-tag_on_failure"]
 ===== `tag_on_failure` 

lib/logstash/filters/elasticsearch.rb

@@ -71,6 +71,8 @@ def register
       @query_dsl = file.read
     end
 
+    @normalised_hosts = normalise_hosts(@hosts, @ssl)
+
     test_connection!
   end # def register
 
@@ -140,8 +142,7 @@ def filter(event)
   private
   def client_options
     {
-      :ssl => @ssl,
-      :hosts => @hosts,
+      :hosts => @normalised_hosts,
       :ca_file => @ca_file,
       :logger => @logger
     }
@@ -191,4 +192,55 @@ def extract_total_from_hits(hits)
   def test_connection!
     get_client.client.ping
   end
+
+  private
+
+  ##
+  # Map the provided array-of-strings to an array of `URI::Generic`
+  # instances, which the Elasticsearch client can use to establish
+  # connections.
+  #
+  # @param hosts [Array<String>]: one or more hosts, each is one of:
+  #                               - a bare hostname or ip, optionally
+  #                                 followed by a colon and port number
+  #                               - a square-bracketed ipv6 literal,
+  #                                 optionally followed by a colon and port
+  #                                 number
+  #                               - a qualified URL with http/https schema
+  # @param force_ssl [Boolean]: true to force SSL; will cause failure if one
+  #                             or more hosts explicitly supplies non-SSL
+  #                             scheme (e.g., `http`).
+  #
+  # @return [Array<URI::Generic>]
+  def normalise_hosts(hosts, force_ssl)
+    hosts.map do |input|
+      if force_ssl && input.start_with?('http://')
+        logger.error("Plugin configured to force SSL with `ssl => true`, " +
+                     "but a host explicitly declared non-https URL `#{input}`")
+
+        raise LogStash::ConfigurationError, "Aborting due to conflicting configuration"
+      end
+
+      if input.start_with?('http://','https://')
+        URI::Generic.new(*URI.split(input))
+      else
+        if input =~ /^(\[[a-z0-9:]+\])(?::([0-9]+))?$/
+          host, port = Regexp::last_match.captures
+        else
+          host, port = input.split(':')
+        end
+        URI::Generic.new(
+          force_ssl ? 'https' : 'http',
+          nil, # userinfo,
+          host,
+          port,
+          nil, # registry
+          nil, # path
+          nil, # opaque
+          nil, # query
+          nil  # fragment
+        )
+      end
+    end
+  end
 end #class LogStash::Filters::Elasticsearch

lib/logstash/filters/elasticsearch/client.rb

@@ -11,17 +11,15 @@ class ElasticsearchClient
       attr_reader :client
 
       def initialize(user, password, options={})
-        ssl     = options.fetch(:ssl, false)
-        hosts   = options[:hosts]
-        @logger = options[:logger]
+        hosts   = options.fetch(:hosts)
+        @logger = options.fetch(:logger)
 
         transport_options = {}
         if user && password
           token = ::Base64.strict_encode64("#{user}:#{password.value}")
           transport_options[:headers] = { Authorization: "Basic #{token}" }
         end
 
-        hosts.map! {|h| { host: h, scheme: 'https' } } if ssl
         # set ca_file even if ssl isn't on, since the host can be an https url
         ssl_options = { ssl: true, ca_file: options[:ca_file] } if options[:ca_file]
         ssl_options ||= {}

spec/filters/elasticsearch_spec.rb

@@ -4,17 +4,225 @@
 require "logstash/filters/elasticsearch"
 require "logstash/json"
 
+RSpec::Matchers.define(:hash_with_member) do |member, matcher=nil|
+  match do |actual|
+    expect(actual).to be_a Hash
+    expect(actual).to include member
+
+    value = actual[member]
+
+    block_arg && block_arg.call(value)
+    
+    matcher.nil? || matcher.matches?(value)
+  end
+
+  description do
+    desc = "hash with member `#{member.inspect}`"
+    caveats = []
+    caveats << 'the provided block' unless block_arg.nil?
+    caveats << matcher.description unless matcher.nil?
+    
+    desc += ' satisfying ' unless caveats.empty?
+    desc += caveats.join(' and ')
+    desc
+  end
+end
+
 describe LogStash::Filters::Elasticsearch do
 
   context "registration" do
 
-    let(:plugin) { LogStash::Plugin.lookup("filter", "elasticsearch").new({}) }
-    before do
-      allow(plugin).to receive(:test_connection!)
+    let(:plugin_class) { LogStash::Plugin.lookup("filter", "elasticsearch") }
+    let(:plugin) { plugin_class.new(config) }
+    let(:config) { Hash.new }
+
+    context 'with defaults' do
+      before do
+        allow(plugin).to receive(:test_connection!)
+      end
+
+      it "should not raise an exception" do
+        expect {plugin.register}.to_not raise_error
+      end
     end
 
-    it "should not raise an exception" do
-      expect {plugin.register}.to_not raise_error
+    context 'hosts' do
+      let(:config) do
+        super().merge(
+            'hosts' => hosts
+          )
+      end
+      let(:hosts) do
+        fail NotImplementedError, 'spec or spec group must define `hosts`.'
+      end
+
+      let(:client_stub) { double(:client).as_null_object }
+      let(:logger_stub) { double(:logger).as_null_object }
+
+      before(:each) do
+        allow(plugin).to receive(:logger).and_return(logger_stub)
+      end
+
+      context 'with schema://hostname' do
+        let(:hosts) { ['http://foo.local', 'http://bar.local'] }
+
+        it 'creates client with URIs that do not include a port' do
+          expect(::Elasticsearch::Client).to receive(:new) do |options|
+            expect(options).to include :hosts
+            expect(options[:hosts]).to be_an Array
+            expect(options[:hosts]).to include(having_attributes(host: 'foo.local', scheme: 'http', port: nil))
+            expect(options[:hosts]).to include(having_attributes(host: 'bar.local', scheme: 'http', port: nil))
+          end.and_return(client_stub)
+
+          plugin.register
+        end
+      end
+
+      context 'with `ssl => true`' do
+        let(:config) { super().merge('ssl' => 'true') }
+        context 'and one or more explicitly-http hosts' do
+          let(:hosts) { ['https://foo.local', 'http://bar.local'] }
+
+          it 'raises an exception' do
+            expect { plugin.register }.to raise_error(LogStash::ConfigurationError)
+          end
+
+          it 'emits a helpful log message' do
+            plugin.register rescue nil
+            expect(plugin.logger).to have_received(:error).with(match(/force SSL/))
+          end
+        end
+
+        context 'and all explicitly-https hosts' do
+          let(:hosts) { ['https://foo.local', 'https://bar.local'] }
+
+          it 'sets the schemas on all to https' do
+            expect(::Elasticsearch::Client).to receive(:new) do |options|
+              expect(options).to include :hosts
+              expect(options[:hosts]).to be_an Array
+              options[:hosts].each do |host|
+                expect(host).to be_an URI
+                expect(host.scheme).to eq 'https'
+              end
+            end.and_return(client_stub)
+
+            plugin.register
+          end
+        end
+
+        context 'and one or more schemaless hosts' do
+          let(:hosts) { ['https://foo.local', 'bar.local'] }
+
+          it 'sets the schemas on all to https' do
+            expect(::Elasticsearch::Client).to receive(:new) do |options|
+              expect(options).to include :hosts
+              expect(options[:hosts]).to be_an Array
+              options[:hosts].each do |host|
+                expect(host).to be_an URI
+                expect(host.scheme).to eq 'https'
+              end
+            end.and_return(client_stub)
+
+            plugin.register
+          end
+        end
+
+        context 'with one or more square-bracketed ipv6 literals' do
+            let(:hosts) { ['[::1]', '[::2]:9201', 'https://[::3]:9202'] }
+            it 'defaults to the http protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: '[::1]', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: '[::2]', port: 9201))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: '[::3]', port: 9202))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+      end
+
+      {
+        'with `ssl => false' => {'ssl' => 'false'},
+        'without `ssl` directive' => {}
+      }.each do |context_string, config_override|
+        context(context_string) do
+          let(:config) { super().merge(config_override) }
+
+          context 'with a mix of http and https hosts' do
+            let(:hosts) { ['https://foo.local', 'http://bar.local'] }
+            it 'does not modify the protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: 'foo.local', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http',  host: 'bar.local', port: nil))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+
+          context 'with https-only hosts' do
+            let(:hosts) { ['https://foo.local', 'https://bar.local'] }
+            it 'does not modify the protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: 'foo.local', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: 'bar.local', port: nil))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+
+          context 'with http-only hosts' do
+            let(:hosts) { ['http://foo.local', 'http://bar.local'] }
+            it 'does not modify the protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http', host: 'foo.local', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http', host: 'bar.local', port: nil))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+
+          context 'with one or more schemaless hosts' do
+            let(:hosts) { ['foo.local', 'bar.local' ] }
+            it 'defaults to the http protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http', host: 'foo.local', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http', host: 'bar.local', port: nil))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+
+          context 'with one or more square-bracketed ipv6 literals' do
+            let(:hosts) { ['[::1]', '[::2]:9201', 'http://[::3]','https://[::4]:9202'] }
+            it 'defaults to the http protocol' do
+              expect(::Elasticsearch::Client).to receive(:new) do |options|
+                expect(options).to include :hosts
+                expect(options[:hosts]).to be_an Array
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http',  host: '[::1]', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http',  host: '[::2]', port: 9201))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'http',  host: '[::3]', port: nil))
+                expect(options[:hosts]).to include(having_attributes(scheme: 'https', host: '[::4]', port: 9202))
+              end.and_return(client_stub)
+
+              plugin.register
+            end
+          end
+        end
+      end
     end
   end