diff --git a/lib/fluent/plugin/storage_local.rb b/lib/fluent/plugin/storage_local.rb
index 399702f2b0..7ccad620db 100644
--- a/lib/fluent/plugin/storage_local.rb
+++ b/lib/fluent/plugin/storage_local.rb
@@ -64,7 +64,8 @@ def configure(conf)
             end
           end
         elsif root_dir = owner.plugin_root_dir
-          @path = File.join(root_dir, 'storage.json')
+          basename = (conf.arg && !conf.arg.empty?) ? "storage.#{conf.arg}.json" : "storage.json"
+          @path = File.join(root_dir, basename)
           @multi_workers_available = true
         else
           if @persistent
diff --git a/lib/fluent/plugin_helper/storage.rb b/lib/fluent/plugin_helper/storage.rb
index f3f6d92920..a41c594c47 100644
--- a/lib/fluent/plugin_helper/storage.rb
+++ b/lib/fluent/plugin_helper/storage.rb
@@ -59,7 +59,7 @@ def storage_create(usage: '', type: nil, conf: nil, default_type: nil)
                      conf = Hash[conf.map{|k,v| [k.to_s, v]}]
                      Fluent::Config::Element.new('storage', usage, conf, [])
                    when nil
-                     Fluent::Config::Element.new('storage', usage, {}, [])
+                     Fluent::Config::Element.new('storage', usage, {'@type' => type}, [])
                    else
                      raise ArgumentError, "BUG: conf must be a Element, Hash (or unspecified), but '#{conf.class}'"
                    end
diff --git a/test/plugin/test_storage_local.rb b/test/plugin/test_storage_local.rb
index bd00721b34..773e1aaa7b 100644
--- a/test/plugin/test_storage_local.rb
+++ b/test/plugin/test_storage_local.rb
@@ -186,6 +186,20 @@ class MyInput < Fluent::Plugin::Input
       assert_equal '2', @p.get('key1')
       assert_equal 4, @p.get('key2')
     end
+
+    test 'works with customized path by specified usage' do
+      root_dir = File.join(TMP_DIR, 'root')
+      expected_storage_path = File.join(root_dir, 'worker0', 'local_storage_test', 'storage.usage.json')
+      conf = config_element('ROOT', 'usage', {'@id' => 'local_storage_test'})
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => root_dir) do
+        @d.configure(conf)
+      end
+      @d.start
+      @p = @d.storage_create(usage: 'usage', type: 'local')
+
+      assert_equal expected_storage_path, @p.path
+      assert @p.store.empty?
+    end
   end
 
   sub_test_case 'configured with root-dir and plugin id, and multi workers' do