Add integration tests for GZIP recovery cases.

Author: mashhurs

lib/logstash/outputs/s3.rb

@@ -398,7 +398,6 @@ def restore_from_crash
     @crash_uploader = Uploader.new(bucket_resource, @logger, CRASH_RECOVERY_THREADPOOL)
 
     temp_folder_path = Pathname.new(@temporary_directory)
-
     files = Dir.glob(::File.join(@temporary_directory, "**/*"))
                .select { |file_path| ::File.file?(file_path) }
     under_recovery_files = get_under_recovery_files(files)
@@ -411,15 +410,16 @@ def restore_from_crash
         end
       else
         temp_file = TemporaryFile.create_from_existing_file(file_path, temp_folder_path)
-        if temp_file.size > 0
-          @logger.debug? && @logger.debug("Recovering from crash and uploading", :path => temp_file.path)
-          @crash_uploader.upload_async(temp_file,
-                                       :on_complete => method(:clean_temporary_file),
-                                       :upload_options => upload_options)
-        end
-        # do not remove if Logstash tries to recover but fails
-        if @encoding != GZIP_ENCODING && temp_file.size != 0
-          clean_temporary_file(temp_file)
+        # do not remove or upload if Logstash tries to recover file but fails
+        if temp_file.recoverable?
+          if temp_file.size > 0
+            @logger.debug? && @logger.debug("Recovering from crash and uploading", :path => temp_file.path)
+            @crash_uploader.upload_async(temp_file,
+                                         :on_complete => method(:clean_temporary_file),
+                                         :upload_options => upload_options)
+          else
+            clean_temporary_file(temp_file)
+          end
         end
       end
     end

lib/logstash/outputs/s3/temporary_file.rb

@@ -68,6 +68,11 @@ def empty?
           size == 0
         end
 
+        # only to cover the case where LS cannot restore corrupted file, file is not exist
+        def recoverable?
+          !@fd.nil?
+        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)
 

spec/integration/restore_from_crash_spec.rb

@@ -7,18 +7,17 @@
 describe "Restore from crash", :integration => true do
   include_context "setup plugin"
 
-  let(:options) { main_options.merge({ "restore" => true, "canned_acl" => "public-read-write" }) }
-
   let(:number_of_files) { 5 }
   let(:dummy_content) { "foobar\n" * 100 }
-  let(:factory) { LogStash::Outputs::S3::TemporaryFileFactory.new(prefix, tags, "none", temporary_directory)}
 
   before do
     clean_remote_files(prefix)
   end
 
-
   context 'with a non-empty tempfile' do
+    let(:options) { main_options.merge({ "restore" => true, "canned_acl" => "public-read-write" }) }
+    let(:factory) { LogStash::Outputs::S3::TemporaryFileFactory.new(prefix, tags, "none", temporary_directory)}
+
     before do
       # Creating a factory always create a file
       factory.current.write(dummy_content)
@@ -41,6 +40,9 @@
   end
 
   context 'with an empty tempfile' do
+    let(:options) { main_options.merge({ "restore" => true, "canned_acl" => "public-read-write" }) }
+    let(:factory) { LogStash::Outputs::S3::TemporaryFileFactory.new(prefix, tags, "none", temporary_directory)}
+
     before do
       factory.current
       factory.rotate!
@@ -63,5 +65,68 @@
       expect(bucket_resource.objects(:prefix => prefix).count).to eq(0)
     end
   end
+
+  context "#gzip encoding" do
+    let(:options) { main_options.merge({ "restore" => true, "canned_acl" => "public-read-write", "encoding" => "gzip" }) }
+    let(:factory) { LogStash::Outputs::S3::TemporaryFileFactory.new(prefix, tags, "gzip", temporary_directory)}
+    describe "with empty recovered file" do
+      before do
+        # Creating a factory always create a file
+        factory.current.write('')
+        factory.current.fsync
+        factory.current.close
+      end
+
+      it 'should not upload and not remove temp file' do
+        subject.register
+        try(20) do
+          expect(bucket_resource.objects(:prefix => prefix).count).to eq(0)
+          expect(Dir.glob(File.join(temporary_directory, "*")).size).to eq(1)
+        end
+      end
+    end
+
+    describe "with healthy recovered, size is greater than zero file" do
+      before do
+        # Creating a factory always create a file
+        factory.current.write(dummy_content)
+        factory.current.fsync
+        factory.current.close
+
+        (number_of_files - 1).times do
+          factory.rotate!
+          factory.current.write(dummy_content)
+          factory.current.fsync
+          factory.current.close
+        end
+      end
+
+      it 'should recover, upload to S3 and remove temp file' do
+        subject.register
+        try(20) do
+          expect(bucket_resource.objects(:prefix => prefix).count).to eq(number_of_files)
+          expect(Dir.glob(File.join(temporary_directory, "*")).size).to eq(0)
+          expect(bucket_resource.objects(:prefix => prefix).first.acl.grants.collect(&:permission)).to include("READ", "WRITE")
+        end
+      end
+    end
+
+    describe "with failure when recovering" do
+      before do
+        # Creating a factory always create a file
+        factory.current.write(dummy_content)
+        factory.current.fsync
+      end
+
+      it 'should not upload to S3 and not remove temp file' do
+        subject.register
+        try(20) do
+          expect(bucket_resource.objects(:prefix => prefix).count).to eq(0)
+          expect(Dir.glob(File.join(temporary_directory, "*")).size).to eq(1)
+        end
+      end
+    end
+  end
+
 end
 

spec/supports/helpers.rb

@@ -5,6 +5,7 @@
   let(:bucket) { ENV["AWS_LOGSTASH_TEST_BUCKET"] }
   let(:access_key_id) {  ENV["AWS_ACCESS_KEY_ID"] }
   let(:secret_access_key) { ENV["AWS_SECRET_ACCESS_KEY"] }
+  let(:session_token) { ENV["AWS_SESSION_TOKEN"] }
   let(:size_file) { 100 }
   let(:time_file) { 100 }
   let(:tags) { [] }
@@ -18,14 +19,15 @@
       "temporary_directory" => temporary_directory,
       "access_key_id" => access_key_id,
       "secret_access_key" => secret_access_key,
+      "session_token" => session_token,
       "size_file" => size_file,
       "time_file" => time_file,
       "region" => region,
       "tags" => []
     }
   end
 
-  let(:client_credentials) { Aws::Credentials.new(access_key_id, secret_access_key) }
+  let(:client_credentials) { Aws::Credentials.new(access_key_id, secret_access_key, session_token) }
   let(:bucket_resource) { Aws::S3::Bucket.new(bucket, { :credentials => client_credentials, :region => region }) }
 
   subject { LogStash::Outputs::S3.new(options) }