fix broken gzip file produced sometimes

Author: nit23uec

Diff for: lib/logstash/outputs/file.rb

@@ -77,6 +77,8 @@ class LogStash::Outputs::File < LogStash::Outputs::Base
   default :codec, "json_lines"
 
   def register
+
+
     require "fileutils" # For mkdir_p
 
     @files = {}
@@ -119,24 +121,49 @@ def multi_receive_encoded(events_and_encoded)
           fd.write(chunks.last)
         else
           # append to the file
-          chunks.each {|chunk| fd.write(chunk) }
+          chunks.each {|chunk|
+           fd.write(chunk) 
+          }
         end
-        fd.flush unless @flusher && @flusher.alive?
+        onflush(fd, path) unless @flusher && @flusher.alive?
       end
-
       close_stale_files
     end
   end
 
+  def onflush(fd, path)
+    fd.flush
+    if @files[path][:isTempFile]
+      copy_to_gzip(fd, path)
+    end
+  end
+
+  def copy_to_gzip(fd, path)
+    zipfd = getFile(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)
+          fileObj[:fd].close
+          @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 +226,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 +245,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 +269,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
+      makeDir(originalPath)
+      gzFile = getFile(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)
+    makeDir(path)
+    fd = getFile(path)
+    @files[originalPath] = {:fd => IOWriter.new(fd), :isTempFile => gzip}
+    return @files[originalPath][:fd]
+  end
 
+  def makeDir(path)
     dir = File.dirname(path)
     if !Dir.exist?(dir)
       @logger.info("Creating directory", :directory => dir)
@@ -260,7 +313,9 @@ def open(path)
         FileUtils.mkdir_p(dir)
       end
     end
+  end
 
+  def getFile(path)
     # work around a bug opening fifos (bug JRUBY-6280)
     stat = File.stat(path) rescue nil
     if stat && stat.ftype == "fifo"
@@ -272,10 +327,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
 
   ##
@@ -380,4 +432,4 @@ def method_missing(method_name, *args, &block)
       super
     end
   end
-end
+end