Fix race conditions in #open

Author: andrewvc

lib/logstash/outputs/file.rb

@@ -39,13 +39,7 @@ class LogStash::Outputs::File < LogStash::Outputs::Base
   # E.g: `/%{myfield}/`, `/test-%{myfield}/` are not valid paths
   config :path, :validate => :string, :required => true
 
-  # The format to use when writing events to the file. This value
-  # supports any string and can include `%{name}` and other dynamic
-  # strings.
-  #
-  # If this setting is omitted, the full json representation of the
-  # event will be written as a single line.
-  config :message_format, :validate => :string, :deprecated => "You can achieve the same behavior with the 'line' codec"
+  config :message_format, :validate => :string, :obsolete => "You can achieve the same behavior with the 'line' codec"
 
   # Flush interval (in seconds) for flushing writes to log files.
   # 0 will flush on every message.
@@ -81,10 +75,8 @@ def register
     require "fileutils" # For mkdir_p
 
     @files = {}
-    @files_lock = java.util.concurrent.locks.ReentrantReadWriteLock.new
-    @files_read_lock = @files_lock.readLock
-    @files_write_lock = @files_lock.writeLock
-
+    @io_mutex = Mutex.new
+    
     @path = File.expand_path(path)
 
     validate_path
@@ -131,33 +123,28 @@ def root_directory
   public
   def multi_receive_encoded(events_and_encoded)
     encoded_by_path = Hash.new {|h,k| h[k] = []}
-    
+
     events_and_encoded.each do |event,encoded|
-      file_output_path = write_event(event)
+      file_output_path = event_path(event)
       encoded_by_path[file_output_path] << encoded
     end
 
-    encoded_by_path.each do |path,chunks|
-      fd = open(path)
-      chunks.each(&fd.method(:write))
-      flush(fd)
-    end
-    
-    close_stale_files
+    @io_mutex.synchronize do
+      encoded_by_path.each do |path,chunks|
+        fd = open(path)
+        chunks.each {|chunk| fd.write(chunk) }
+        fd.flush
+      end
+      
+      close_stale_files
+    end   
   end # def receive
 
-  def with_lock(lock)
-    lock.lock
-    yield
-  ensure
-    lock.unlock
-  end
-
   public
   def close
-    @logger.debug("Close: closing files")
-    
-    with_lock(@files_read_lock) do
+    @io_mutex.synchronize do
+      @logger.debug("Close: closing files")
+      
       @files.each do |path, fd|
         begin
           fd.close
@@ -176,7 +163,7 @@ def inside_file_root?(log_path)
   end
 
   private
-  def write_event(event)
+  def event_path(event)
     file_output_path = generate_filepath(event)
     if path_with_field_ref? && !inside_file_root?(file_output_path)
       @logger.warn("File: the event tried to write outside the files root, writing the event to the failure file",  :event => event, :filename => @failure_path)
@@ -185,7 +172,6 @@ def write_event(event)
       file_output_path = @failure_path
     end
     @logger.debug("File, writing event to file.", :filename => file_output_path)
-
 
     file_output_path    
   end
@@ -221,11 +207,9 @@ def flush_pending_files
     return unless Time.now - @last_flush_cycle >= flush_interval
     @logger.debug("Starting flush cycle")
 
-    with_lock(@files_read_lock) do
-      @files.each do |path, fd|
-        @logger.debug("Flushing file", :path => path, :fd => fd)
-        fd.flush
-      end
+    @files.each do |path, fd|
+      @logger.debug("Flushing file", :path => path, :fd => fd)
+      fd.flush
     end
 
     @last_flush_cycle = Time.now
@@ -237,24 +221,22 @@ def close_stale_files
     now = Time.now
     return unless now - @last_stale_cleanup_cycle >= @stale_cleanup_interval
 
-    with_lock(@files_write_lock) do
-      @logger.info("Starting stale files cleanup cycle", :files => @files)
-      inactive_files = @files.select { |path, fd| not fd.active }
-      @logger.debug("%d stale files found" % inactive_files.count, :inactive_files => inactive_files)
-      inactive_files.each do |path, fd|
-        @logger.info("Closing file %s" % path)
-        fd.close
-        @files.delete(path)
-      end
-      # mark all files as inactive, a call to write will mark them as active again
-      @files.each { |path, fd| fd.active = false }
-      @last_stale_cleanup_cycle = now
+    @logger.info("Starting stale files cleanup cycle", :files => @files)
+    inactive_files = @files.select { |path, fd| not fd.active }
+    @logger.debug("%d stale files found" % inactive_files.count, :inactive_files => inactive_files)
+    inactive_files.each do |path, fd|
+      @logger.info("Closing file %s" % path)
+      fd.close
+      @files.delete(path)
     end
+    # mark all files as inactive, a call to write will mark them as active again
+    @files.each { |path, fd| fd.active = false }
+    @last_stale_cleanup_cycle = now
   end
 
   private
   def cached?(path)
-    with_lock(@files_read_lock) { @files.include?(path) && !@files[path].nil? }
+    @files.include?(path) && !@files[path].nil?
   end
 
   private
@@ -265,19 +247,20 @@ def deleted?(path)
   private
   def open(path)
     if !deleted?(path) && cached?(path)
-      with_lock(@files_read_lock) { return @files[path] }
-    elsif deleted?(path)
-      with_lock(@files_write_lock) do
-        if @create_if_deleted
-          @logger.debug("Required path was deleted, creating the file again", :path => path)
-          @files.delete(path)
-        else
-          return @files[path] if cached?(path)
-        end
+      return @files[path]
+    end
+
+    if deleted?(path)
+      if @create_if_deleted
+        @logger.debug("Required path was deleted, creating the file again", :path => path)
+        @files.delete(path)
+      else
+        return @files[path] if cached?(path)
       end
     end
-    @logger.info("Opening file", :path => path)
 
+    @logger.info("Opening file", :path => path)
+    
     dir = File.dirname(path)
     if !Dir.exist?(dir)
       @logger.info("Creating directory", :directory => dir)
@@ -287,6 +270,7 @@ def open(path)
         FileUtils.mkdir_p(dir)
       end
     end
+    
     # work around a bug opening fifos (bug JRUBY-6280)
     stat = File.stat(path) rescue nil
     if stat && stat.ftype == "fifo" && LogStash::Environment.jruby?
@@ -301,21 +285,18 @@ def open(path)
     if gzip
       fd = Zlib::GzipWriter.new(fd)
     end
-    with_lock(@files_write_lock) { @files[path] = IOWriter.new(fd) }
+    @files[path] = IOWriter.new(fd)
   end
 end # class LogStash::Outputs::File
 
 # wrapper class
 class IOWriter
   def initialize(io)
     @io = io
-    @write_mutex = Mutex.new
   end
   def write(string)
-    @write_mutex.synchronize do
-      @io.write(string)
-      @active = true
-    end
+    @io.write(string)
+    @active = true
   end
   def flush
     @io.flush
@@ -325,6 +306,7 @@ def flush
   end
   def method_missing(method_name, *args, &block)
     if @io.respond_to?(method_name)
+
       @io.send(method_name, *args, &block)
     else
       super

spec/outputs/file_spec.rb

@@ -47,7 +47,7 @@
 
   describe "ship lots of events to a file gzipped" do
     Stud::Temporary.file('logstash-spec-output-file') do |tmp_file|
-      event_count = 10000 + rand(500)
+      event_count = 100000 + rand(500)
 
       config <<-CONFIG
         input {
@@ -119,19 +119,20 @@
         { "path" => temp_file.path, "flush_interval" => 0 }
       end
 
-      it "should recreate the required file if deleted" do
+      it "should recreate the required file if deleted" do        
         output = LogStash::Outputs::File.new(config)
         output.register
 
         10.times do |i|
           event = LogStash::Event.new("event_id" => i)
-          output.receive(event)
+          output.multi_receive([event])
         end
         FileUtils.rm(temp_file)
-        10.times do |i|
+        10.times do |i|          
           event = LogStash::Event.new("event_id" => i+10)
-          output.receive(event)
+          output.multi_receive([event])
         end
+        
         expect(FileTest.size(temp_file.path)).to be > 0
       end
 
@@ -147,12 +148,12 @@
 
           10.times do |i|
             event = LogStash::Event.new("event_id" => i)
-            output.receive(event)
+            output.multi_receive([event])
           end
           FileUtils.rm(temp_file)
           10.times do |i|
             event = LogStash::Event.new("event_id" => i+10)
-            output.receive(event)
+            output.multi_receive([event])
           end
           expect(FileTest.exist?(temp_file.path)).to be_falsey
           expect(FileTest.size(output.failure_path)).to be > 0
@@ -184,7 +185,7 @@
 
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(bad_event)
+            output.multi_receive([bad_event])
 
             error_file = File.join(path, config["filename_failure"])
 
@@ -202,10 +203,10 @@
             output.register
 
             bad_event.set('error', encoded_once)
-            output.receive(bad_event)
+            output.multi_receive([bad_event])
 
             bad_event.set('error', encoded_twice)
-            output.receive(bad_event)
+            output.multi_receive([bad_event])
 
             expect(Dir.glob(File.join(path, "*")).size).to eq(2)
             output.close
@@ -218,7 +219,7 @@
             output.register
 
             bad_event.set('error', '../..//test')
-            output.receive(bad_event)
+            output.multi_receive([bad_event])
 
             expect(Dir.glob(File.join(path, "*")).size).to eq(1)
             output.close
@@ -235,7 +236,7 @@
             config = { "path" => "#{path}/%{error}" }
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
 
             good_file = File.join(path, good_event.get('error'))
             expect(File.exist?(good_file)).to eq(true)
@@ -254,7 +255,7 @@
             config = { "path" => dynamic_path }
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
 
             expect(File.exist?(expected_path)).to eq(true)
             output.close
@@ -276,7 +277,7 @@
 
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
 
             expect(File.exist?(expected_path)).to eq(true)
             output.close
@@ -291,7 +292,7 @@
             config = { "path" => "#{path}/%{error}" }
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
 
             good_file = File.join(path, good_event.get('error'))
             expect(File.exist?(good_file)).to eq(true)
@@ -310,7 +311,7 @@
             config = { "path" => "#{path}/output.txt" }
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
             good_file = File.join(path, 'output.txt')
             expect(File.exist?(good_file)).to eq(true)
             output.close #teardown first to allow reading the file
@@ -328,30 +329,9 @@
 
           Stud::Temporary.directory do |path|
             config = { "path" => "#{path}/output.txt" }
-            output = LogStash::Outputs::File.new(config)
-            output.codec = LogStash::Codecs::Line.new({ "format" => "Custom format: %{message}"})
-            output.register
-            output.receive(good_event)
-            good_file = File.join(path, 'output.txt')
-            expect(File.exist?(good_file)).to eq(true)
-            output.close #teardown first to allow reading the file
-            File.open(good_file) {|f|
-              line = f.readline
-              expect(line).to eq("Custom format: hello world\n")
-            }
-          end
-        end
-      end
-      context "when using deprecated message_format config" do
-        it 'falls back to line codec' do
-          good_event = LogStash::Event.new
-          good_event.set('message', 'hello world')
-
-          Stud::Temporary.directory do |path|
-            config = { "path" => "#{path}/output.txt", "message_format" => "Custom format: %{message}" }
-            output = LogStash::Outputs::File.new(config)
+            output = LogStash::Outputs::File.new(config.merge("codec" => LogStash::Codecs::Line.new({ "format" => "Custom format: %{message}"})))
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
             good_file = File.join(path, 'output.txt')
             expect(File.exist?(good_file)).to eq(true)
             output.close #teardown first to allow reading the file
@@ -375,7 +355,7 @@
             }
             output = LogStash::Outputs::File.new(config)
             output.register
-            output.receive(good_event)
+            output.multi_receive([good_event])
             good_file = File.join(path, 'is/nested/output.txt')
             expect(File.exist?(good_file)).to eq(true)
             expect(File.stat(good_file).mode.to_s(8)[-3..-1]).to eq('610')