Fix: sincedb_clean_after not being respected (#276)

In certain cases, in read mode sincedb is not updated.

There was no periodic check for sincedb updates that would cause the sincedb_clean_after entries to cleanup.
The cleanup relied on new files being discovered or new content being added to existing files -> causing sincedb updates.

The fix here is to periodically flush sincedb (from the watch loop).
Besides, to make the process more deterministic, there's a minor change to make sure the same "updated" timestamp is used to mark the last changed time.

resolves #250
expected to also resolve #260

Author: kares

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.2.2
+  - Fix: sincedb_clean_after not being respected [#276](https://github.com/logstash-plugins/logstash-input-file/pull/276)
+
 ## 4.2.1
   - Fix: skip sincedb eviction if read mode completion deletes file during flush [#273](https://github.com/logstash-plugins/logstash-input-file/pull/273)
 

lib/filewatch/observing_base.rb

@@ -83,15 +83,5 @@ def quit
       # sincedb_write("shutting down")
     end
 
-    # close_file(path) is to be used by external code
-    # when it knows that it is completely done with a file.
-    # Other files or folders may still be being watched.
-    # Caution, once unwatched, a file can't be watched again
-    # unless a new instance of this class begins watching again.
-    # The sysadmin should rename, move or delete the file.
-    def close_file(path)
-      @watch.unwatch(path)
-      sincedb_write
-    end
   end
 end

lib/filewatch/sincedb_collection.rb

@@ -180,17 +180,17 @@ def watched_file_unset?(key)
       get(key).watched_file.nil?
     end
 
-    private
-
     def flush_at_interval
-      now = Time.now.to_i
-      delta = now - @sincedb_last_write
+      now = Time.now
+      delta = now.to_i - @sincedb_last_write
       if delta >= @settings.sincedb_write_interval
         logger.debug("writing sincedb (delta since last write = #{delta})")
         sincedb_write(now)
       end
     end
 
+    private
+
     def handle_association(sincedb_value, watched_file)
       watched_file.update_bytes_read(sincedb_value.position)
       sincedb_value.set_watched_file(watched_file)
@@ -210,33 +210,33 @@ def set_key_value(key, value)
       end
     end
 
-    def sincedb_write(time = Time.now.to_i)
-      logger.trace("sincedb_write: to: #{path}")
+    def sincedb_write(time = Time.now)
+      logger.trace("sincedb_write: #{path} (time = #{time})")
       begin
-        @write_method.call
+        @write_method.call(time)
         @serializer.expired_keys.each do |key|
           @sincedb[key].unset_watched_file
           delete(key)
           logger.trace? && logger.trace("sincedb_write: cleaned", :key => key)
         end
-        @sincedb_last_write = time
+        @sincedb_last_write = time.to_i
         @write_requested = false
       rescue Errno::EACCES
         # no file handles free perhaps
         # maybe it will work next time
-        logger.trace("sincedb_write: error: #{path}: #{$!}")
+        logger.trace("sincedb_write: #{path} error: #{$!}")
       end
     end
 
-    def atomic_write
+    def atomic_write(time)
       FileHelper.write_atomically(@full_path) do |io|
-        @serializer.serialize(@sincedb, io)
+        @serializer.serialize(@sincedb, io, time.to_f)
       end
     end
 
-    def non_atomic_write
+    def non_atomic_write(time)
       IO.open(IO.sysopen(@full_path, "w+")) do |io|
-        @serializer.serialize(@sincedb, io)
+        @serializer.serialize(@sincedb, io, time.to_f)
       end
     end
   end

lib/filewatch/watch.rb

@@ -51,7 +51,10 @@ def subscribe(observer, sincedb_collection)
           glob = 0
         end
         break if quit?
+        # NOTE: maybe the plugin should validate stat_interval <= sincedb_write_interval <= sincedb_clean_after
         sleep(@settings.stat_interval)
+        # we need to check potential expired keys (sincedb_clean_after) periodically
+        sincedb_collection.flush_at_interval
       end
       sincedb_collection.write_if_requested # does nothing if no requests to write were lodged.
       @watched_files_collection.close_all

logstash-input-file.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-file'
-  s.version         = '4.2.1'
+  s.version         = '4.2.2'
   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"

spec/inputs/file_read_spec.rb

@@ -301,25 +301,69 @@
       watched_files = plugin.watcher.watch.watched_files_collection
       expect( watched_files ).to be_empty
     end
+  end
 
-    private
+  describe 'sincedb cleanup' do
 
-    def wait_for_start_processing(run_thread, timeout: 1.0)
-      begin
-        Timeout.timeout(timeout) do
-          sleep(0.01) while run_thread.status != 'sleep'
-          sleep(timeout) unless plugin.queue
-        end
-      rescue Timeout::Error
-        raise "plugin did not start processing (timeout: #{timeout})" unless plugin.queue
-      else
-        raise "plugin did not start processing" unless plugin.queue
+    let(:options) do
+      super.merge(
+          'sincedb_path' => sincedb_path,
+          'sincedb_clean_after' => '1.0 seconds',
+          'sincedb_write_interval' => 0.25,
+          'stat_interval' => 0.1,
+      )
+    end
+
+    let(:sincedb_path) { "#{temp_directory}/.sincedb" }
+
+    let(:sample_file) { File.join(temp_directory, "sample.txt") }
+
+    before do
+      plugin.register
+      @run_thread = Thread.new(plugin) do |plugin|
+        Thread.current.abort_on_exception = true
+        plugin.run queue
       end
+
+      File.open(sample_file, 'w') { |fd| fd.write("line1\nline2\n") }
+
+      wait_for_start_processing(@run_thread)
     end
 
-    def wait_for_file_removal(path, timeout: 3 * interval)
-      wait(timeout).for { File.exist?(path) }.to be_falsey
+    after { plugin.stop }
+
+    it 'cleans up sincedb entry' do
+      wait_for_file_removal(sample_file) # watched discovery
+
+      sincedb_content = File.read(sincedb_path).strip
+      expect( sincedb_content ).to_not be_empty
+
+      Stud.try(3.times) do
+        sleep(1.5) # > sincedb_clean_after
+
+        sincedb_content = File.read(sincedb_path).strip
+        expect( sincedb_content ).to be_empty
+      end
     end
 
   end
+
+  private
+
+  def wait_for_start_processing(run_thread, timeout: 1.0)
+    begin
+      Timeout.timeout(timeout) do
+        sleep(0.01) while run_thread.status != 'sleep'
+        sleep(timeout) unless plugin.queue
+      end
+    rescue Timeout::Error
+      raise "plugin did not start processing (timeout: #{timeout})" unless plugin.queue
+    else
+      raise "plugin did not start processing" unless plugin.queue
+    end
+  end
+
+  def wait_for_file_removal(path, timeout: 3 * interval)
+    wait(timeout).for { File.exist?(path) }.to be_falsey
+  end
 end