EventRouter is responsible to route events to a collector.
It has a list of MatchPattern and Collector pairs:
+----------------+ +-----------------+ | MatchPattern | | Collector | +----------------+ +-----------------+ | access.** ---------> type forward | | logs.** ---------> type copy | | archive.** ---------> type s3 | +----------------+ +-----------------+
EventRouter does:
1) receive an event at `#emit` methods 2) match the event's tag with the MatchPatterns 3) forward the event to the corresponding Collector
Collector is either of Output, Filter or other EventRouter.
# File lib/fluent/event_router.rb, line 44 def initialize(default_collector, emit_error_handler) @match_rules = [] @match_cache = MatchCache.new @default_collector = default_collector @emit_error_handler = emit_error_handler end
called by Agent to add new match pattern and collector
# File lib/fluent/event_router.rb, line 75 def add_rule(pattern, collector) @match_rules << Rule.new(pattern, collector) end
# File lib/fluent/event_router.rb, line 79 def emit(tag, time, record) unless record.nil? emit_stream(tag, OneEventStream.new(time, record)) end end
# File lib/fluent/event_router.rb, line 85 def emit_array(tag, array) emit_stream(tag, ArrayEventStream.new(array)) end
# File lib/fluent/event_router.rb, line 95 def emit_error_event(tag, time, record, error) @emit_error_handler.emit_error_event(tag, time, record, error) end
# File lib/fluent/event_router.rb, line 89 def emit_stream(tag, es) match(tag).emit_events(tag, es) rescue => e @emit_error_handler.handle_emits_error(tag, es, e) end
# File lib/fluent/event_router.rb, line 103 def match(tag) collector = @match_cache.get(tag) { find(tag) || @default_collector } collector end
# File lib/fluent/event_router.rb, line 99 def match?(tag) !!find(tag) end
# File lib/fluent/event_router.rb, line 235 def find(tag) pipeline = nil @match_rules.each_with_index { |rule, i| if rule.match?(tag) if rule.collector.is_a?(Plugin::Filter) pipeline ||= Pipeline.new pipeline.add_filter(rule.collector) else if pipeline pipeline.set_output(rule.collector) else # Use Output directly when filter is not matched pipeline = rule.collector end return pipeline end end } if pipeline # filter is matched but no match pipeline.set_output(@default_collector) pipeline else nil end end