lib/fluent/plugin/in_tail.rb in fluentd-0.14.23.rc1 vs lib/fluent/plugin/in_tail.rb in fluentd-0.14.23

@@ -32,10 +32,20 @@
   class TailInput < Fluent::Plugin::Input
     Fluent::Plugin.register_input('tail', self)
 
     helpers :timer, :event_loop, :parser, :compat_parameters
 
+    class WatcherSetupError < StandardError
+      def initialize(msg)
+        @message = msg
+      end
+
+      def to_s
+        @message
+      end
+    end
+
     FILE_PERMISSION = 0644
 
     def initialize
       super
       @paths = []
@@ -248,10 +258,16 @@
       tw.attach do |watcher|
         watcher.timer_trigger = timer_execute(:in_tail_timer_trigger, 1, &watcher.method(:on_notify)) if watcher.enable_watch_timer
         event_loop_attach(watcher.stat_trigger)
       end
       tw
+    rescue => e
+      if tw
+        tw.detach
+        tw.close
+      end
+      raise e
     end
 
     def start_watchers(paths)
       paths.each { |path|
         pe = nil
@@ -264,11 +280,17 @@
               $log.warn "#{path} not found. Continuing without tailing it."
             end
           end
         end
 
-        @tails[path] = setup_watcher(path, pe)
+        begin
+          tw = setup_watcher(path, pe)
+        rescue WatcherSetupError => e
+          log.warn "Skip #{path} because unexpected setup error happens: #{e}"
+          next
+        end
+        @tails[path] = tw
       }
     end
 
     def stop_watchers(paths, immediate: false, unwatched: false, remove_watcher: true)
       paths.each { |path|
@@ -478,12 +500,12 @@
         on_notify
         yield self
       end
 
       def detach
-        @timer_trigger.detach if @enable_watch_timer && @timer_trigger.attached?
-        @stat_trigger.detach if @stat_trigger.attached?
+        @timer_trigger.detach if @enable_watch_timer && @timer_trigger && @timer_trigger.attached?
+        @stat_trigger.detach if @stat_trigger && @stat_trigger.attached?
         @io_handler.on_notify if @io_handler
       end
 
       def close
         if @io_handler
@@ -696,10 +718,13 @@
 
         def open
           io = Fluent::FileWrapper.open(@watcher.path)
           io.seek(@watcher.pe.read_pos + @fifo.bytesize)
           io
+        rescue RangeError
+          io.close if io
+          raise WatcherSetupError, "seek error with #{@watcher.path}: file position = #{@watcher.pe.read_pos.to_s(16)}, reading bytesize = #{@fifo.bytesize.to_s(16)}"
         rescue Errno::ENOENT
           nil
         end
 
         def with_io
@@ -713,9 +738,12 @@
               end
             else
               @io ||= open
               yield @io
             end
+          rescue WatcherSetupError => e
+            close
+            raise e
           rescue
             @watcher.log.error $!.to_s
             @watcher.log.error_backtrace
             close
           end