Use bucket based on placeholder

Author: rmontenegroo

lib/fluent/plugin/out_s3.rb

@@ -252,6 +252,13 @@ def start
       s3_client = Aws::S3::Client.new(options)
       @s3 = Aws::S3::Resource.new(client: s3_client)
 
+      if not @s3_bucket =~ /\$\{.*\}/
+        @bucket = @s3.bucket(@s3_bucket)
+        check_apikeys(@bucket) if @check_apikey_on_start
+        ensure_bucket(@bucket) if @check_bucket
+        ensure_bucket_lifecycle(@bucket)
+      end
+
       super
     end
 
@@ -270,41 +277,36 @@ def write(chunk)
                    else
                      @time_slice_with_tz.call(metadata.timekey)
                    end
-
+      
       bucket_name = nil
 
-      if @s3_bucket.match(/\$\{.*\}/)
-
-        bucket_name = extract_placeholders(@s3_bucket, chunk)
-
-        if bucket_name.match(/\$\{.*\}/)
-
-            log.warn "Trying to use @s3_bucket_fallback as a fallback bucket name"
-            
+      if @s3_bucket =~ /\$\{.*\}/
+        @s3_bucket.scan(/\$\{([^\$\{\}]+)\}/) do |placeholder|
+          placeholder = placeholder.join
+          if (not chunk.metadata.variables) or (not chunk.metadata.variables.keys.include?(placeholder.to_sym))
+            log.warn "There is no placeholder '#{placeholder}'" 
             if @s3_bucket_fallback
-              
               bucket_name = @s3_bucket_fallback
-            
+              log.warn "Using @s3_bucket_fallback ('#{@s3_bucket_fallback}') as a fallback bucket name."
+              break
             else
-
-              raise "It was possible to extract_placeholder from @s3_bucket and there is no @s3_bucket_fallback set"
-
+              raise "It was possible to extract placeholder '#{placeholder}' from chunk and @s3_bucket_fallback is not set."
             end
-
+          end
         end
-         
-      else
 
-        bucket_name = @s3_bucket
+        if not bucket_name
+          bucket_name = extract_placeholders(@s3_bucket, chunk)
+        end
 
+        bucket = @s3.bucket(bucket_name)
+        check_apikeys(bucket) if @check_apikey_on_start
+        ensure_bucket(bucket) if @check_bucket
+        ensure_bucket_lifecycle(bucket)
+      else
+        bucket = @bucket
       end
 
-      bucket = @s3.bucket(bucket_name)
-
-      check_apikeys(bucket) if @check_apikey_on_start
-      ensure_bucket(bucket) if @check_bucket
-      ensure_bucket_lifecycle(bucket)
-
       if @check_object
         begin
           @values_for_s3_object_chunk[chunk.unique_id] ||= {
@@ -465,7 +467,7 @@ def get_bucket_lifecycle_rules(bucket)
       rescue Aws::S3::Errors::NoSuchLifecycleConfiguration
         []
       end
-     end
+    end
 
     def process_s3_object_key_format
       %W(%{uuid} %{uuid:random} %{uuid:hostname} %{uuid:timestamp}).each { |ph|

test/test_out_s3.rb

@@ -44,10 +44,10 @@ def write(chunk)
 
       private
 
-      def ensure_bucket
+      def ensure_bucket(bucket)
       end
 
-      def check_apikeys
+      def check_apikeys(bucket)
       end
     end.configure(conf)
   end
@@ -287,7 +287,7 @@ def write(chunk)
 
       private
 
-      def check_apikeys
+      def check_apikeys(bucket)
       end
     end.configure(conf)
   end
@@ -427,7 +427,7 @@ def setup_mocks(exists_return = false)
     mock(Aws::S3::Client).new(anything).at_least(0) { @s3_client }
     @s3_resource = mock(Aws::S3::Resource.new(client: @s3_client))
     mock(Aws::S3::Resource).new(client: @s3_client) { @s3_resource }
-    @s3_bucket = mock(Aws::S3::Bucket.new(name: "test",
+    @s3_bucket = mock(Aws::S3::Bucket.new(name: "test_bucket",
                                           client: @s3_client))
     @s3_bucket.exists? { exists_return }
     @s3_object = mock(Aws::S3::Object.new(bucket_name: "test_bucket",