out_http: Add headers_from_placeholders. fix #3237

lib/fluent/plugin/out_http.rb

@@ -51,6 +51,8 @@ class RetryableResponse < StandardError; end
     config_param :json_array, :bool, default: false
     desc 'Additional headers for HTTP request'
     config_param :headers, :hash, default: nil
+    desc 'Additional placeholder based headers for HTTP request'
+    config_param :headers_from_placeholders, :hash, default: nil
 
     desc 'The connection open timeout in seconds'
     config_param :open_timeout, :integer, default: nil
@@ -213,12 +215,17 @@ def parse_endpoint(chunk)
       URI.parse(endpoint)
     end
 
-    def set_headers(req)
+    def set_headers(req, chunk)
       if @headers
         @headers.each do |k, v|
           req[k] = v
         end
       end
+      if @headers_from_placeholders
+        @headers_from_placeholders.each do |k, v|
+          req[k] = extract_placeholders(v, chunk)
+        end
+      end
       req['Content-Type'] = @content_type
     end
 
@@ -232,7 +239,7 @@ def create_request(chunk, uri)
       if @auth
         req.basic_auth(@auth.username, @auth.password)
       end
-      set_headers(req)
+      set_headers(req, chunk)
       req.body = @json_array ? "[#{chunk.read.chop}]" : chunk.read
       req
     end

test/plugin/test_out_http.rb

@@ -280,6 +280,25 @@ def test_write_with_headers
     assert_equal "fluentd!", result.headers['test_header']
   end
 
+  def test_write_with_headers_from_placeholders
+    d = create_driver(config + %[
+      headers_from_placeholders {"x-test":"${$.foo.bar}-test","x-tag":"${tag}"}
+      <buffer tag,$.foo.bar>
+      </buffer>
+    ])
+    d.run(default_tag: 'test.http') do
+      test_events.each { |event|
+        ev = event.dup
+        ev['foo'] = {'bar' => 'abcd'}
+        d.feed(ev)
+      }
+    end
+
+    result = @@result
+    assert_equal "abcd-test", result.headers['x-test']
+    assert_equal "test.http", result.headers['x-tag']
+  end
+
   def test_write_with_retryable_response
     old_report_on_exception = Thread.report_on_exception
     Thread.report_on_exception = false # thread finished as invalid state since RetryableResponse raises.