out_s3.rb in fluent-plugin-s3-1.2.0

- old
+ new

@@ -122,10 +122,18 @@
     config_param :signature_version, :string, default: nil # use nil to follow SDK default configuration
     desc "Given a threshold to treat events as delay, output warning logs if delayed events were put into s3"
     config_param :warn_for_delay, :time, default: nil
     desc "Arbitrary S3 metadata headers to set for the object"
     config_param :s3_metadata, :hash, default: nil
+    config_section :bucket_lifecycle_rule, param_name: :bucket_lifecycle_rules, multi: true do
+      desc "A unique ID for this rule"
+      config_param :id, :string
+      desc "Objects whose keys begin with this prefix will be affected by the rule. If not specified all objects of the bucket will be affected"
+      config_param :prefix, :string, default: ''
+      desc "The number of days before the object will expire"
+      config_param :expiration_days, :integer
+    end
 
     DEFAULT_FORMAT_TYPE = "out_file"
 
     config_section :format do
       config_set_default :@type, DEFAULT_FORMAT_TYPE
@@ -215,10 +223,11 @@
       @s3 = Aws::S3::Resource.new(client: s3_client)
       @bucket = @s3.bucket(@s3_bucket)
 
       check_apikeys if @check_apikey_on_start
       ensure_bucket if @check_bucket
+      ensure_bucket_lifecycle
 
       super
     end
 
     def format(tag, time, record)
@@ -368,9 +377,33 @@
           log.info "Creating bucket #{@s3_bucket} on #{@s3_endpoint}"
           @s3.create_bucket(bucket: @s3_bucket)
         else
           raise "The specified bucket does not exist: bucket = #{@s3_bucket}"
         end
+      end
+    end
+
+    def ensure_bucket_lifecycle
+      unless @bucket_lifecycle_rules.empty?
+        old_rules = get_bucket_lifecycle_rules
+        new_rules = @bucket_lifecycle_rules.sort_by { |rule| rule.id }.map do |rule|
+          { id: rule.id, expiration: { days: rule.expiration_days }, prefix: rule.prefix, status: "Enabled" }
+        end
+
+        unless old_rules == new_rules
+          log.info "Configuring bucket lifecycle rules for #{@s3_bucket} on #{@s3_endpoint}"
+          @bucket.lifecycle_configuration.put({ lifecycle_configuration: { rules: new_rules } })
+        end
+      end
+    end
+
+    def get_bucket_lifecycle_rules
+      begin
+        @bucket.lifecycle_configuration.rules.sort_by { |rule| rule[:id] }.map do |rule|
+          { id: rule[:id], expiration: { days: rule[:expiration][:days] }, prefix: rule[:prefix], status: rule[:status] }
+        end
+      rescue Aws::S3::Errors::NoSuchLifecycleConfiguration
+        []
       end
     end
 
     def process_s3_object_key_format
       %W(%{uuid} %{uuid:random} %{uuid:hostname} %{uuid:timestamp}).each { |ph|