# # Fluentd # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # require 'fluent/configurable' require 'fluent/plugin' require 'fluent/output' module Fluent # # Agent is a resource unit who manages emittable plugins # # Next step: `fluentd/root_agent.rb` # Next step: `fluentd/label.rb` # class Agent include Configurable def initialize(log:) super() @context = nil @outputs = [] @filters = [] @lifecycle_control_list = nil # lifecycle_control_list is the list of plugins in this agent, and ordered # from plugins which DOES emit, then DOESN'T emit # (input -> output w/ router -> filter -> output w/o router) # for start: use this order DESC # (because plugins which appears later in configurations will receive events from plugins which appears ealier) # for stop/before_shutdown/shutdown/after_shutdown/close/terminate: use this order ASC @lifecycle_cache = nil @log = log @event_router = EventRouter.new(NoMatchMatch.new(log), self) @error_collector = nil end attr_reader :log attr_reader :outputs attr_reader :filters attr_reader :context attr_reader :event_router attr_reader :error_collector def configure(conf) super # initialize and elements conf.elements('filter', 'match').each { |e| pattern = e.arg.empty? ? '**' : e.arg type = e['@type'] raise ConfigError, "Missing '@type' parameter on <#{e.name}> directive" unless type if e.name == 'filter' add_filter(type, pattern, e) else add_match(type, pattern, e) end } end def lifecycle_control_list return @lifecycle_control_list if @lifecycle_control_list lifecycle_control_list = { input: [], output_with_router: [], filter: [], output: [], } if self.respond_to?(:inputs) inputs.each do |i| lifecycle_control_list[:input] << i end end recursive_output_traverse = ->(o) { if o.has_router? lifecycle_control_list[:output_with_router] << o else lifecycle_control_list[:output] << o end if o.respond_to?(:outputs) o.outputs.each do |store| recursive_output_traverse.call(store) end end } outputs.each do |o| recursive_output_traverse.call(o) end filters.each do |f| lifecycle_control_list[:filter] << f end @lifecycle_control_list = lifecycle_control_list end def lifecycle(desc: false) kind_list = if desc [:output, :filter, :output_with_router] else [:output_with_router, :filter, :output] end kind_list.each do |kind| list = if desc lifecycle_control_list[kind].reverse else lifecycle_control_list[kind] end display_kind = (kind == :output_with_router ? :output : kind) list.each do |instance| yield instance, display_kind end end end def add_match(type, pattern, conf) log.info "adding match#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type output = Plugin.new_output(type) output.router = @event_router if output.respond_to?(:router=) output.configure(conf) @outputs << output if output.respond_to?(:outputs) && (output.respond_to?(:multi_output?) && output.multi_output? || output.is_a?(Fluent::MultiOutput)) # TODO: ruby 2.3 or later: replace `output.respond_to?(:multi_output?) && output.multi_output?` with output&.multi_output? @outputs.push(*output.outputs) end @event_router.add_rule(pattern, output) output end def add_filter(type, pattern, conf) log.info "adding filter#{@context.nil? ? '' : " in #{@context}"}", pattern: pattern, type: type filter = Plugin.new_filter(type) filter.router = @event_router filter.configure(conf) @filters << filter @event_router.add_rule(pattern, filter) filter end # For handling invalid record def emit_error_event(tag, time, record, error) end def handle_emits_error(tag, es, error) end class NoMatchMatch def initialize(log) @log = log @count = 0 end def emit_events(tag, es) # TODO use time instead of num of records c = (@count += 1) if c < 512 if Math.log(c) / Math.log(2) % 1.0 == 0 @log.warn "no patterns matched", tag: tag return end else if c % 512 == 0 @log.warn "no patterns matched", tag: tag return end end @log.on_trace { @log.trace "no patterns matched", tag: tag } end def start end def shutdown end end end end