Add new ZST compression plugin

README.md

@@ -40,7 +40,6 @@ Simply use RubyGems:
     # For v0.12. This is for old v0.12 users. Don't use v0.12 for new deployment
     $ gem install fluent-plugin-s3 -v "~> 0.8" --no-document # for fluentd v0.12
 
-
 ## Configuration: credentials
 
 Both S3 input/output plugin provide several credential methods for authentication/authorization.

lib/fluent/plugin/s3_compressor_zst.rb

@@ -0,0 +1,35 @@
+module Fluent::Plugin
+  class S3Output
+    class ZSTCompressor < Compressor
+      S3Output.register_compressor('zst', self)
+
+      config_param :command_parameter, :string, default: '-qq -f -7'
+
+      def configure(conf)
+        super
+        check_command('zstd')
+      end
+
+      def ext
+        'zst'.freeze
+      end
+
+      def content_type
+        'application/x-zst'.freeze
+      end
+
+      def compress(chunk, tmp)
+        w = Tempfile.new("chunk-tmp")
+        w.binmode
+        chunk.write_to(w)
+        w.close
+
+        # We don't check the return code because we can't recover zstd failure.
+        system "zstd #{@command_parameter} -o #{tmp.path} #{w.path}"
+      ensure
+        w.close rescue nil
+        w.unlink rescue nil
+      end
+    end
+  end
+end

lib/fluent/plugin/s3_extractor_zst.rb

@@ -0,0 +1,40 @@
+module Fluent::Plugin
+  class S3Input
+    class ZSTExtractor < Extractor
+      S3Input.register_extractor('zst', self)
+
+      config_param :command_parameter, :string, default: '-qq -d -f'
+
+      def configure(conf)
+        super
+        check_command('zstd')
+      end
+
+      def ext
+        'zst'.freeze
+      end
+
+      def content_type
+        'application/x-zst'.freeze
+      end
+
+      def extract(io)
+        path = if io.respond_to?(path)
+                 io.path
+               else
+                 temp = Tempfile.new("zst-temp")
+                 temp.write(io.read)
+                 temp.close
+                 temp.path
+               end
+
+        stdout, succeeded = Open3.capture2("zstd #{@command_parameter} #{path}")
+        if succeeded
+          stdout
+        else
+          raise "Failed to extract #{path} with zstd command."
+        end
+      end
+    end
+  end
+end

test/test_in_s3.rb

@@ -94,7 +94,8 @@ def test_unknown_store_as
          "gzip" => ["gzip", "gz", "application/x-gzip"],
          "gzip_command" => ["gzip_command", "gz", "application/x-gzip"],
          "lzo" => ["lzo", "lzo", "application/x-lzop"],
-         "lzma2" => ["lzma2", "xz", "application/x-xz"])
+         "lzma2" => ["lzma2", "xz", "application/x-xz"],
+         "zst" => ["zst", "zst", "application/x-zst"])
     def test_extractor(data)
       store_type, ext, content_type = data
       config = CONFIG + "\nstore_as #{store_type}\n"