Synchronized Log Rotation and Invariant (#31)

 - Refactoring work to add locks to file rotation and writing.
    - Fixes [#2](#2) - Plugin crashes on file rotation.
    - Fixes [#19](#19) - Deleted files remain in use by the system eventually filling up disk space.

Author: josephlewis42

.gitignore

@@ -2,3 +2,5 @@
 Gemfile.lock
 .bundle
 vendor
+.idea
+

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 3.2.1
+  - Refactoring work to add locks to file rotation and writing.
+    - Fixes [#2](https://github.com/logstash-plugins/logstash-output-google_cloud_storage/issues/2) - Plugin crashes on file rotation.
+    - Fixes [#19](https://github.com/logstash-plugins/logstash-output-google_cloud_storage/issues/19) - Deleted files remain in use by the system eventually filling up disk space.
+
 ## 3.2.0
   - Change uploads to use a job pool for better performance
     - Fixes [#22](https://github.com/logstash-plugins/logstash-output-google_cloud_storage/issues/22) - Refactor Job Queue Architecture

lib/logstash/outputs/gcs/log_rotate.rb

@@ -0,0 +1,76 @@
+# encoding: utf-8
+require 'logstash/outputs/gcs/temp_log_file'
+require 'concurrent'
+
+module LogStash
+  module Outputs
+    module Gcs
+      class LogRotate
+        def initialize(path_factory, max_file_size_bytes, gzip, flush_interval_secs)
+          @path_factory = path_factory
+          @max_file_size_bytes = max_file_size_bytes
+          @gzip = gzip
+          @flush_interval_secs = flush_interval_secs
+
+          @lock = Concurrent::ReentrantReadWriteLock.new
+          @rotate_callback = nil
+
+          rotate_log!
+        end
+
+        # writeln writes a message and carriage-return character to the open
+        # log file, rotating and syncing it if necessary.
+        #
+        # nil messages do not get written, but may cause the log to rotate
+        def writeln(message=nil)
+          @lock.with_write_lock do
+            rotate_log! if should_rotate?
+
+            @temp_file.write(message, "\n") unless message.nil?
+
+            @temp_file.fsync if @temp_file.time_since_sync >= @flush_interval_secs
+          end
+        end
+
+        # rotate_log! closes the current log (if it exists), notifies the
+        # handler, rolls the path over and opens a new log.
+        #
+        # Invariant: the old log will ALWAYS be closed and a new one will
+        # ALWAYS be open at the completion of this function.
+        def rotate_log!
+          @lock.with_write_lock do
+            unless @temp_file.nil?
+              @temp_file.close!
+              @rotate_callback.call(@temp_file.path) unless @rotate_callback.nil?
+            end
+
+            @path_factory.rotate_path!
+
+            path = @path_factory.current_path
+            @temp_file = LogStash::Outputs::Gcs::LogFileFactory.create(path, @gzip)
+          end
+        end
+
+        # on_rotate sets a handler to be called when the log gets rotated.
+        # The handler receives the path to the rotated out log as a string.
+        def on_rotate(&block)
+          @lock.with_write_lock do
+            @rotate_callback = block
+          end
+        end
+
+        private
+
+        def should_rotate?
+          @lock.with_read_lock do
+            path_changed = @path_factory.should_rotate?
+            rotate_on_size = @max_file_size_bytes > 0
+            too_big = @temp_file.size >= @max_file_size_bytes
+
+            path_changed || (rotate_on_size && too_big)
+          end
+        end
+      end
+    end
+  end
+end

lib/logstash/outputs/gcs/path_factory.rb

@@ -30,13 +30,16 @@ def initialize(directory, prefix, include_host, date_pattern, include_part, incl
 
         # Rotates the path to the next one in sequence. If the path has a part number
         # and the base path (date/hostname) haven't changed the part number is incremented.
+        # Returns the path that was rotated out
         def rotate_path!
+          last_path = current_path
+
           @path_lock.synchronize {
             @part_number = (next_base == current_base) ? @part_number + 1 : 0
             @current = template_variables
           }
 
-          current_path
+          last_path
         end
 
         # Checks if the file is ready to rotate because the timestamp changed.

lib/logstash/outputs/gcs/temp_log_file.rb

@@ -0,0 +1,110 @@
+# encoding: utf-8
+require 'zlib'
+require 'concurrent'
+require 'time'
+
+module LogStash
+  module Outputs
+    module Gcs
+      # LogFileFactory creates a LogFile according to user specification
+      # optionally gzipping it and creating mutexes around modification
+      # points.
+      class LogFileFactory
+        def self.create(path, gzip, synchronize=true)
+          lf = LogStash::Outputs::Gcs::PlainLogFile.new(path)
+          lf = LogStash::Outputs::Gcs::GzipLogFile.new(lf) if gzip
+          lf = LogStash::Outputs::Gcs::SynchronizedLogFile.new(lf) if synchronize
+
+          lf
+        end
+      end
+
+      # PlainLogFile writes events to a plain text file.
+      class PlainLogFile
+        attr_reader :path, :fd
+
+        def initialize(path)
+          @path = path
+          @fd = ::File.new(path, 'a+')
+          @last_sync = Time.now
+        end
+
+        def write(*contents)
+          contents.each { |c| @fd.write(c) }
+        end
+
+        def fsync
+          @fd.fsync
+          @last_sync = Time.now
+        end
+
+        def close!
+          @fd.fsync
+          @fd.close
+        end
+
+        def size
+          ::File.stat(@path).size
+        end
+
+        def time_since_sync
+          Time.now - @last_sync
+        end
+      end
+
+      # GzipLogFile wraps another log file and writes events through it.
+      class GzipLogFile
+        attr_reader :fd
+
+        def initialize(child)
+          @child = child
+          @fd = Zlib::GzipWriter.new(child.fd)
+        end
+
+        def write(*contents)
+          contents.each { |c| @fd.write(c) }
+        end
+
+        def fsync
+          @fd.flush
+          @child.fsync
+        end
+
+        def close!
+          fsync
+          # The Gzip writer closes the underlying IO after
+          # appending the Gzip footer.
+          @fd.close
+        end
+
+        def method_missing(method_name, *args, &block)
+          @child.send(method_name, *args, &block)
+        end
+      end
+
+      # SynchronizedLogFile wraps another log file and uses reentrant locks
+      # around its methods to prevent concurrent modification.
+      class SynchronizedLogFile
+        def initialize(child)
+          @child = child
+          @lock = Concurrent::ReentrantReadWriteLock.new
+        end
+
+        def time_since_sync
+          @lock.with_read_lock { @child.time_since_sync }
+        end
+
+        def path
+          @lock.with_read_lock { @child.path }
+        end
+
+        def method_missing(method_name, *args, &block)
+          # unless otherwise specified, get a write lock
+          @lock.with_write_lock do
+            @child.send(method_name, *args, &block)
+          end
+        end
+      end
+    end
+  end
+end