WIP: Better support for NFS, step 1 (#199)

* switch to loop controlled reading

* fix test

Fixes #189

Author: Guy Boertje

Diff for: lib/filewatch/bootstrap.rb

@@ -42,6 +42,7 @@ def to_s
   end
 
   BufferExtractResult = Struct.new(:lines, :warning, :additional)
+  LoopControlResult = Struct.new(:count, :size, :more)
 
   class NoSinceDBPathGiven < StandardError; end
 

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

@@ -5,42 +5,51 @@ class ReadFile < Base
     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)
-        changed = false
-        logger.trace("reading...", "amount" => watched_file.read_bytesize_description, "filename" => watched_file.filename)
-        watched_file.read_loop_count.times do
+        loop do
           break if quit?
-          begin
-            # expect BufferExtractResult
-            result = watched_file.read_extract_lines
-            # read_extract_lines will increment bytes_read
-            logger.trace(result.warning, result.additional) unless result.warning.empty?
-            changed = true
-            result.lines.each do |line|
-              watched_file.listener.accept(line)
-              # sincedb position is independent from the watched_file bytes_read
-              sincedb_collection.increment(watched_file.sincedb_key, line.bytesize + @settings.delimiter_byte_size)
-            end
-            sincedb_collection.request_disk_flush
-          rescue EOFError
-            # flush the buffer now in case there is no final delimiter
-            line = watched_file.buffer.flush
-            watched_file.listener.accept(line) unless line.empty?
-            watched_file.listener.eof
-            watched_file.file_close
-            key = watched_file.sincedb_key
-            sincedb_collection.reading_completed(key)
-            sincedb_collection.clear_watched_file(key)
-            watched_file.listener.deleted
-            watched_file.unwatch
-            break
-          rescue Errno::EWOULDBLOCK, Errno::EINTR
-            watched_file.listener.error
-            break
-          rescue => e
-            logger.error("read_to_eof: general error reading file", "path" => watched_file.path, "error" => e.inspect, "backtrace" => e.backtrace.take(8))
-            watched_file.listener.error
-            break
+          loop_control = watched_file.loop_control_adjusted_for_stat_size
+          controlled_read(watched_file, loop_control)
+          sincedb_collection.request_disk_flush
+          break unless loop_control.more
+        end
+        if watched_file.all_read?
+          # flush the buffer now in case there is no final delimiter
+          line = watched_file.buffer.flush
+          watched_file.listener.accept(line) unless line.empty?
+          watched_file.listener.eof
+          watched_file.file_close
+          key = watched_file.sincedb_key
+          sincedb_collection.reading_completed(key)
+          sincedb_collection.clear_watched_file(key)
+          watched_file.listener.deleted
+          watched_file.unwatch
+        end
+      end
+    end
+
+    def controlled_read(watched_file, loop_control)
+      logger.trace("reading...", "iterations" => loop_control.count, "amount" => loop_control.size, "filename" => watched_file.filename)
+      loop_control.count.times do
+        begin
+          result = watched_file.read_extract_lines(loop_control.size) # expect BufferExtractResult
+          logger.info(result.warning, result.additional) unless result.warning.empty?
+          result.lines.each do |line|
+            watched_file.listener.accept(line)
+            # sincedb position is independent from the watched_file bytes_read
+            delta = line.bytesize + @settings.delimiter_byte_size
+            sincedb_collection.increment(watched_file.sincedb_key, delta)
           end
+        rescue EOFError
+          logger.error("controlled_read: eof error reading file", "path" => watched_file.path, "error" => e.inspect, "backtrace" => e.backtrace.take(8))
+          break
+        rescue Errno::EWOULDBLOCK, Errno::EINTR
+          logger.error("controlled_read: block or interrupt error reading file", "path" => watched_file.path, "error" => e.inspect, "backtrace" => e.backtrace.take(8))
+          watched_file.listener.error
+          break
+        rescue => e
+          logger.error("controlled_read: general error reading file", "path" => watched_file.path, "error" => e.inspect, "backtrace" => e.backtrace.take(8))
+          watched_file.listener.error
+          break
         end
       end
     end

Diff for: lib/filewatch/tail_mode/handlers/base.rb

@@ -30,19 +30,19 @@ def update_existing_specifically(watched_file, sincedb_value)
 
     private
 
-    def read_to_eof(watched_file)
-      logger.trace("reading...", "amount" => watched_file.read_bytesize_description, "filename" => watched_file.filename)
+    def controlled_read(watched_file, loop_control)
       changed = false
+      logger.trace("reading...", "iterations" => loop_control.count, "amount" => loop_control.size, "filename" => watched_file.filename)
       # from a real config (has 102 file inputs)
       # -- This cfg creates a file input for every log file to create a dedicated file pointer and read all file simultaneously
       # -- If we put all log files in one file input glob we will have indexing delay, because Logstash waits until the first file becomes EOF
       # by allowing the user to specify a combo of `file_chunk_count` X `file_chunk_size`...
       # we enable the pseudo parallel processing of each file.
       # user also has the option to specify a low `stat_interval` and a very high `discover_interval`to respond
       # quicker to changing files and not allowing too much content to build up before reading it.
-      watched_file.read_loop_count.times do
+      loop_control.count.times do
         begin
-          result = watched_file.read_extract_lines # expect BufferExtractResult
+          result = watched_file.read_extract_lines(loop_control.size) # expect BufferExtractResult
           logger.trace(result.warning, result.additional) unless result.warning.empty?
           changed = true
           result.lines.each do |line|

Diff for: lib/filewatch/tail_mode/handlers/grow.rb

@@ -4,7 +4,11 @@ module FileWatch module TailMode module Handlers
   class Grow < Base
     def handle_specifically(watched_file)
       watched_file.file_seek(watched_file.bytes_read)
-      read_to_eof(watched_file)
+      loop do
+        loop_control = watched_file.loop_control_adjusted_for_stat_size
+        controlled_read(watched_file, loop_control)
+        break unless loop_control.more
+      end
     end
   end
 end end end

Diff for: lib/filewatch/tail_mode/handlers/shrink.rb

@@ -3,10 +3,13 @@
 module FileWatch module TailMode module Handlers
   class Shrink < Base
     def handle_specifically(watched_file)
-      sdbv = add_or_update_sincedb_collection(watched_file)
+      add_or_update_sincedb_collection(watched_file)
       watched_file.file_seek(watched_file.bytes_read)
-      read_to_eof(watched_file)
-      logger.trace("handle_specifically: after read_to_eof", "watched file" => watched_file.details, "sincedb value" => sdbv)
+      loop do
+        loop_control = watched_file.loop_control_adjusted_for_stat_size
+        controlled_read(watched_file, loop_control)
+        break unless loop_control.more
+      end
     end
 
     def update_existing_specifically(watched_file, sincedb_value)

Diff for: lib/filewatch/tail_mode/processor.rb

@@ -155,9 +155,9 @@ def process_rotation_in_progress(watched_files)
             # we need to keep reading the open file, if we close it we lose it because the path is now pointing at a different file.
             logger.trace(">>> Rotation In Progress - inode change detected and original content is not fully read, reading all", "watched_file details" => watched_file.details)
             # need to fully read open file while we can
-            watched_file.set_depth_first_read_loop
+            watched_file.set_maximum_read_loop
             grow(watched_file)
-            watched_file.set_user_defined_read_loop
+            watched_file.set_standard_read_loop
           else
             logger.warn(">>> Rotation In Progress - inode change detected and original content is not fully read, file is closed and path points to new content", "watched_file details" => watched_file.details)
           end

Diff for: lib/filewatch/watched_file.rb

@@ -7,7 +7,8 @@ class WatchedFile
 
     attr_reader :bytes_read, :state, :file, :buffer, :recent_states, :bytes_unread
     attr_reader :path, :accessed_at, :modified_at, :pathname, :filename
-    attr_reader :listener, :read_loop_count, :read_chunk_size, :stat, :read_bytesize_description
+    attr_reader :listener, :read_loop_count, :read_chunk_size, :stat
+    attr_reader :loop_count_type, :loop_count_mode
     attr_accessor :last_open_warning_at
 
     # this class represents a file that has been discovered
@@ -19,7 +20,7 @@ def initialize(pathname, stat, settings)
       @filename = @pathname.basename.to_s
       full_state_reset(stat)
       watch
-      set_user_defined_read_loop
+      set_standard_read_loop
       set_accessed_at
     end
 
@@ -52,7 +53,7 @@ def full_state_reset(this_stat = nil)
 
     def rotate_from(other)
       # move all state from other to this one
-      set_user_defined_read_loop
+      set_standard_read_loop
       file_close
       @bytes_read = other.bytes_read
       @bytes_unread = other.bytes_unread
@@ -231,8 +232,8 @@ def reset_buffer
       @buffer.flush
     end
 
-    def read_extract_lines
-      data = file_read
+    def read_extract_lines(amount)
+      data = file_read(amount)
       result = buffer_extract(data)
       increment_bytes_read(data.bytesize)
       result
@@ -342,16 +343,39 @@ def expiry_ignore_enabled?
       !@settings.ignore_older.nil?
     end
 
-    def set_depth_first_read_loop
-      @read_loop_count = FileWatch::MAX_ITERATIONS
-      @read_chunk_size = FileWatch::FILE_READ_SIZE
-      @read_bytesize_description = "All"
-    end
-
-    def set_user_defined_read_loop
+    def set_standard_read_loop
       @read_loop_count = @settings.file_chunk_count
       @read_chunk_size = @settings.file_chunk_size
-      @read_bytesize_description = @read_loop_count == FileWatch::MAX_ITERATIONS ? "All" : (@read_loop_count * @read_chunk_size).to_s
+      # e.g. 1 * 10 bytes -> 10 or 256 * 65536 -> 1677716 or 140737488355327 * 32768 -> 4611686018427355136
+      @standard_loop_max_bytes = @read_loop_count * @read_chunk_size
+    end
+
+    def set_maximum_read_loop
+      # used to quickly fully read an open file when rotation is detected
+      @read_loop_count = FileWatch::MAX_ITERATIONS
+      @read_chunk_size = FileWatch::FILE_READ_SIZE
+      @standard_loop_max_bytes = @read_loop_count * @read_chunk_size
+    end
+
+    def loop_control_adjusted_for_stat_size
+      more = false
+      to_read = current_size - @bytes_read
+      return LoopControlResult.new(0, 0, more) if to_read < 1
+      return LoopControlResult.new(1, to_read, more) if to_read < @read_chunk_size
+      # set as if to_read is greater than or equal to max_bytes
+      # use the ones from settings and don't indicate more
+      count = @read_loop_count
+      if to_read < @standard_loop_max_bytes
+        # if the defaults are used then this branch will be taken
+        # e.g. to_read is 100 and max_bytes is 4 * 30 -> 120
+        # will overrun and trigger EOF, build less iterations
+        # will generate 3 * 30 -> 90 this time and we indicate more
+        # a 2GB file in read mode will get one loop of 64666 x 32768 (2119006656 / 32768)
+        # and a second loop with 1 x 31168
+        count = to_read / @read_chunk_size
+        more = true
+      end
+      LoopControlResult.new(count, @read_chunk_size, more)
     end
 
     def reset_bytes_unread

Diff for: spec/filewatch/read_mode_handlers_read_file_spec.rb

@@ -20,7 +20,7 @@ module FileWatch
       let(:watch) { double("watch", :quit? => false) }
       it "calls 'sincedb_write' exactly 2 times" do
         allow(FileOpener).to receive(:open).with(watched_file.path).and_return(file)
-        expect(sdb_collection).to receive(:sincedb_write).exactly(2).times
+        expect(sdb_collection).to receive(:sincedb_write).exactly(1).times
         watched_file.activate
         processor.initialize_handlers(sdb_collection, TestObserver.new)
         processor.read_file(watched_file)