[WIP] v0.14 plugin APIs

.travis.yml

@@ -1,8 +1,8 @@
 language: ruby
 
 rvm:
-  - 2.0.0
   - 2.1
+  - 2.1.5
   - 2.2.3
   - 2.3.0
   - ruby-head

example/in_status.conf

@@ -0,0 +1,10 @@
+<source>
+  @type status
+  emit_interval 3
+  tag "test"
+</source>
+
+<match test>
+  @type stdout
+</match>
+

lib/fluent/command/debug.rb

@@ -85,7 +85,6 @@
 puts "Usage:"
 puts "    Engine.match('some.tag').output  : get an output plugin instance"
 puts "    Engine.sources[i]                : get input plugin instances"
-puts "    Plugin.load_plugin(type,name)    : load plugin class (use this if you get DRb::DRbUnknown)"
 puts ""
 
 Encoding.default_internal = nil if Encoding.respond_to?(:default_internal)

lib/fluent/command/fluentd.rb

@@ -92,6 +92,10 @@
   opts[:without_source] = b
 }
 
+op.on('--plugin-storage-path DIR_PATH', "directory path which is used for storages of plugin internal data") {|s|
+  opts[:plugin_storage_path] = s
+}
+
 op.on('--use-v1-config', "Use v1 configuration format (default)", TrueClass) {|b|
   opts[:use_v1_config] = b
 }

lib/fluent/config.rb

@@ -38,17 +38,4 @@ def self.new(name = '')
       Element.new(name, '', {}, [])
     end
   end
-
-  require 'fluent/configurable'
-
-  module PluginId
-    def configure(conf)
-      @id = conf['@id'] || conf['id']
-      super
-    end
-
-    def plugin_id
-      @id ? @id : "object:#{object_id.to_s(16)}"
-    end
-  end
 end

lib/fluent/config/configure_proxy.rb

@@ -77,19 +77,7 @@ def merge(other) # self is base class, other is subclass
         merged.argument = other.argument || self.argument
         merged.params = self.params.merge(other.params)
         merged.defaults = self.defaults.merge(other.defaults)
-        merged.sections = {}
-        (self.sections.keys + other.sections.keys).uniq.each do |section_key|
-          self_section = self.sections[section_key]
-          other_section = other.sections[section_key]
-          merged_section = if self_section && other_section
-                             self_section.merge(other_section)
-                           elsif self_section || other_section
-                             self_section || other_section
-                           else
-                             raise "BUG: both of self and other section are nil"
-                           end
-          merged.sections[section_key] = merged_section
-        end
+        merged.sections = self.sections.merge(other.sections)
 
         merged
       end

lib/fluent/config/section.rb

@@ -33,6 +33,10 @@ def initialize(params = {})
 
       alias :object_id :__id__
 
+      def to_s
+        inspect
+      end
+
       def inspect
         "<Fluent::Config::Section #{@params.to_json}>"
       end
@@ -62,6 +66,19 @@ def [](key)
         @params[key.to_sym]
       end
 
+      def respond_to?(symbol, include_all=false)
+        case symbol
+        when :inspect, :nil?, :to_h, :+, :instance_of?, :kind_of?, :[], :respond_to?, :respond_to_missing?, :method_missing,
+          true
+        when :!, :!= , :==, :equal?, :instance_eval, :instane_exec
+          true
+        when :method_missing, :singleton_method_added, :singleton_method_removed, :singleton_method_undefined
+          include_all
+        else
+          false
+        end
+      end
+
       def respond_to_missing?(symbol, include_private)
         @params.has_key?(symbol)
       end
@@ -70,7 +87,7 @@ def method_missing(name, *args)
         if @params.has_key?(name)
           @params[name]
         else
-          super
+          ::Kernel.raise ::NoMethodError, "undefined method `#{name}' for #{self.inspect}"
         end
       end
     end

lib/fluent/engine.rb

@@ -18,6 +18,7 @@ module Fluent
   require 'fluent/event_router'
   require 'fluent/root_agent'
   require 'fluent/time'
+  require 'fluent/system_config'
 
   class EngineClass
     def initialize
@@ -34,26 +35,30 @@ def initialize
 
       @msgpack_factory = MessagePack::Factory.new
       @msgpack_factory.register_type(Fluent::EventTime::TYPE, Fluent::EventTime)
+      @system_config = SystemConfig.new({})
     end
 
     MATCH_CACHE_SIZE = 1024
     LOG_EMIT_INTERVAL = 0.1
 
+    attr_reader :system_config
     attr_reader :root_agent
     attr_reader :matches, :sources
     attr_reader :msgpack_factory
 
     def init(opts = {})
       BasicSocket.do_not_reverse_lookup = true
-      Plugin.load_plugins
       if defined?(Encoding)
         Encoding.default_internal = 'ASCII-8BIT' if Encoding.respond_to?(:default_internal)
         Encoding.default_external = 'ASCII-8BIT' if Encoding.respond_to?(:default_external)
       end
 
+      @system_config = opts[:system_config] if opts[:system_config]
+
       suppress_interval(opts[:suppress_interval]) if opts[:suppress_interval]
       @suppress_config_dump = opts[:suppress_config_dump] if opts[:suppress_config_dump]
       @without_source = opts[:without_source] if opts[:without_source]
+      @plugin_storage_path = opts[:plugin_storage_path] if opts[:plugin_storage_path]
 
       @root_agent = RootAgent.new(opts)
 
@@ -113,22 +118,20 @@ def configure(conf)
       end
     end
 
-    def load_plugin_dir(dir)
-      Plugin.load_plugin_dir(dir)
+    def add_plugin_dir(dir)
+      Plugin.add_plugin_dir(dir)
     end
 
     def emit(tag, time, record)
-      unless record.nil?
-        emit_stream tag, OneEventStream.new(time, record)
-      end
+      raise "BUG: use router.emit instead of Engine.emit"
     end
 
     def emit_array(tag, array)
-      emit_stream tag, ArrayEventStream.new(array)
+      raise "BUG: use router.emit_array instead of Engine.emit_array"
     end
 
     def emit_stream(tag, es)
-      @event_router.emit_stream(tag, es)
+      raise "BUG: use router.emit_stream instead of Engine.emit_stream"
     end
 
     def flush!

lib/fluent/env.rb

@@ -18,9 +18,9 @@ module Fluent
   DEFAULT_CONFIG_PATH = ENV['FLUENT_CONF'] || '/etc/fluent/fluent.conf'
   DEFAULT_PLUGIN_DIR = ENV['FLUENT_PLUGIN'] || '/etc/fluent/plugin'
   DEFAULT_SOCKET_PATH = ENV['FLUENT_SOCKET'] || '/var/run/fluent/fluent.sock'
-  DEFAULT_LISTEN_PORT = 24224
-  DEFAULT_FILE_PERMISSION = 0644
-  DEFAULT_DIR_PERMISSION = 0755
+  DEFAULT_LISTEN_PORT = 24224 # TODO: obsolete
+  DEFAULT_FILE_PERMISSION = 0644 # TODO: configurable w/ <system>
+  DEFAULT_DIR_PERMISSION = 0755 # TODO: configurable w/ <system>
   IS_WINDOWS = /mswin|mingw/ === RUBY_PLATFORM
   private_constant :IS_WINDOWS
 

lib/fluent/filter.rb

@@ -16,6 +16,7 @@
 
 module Fluent
   class Filter
+    # TODO: move to plugin/filter.rb, and make interoperability layer here
     include Configurable
     include PluginId
     include PluginLoggerMixin

lib/fluent/input.rb

@@ -14,33 +14,10 @@
 #    limitations under the License.
 #
 
-module Fluent
-  class Input
-    include Configurable
-    include PluginId
-    include PluginLoggerMixin
-
-    attr_accessor :router
-
-    def initialize
-      super
-    end
-
-    def configure(conf)
-      super
+require 'fluent/plugin/input'
 
-      if label_name = conf['@label']
-        label = Engine.root_agent.find_label(label_name)
-        @router = label.event_router
-      elsif @router.nil?
-        @router = Engine.root_agent.event_router
-      end
-    end
-
-    def start
-    end
-
-    def shutdown
-    end
+module Fluent
+  class Input < Plugin::Input
+    # TODO: add interoperability layer
   end
 end

lib/fluent/load.rb

@@ -30,8 +30,8 @@
 require 'fluent/parser'
 require 'fluent/formatter'
 require 'fluent/event'
-require 'fluent/buffer'
-require 'fluent/input'
-require 'fluent/output'
-require 'fluent/filter'
+require 'fluent/plugin/buffer'
+require 'fluent/plugin/input'
+require 'fluent/plugin/output'
+require 'fluent/plugin/filter'
 require 'fluent/match'

lib/fluent/log.rb

@@ -65,6 +65,9 @@ def initialize(out=STDERR, level=LEVEL_TRACE, opts={})
       # TODO: This variable name is unclear so we should change to better name.
       @threads_exclude_events = []
 
+      @optional_header = nil
+      @optional_attrs = nil
+
       if opts.has_key?(:suppress_repeated_stacktrace)
         @suppress_repeated_stacktrace = opts[:suppress_repeated_stacktrace]
       end
@@ -74,6 +77,7 @@ def initialize(out=STDERR, level=LEVEL_TRACE, opts={})
     attr_accessor :level
     attr_accessor :tag
     attr_accessor :time_format
+    attr_accessor :optional_header, :optional_attrs
 
     def enable_debug(b=true)
       @debug_mode = b
@@ -263,8 +267,8 @@ def dump_stacktrace(backtrace, level)
 
     def event(level, args)
       time = Time.now
-      message = ''
-      map = {}
+      message = @optional_header ? @optional_header.dup : ''
+      map = @optional_attrs ? @optional_attrs.dup : {}
       args.each {|a|
         if a.is_a?(Hash)
           a.each_pair {|k,v|
@@ -364,6 +368,13 @@ def configure(conf)
           @log = PluginLogger.new($log)
         end
         @log.level = @log_level
+        @log.optional_header = "[#{self.class.name}#{@_id_configured ? "(" + @id + ")" : ""}] "
+        @log.optional_attrs = {
+          'plugin_type' => self.class.name,
+        }
+        if @_id_configured
+          @log.optional_attrs.update({'plugin_id' => @id})
+        end
       end
     end
   end