in_tail: Safely skip files used by another process on Windows

lib/fluent/plugin/file_wrapper.rb

@@ -52,6 +52,8 @@ class Win32Error < StandardError
 
     attr_reader :errcode, :msg
 
+    WSABASEERR = 10000
+
     def initialize(errcode, msg = nil)
       @errcode = errcode
       @msg = msg
@@ -80,6 +82,10 @@ def ==(other)
       return false if other.class != Win32Error
       @errcode == other.errcode && @msg == other.msg
     end
+
+    def wsaerr?
+      @errcode >= WSABASEERR
+    end
   end
 
   # To open and get stat with setting FILE_SHARE_DELETE
@@ -113,11 +119,14 @@ def initialize(path, mode='r', sharemode=FILE_SHARE_READ|FILE_SHARE_WRITE|FILE_S
       @file_handle = CreateFile.call(@path, access, sharemode,
                      0, creationdisposition, FILE_ATTRIBUTE_NORMAL, 0)
       if @file_handle == INVALID_HANDLE_VALUE
-        err = Win32::API.last_error
-        if err == ERROR_FILE_NOT_FOUND || err == ERROR_PATH_NOT_FOUND || err == ERROR_ACCESS_DENIED
-          raise Errno::ENOENT
+        win32err = Win32Error.new(Win32::API.last_error, path)
+        errno = ServerEngine::RbWinSock.rb_w32_map_errno(win32err.errcode)
+        if errno == Errno::EINVAL::Errno || win32err.wsaerr?
+          # maybe failed to map
+          raise win32err
+        else
+          raise SystemCallError.new(win32err.message, errno)
         end
-        raise Win32Error.new(err, path)
       end
     end
 

lib/fluent/plugin/in_tail.rb

@@ -302,7 +302,7 @@ def expand_paths
                 end
                 false
               end
-            rescue Errno::ENOENT
+            rescue Errno::ENOENT, Errno::EACCES
               log.debug("#{p} is missing after refresh file list")
               false
             end
@@ -334,8 +334,8 @@ def expand_paths
           else
             hash[target_info.path] = target_info
           end
-        rescue Errno::ENOENT
-          $log.warn "expand_paths: stat() for #{path} failed with ENOENT. Skip file."
+        rescue Errno::ENOENT, Errno::EACCES  => e
+          $log.warn "expand_paths: stat() for #{path} failed with #{e.class.name}. Skip file."
         end
       }
       hash
@@ -411,8 +411,8 @@ def construct_watcher(target_info)
         if @read_from_head && pe.read_inode.zero?
           begin
             pe.update(Fluent::FileWrapper.stat(target_info.path).ino, 0)
-          rescue Errno::ENOENT
-            $log.warn "#{target_info.path} not found. Continuing without tailing it."
+          rescue Errno::ENOENT, Errno::EACCES
+            $log.warn "stat() for #{target_info.path} failed. Continuing without tailing it."
           end
         end
       end
@@ -427,8 +427,8 @@ def construct_watcher(target_info)
       begin
         target_info = TargetInfo.new(target_info.path, Fluent::FileWrapper.stat(target_info.path).ino)
         @tails[target_info] = tw
-      rescue Errno::ENOENT
-        $log.warn "stat() for #{target_info.path} failed with ENOENT. Drop tail watcher for now."
+      rescue Errno::ENOENT, Errno::EACCES => e
+        $log.warn "stat() for #{target_info.path} failed with #{e.class.name}. Drop tail watcher for now."
         # explicitly detach and unwatch watcher `tw`.
         tw.unwatched = true
         detach_watcher(tw, target_info.ino, false)
@@ -732,7 +732,7 @@ def close
       def on_notify
         begin
           stat = Fluent::FileWrapper.stat(@path)
-        rescue Errno::ENOENT
+        rescue Errno::ENOENT, Errno::EACCES
           # moved or deleted
           stat = nil
         end
@@ -999,7 +999,7 @@ def open
         rescue RangeError
           io.close if io
           raise WatcherSetupError, "seek error with #{@path}: file position = #{@watcher.pe.read_pos.to_s(16)}, reading bytesize = #{@fifo.bytesize.to_s(16)}"
-        rescue Errno::ENOENT
+        rescue Errno::ENOENT, Errno::EACCES
           nil
         end
 

test/plugin/test_file_wrapper.rb

@@ -58,6 +58,15 @@ def teardown
       assert_equal("#<Fluent::Win32Error: code: 32, The process cannot access the file because it is being used by another process. - C:\file.txt>",
                    Fluent::Win32Error.new(ERROR_SHARING_VIOLATION, "C:\file.txt").inspect)
     end
+
+    data('0' => [false, 0],
+         '9999' => [false, 9999],
+         '10000' => [true, 10000],
+         '10001' => [true, 10001])
+    test 'wsaerr?' do |data|
+      expected, code = data
+      assert_equal(expected, Fluent::Win32Error.new(code).wsaerr?)
+    end
   end
 
   sub_test_case 'WindowsFile exceptions' do
@@ -92,7 +101,8 @@ def teardown
         path = "#{TMP_DIR}/test_windows_file.txt"
         file1 = file2 = nil
         file1 = File.open(path, "wb")
-        assert_raise(Fluent::Win32Error.new(ERROR_SHARING_VIOLATION, path)) do
+        win32err = Fluent::Win32Error.new(ERROR_SHARING_VIOLATION, path)
+        assert_raise(Errno::EACCES.new(win32err.message)) do
           file2 = Fluent::WindowsFile.new(path, 'r', FILE_SHARE_READ)
         ensure
           file2.close if file2

test/plugin/test_in_tail.rb

@@ -2068,6 +2068,50 @@ def test_ENOENT_error_after_setup_watcher
       d.run(shutdown: false) {}
     end
     d.instance_shutdown
-    assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed with ENOENT. Drop tail watcher for now.\n") })
+    assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed with Errno::ENOENT. Drop tail watcher for now.\n") })
+  end
+
+  def test_EACCES_error_after_setup_watcher
+    path = "#{TMP_DIR}/noaccess/tail.txt"
+    begin
+      FileUtils.mkdir_p("#{TMP_DIR}/noaccess")
+      FileUtils.chmod(0755, "#{TMP_DIR}/noaccess")
+      FileUtils.touch(path)
+      config = config_element('', '', {
+                                'tag' => "tail",
+                                'path' => path,
+                                'format' => 'none',
+                              })
+      d = create_driver(config, false)
+      mock.proxy(d.instance).setup_watcher(anything, anything) do |tw|
+        FileUtils.chmod(0000, "#{TMP_DIR}/noaccess")
+        tw
+      end
+      assert_nothing_raised do
+        d.run(shutdown: false) {}
+      end
+      d.instance_shutdown
+      assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed with Errno::EACCES. Drop tail watcher for now.\n") })
+    end
+  ensure
+    FileUtils.chmod(0755, "#{TMP_DIR}/noaccess")
+    FileUtils.rm_rf("#{TMP_DIR}/noaccess")
+  end unless Fluent.windows?
+
+  def test_EACCES
+    path = "#{TMP_DIR}/tail.txt"
+    FileUtils.touch(path)
+    config = config_element('', '', {
+                              'format' => 'none',
+                            })
+    d = create_driver(config)
+    mock.proxy(Fluent::FileWrapper).stat(path) do |stat|
+      raise Errno::EACCES
+    end.at_least(1)
+    assert_nothing_raised do
+      d.run(shutdown: false) {}
+    end
+    d.instance_shutdown
+    assert($log.out.logs.any?{|log| log.include?("expand_paths: stat() for #{path} failed with Errno::EACCES. Skip file.\n") })
   end
 end