Class: Fluent::EventRouter

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/event_router.rb

Overview

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.

Defined Under Namespace

Classes: MatchCache, Pipeline, Rule

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(default_collector, emit_error_handler) ⇒ EventRouter

Returns a new instance of EventRouter.



45
46
47
48
49
50
51
52
 
# File 'lib/fluent/event_router.rb', line 45

def initialize(default_collector, emit_error_handler)
  @match_rules = []
  @match_cache = MatchCache.new
  @default_collector = default_collector
  @emit_error_handler = emit_error_handler
  @metric_callbacks = {}
  @caller_plugin_id = nil
end

Instance Attribute Details

#default_collector ⇒ Object

Returns the value of attribute default_collector.



54
55
56
 
# File 'lib/fluent/event_router.rb', line 54

def default_collector
  @default_collector
end

#emit_error_handler ⇒ Object

Returns the value of attribute emit_error_handler.



55
56
57
 
# File 'lib/fluent/event_router.rb', line 55

def emit_error_handler
  @emit_error_handler
end

Instance Method Details

#add_metric_callbacks(caller_plugin_id, callback) ⇒ Object 



88
89
90
 
# File 'lib/fluent/event_router.rb', line 88

def add_metric_callbacks(caller_plugin_id, callback)
  @metric_callbacks[caller_plugin_id] = callback
end

#add_rule(pattern, collector) ⇒ Object

called by Agent to add new match pattern and collector



84
85
86
 
# File 'lib/fluent/event_router.rb', line 84

def add_rule(pattern, collector)
  @match_rules << Rule.new(pattern, collector)
end

#caller_plugin_id=(caller_plugin_id) ⇒ Object 



92
93
94
 
# File 'lib/fluent/event_router.rb', line 92

def caller_plugin_id=(caller_plugin_id)
  @caller_plugin_id = caller_plugin_id
end

#emit(tag, time, record) ⇒ Object 



104
105
106
107
108
 
# File 'lib/fluent/event_router.rb', line 104

def emit(tag, time, record)
  unless record.nil?
    emit_stream(tag, OneEventStream.new(time, record))
  end
end

#emit_array(tag, array) ⇒ Object 



110
111
112
 
# File 'lib/fluent/event_router.rb', line 110

def emit_array(tag, array)
  emit_stream(tag, ArrayEventStream.new(array))
end

#emit_error_event(tag, time, record, error) ⇒ Object 



125
126
127
 
# File 'lib/fluent/event_router.rb', line 125

def emit_error_event(tag, time, record, error)
  @emit_error_handler.emit_error_event(tag, time, record, error)
end

#emit_stream(tag, es) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
 
# File 'lib/fluent/event_router.rb', line 114

def emit_stream(tag, es)
  match(tag).emit_events(tag, es)
  if callback = find_callback
    callback.call(es)
  end
rescue Pipeline::OutputError => e
  @emit_error_handler.handle_emits_error(tag, e.processed_es, e.internal_error)
rescue => e
  @emit_error_handler.handle_emits_error(tag, es, e)
end

#find_callback ⇒ Object 



96
97
98
99
100
101
102
 
# File 'lib/fluent/event_router.rb', line 96

def find_callback
  if @caller_plugin_id
    @metric_callbacks[@caller_plugin_id]
  else
    nil
  end
end

#match(tag) ⇒ Object 



133
134
135
136
137
138
 
# File 'lib/fluent/event_router.rb', line 133

def match(tag)
  collector = @match_cache.get(tag) {
    find(tag) || @default_collector
  }
  collector
end

#match?(tag) ⇒ Boolean

Returns:

  • (Boolean) 


129
130
131
 
# File 'lib/fluent/event_router.rb', line 129

def match?(tag)
  !!find(tag)
end

#suppress_missing_match! ⇒ Object 



77
78
79
80
81
 
# File 'lib/fluent/event_router.rb', line 77

def suppress_missing_match!
  if @default_collector.respond_to?(:suppress_missing_match!)
    @default_collector.suppress_missing_match!
  end
end