diff --git a/x-pack/lib/config_management/elasticsearch_source.rb b/x-pack/lib/config_management/elasticsearch_source.rb
index 84b569541..b4a436140 100644
--- a/x-pack/lib/config_management/elasticsearch_source.rb
+++ b/x-pack/lib/config_management/elasticsearch_source.rb
@@ -119,7 +119,7 @@ module LogStash
       # But we have to silence the logger from the plugin, to make sure the
       # log originate from the `ElasticsearchSource`
       def build_client
-        es = LogStash::Outputs::ElasticSearch.new(@es_options)
+        es = LogStash::Outputs::ElasticSearch.new(es_options_with_product_origin_header(@es_options))
         new_logger = logger
         es.instance_eval { @logger = new_logger }
         es.build_client
diff --git a/x-pack/lib/helpers/elasticsearch_options.rb b/x-pack/lib/helpers/elasticsearch_options.rb
index 397632a06..8ecbed03c 100644
--- a/x-pack/lib/helpers/elasticsearch_options.rb
+++ b/x-pack/lib/helpers/elasticsearch_options.rb
@@ -79,6 +79,16 @@ module LogStash module Helpers
       opts
     end
 
+    # when the Elasticsearch Output client is used exclusively to
+    # perform Logstash-defined actions without user input, adding
+    # a product origin header allows us to reduce log noise.
+    def es_options_with_product_origin_header(es_options)
+      custom_headers = es_options.delete('custom_headers') { Hash.new }
+                                 .merge('x-elastic-product-origin' => 'logstash')
+
+      es_options.merge('custom_headers' => custom_headers)
+    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
diff --git a/x-pack/lib/license_checker/license_reader.rb b/x-pack/lib/license_checker/license_reader.rb
index 29e2e873e..194c8ff98 100644
--- a/x-pack/lib/license_checker/license_reader.rb
+++ b/x-pack/lib/license_checker/license_reader.rb
@@ -15,7 +15,9 @@ module LogStash
       def initialize(settings, feature, options)
         @namespace = "xpack.#{feature}"
         @settings = settings
-        @es_options = options.merge('resurrect_delay' => 30)
+
+        es_options = options.merge('resurrect_delay' => 30)
+        @es_options = Helpers::ElasticsearchOptions::es_options_with_product_origin_header(es_options)
       end
 
       ##
diff --git a/x-pack/lib/template.cfg.erb b/x-pack/lib/template.cfg.erb
index 1926a6324..2335c3b1c 100644
--- a/x-pack/lib/template.cfg.erb
+++ b/x-pack/lib/template.cfg.erb
@@ -54,5 +54,8 @@ output {
     # the reason being that the user can still turn ssl on by using https in their URL
     # This causes the ES output to throw an error due to conflicting messages
   <% end %>
+    # Since this config is generated by Logstash and cannot be directly
+    # controlled by the user, we include a product origin header
+    custom_headers => { "x-elastic-product-origin" => "logstash" }
   }
 }
diff --git a/x-pack/spec/config_management/elasticsearch_source_spec.rb b/x-pack/spec/config_management/elasticsearch_source_spec.rb
index 1fa0a0516..d4510aef6 100644
--- a/x-pack/spec/config_management/elasticsearch_source_spec.rb
+++ b/x-pack/spec/config_management/elasticsearch_source_spec.rb
@@ -192,6 +192,27 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
         end
       end
     end
+
+    context "valid settings" do
+      let(:settings) do
+        {
+          "xpack.management.enabled" => true,
+          "xpack.management.pipeline.id" => "main",
+          "xpack.management.elasticsearch.hosts" => elasticsearch_url,
+          "xpack.management.elasticsearch.username" => elasticsearch_username,
+          "xpack.management.elasticsearch.password" => elasticsearch_password,
+        }
+      end
+
+      # This spec is certainly indirect, and assumes that communication with Elasticsearch
+      # will be done through an ES-output HttpClient available at described_class.client
+      it 'creates an Elasticsearch HttpClient with product origin custom headers' do
+        internal_client = subject.send(:client) # internal reach
+        expect(internal_client)
+          .to be_a_kind_of(LogStash::Outputs::ElasticSearch::HttpClient)
+          .and(have_attributes(:client_settings => hash_including(:headers => hash_including("x-elastic-product-origin" => "logstash"))))
+      end
+    end
   end
 
   describe LogStash::ConfigManagement::SystemIndicesFetcher do
diff --git a/x-pack/spec/license_checker/license_reader_spec.rb b/x-pack/spec/license_checker/license_reader_spec.rb
index ccb0af5a4..92a03b889 100644
--- a/x-pack/spec/license_checker/license_reader_spec.rb
+++ b/x-pack/spec/license_checker/license_reader_spec.rb
@@ -20,6 +20,7 @@ describe LogStash::LicenseChecker::LicenseReader do
       apply_settings(settings, system_settings) # apply `settings`
     end
   end
+  let(:product_origin_header) { { "x-elastic-product-origin" => "logstash" } }
 
   let(:settings) do
     {
@@ -112,6 +113,7 @@ describe LogStash::LicenseChecker::LicenseReader do
     expect( subject.client.options[:hosts].size ).to eql 1
     expect( subject.client.options[:hosts][0].to_s ).to eql elasticsearch_url # URI#to_s
     expect( subject.client.options ).to include(:user => elasticsearch_username, :password => elasticsearch_password)
+    expect( subject.client.client_settings[:headers] ).to include(product_origin_header)
   end
 
   context 'with cloud_id' do
@@ -134,6 +136,7 @@ describe LogStash::LicenseChecker::LicenseReader do
       expect( subject.client.options[:hosts].size ).to eql 1
       expect( subject.client.options[:hosts][0].to_s ).to eql 'https://e1e631201fb64d55a75f431eb6349589.westeurope.azure.elastic-cloud.com:9243'
       expect( subject.client.options ).to include(:user => 'elastic', :password => 'LnWMLeK3EQPTf3G3F1IBdFvO')
+      expect( subject.client.client_settings[:headers] ).to include(product_origin_header)
     end
   end
 
@@ -148,7 +151,8 @@ describe LogStash::LicenseChecker::LicenseReader do
     end
 
     it "builds ES client" do
-      expect( subject.client.options[:client_settings][:headers] ).to include("Authorization" => "ApiKey Zm9vOmJhcg==")
+      expect( subject.client.client_settings[:headers] ).to include("Authorization" => "ApiKey Zm9vOmJhcg==")
+      expect( subject.client.client_settings[:headers] ).to include(product_origin_header)
     end
   end
 end