Support placeholder for s3_bucket

Closes: #350

e.g.

 s3_bucket test_${tag}
 <buffer tag>
  @type file
  path foo
 </buffer>

lib/fluent/plugin/out_s3.rb

@@ -249,11 +249,17 @@ def start
 
       s3_client = Aws::S3::Client.new(options)
       @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
+      @s3_bucket_template = @s3_bucket
+
+      unless @s3_bucket_template.match?(CHUNK_TAG_PLACEHOLDER_PATTERN)
+        @bucket = @s3.bucket(@s3_bucket)
+        check_apikeys if @check_apikey_on_start
+        ensure_bucket if @check_bucket
+        ensure_bucket_lifecycle
+        @dynamic_bucket = false
+      else
+        @dynamic_bucket = true
+      end
 
       super
     end
@@ -264,6 +270,12 @@ def format(tag, time, record)
     end
 
     def write(chunk)
+      if @dynamic_bucket
+        @s3_bucket = extract_placeholders(@s3_bucket_template, chunk)
+        @bucket = @s3.bucket(@s3_bucket)
+        ensure_bucket if @check_bucket
+        ensure_bucket_lifecycle
+      end
       i = 0
       metadata = chunk.metadata
       previous_path = nil

test/test_out_s3.rb

@@ -150,6 +150,13 @@ def test_configure_with_grant
     assert_equal "id='3456789012'", d.instance.grant_write_acp
   end
 
+  def test_configure_with_s3_bucket_tag
+    conf = CONFIG.clone
+    conf.sub!(/s3_bucket test_bucket/, "s3_bucket test_bucket_\${tag}")
+    d = create_driver(conf)
+    assert_equal "test_bucket_\${tag}", d.instance.s3_bucket
+  end
+
   def test_format
     d = create_driver
 
@@ -411,6 +418,22 @@ def test_write_with_custom_s3_object_key_format_containing_hex_random_placeholde
     FileUtils.rm_f(s3_local_file_path)
   end
 
+  def test_write_with_custom_s3_bucket_placeholder
+    config = CONFIG_TIME_SLICE.gsub(/s3_bucket test_bucket/,"s3_bucket test_bucket_\${tag}")
+    config << "<buffer tag>\n@type file\npath tmp\n</buffer>"
+
+    setup_mocks(true)
+
+    d = create_time_sliced_driver(config)
+    time = event_time("2011-01-02 13:14:15 UTC")
+    d.run(default_tag: "test") do
+      d.feed(time, {"a"=>1})
+      d.feed(time, {"a"=>2})
+    end
+
+    assert_equal "test_bucket_test", d.instance.s3_bucket
+  end
+
   class MockResponse
     attr_reader :data