out_s3: stop unnecessary re-compression when chunk is already compressed

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/out_s3.rb

@@ -203,7 +203,7 @@ def configure(conf)
 
       begin
         buffer_type = @buffer_config[:@type]
-        @compressor = COMPRESSOR_REGISTRY.lookup(@store_as).new(buffer_type: buffer_type, log: log)
+        @compressor = COMPRESSOR_REGISTRY.lookup(@store_as).new(buffer_type: buffer_type, buffer_compressed_type: @buffer.compress, log: log)
       rescue => e
         log.warn "'#{@store_as}' not supported. Use 'text' instead: error = #{e.message}"
         @compressor = TextCompressor.new
@@ -600,6 +600,7 @@ class Compressor
       def initialize(opts = {})
         super()
         @buffer_type = opts[:buffer_type]
+        @buffer_compressed_type = opts[:buffer_compressed_type]
         @log = opts[:log]
       end
 
@@ -642,11 +643,18 @@ def content_type
       end
 
       def compress(chunk, tmp)
+        if @buffer_compressed_type == :gzip
+          chunk.write_to(tmp, compressed: @buffer_compressed_type)
+          return
+        end
+
         w = Zlib::GzipWriter.new(tmp)
         chunk.write_to(w)
         w.finish
       ensure
-        w.finish rescue nil
+        if w
+          w.finish rescue nil
+        end
       end
     end
 

lib/fluent/plugin/s3_compressor_gzip_command.rb

@@ -19,6 +19,11 @@ def content_type
       end
 
       def compress(chunk, tmp)
+        if @buffer_compressed_type == :gzip
+          chunk.write_to(tmp, compressed: @buffer_compressed_type)
+          return
+        end
+
         chunk_is_file = @buffer_type == 'file'
         path = if chunk_is_file
                  chunk.path

lib/fluent/plugin/s3_compressor_zstd.rb

@@ -25,6 +25,11 @@ def content_type
       end
 
       def compress(chunk, tmp)
+        if @buffer_compressed_type == :zstd
+          chunk.write_to(tmp, compressed: @buffer_compressed_type)
+          return
+        end
+
         compressed = Zstd.compress(chunk.read, level: @compress_config.level)
         tmp.write(compressed)
       rescue => e