add feature to detect too low collection ratio

Signed-off-by: Daijiro Fukuda <fukuda@clear-code.com>

Author: daipom

Gemfile

@@ -9,3 +9,4 @@ gem "irb"
 gem "rake", "~> 13.0"
 
 gem "test-unit", "~> 3.0"
+gem 'test-unit-rr'

fluent-tail_checker.gemspec

@@ -31,9 +31,5 @@ Gem::Specification.new do |spec|
   spec.executables = spec.files.grep(%r{\Aexe/}) { |f| File.basename(f) }
   spec.require_paths = ["lib"]
 
-  # Uncomment to register a new dependency of your gem
-  # spec.add_dependency "example-gem", "~> 1.0"
-
-  # For more information and examples about making a new gem, check out our
-  # guide at: https://bundler.io/guides/creating_gem.html
+  spec.add_dependency "fluentd", "~> 1.0"
 end

lib/fluent/tail_checker/collection_ratio_checker.rb

@@ -0,0 +1,107 @@
+# frozen_string_literal: true
+
+# Copyright 2025 Daijiro Fukuda
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+require "fluent/env"
+
+# For compatibility with Fluentd v1.15.2 and earlier.
+# See https://github.com/fluent/fluentd/pull/3883
+if Fluent.windows?
+  require "fluent/file_wrapper"
+else
+  Fluent::FileWrapper = File
+end
+
+module Fluent
+  module TailChecker
+    class CollectionRatioChecker
+      def initialize(posfile, follow_inodes)
+        @posfile = posfile
+        @follow_inodes = follow_inodes
+        @collection_ratio_threshold = 0.8
+      end
+
+      def check
+        unacceptable_collection_ratio_found = false
+        unacceptable_collection_ratio_path_and_ratio_list = []
+        checked_file_counts = 0
+
+        @posfile.watching_entries.each do |entry|
+          size = get_file_size(entry)
+          next if size.nil?
+
+          checked_file_counts += 1
+          ratio = collection_ratio(entry.pos, size)
+          if ratio < @collection_ratio_threshold
+            unacceptable_collection_ratio_found = true
+            unacceptable_collection_ratio_path_and_ratio_list.append([entry.path, ratio])
+          end
+        end
+
+        puts "Checked collection ratio of #{checked_file_counts} files."
+
+        if unacceptable_collection_ratio_found
+          log_issue(unacceptable_collection_ratio_path_and_ratio_list)
+          return false
+        end
+
+        true
+      end
+
+      def get_file_size(pos_entry)
+        if @follow_inodes
+          get_file_size_from_inode(pos_entry)
+        else
+          get_file_size_from_path(pos_entry)
+        end
+      end
+
+      def get_file_size_from_path(pos_entry)
+        FileTest.size?(pos_entry.path)
+      end
+
+      def get_file_size_from_inode(pos_entry)
+        return nil unless FileTest.exist?(pos_entry.path)
+
+        stat = Fluent::FileWrapper.stat(pos_entry.path)
+        # If follow_inodes is enabled, the inode of the current logfile should match the inode in the pos_file.
+        # It may not match for the rotated logfiles because the path info in the pos_file is not updated.
+        # So, at least, check the current logfile.
+        # For rotated logfiles, check them if inode matches.
+        return nil unless stat.ino == pos_entry.ino
+
+        stat.size
+      end
+
+      def collection_ratio(pos, file_size)
+        return 1.0 if file_size == 0
+
+        pos.to_f / file_size
+      end
+
+      def log_issue(path_and_ratio_list)
+        puts "Collection ratio of some files are too low. Collection of those files may have stopped or may not be keeping up."
+        if @follow_inodes
+          puts "This can be a known log loss issue of the follow_inodes feature that was fixed in Fluentd v1.16.2."
+        end
+
+        puts "Filepaths with too low collection ratio (threshold: #{@collection_ratio_threshold}):"
+        path_and_ratio_list.each do |path, ratio|
+          puts "  #{path} (ratio: #{ratio})"
+        end
+      end
+    end
+  end
+end

lib/fluent/tail_checker/duplicated_pos_checker.rb

@@ -28,16 +28,20 @@ def check
         duplicated_key_found = false
         key_set = Set.new
         duplicated_key_set = Set.new
+        checked_key_counts = 0
 
         @posfile.watching_entries.map do |entry|
           @follow_inodes ? entry.ino : entry.path
         end.each do |key|
+          checked_key_counts += 1
           next if key_set.add?(key)
 
           duplicated_key_found = true
           duplicated_key_set.add(key)
         end
 
+        puts "Checked duplicated pos for #{checked_key_counts} PosEntries."
+
         if duplicated_key_found
           log_issue(duplicated_key_set)
           return false

lib/fluent/tail_checker/pos.rb

@@ -22,8 +22,12 @@ module TailChecker
     class PosFile
       attr_reader :entries
 
-      def initialize(path)
-        @entries = PosFile.load_entries(path)
+      def initialize(entries)
+        @entries = entries
+      end
+
+      def self.load(path)
+        PosFile.new(PosFile.load_entries(path))
       end
 
       def self.load_entries(path)

lib/fluent/tail_checker/tail_check.rb

@@ -18,6 +18,7 @@
 
 require_relative "pos"
 require_relative "duplicated_pos_checker"
+require_relative "collection_ratio_checker"
 
 module Fluent
   module TailChecker
@@ -75,7 +76,11 @@ def check
 
         @pos_filepaths.each do |path|
           puts "\nCheck #{path}."
-          succeeded = check_pos_file(path) && succeeded
+          pos_file = try_to_open_pos_file(path)
+          next if pos_file.nil?
+
+          succeeded = DuplicatedPosChecker.new(pos_file, @follow_inodes).check && succeeded
+          succeeded = CollectionRatioChecker.new(pos_file, @follow_inodes).check && succeeded
         end
 
         puts "\nAll check completed."
@@ -90,11 +95,8 @@ def check
         true
       end
 
-      def check_pos_file(path)
-        succeeded = true
-        posfile = PosFile.new(path)
-        succeeded = DuplicatedPosChecker.new(posfile, @follow_inodes).check && succeeded
-        succeeded
+      def try_to_open_pos_file(path)
+        PosFile.load(path)
       rescue => e
         $stderr.puts "Can not open the file. Skipped. Path: #{path}, Error: #{e}"
       end

test/data/log/bar.log

@@ -0,0 +1,3 @@
+barbar
+barbarbarbar
+barbarbarbarbarbarbarbar

test/data/log/foo.log

@@ -0,0 +1,4 @@
+foo
+foofoo
+foofoofoo
+foofoofoofoo

test/data/log/foo.log.1

@@ -0,0 +1,13 @@
+foo
+foofoo
+foofoofoo
+foofoofoofoo
+foo
+foofoo
+foofoofoo
+foofoofoofoo
+foo
+foofoo
+foofoofoo
+foofoofoofoo
+foo

test/fluent/tail_checker/collection_ratio_checker_test.rb

@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class Fluent::TailChecker::CollectionRatioCheckerTest < Test::Unit::TestCase
+  data("Normal ratio", [["test/data/log/foo.log", "test/data/log/bar.log"], 0.9, false, true])
+  data("Too low ratio", [["test/data/log/foo.log", "test/data/log/bar.log"], 0.7, false, false])
+  data("Normal ratio with follow_inodes", [["test/data/log/foo.log", "test/data/log/bar.log"], 0.9, true, true])
+  data("Too low ratio with follow_inodes", [["test/data/log/foo.log", "test/data/log/foo.log.1"], 0.7, true, false])
+  test "Check should return false when too low collection ratio is detected" do |(paths, ratio, follow_inodes, expected)|
+    pos_entries = paths.map do |path|
+      stat = Fluent::FileWrapper.stat(path)
+      Fluent::TailChecker::PosEntry.new(path, stat.size * ratio, stat.ino)
+    end
+    pos_file = Fluent::TailChecker::PosFile.new(pos_entries)
+    checker = Fluent::TailChecker::CollectionRatioChecker.new(pos_file, follow_inodes)
+
+    result = checker.check
+
+    assert_equal(expected, result)
+  end
+end