186-custom-headers

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.17.0
+ - Added support for custom headers [#187](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/187)
+
 ## 3.16.2
   - Add `x-elastic-product-origin` header to Elasticsearch requests [#185](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/185)
 

docs/index.asciidoc

@@ -132,6 +132,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-cloud_auth>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-cloud_id>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-docinfo_fields>> |<<hash,hash>>|No
+| <<plugins-{type}s-{plugin}-custom_headers>> |<<hash,hash>>|No
 | <<plugins-{type}s-{plugin}-enable_sort>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-fields>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-hosts>> |<<array,array>>|No
@@ -247,6 +248,15 @@ Example:
       }
     }
 
+[id="plugins-{type}s-{plugin}-custom_headers"]
+===== `custom_headers`
+
+  * Value type is <<hash,hash>>
+  * Default value is empty
+
+Pass a set of key value pairs as the headers sent in each request to an elasticsearch node.
+The values of these custom headers will override any headers previously set by the plugin such as the User Agent or Authorization headers.
+
 [id="plugins-{type}s-{plugin}-enable_sort"]
 ===== `enable_sort` 
 

lib/logstash/filters/elasticsearch.rb

@@ -36,6 +36,9 @@ class LogStash::Filters::Elasticsearch < LogStash::Filters::Base
   # Hash of docinfo fields to copy from old event (found via elasticsearch) into new event
   config :docinfo_fields, :validate => :hash, :default => {}
 
+  # Custom headers for Elasticsearch requests
+  config :custom_headers, :validate => :hash, :default => {}
+
   # Hash of aggregation names to copy from elasticsearch response into Logstash event fields
   config :aggregation_fields, :validate => :hash, :default => {}
 
@@ -269,7 +272,9 @@ def client_options
       :ssl => client_ssl_options,
       :retry_on_failure => @retry_on_failure,
       :retry_on_status => @retry_on_status,
-      :user_agent => prepare_user_agent
+      :user_agent => prepare_user_agent,
+      :custom_headers => @custom_headers
+
     }
   end
 

lib/logstash/filters/elasticsearch/client.rb

@@ -20,14 +20,15 @@ def initialize(logger, hosts, options = {})
         api_key = options.fetch(:api_key, nil)
         proxy = options.fetch(:proxy, nil)
         user_agent = options[:user_agent]
+        custom_headers = options[:custom_headers]
 
         transport_options = { }
         transport_options[:headers] = options.fetch(:serverless, false) ?  DEFAULT_EAV_HEADER.dup : {}
         transport_options[:headers].merge!(setup_basic_auth(user, password))
         transport_options[:headers].merge!(setup_api_key(api_key))
         transport_options[:headers].merge!({ 'user-agent' => "#{user_agent}" })
         transport_options[:headers].merge!(INTERNAL_ORIGIN_HEADER)
-
+        transport_options[:headers].merge!(custom_headers)
         transport_options[:pool_max] = 1000
         transport_options[:pool_max_per_route] = 100
 

spec/filters/elasticsearch_spec.rb

@@ -82,6 +82,23 @@
         expect {plugin.register}.to raise_error(LogStash::ConfigurationError)
       end
     end
+
+
+    context "with custom headers" do
+      let(:config) do
+        {
+          "schedule" => "* * * * * UTC",
+          "custom_headers" => { "Custom-Header-1" => "Custom Value 1", "Custom-Header-2" => "Custom Value 2" }
+        }
+      end
+
+
+      it "sets custom headers" do
+        plugin.register
+        client = plugin.send(:client)
+        expect( extract_transport(client).options[:transport_options][:headers] ).to match hash_including(config["custom_headers"])
+      end
+    end
   end
 
   describe "data fetch" do