Specify the dir mode explicitly

lib/fluent/plugin/in_tail.rb

@@ -49,6 +49,7 @@ def to_s
     end
 
     FILE_PERMISSION = 0644
+    DIR_PERMISSION = 0755
 
     def initialize
       super
@@ -168,6 +169,7 @@ def configure(conf)
                            method(:parse_singleline)
                          end
       @file_perm = system_config.file_permission || FILE_PERMISSION
+      @dir_perm = system_config.dir_permission || DIR_PERMISSION
       # parser is already created by parser helper
       @parser = parser_create(usage: parser_config['usage'] || @parser_configs.first.usage)
     end
@@ -210,7 +212,7 @@ def start
 
       if @pos_file
         pos_file_dir = File.dirname(@pos_file)
-        FileUtils.mkdir_p(pos_file_dir) unless Dir.exist?(pos_file_dir)
+        FileUtils.mkdir_p(pos_file_dir, mode: @dir_perm) unless Dir.exist?(pos_file_dir)
         @pf_file = File.open(@pos_file, File::RDWR|File::CREAT|File::BINARY, @file_perm)
         @pf_file.sync = true
         @pf = PositionFile.load(@pf_file, logger: log)

lib/fluent/plugin/output.rb

@@ -1203,7 +1203,7 @@ def backup_chunk(chunk, using_secondary, delayed_commit)
           backup_dir = File.dirname(backup_file)
 
           log.warn "bad chunk is moved to #{backup_file}"
-          FileUtils.mkdir_p(backup_dir) unless Dir.exist?(backup_dir)
+          FileUtils.mkdir_p(backup_dir, mode: system_config.dir_permission || 0755) unless Dir.exist?(backup_dir)
           File.open(backup_file, 'ab', system_config.file_permission || 0644) { |f|
             chunk.write_to(f)
           }

lib/fluent/plugin_id.rb

@@ -76,7 +76,7 @@ def plugin_root_dir
 
       # Fluent::Plugin::Base#fluentd_worker_id
       dir = File.join(system_config.root_dir, "worker#{fluentd_worker_id}", plugin_id)
-      FileUtils.mkdir_p(dir) unless Dir.exist?(dir)
+      FileUtils.mkdir_p(dir, mode: system_config.dir_permission || 0755) unless Dir.exist?(dir)
       @_plugin_root_dir = dir.freeze
       dir
     end

lib/fluent/supervisor.rb

@@ -439,9 +439,13 @@ def reopen!
         self
       end
 
-      def apply_options(format: nil, time_format: nil)
+      def apply_options(format: nil, time_format: nil, log_dir_perm: nil)
         $log.format = format if format
         $log.time_format = time_format if time_format
+
+        if @path && log_dir_perm
+          File.chmod(log_dir_perm || 0755, File.dirname(@path))
+        end
       end
 
       def level=(level)
@@ -529,7 +533,7 @@ def run_supervisor(dry_run: false)
           end
         else
           begin
-            FileUtils.mkdir_p(root_dir)
+            FileUtils.mkdir_p(root_dir, mode: @system_config.dir_permission || 0755)
           rescue => e
             raise Fluent::InvalidRootDirectory, "failed to create root directory:#{root_dir}, #{e.inspect}"
           end
@@ -620,7 +624,11 @@ def configure(supervisor: false)
       @system_config = build_system_config(@conf)
 
       @log.level = @system_config.log_level
-      @log.apply_options(format: @system_config.log.format, time_format: @system_config.log.time_format)
+      @log.apply_options(
+        format: @system_config.log.format,
+        time_format: @system_config.log.time_format,
+        log_dir_perm: @system_config.dir_permission,
+      )
 
       $log.info :supervisor, 'parsing config file is succeeded', path: @config_path
 

test/command/test_fluentd.rb

@@ -694,6 +694,23 @@ def write(chunk)
       )
     end
 
+    test 'success to start a worker2 with worker specific configuration' do
+      conf = <<CONF
+<system>
+  root_dir #{@root_path}
+  dir_permission 0744
+</system>
+CONF
+      conf_path = create_conf_file('worker_section0.conf', conf)
+
+      FileUtils.rm_rf(@root_path) rescue nil
+
+      assert_path_not_exist(@root_path)
+      assert_log_matches(create_cmdline(conf_path), 'spawn command to main') # any message is ok
+      assert_path_exist(@root_path)
+      assert_equal '744', File.stat(@root_path).mode.to_s(8)[-3, 3]
+    end
+
     test 'success to start a worker with worker specific configuration' do
       conf = <<CONF
 <system>

test/plugin/test_in_tail.rb

@@ -1074,14 +1074,37 @@ def test_pos_file_dir_creation
       "read_from_head" => true,
       "refresh_interval" => 1
     })
+
+    assert_path_not_exist("#{TMP_DIR}/pos")
     d = create_driver(config, false)
-    d.run(expect_emits: 1, shutdown: false) do
-      File.open("#{TMP_DIR}/tail.txt", "ab") { |f| f.puts "test3\n" }
-    end
-    assert_path_exist("#{TMP_DIR}/pos/tail.pos")
+    d.run
+    assert_path_exist("#{TMP_DIR}/pos")
+    assert_equal '755', File.stat("#{TMP_DIR}/pos").mode.to_s(8)[-3, 3]
+  ensure
     cleanup_directory(TMP_DIR)
+  end
 
-    d.instance_shutdown
+  def test_pos_file_dir_creation_with_system_dir_permission
+    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
+    })
+
+    assert_path_not_exist("#{TMP_DIR}/pos")
+
+    Fluent::SystemConfig.overwrite_system_config({ "dir_permission" => "744" }) do
+      d = create_driver(config, false)
+      d.run
+    end
+
+    assert_path_exist("#{TMP_DIR}/pos")
+    assert_equal '744', File.stat("#{TMP_DIR}/pos").mode.to_s(8)[-3, 3]
+  ensure
+    cleanup_directory(TMP_DIR)
   end
 
   def test_z_refresh_watchers

test/plugin/test_output_as_buffered_backup.rb

@@ -232,6 +232,52 @@ def wait_flush(target_file)
       end
     end
 
+    test 'create directory' do
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => TMP_DIR) do
+        id = 'backup_test_with_same_secondary'
+        hash = { 'flush_interval' => 1, 'flush_thread_burst_interval' => 0.1 }
+        chunk_id = nil
+        secconf = config_element('secondary', '', { '@type' => 'backup_output' })
+        @i.configure(config_element('ROOT', '', { '@id' => id }, [config_element('buffer', 'tag', hash), secconf]))
+        @i.register(:write) { |chunk|
+          chunk_id = chunk.unique_id;
+          raise Fluent::UnrecoverableError, "yay, your #write must fail"
+        }
+
+        flush_chunks
+
+        target = "#{TMP_DIR}/backup/worker0/#{id}/#{@i.dump_unique_id_hex(chunk_id)}.log"
+        target_dir = File.dirname(target)
+        wait_flush(target)
+
+        assert_path_exist(target_dir)
+        assert_equal '755', File.stat(target_dir).mode.to_s(8)[-3, 3]
+      end
+    end
+
+    test 'create directory with specific mode' do
+      Fluent::SystemConfig.overwrite_system_config('root_dir' => TMP_DIR, 'dir_permission' => '744') do
+        id = 'backup_test_with_same_secondary'
+        hash = { 'flush_interval' => 1, 'flush_thread_burst_interval' => 0.1 }
+        chunk_id = nil
+        secconf = config_element('secondary', '', { '@type' => 'backup_output' })
+        @i.configure(config_element('ROOT', '', { '@id' => id }, [config_element('buffer', 'tag', hash), secconf]))
+        @i.register(:write) { |chunk|
+          chunk_id = chunk.unique_id;
+          raise Fluent::UnrecoverableError, "yay, your #write must fail"
+        }
+
+        flush_chunks
+
+        target = "#{TMP_DIR}/backup/worker0/#{id}/#{@i.dump_unique_id_hex(chunk_id)}.log"
+        target_dir = File.dirname(target)
+        wait_flush(target)
+
+        assert_path_exist(target_dir)
+        assert_equal '744', File.stat(target_dir).mode.to_s(8)[-3, 3]
+      end
+    end
+
     test 'backup chunk with different type secondary' do
       Fluent::SystemConfig.overwrite_system_config('root_dir' => TMP_DIR) do
         id = 'backup_test_with_diff_secondary'

test/test_logger_initializer.rb

@@ -17,10 +17,28 @@ class LoggerInitializerTest < ::Test::Unit::TestCase
 
     assert_false File.exist?(TMP_DIR)
     logger = Fluent::Supervisor::LoggerInitializer.new(path, Fluent::Log::LEVEL_DEBUG, nil, nil, {})
+    mock.proxy(File).chmod(0o777, TMP_DIR).never
 
     assert_nothing_raised do
       logger.init(:supervisor, 0)
     end
+
+    assert_true File.exist?(TMP_DIR)
+  end
+
+  test 'apply_options with log_dir_perm' do
+    path = File.join(TMP_DIR, 'fluent_with_path.log')
+
+    assert_false File.exist?(TMP_DIR)
+    logger = Fluent::Supervisor::LoggerInitializer.new(path, Fluent::Log::LEVEL_DEBUG, nil, nil, {})
+    mock.proxy(File).chmod(0o777, TMP_DIR).once
+
+    assert_nothing_raised do
+      logger.init(:supervisor, 0)
+    end
+
+    logger.apply_options(log_dir_perm: 0o777)
     assert_true File.exist?(TMP_DIR)
+    assert_equal 0o777, (File.stat(TMP_DIR).mode & 0xFFF)
   end
 end

test/test_plugin_id.rb

@@ -97,5 +97,21 @@ class MyPlugin < Fluent::Plugin::Base
         ENV['SERVERENGINE_WORKER_ID'] = prev_env_val
       end
     end
+
+    test '#plugin_root_dir create dirctory with specify mode if not exists ' do
+      root_dir = Fluent::SystemConfig.overwrite_system_config({ 'root_dir' => File.join(TMP_DIR, "myroot"), 'dir_permission' => '0777' }) do
+        @p.plugin_root_dir
+      end
+
+      assert_equal '777', File.stat(root_dir).mode.to_s(8)[-3, 3]
+    end
+
+    test '#plugin_root_dir create dirctory with default permission if not exists ' do
+      root_dir = Fluent::SystemConfig.overwrite_system_config({ 'root_dir' => File.join(TMP_DIR, "myroot") }) do
+        @p.plugin_root_dir
+      end
+
+      assert_equal '755', File.stat(root_dir).mode.to_s(8)[-3, 3]
+    end
   end
 end