Fix ReadFile handler to consider the value stored in sincedb on plugin restart (#307)

Fixes read mode to restart the read from reference stored in sincedb in case the file wasn't completely consumed

Update the file pointer of a read mode file to the max between the read bytes or the sincedb reference for the same file.
This solves a problem, that when a pipeline is restarted, it's able to recover from the last known reference, without restarting from the beginning, and reprocessing already processed lines.

Co-authored-by: João Duarte <[email protected]>

Author: andsel

Diff for: CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.4.3
+  - Fixes read mode to restart the read from reference stored in sincedb in case the file wasn't completely consumed. [#307](https://github.com/logstash-plugins/logstash-input-file/pull/307)
+
 ## 4.4.2
   - Doc: Fix attribute by removing extra character [#310](https://github.com/logstash-plugins/logstash-input-file/pull/310)
 

Diff for: lib/filewatch/read_mode/handlers/read_file.rb

@@ -2,9 +2,19 @@
 
 module FileWatch module ReadMode module Handlers
   class ReadFile < Base
+
+    # seek file to which ever is furthest: either current bytes read or sincedb position
+    private
+    def seek_to_furthest_position(watched_file)
+      previous_pos = sincedb_collection.find(watched_file).position
+      watched_file.file_seek([watched_file.bytes_read, previous_pos].max)
+    end
+
+    public
     def handle_specifically(watched_file)
       if open_file(watched_file)
         add_or_update_sincedb_collection(watched_file) unless sincedb_collection.member?(watched_file.sincedb_key)
+        seek_to_furthest_position(watched_file)
         loop do
           break if quit?
           loop_control = watched_file.loop_control_adjusted_for_stat_size

Diff for: logstash-input-file.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-file'
-  s.version         = '4.4.2'
+  s.version         = '4.4.3'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Streams events from files"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

Diff for: spec/filewatch/read_mode_handlers_read_file_spec.rb

@@ -36,5 +36,45 @@ module FileWatch
         processor.read_file(watched_file)
       end
     end
+
+    context "when restart from existing sincedb" do
+      let(:settings) do
+        Settings.from_options(
+          :sincedb_write_interval => 0,
+          :sincedb_path => File::NULL,
+          :file_chunk_size => 10
+        )
+      end
+
+      let(:processor) { double("fake processor") }
+      let(:observer) { TestObserver.new }
+      let(:watch) { double("watch") }
+
+      before(:each) {
+        allow(watch).to receive(:quit?).and_return(false)#.and_return(false).and_return(true)
+        allow(processor).to receive(:watch).and_return(watch)
+      }
+
+      it "read from where it left" do
+        listener = observer.listener_for(Pathname.new(pathname).to_path)
+        sut = ReadMode::Handlers::ReadFile.new(processor, sdb_collection, observer, settings)
+
+        # simulate a previous partial read of the file
+        sincedb_value = SincedbValue.new(0)
+        sincedb_value.set_watched_file(watched_file)
+        sdb_collection.set(watched_file.sincedb_key, sincedb_value)
+
+
+        # simulate a consumption of first line, (size + newline) bytes
+        sdb_collection.increment(watched_file.sincedb_key, File.readlines(pathname)[0].size + 2)
+
+        # exercise
+        sut.handle(watched_file)
+
+        # verify
+        expect(listener.lines.size).to eq(1)
+        expect(listener.lines[0]).to start_with("2010-03-12   23:51:21   SEA4   192.0.2.222   play   3914   OK")
+      end
+    end
   end
 end

Diff for: spec/filewatch/spec_helper.rb

@@ -80,6 +80,8 @@ def sysread(amount)
       multiplier = amount / string.length
       string * multiplier
     end
+    def sysseek(offset, whence)
+    end
   end
 
   FIXTURE_DIR = File.join('spec', 'fixtures')

Diff for: spec/helpers/spec_helper.rb

@@ -31,7 +31,13 @@ def initialize
 
     def trace_for(symbol)
       params = @tracer.map {|k,v| k == symbol ? v : nil}.compact
-      params.empty? ? false : params
+      if params.empty?
+        false
+      else
+        # merge all params with same key
+        # there could be multiple instances of same call, e.g. [[:accept, true], [:auto_flush, true], [:close, true], [:auto_flush, true]]
+        params.reduce {|b1, b2| b1 and b2}
+      end
     end
 
     def clear

Diff for: spec/inputs/file_tail_spec.rb

@@ -332,7 +332,7 @@
           .then("wait accept") do
             wait(0.75).for {
               subject.codec.identity_map[tmpfile_path].codec.trace_for(:accept)
-            }.to eq([true]), "accept didn't"
+            }.to eq(true), "accept didn't"
           end
           .then("request a stop") do
             # without this the subject.run doesn't invokes the #exit_flush which is the only @codec.flush_mapped invocation
@@ -341,12 +341,11 @@
           .then("wait for auto_flush") do
             wait(2).for {
               subject.codec.identity_map[tmpfile_path].codec.trace_for(:auto_flush)
-                .reduce {|b1, b2| b1 and b2} # there could be multiple instances of same call, e.g. [[:accept, true], [:auto_flush, true], [:close, true], [:auto_flush, true]]
             }.to eq(true), "autoflush didn't"
           end
         subject.run(events)
         actions.assert_no_errors
-        expect(subject.codec.identity_map[tmpfile_path].codec.trace_for(:accept)).to eq([true])
+        expect(subject.codec.identity_map[tmpfile_path].codec.trace_for(:accept)).to eq(true)
       end
     end