plugin: Added filter feature per record.

lib/fluent/filter.rb

@@ -0,0 +1,34 @@
+#!/usr/bin/env ruby
+#
+# Fluent
+#
+# Copyright (C) 2011-2012 FURUHASHI Sadayuki
+# Copyright (C) 2012 NTT Communications
+#
+#    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.
+#
+
+module Fluent
+
+class Filter
+
+  class UserDefinedScirptError < StandardError
+  end
+
+  def filter(time, record)
+    raise UserDefinedScirptError, "'filter' method is undefined."
+  end
+
+end
+
+end

lib/fluent/load.rb

@@ -23,3 +23,4 @@
 require 'fluent/input'
 require 'fluent/output'
 require 'fluent/match'
+require 'fluent/filter'

lib/fluent/plugin.rb

@@ -23,6 +23,7 @@ def initialize
     @input = {}
     @output = {}
     @buffer = {}
+    @filter = {}
   end
 
   def register_input(type, klass)
@@ -37,6 +38,10 @@ def register_buffer(type, klass)
     register_impl('buffer', @buffer, type, klass)
   end
 
+  def register_filter(type, klass)
+    register_impl('filter', @filter, type, klass)
+  end
+
   def new_input(type)
     new_impl('input', @input, type)
   end
@@ -49,6 +54,10 @@ def new_buffer(type)
     new_impl('buffer', @buffer, type)
   end
 
+  def new_filter(type, filter)
+    new_impl('filter', @filter, type, {path: filter})
+  end
+
   def load_plugins
     dir = File.join(File.dirname(__FILE__), "plugin")
     load_plugin_dir(dir)
@@ -86,25 +95,31 @@ def register_impl(name, map, type, klass)
     nil
   end
 
-  def new_impl(name, map, type)
+  def new_impl(name, map, type, opt = {})
     if klass = map[type]
       return klass.new
     end
-    try_load_plugin(name, type)
+    try_load_plugin(name, type, opt)
     if klass = map[type]
       return klass.new
     end
     raise ConfigError, "Unknown #{name} plugin '#{type}'. Run 'gem search -rd fluent-plugin' to find plugins"
   end
 
-  def try_load_plugin(name, type)
+  def try_load_plugin(name, type, opt = {})
     case name
     when 'input'
       path = "fluent/plugin/in_#{type}"
     when 'output'
       path = "fluent/plugin/out_#{type}"
     when 'buffer'
       path = "fluent/plugin/buf_#{type}"
+    when 'filter'
+      path = opt[:path]
+      file = File.exist?(path) ? path : nil
+      require file if file
+      $log.trace file
+      return
     else
       return
     end

lib/fluent/plugin/in_tail.rb

@@ -1,7 +1,8 @@
 #
 # Fluent
 #
-# Copyright (C) 2011 FURUHASHI Sadayuki
+# Copyright (C) 2011-2012 FURUHASHI Sadayuki
+# Copyright (C) 2012 NTT Communications
 #
 #    Licensed under the Apache License, Version 2.0 (the "License");
 #    you may not use this file except in compliance with the License.
@@ -30,6 +31,8 @@ def initialize
   config_param :tag, :string
   config_param :rotate_wait, :time, :default => 5
   config_param :pos_file, :string, :default => nil
+  config_param :filter_script, :string
+  config_param :filter_type, :string
 
   def configure(conf)
     super
@@ -48,6 +51,19 @@ def configure(conf)
       $log.warn "this parameter is highly recommended to save the position to resume tailing."
     end
 
+    @filter_script = conf['filter_script']
+    @filter_type = conf['filter_type']
+    $log.warn @filter_type
+    $log.warn @filter_script
+    if @filter_script
+      begin
+        @filter_instance = Plugin.new_filter(@filter_type, @filter_script)
+      rescue
+        $log.error "illigal script error:", :error=>$!.to_s
+        $log.error_backtrace
+      end
+    end
+
     configure_parser(conf)
   end
 
@@ -84,13 +100,21 @@ def run
     $log.error_backtrace
   end
 
+  def user_defined_filter(time, record)
+    @filter_instance.filter(time, record) if @filter_instance
+  rescue
+    $log.warn line.dump, :error=>$!.to_s
+    $log.debug_backtrace
+  end
+
   def receive_lines(lines)
     es = MultiEventStream.new
     lines.each {|line|
       begin
         line.rstrip!  # remove \n
         time, record = parse_line(line)
         if time && record
+          user_defined_filter(time, record)
           es.add(time, record)
         end
       rescue