Fix: hosts => "es_host:port" regression (when ssl => true) (#156)

.ci/Dockerfile.elasticsearch

@@ -3,5 +3,19 @@ FROM docker.elastic.co/elasticsearch/elasticsearch$distribution_suffix:$ELASTIC_
 
 ARG es_path=/usr/share/elasticsearch
 ARG es_yml=$es_path/config/elasticsearch.yml
+ARG SECURE_INTEGRATION
+ARG ES_SSL_SUPPORTED_PROTOCOLS
 
 RUN rm -f $es_path/config/scripts
+
+COPY --chown=elasticsearch:elasticsearch spec/filters/fixtures/test_certs/* $es_path/config/test_certs/
+
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then echo "xpack.security.http.ssl.enabled: true" >> $es_yml; fi
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then echo "xpack.security.http.ssl.key: $es_path/config/test_certs/es.key" >> $es_yml; fi
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then echo "xpack.security.http.ssl.certificate: $es_path/config/test_certs/es.crt" >> $es_yml; fi
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then echo "xpack.security.http.ssl.certificate_authorities: [ '$es_path/config/test_certs/ca.crt' ]" >> $es_yml; fi
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then echo "xpack.security.http.ssl.verification_mode: certificate" >> $es_yml; fi
+
+RUN cat $es_yml
+
+RUN if [ "$SECURE_INTEGRATION" = "true" ] ; then $es_path/bin/elasticsearch-users useradd tests -p Tests123 -r superuser; fi

.ci/docker-compose.override.yml

@@ -5,11 +5,12 @@ services:
   logstash:
     command: /usr/share/plugins/plugin/.ci/logstash-run.sh
     environment:
+        - ELASTIC_STACK_VERSION=$ELASTIC_STACK_VERSION
         - INTEGRATION=${INTEGRATION:-false}
         - SECURE_INTEGRATION=${SECURE_INTEGRATION:-false}
-        - ELASTIC_STACK_VERSION=$ELASTIC_STACK_VERSION
         - ELASTIC_SECURITY_ENABLED=$ELASTIC_SECURITY_ENABLED
         - ELASTIC_PASSWORD=$ELASTIC_PASSWORD
+
   elasticsearch:
     build:
       context: ../

.ci/logstash-run.sh

@@ -3,11 +3,16 @@ set -ex
 
 export PATH=$BUILD_DIR/gradle/bin:$PATH
 
+CURL_OPTS="-k --tlsv1.2"
+
 wait_for_es() {
   echo "Waiting for elasticsearch to respond..."
   es_url="http://elasticsearch:9200"
+  if [[ "$SECURE_INTEGRATION" == "true" ]]; then
+    es_url="https://elasticsearch:9200"
+  fi
   count=120
-  while ! curl -u elastic:$ELASTIC_PASSWORD --silent $es_url && [[ $count -ne 0 ]]; do
+  while ! curl $CURL_OPTS -u elastic:$ELASTIC_PASSWORD --silent $es_url && [[ $count -ne 0 ]]; do
     count=$(( $count - 1 ))
     [[ $count -eq 0 ]] && return 1
     sleep 1

.travis.yml

@@ -3,8 +3,11 @@ import:
 
 env:
   - INTEGRATION=false ELASTIC_STACK_VERSION=7.x
+  - INTEGRATION=false ELASTIC_STACK_VERSION=8.x
+  - INTEGRATION=false ELASTIC_STACK_VERSION=7.16.3 MANTICORE_VERSION=0.7.1 ELASTICSEARCH_VERSION=7.15.0
   - INTEGRATION=true ELASTIC_STACK_VERSION=7.x
   - INTEGRATION=true ELASTIC_STACK_VERSION=7.x ELASTIC_PASSWORD=password ELASTIC_SECURITY_ENABLED=true
-  - INTEGRATION=true ELASTIC_STACK_VERSION=7.x SNAPSHOT=true
-  - INTEGRATION=true ELASTIC_STACK_VERSION=8.x SNAPSHOT=true
-  - INTEGRATION=true ELASTIC_STACK_VERSION=7.14.0 ELASTIC_PASSWORD=password ELASTIC_SECURITY_ENABLED=true
+  - SECURE_INTEGRATION=true INTEGRATION=true ELASTIC_STACK_VERSION=7.x ELASTIC_SECURITY_ENABLED=true LOG_LEVEL=info
+  - SECURE_INTEGRATION=true INTEGRATION=true ELASTIC_STACK_VERSION=7.14.2 MANTICORE_VERSION=0.7.1 ELASTICSEARCH_VERSION=7.14.1 ELASTIC_SECURITY_ENABLED=true LOG_LEVEL=info
+  - INTEGRATION=true ELASTIC_STACK_VERSION=7.x SNAPSHOT=true LOG_LEVEL=info
+  - INTEGRATION=true ELASTIC_STACK_VERSION=8.x SNAPSHOT=true LOG_LEVEL=info

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.11.1
+  -  Fix: hosts => "es_host:port" regression [#156](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/156)
+
 ## 3.11.0
   - Feat: update Elasticsearch client to 7.14.0 [#150](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/150)
 

Gemfile

@@ -9,3 +9,6 @@ if Dir.exist?(logstash_path) && use_logstash_source
   gem 'logstash-core', :path => "#{logstash_path}/logstash-core"
   gem 'logstash-core-plugin-api', :path => "#{logstash_path}/logstash-core-plugin-api"
 end
+
+gem 'manticore', ENV['MANTICORE_VERSION'] if ENV['MANTICORE_VERSION']
+gem 'elasticsearch', ENV['ELASTICSEARCH_VERSION'] if ENV['ELASTICSEARCH_VERSION']

lib/logstash/filters/elasticsearch/client.rb

@@ -26,7 +26,7 @@ def initialize(logger, hosts, options = {})
         logger.warn "Supplied proxy setting (proxy => '') has no effect" if @proxy.eql?('')
         transport_options[:proxy] = proxy.to_s if proxy && !proxy.eql?('')
 
-        hosts = hosts.map { |host| { host: host, scheme: 'https' } } if ssl
+        hosts = setup_hosts(hosts, 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 ||= {}
@@ -41,6 +41,17 @@ def search(params)
 
       private
 
+      def setup_hosts(hosts, ssl)
+        hosts.map do |h|
+          if h.start_with?('http:/', 'https:/')
+            h
+          else
+            host, port = h.split(':')
+            { host: host, port: port, scheme: (ssl ? 'https' : 'http') }
+          end
+        end
+      end
+
       def setup_basic_auth(user, password)
         return {} unless user && password && password.value
 

logstash-filter-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-filter-elasticsearch'
-  s.version         = '3.11.0'
+  s.version         = '3.11.1'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Copies fields from previous log events in Elasticsearch to current events "
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/es_helper.rb

@@ -7,12 +7,26 @@ def self.get_host_port
     end
   end
 
-  def self.get_client(credentials)
-    require 'elasticsearch/transport/transport/http/faraday' # supports user/password options
-    host, port = get_host_port.split(':')
-    host_opts = credentials.inject({}) { |h, (k, v)| h[k.to_sym] = v; h } # user: _, password: _
-    host_opts.merge! host: host, port: port, scheme: 'http'
-    Elasticsearch::Client.new(hosts: [host_opts], transport_class: Elasticsearch::Transport::Transport::HTTP::Faraday)
+  def self.curl_and_get_json_response(url, method: :get, args: nil); require 'open3'
+    cmd = "curl -s -v --show-error #{args} -X #{method.to_s.upcase} -k #{url}"
+    begin
+      out, err, status = Open3.capture3(cmd)
+    rescue Errno::ENOENT
+      fail "curl not available, make sure curl binary is installed and available on $PATH"
+    end
+
+    if status.success?
+      http_status = err.match(/< HTTP\/1.1 (.*?)/)[1] || '0' # < HTTP/1.1 200 OK\r\n
+      if http_status.strip[0].to_i > 2
+        warn out
+        fail "#{cmd.inspect} unexpected response: #{http_status}\n\n#{err}"
+      end
+
+      LogStash::Json.load(out)
+    else
+      warn out
+      fail "#{cmd.inspect} process failed: #{status}\n\n#{err}"
+    end
   end
 
   def self.doc_type
@@ -25,12 +39,6 @@ def self.doc_type
     end
   end
 
-  def self.index_doc(es, params)
-    type = doc_type
-    params[:type] = doc_type unless type.nil?
-    es.index(params)
-  end
-
   def self.es_version
     ENV['ES_VERSION'] || ENV['ELASTIC_STACK_VERSION']
   end

spec/filters/elasticsearch_spec.rb

@@ -9,6 +9,10 @@
 
 describe LogStash::Filters::Elasticsearch do
 
+  subject(:plugin) { described_class.new(config) }
+
+  let(:event)  { LogStash::Event.new({}) }
+
   context "registration" do
 
     let(:plugin) { LogStash::Plugin.lookup("filter", "elasticsearch").new({}) }
@@ -53,8 +57,6 @@
         "aggregation_fields" => { "bytes_avg" => "bytes_avg_ls_field" }
       }
     end
-    let(:plugin) { described_class.new(config) }
-    let(:event)  { LogStash::Event.new({}) }
 
     let(:response) do
       LogStash::Json.load(File.read(File.join(File.dirname(__FILE__), "fixtures", "request_x_1.json")))
@@ -569,7 +571,9 @@ def wait_receive_request
     it "should set localhost:9200 as hosts" do
       plugin.register
       client = plugin.send(:get_client).client
-      expect( extract_transport(client).hosts ).to eql [{ :host => "localhost", :port => 9200, :protocol => "http"}]
+      hosts = extract_transport(client).hosts
+      expect( hosts.size ).to be 1
+      expect( hosts[0] ).to include(:host => "localhost", :port => 9200, :scheme => "http")
     end
   end
 

spec/filters/fixtures/test_certs/ca.crt

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDSTCCAjGgAwIBAgIUUcAg9c8B8jiliCkOEJyqoAHrmccwDQYJKoZIhvcNAQEL
+BQAwNDEyMDAGA1UEAxMpRWxhc3RpYyBDZXJ0aWZpY2F0ZSBUb29sIEF1dG9nZW5l
+cmF0ZWQgQ0EwHhcNMjEwODEyMDUxNDU1WhcNMjQwODExMDUxNDU1WjA0MTIwMAYD
+VQQDEylFbGFzdGljIENlcnRpZmljYXRlIFRvb2wgQXV0b2dlbmVyYXRlZCBDQTCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAK1HuusRuGNsztd4EQvqwcMr
+8XvnNNaalerpMOorCGySEFrNf0HxDIVMGMCrOv1F8SvlcGq3XANs2MJ4F2xhhLZr
+PpqVHx+QnSZ66lu5R89QVSuMh/dCMxhNBlOA/dDlvy+EJBl9H791UGy/ChhSgaBd
+OKVyGkhjErRTeMIq7rR7UG6GL/fV+JGy41UiLrm1KQP7/XVD9UzZfGq/hylFkTPe
+oox5BUxdxUdDZ2creOID+agtIYuJVIkelKPQ+ljBY3kWBRexqJQsvyNUs1gZpjpz
+YUCzuVcXDRuJXYQXGqWXhsBPfJv+ZcSyMIBUfWT/G13cWU1iwufPy0NjajowPZsC
+AwEAAaNTMFEwHQYDVR0OBBYEFMgkye5+2l+TE0I6RsXRHjGBwpBGMB8GA1UdIwQY
+MBaAFMgkye5+2l+TE0I6RsXRHjGBwpBGMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZI
+hvcNAQELBQADggEBAIgtJW8sy5lBpzPRHkmWSS/SCZIPsABW+cHqQ3e0udrI3CLB
+G9n7yqAPWOBTbdqC2GM8dvAS/Twx4Bub/lWr84dFCu+t0mQq4l5kpJMVRS0KKXPL
+DwJbUN3oPNYy4uPn5Xi+XY3BYFce5vwJUsqIxeAbIOxVTNx++k5DFnB0ESAM23QL
+sgUZl7xl3/DkdO4oHj30gmTRW9bjCJ6umnHIiO3JoJatrprurUIt80vHC4Ndft36
+NBQ9mZpequ4RYjpSZNLcVsxyFAYwEY4g8MvH0MoMo2RRLfehmMCzXnI/Wh2qEyYz
+emHprBii/5y1HieKXlX9CZRb5qEPHckDVXW3znw=
+-----END CERTIFICATE-----

spec/filters/fixtures/test_certs/ca.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEArUe66xG4Y2zO13gRC+rBwyvxe+c01pqV6ukw6isIbJIQWs1/
+QfEMhUwYwKs6/UXxK+VwardcA2zYwngXbGGEtms+mpUfH5CdJnrqW7lHz1BVK4yH
+90IzGE0GU4D90OW/L4QkGX0fv3VQbL8KGFKBoF04pXIaSGMStFN4wirutHtQboYv
+99X4kbLjVSIuubUpA/v9dUP1TNl8ar+HKUWRM96ijHkFTF3FR0NnZyt44gP5qC0h
+i4lUiR6Uo9D6WMFjeRYFF7GolCy/I1SzWBmmOnNhQLO5VxcNG4ldhBcapZeGwE98
+m/5lxLIwgFR9ZP8bXdxZTWLC58/LQ2NqOjA9mwIDAQABAoIBABmBC0P6Ebegljkk
+lO26GdbOKvbfqulDS3mN5QMyXkUMopea03YzMnKUJriE+2O33a1mUcuDPWnLpYPK
+BTiQieYHlulNtY0Bzf+R69igRq9+1WpZftGnzrlu7NVxkOokRqWJv3546ilV7QZ0
+f9ngmu+tiN7hEnlBC8m613VMuGGb3czwbCizEVZxlZX0Dk2GExbH7Yf3NNs/aOP/
+8x6CqgL+rhrtOQ80xwRrOlEF8oSSjXCzypa3nFv21YO3J2lVo4BoIwnHgOzyz46A
+b37gekqXXajIYQ0HAB+NDgVoCRFFJ7Xe16mgB3DpyUpUJzwiMedJkeQ0TprIownQ
++1mPe9ECgYEA/K4jc0trr3sk8KtcZjOYdpvwrhEqSSGEPeGfFujZaKOb8PZ8PX6j
+MbCTV12nEgm8FEhZQ3azxLnO17gbJ2A+Ksm/IIwnTWlqvvMZD5qTQ7L3qZuCtbWQ
++EGC/H1SDjhiwvjHcXP61/tYL/peApBSoj0L4kC+U/VaNyvicudKk08CgYEAr46J
+4VJBJfZ4ZaUBRy53+fy+mknOfaj2wo8MnD3u+/x4YWTapqvDOPN2nJVtKlIsxbS4
+qCO+fzUV17YHlsQmGULNbtFuXWJkP/RcLVbe8VYg/6tmk0dJwNAe90flagX2KJov
+8eDX129nNpuUqrNNWsfeLmPmH6vUzpKlga+1zfUCgYBrbUHHJ96dmbZn2AMNtIvy
+iXP3HXcj5msJwB3aKJ8eHMkU1kaWAnwxiQfrkfaQ9bCP0v6YbyQY1IJ7NlvdDs7/
+dAydMtkW0WW/zyztdGN92d3vrx0QUiRTV87vt/wl7ZUXnZt1wcB5CPRCWaiUYHWx
+YlDmHW6N1XdIk5DQF0OegwKBgEt7S8k3Zo9+A5IgegYy8p7njsQjy8a3qTFJ9DAR
+aPmrOc8WX/SdkVihRXRZwxAZOOrgoyyYAcYL+xI+T9EBESh3UoC9R2ibb2MYG7Ha
+0gyN7a4/8eCNHCbs1QOZRAhr+8TFVqv28pbMbWJLToZ+hVns6Zikl0MyzFLtNoAm
+HlMpAoGBAIOkqnwwuRKhWprL59sdcJfWY26os9nvuDV4LoKFNEFLJhj2AA2/3UlV
+v85gqNSxnMNlHLZC9l2HZ3mKv/mfx1aikmFvyhJAnk5u0f9KkexmCPLjQzS5q3ba
+yFuxK2DXwN4x46RgQPFlLjOTCX0BG6rkEu4JdonF8ETSjoCtGEU8
+-----END RSA PRIVATE KEY-----

spec/filters/fixtures/test_certs/es.crt

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDNjCCAh6gAwIBAgIUF9wE+oqGSbm4UVn1y9gEjzyaJFswDQYJKoZIhvcNAQEL
+BQAwNDEyMDAGA1UEAxMpRWxhc3RpYyBDZXJ0aWZpY2F0ZSBUb29sIEF1dG9nZW5l
+cmF0ZWQgQ0EwHhcNMjEwODEyMDUxNTI3WhcNMjQwODExMDUxNTI3WjANMQswCQYD
+VQQDEwJlczCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAK2S2by0lgyu
+1JfgGgZ41PNXbH2qMPMzowguVVdtZ16WM0CaEG7lnLxmMcC+2Q7NnGuFnPAVQo9T
+Q3bh7j+1PkCJVHUKZfJIeWtGc9+qXBcO1MhedfwM1osSa4bfwM85G+XKWbRNtmSt
+CoUuKArIyZkzdBAAQLBoQyPf3DIza1Au4j9Hb3zrswD6e7n2PN4ffIyil1GFduLJ
+2275qqFiOhkEDUhv7BKNftVBh/89O/5lSqAQGuQ1aDRr8TdHwhO71u4ZIU/Pn6yX
+LGBWrQG53+qpdCsxGvJTfbtIEYUDTN83CirIxDKJgc1QXOEldylztHf4xnQ7ZarJ
+tqF6pUzHbRsCAwEAAaNnMGUwHQYDVR0OBBYEFFQUK+6Cg2kExRj1xSDzEi4kkgKX
+MB8GA1UdIwQYMBaAFMgkye5+2l+TE0I6RsXRHjGBwpBGMBgGA1UdEQQRMA+CDWVs
+YXN0aWNzZWFyY2gwCQYDVR0TBAIwADANBgkqhkiG9w0BAQsFAAOCAQEAinaknZIc
+7xtQNwUwa+kdET+I4lMz+TJw9vTjGKPJqe082n81ycKU5b+a/OndG90z+dTwhShW
+f0oZdIe/1rDCdiRU4ceCZA4ybKrFDIbW8gOKZOx9rsgEx9XNELj4ocZTBqxjQmNE
+Ho91fli5aEm0EL2vJgejh4hcfDeElQ6go9gtvAHQ57XEADQSenvt69jOICOupnS+
+LSjDVhv/VLi3CAip0B+lD5fX/DVQdrJ62eRGuQYxoouE3saCO58qUUrKB39yD9KA
+qRA/sVxyLogxaU+5dLfc0NJdOqSzStxQ2vdMvAWo9tZZ2UBGFrk5SdwCQe7Yv5mX
+qi02i4q6meHGcw==
+-----END CERTIFICATE-----

spec/filters/fixtures/test_certs/es.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEArZLZvLSWDK7Ul+AaBnjU81dsfaow8zOjCC5VV21nXpYzQJoQ
+buWcvGYxwL7ZDs2ca4Wc8BVCj1NDduHuP7U+QIlUdQpl8kh5a0Zz36pcFw7UyF51
+/AzWixJrht/Azzkb5cpZtE22ZK0KhS4oCsjJmTN0EABAsGhDI9/cMjNrUC7iP0dv
+fOuzAPp7ufY83h98jKKXUYV24snbbvmqoWI6GQQNSG/sEo1+1UGH/z07/mVKoBAa
+5DVoNGvxN0fCE7vW7hkhT8+frJcsYFatAbnf6ql0KzEa8lN9u0gRhQNM3zcKKsjE
+MomBzVBc4SV3KXO0d/jGdDtlqsm2oXqlTMdtGwIDAQABAoIBAQCm/VBDz41ImG7p
+yu3e6iMeFi7HW5SKdlRUS5dJbHT1uBWJAm/q8TbwvnUBVdsn9cKWY06QYDPQBjAy
+0LxRSIKivjyl+aIJDZbbEUXrmk/M0zT9rHtgSc2isM8ITH6IHw5q7lmNMPLYOu6T
+IMvfTDtADBOOTV/vF+/4NKf5GCUXVt1XTzLBFMK0p/ZoI7Fsw7fhH6FR12vk0xA4
+BEC4pwRbGfHo7P31ii0by8epkve93tF4IZuFmN92A84bN1z7Kc4TYaSbua2rgguz
+FzMyWpsTxr363HzCK1xOJb6JyJOiXbq4+j2oqtne3GIvyozJeiyKRgjLIMoe/LV7
+fPPc5wlhAoGBAOD3z0JH2eyR/1RHILFsWInH2nDbKHHuCjhFIL2XloeXsJkiJZ95
+BpdjExMZCqD44tPNRW/GgWKwoVwltm6zB0aq0aW/OfOzw6fhKt1W+go47L7Tpwap
+VQgy6BFXSueUKfQDlZEWV4E2gakf8vOl0/VRQExae/CeKf1suEedQaErAoGBAMWE
+LOmNDEU2NFqghfNBAFYyFJst3YnBmSmlL7W22+OsfSK/PhxnJbuNHxMgxpg9rieW
+tVyjuZRo/i7WLVm3uG+dK1RJ9t8Y6kpYkCRKpi9G8DBOj3PSulOybBr+fdRfW9mf
+8UmqOjOkrhxXPkchc9TY4EM7/1XeKvEidlIp0gvRAoGAAurz4zYvW2QhXaR2hhaT
+p2XSLXiKM8AUndo3rH3U0/lhrvrEZicZsMj2LF88xg20U27sIaD/eJo13Y4XqaPk
+ykPY6D9srv574SeIeMpx/8PxPiBcoDd+BNc0L1VkgVBoouORAwq5I9HjKKBjdEmI
+UDw3i0X5KYvDm6fXVAZ0HXUCgYBWc4To8KiXPqNpq2sVzrSkBaWJSmj2G7u7Q6b/
+RTs3is72v3gjHG6iiaE5URY7mnu4rjlRhAP9Vnsy6uHMrCJZEBTf/sPEYHZj9iGZ
+EOduOAF3U1tsmaaebbDtm8hdhSOBvITy9kQlSIZAt1r17Ulytz5pj0AySFzJUIkz
+a0SZkQKBgCWixtUxiK8PAdWhyS++90WJeJn8eqjuSAz+VMtFQFRRWDUbkiHvGMRu
+o/Hhk6zS46gSF2Evb1d26uUEenXnJlIp6YWzb0DLPrfy5P53kPA6YEvYq5MSAg3l
+DZOJUF+ko7cWXSZkeTIBH/jrGOdP4tTALZt6DNt+Gz7xwPO5tGgV
+-----END RSA PRIVATE KEY-----

spec/filters/integration/elasticsearch_spec.rb

@@ -7,39 +7,53 @@
 describe LogStash::Filters::Elasticsearch, :integration => true do
 
   ELASTIC_SECURITY_ENABLED = ENV['ELASTIC_SECURITY_ENABLED'].eql? 'true'
+  SECURE_INTEGRATION = ENV['SECURE_INTEGRATION'].eql? 'true'
 
   let(:base_config) do
     {
         "index" => 'logs',
-        "hosts" => [ESHelper.get_host_port],
+        "hosts" => ["http#{SECURE_INTEGRATION ? 's' : nil}://#{ESHelper.get_host_port}"],
         "query" => "response: 404",
         "sort" => "response",
         "fields" => [ ["response", "code"] ],
     }
   end
 
   let(:credentials) do
-    { 'user' => 'elastic', 'password' => ENV['ELASTIC_PASSWORD'] }
+    if SECURE_INTEGRATION
+      { 'user' => 'tests', 'password' => 'Tests123' } # added user
+    else
+      { 'user' => 'elastic', 'password' => ENV['ELASTIC_PASSWORD'] }
+    end
   end
 
   let(:config) do
-    ELASTIC_SECURITY_ENABLED ? base_config.merge(credentials) : base_config
+    config = ELASTIC_SECURITY_ENABLED ? base_config.merge(credentials) : base_config
+    config = { 'ca_file' => ca_path }.merge(config) if SECURE_INTEGRATION
+    config
+  end
+
+  let(:ca_path) do
+    File.expand_path('../fixtures/test_certs/ca.crt', File.dirname(__FILE__))
   end
 
   let(:plugin) { described_class.new(config) }
   let(:event)  { LogStash::Event.new({}) }
 
   before(:each) do
-    @es = ESHelper.get_client(ELASTIC_SECURITY_ENABLED ? credentials : {})
-    # Delete all templates first.
+    es_url = ESHelper.get_host_port
+    es_url = SECURE_INTEGRATION ? "https://#{es_url}" : "http://#{es_url}"
+    args = ELASTIC_SECURITY_ENABLED ? "-u #{credentials['user']}:#{credentials['password']}" : ''
     # Clean ES of data before we start.
-    @es.indices.delete_template(:name => "*")
+    # Delete all templates first.
+    ESHelper.curl_and_get_json_response "#{es_url}/_index_template/*", method: 'DELETE', args: args
     # This can fail if there are no indexes, ignore failure.
-    @es.indices.delete(:index => "*") rescue nil
+    ESHelper.curl_and_get_json_response "#{es_url}/_index/*", method: 'DELETE', args: args
+    doc_args = "#{args} -H 'Content-Type: application/json' -d '{\"response\": 404, \"this\":\"that\"}'"
     10.times do
-      ESHelper.index_doc(@es, :index => 'logs', :body => { :response => 404, :this => 'that'})
+      ESHelper.curl_and_get_json_response "#{es_url}/logs/_doc", method: 'POST', args: doc_args
     end
-    @es.indices.refresh
+    ESHelper.curl_and_get_json_response "#{es_url}/_refresh", method: 'POST', args: args
   end
 
   it "should enhance the current event with new data" do
@@ -69,10 +83,23 @@
       super().reject { |key, _| key == 'password' }
     end
 
-    it "should enhance the current event with new data" do
+    it "fails to register plugin" do
       expect { plugin.register }.to raise_error Elasticsearch::Transport::Transport::Errors::Unauthorized
     end
 
   end if ELASTIC_SECURITY_ENABLED
 
+  context 'setting host:port (and ssl)' do # reproduces GH-155
+
+    let(:config) do
+      super().merge "hosts" => [ESHelper.get_host_port], "ssl" => SECURE_INTEGRATION
+    end
+
+    it "works" do
+      expect { plugin.register }.to_not raise_error
+      plugin.filter(event)
+    end
+
+  end
+
 end