如何在Fluentd S3插件中实现zstd压缩？启动报错求助

需求与现状 #

需要将Fluentd采集的数据以zstd压缩格式上传至S3，自行开发了对应压缩插件但启动Fluentd时报错。已完成以下操作：

• 在Ubuntu 18.04.6 LTS系统安装td-agent 4.4.2（对应Fluentd 1.15.3）
• 安装zstd-ruby(1.5.2.3)和zstd gem
• 编写s3_compressor_zstd.rb压缩插件文件，并放置到fluent-plugin-s3插件目录，但启动失败

环境信息 #

• OS: Ubuntu 18.04.6 LTS
• td-agent版本: 4.4.2
• Fluentd版本: 1.15.3
• zstd版本: 1.1.2.1
• zstd-ruby版本: 1.5.2.3

原有插件代码 #

require "zstd"
module Fluent::Plugin
  class S3Output
    class ZstdCompressor < Compressor
      S3Output.register_compressor('zstd', self)
      def initialize(options = {})
        begin
          require "zstd"
        rescue LoadError
          raise Fluent::ConfigError, "Install zstd-ruby gem before using zstd compressor"
        end
      end
      def ext
        ".zst"
      end
      def compress(data)
        compressed_data = ''
        Zstd::Writer.open(StringIO.new(compressed_data)) do |compressed_stream|
          compressed_stream.write(data)
        end
        compressed_data
      end
    end
  end
end

问题分析与修正方案 #

问题点 #

1. StringIO使用错误：原代码中StringIO.new(compressed_data)基于空字符串创建流，写入操作不会修改原compressed_data变量，导致最终返回空的压缩数据
2. 缺少标准库依赖：未引入stringio标准库，可能引发加载错误
3. 初始化逻辑冗余：重复执行require "zstd"无意义
4. 无压缩级别配置：无法灵活调整压缩效率与速度的平衡

修正后的插件代码 #

require "zstd"
require "stringio"

module Fluent::Plugin
  class S3Output
    class ZstdCompressor < Compressor
      S3Output.register_compressor('zstd', self)

      # 添加压缩级别配置，默认3（平衡速度与压缩率）
      config_param :compression_level, :integer, default: 3

      def initialize(options = {})
        super
        # 检查zstd库是否加载
        unless defined?(Zstd)
          raise Fluent::ConfigError, "Install zstd-ruby gem before using zstd compressor"
        end
      end

      def ext
        ".zst"
      end

      def compress(data)
        # 使用StringIO捕获压缩输出
        output_stream = StringIO.new
        Zstd::Writer.open(output_stream, level: @compression_level) do |writer|
          writer.write(data)
        end
        # 返回压缩后的字符串
        output_stream.string
      end
    end
  end
end

部署与配置步骤 #

1. 放置插件文件：将修正后的s3_compressor_zstd.rb放到fluent-plugin-s3的插件目录，示例路径：

   /opt/td-agent/lib/ruby/gems/2.7.0/gems/fluent-plugin-s3-1.7.2/lib/fluent/plugin/
   

   （请根据实际安装路径调整）

2. 修改Fluentd配置：在S3输出插件的配置块中指定zstd压缩：

   <match **>
     @type s3
     # 以下为你的S3基础配置，请按需修改
     aws_key_id YOUR_AWS_KEY
     aws_sec_key YOUR_AWS_SECRET
     s3_bucket YOUR_BUCKET_NAME
     path logs/%Y/%m/%d/
     s3_region ap-southeast-1
   
     # 启用zstd压缩
     compressor zstd
     # 可选，设置压缩级别（1-22，级别越高压缩率越高但速度越慢）
     compression_level 5
   </match>
   

3. 重启服务并验证：

   sudo systemctl restart td-agent
   # 查看日志确认启动状态
   sudo tail -f /var/log/td-agent/td-agent.log
   

内容的提问来源于stack exchange，提问作者Akash Saini