config: fix duplicate config file loading in config_include_dir

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

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.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(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_supervisor.rb

@@ -1085,7 +1085,8 @@ def test_stop_parallel_old_supervisor_after_delay
       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]) }
+                                 type: anything,
+                                 on_file_parsed: anything) { config_element('ROOT', '', {}, [c]) }
       sources.each do |type|
         config = <<~EOF
         <source>
@@ -1111,7 +1112,8 @@ def test_stop_parallel_old_supervisor_after_delay
       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]) }
+                                 type: anything,
+                                 on_file_parsed: anything) { config_element('ROOT', '', {}, [c]) }
       sources.each do |type|
         config = <<~EOF
         config:
@@ -1139,7 +1141,8 @@ def test_stop_parallel_old_supervisor_after_delay
       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]) }
+                                 type: anything,
+                                 on_file_parsed: anything) { config_element('ROOT', '', {}, [c]) }
       sources.each do |type|
         if yaml
           config = <<~EOF