Class: Fluent::Plugin::MultiOutput

Inherits:
Base
  • Object
show all
Includes:
Fluent::PluginHelper::Mixin, Fluent::PluginId, Fluent::PluginLoggerMixin
Defined in:
lib/fluent/plugin/multi_output.rb

Direct Known Subclasses

RoundRobinOutput

Constant Summary

Constants included from Configurable

Configurable::CONFIG_TYPE_REGISTRY

Instance Attribute Summary collapse

Attributes included from Fluent::PluginLoggerMixin

#log

Instance Method Summary collapse

Methods included from Fluent::PluginHelper::Mixin

included

Methods included from Fluent::PluginLoggerMixin

included, #start, #terminate

Methods included from Fluent::PluginId

#plugin_id, #plugin_id_configured?, #plugin_id_for_test?

Methods inherited from Base

#after_shutdown, #after_shutdown?, #after_start, #after_started?, #before_shutdown, #before_shutdown?, #close, #closed?, #configured?, #has_router?, #inspect, #shutdown, #shutdown?, #start, #started?, #stop, #stopped?, #terminate, #terminated?

Methods included from SystemConfig::Mixin

#system_config, #system_config_override

Methods included from Configurable

#config, included, lookup_type, register_type

Constructor Details

#initialize ⇒ MultiOutput

Returns a new instance of MultiOutput.



41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/fluent/plugin/multi_output.rb', line 41

def initialize
  super
  @outputs = []

  @compat = false

  @counters_monitor = Monitor.new
  # TODO: well organized counters
  @num_errors = 0
  @emit_count = 0
  @emit_records = 0
  # @write_count = 0
  # @rollback_count = 0
end

Instance Attribute Details

#outputs ⇒ Object (readonly)

Returns the value of attribute outputs.



35
36
37
 
# File 'lib/fluent/plugin/multi_output.rb', line 35

def outputs
  @outputs
end

Instance Method Details

#configure(conf) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/fluent/plugin/multi_output.rb', line 56

def configure(conf)
  super

  @stores.each do |store|
    store_conf = store.corresponding_config_element
    type = store_conf['@type']
    unless type
      raise Fluent::ConfigError, "Missing '@type' parameter in <store> section"
    end

    log.debug "adding store", type: type

    output = Fluent::Plugin.new_output(type)
    if output.has_router?
      output.router = router
    end
    output.configure(store_conf)
    @outputs << output
  end
end

#emit_sync(tag, es) ⇒ Object Also known as: emit_events

Child plugin’s lifecycles are controlled by agent automatically. It calls ‘outputs` to traverse plugins, and invoke start/stop/*shutdown/close/terminate on these directly.

  • ‘start` of this plugin will be called after child plugins

  • ‘stop`, `*shutdown`, `close` and `terminate` of this plugin will be called before child plugins



82
83
84
85
86
87
88
89
90
91
 
# File 'lib/fluent/plugin/multi_output.rb', line 82

def emit_sync(tag, es)
  @counters_monitor.synchronize{ @emit_count += 1 }
  begin
    process(tag, es)
    @counters_monitor.synchronize{ @emit_records += es.size }
  rescue
    @counters_monitor.synchronize{ @num_errors += 1 }
    raise
  end
end

#process(tag, es) ⇒ Object

Raises:

  • (NotImplementedError) 


37
38
39
 
# File 'lib/fluent/plugin/multi_output.rb', line 37

def process(tag, es)
  raise NotImplementedError, "BUG: output plugins MUST implement this method"
end