diff --git a/lib/fluent/plugin/out_http.rb b/lib/fluent/plugin/out_http.rb index 4d45b9706c..44e258b054 100644 --- a/lib/fluent/plugin/out_http.rb +++ b/lib/fluent/plugin/out_http.rb @@ -55,6 +55,8 @@ class RetryableResponse < StandardError; end config_param :headers, :hash, default: nil desc 'Additional placeholder based headers for HTTP request' config_param :headers_from_placeholders, :hash, default: nil + desc 'Compress HTTP request body' + config_param :compress, :enum, list: [:text, :gzip], default: :text desc 'The connection open timeout in seconds' config_param :open_timeout, :integer, default: nil @@ -290,6 +292,9 @@ def set_headers(req, uri, chunk) req[k] = extract_placeholders(v, chunk) end end + if @compress == :gzip + req['Content-Encoding'] = "gzip" + end req['Content-Type'] = @content_type end @@ -323,7 +328,24 @@ def create_request(chunk, uri) Net::HTTP::Put.new(uri.request_uri) end set_headers(req, uri, chunk) - req.body = @json_array ? "[#{chunk.read.chop}]" : chunk.read + + compress_body = @compress != :text + + case + when @json_array + req.body = "[#{chunk.read.chop}]" + when @compress != @buffer.compress + req.body = chunk.read + when @compress == @buffer.compress + req.body = chunk.read(compressed: @compress) + compress_body = false + end + + if compress_body + gz = Zlib::GzipWriter.new(StringIO.new) + gz << req.body + req.body = gz.close.string + end # At least one authentication method requires the body and other headers, so the order of this call matters set_auth(req, uri) diff --git a/test/plugin/test_out_http.rb b/test/plugin/test_out_http.rb index a0b3969a8d..a8c39fdbee 100644 --- a/test/plugin/test_out_http.rb +++ b/test/plugin/test_out_http.rb @@ -75,21 +75,30 @@ def run_http_server @@result.headers[key] = value end + body = "" data = [] + + case req['content-encoding'] + when 'gzip' + body = Zlib::GzipReader.new(StringIO.new(req.body)).read + else + body = req.body + end + case req.content_type when 'application/x-ndjson' - req.body.each_line { |l| + body.each_line { |l| data << JSON.parse(l) } when 'application/json' - data = JSON.parse(req.body) + data = JSON.parse(body) when 'text/plain' # Use single_value in this test - req.body.each_line { |line| + body.each_line { |line| data << line.chomp } else - data << req.body + data << body end @@result.data = data @@ -519,6 +528,31 @@ def test_write_with_https end end + sub_test_case 'GZIP' do + def server_port + 19882 + end + + def test_write_with_gzip + d = create_driver(%[ + endpoint http://127.0.0.1:#{server_port}/test + compress gzip + ]) + d.run(default_tag: 'test.http') do + test_events.each { |event| + d.feed(event) + } + end + + result = @@result + assert_equal 'POST', result.method + assert_equal 'application/x-ndjson', result.content_type + assert_equal 'gzip', result.headers['content-encoding'] + assert_equal test_events, result.data + assert_not_empty result.headers + end + end + sub_test_case 'connection_reuse' do def server_port 19883