Fix: watched files performance with huge filesets (logstash-plugins#268)

Plugin tracks all discovered files in a map-like collection
(`FileWatch::WatchedFilesCollection`).
The collection design was to re-sort all elements every time the
collection gets updated ... this obviously does not scale with 10000s of
watched files.
Starting up LS with 60k files in a watched directory, locally, takes
**\~ an hour before these changes**.
**After these changes** for the plugin to start processing files it
takes  **~2 seconds**.

Since the collection is known to have a noticeable (memory) footprint,
there's changes towards reducing intermediate garbage - which also
contributed to the decision to move the collection to native.

Notable changes to tracking watched files with
`FileWatch::WatchedFilesCollection`:
- predictable `log(size)` modification (and access) times
- implementation has clear locking semantics, previously operations
weren't 100% atomic
- hopefully, cleaner API as the collection re-resembles a *WatchedFile
-> String (path)* map
- the collection now needs an explicit update whenever a file changes
(on re-stat),
previous implicitness of picking up changes might have been the
reasoning behind re-sorts
(could be decoupled further by making `WatchedFile` immutable but
there's other state anyway)

---

Apart from the necessary changes to resolve the performance bottleneck,
there's also a quick review of (trace) logging - annoying to not only
see partial traces even in debug/trace levels.

Few unused methods and instance variables were removed for a better code
reading experience.

Author: kares

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 4.2.0
+  - Fix: watched files performance with huge filesets [#268](https://github.com/logstash-plugins/logstash-input-file/pull/268) 
+  - Updated logging to include full traces in debug (and trace) levels
+
 ## 4.1.18
   - Fix: release watched files on completion (in read-mode) [#271](https://github.com/logstash-plugins/logstash-input-file/pull/271)
 

lib/filewatch/discoverer.rb

@@ -9,6 +9,8 @@ class Discoverer
     # associated with a sincedb entry if one can be found
     include LogStash::Util::Loggable
 
+    attr_reader :watched_files_collection
+
     def initialize(watched_files_collection, sincedb_collection, settings)
       @watching = Concurrent::Array.new
       @exclude = Concurrent::Array.new
@@ -37,8 +39,7 @@ def can_exclude?(watched_file, new_discovery)
       @exclude.each do |pattern|
         if watched_file.pathname.basename.fnmatch?(pattern)
           if new_discovery
-            logger.trace("Discoverer can_exclude?: #{watched_file.path}: skipping " +
-              "because it matches exclude #{pattern}")
+            logger.trace("skipping file because it matches exclude", :path => watched_file.path, :pattern => pattern)
           end
           watched_file.unwatch
           return true
@@ -56,13 +57,13 @@ def discover_files_ongoing(path)
     end
 
     def discover_any_files(path, ongoing)
-      fileset = Dir.glob(path).select{|f| File.file?(f)}
-      logger.trace("discover_files",  "count" => fileset.size)
+      fileset = Dir.glob(path).select { |f| File.file?(f) }
+      logger.trace("discover_files", :count => fileset.size)
       fileset.each do |file|
-        pathname = Pathname.new(file)
         new_discovery = false
-        watched_file = @watched_files_collection.watched_file_by_path(file)
+        watched_file = @watched_files_collection.get(file)
         if watched_file.nil?
+          pathname = Pathname.new(file)
           begin
             path_stat = PathStatClass.new(pathname)
           rescue Errno::ENOENT
@@ -74,7 +75,7 @@ def discover_any_files(path, ongoing)
         # if it already unwatched or its excluded then we can skip
         next if watched_file.unwatched? || can_exclude?(watched_file, new_discovery)
 
-        logger.trace("discover_files handling:", "new discovery"=> new_discovery, "watched_file details" => watched_file.details)
+        logger.trace? && logger.trace("handling:", :new_discovery => new_discovery, :watched_file => watched_file.details)
 
         if new_discovery
           watched_file.initial_completed if ongoing
@@ -86,7 +87,7 @@ def discover_any_files(path, ongoing)
           #   associated with a different watched_file
           if @sincedb_collection.associate(watched_file)
             if watched_file.file_ignorable?
-              logger.trace("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
+              logger.trace("skipping file because it was last modified more than #{@settings.ignore_older} seconds ago", :path => file)
               # on discovery ignorable watched_files are put into the ignored state and that
               # updates the size from the internal stat
               # so the existing contents are not read.

lib/filewatch/observing_base.rb

@@ -62,8 +62,7 @@ def build_watch_and_dependencies
       @sincedb_collection = SincedbCollection.new(@settings)
       @sincedb_collection.open
       discoverer = Discoverer.new(watched_files_collection, @sincedb_collection, @settings)
-      @watch = Watch.new(discoverer, watched_files_collection, @settings)
-      @watch.add_processor build_specific_processor(@settings)
+      @watch = Watch.new(discoverer, build_specific_processor(@settings), @settings)
     end
 
     def watch_this(path)

lib/filewatch/processor.rb

@@ -0,0 +1,55 @@
+# encoding: utf-8
+require "logstash/util/loggable"
+require 'concurrent/atomic/atomic_reference'
+
+module FileWatch
+  class Processor
+    include LogStash::Util::Loggable
+
+    attr_reader :watch
+
+    def initialize(settings)
+      @settings = settings
+      @deletable_paths = Concurrent::AtomicReference.new []
+    end
+
+    def add_watch(watch)
+      @watch = watch
+      self
+    end
+
+    def clear_deletable_paths
+      @deletable_paths.get_and_set []
+    end
+
+    def add_deletable_path(path)
+      @deletable_paths.get << path
+    end
+
+    def restat(watched_file)
+      changed = watched_file.restat!
+      if changed
+        # the collection (when sorted by modified_at) needs to re-sort every time watched-file is modified,
+        # we can perform these update operation while processing files (stat interval) instead of having to
+        # re-sort the whole collection every time an entry is accessed
+        @watch.watched_files_collection.update(watched_file)
+      end
+    end
+
+    private
+
+    def error_details(error, watched_file)
+      details = { :path => watched_file.path,
+                  :exception => error.class,
+                  :message => error.message,
+                  :backtrace => error.backtrace }
+      if logger.debug?
+        details[:file] = watched_file
+      else
+        details[:backtrace] = details[:backtrace].take(8) if details[:backtrace]
+      end
+      details
+    end
+
+  end
+end

lib/filewatch/read_mode/handlers/base.rb

@@ -34,19 +34,21 @@ def handle_specifically(watched_file)
 
     def open_file(watched_file)
       return true if watched_file.file_open?
-      logger.trace("opening #{watched_file.path}")
+      logger.trace? && logger.trace("opening", :path => watched_file.path)
       begin
         watched_file.open
-      rescue
+      rescue => e
         # don't emit this message too often. if a file that we can't
         # read is changing a lot, we'll try to open it more often, and spam the logs.
         now = Time.now.to_i
         logger.trace("opening OPEN_WARN_INTERVAL is '#{OPEN_WARN_INTERVAL}'")
         if watched_file.last_open_warning_at.nil? || now - watched_file.last_open_warning_at > OPEN_WARN_INTERVAL
-          logger.warn("failed to open #{watched_file.path}: #{$!.inspect}, #{$!.backtrace.take(3)}")
+          backtrace = e.backtrace
+          backtrace = backtrace.take(3) if backtrace && !logger.debug?
+          logger.warn("failed to open", :path => watched_file.path, :exception => e.class, :message => e.message, :backtrace => backtrace)
           watched_file.last_open_warning_at = now
         else
-          logger.trace("suppressed warning for `failed to open` #{watched_file.path}: #{$!.inspect}")
+          logger.trace("suppressed warning (failed to open)", :path => watched_file.path, :exception => e.class, :message => e.message)
         end
         watched_file.watch # set it back to watch so we can try it again
       end
@@ -75,7 +77,7 @@ def add_or_update_sincedb_collection(watched_file)
           watched_file.update_bytes_read(sincedb_value.position)
         else
           sincedb_value.set_watched_file(watched_file)
-          logger.trace("add_or_update_sincedb_collection: switching from...", "watched_file details" => watched_file.details)
+          logger.trace("add_or_update_sincedb_collection: switching from", :watched_file => watched_file.details)
           watched_file.rotate_from(existing_watched_file)
         end
 
@@ -92,7 +94,7 @@ def update_existing_sincedb_collection_value(watched_file, sincedb_value)
     def add_new_value_sincedb_collection(watched_file)
       sincedb_value = SincedbValue.new(0)
       sincedb_value.set_watched_file(watched_file)
-      logger.trace("add_new_value_sincedb_collection: #{watched_file.path}", "position" => sincedb_value.position)
+      logger.trace("add_new_value_sincedb_collection:", :path => watched_file.path, :position => sincedb_value.position)
       sincedb_collection.set(watched_file.sincedb_key, sincedb_value)
     end
   end

lib/filewatch/read_mode/handlers/read_file.rb

@@ -31,7 +31,7 @@ def handle_specifically(watched_file)
     end
 
     def controlled_read(watched_file, loop_control)
-      logger.trace("reading...", "iterations" => loop_control.count, "amount" => loop_control.size, "filename" => watched_file.filename)
+      logger.trace? && logger.trace("reading...", :filename => watched_file.filename, :iterations => loop_control.count, :amount => loop_control.size)
       loop_control.count.times do
         break if quit?
         begin
@@ -43,22 +43,35 @@ def controlled_read(watched_file, loop_control)
             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))
+        rescue EOFError => e
+          log_error("controlled_read: eof error reading file", watched_file, e)
           loop_control.flag_read_error
           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))
+        rescue Errno::EWOULDBLOCK, Errno::EINTR => e
+          log_error("controlled_read: block or interrupt error reading file", watched_file, e)
           watched_file.listener.error
           loop_control.flag_read_error
           break
         rescue => e
-          logger.error("controlled_read: general error reading file", "path" => watched_file.path, "error" => e.inspect, "backtrace" => e.backtrace.take(8))
+          log_error("controlled_read: general error reading file", watched_file, e)
           watched_file.listener.error
           loop_control.flag_read_error
           break
         end
       end
     end
+
+    def log_error(msg, watched_file, error)
+      details = { :path => watched_file.path,
+                  :exception => error.class,
+                  :message => error.message,
+                  :backtrace => error.backtrace }
+      if logger.debug?
+        details[:file] = watched_file
+      else
+        details[:backtrace] = details[:backtrace].take(8) if details[:backtrace]
+      end
+      logger.error(msg, details)
+    end
   end
 end end end

lib/filewatch/read_mode/handlers/read_zip_file.rb

@@ -1,13 +1,15 @@
 # encoding: utf-8
 require 'java'
-java_import java.io.InputStream
-java_import java.io.InputStreamReader
-java_import java.io.FileInputStream
-java_import java.io.BufferedReader
-java_import java.util.zip.GZIPInputStream
-java_import java.util.zip.ZipException
 
 module FileWatch module ReadMode module Handlers
+
+  java_import java.io.InputStream
+  java_import java.io.InputStreamReader
+  java_import java.io.FileInputStream
+  java_import java.io.BufferedReader
+  java_import java.util.zip.GZIPInputStream
+  java_import java.util.zip.ZipException
+
   class ReadZipFile < Base
     def handle_specifically(watched_file)
       add_or_update_sincedb_collection(watched_file) unless sincedb_collection.member?(watched_file.sincedb_key)

lib/filewatch/read_mode/processor.rb

@@ -1,6 +1,5 @@
 # encoding: utf-8
-require "logstash/util/loggable"
-
+require 'filewatch/processor'
 require_relative "handlers/base"
 require_relative "handlers/read_file"
 require_relative "handlers/read_zip_file"
@@ -9,20 +8,7 @@ module FileWatch module ReadMode
   # Must handle
   #   :read_file
   #   :read_zip_file
-  class Processor
-    include LogStash::Util::Loggable
-
-    attr_reader :watch, :deletable_filepaths
-
-    def initialize(settings)
-      @settings = settings
-      @deletable_filepaths = []
-    end
-
-    def add_watch(watch)
-      @watch = watch
-      self
-    end
+  class Processor < FileWatch::Processor
 
     def initialize_handlers(sincedb_collection, observer)
       # we deviate from the tail mode handler initialization here
@@ -48,24 +34,23 @@ def process_all_states(watched_files)
     private
 
     def process_watched(watched_files)
-      logger.trace("Watched processing")
+      logger.trace(__method__.to_s)
       # Handles watched_files in the watched state.
       # for a slice of them:
       #   move to the active state
       #   should never have been active before
       # how much of the max active window is available
-      to_take = @settings.max_active - watched_files.count{|wf| wf.active?}
+      to_take = @settings.max_active - watched_files.count { |wf| wf.active? }
       if to_take > 0
-        watched_files.select {|wf| wf.watched?}.take(to_take).each do |watched_file|
-          path = watched_file.path
+        watched_files.select(&:watched?).take(to_take).each do |watched_file|
           begin
-            watched_file.restat
+            restat(watched_file)
             watched_file.activate
           rescue Errno::ENOENT
-            common_deleted_reaction(watched_file, "Watched")
+            common_deleted_reaction(watched_file, __method__)
             next
           rescue => e
-            common_error_reaction(path, e, "Watched")
+            common_error_reaction(watched_file, e, __method__)
             next
           end
           break if watch.quit?
@@ -74,7 +59,7 @@ def process_watched(watched_files)
         now = Time.now.to_i
         if (now - watch.lastwarn_max_files) > MAX_FILES_WARN_INTERVAL
           waiting = watched_files.size - @settings.max_active
-          logger.warn(@settings.max_warn_msg + ", files yet to open: #{waiting}")
+          logger.warn("#{@settings.max_warn_msg}, files yet to open: #{waiting}")
           watch.lastwarn_max_files = now
         end
       end
@@ -83,17 +68,18 @@ def process_watched(watched_files)
     ## TODO add process_rotation_in_progress
 
     def process_active(watched_files)
-      logger.trace("Active processing")
+      logger.trace(__method__.to_s)
       # Handles watched_files in the active state.
-      watched_files.select {|wf| wf.active? }.each do |watched_file|
-        path = watched_file.path
+      watched_files.each do |watched_file|
+        next unless watched_file.active?
+
         begin
-          watched_file.restat
+          restat(watched_file)
         rescue Errno::ENOENT
-          common_deleted_reaction(watched_file, "Active")
+          common_deleted_reaction(watched_file, __method__)
           next
         rescue => e
-          common_error_reaction(path, e, "Active")
+          common_error_reaction(watched_file, e, __method__)
           next
         end
         break if watch.quit?
@@ -114,19 +100,19 @@ def process_active(watched_files)
     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")
+      add_deletable_path watched_file.path
+      logger.trace? && logger.trace("whole file read, removing from collection", :path => watched_file.path)
     end
 
     def common_deleted_reaction(watched_file, action)
       # file has gone away or we can't read it anymore.
       watched_file.unwatch
-      deletable_filepaths << watched_file.path
-      logger.trace("#{action} - stat failed: #{watched_file.path}, removing from collection")
+      add_deletable_path watched_file.path
+      logger.trace? && logger.trace("#{action} - stat failed, removing from collection", :path => watched_file.path)
     end
 
-    def common_error_reaction(path, error, action)
-      logger.error("#{action} - other error #{path}: (#{error.message}, #{error.backtrace.take(8).inspect})")
+    def common_error_reaction(watched_file, error, action)
+      logger.error("#{action} - other error", error_details(error, watched_file))
     end
   end
 end end