Merge pull request #89 from fluent/full-v1-api

out_sql: Use v1 API completely

Author: repeatedly

lib/fluent/plugin/out_sql.rb

@@ -1,11 +1,12 @@
 require "fluent/plugin/output"
 
+require 'active_record'
+require 'activerecord-import'
+
 module Fluent::Plugin
   class SQLOutput < Output
     Fluent::Plugin.register_output('sql', self)
 
-    DEFAULT_BUFFER_TYPE = "memory"
-
     helpers :inject, :compat_parameters, :event_emitter
 
     desc 'RDBMS host'
@@ -33,7 +34,7 @@ class SQLOutput < Output
     config_param :timeout, :integer, default: 5000
 
     config_section :buffer do
-      config_set_default :@type, DEFAULT_BUFFER_TYPE
+      config_set_default :chunk_keys, ["tag"]
     end
 
     attr_accessor :tables
@@ -86,11 +87,12 @@ def init(base_model)
         # @model.column_names
       end
 
-      def import(chunk)
+      def import(chunk, output)
+        tag = chunk.metadata.tag
         records = []
-        chunk.msgpack_each { |tag, time, data|
+        chunk.msgpack_each { |time, data|
           begin
-            # format process should be moved to emit / format after supports error stream.
+            data = output.inject_values_to_record(tag, time, data)
             records << @model.new(@format_proc.call(data))
           rescue => e
             args = {error: e, table: @table, record: Yajl.dump(data)}
@@ -105,7 +107,7 @@ def import(chunk)
             @log.warn "Got deterministic error. Fallback to one-by-one import", error: e
             one_by_one_import(records)
           else
-            $log.warn "Got deterministic error. Fallback is disabled", error: e
+            @log.warn "Got deterministic error. Fallback is disabled", error: e
             raise e
           end
         end
@@ -147,8 +149,6 @@ def parse_column_mapping(column_mapping_conf)
 
     def initialize
       super
-      require 'active_record'
-      require 'activerecord-import'
     end
 
     def configure(conf)
@@ -174,7 +174,6 @@ def configure(conf)
           @tables << te
         end
       }
-      @only_default = @tables.empty?
 
       if @default_table.nil?
         raise Fluent::ConfigError, "There is no default table. <table> is required in sql output"
@@ -215,19 +214,6 @@ def shutdown
       super
     end
 
-    def emit(tag, es, chain)
-      if @only_default
-        super(tag, es, chain)
-      else
-        super(tag, es, chain, format_tag(tag))
-      end
-    end
-
-    def format(tag, time, record)
-      record = inject_values_to_record(tag, time, record)
-      [tag, time, record].to_msgpack
-    end
-
     def formatted_to_msgpack_binary
       true
     end
@@ -236,11 +222,12 @@ def write(chunk)
       ActiveRecord::Base.connection_pool.with_connection do
 
         @tables.each { |table|
-          if table.pattern.match(chunk.key)
-            return table.import(chunk)
+          tag = format_tag(chunk.metadata.tag)
+          if table.pattern.match(tag)
+            return table.import(chunk, self)
           end
         }
-        @default_table.import(chunk)
+        @default_table.import(chunk, self)
       end
     end
 
@@ -259,7 +246,7 @@ def init_table(te, base_model)
     end
 
     def format_tag(tag)
-      if @remove_tag_prefix
+      if tag && @remove_tag_prefix
         tag.gsub(@remove_tag_prefix, '')
       else
         tag