out_s3.rb in fluent-plugin-s3-1.7.0

- old
+ new

@@ -145,10 +145,12 @@
     config_param :compute_checksums, :bool, default: nil # use nil to follow SDK default configuration
     desc "Signature version for API Request (s3,v4)"
     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 tag-set for the object"
+    config_param :tagging, :string, 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
@@ -358,10 +360,11 @@
         put_options[:acl] = @acl if @acl
         put_options[:grant_full_control] = @grant_full_control if @grant_full_control
         put_options[:grant_read] = @grant_read if @grant_read
         put_options[:grant_read_acp] = @grant_read_acp if @grant_read_acp
         put_options[:grant_write_acp] = @grant_write_acp if @grant_write_acp
+        put_options[:tagging] = @tagging if @tagging
 
         if @s3_metadata
           put_options[:metadata] = {}
           @s3_metadata.each do |k, v|
             put_options[:metadata][k] = extract_placeholders(v, chunk).gsub(%r(%{[^}]+}), {"%{index}" => sprintf(@index_format, i - 1)})
@@ -459,11 +462,12 @@
     def check_s3_path_safety(conf)
       unless conf.has_key?('s3_object_key_format')
         log.warn "The default value of s3_object_key_format will use ${chunk_id} instead of %{index} to avoid object conflict in v2"
       end
 
-      if (@buffer_config.flush_thread_count > 1) && ['${chunk_id}', '%{uuid_flush}'].none? { |key| @s3_object_key_format.include?(key) }
-        log.warn "No ${chunk_id} or %{uuid_flush} in s3_object_key_format with multiple flush threads. Recommend to set ${chunk_id} or %{uuid_flush} to avoid data lost by object conflict"
+      is_working_on_parallel = @buffer_config.flush_thread_count > 1 || system_config.workers > 1
+      if is_working_on_parallel && ['${chunk_id}', '%{uuid_flush}'].none? { |key| @s3_object_key_format.include?(key) }
+        log.warn "No ${chunk_id} or %{uuid_flush} in s3_object_key_format with multiple flush threads or multiple workers. Recommend to set ${chunk_id} or %{uuid_flush} to avoid data lost by object conflict"
       end
     end
 
     def check_apikeys
       @bucket.objects(prefix: @path, :max_keys => 1).first