fix broken gzip file produced sometimes

Author: nit23uec

lib/logstash/outputs/file.rb

@@ -121,22 +121,53 @@ def multi_receive_encoded(events_and_encoded)
           # append to the file
           chunks.each {|chunk| fd.write(chunk) }
         end
-        fd.flush unless @flusher && @flusher.alive?
+        on_flush(fd, path) unless @flusher && @flusher.alive?
       end
 
       close_stale_files
     end
   end
 
+  def on_flush(fd, path)
+    fd.flush
+    if @files[path][:isTempFile]
+      copy_to_gzip(fd, path)
+    end
+  end
+
+  def copy_to_gzip(fd, path)
+    zipfd = get_file(path)
+    zipfd = Zlib::GzipWriter.new(zipfd)
+    fd.seek(0, IO::SEEK_SET)
+    data = fd.read
+    fd.truncate(0)
+    fd.seek(0, IO::SEEK_SET)
+    if @write_behavior == "overwrite"
+      zipfd.truncate(0)
+      zipfd.seek(0, IO::SEEK_SET)
+    end
+    zipfd.write(data)
+    zipfd.flush
+    zipfd.to_io.flush
+    zipfd.close
+  end
+
   def close
     @flusher.stop unless @flusher.nil?
     @io_mutex.synchronize do
       @logger.debug("Close: closing files")
 
-      @files.each do |path, fd|
+      @files.each do |path, fileObj|
         begin
-          fd.close
-          @logger.debug("Closed file #{path}", :fd => fd)
+          if fileObj[:isTempFile]
+             copy_to_gzip(fileObj[:fd], path)
+          end
+          fileObj[:fd].close
+          if fileObj[:isTempFile] && File.exist?(fileObj[:fd].path)
+             File.delete(fileObj[:fd].path)
+             @logger.debug("Deleted temp file ", :path => fileObj[:fd].path)
+          end
+          @logger.debug("Closed file #{fileObj[:fd]}", :fd => fileObj[:fd])
         rescue Exception => e
           @logger.error("Exception while flushing and closing files.", :exception => e)
         end
@@ -199,9 +230,12 @@ def flush_pending_files
     @io_mutex.synchronize do
       @logger.debug("Starting flush cycle")
 
-      @files.each do |path, fd|
-        @logger.debug("Flushing file", :path => path, :fd => fd)
-        fd.flush
+      @files.each do |path, fileObj|
+        @logger.debug("Flushing file", :path => path, :fd => fileObj[:fd])
+        fileObj[:fd].flush
+        if fileObj[:isTempFile]
+          copy_to_gzip(fileObj[:fd], path)
+        end
       end
     end
   rescue => e
@@ -215,15 +249,18 @@ def close_stale_files
     return unless now - @last_stale_cleanup_cycle >= @stale_cleanup_interval
 
     @logger.debug("Starting stale files cleanup cycle", :files => @files)
-    inactive_files = @files.select { |path, fd| not fd.active }
+    inactive_files = @files.select { |path, fileObj| not fileObj[:fd].active }
     @logger.debug("%d stale files found" % inactive_files.count, :inactive_files => inactive_files)
-    inactive_files.each do |path, fd|
+    inactive_files.each do |path, fileObj|
       @logger.info("Closing file %s" % path)
-      fd.close
+      fileObj[:fd].close
+      if fileObj[:isTempFile] && File.exist?(fileObj[:fd].path)
+        File.delete(fileObj[:fd].path)
+      end
       @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 }
+    @files.each { |path, fileObj| fileObj[:fd].active = false }
     @last_stale_cleanup_cycle = now
   end
 
@@ -236,21 +273,41 @@ def deleted?(path)
   end
 
   def open(path)
+    originalPath = path
     if !deleted?(path) && cached?(path)
-      return @files[path]
+      return @files[path][:fd]
     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)
+        return @files[path][:fd] if cached?(path)
+      end
+    end
+
+    #Fix for broken gzip issue.
+    if gzip
+      tmpfile = java.io.File.createTempFile("outfile-", "-temp");
+      path = tmpfile.path
+      #create file at original path also, so that temp file is not created again
+      make_dir(originalPath)
+      gzFile = get_file(originalPath)
+      #if gzFile is fifo type, file writer object is returned that needs to closed.
+      if gzFile.class == Java::JavaIo::FileWriter
+        gzFile.close
       end
     end
 
     @logger.info("Opening file", :path => path)
+    make_dir(path)
+    fd = get_file(path)
+    @files[originalPath] = {:fd => IOWriter.new(fd), :isTempFile => gzip}
+    return @files[originalPath][:fd]
+  end
 
+  def make_dir(path)
     dir = File.dirname(path)
     if !Dir.exist?(dir)
       @logger.info("Creating directory", :directory => dir)
@@ -260,7 +317,9 @@ def open(path)
         FileUtils.mkdir_p(dir)
       end
     end
+  end
 
+  def get_file(path)
     # work around a bug opening fifos (bug JRUBY-6280)
     stat = File.stat(path) rescue nil
     if stat && stat.ftype == "fifo"
@@ -272,10 +331,7 @@ def open(path)
         fd = File.new(path, "a+")
       end
     end
-    if gzip
-      fd = Zlib::GzipWriter.new(fd)
-    end
-    @files[path] = IOWriter.new(fd)
+    return fd
   end
 
   ##

spec/outputs/file_spec.rb

@@ -69,9 +69,21 @@
 
       agent do
         line_num = 0
+        global_events = []
         # Now check all events for order and correctness.
-        events = Zlib::GzipReader.open(tmp_file.path).map {|line| LogStash::Event.new(LogStash::Json.load(line)) }
-        sorted = events.sort_by {|e| e.get("sequence")}
+        File.open(tmp_file.path) do |file|
+          zio = file
+          loop do
+            io = Zlib::GzipReader.new(zio)
+            events = io.map {|line| LogStash::Event.new(LogStash::Json.load(line)) }
+            global_events = global_events + events
+            unused = io.unused
+            io.finish
+            break if unused.nil?
+            zio.pos -= unused.length
+          end
+        end
+        sorted = global_events.sort_by {|e| e.get("sequence")}
         sorted.each do |event|
           insist {event.get("message")} == "hello world"
           insist {event.get("sequence")} == line_num