buf_file/buf_file_single: fix to ignore placeholder based path. fix #2582

lib/fluent/plugin/buf_file.rb

@@ -132,7 +132,7 @@ def resume
 
         patterns = [@path]
         patterns.unshift @additional_resume_path if @additional_resume_path
-        Dir.glob(patterns) do |path|
+        Dir.glob(escaped_patterns(patterns)) do |path|
           next unless File.file?(path)
 
           log.debug { "restoring buffer file: path = #{path}" }
@@ -183,6 +183,15 @@ def handle_broken_files(path, mode, e)
         # After support 'backup_dir' feature, these files are moved to backup_dir instead of unlink.
         File.unlink(path, path + '.meta') rescue nil
       end
+
+      private
+
+      def escaped_patterns(patterns)
+        patterns.map { |pattern|
+          # '{' '}' are special character in Dir.glob
+          pattern.gsub(/[\{\}]/) { |c| "\\#{c}" }
+        }
+      end
     end
   end
 end

lib/fluent/plugin/buf_file_single.rb

@@ -155,7 +155,7 @@ def resume
 
         patterns = [@path]
         patterns.unshift @additional_resume_path if @additional_resume_path
-        Dir.glob(patterns) do |path|
+        Dir.glob(escaped_patterns(patterns)) do |path|
           next unless File.file?(path)
 
           log.debug { "restoring buffer file: path = #{path}" }
@@ -206,6 +206,15 @@ def handle_broken_files(path, mode, e)
         # After support 'backup_dir' feature, these files are moved to backup_dir instead of unlink.
         File.unlink(path) rescue nil
       end
+
+      private
+
+      def escaped_patterns(patterns)
+        patterns.map { |pattern|
+          # '{' '}' are special character in Dir.glob
+          pattern.gsub(/[\{\}]/) { |c| "\\#{c}" }
+        }
+      end
     end
   end
 end

test/plugin/test_buf_file.rb

@@ -537,6 +537,60 @@ def write_metadata(path, chunk_id, metadata, size, ctime, mtime)
     end
   end
 
+  sub_test_case 'there are some existing file chunks with placeholders path' do
+    setup do
+      @bufdir = File.expand_path('../../tmp/buffer_${test}_file', __FILE__)
+      FileUtils.rm_rf(@bufdir)
+      FileUtils.mkdir_p(@bufdir)
+
+      @c1id = Fluent::UniqueId.generate
+      p1 = File.join(@bufdir, "etest.q#{Fluent::UniqueId.hex(@c1id)}.log")
+      File.open(p1, 'wb') do |f|
+        f.write ["t1.test", event_time('2016-04-17 13:58:15 -0700').to_i, {"message" => "yay"}].to_json + "\n"
+      end
+      write_metadata(
+        p1 + '.meta', @c1id, metadata(timekey: event_time('2016-04-17 13:58:00 -0700').to_i),
+        1, event_time('2016-04-17 13:58:00 -0700').to_i, event_time('2016-04-17 13:58:22 -0700').to_i
+      )
+
+      @c2id = Fluent::UniqueId.generate
+      p2 = File.join(@bufdir, "etest.b#{Fluent::UniqueId.hex(@c2id)}.log")
+      File.open(p2, 'wb') do |f|
+        f.write ["t1.test", event_time('2016-04-17 14:00:15 -0700').to_i, {"message" => "yay"}].to_json + "\n"
+      end
+      write_metadata(
+        p2 + '.meta', @c2id, metadata(timekey: event_time('2016-04-17 14:00:00 -0700').to_i),
+        1, event_time('2016-04-17 14:00:00 -0700').to_i, event_time('2016-04-17 14:00:28 -0700').to_i
+      )
+
+      @bufpath = File.join(@bufdir, 'etest.*.log')
+
+      Fluent::Test.setup
+      @d = FluentPluginFileBufferTest::DummyOutputPlugin.new
+      @p = Fluent::Plugin::FileBuffer.new
+      @p.owner = @d
+      @p.configure(config_element('buffer', '', {'path' => @bufpath}))
+      @p.start
+    end
+
+    teardown do
+      if @p
+        @p.stop unless @p.stopped?
+        @p.before_shutdown unless @p.before_shutdown?
+        @p.shutdown unless @p.shutdown?
+        @p.after_shutdown unless @p.after_shutdown?
+        @p.close unless @p.closed?
+        @p.terminate unless @p.terminated?
+      end
+      FileUtils.rm_rf(@bufdir)
+    end
+
+    test '#resume returns staged/queued chunks with metadata' do
+      assert_equal 1, @p.stage.size
+      assert_equal 1, @p.queue.size
+    end
+  end
+
   sub_test_case 'there are some existing file chunks, both in specified path and per-worker directory under specified path, configured as multi workers' do
     setup do
       @bufdir = File.expand_path('../../tmp/buffer_file/path', __FILE__)

test/plugin/test_buf_file_single.rb

@@ -502,6 +502,54 @@ def create_driver(conf = TAG_CONF, klass = FluentPluginFileSingleBufferTest::Dum
     end
   end
 
+  sub_test_case 'there are some existing file chunks with placeholders path' do
+    setup do
+      @buf_ph_dir = File.expand_path('../../tmp/buffer_${test}_file_single_dir', __FILE__)
+      FileUtils.rm_rf(@buf_ph_dir)
+      FileUtils.mkdir_p(@buf_ph_dir)
+
+      @c1id = Fluent::UniqueId.generate
+      p1 = File.join(@buf_ph_dir, "fsb.testing.q#{Fluent::UniqueId.hex(@c1id)}.buf")
+      File.open(p1, 'wb') do |f|
+        f.write ["t1.test", event_time('2016-04-17 13:58:15 -0700').to_i, {"message" => "yay"}].to_json + "\n"
+      end
+      t = Time.now - 50000
+      File.utime(t, t, p1)
+
+      @c2id = Fluent::UniqueId.generate
+      p2 = File.join(@buf_ph_dir, "fsb.testing.b#{Fluent::UniqueId.hex(@c2id)}.buf")
+      File.open(p2, 'wb') do |f|
+        f.write ["t1.test", event_time('2016-04-17 14:00:15 -0700').to_i, {"message" => "yay"}].to_json + "\n"
+      end
+    end
+
+    teardown do
+      if @p
+        @p.stop unless @p.stopped?
+        @p.before_shutdown unless @p.before_shutdown?
+        @p.shutdown unless @p.shutdown?
+        @p.after_shutdown unless @p.after_shutdown?
+        @p.close unless @p.closed?
+        @p.terminate unless @p.terminated?
+      end
+      FileUtils.rm_rf(@buf_ph_dir)
+    end
+
+    test '#resume returns staged/queued chunks with metadata' do
+      @d = create_driver(%[
+        <buffer tag>
+          @type file_single
+          path #{@buf_ph_dir}
+        </buffer>
+      ])
+      @p = @d.instance.buffer
+      @p.start
+
+      assert_equal 1, @p.stage.size
+      assert_equal 1, @p.queue.size
+    end
+  end
+
   sub_test_case 'there are some existing msgpack file chunks' do
     setup do
       packer = Fluent::MessagePackFactory.packer