add feature to detect duplicated pos entries

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

Author: daipom

.github/workflows/main.yml

@@ -42,6 +42,5 @@ jobs:
           bundler-cache: true
       - name: Install
         run: rake install
-      - name: Run exe
-        run: |
-          tailcheck
+      - name: Test command
+        run: test/script/command-test.bash

exe/tailcheck

@@ -1,9 +1,7 @@
 #!/usr/bin/env ruby
 # frozen_string_literal: true
 
-require 'fluent/tail_checker'
-
-puts Fluent::TailChecker::VERSION
+require "fluent/tail_checker"
 
 command = Fluent::TailChecker::TailCheck.new
 exit(command.run)

lib/fluent/tail_checker/duplicated_pos_checker.rb

@@ -0,0 +1,60 @@
+# 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 "set"
+
+module Fluent
+  module TailChecker
+    class DuplicatedPosChecker
+      def initialize(posfile, follow_inodes)
+        @posfile = posfile
+        @follow_inodes = follow_inodes
+      end
+
+      def check
+        duplicated_key_found = false
+        key_set = Set.new
+        duplicated_key_set = Set.new
+
+        @posfile.entries.filter do |entry|
+          not entry.unwatched?
+        end.map do |entry|
+          @follow_inodes ? entry.ino : entry.path
+        end.each do |key|
+          next if key_set.add?(key)
+
+          duplicated_key_found = true
+          duplicated_key_set.add(key)
+        end
+
+        if duplicated_key_found
+          log_issue(duplicated_key_set)
+          return false
+        end
+
+        true
+      end
+
+      def log_issue(duplicated_keys)
+        if @follow_inodes
+          puts "Duplicated PosEntries are found with follow_inodes. Unknown anomalies may be occurring."
+        else
+          puts "Duplicated PosEntries are found. This is a known log loss issue that was fixed in Fluentd v1.16.3."
+        end
+      end
+    end
+  end
+end

lib/fluent/tail_checker/pos.rb

@@ -0,0 +1,63 @@
+# 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.
+
+module Fluent
+  module TailChecker
+    UNWATCHED_POSITION = 0xffffffffffffffff
+    POSITION_FILE_ENTRY_REGEX = /^([^\t]+)\t([0-9a-fA-F]+)\t([0-9a-fA-F]+)/
+
+    class PosFile
+      attr_reader :entries
+
+      def initialize(path)
+        @entries = PosFile.load_entries(path)
+      end
+
+      def self.load_entries(path)
+        entries = []
+
+        File.open(path, File::RDONLY|File::BINARY) do |file|
+          file.each_line do |line|
+            m = POSITION_FILE_ENTRY_REGEX.match(line)
+            next if m.nil?
+
+            entries << PosEntry.new(
+              m[1],
+              m[2].to_i(16),
+              m[3].to_i(16),
+            )
+          end
+        end
+
+        entries
+      end
+    end
+
+    class PosEntry
+      attr_reader :path, :pos, :ino
+
+      def initialize(path, pos, ino)
+        @path = path
+        @pos = pos
+        @ino = ino
+      end
+
+      def unwatched?
+        @pos == UNWATCHED_POSITION
+      end
+    end
+  end
+end

lib/fluent/tail_checker/tail_check.rb

@@ -14,14 +14,90 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+require "optparse"
+
+require_relative "pos"
+require_relative "duplicated_pos_checker"
+
 module Fluent
   module TailChecker
     class TailCheck
-      def run
-        # TODO
-        puts "TailCheck command run."
+      def initialize
+        @pos_filepaths = []
+        @follow_inodes = false
+      end
+
+      def run(argv=ARGV)
+        parse_command_line(argv)
+        @pos_filepaths = validate_paths(@pos_filepaths).to_a
+        check
+      end
+
+      def parse_command_line(argv)
+        parser = OptionParser.new
+        parser.version = VERSION
+        parser.banner = <<~BANNER
+          Usage: tailcheck [OPTIONS] POS_FILE...
+          Example: tailcheck /path/to/pos1 /path/to/pos2
+          Example: tailcheck /path/to/pos/*
+          Example: tailcheck --follow_inodes /path/to/pos_with_follow_inodes
+          Options:
+
+        BANNER
+
+        parser.on("--follow_inodes", "Check the specified pos files with the condition that the follow_inodes feature is enabled.", "Default: Disabled") do
+          @follow_inodes = true
+        end
+
+        @pos_filepaths = parser.parse(argv)
+      end
+
+      def validate_paths(paths)
+        Enumerator.new do |y|
+          paths.each do |path|
+            unless FileTest.exist?(path)
+              $stderr.puts "File does not exist. Skipped. Path: #{path}"
+              next
+            end
+
+            y << path
+          end
+        end
+      end
+
+      def check
+        if @pos_filepaths.empty?
+          $stderr.puts "No pos_file to be checked. Please specify valid pos_file paths."
+          return false
+        end
+
+        succeeded = true
+
+        @pos_filepaths.each do |path|
+          puts "\nCheck #{path}."
+          succeeded = check_pos_file(path) && succeeded
+        end
+
+        puts "\nAll check completed."
+
+        unless succeeded
+          puts "Some anomalies are found. Please check whether there is any log loss."
+          # TODO add message about how to concact the community or us.
+          return false
+        end
+
+        puts "There is no anomalies."
         true
       end
+
+      def check_pos_file(path)
+        succeeded = true
+        posfile = PosFile.new(path)
+        succeeded = DuplicatedPosChecker.new(posfile, @follow_inodes).check && succeeded
+        succeeded
+      rescue => e
+        $stderr.puts "Can not open the file. Skipped. Path: #{path}, Error: #{e}"
+      end
     end
   end
 end

test/data/pos_duplicate_unwatched_inode

@@ -0,0 +1,8 @@
+/test/foo.log	ffffffffffffffff	00000000060e3795
+/test/foo.log.1	ffffffffffffffff	00000000060e3794
+/test/foo.log.2	000000000000000c	00000000060e3793
+/test/foo.log.3	ffffffffffffffff	00000000060e3796
+/test/foo.log	0000000000000004	00000000060e379a
+/test/foo.log.1	000000000000000c	00000000060e3795
+/test/foo.log	0000000000000004	00000000060e3796
+/test/foo.log	0000000000000004	00000000060e3793

test/data/pos_duplicate_unwatched_path

@@ -0,0 +1,5 @@
+/test/foo.log	ffffffffffffffff	00000000060e3789
+/test/bar.log	0000000000000008	00000000060e3794
+/test/foo.log	0000000000000008	00000000060e3789
+/test/foo.log	ffffffffffffffff	0000000000000000
+/test/foo.log	000000000000000c	00000000060e378c

test/data/pos_follow_inodes_normal

@@ -0,0 +1,8 @@
+/test/foo.log	ffffffffffffffff	00000000060e3795
+/test/foo.log.1	ffffffffffffffff	00000000060e3794
+/test/foo.log.2	ffffffffffffffff	00000000060e3793
+/test/foo.log.3	ffffffffffffffff	00000000060e3796
+/test/foo.log	0000000000000004	00000000060e379a
+/test/foo.log.1	000000000000000c	00000000060e3795
+/test/foo.log	0000000000000004	00000000060e3796
+/test/foo.log	0000000000000004	00000000060e3793

test/data/pos_normal

@@ -0,0 +1,3 @@
+/test/foo.log	ffffffffffffffff	00000000060e3789
+/test/bar.log	0000000000000008	00000000060e3794
+/test/foo.log	0000000000000004	00000000060e378a

test/fluent/tail_checker_test.rb

@@ -9,7 +9,78 @@ class Fluent::TailCheckerTest < Test::Unit::TestCase
     end
   end
 
-  test "something useful" do
-    assert_equal("actual", "actual")
+  sub_test_case "Parse args" do
+    data(
+      "Minimum",
+      [
+        ["/path/to/pos"],
+        { pos_filepaths: ["/path/to/pos"], follow_inodes: false },
+      ]
+    )
+    data(
+      "Full",
+      [
+        ["--follow_inodes", "/path/to/pos", "/path/to/pos2"],
+        { pos_filepaths: ["/path/to/pos", "/path/to/pos2"], follow_inodes: true },
+      ]
+    )
+    test "Correct args" do |(args, expected)|
+      tail_check = Fluent::TailChecker::TailCheck.new
+
+      tail_check.parse_command_line(args)
+
+      result = {
+        pos_filepaths: tail_check.instance_variable_get(:@pos_filepaths),
+        follow_inodes: tail_check.instance_variable_get(:@follow_inodes),
+      }
+      assert_equal(expected, result)
+    end
+
+    data("Invalid options", ["--foo", "/path/to/pos"])
+    test "Raise error for invalid options" do |args|
+      tail_check = Fluent::TailChecker::TailCheck.new
+
+      assert_raise(OptionParser::InvalidOption) do
+        tail_check.parse_command_line(args)
+      end
+    end
+  end
+
+  sub_test_case "Validate pos_file paths" do
+    data(
+      "Mixed existant and nonexistant paths",
+      [
+        ["test/data/pos_normal", "foo", "test/data/pos_follow_inodes_normal", "bar"],
+        ["test/data/pos_normal", "test/data/pos_follow_inodes_normal"],
+      ]
+    )
+    test "Exclude nonexistant paths" do |(paths, expected)|
+      tail_check = Fluent::TailChecker::TailCheck.new
+
+      validated_paths = tail_check.validate_paths(paths).to_a
+
+      assert_equal(expected, validated_paths)
+    end
+  end
+
+  sub_test_case "Check" do
+    data("No pos_file", [[], false, false])
+    data("No pos_file with follow_inodes", [[], true, false])
+    data("Normal", [["test/data/pos_normal"], false, true])
+    data("Duplicated unwatched paths", [["test/data/pos_duplicate_unwatched_path"], false, false])
+    data("Normal with follow_inodes", [["test/data/pos_follow_inodes_normal"], true, true])
+    data("Duplicated unwatched inodes with follow_inodes", [["test/data/pos_duplicate_unwatched_inode"], true, false])
+    data("Duplicated unwatched paths with follow_inodes", [["test/data/pos_duplicate_unwatched_path"], true, true])
+    data("Normal multiple", [["test/data/pos_follow_inodes_normal", "test/data/pos_duplicate_unwatched_path"], true, true])
+    data("Anomaly multiple", [["test/data/pos_follow_inodes_normal", "test/data/pos_duplicate_unwatched_inode"], true, false])
+    test "Return false when an anomaly is detected or there is no target" do |(paths, follow_inodes, expected)|
+      tail_check = Fluent::TailChecker::TailCheck.new
+      tail_check.instance_variable_set(:@pos_filepaths, paths)
+      tail_check.instance_variable_set(:@follow_inodes, follow_inodes)
+
+      result = tail_check.check
+
+      assert_equal(expected, result)
+    end
   end
 end