system_config: support built-in config files

lib/fluent/env.rb

@@ -21,6 +21,7 @@
 
 module Fluent
   DEFAULT_CONFIG_PATH = ENV['FLUENT_CONF'] || '/etc/fluent/fluent.conf'
+  DEFAULT_CONFIG_INCLUDE_DIR = ENV["FLUENT_CONF_INCLUDE_DIR"] || '/etc/fluent/conf.d'
   DEFAULT_PLUGIN_DIR = ENV['FLUENT_PLUGIN'] || '/etc/fluent/plugin'
   DEFAULT_SOCKET_PATH = ENV['FLUENT_SOCKET'] || '/var/run/fluent/fluent.sock'
   DEFAULT_BACKUP_DIR = ENV['FLUENT_BACKUP_DIR'] || '/tmp/fluent'

lib/fluent/supervisor.rb

@@ -17,6 +17,7 @@
 require 'fileutils'
 require 'open3'
 require 'pathname'
+require 'find'
 
 require 'fluent/config'
 require 'fluent/counter'
@@ -806,6 +807,10 @@ def configure(supervisor: false)
 
       $log.info :supervisor, 'parsing config file is succeeded', path: @config_path
 
+      build_additional_configurations do |additional_conf|
+        @conf += additional_conf
+      end
+
       @libs.each do |lib|
         require lib
       end
@@ -1090,6 +1095,10 @@ def reload_config
             type: @config_file_type,
           )
 
+          build_additional_configurations do |additional_conf|
+            conf += additional_conf
+          end
+
           Fluent::VariableStore.try_to_reset do
             Fluent::Engine.reload_config(conf)
           end
@@ -1196,6 +1205,28 @@ def build_system_config(conf)
       system_config
     end
 
+    def build_additional_configurations
+      if @system_config.config_include_dir&.empty?
+        $log.info :supervisor, 'configuration include directory is disabled'
+        return
+      end
+      begin
+        Find.find(@system_config.config_include_dir) do |path|
+          next if File.directory?(path)
+          next unless [".conf", ".yaml", ".yml"].include?(File.extname(path))
+          # NOTE: both types of normal config (.conf) and YAML will be loaded.
+          # Thus, it does not care whether @config_path is .conf or .yml.
+          $log.info :supervisor, 'loading additional configuration file', path: path
+          yield Fluent::Config.build(config_path: path,
+                                     encoding: @conf_encoding,
+                                     use_v1_config: @use_v1_config,
+                                     type: :guess)
+        end
+      rescue Errno::ENOENT
+        $log.info :supervisor, 'inaccessible include directory was specified', path: @system_config.config_include_dir
+      end
+    end
+
     RUBY_ENCODING_OPTIONS_REGEX = %r{\A(-E|--encoding=|--internal-encoding=|--external-encoding=)}.freeze
 
     def build_spawn_command

lib/fluent/system_config.rb

@@ -16,6 +16,7 @@
 
 require 'fluent/configurable'
 require 'fluent/config/element'
+require 'fluent/env'
 
 module Fluent
   class SystemConfig
@@ -28,7 +29,8 @@ class SystemConfig
       :without_source, :with_source_only, :rpc_endpoint, :enable_get_dump, :process_name,
       :file_permission, :dir_permission, :counter_server, :counter_client,
       :strict_config_value, :enable_msgpack_time_support, :disable_shared_socket,
-      :metrics, :enable_input_metrics, :enable_size_metrics, :enable_jit, :source_only_buffer
+      :metrics, :enable_input_metrics, :enable_size_metrics, :enable_jit, :source_only_buffer,
+      :config_include_dir
     ]
 
     config_param :workers,   :integer, default: 1
@@ -58,6 +60,7 @@ class SystemConfig
     config_param :dir_permission, default: nil do |v|
       v.to_i(8)
     end
+    config_param :config_include_dir, default: Fluent::DEFAULT_CONFIG_INCLUDE_DIR
     config_section :log, required: false, init: true, multi: false do
       config_param :path, :string, default: nil
       config_param :format, :enum, list: [:text, :json], default: :text

test/command/test_fluentd.rb

@@ -1536,4 +1536,76 @@ def send_end(port)
       end_thread&.kill
     end
   end
+
+  def create_config_include_dir_configuration(config_path, config_dir, yaml_format = false)
+    if yaml_format
+      conf = <<CONF
+      system:
+        config_include_dir: "#{config_dir}"
+CONF
+    else
+      conf = <<CONF
+      <system>
+        config_include_dir #{config_dir}
+      </system>
+CONF
+    end
+    create_conf_file(config_path, conf)
+  end
+
+  sub_test_case "test additional configuration directory" do
+    setup do
+      FileUtils.mkdir_p(File.join(@tmp_dir, "conf.d"))
+    end
+
+    test "disable additional configuration directory" do
+      conf_path = create_config_include_dir_configuration("disabled_config_include_dir.conf", "")
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: configuration include directory is disabled")
+    end
+
+    test "inaccessible include directory error" do
+      conf_path = create_config_include_dir_configuration("inaccessible_include.conf", "/nonexistent")
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: inaccessible include directory was specified")
+    end
+
+    data("include additional configuration with relative conf.d" => {"relative_path" => true},
+         "include additional configuration with full-path conf.d" => {"relative_path" => false})
+    test "additional configuration file (conf.d/child.conf) was loaded" do |option|
+      conf_dir = option["relative_path"] ? "conf.d" : "#{@tmp_dir}/conf.d"
+      conf_path = create_config_include_dir_configuration("parent.conf", conf_dir)
+      create_conf_file('conf.d/child.conf', "")
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: loading additional configuration file path=\"#{conf_dir}/child.conf\"")
+    end
+  end
+
+  sub_test_case "test additional configuration directory (YAML)" do
+    setup do
+      FileUtils.mkdir_p(File.join(@tmp_dir, "conf.d"))
+    end
+
+    test "disable additional configuration directory" do
+      conf_path = create_config_include_dir_configuration("disabled_config_include_dir.yml", "", true)
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: configuration include directory is disabled")
+    end
+
+    test "inaccessible include directory error" do
+      conf_path = create_config_include_dir_configuration("inaccessible_include.yml", "/nonexistent", true)
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: inaccessible include directory was specified")
+    end
+
+    data("include additional YAML configuration with relative conf.d" => {"relative_path" => true},
+         "include additional YAML configuration with full path conf.d" => {"relative_path" => false})
+    test "additional relative configuration file (conf.d/child.yml) was loaded" do |option|
+      conf_dir = option["relative_path"] ? "conf.d" : "#{@tmp_dir}/conf.d"
+      conf_path = create_config_include_dir_configuration("parent.yml", conf_dir, true)
+      create_conf_file('conf.d/child.yml', "")
+      assert_log_matches(create_cmdline(conf_path),
+                         "[info]: loading additional configuration file path=\"#{conf_dir}/child.yml\"")
+    end
+  end
 end

test/test_supervisor.rb

@@ -1054,6 +1054,112 @@ def test_stop_parallel_old_supervisor_after_delay
     end
   end
 
+  sub_test_case "include additional configuration" do
+    setup do
+      @config_include_dir = File.join(@tmp_dir, "conf.d")
+      FileUtils.mkdir_p(@config_include_dir)
+    end
+
+    test "no additional configuration" do
+      c = Fluent::Config::Element.new('system', '', { 'config_include_dir' => '' }, [])
+      stub(Fluent::Config).build { config_element('ROOT', '', {}, [c]) }
+      supervisor = Fluent::Supervisor.new({})
+      supervisor.configure(supervisor: true)
+      assert_equal([c], supervisor.instance_variable_get(:@conf).elements)
+    end
+
+    data(
+      "single source" => ["forward"],
+      "multiple sources" => ["forward", "tcp"])
+    test "additional configuration" do |sources|
+      c = Fluent::Config::Element.new('system', '',
+                                      { 'config_include_dir' => @config_include_dir }, [])
+      config_path = "#{@config_include_dir}/dummy.conf"
+      stub.proxy(Fluent::Config).build
+      stub(Fluent::Config).build(config_path: "/etc/fluent/fluent.conf", encoding: "utf-8",
+                                 additional_config: anything, use_v1_config: anything,
+                                 type: anything) { config_element('ROOT', '', {}, [c]) }
+      sources.each do |type|
+        config = <<~EOF
+        <source>
+          @type #{type}
+        </source>
+        EOF
+        additional_config_path = "#{@config_include_dir}/#{type}.conf"
+        write_config(additional_config_path, config)
+      end
+      supervisor = Fluent::Supervisor.new({})
+      supervisor.configure(supervisor: true)
+      expected = [c].concat(sources.collect { |type| {"@type" => type} })
+      assert_equal(expected, supervisor.instance_variable_get(:@conf).elements)
+    end
+
+    data(
+      "single YAML source" => ["forward"],
+      "multiple YAML sources" => ["forward", "tcp"])
+    test "additional YAML configuration" do |sources|
+      c = Fluent::Config::Element.new('system', '',
+                                      { 'config_include_dir' => @config_include_dir }, [])
+      config_path = "#{@config_include_dir}/dummy.yml"
+      stub.proxy(Fluent::Config).build
+      stub(Fluent::Config).build(config_path: "/etc/fluent/fluent.conf", encoding: "utf-8",
+                                 additional_config: anything, use_v1_config: anything,
+                                 type: anything) { config_element('ROOT', '', {}, [c]) }
+      sources.each do |type|
+        config = <<~EOF
+        config:
+          - source:
+              $type: #{type}
+        EOF
+        additional_config_path = "#{@config_include_dir}/#{type}.yml"
+        write_config(additional_config_path, config)
+      end
+      supervisor = Fluent::Supervisor.new({})
+      supervisor.configure(supervisor: true)
+      expected = [c].concat(sources.collect { |type| {"@type" => type} })
+      assert_equal(expected, supervisor.instance_variable_get(:@conf).elements)
+    end
+
+    data(
+      "single source" => [false, ["forward"]],
+      "multiple sources" => [false, ["forward", "tcp"]],
+      "single YAML source" => [true, ["forward"]],
+      "multiple YAML sources" => [true, ["forward", "tcp"]])
+    test "reload with additional configuration" do |(yaml, sources)|
+      c = Fluent::Config::Element.new('system', '',
+                                      { 'config_include_dir' => @config_include_dir }, [])
+      config_path = "#{@config_include_dir}/dummy.yml"
+      stub.proxy(Fluent::Config).build
+      stub(Fluent::Config).build(config_path: "/etc/fluent/fluent.conf", encoding: "utf-8",
+                                 additional_config: anything, use_v1_config: anything,
+                                 type: anything) { config_element('ROOT', '', {}, [c]) }
+      sources.each do |type|
+        if yaml
+          config = <<~EOF
+          config:
+            - source:
+                $type: #{type}
+          EOF
+          additional_config_path = "#{@config_include_dir}/#{type}.yml"
+          write_config(additional_config_path, config)
+        else
+          config = <<~EOF
+          <source>
+            @type #{type}
+          </source>
+          EOF
+          additional_config_path = "#{@config_include_dir}/#{type}.conf"
+          write_config(additional_config_path, config)
+        end
+      end
+      supervisor = Fluent::Supervisor.new({})
+      supervisor.configure(supervisor: true)
+      supervisor.__send__(:reload_config)
+      expected = [c].concat(sources.collect { |type| {"@type" => type} })
+      assert_equal(expected, supervisor.instance_variable_get(:@conf).elements)
+    end
+  end
+
   def create_debug_dummy_logger
     dl_opts = {}
     dl_opts[:log_level] = ServerEngine::DaemonLogger::DEBUG