in_tail: Safely skip files used by another process on Windows

On Windows, a win32 error `ERROR_SHARING_VIOLATION` is occured on
`Fluent::FileWrapper.stat(path)` when a file is temporarily used by
another process. Such files should be skipped safely until the file is
released as same as ENOENT case.

This patch rescues also other EACCES cases although they rarely occur.

Signed-off-by: Takuro Ashie <ashie@clear-code.com>

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