Set x-elastic-product-origin header for ES requests

This commit updates the `Elasticsearch::Client` used to make requests to ES to
send along a header identifying the request as originating from an internal
component.

lib/logstash/inputs/elasticsearch.rb

@@ -271,6 +271,7 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
 
   BUILD_FLAVOR_SERVERLESS = 'serverless'.freeze
   DEFAULT_EAV_HEADER = { "Elastic-Api-Version" => "2023-10-31" }.freeze
+  INTERNAL_ORIGIN_HEADER = { 'x-elastic-product-origin' => 'logstash-input-elasticsearch'}.freeze
 
   def initialize(params={})
     super(params)
@@ -300,6 +301,7 @@ def register
     fill_user_password_from_cloud_auth
 
     transport_options = {:headers => {}}
+    transport_options[:headers].merge!(INTERNAL_ORIGIN_HEADER)
     transport_options[:headers].merge!(setup_basic_auth(user, password))
     transport_options[:headers].merge!(setup_api_key(api_key))
     transport_options[:headers].merge!({'user-agent' => prepare_user_agent()})
@@ -622,7 +624,7 @@ def setup_serverless
       @client = Elasticsearch::Client.new(@client_options)
       @client.info
     end
-  rescue => e
+  rescue => egi
     @logger.error("Failed to retrieve Elasticsearch info", message: e.message, exception: e.class, backtrace: e.backtrace)
     raise LogStash::ConfigurationError, "Could not connect to a compatible version of Elasticsearch"
   end

spec/inputs/elasticsearch_spec.rb

@@ -50,6 +50,12 @@
         client = plugin.send(:client)
         expect( extract_transport(client).options[:transport_options][:headers] ).not_to match hash_including("Elastic-Api-Version" => "2023-10-31")
       end
+
+      it "sets an x-elastic-product-origin header identifying this as an internal plugin request" do
+        plugin.register
+        client = plugin.send(:client)
+        expect( extract_transport(client).options[:transport_options][:headers] ).to match hash_including("x-elastic-product-origin"=>"logstash-input-elasticsearch")
+      end
     end
 
     context "against not authentic Elasticsearch" do
@@ -90,6 +96,12 @@
           client = plugin.send(:client)
           expect( extract_transport(client).options[:transport_options][:headers] ).to match hash_including("Elastic-Api-Version" => "2023-10-31")
         end
+
+        it "sets an x-elastic-product-origin header identifying this as an internal plugin request" do
+          plugin.register
+          client = plugin.send(:client)
+          expect( extract_transport(client).options[:transport_options][:headers] ).to match hash_including("x-elastic-product-origin"=>"logstash-input-elasticsearch")
+        end
       end
     end