make sure the test pass when using java classes

Fixes #102

Author: ph

lib/logstash/outputs/s3.rb

@@ -11,7 +11,10 @@
 require "fileutils"
 require "set"
 require "pathname"
+require "aws-sdk"
+require "logstash/outputs/s3/patch"
 
+Aws.eager_autoload!
 
 # INFORMATION:
 #
@@ -118,7 +121,7 @@ class LogStash::Outputs::S3 < LogStash::Outputs::Base
   ## This is hack for not destroy the new files after restoring the initial files.
   ## If you do not specify "restore => true" when logstash crashes or is restarted, the files are not sent into the bucket,
   ## for example if you have single Instance.
-  config :restore, :validate => :boolean, :default => false
+  config :restore, :validate => :boolean, :default => true
 
   # The S3 canned ACL to use when putting the file. Defaults to "private".
   config :canned_acl, :validate => ["private", "public_read", "public_read_write", "authenticated_read"],
@@ -191,20 +194,21 @@ def register
     @file_repository = FileRepository.new(@tags, @encoding, @temporary_directory)
 
     @rotation = rotation_strategy
-    @uploader = Uploader.new(bucket_resource, @logger,  Concurrent::ThreadPoolExecutor.new({
-                                                                                             :min_threads => 1,
-                                                                                             :max_threads => @upload_workers_count,
-                                                                                             :max_queue => @upload_queue_size,
-                                                                                             :fallback_policy => :caller_runs
-                                                                                           }))
+
+    executor = Concurrent::ThreadPoolExecutor.new({ :min_threads => 1,
+                                                    :max_threads => @upload_workers_count,
+                                                    :max_queue => @upload_queue_size,
+                                                    :fallback_policy => :caller_runs })
+
+    @uploader = Uploader.new(bucket_resource, @logger, executor)
 
     # Restoring from crash will use a new threadpool to slowly recover
     # New events should have more priority.
     restore_from_crash if @restore
 
     # If we need time based rotation we need to do periodic check on the file
     # to take care of file that were not updated recently
-    start_periodic_check if @rotation.need_periodic?
+    start_periodic_check if @rotation.needs_periodic?
   end
 
   def multi_receive_encoded(events_and_encoded)
@@ -229,7 +233,7 @@ def multi_receive_encoded(events_and_encoded)
   end
 
   def close
-    stop_periodic_check if @rotation.need_periodic?
+    stop_periodic_check if @rotation.needs_periodic?
 
     @logger.debug("Uploading current workspace")
 
@@ -294,7 +298,8 @@ def upload_options
 
   def rotate_if_needed(prefixes)
     prefixes.each do |prefix|
-      # Each file access is thread safe, until the rotation is done then only
+      # Each file access is thread safe,
+      # until the rotation is done then only
       # one thread has access to the resource.
       @file_repository.get_factory(prefix) do |factory|
         temp_file = factory.current
@@ -316,7 +321,7 @@ def upload_file(temp_file)
     @logger.debug("Queue for upload", :path => temp_file.path)
 
     # if the queue is full the calling thread will be used to upload
-    temp_file.fsync # make sure we flush the fd before uploading it.
+    temp_file.close # make sure the content is on disk
     if temp_file.size > 0
       @uploader.upload_async(temp_file,
                              :on_complete => method(:clean_temporary_file),
@@ -346,14 +351,12 @@ def restore_from_crash
     @crash_uploader = Uploader.new(bucket_resource, @logger, CRASH_RECOVERY_THREADPOOL)
 
     temp_folder_path = Pathname.new(@temporary_directory)
-    Dir.glob(::File.join(@temporary_directory, "**/*")) do |file|
-      if ::File.file?(file)
-        key_parts = Pathname.new(file).relative_path_from(temp_folder_path).to_s.split(::File::SEPARATOR)
-        temp_file = TemporaryFile.new(key_parts.slice(1, key_parts.size).join("/"), ::File.open(file, "r"), key_parts.slice(0, 1))
-
-        @logger.debug("Recovering from crash and uploading", :file => temp_file.path)
-        @crash_uploader.upload_async(temp_file, :on_complete => method(:clean_temporary_file))
-      end
+    Dir.glob(::File.join(@temporary_directory, "**/*"))
+      .select { |file| ::File.file?(file) }
+      .each do |file|
+      temp_file = TemporaryFile.create_from_existing_file(file, temp_folder_path)
+      @logger.debug("Recovering from crash and uploading", :file => temp_file.path)
+      @crash_uploader.upload_async(temp_file, :on_complete => method(:clean_temporary_file))
     end
   end
 end

lib/logstash/outputs/s3/file_repository.rb

@@ -15,15 +15,15 @@ class FileRepository
         # Ensure that all access or work done
         # on a factory is threadsafe
         class PrefixedValue
-          def initialize(factory, stale_time)
-            @factory = factory
+          def initialize(file_factory, stale_time)
+            @file_factory = file_factory
             @lock = Mutex.new
             @stale_time = stale_time
           end
 
           def with_lock
             @lock.synchronize {
-              yield @factory
+              yield @file_factory
             }
           end
 

lib/logstash/outputs/s3/patch.rb

@@ -0,0 +1,22 @@
+# This is patch related to the autoloading and ruby
+#
+# The fix exist in jruby 9k but not in the current jruby, not sure when or it will be backported
+# https://github.com/jruby/jruby/issues/3645
+#
+# AWS is doing tricky name discovery in the module to generate the correct error class and
+# this strategy is bogus in jruby and `eager_autoload` don't fix this issue.
+#
+# This will be a short lived patch since AWS is removing the need.
+# see: https://github.com/aws/aws-sdk-ruby/issues/1301#issuecomment-261115960
+old_stderr = $stderr
+
+$stderr = StringIO.new
+begin
+  module Aws
+    const_set(:S3, Aws::S3)
+  end
+ensure
+  $stderr = old_stderr
+end
+
+

lib/logstash/outputs/s3/size_and_time_rotation_policy.rb

@@ -15,7 +15,7 @@ def rotate?(file)
           @size_strategy.rotate?(file) || @time_strategy.rotate?(file)
         end 
 
-        def need_periodic?
+        def needs_periodic?
           true
         end
       end

lib/logstash/outputs/s3/size_rotation_policy.rb

@@ -17,7 +17,7 @@ def rotate?(file)
           file.size >= size_file
         end
 
-        def need_periodic?
+        def needs_periodic?
           false
         end
       end

lib/logstash/outputs/s3/temporary_file.rb

@@ -10,9 +10,10 @@ class S3
       # It make it more OOP and easier to reason with the paths.
       class TemporaryFile
         extend Forwardable
-        DELEGATES_METHODS = [:path, :write, :close, :size, :fsync]
 
-        def_delegators :@fd, *DELEGATES_METHODS
+        def_delegators :@fd, :path, :write, :close, :fsync
+
+        attr_reader :fd
 
         def initialize(key, fd, temp_path)
           @fd = fd
@@ -29,6 +30,17 @@ def temp_path
           @temp_path
         end
 
+        def size
+          # Use the fd size to get the accurate result,
+          # so we dont have to deal with fsync
+          # if the file is close we will use the File::size
+          begin
+            @fd.size
+          rescue IOError
+            ::File.size(path)
+          end
+        end
+
         def key
           @key.gsub(/^\//, "")
         end
@@ -45,6 +57,14 @@ def delete!
         def empty?
           size == 0
         end
+
+        def self.create_from_existing_file(file_path, temporary_folder)
+          key_parts = Pathname.new(file_path).relative_path_from(temporary_folder).to_s.split(::File::SEPARATOR)
+
+          TemporaryFile.new(key_parts.slice(1, key_parts.size).join("/"),
+                         ::File.open(file_path, "r"),
+                         ::File.join(temporary_folder, key_parts.slice(0, 1)))
+        end
       end
     end
   end

lib/logstash/outputs/s3/temporary_file_factory.rb

@@ -2,6 +2,8 @@
 require "socket"
 require "securerandom"
 require "fileutils"
+require "zlib"
+require "forwardable"
 
 module LogStash
   module Outputs
@@ -80,13 +82,41 @@ def new_file
           FileUtils.mkdir_p(::File.join(path, prefix))
 
           io = if gzip?
-                 Zlib::GzipWriter.open(::File.join(path, key))
+                 # We have to use this wrapper because we cannot access the size of the
+                 # file directly on the gzip writer.
+                 IOWrappedGzip.new(::File.open(::File.join(path, key), FILE_MODE))
                else
                  ::File.open(::File.join(path, key), FILE_MODE)
                end
 
           TemporaryFile.new(key, io, path)
         end
+
+        class IOWrappedGzip
+          extend Forwardable
+
+          def_delegators :@gzip_writer, :write, :close
+          attr_reader :file_io, :gzip_writer
+
+          def initialize(file_io)
+            @file_io = file_io
+            @gzip_writer = Zlib::GzipWriter.open(file_io)
+          end
+
+          def path
+            @gzip_writer.to_io.path
+          end
+
+          def size
+            # to get the current file size
+            @gzip_writer.flush
+            @gzip_writer.to_io.size
+          end
+
+          def fsync
+            @gzip_writer.to_io.fsync
+          end
+        end
       end
     end
   end

lib/logstash/outputs/s3/time_rotation_policy.rb

@@ -17,7 +17,7 @@ def rotate?(file)
           file.size > 0 && (Time.now - file.ctime) >= time_file
         end
 
-        def need_periodic?
+        def needs_periodic?
           true
         end
       end

lib/logstash/outputs/s3/uploader.rb

@@ -1,6 +1,6 @@
 # encoding: utf-8
 require "logstash/util"
-require "aws-sdk-resources"
+require "aws-sdk"
 
 module LogStash
   module Outputs
@@ -42,8 +42,7 @@ def upload(file, options = {})
             #
             # Thread might be stuck here, but I think its better than losing anything
             # its either a transient errors or something bad really happened.
-            sleep(TIME_BEFORE_RETRYING_SECONDS)
-            logger.error("Uploading failed, retrying", :exception => e, :path => file.path)
+            logger.error("Uploading failed, retrying", :exception => e, :path => file.path, :backtrace => e.backtrace)
             retry
           end
 

spec/integration/dynamic_prefix_spec.rb

@@ -85,7 +85,7 @@
     end
 
     it "creates dated path" do
-      re = /^#{sandbox}\/\d{4}-\d{2}-\d{2}\/ls\.s3\./
+      re = /^#{sandbox}\/\d{4}-\d{2}-\d{1,2}\/ls\.s3\./
       expect(bucket_resource.objects(:prefix => sandbox).first.key).to match(re)
     end
   end

spec/integration/gzip_file_spec.rb

@@ -0,0 +1,62 @@
+# encoding: utf-8
+require_relative "../spec_helper"
+require "logstash/outputs/s3"
+require "logstash/codecs/line"
+require "stud/temporary"
+
+describe "Gzip File Time rotation with constant write", :integration => true do
+  include_context "setup plugin"
+
+  let(:time_file) { 0.004 }
+  let(:options) { main_options.merge({ "encoding" => "gzip",
+                                       "rotation_strategy" => "time" }) }
+  let(:number_of_events) { 5000 }
+  let(:batch_size) { 125 }
+  let(:event_encoded) { "Hello world" }
+  let(:batch) do
+    b = {}
+    number_of_events.times do
+      event = LogStash::Event.new({ "message" => event_encoded })
+      b[event] = "#{event_encoded}\n"
+    end
+    b
+  end
+  let(:minimum_number_of_time_rotation) { 3 }
+  let(:batch_step) { (number_of_events / minimum_number_of_time_rotation).ceil }
+
+  before do
+    clean_remote_files(prefix)
+    subject.register
+
+    # simulate batch read/write
+    batch.each_slice(batch_step) do |batch_time|
+      batch_time.each_slice(batch_size) do |smaller_batch|
+        subject.multi_receive_encoded(smaller_batch)
+      end
+      sleep(1)
+    end
+
+    subject.close
+  end
+
+  it "creates multiples files" do
+    # using close will upload the current file
+    expect(bucket_resource.objects(:prefix => prefix).count).to be_between(minimum_number_of_time_rotation, minimum_number_of_time_rotation + 1).inclusive
+  end
+
+  it "Persists all events" do
+    download_directory = Stud::Temporary.pathname
+
+    FileUtils.rm_rf(download_directory)
+    FileUtils.mkdir_p(download_directory)
+
+    counter = 0
+    bucket_resource.objects(:prefix => prefix).each do |object|
+      target = File.join(download_directory, "#{counter}.gz")
+      object.get(:response_target => target)
+      counter += 1
+    end
+
+    expect(Dir.glob(File.join(download_directory, "**", "*.gz")).inject(0) { |sum, f| sum + Zlib::GzipReader.new(File.open(f)).readlines.size }).to eq(number_of_events)
+  end
+end