Implement Compressable to compress data

lib/fluent/plugin/buffer.rb

@@ -17,6 +17,7 @@
 require 'fluent/plugin/base'
 require 'fluent/plugin/owned_by_mixin'
 require 'fluent/unique_id'
+require 'fluent/plugin/buffer/compressable'
 
 require 'monitor'
 
@@ -91,6 +92,14 @@ def configure(conf)
         unless @queue_length_limit.nil?
           @total_limit_size = @chunk_limit_size * @queue_length_limit
         end
+
+        if @compress
+          if owner && owner.remote_write?
+            extend(RemoteCompressable)
+          else
+            extend(Compressable)
+          end
+        end
       end
 
       def start

lib/fluent/plugin/buffer/compressable.rb

@@ -0,0 +1,60 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'zlib'
+
+module Fluent
+  module Plugin
+    module Compressable
+      def  generate_chunk(metadata)
+        chunk = super(metadata)
+        chunk.extend(ChunkCompressable)
+        chunk.extend(ChunkUncompressable)
+        chunk
+      end
+    end
+
+    module RemoteCompressable
+      def  generate_chunk(metadata)
+        chunk = super(metadata)
+        chunk.extend(ChunkCompressable)
+        # Do not uncompress chunk when plugin send chunk on the network, to reduce consuming the bandwidth of the network.
+        chunk
+      end
+    end
+
+    module ChunkCompressable
+      def concat(bulk, bulk_size)
+        super(compress(bulk), bulk_size)
+      end
+
+      def compress(data)
+        Zlib::Deflate.deflate(data)
+      end
+    end
+
+    module ChunkUncompressable
+      def write_to(io)
+        @chunk = uncompress(@chunk)
+        super
+      end
+
+      def uncompress(compressed_data)
+        Zlib::Inflate.inflate(compressed_data)
+      end
+    end
+  end
+end

lib/fluent/plugin/output.rb

@@ -1000,6 +1000,10 @@ def flush_thread_run(state)
           raise
         end
       end
+
+      def remote_write?
+        false
+      end
     end
   end
 end