feat: Add directory configuration for pos_file

This commit adds new configuration options for the tail input plugin
to control the pos_file directory permissions and ownership:

- Add pos_dir_perm parameter for directory permissions
- Add pos_dir_owner parameter for directory owner
- Add pos_dir_group parameter for directory group
- Use Etc.getpwnam and Etc.getgrnam for proper user/group resolution

Example config:
<source>
  @type tail
  pos_file /var/log/td-agent/httpd/file.pos
  pos_dir_perm 0770
  pos_dir_owner 'root'
  pos_dir_group 'wheel'
</source>

Closes fluent#4822

Signed-off-by: kushynoda <egemen.utku3@gmail.com>

lib/fluent/plugin/in_tail.rb

@@ -15,6 +15,7 @@
 #
 
 require 'cool.io'
+require 'etc'
 
 require 'fluent/plugin/input'
 require 'fluent/config/error'
@@ -116,6 +117,12 @@ def initialize
     config_param :follow_inodes, :bool, default: false
     desc 'Maximum length of line. The longer line is just skipped.'
     config_param :max_line_size, :size, default: nil
+    desc 'The permission of pos file directory'
+    config_param :pos_dir_perm, :string, default: '0755'
+    desc 'The owner of pos file directory'
+    config_param :pos_dir_owner, :string, default: nil
+    desc 'The group of pos file directory' 
+    config_param :pos_dir_group, :string, default: nil
 
     config_section :parse, required: false, multi: true, init: true, param_name: :parser_configs do
       config_argument :usage, :string, default: 'in_tail_parser'
@@ -250,7 +257,14 @@ def start
 
       if @pos_file
         pos_file_dir = File.dirname(@pos_file)
-        FileUtils.mkdir_p(pos_file_dir, mode: @dir_perm) unless Dir.exist?(pos_file_dir)
+        unless Dir.exist?(pos_file_dir)
+          FileUtils.mkdir_p(pos_file_dir, mode: @pos_dir_perm ? @pos_dir_perm.to_i(8) : @dir_perm)
+          if @pos_dir_owner || @pos_dir_group
+            owner = @pos_dir_owner ? Etc.getpwnam(@pos_dir_owner).uid : Process.uid
+            group = @pos_dir_group ? Etc.getgrnam(@pos_dir_group).gid : Process.gid
+            File.chown(owner, group, pos_file_dir)
+          end
+        end
         @pf_file = File.open(@pos_file, File::RDWR|File::CREAT|File::BINARY, @file_perm)
         @pf_file.sync = true
         @pf = PositionFile.load(@pf_file, @follow_inodes, expand_paths, logger: log)
@@ -913,35 +927,18 @@ def on_notify
       end
 
       def on_rotate(stat)
+        if stat.nil?
+          inode = nil
+          fsize = 0
+        else
+          inode = stat.ino
+          fsize = stat.size
+        end
+
         if @io_handler.nil?
           if stat
             # first time
-            fsize = stat.size
-            inode = stat.ino
-
-            last_inode = @pe.read_inode
-            if inode == last_inode
-              # rotated file has the same inode number with the last file.
-              # assuming following situation:
-              #   a) file was once renamed and backed, or
-              #   b) symlink or hardlink to the same file is recreated
-              # 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
-            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)
-            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)
-            end
+            @pe.update(inode, 0)
             @io_handler = io_handler
           else
             @io_handler = NullIOHandler.new
@@ -950,8 +947,7 @@ def on_rotate(stat)
           watcher_needs_update = false
 
           if stat
-            inode = stat.ino
-            if inode == @pe.read_inode # truncated
+            if @pe.read_inode == inode # truncated
               @pe.update_pos(0)
               @io_handler.close
             elsif !@io_handler.opened? # There is no previous file. Reuse TailWatcher

test/plugin/test_in_tail.rb

@@ -3402,4 +3402,104 @@ def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait
                     d.logs[-2..]
                   ])
   end
+
+  sub_test_case "directory permissions and ownership" do
+    setup do
+      @dir = File.expand_path("#{@tmp_dir}/tail#{rand(0..10000)}")
+      @pos_dir = File.expand_path("#{@tmp_dir}/pos#{rand(0..10000)}")
+      FileUtils.mkdir_p(@dir)
+    end
+
+    teardown do
+      FileUtils.rm_rf(@dir) if File.exist?(@dir)
+      FileUtils.rm_rf(@pos_dir) if File.exist?(@pos_dir)
+    end
+
+    test 'creates pos_file directory with default permissions when not specified' do
+      pos_file = "#{@pos_dir}/tail.pos"
+      config = config_element("ROOT", "", {
+                                "path" => "#{@dir}/tail.txt",
+                                "pos_file" => pos_file,
+                                "tag" => "t1",
+                                "format" => "none",
+                                "read_from_head" => "true"
+                              })
+      d = create_driver(config)
+      d.run(expect_emits: 1) do
+        File.open("#{@dir}/tail.txt", "wb") {|f| f.puts "test default permissions" }
+      end
+
+      assert_equal(true, File.directory?(@pos_dir))
+      assert_equal(0755, File.stat(@pos_dir).mode & 0777)
+    end
+
+    test 'creates pos_file directory with custom permissions' do
+      pos_file = "#{@pos_dir}/tail.pos"
+      config = config_element("ROOT", "", {
+                                "path" => "#{@dir}/tail.txt",
+                                "pos_file" => pos_file,
+                                "tag" => "t1",
+                                "format" => "none",
+                                "read_from_head" => "true",
+                                "pos_dir_perm" => "0770"
+                              })
+      d = create_driver(config)
+      d.run(expect_emits: 1) do
+        File.open("#{@dir}/tail.txt", "wb") {|f| f.puts "test custom permissions" }
+      end
+
+      assert_equal(true, File.directory?(@pos_dir))
+      assert_equal(0770, File.stat(@pos_dir).mode & 0777)
+    end
+
+    test 'creates pos_file directory with custom ownership' do
+      omit "Test requires root privileges" unless Process.uid == 0
+
+      pos_file = "#{@pos_dir}/tail.pos"
+      config = config_element("ROOT", "", {
+                                "path" => "#{@dir}/tail.txt",
+                                "pos_file" => pos_file,
+                                "tag" => "t1",
+                                "format" => "none",
+                                "read_from_head" => "true",
+                                "pos_dir_owner" => Process.uid.to_s,
+                                "pos_dir_group" => Process.gid.to_s
+                              })
+      d = create_driver(config)
+      d.run(expect_emits: 1) do
+        File.open("#{@dir}/tail.txt", "wb") {|f| f.puts "test custom ownership" }
+      end
+
+      assert_equal(true, File.directory?(@pos_dir))
+      stat = File.stat(@pos_dir)
+      assert_equal(Process.uid, stat.uid)
+      assert_equal(Process.gid, stat.gid)
+    end
+
+    test 'creates pos_file directory with both custom permissions and ownership' do
+      omit "Test requires root privileges" unless Process.uid == 0
+
+      pos_file = "#{@pos_dir}/tail.pos"
+      config = config_element("ROOT", "", {
+                                "path" => "#{@dir}/tail.txt",
+                                "pos_file" => pos_file,
+                                "tag" => "t1",
+                                "format" => "none",
+                                "read_from_head" => "true",
+                                "pos_dir_perm" => "0770",
+                                "pos_dir_owner" => Process.uid.to_s,
+                                "pos_dir_group" => Process.gid.to_s
+                              })
+      d = create_driver(config)
+      d.run(expect_emits: 1) do
+        File.open("#{@dir}/tail.txt", "wb") {|f| f.puts "test custom permissions and ownership" }
+      end
+
+      assert_equal(true, File.directory?(@pos_dir))
+      stat = File.stat(@pos_dir)
+      assert_equal(0770, stat.mode & 0777)
+      assert_equal(Process.uid, stat.uid)
+      assert_equal(Process.gid, stat.gid)
+    end
+  end
 end