Merge pull request #3542 from majimenez-stratio/read_from_head

in_tail: Always read from head for new/rotated files after startup

lib/fluent/plugin/in_tail.rb

@@ -59,6 +59,7 @@ def initialize
       @ignore_list = []
       @shutdown_start_time = nil
       @metrics = nil
+      @startup = true
     end
 
     desc 'The paths to read. Multiple paths can be specified, separated by comma.'
@@ -381,11 +382,13 @@ def refresh_watchers
 
       stop_watchers(unwatched_hash, immediate: false, unwatched: true) unless unwatched_hash.empty?
       start_watchers(added_hash) unless added_hash.empty?
+      @startup = false if @startup
     end
 
     def setup_watcher(target_info, pe)
       line_buffer_timer_flusher = @multiline_mode ? TailWatcher::LineBufferTimerFlusher.new(log, @multiline_flush_interval, &method(:flush_buffer)) : nil
-      tw = TailWatcher.new(target_info, pe, log, @read_from_head, @follow_inodes, method(:update_watcher), line_buffer_timer_flusher, method(:io_handler), @metrics)
+      read_from_head = !@startup || @read_from_head
+      tw = TailWatcher.new(target_info, pe, log, read_from_head, @follow_inodes, method(:update_watcher), line_buffer_timer_flusher, method(:io_handler), @metrics)
 
       if @enable_watch_timer
         tt = TimerTrigger.new(1, log) { tw.on_notify }

test/plugin/test_in_tail.rb

@@ -651,6 +651,23 @@ def test_emit_with_disable_stat_watcher(data)
       assert_equal({"message" => "test3"}, events[0][2])
       assert_equal({"message" => "test4"}, events[1][2])
     end
+
+    def test_always_read_from_head_on_detecting_a_new_file
+      d = create_driver(SINGLE_LINE_CONFIG)
+
+      d.run(expect_emits: 1, timeout: 3) do
+        File.open("#{TMP_DIR}/tail.txt", "wb") {|f|
+          f.puts "test1\ntest2\n"
+        }
+      end
+
+      assert_equal(
+        [
+          {"message" => "test1"},
+          {"message" => "test2"},
+        ],
+        d.events.collect { |event| event[2] })
+    end
   end
 
   class TestWithSystem < self