Added settings flag exit_after_read that let to terminate the pipeline once it red all files. Closes #212

Fixes #240

Author: BigYellowHammer

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 4.1.16
+  - Added configuration setting exit_after_read to read to EOF and terminate
+  the input [#240](https://github.com/logstash-plugins/logstash-input-file/pull/240)
+
 ## 4.1.15
   - Fixed bug in conversion of sincedb_clean_after setting [#257](https://github.com/logstash-plugins/logstash-input-file/pull/257)
 

docs/index.asciidoc

@@ -168,6 +168,7 @@ see <<plugins-{type}s-{plugin}-string_duration,string_duration>> for the details
 | <<plugins-{type}s-{plugin}-delimiter>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-discover_interval>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-exclude>> |<<array,array>>|No
+| <<plugins-{type}s-{plugin}-exit_after_read>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-file_chunk_count>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-file_chunk_size>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-file_completed_action>> |<<string,string>>, one of `["delete", "log", "log_and_delete"]`|No
@@ -241,6 +242,19 @@ In Tail mode, you might want to exclude gzipped files:
 [source,ruby]
     exclude => "*.gz"
 
+[id="plugins-{type}s-{plugin}-exit_after_read"]
+===== `exit_after_read`
+
+  * Value type is <<boolean,boolean>>
+  * Default value is `false`
+
+This option can be used in `read` mode to enforce closing all watchers when file gets read.
+Can be used in situation when content of the file is static and won't change during execution.
+When set to `true` it also disables active discovery of the files - only files that were in 
+the directories when process was started will be read.
+It supports `sincedb` entries. When file was processed once, then modified - next run will only
+read newly added entries.
+
 [id="plugins-{type}s-{plugin}-file_chunk_count"]
 ===== `file_chunk_count`
 

lib/filewatch/read_mode/processor.rb

@@ -103,10 +103,21 @@ def process_active(watched_files)
         else
           read_file(watched_file)
         end
+        
+        if @settings.exit_after_read
+          common_detach_when_allread(watched_file)
+        end
         # handlers take care of closing and unwatching
       end
     end
 
+    def common_detach_when_allread(watched_file)
+      watched_file.unwatch
+      watched_file.listener.reading_completed
+      deletable_filepaths << watched_file.path
+      logger.trace("Whole file read: #{watched_file.path}, removing from collection")
+    end
+
     def common_deleted_reaction(watched_file, action)
       # file has gone away or we can't read it anymore.
       watched_file.unwatch

lib/filewatch/settings.rb

@@ -8,6 +8,7 @@ class Settings
     attr_reader :exclude, :start_new_files_at, :file_chunk_count, :file_chunk_size
     attr_reader :sincedb_path, :sincedb_write_interval, :sincedb_expiry_duration
     attr_reader :file_sort_by, :file_sort_direction
+    attr_reader :exit_after_read
 
     def self.from_options(opts)
       new.add_options(opts)
@@ -50,6 +51,7 @@ def add_options(opts)
       @sincedb_expiry_duration =  @opts.fetch(:sincedb_clean_after)
       @file_sort_by = @opts[:file_sort_by]
       @file_sort_direction = @opts[:file_sort_direction]
+      @exit_after_read = @opts[:exit_after_read]
       self
     end
 

lib/filewatch/watch.rb

@@ -43,10 +43,11 @@ def subscribe(observer, sincedb_collection)
       reset_quit
       until quit?
         iterate_on_state
+        # Don't discover new files when files to read are known at the beginning
         break if quit?
         sincedb_collection.write_if_requested
         glob += 1
-        if glob == interval
+        if glob == interval && !@settings.exit_after_read
           discover
           glob = 0
         end
@@ -76,7 +77,10 @@ def quit
     end
 
     def quit?
-      @quit.true?
+      if @settings.exit_after_read
+        @exit = @watched_files_collection.empty?
+      end
+      @quit.true? || @exit
     end
 
     private

lib/logstash/inputs/file.rb

@@ -222,6 +222,11 @@ class File < LogStash::Inputs::Base
   # perhaps path + asc will help to achieve the goal of controlling the order of file ingestion
   config :file_sort_direction, :validate => ["asc", "desc"], :default => "asc"
 
+  # When in 'read' mode - this option is closing all file watchers when EOF is hit
+  # This option also disables discovery of new/changes files. It works only on files found at the beginning
+  # Sincedb still works, if you run LS once again after doing some changes - only new values will be read
+  config :exit_after_read, :validate => :boolean, :default => false
+
   public
 
   class << self
@@ -260,6 +265,7 @@ def register
       :file_chunk_size => @file_chunk_size,
       :file_sort_by => @file_sort_by,
       :file_sort_direction => @file_sort_direction,
+      :exit_after_read => @exit_after_read,
     }
 
     @completed_file_handlers = []
@@ -280,7 +286,7 @@ def register
         raise ArgumentError.new("The \"sincedb_path\" argument must point to a file, received a directory: \"#{@sincedb_path}\"")
       end
     end
-
+    
     @filewatch_config[:sincedb_path] = @sincedb_path
 
     @filewatch_config[:start_new_files_at] = @start_position.to_sym
@@ -301,6 +307,9 @@ def register
     end
 
     if tail_mode?
+      if @exit_after_read
+        raise ArgumentError.new('The "exit_after_read" setting only works when the "mode" is set to "read"')
+      end
       @watcher_class = FileWatch::ObservingTail
     else
       @watcher_class = FileWatch::ObservingRead

lib/logstash/inputs/file_listener.rb

@@ -21,6 +21,9 @@ def eof
     def error
     end
 
+    def reading_completed
+    end
+
     def timed_out
       input.codec.evict(path)
     end

logstash-input-file.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-file'
-  s.version         = '4.1.15'
+  s.version         = '4.1.16'
   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/filewatch/reading_spec.rb

@@ -147,6 +147,65 @@ module FileWatch
       end
     end
 
+    context "when watching a directory with files using exit_after_read" do
+      let(:opts) { super.merge(:exit_after_read => true, :max_open_files => 2) }
+      let(:file_path3) { ::File.join(directory, "3.log") }
+      let(:file_path4) { ::File.join(directory, "4.log") }
+      let(:file_path5) { ::File.join(directory, "5.log") }
+      let(:lines) { [] }
+      let(:observer) { TestObserver.new(lines) }
+      let(:listener3) { observer.listener_for(file_path3) }
+      let(:file_path6) { ::File.join(directory, "6.log") }
+      let(:listener6) { observer.listener_for(file_path6) }
+
+      it "the file is read" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        expect(listener3.lines).to eq(["line1", "line2"])
+      end
+      it "multiple files are read" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        File.open(file_path4, "w") { |file| file.write("line3\nline4\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        expect(listener3.lines.sort).to eq(["line1", "line2", "line3", "line4"])
+      end
+      it "multiple files are read even if max_open_files is smaller then number of files" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        File.open(file_path4, "w") { |file| file.write("line3\nline4\n") }
+        File.open(file_path5, "w") { |file| file.write("line5\nline6\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        expect(listener3.lines.sort).to eq(["line1", "line2", "line3", "line4", "line5", "line6"])
+      end
+      it "file as marked as reading_completed" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        expect(listener3.calls).to eq([:open, :accept, :accept, :eof, :delete, :reading_completed])
+      end
+      it "sincedb works correctly" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        sincedb_record_fields = File.read(sincedb_path).split(" ")
+        position_field_index = 3
+        expect(sincedb_record_fields[position_field_index]).to eq("12")
+      end
+      it "does not include new files added after start" do
+        File.open(file_path3, "w") { |file|  file.write("line1\nline2\n") }
+        reading.watch_this(watch_dir)
+        reading.subscribe(observer)
+        File.open(file_path6, "w") { |file|  file.write("foob\nbar\n") }
+        expect(listener3.lines).to eq(["line1", "line2"])
+        expect(listener3.calls).to eq([:open, :accept, :accept, :eof, :delete, :reading_completed])
+        expect(listener6.calls).to eq([])
+
+      end
+    
+    end
+
     describe "reading fixtures" do
       let(:directory) { FIXTURE_DIR }
       let(:actions) do

spec/filewatch/spec_helper.rb

@@ -152,6 +152,10 @@ def eof
       def timed_out
         @calls << :timed_out
       end
+
+      def reading_completed
+        @calls << :reading_completed
+      end
     end
 
     attr_reader :listeners