in_systemd: pass storage conf to storage_create to support named storage

lib/fluent/plugin/in_systemd.rb

@@ -50,7 +50,8 @@ class SystemdInput < Input # rubocop:disable Metrics/ClassLength
       def configure(conf)
         super
         @journal = nil
-        @pos_storage = storage_create(usage: 'positions')
+        config = conf.elements.find { |e| e.name == 'storage' }
+        @pos_storage = storage_create(usage: 'positions', conf: config, default_type: DEFAULT_STORAGE_TYPE)
         @mutator = SystemdEntryMutator.new(**@entry_opts.to_h)
         @mutator.warnings.each { |warning| log.warn(warning) }
       end

test/plugin/test_in_systemd.rb

@@ -268,4 +268,58 @@ def test_reading_from_a_journal_with_corrupted_entries
     # Since libsystemd v250, it can read this corrupted record.
     assert { d.events.size == 460 or d.events.size == 461 }
   end
+
+  def test_configure_storage_with_conf_arg_does_not_raise
+    config = base_config + %(
+      @id in_systemd
+      <storage system_cursor>
+        @type local
+        persistent true
+      </storage>
+    )
+    Dir.mktmpdir('root_dir') do |root_dir|
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => root_dir) do
+        assert_nothing_raised do
+          create_driver(config)
+        end
+      end
+    end
+  end
+
+  def test_storage_path_reflects_conf_arg
+    config = base_config + %(
+      @id in_systemd
+      <storage system_cursor>
+        @type local
+        persistent true
+      </storage>
+    )
+    Dir.mktmpdir('root_dir') do |root_dir|
+      expected = File.join(root_dir, 'worker0', 'in_systemd', 'storage.system_cursor.json')
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => root_dir) do
+        d = create_driver(config)
+        d.run(expect_emits: 1)
+        assert_path_exist expected
+      end
+    end
+  end
+
+  def test_storage_path_without_conf_arg
+    config = base_config + %(
+      read_from_head true
+      @id in_systemd
+      <storage>
+        @type local
+        persistent true
+      </storage>
+    )
+    Dir.mktmpdir('root_dir') do |root_dir|
+      expected = File.join(root_dir, 'worker0', 'in_systemd', 'storage.json')
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => root_dir) do
+        d = create_driver(config)
+        d.run(expect_emits: 1)
+        assert_path_exist expected
+      end
+    end
+  end
 end