config: fix duplicate config file loading in config_include_dir (#5228)

**Which issue(s) this PR fixes**: 
Fixes #

Ref. #5100

**What this PR does / why we need it**: 
Currently, if a user explicitly `@include`s a file that is also located
in `config_include_dir`, the file is loaded twice.
This causes startup failures due to port conflicts or ID duplication.

This commit introduces a deduplication mechanism using a callback:

1. Add `on_file_parsed` callback to `Fluent::Config.build` to track
files loaded during the user config parsing phase.
2. In `Supervisor`, record loaded files via this callback.
3. When expanding `config_include_dir`, skip files that have already
been loaded in the user configuration.

**Docs Changes**:

**Release Note**:

---------

Signed-off-by: Shizuo Fujita <[email protected]>

Author: Watson1978

lib/fluent/config.rb

@@ -26,7 +26,7 @@ module Config
     # @param additional_config [String] config which is added to last of config body
     # @param use_v1_config [Bool] config is formatted with v1 or not
     # @return [Fluent::Config]
-    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true, type: nil)
+    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true, type: nil, on_file_parsed: nil)
       if type == :guess
         config_file_ext = File.extname(config_path)
         if config_file_ext == '.yaml' || config_file_ext == '.yml'
@@ -35,7 +35,7 @@ def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_c
       end
 
       if type == :yaml || type == :yml
-        return Fluent::Config::YamlParser.parse(config_path)
+        return Fluent::Config::YamlParser.parse(config_path, on_file_parsed: on_file_parsed)
       end
 
       config_fname = File.basename(config_path)
@@ -49,10 +49,10 @@ def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_c
         s
       end
 
-      Fluent::Config.parse(config_data, config_fname, config_basedir, use_v1_config)
+      Fluent::Config.parse(config_data, config_fname, config_basedir, use_v1_config, on_file_parsed: on_file_parsed)
     end
 
-    def self.parse(str, fname, basepath = Dir.pwd, v1_config = nil, syntax: :v1)
+    def self.parse(str, fname, basepath = Dir.pwd, v1_config = nil, syntax: :v1, on_file_parsed: nil)
       parser = if fname =~ /\.rb$/ || syntax == :ruby
                  :ruby
                elsif v1_config.nil?
@@ -68,7 +68,7 @@ def self.parse(str, fname, basepath = Dir.pwd, v1_config = nil, syntax: :v1)
       case parser
       when :v1
         require 'fluent/config/v1_parser'
-        V1Parser.parse(str, fname, basepath, Kernel.binding)
+        V1Parser.parse(str, fname, basepath, Kernel.binding, on_file_parsed: on_file_parsed)
       when :v0
         # TODO: show deprecated message in v1
         require 'fluent/config/parser'

lib/fluent/config/v1_parser.rb

@@ -27,17 +27,18 @@ module Config
     class V1Parser < LiteralParser
       ELEMENT_NAME = /[a-zA-Z0-9_]+/
 
-      def self.parse(data, fname, basepath = Dir.pwd, eval_context = nil)
+      def self.parse(data, fname, basepath = Dir.pwd, eval_context = nil, on_file_parsed: nil)
         ss = StringScanner.new(data)
-        ps = V1Parser.new(ss, basepath, fname, eval_context)
+        ps = V1Parser.new(ss, basepath, fname, eval_context, on_file_parsed: on_file_parsed)
         ps.parse!
       end
 
-      def initialize(strscan, include_basepath, fname, eval_context)
+      def initialize(strscan, include_basepath, fname, eval_context, on_file_parsed: nil)
         super(strscan, eval_context)
         @include_basepath = include_basepath
         @fname = fname
         @logger = defined?($log) ? $log : nil
+        @on_file_parsed = on_file_parsed
       end
 
       def parse!
@@ -138,6 +139,8 @@ def parse_element(root_element, elem_name, attrs = {}, elems = [])
           end
         end
 
+        @on_file_parsed&.call(File.expand_path(File.join(@include_basepath, @fname))) if root_element
+
         return attrs, elems
       end
 
@@ -166,7 +169,7 @@ def eval_include(attrs, elems, uri)
             data = File.read(entry)
             data.force_encoding('UTF-8')
             ss = StringScanner.new(data)
-            V1Parser.new(ss, basepath, fname, @eval_context).parse_element(true, nil, attrs, elems)
+            V1Parser.new(ss, basepath, fname, @eval_context, on_file_parsed: @on_file_parsed).parse_element(true, nil, attrs, elems)
           }
         else
           require 'open-uri'
@@ -175,7 +178,7 @@ def eval_include(attrs, elems, uri)
           data = u.open { |f| f.read }
           data.force_encoding('UTF-8')
           ss = StringScanner.new(data)
-          V1Parser.new(ss, basepath, fname, @eval_context).parse_element(true, nil, attrs, elems)
+          V1Parser.new(ss, basepath, fname, @eval_context, on_file_parsed: @on_file_parsed).parse_element(true, nil, attrs, elems)
         end
       rescue SystemCallError => e
         cpe = ConfigParseError.new("include error #{uri} - #{e}")

lib/fluent/config/yaml_parser.rb

@@ -21,7 +21,7 @@
 module Fluent
   module Config
     module YamlParser
-      def self.parse(path)
+      def self.parse(path, on_file_parsed: nil)
         context = Kernel.binding
 
         unless context.respond_to?(:use_nil)
@@ -48,7 +48,7 @@ def self.parse(path)
           end
         end
 
-        s = Fluent::Config::YamlParser::Loader.new(context).load(Pathname.new(path))
+        s = Fluent::Config::YamlParser::Loader.new(context, on_file_parsed: on_file_parsed).load(Pathname.new(path))
         Fluent::Config::YamlParser::Parser.new(s).build.to_element
       end
     end

lib/fluent/config/yaml_parser/loader.rb

@@ -30,9 +30,10 @@ class Loader
         FLUENT_STR_TAG = 'tag:fluent/s'.freeze
         SHOVEL = '<<'.freeze
 
-        def initialize(context = Kernel.binding)
+        def initialize(context = Kernel.binding, on_file_parsed: nil)
           @context = context
           @current_path = nil
+          @on_file_parsed = on_file_parsed
         end
 
         # @param [String] path
@@ -55,9 +56,13 @@ def load(path)
             Fluent::Config::YamlParser::FluentValue::StringValue.new(val, @context)
           end
 
-          path.open do |f|
+          config = path.open do |f|
             visitor.accept(Psych.parse(f))
           end
+
+          @on_file_parsed&.call(File.expand_path(path.to_s))
+
+          config
         end
 
         def eval_include(path, parent)

lib/fluent/supervisor.rb

@@ -18,6 +18,7 @@
 require 'open3'
 require 'pathname'
 require 'find'
+require 'set'
 
 require 'fluent/config'
 require 'fluent/counter'
@@ -796,18 +797,20 @@ def configure(supervisor: false)
         $log.warn('the value "-" for `inline_config` is deprecated. See https://github.com/fluent/fluentd/issues/2711')
         @inline_config = STDIN.read
       end
+      parsed_files = Set.new
       @conf = Fluent::Config.build(
         config_path: @config_path,
         encoding: @conf_encoding,
         additional_config: @inline_config,
         use_v1_config: @use_v1_config,
         type: @config_file_type,
+        on_file_parsed: ->(path) { parsed_files << path },
       )
       @system_config = build_system_config(@conf)
 
       $log.info :supervisor, 'parsing config file is succeeded', path: @config_path
 
-      build_additional_configurations do |additional_conf|
+      build_additional_configurations(parsed_files) do |additional_conf|
         @conf += additional_conf
       end
 
@@ -854,6 +857,7 @@ def setup_global_logger(supervisor: false)
         additional_config: @inline_config,
         use_v1_config: @use_v1_config,
         type: @config_file_type,
+        on_file_parsed: nil,
       )
       system_config = build_system_config(conf)
 
@@ -1088,15 +1092,17 @@ def reload_config
         $log.debug('worker got SIGUSR2')
 
         begin
+          parsed_files = Set.new
           conf = Fluent::Config.build(
             config_path: @config_path,
             encoding: @conf_encoding,
             additional_config: @inline_config,
             use_v1_config: @use_v1_config,
             type: @config_file_type,
+            on_file_parsed: ->(path) { parsed_files << path },
           )
 
-          build_additional_configurations do |additional_conf|
+          build_additional_configurations(parsed_files) do |additional_conf|
             conf += additional_conf
           end
 
@@ -1206,7 +1212,7 @@ def build_system_config(conf)
       system_config
     end
 
-    def build_additional_configurations
+    def build_additional_configurations(parsed_files)
       if @system_config.config_include_dir&.empty?
         $log.info :supervisor, 'configuration include directory is disabled'
         return
@@ -1218,11 +1224,17 @@ def build_additional_configurations
           next unless supported_suffixes.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.
+          if parsed_files.include?(path)
+            $log.info :supervisor, 'skip auto loading, it was already loaded', path: path
+            next
+          end
+
           $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)
+                                     type: :guess,
+                                     on_file_parsed: nil)
         end
       rescue Errno::ENOENT
         $log.info :supervisor, 'inaccessible include directory was specified', path: @system_config.config_include_dir

test/test_config.rb

@@ -520,5 +520,118 @@ def write_config(path, data, encoding: 'utf-8')
       assert_equal('value', c['key'])
       assert_equal('value2', c['key2'])
     end
+
+    sub_test_case 'on_file_parsed' do
+      test "calling order on normal configuration files" do
+        write_config("#{TMP_DIR}/build/common_param.conf", <<~EOS)
+          flush_interval 5s
+          total_limit_size 100m
+          chunk_limit_size 1m
+        EOS
+        write_config("#{TMP_DIR}/build/server.conf", <<~EOS)
+          <server>
+            host 127.0.0.1
+            port 24224
+          </server>
+        EOS
+        write_config("#{TMP_DIR}/build/forward.conf", <<~EOS)
+          <match test.*>
+            @type forward
+
+            @include server.conf
+          </match>
+        EOS
+        write_config("#{TMP_DIR}/build/inline.conf", <<~EOS)
+          <source>
+            @type stdout
+            tag test
+          </source>
+        EOS
+        write_config("#{TMP_DIR}/build/fluent.conf", <<~EOS)
+          <match sample.*>
+            @type file
+            <buffer>
+              @include common_param.conf
+            </buffer>
+          </match>
+          <match debug.*>
+            @type stdout
+            <buffer>
+              @include common_param.conf
+            </buffer>
+          </match>
+
+          @include forward.conf
+        EOS
+
+        # parsed_files contains file paths in the order of file parsed
+        parsed_files = []
+        Fluent::Config.build(
+          config_path: "#{TMP_DIR}/build/fluent.conf",
+          additional_config: "@include inline.conf",
+          on_file_parsed: ->(path) { parsed_files << path },
+        )
+
+        assert_equal(
+          [
+            "#{TMP_DIR}/build/common_param.conf",
+            "#{TMP_DIR}/build/common_param.conf",
+            "#{TMP_DIR}/build/server.conf",
+            "#{TMP_DIR}/build/forward.conf",
+            "#{TMP_DIR}/build/inline.conf",
+            "#{TMP_DIR}/build/fluent.conf"
+          ],
+          parsed_files
+        )
+      end
+
+      test "calling order on YAML configuration files" do
+        write_config("#{TMP_DIR}/build/common_buffer.yaml", <<~EOS)
+          - buffer:
+              flush_interval: 5s
+              total_limit_size: 100m
+              chunk_limit_size: 1m
+        EOS
+        write_config("#{TMP_DIR}/build/forward.yaml", <<~EOS)
+          - match:
+              $tag: test.*
+              server:
+                host: 127.0.0.1
+                port: 24224
+        EOS
+        write_config("#{TMP_DIR}/build/fluent.yaml", <<~EOS)
+          config:
+            - match:
+                $tag: sample.*
+                $type: file
+                <<: !include common_buffer.yaml
+            - match:
+                $tag: debug.*
+                $type: stdout
+                <<: !include common_buffer.yaml
+            - !include forward.yaml
+        EOS
+
+        # parsed_files contains file paths in the order of file parsed
+        # `additional_config` does not support YAML config
+        parsed_files = []
+        Fluent::Config.build(
+          config_path: "#{TMP_DIR}/build/fluent.yaml",
+          type: :yaml,
+          on_file_parsed: ->(path) { parsed_files << path },
+          )
+
+        assert_equal(
+          [
+            "#{TMP_DIR}/build/common_buffer.yaml",
+            "#{TMP_DIR}/build/common_buffer.yaml",
+            "#{TMP_DIR}/build/forward.yaml",
+            "#{TMP_DIR}/build/fluent.yaml"
+          ],
+          parsed_files
+        )
+      end
+
+    end
   end
 end