From 427ea1f6f699c1c9c5f2e3dd8135b9e0c67c9ca3 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Tue, 29 Oct 2019 10:06:49 +0000 Subject: [PATCH 1/2] Add size for each log file Remove debug log Add size file Update size parameter FilePositionEntry adding logs update pos file function in memory add debug size log update file size from pos_file read Save size inside the file and keep it backward compatible Revert "Add log throttling per file" This reverts commit f5bb69b72d5b9608d2505fb046a6fd0748563df7. Remove debug log --- lib/fluent/plugin/in_tail.rb | 60 +++++++++++++++++++++++------------- 1 file changed, 38 insertions(+), 22 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 6a226a544d..7e46f3d729 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -307,7 +307,7 @@ def start_watchers(paths) pe = @pf[path] if @read_from_head && pe.read_inode.zero? begin - pe.update(Fluent::FileWrapper.stat(path).ino, 0) + pe.update(Fluent::FileWrapper.stat(path).ino, 0, 0) rescue Errno::ENOENT $log.warn "#{path} not found. Continuing without tailing it." end @@ -372,7 +372,7 @@ def detach_watcher(tw, close_io = true) tw.close if close_io flush_buffer(tw) if tw.unwatched && @pf - @pf[tw.path].update_pos(PositionFile::UNWATCHED_POSITION) + @pf[tw.path].update_pos(PositionFile::UNWATCHED_POSITION, 0) end end @@ -580,19 +580,19 @@ def on_rotate(stat) # in either case of a and b, seek to the saved position # c) file was once renamed, truncated and then backed # in this case, consider it truncated - @pe.update(inode, 0) if fsize < @pe.read_pos + @pe.update(inode, 0, fsize) if fsize < @pe.read_pos elsif last_inode != 0 # this is FilePositionEntry and fluentd once started. # read data from the head of the rotated file. # logs never duplicate because this file is a rotated new file. - @pe.update(inode, 0) + @pe.update(inode, 0, fize) else # this is MemoryPositionEntry or this is the first time fluentd started. # seek to the end of the any files. # logs may duplicate without this seek because it's not sure the file is # existent file or rotated new file. pos = @read_from_head ? 0 : fsize - @pe.update(inode, pos) + @pe.update(inode, pos, fsize) end @io_handler = IOHandler.new(self, &method(:wrap_receive_lines)) else @@ -604,10 +604,10 @@ def on_rotate(stat) if stat inode = stat.ino if inode == @pe.read_inode # truncated - @pe.update_pos(0) + @pe.update_pos(0, stat.size) @io_handler.close elsif !@io_handler.opened? # There is no previous file. Reuse TailWatcher - @pe.update(inode, 0) + @pe.update(inode, 0, stat.size) else # file is rotated and new file found watcher_needs_update = true # Handle the old log file before renewing TailWatcher [fluentd#1055] @@ -634,7 +634,7 @@ def on_rotate(stat) def swap_state(pe) # Use MemoryPositionEntry for rotated file temporary mpe = MemoryPositionEntry.new - mpe.update(pe.read_inode, pe.read_pos) + mpe.update(pe.read_inode, pe.read_pos, pe.read_size) @pe = mpe pe # This pe will be updated in on_rotate after TailWatcher is initialized end @@ -766,7 +766,7 @@ def handle_notify unless @lines.empty? if @receive_lines.call(@lines) - @watcher.pe.update_pos(io.pos - @fifo.bytesize) + @watcher.pe.update_pos(io.pos - @fifo.bytesize, io.size) @lines.clear else read_more = false @@ -912,10 +912,10 @@ def [](path) @file_mutex.synchronize { @file.pos = @last_pos - @file.write "#{path}\t0000000000000000\t0000000000000000\n" + @file.write "#{path}\t0000000000000000\t0000000000000000\t0000000000000000\n" seek = @last_pos + path.bytesize + 1 @last_pos = @file.pos - @map[path] = FilePositionEntry.new(@file, @file_mutex, seek, 0, 0) + @map[path] = FilePositionEntry.new(@file, @file_mutex, seek, 0, 0, 0) } end @@ -926,7 +926,7 @@ def self.parse(file) map = {} file.pos = 0 file.each_line {|line| - m = /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)/.match(line) + m = /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)[\t]?([0-9a-fA-F]+)?/.match(line) unless m $log.warn "Unparsable line in pos_file: #{line}" next @@ -934,8 +934,9 @@ def self.parse(file) path = m[1] pos = m[2].to_i(16) ino = m[3].to_i(16) + size = m[4].to_i(16) seek = file.pos - line.bytesize + path.bytesize + 1 - map[path] = FilePositionEntry.new(file, file_mutex, seek, pos, ino) + map[path] = FilePositionEntry.new(file, file_mutex, seek, pos, size, ino) } new(file, file_mutex, map, file.pos) end @@ -944,7 +945,7 @@ def self.parse(file) def self.compact(file) file.pos = 0 existent_entries = file.each_line.map { |line| - m = /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)/.match(line) + m = /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)[\t]?([0-9a-fA-F]+)?/.match(line) unless m $log.warn "Unparsable line in pos_file: #{line}" next @@ -952,8 +953,9 @@ def self.compact(file) path = m[1] pos = m[2].to_i(16) ino = m[3].to_i(16) + m[4].nil? ? size = 0 : size = m[4].to_i(16) # 32bit inode converted to 64bit at this phase - pos == UNWATCHED_POSITION ? nil : ("%s\t%016x\t%016x\n" % [path, pos, ino]) + pos == UNWATCHED_POSITION ? nil : ("%s\t%016x\t%016x\t%016x\n" % [path, pos, ino, size]) }.compact file.pos = 0 @@ -971,35 +973,42 @@ class FilePositionEntry LN_OFFSET = 33 SIZE = 34 - def initialize(file, file_mutex, seek, pos, inode) + def initialize(file, file_mutex, seek, pos, size, inode) @file = file @file_mutex = file_mutex @seek = seek @pos = pos + @size = size @inode = inode end - def update(ino, pos) + def update(ino, pos, size) @file_mutex.synchronize { @file.pos = @seek - @file.write "%016x\t%016x" % [pos, ino] + @file.write "%016x\t%016x\t%016x" % [pos, ino, size] } @pos = pos @inode = ino + @size = size end - def update_pos(pos) + def update_pos(pos, size) @file_mutex.synchronize { @file.pos = @seek - @file.write "%016x" % pos + @file.write "%016x\t%016x\t%016x" % [pos, @inode, size] } @pos = pos + @size = size end def read_inode @inode end + def read_size + @size + end + def read_pos @pos end @@ -1009,21 +1018,28 @@ class MemoryPositionEntry def initialize @pos = 0 @inode = 0 + @size = 0 end - def update(ino, pos) + def update(ino, pos, size) @inode = ino @pos = pos + @size = size end - def update_pos(pos) + def update_pos(pos, size) @pos = pos + @size = size end def read_pos @pos end + def read_size + @size + end + def read_inode @inode end From 902af996dadd0c482bfc803391b2ce21ff87d576 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Mon, 4 Nov 2019 15:30:22 +0000 Subject: [PATCH 2/2] Add legacy migration pos file test --- test/plugin/test_in_tail.rb | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index d67cbdce3c..8d506bf40d 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -1050,6 +1050,32 @@ def test_pos_file_dir_creation d.instance_shutdown end + def test_pos_file_legacy_file_format_migrate_to_new_pos_file_format + config = config_element("", "", { + "tag" => "tail", + "path" => "#{TMP_DIR}/*.txt", + "format" => "none", + "pos_file" => "#{TMP_DIR}/pos/tail.pos", + "read_from_head" => true, + "refresh_interval" => 1 + }) + d = create_driver(config, false) + d.run(expect_emits: 1, shutdown: false) do + File.open("#{TMP_DIR}/pos/tail.pos", "wb") { |f| + f.puts "#{TMP_DIR}/tail.txt\t0000000000000000\t0000000000000000\n" + } + File.open("#{TMP_DIR}/tail.txt", "ab") { |f| f.puts "test3\n" } + end + assert_path_exist("#{TMP_DIR}/pos/tail.pos") + + first_line_pos_file = '' + File.open("#{TMP_DIR}/pos/tail.pos", "r") { |f| first_line_pos_file = f.readline } + assert_equal(5, /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)/.match(first_line_pos_file).length) + + cleanup_directory(TMP_DIR) + d.instance_shutdown + end + def test_z_refresh_watchers plugin = create_driver(EX_CONFIG, false).instance sio = StringIO.new