diff --git a/docs/static/settings/configuration-management-settings.asciidoc b/docs/static/settings/configuration-management-settings.asciidoc
index 803354460..c83863d75 100644
--- a/docs/static/settings/configuration-management-settings.asciidoc
+++ b/docs/static/settings/configuration-management-settings.asciidoc
@@ -99,3 +99,10 @@ and `xpack.management.elasticsearch.password`. If `cloud_auth` is configured,
 those settings should not be used.
 The credentials you specify here should be for a user with the `logstash_admin` role, which
 provides access to `.logstash-*` indices for managing configurations.
+
+`xpack.management.elasticsearch.api_key`::
+
+Authenticate using an Elasticsearch API key. Note that this option also requires using SSL.
+
+The API key Format is `id:api_key` where `id` and `api_key` are as returned by the Elasticsearch
+https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key API].
diff --git a/docs/static/settings/monitoring-settings-legacy.asciidoc b/docs/static/settings/monitoring-settings-legacy.asciidoc
index cbab903db..72f836e29 100644
--- a/docs/static/settings/monitoring-settings-legacy.asciidoc
+++ b/docs/static/settings/monitoring-settings-legacy.asciidoc
@@ -104,3 +104,10 @@ If you're using {es} in {ecloud}, you can set your auth credentials here.
 This setting is an alternative to both `xpack.monitoring.elasticsearch.username`
 and `xpack.monitoring.elasticsearch.password`. If `cloud_auth` is configured,
 those settings should not be used.
+
+`xpack.monitoring.elasticsearch.api_key`::
+
+Authenticate using an Elasticsearch API key. Note that this option also requires using SSL.
+
+The API key Format is `id:api_key` where `id` and `api_key` are as returned by the Elasticsearch
+https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key API].
diff --git a/x-pack/lib/config_management/elasticsearch_source.rb b/x-pack/lib/config_management/elasticsearch_source.rb
index 6db650646..edc4af3b1 100644
--- a/x-pack/lib/config_management/elasticsearch_source.rb
+++ b/x-pack/lib/config_management/elasticsearch_source.rb
@@ -35,22 +35,9 @@ module LogStash
 
       def initialize(settings)
         super(settings)
-        if @settings.get("xpack.management.enabled")
-          if @settings.get_setting("xpack.management.elasticsearch.cloud_id").set?
-            if !@settings.get_setting("xpack.management.elasticsearch.cloud_auth").set?
-              raise ArgumentError.new("You must set credentials using \"xpack.management.elasticsearch.cloud_auth\", " +
-                                      "when using \"xpack.management.elasticsearch.cloud_id\" in logstash.yml")
-            end
-          else
-            if !@settings.get_setting("xpack.management.elasticsearch.password").set?
-              raise ArgumentError.new("You must set the password using \"xpack.management.elasticsearch.password\" in logstash.yml")
-            end
-          end
-        end
-
-        @es_options = es_options_from_settings('management', settings)
 
         if enabled?
+          @es_options = es_options_from_settings('management', settings)
           setup_license_checker(FEATURE_INTERNAL)
           license_check(true)
         end
diff --git a/x-pack/lib/config_management/extension.rb b/x-pack/lib/config_management/extension.rb
index 9c694efcb..80ca77408 100644
--- a/x-pack/lib/config_management/extension.rb
+++ b/x-pack/lib/config_management/extension.rb
@@ -29,6 +29,7 @@ module LogStash
         settings.register(LogStash::Setting::ArrayCoercible.new("xpack.management.elasticsearch.hosts", String, [ "https://localhost:9200" ] ))
         settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.cloud_id"))
         settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.cloud_auth"))
+        settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.api_key"))
         settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.proxy"))
         settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.ssl.certificate_authority"))
         settings.register(LogStash::Setting::NullableString.new("xpack.management.elasticsearch.ssl.truststore.path"))
diff --git a/x-pack/lib/helpers/elasticsearch_options.rb b/x-pack/lib/helpers/elasticsearch_options.rb
index f5d5b9ee9..26bd6d82a 100644
--- a/x-pack/lib/helpers/elasticsearch_options.rb
+++ b/x-pack/lib/helpers/elasticsearch_options.rb
@@ -6,17 +6,34 @@ module LogStash module Helpers
   module ElasticsearchOptions
     extend self
 
-    ES_SETTINGS =%w(
-        ssl.certificate_authority
-        ssl.truststore.path
-        ssl.keystore.path
-        hosts
-        username
-        password
-        cloud_id
-        cloud_auth
-        proxy
-      )
+    ES_SETTINGS = %w(
+      ssl.certificate_authority
+      ssl.truststore.path
+      ssl.keystore.path
+      hosts
+      username
+      password
+      cloud_id
+      cloud_auth
+      api_key
+      proxy
+    )
+
+    # xpack setting to ES output setting
+    SETTINGS_MAPPINGS = {
+      "cloud_id" => "cloud_id",
+      "cloud_auth" => "cloud_auth",
+      "username" => "user",
+      "password" => "password",
+      "api_key" => "api_key",
+      "proxy" => "proxy",
+      "sniffing" => "sniffing",
+      "ssl.certificate_authority" => "cacert",
+      "ssl.truststore.path" => "truststore",
+      "ssl.truststore.password" => "truststore_password",
+      "ssl.keystore.path" => "keystore",
+      "ssl.keystore.password" => "keystore_password",
+    }
 
     # Retrieve elasticsearch options from either specific settings, or modules if the setting is not there and the
     # feature supports falling back to modules if the feature is not specified in logstash.yml
@@ -27,53 +44,53 @@ module LogStash module Helpers
     # Populate the Elasticsearch options from LogStashSettings file, based on the feature that is being used.
     # @return Hash
     def es_options_from_settings(feature, settings)
-      prefix = if feature == "monitoring" &&
-                    LogStash::MonitoringExtension.use_direct_shipping?(settings)
-                 ""
-               else
-                 "xpack."
-               end
+      prefix = (feature == "monitoring" && LogStash::MonitoringExtension.use_direct_shipping?(settings)) ? "" : "xpack."
       opts = {}
 
-      if cloud_id = settings.get("#{prefix}#{feature}.elasticsearch.cloud_id")
-        opts['cloud_id'] = cloud_id
-        check_cloud_id_configuration!(feature, settings, prefix)
-      else
+      validate_authentication!(feature, settings, prefix)
+
+      # transpose all directly mappable settings
+      SETTINGS_MAPPINGS.each do |xpack_setting, es_setting|
+        v = settings.get("#{prefix}#{feature}.elasticsearch.#{xpack_setting}")
+        opts[es_setting] = v unless v.nil?
+      end
+
+      # process remaining settings
+
+      unless settings.get("#{prefix}#{feature}.elasticsearch.cloud_id")
         opts['hosts'] = settings.get("#{prefix}#{feature}.elasticsearch.hosts")
       end
-      if cloud_auth = settings.get("#{prefix}#{feature}.elasticsearch.cloud_auth")
-        opts['cloud_auth'] = cloud_auth
-        check_cloud_auth_configuration!(feature, settings, prefix)
-      else
-        opts['user'] = settings.get("#{prefix}#{feature}.elasticsearch.username")
-        opts['password'] = settings.get("#{prefix}#{feature}.elasticsearch.password")
-      end
-      if proxysetting = settings.get("#{prefix}#{feature}.elasticsearch.proxy")
-        opts['proxy'] = proxysetting
-      end
-
-      opts['sniffing'] = settings.get("#{prefix}#{feature}.elasticsearch.sniffing")
       opts['ssl_certificate_verification'] = settings.get("#{prefix}#{feature}.elasticsearch.ssl.verification_mode") == 'certificate'
 
-      if cacert = settings.get("#{prefix}#{feature}.elasticsearch.ssl.certificate_authority")
-        opts['cacert'] = cacert
+      # if all hosts are using https or any of the ssl related settings are set
+      if ssl?(feature, settings, prefix)
         opts['ssl'] = true
       end
 
-      if truststore = settings.get("#{prefix}#{feature}.elasticsearch.ssl.truststore.path")
-        opts['truststore'] = truststore
-        opts['truststore_password'] = settings.get("#{prefix}#{feature}.elasticsearch.ssl.truststore.password")
-        opts['ssl'] = true
+      # the username setting has a default value and should not be included when using another authentication
+      # it is safe to silently remove here since all authentication verifications have been validated at this point.
+      if settings.set?("#{prefix}#{feature}.elasticsearch.cloud_auth") || settings.set?("#{prefix}#{feature}.elasticsearch.api_key")
+        opts.delete('user')
       end
 
-      if keystore = settings.get("#{prefix}#{feature}.elasticsearch.ssl.keystore.path")
-        opts['keystore'] = keystore
-        opts['keystore_password']= settings.get("#{prefix}#{feature}.elasticsearch.ssl.keystore.password")
-        opts['ssl'] = true
-      end
       opts
     end
 
+    def ssl?(feature, settings, prefix)
+      return true if verify_https_scheme(feature, settings, prefix)
+      return true if settings.set?("#{prefix}#{feature}.elasticsearch.cloud_id") # cloud_id always resolves to https hosts
+      return true if settings.set?("#{prefix}#{feature}.elasticsearch.ssl.certificate_authority")
+      return true if settings.set?("#{prefix}#{feature}.elasticsearch.ssl.truststore.path") && settings.set?("#{prefix}#{feature}.elasticsearch.ssl.truststore.password")
+      return true if settings.set?("#{prefix}#{feature}.elasticsearch.ssl.keystore.path") && settings.set?("#{prefix}#{feature}.elasticsearch.ssl.keystore.password")
+
+      return false
+    end
+
+    HTTPS_SCHEME = /^https:\/\/.+/
+    def verify_https_scheme(feature, settings, prefix)
+      hosts = Array(settings.get("#{prefix}#{feature}.elasticsearch.hosts"))
+      hosts.all? {|host| host.match?(HTTPS_SCHEME)}
+    end
 
     # Elasticsearch settings can be extracted from the modules settings inside the configuration.
     # Few options will be supported, however - the modules security configuration is
@@ -113,9 +130,6 @@ module LogStash module Helpers
     end
 
     # If no settings are configured, then assume that the feature has not been configured.
-    # The assumption is that with security setup, at least one setting (password or certificates)
-    # should be configured. If security is not setup, and defaults 'just work' for monitoring, then
-    # this will need to be reconsidered.
     def feature_configured?(feature, settings)
       ES_SETTINGS.each do |option|
         return true if settings.set?("xpack.#{feature}.elasticsearch.#{option}")
@@ -146,20 +160,62 @@ module LogStash module Helpers
 
     private
 
-    def check_cloud_id_configuration!(feature, settings, prefix)
-      return if !settings.set?("#{prefix}#{feature}.elasticsearch.hosts")
+    def validate_authentication!(feature, settings, prefix)
+      provided_cloud_id = settings.set?("#{prefix}#{feature}.elasticsearch.cloud_id")
+      provided_hosts = settings.set?("#{prefix}#{feature}.elasticsearch.hosts")
+      provided_cloud_auth = settings.set?("#{prefix}#{feature}.elasticsearch.cloud_auth")
+      provided_api_key = settings.set?("#{prefix}#{feature}.elasticsearch.api_key")
+      provided_username = settings.set?("#{prefix}#{feature}.elasticsearch.username")
+      provided_password = settings.set?("#{prefix}#{feature}.elasticsearch.password")
 
-      raise ArgumentError.new("Both \"#{prefix}#{feature}.elasticsearch.cloud_id\" and " +
-                              "\"#{prefix}#{feature}.elasticsearch.hosts\" specified, please only use one of those.")
+      # note that the username setting has a default value and in the verifications below
+      # we can test on the password option being set as a proxy to using basic auth because
+      # if the username is not explicitly set it will use its default value.
+
+      if provided_cloud_auth && (provided_username || provided_password)
+        raise ArgumentError.new(
+          "Both #{prefix}#{feature}.elasticsearch.cloud_auth and " +
+          "#{prefix}#{feature}.elasticsearch.username/password " +
+          "specified, please only use one of those"
+        )
+      end
+
+      if provided_username && !provided_password
+        raise(ArgumentError,
+          "When using #{prefix}#{feature}.elasticsearch.username, " +
+          "#{prefix}#{feature}.elasticsearch.password must also be set"
+        )
+      end
+
+      if provided_cloud_id
+        if provided_hosts
+          raise(ArgumentError,
+            "Both #{prefix}#{feature}.elasticsearch.cloud_id and " +
+            "#{prefix}#{feature}.elasticsearch.hosts specified, please only use one of those"
+          )
+        end
+      end
+
+      authentication_count = 0
+      authentication_count += 1 if provided_cloud_auth
+      authentication_count += 1 if provided_password
+      authentication_count += 1 if provided_api_key
+
+      if authentication_count == 0
+        # when no explicit authentication is set it is relying on default username
+        # but without and explicit password set
+        raise(ArgumentError,
+          "With the default #{prefix}#{feature}.elasticsearch.username, " +
+          "#{prefix}#{feature}.elasticsearch.password must be set"
+        )
+      end
+
+      if authentication_count > 1
+        raise(ArgumentError, "Multiple authentication options are specified, please only use one of #{prefix}#{feature}.elasticsearch.username/password, #{prefix}#{feature}.elasticsearch.cloud_auth or #{prefix}#{feature}.elasticsearch.api_key")
+      end
+
+      if provided_api_key && !ssl?(feature, settings, prefix)
+        raise(ArgumentError, "Using api_key authentication requires SSL/TLS secured communication")
+      end
     end
-
-    def check_cloud_auth_configuration!(feature, settings, prefix)
-      return if !settings.set?("#{prefix}#{feature}.elasticsearch.username") &&
-                !settings.set?("#{prefix}#{feature}.elasticsearch.password")
-
-      raise ArgumentError.new("Both \"#{prefix}#{feature}.elasticsearch.cloud_auth\" and " +
-                              "\"#{prefix}#{feature}.elasticsearch.username\"/\"#{prefix}#{feature}.elasticsearch.password\" " +
-                              "specified, please only use one of those.")
-    end
-
   end end end
diff --git a/x-pack/lib/monitoring/internal_pipeline_source.rb b/x-pack/lib/monitoring/internal_pipeline_source.rb
index bbf2133d5..256e1c07d 100644
--- a/x-pack/lib/monitoring/internal_pipeline_source.rb
+++ b/x-pack/lib/monitoring/internal_pipeline_source.rb
@@ -13,10 +13,10 @@ module LogStash module Monitoring
     include LogStash::Util::Loggable
     FEATURE = 'monitoring'
 
-    def initialize(pipeline_config, agent)
+    def initialize(pipeline_config, agent, settings)
       super(pipeline_config.settings)
       @pipeline_config = pipeline_config
-      @settings = LogStash::SETTINGS.clone
+      @settings = settings
       @agent = agent
       @es_options = es_options_from_settings_or_modules(FEATURE, @settings)
       setup_license_checker(FEATURE)
diff --git a/x-pack/lib/monitoring/monitoring.rb b/x-pack/lib/monitoring/monitoring.rb
index ec4924e0e..4d1ff269e 100644
--- a/x-pack/lib/monitoring/monitoring.rb
+++ b/x-pack/lib/monitoring/monitoring.rb
@@ -32,7 +32,9 @@ module LogStash
         @password = es_settings['password']
         @cloud_id = es_settings['cloud_id']
         @cloud_auth = es_settings['cloud_auth']
+        @api_key = es_settings['api_key']
         @proxy = es_settings['proxy']
+        @ssl = es_settings['ssl']
         @ca_path = es_settings['cacert']
         @truststore_path = es_settings['truststore']
         @truststore_password = es_settings['truststore_password']
@@ -42,8 +44,8 @@ module LogStash
         @ssl_certificate_verification = (es_settings['verification_mode'] == 'certificate')
       end
 
-      attr_accessor :system_api_version, :es_hosts, :user, :password, :node_uuid, :cloud_id, :cloud_auth, :proxy
-      attr_accessor :ca_path, :truststore_path, :truststore_password
+      attr_accessor :system_api_version, :es_hosts, :user, :password, :node_uuid, :cloud_id, :cloud_auth, :api_key
+      attr_accessor :proxy, :ssl, :ca_path, :truststore_path, :truststore_password
       attr_accessor :keystore_path, :keystore_password, :sniffing, :ssl_certificate_verification
 
       def collection_interval
@@ -70,8 +72,12 @@ module LogStash
         user && password
       end
 
+      def api_key?
+        api_key
+      end
+
       def ssl?
-        ca_path || (truststore_path && truststore_password) || (keystore_path && keystore_password)
+        ssl || ca_path || (truststore_path && truststore_password) || (keystore_path && keystore_password)
       end
 
       def truststore?
@@ -133,7 +139,7 @@ module LogStash
 
         logger.trace("registering the metrics pipeline")
         LogStash::SETTINGS.set("node.uuid", runner.agent.id)
-        internal_pipeline_source = LogStash::Monitoring::InternalPipelineSource.new(setup_metrics_pipeline, runner.agent)
+        internal_pipeline_source = LogStash::Monitoring::InternalPipelineSource.new(setup_metrics_pipeline, runner.agent, LogStash::SETTINGS.clone)
         runner.source_loader.add_source(internal_pipeline_source)
       rescue => e
         logger.error("Failed to set up the metrics pipeline", :message => e.message, :backtrace => e.backtrace)
@@ -260,6 +266,7 @@ module LogStash
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.proxy"))
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.cloud_id"))
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.cloud_auth"))
+      settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.api_key"))
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.ssl.certificate_authority"))
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.ssl.truststore.path"))
       settings.register(LogStash::Setting::NullableString.new("#{prefix}monitoring.elasticsearch.ssl.truststore.password"))
diff --git a/x-pack/lib/template.cfg.erb b/x-pack/lib/template.cfg.erb
index af7c50a22..1926a6324 100644
--- a/x-pack/lib/template.cfg.erb
+++ b/x-pack/lib/template.cfg.erb
@@ -12,13 +12,20 @@ input {
 }
 output {
   elasticsearch_monitoring {
+  <% if auth? %>
+    user => "<%= user %>"
+    password => "<%= password %>"
+  <% end %>
+  <% if api_key? %>
+    api_key => "<%= api_key %>"
+  <% end %>
   <% if cloud_id? %>
     cloud_id => "<%= cloud_id %>"
-    <% if cloud_auth %>
-    cloud_auth => "<%= cloud_auth %>"
-    <% end %>
   <% else %>
     hosts => <%= es_hosts %>
+  <% end %>
+  <% if cloud_auth %>
+    cloud_auth => "<%= cloud_auth %>"
   <% end %>
     bulk_path => "<%= monitoring_endpoint %>"
     manage_template => false
@@ -28,10 +35,6 @@ output {
   <% if proxy? %>
     proxy => "<%= proxy %>"
   <% end %>
-  <% if auth? && !cloud_auth? %>
-    user => "<%= user %>"
-    password => "<%= password %>"
-  <% end %>
   <% if ssl? %>
     ssl => true
     <% if ca_path %>
diff --git a/x-pack/spec/config_management/elasticsearch_source_spec.rb b/x-pack/spec/config_management/elasticsearch_source_spec.rb
index 3e204412b..d7d3250ef 100644
--- a/x-pack/spec/config_management/elasticsearch_source_spec.rb
+++ b/x-pack/spec/config_management/elasticsearch_source_spec.rb
@@ -4,6 +4,7 @@
 
 require "spec_helper"
 require "logstash/json"
+require "logstash/runner"
 require "config_management/elasticsearch_source"
 require "config_management/extension"
 require "license_checker/license_manager"
@@ -164,9 +165,12 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
     end
 
     let(:pipeline_id) { "foobar" }
-    let(:settings) { { "xpack.management.pipeline.id" => pipeline_id,
-                       "xpack.management.elasticsearch.password" => "testpassword"
-          } }
+    let(:settings) do
+      {
+        "xpack.management.pipeline.id" => pipeline_id,
+        "xpack.management.elasticsearch.password" => "testpassword"
+      }
+    end
 
     it "generates the path to get the configuration" do
       expect(subject.config_path).to eq("#{described_class::PIPELINE_INDEX}/_mget")
@@ -182,10 +186,13 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
     end
 
     context "when enabled" do
-      let(:settings) { {
-        "xpack.management.enabled" => true,
-        "xpack.management.elasticsearch.password" => "testpassword"
-      } }
+      let(:settings) do
+        {
+          "xpack.management.enabled" => true,
+          "xpack.management.elasticsearch.username" => "testuser",
+          "xpack.management.elasticsearch.password" => "testpassword"
+        }
+      end
 
       it "returns true" do
         expect(subject.match?).to be_truthy
@@ -357,7 +364,8 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
         context 'when security is enabled in Elasticsearch' do
           let(:security_enabled) { true }
           it 'should not raise an error' do
-            expect { subject.pipeline_configs }.not_to raise_error(LogStash::LicenseChecker::LicenseError)
+            expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
+            expect { subject.pipeline_configs }.not_to raise_error
           end
         end
       end
diff --git a/x-pack/spec/helpers/elasticsearch_options_spec.rb b/x-pack/spec/helpers/elasticsearch_options_spec.rb
index 83e73b86c..3fb5b4922 100644
--- a/x-pack/spec/helpers/elasticsearch_options_spec.rb
+++ b/x-pack/spec/helpers/elasticsearch_options_spec.rb
@@ -4,6 +4,7 @@
 
 require "spec_helper"
 require "logstash/json"
+require "logstash/runner"
 require 'helpers/elasticsearch_options'
 require "license_checker/license_manager"
 require 'monitoring/monitoring'
@@ -97,65 +98,267 @@ describe LogStash::Helpers::ElasticsearchOptions do
   end
 
   describe "es_options_from_settings" do
-    let(:settings) do
-      {
+
+    context "with implicit username" do
+      let(:settings) do
+        {
+          "xpack.monitoring.enabled" => true,
+          "xpack.monitoring.elasticsearch.hosts" => elasticsearch_url,
+        }
+      end
+
+      it "fails without password" do
+        expect {
+          test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+        }.to raise_error(ArgumentError, /password must be set/)
+      end
+
+      context "with cloud_auth" do
+        let(:cloud_username) { 'elastic' }
+        let(:cloud_password) { 'passw0rd'}
+        let(:cloud_auth) { "#{cloud_username}:#{cloud_password}" }
+
+        let(:settings) do
+          super.merge(
+            "xpack.monitoring.elasticsearch.cloud_auth" => cloud_auth,
+          )
+        end
+
+        it "silently ignores the default username" do
+          es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          expect(es_options).to include("cloud_auth")
+          expect(es_options).to_not include("user")
+        end
+      end
+
+
+      context "with api_key" do
+        let(:settings) do
+          super.merge(
+            "xpack.monitoring.elasticsearch.api_key" => 'foo:bar'
+          )
+        end
+
+        it "silently ignores the default username" do
+          es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          expect(es_options).to include("api_key")
+          expect(es_options).to_not include("user")
+        end
+
+        context "and explicit password" do
+          let(:settings) do
+            super.merge(
+              "xpack.monitoring.elasticsearch.password" => elasticsearch_password
+            )
+          end
+
+          it "fails for multiple authentications" do
+            expect {
+              test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+            }.to raise_error(ArgumentError, /Multiple authentication options are specified/)
+          end
+        end
+      end
+    end
+
+    context "with explicit username" do
+      let(:settings) do
+        {
+          "xpack.monitoring.enabled" => true,
+          "xpack.monitoring.elasticsearch.hosts" => elasticsearch_url,
+          "xpack.monitoring.elasticsearch.username" => "foo",
+        }
+      end
+
+      it "fails without password" do
+        expect {
+          test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+        }.to raise_error(ArgumentError, /password must also be set/)
+      end
+
+      context "with cloud_auth" do
+        let(:settings) do
+          super.merge(
+            "xpack.monitoring.elasticsearch.password" => "bar",
+            "xpack.monitoring.elasticsearch.cloud_auth" => "foo:bar",
+          )
+        end
+
+        it "fails for multiple authentications" do
+          expect {
+            test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          }.to raise_error(ArgumentError, /Both.*?cloud_auth.*?and.*?username.*?specified/)
+        end
+      end
+
+      context "with api_key" do
+        let(:settings) do
+          super.merge(
+            "xpack.monitoring.elasticsearch.password" => "bar",
+            "xpack.monitoring.elasticsearch.api_key" => 'foo:bar'
+          )
+        end
+
+        it "fails for multiple authentications" do
+          expect {
+            test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          }.to raise_error(ArgumentError, /Multiple authentication options are specified/)
+        end
+      end
+    end
+
+    context "with username and password" do
+      let(:settings) do
+        {
           "xpack.monitoring.enabled" => true,
           "xpack.monitoring.elasticsearch.hosts" => elasticsearch_url,
           "xpack.monitoring.elasticsearch.username" => elasticsearch_username,
           "xpack.monitoring.elasticsearch.password" => elasticsearch_password,
-      }
-    end
-
-    it_behaves_like 'elasticsearch options hash is populated without security'
-    it_behaves_like 'elasticsearch options hash is populated with secure options'
-
-    context 'when cloud id and auth are set' do
-      let(:cloud_name) { 'thebigone'}
-      let(:cloud_domain) { 'elastic.co'}
-      let(:cloud_id) { "monitoring:#{Base64.urlsafe_encode64("#{cloud_domain}$#{cloud_name}$ignored")}" }
-      let(:cloud_username) { 'elastic' }
-      let(:cloud_password) { 'passw0rd'}
-      let(:cloud_auth) { "#{cloud_username}:#{cloud_password}" }
-      let(:expected_url) { ["https://#{cloud_name}.#{cloud_domain}:443"] }
-      let(:settings) do
-        {
-            "xpack.monitoring.enabled" => true,
-            "xpack.monitoring.elasticsearch.cloud_id" => cloud_id,
-            "xpack.monitoring.elasticsearch.cloud_auth" => cloud_auth,
         }
       end
 
-      it "creates the elasticsearch output options hash" do
-        es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
-        expect(es_options).to include("cloud_id" => cloud_id, "cloud_auth" => cloud_auth)
-        expect(es_options.keys).to_not include("hosts")
-        expect(es_options.keys).to_not include("username")
-        expect(es_options.keys).to_not include("password")
+      it_behaves_like 'elasticsearch options hash is populated without security'
+      it_behaves_like 'elasticsearch options hash is populated with secure options'
+    end
+
+
+    context 'when cloud_id' do
+      let(:cloud_name) { 'thebigone'}
+      let(:cloud_domain) { 'elastic.co'}
+      let(:cloud_id) { "monitoring:#{Base64.urlsafe_encode64("#{cloud_domain}$#{cloud_name}$ignored")}" }
+      let(:expected_url) { ["https://#{cloud_name}.#{cloud_domain}:443"] }
+      let(:settings) do
+        {
+          "xpack.monitoring.enabled" => true,
+          "xpack.monitoring.elasticsearch.cloud_id" => cloud_id,
+        }
       end
 
       context 'hosts also set' do
         let(:settings) do
           super.merge(
-              "xpack.monitoring.elasticsearch.hosts" => 'https://localhost:9200'
+            "xpack.monitoring.elasticsearch.hosts" => 'https://localhost:9200'
           )
         end
 
         it "raises due invalid configuration" do
-          expect { test_class.es_options_from_settings_or_modules('monitoring', system_settings) }.
-              to raise_error(ArgumentError, /Both.*?cloud_id.*?and.*?hosts.*?specified/)
+          expect {
+            test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+           }.to raise_error(ArgumentError, /Both.*?cloud_id.*?and.*?hosts.*?specified/)
         end
       end
 
-      context 'username also set' do
+      context "when cloud_auth is set" do
+        let(:cloud_username) { 'elastic' }
+        let(:cloud_password) { 'passw0rd'}
+        let(:cloud_auth) { "#{cloud_username}:#{cloud_password}" }
         let(:settings) do
           super.merge(
-              "xpack.monitoring.elasticsearch.username" => 'elastic'
+            "xpack.monitoring.elasticsearch.cloud_auth" => cloud_auth,
           )
         end
 
-        it "raises due invalid configuration" do
-          expect { test_class.es_options_from_settings_or_modules('monitoring', system_settings) }.
-              to raise_error(ArgumentError, /Both.*?cloud_auth.*?and.*?username.*?specified/)
+        it "creates the elasticsearch output options hash" do
+          es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          expect(es_options).to include("cloud_id" => cloud_id, "cloud_auth" => cloud_auth)
+          expect(es_options.keys).to_not include("hosts")
+          expect(es_options.keys).to_not include("username")
+          expect(es_options.keys).to_not include("password")
+        end
+
+        context 'username also set' do
+          let(:settings) do
+            super.merge(
+                "xpack.monitoring.elasticsearch.username" => 'elastic'
+            )
+          end
+
+          it "raises for invalid configuration" do
+            expect {
+              test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+            }.to raise_error(ArgumentError, /Both.*?cloud_auth.*?and.*?username.*?specified/)
+          end
+        end
+
+        context 'api_key also set' do
+          let(:settings) do
+            super.merge(
+                "xpack.monitoring.elasticsearch.api_key" => 'foo:bar',
+            )
+          end
+
+          it "raises for invalid configuration" do
+            expect {
+              test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+            }.to raise_error(ArgumentError, /Multiple authentication options are specified/)
+          end
+        end
+      end
+
+      context "when cloud_auth is not set" do
+
+        it "raises for invalid configuration" do
+          # if not other authn is provided it will assume basic auth using the default username
+          # but the password is missing.
+          expect {
+            test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          }.to raise_error(ArgumentError, /With the default.*?username,.*?password must be set/)
+        end
+
+        context 'username and password set' do
+          let(:settings) do
+            super.merge(
+              "xpack.monitoring.elasticsearch.username" => 'foo',
+              "xpack.monitoring.elasticsearch.password" => 'bar'
+            )
+          end
+
+          it "creates the elasticsearch output options hash" do
+            es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+            expect(es_options).to include("cloud_id", "user", "password")
+            expect(es_options.keys).to_not include("hosts")
+          end
+        end
+
+        context 'api_key set' do
+          let(:settings) do
+            super.merge(
+              "xpack.monitoring.elasticsearch.api_key" => 'foo:bar'
+            )
+          end
+
+          it "creates the elasticsearch output options hash" do
+            es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+            expect(es_options).to include("cloud_id", "api_key")
+            expect(es_options.keys).to_not include("hosts")
+          end
+        end
+      end
+    end
+
+    context 'when api_key is set' do
+      let(:api_key) { 'foo:bar'}
+      let(:settings) do
+        {
+          "xpack.monitoring.enabled" => true,
+          "xpack.monitoring.elasticsearch.hosts" => elasticsearch_url,
+          "xpack.monitoring.elasticsearch.api_key" => api_key,
+        }
+      end
+
+      it "creates the elasticsearch output options hash" do
+        es_options = test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+        expect(es_options).to include("api_key" => api_key)
+        expect(es_options.keys).to include("hosts")
+      end
+
+      context "with a non https host" do
+        let(:elasticsearch_url) { ["https://host1", "http://host2"] }
+
+        it "fails at options validation" do
+          expect {
+            test_class.es_options_from_settings_or_modules('monitoring', system_settings)
+          }.to raise_error(ArgumentError, /api_key authentication requires SSL\/TLS/)
         end
       end
     end
diff --git a/x-pack/spec/license_checker/license_reader_spec.rb b/x-pack/spec/license_checker/license_reader_spec.rb
index fc25e1de4..5acdecf80 100644
--- a/x-pack/spec/license_checker/license_reader_spec.rb
+++ b/x-pack/spec/license_checker/license_reader_spec.rb
@@ -7,6 +7,7 @@ require 'support/helpers'
 require "license_checker/license_reader"
 require "helpers/elasticsearch_options"
 require "monitoring/monitoring"
+require "logstash/runner"
 
 describe LogStash::LicenseChecker::LicenseReader do
   let(:elasticsearch_url) { "https://localhost:9898" }
@@ -23,7 +24,7 @@ describe LogStash::LicenseChecker::LicenseReader do
   let(:settings) do
     {
       "xpack.monitoring.enabled" => true,
-      "xpack.monitoring.elasticsearch.hosts" => [ elasticsearch_url ],
+      "xpack.monitoring.elasticsearch.hosts" => [ elasticsearch_url],
       "xpack.monitoring.elasticsearch.username" => elasticsearch_username,
       "xpack.monitoring.elasticsearch.password" => elasticsearch_password,
     }
@@ -128,4 +129,19 @@ describe LogStash::LicenseChecker::LicenseReader do
       expect( subject.client.options ).to include(:user => 'elastic', :password => 'LnWMLeK3EQPTf3G3F1IBdFvO')
     end
   end
+
+  context 'with api_key' do
+    let(:api_key) { "foo:bar" }
+    let(:settings) do
+      {
+        "xpack.monitoring.enabled" => true,
+        "xpack.monitoring.elasticsearch.hosts" => [elasticsearch_url],
+        "xpack.monitoring.elasticsearch.api_key" => api_key,
+      }
+    end
+
+    it "builds ES client" do
+      expect( subject.client.options[:client_settings][:headers] ).to include("Authorization" => "ApiKey Zm9vOmJhcg==")
+    end
+  end
 end
diff --git a/x-pack/spec/monitoring/inputs/metrics_spec.rb b/x-pack/spec/monitoring/inputs/metrics_spec.rb
index 47e1db9c3..65d7aecd1 100644
--- a/x-pack/spec/monitoring/inputs/metrics_spec.rb
+++ b/x-pack/spec/monitoring/inputs/metrics_spec.rb
@@ -4,6 +4,7 @@
 
 require 'spec_helper'
 require "logstash-core"
+require "logstash/runner"
 require "logstash/agent"
 require "monitoring/inputs/metrics"
 require "rspec/wait"
diff --git a/x-pack/spec/monitoring/internal_pipeline_source_spec.rb b/x-pack/spec/monitoring/internal_pipeline_source_spec.rb
index 508ccdbb1..3ebaabf12 100644
--- a/x-pack/spec/monitoring/internal_pipeline_source_spec.rb
+++ b/x-pack/spec/monitoring/internal_pipeline_source_spec.rb
@@ -4,7 +4,7 @@
 
 require "logstash-core"
 require "logstash/agent"
-require "logstash/agent"
+require "logstash/runner"
 require "monitoring/inputs/metrics"
 require "logstash/config/pipeline_config"
 require "logstash/config/source/local"
@@ -23,11 +23,17 @@ describe LogStash::Monitoring::InternalPipelineSource do
     let(:options) { { "collection_interval" => xpack_monitoring_interval,
                         "collection_timeout_interval" => 600 } }
 
-    subject { described_class.new(pipeline_config, mock_agent) }
+    subject { described_class.new(pipeline_config, mock_agent, system_settings) }
     let(:mock_agent) { double("agent")}
     let(:mock_license_client) { double("es_client")}
     let(:license_reader) { LogStash::LicenseChecker::LicenseReader.new(system_settings, 'monitoring', es_options)}
-    let(:system_settings) { LogStash::Runner::SYSTEM_SETTINGS.clone }
+    let(:extension) { LogStash::MonitoringExtension.new }
+    let(:system_settings) do
+      LogStash::Runner::SYSTEM_SETTINGS.clone.tap do |system_settings|
+        extension.additionals_settings(system_settings) # register defaults from extension
+        apply_settings(settings, system_settings) # apply `settings`
+      end
+    end
     let(:license_status) { 'active'}
     let(:license_type) { 'trial' }
     let(:license_expiry_date) { Time.now + (60 * 60 * 24)}
@@ -70,7 +76,6 @@ describe LogStash::Monitoring::InternalPipelineSource do
     end
 
     before :each do
-      allow(subject).to receive(:es_options_from_settings_or_modules).and_return(es_options)
       allow(subject).to receive(:license_reader).and_return(license_reader)
       allow(license_reader).to receive(:build_client).and_return(mock_license_client)
     end