Fix: release watched files on complete (in read-mode)

not doing so leads to a steady increase in watched collection's size
over time (esp. in use-cases where user is pulling in new files).

the left-over file is never to be processed again - it's being deleted
anyway using the completion handler.

Author: kares

Diff for: lib/filewatch/watch.rb

@@ -67,7 +67,7 @@ def iterate_on_state
         watched_files = @watched_files_collection.values
         @processor.process_all_states(watched_files)
       ensure
-        @watched_files_collection.delete(@processor.deletable_filepaths)
+        @watched_files_collection.remove_paths(@processor.deletable_filepaths)
         @processor.deletable_filepaths.clear
       end
     end # def each

Diff for: lib/filewatch/watched_files_collection.rb

@@ -15,13 +15,16 @@ def add(watched_file)
       @sort_method.call
     end
 
-    def delete(paths)
-      Array(paths).each do |f|
-        index = @pointers.delete(f)
-        @files.delete_at(index)
-        refresh_pointers
+    def remove_paths(paths)
+      removed_files = Array(paths).map do |path|
+        if index = @pointers.delete(path)
+          watched_file = @files.delete_at(index)
+          refresh_pointers
+          watched_file
+        end
       end
       @sort_method.call
+      removed_files
     end
 
     def close_all

Diff for: lib/logstash/inputs/delete_completed_file_handler.rb

@@ -2,8 +2,13 @@
 
 module LogStash module Inputs
   class DeleteCompletedFileHandler
+    def initialize(watch)
+      @watch = watch
+    end
+
     def handle(path)
       Pathname.new(path).unlink rescue nil
+      @watch.watched_files_collection.remove_paths([path])
     end
   end
 end end

Diff for: lib/logstash/inputs/file.rb

@@ -247,6 +247,8 @@ def validate_value(value, validator)
     end
   end
 
+  attr_reader :queue, :watcher # used in specs
+
   def register
     require "addressable/uri"
     require "digest/md5"
@@ -274,8 +276,6 @@ def register
       :check_archive_validity => @check_archive_validity,
     }
 
-    @completed_file_handlers = []
-
     @path.each do |path|
       if Pathname.new(path).relative?
         raise ArgumentError.new("File paths must be absolute, relative path specified: #{path}")
@@ -319,12 +319,6 @@ def register
       @watcher_class = FileWatch::ObservingTail
     else
       @watcher_class = FileWatch::ObservingRead
-      if @file_completed_action.include?('log')
-        @completed_file_handlers << LogCompletedFileHandler.new(@file_completed_log_path)
-      end
-      if @file_completed_action.include?('delete')
-        @completed_file_handlers << DeleteCompletedFileHandler.new
-      end
     end
     @codec = LogStash::Codecs::IdentityMapCodec.new(@codec)
     @completely_stopped = Concurrent::AtomicBoolean.new
@@ -344,7 +338,19 @@ def start_processing
     # if the pipeline restarts this input,
     # make sure previous files are closed
     stop
+
     @watcher = @watcher_class.new(@filewatch_config)
+
+    @completed_file_handlers = []
+    if read_mode?
+      if @file_completed_action.include?('log')
+        @completed_file_handlers << LogCompletedFileHandler.new(@file_completed_log_path)
+      end
+      if @file_completed_action.include?('delete')
+        @completed_file_handlers << DeleteCompletedFileHandler.new(@watcher.watch)
+      end
+    end
+
     @path.each { |path| @watcher.watch_this(path) }
   end
 

Diff for: spec/filewatch/watched_files_collection_spec.rb

@@ -85,9 +85,9 @@ module FileWatch
         collection.add(wf3)
         expect(collection.keys).to eq([filepath1, filepath2, filepath3])
 
-        collection.delete([filepath2,filepath3])
+        collection.remove_paths([filepath2,filepath3])
         expect(collection.keys).to eq([filepath1])
-
+        expect(collection.values.size).to eq 1
       end
     end
 

Diff for: spec/inputs/file_read_spec.rb

@@ -247,4 +247,64 @@
       end
     end
   end
+
+  let(:temp_directory) { Stud::Temporary.directory }
+  let(:interval) { 0.1 }
+  let(:options) do
+    {
+        'mode' => "read",
+        'path' => "#{temp_directory}/*",
+        'stat_interval' => interval,
+        'discover_interval' => interval,
+        'sincedb_path' => "#{temp_directory}/.sincedb",
+        'sincedb_write_interval' => interval
+    }
+  end
+
+  let(:queue) { Queue.new }
+  let(:plugin) { LogStash::Inputs::File.new(options) }
+
+  describe 'delete on complete' do
+
+    let(:options) do
+      super.merge({ 'file_completed_action' => "delete", 'exit_after_read' => false })
+    end
+
+    let(:sample_file) { File.join(temp_directory, "sample.log") }
+
+    before do
+      plugin.register
+      Thread.new(plugin) do |plugin|
+        Thread.current.abort_on_exception = true
+        plugin.run queue
+      end
+
+      File.open(sample_file, 'w') { |fd| fd.write("sample-content\n") }
+    end
+
+    after { plugin.stop }
+
+    it 'processes a file' do
+      wait_for_watcher_discovery
+
+      expect( plugin.queue.size ).to eql 1
+      event = plugin.queue.pop
+      expect( event.get('message') ).to eql 'sample-content'
+    end
+
+    it 'removes watched file from collection' do
+      wait_for_watcher_discovery
+
+      watched_files = plugin.watcher.watch.watched_files_collection
+      expect( watched_files ).to be_empty
+    end
+
+    private
+
+    def wait_for_watcher_discovery(interval: self.interval)
+      sleep(3 * interval) # let the watched discover files
+      raise "plugin did not start processing" unless plugin.queue
+    end
+
+  end
 end