out_sql.rb in fluent-plugin-sql-0.5.0

- old
+ new

@@ -3,17 +3,33 @@
     Plugin.register_output('sql', self)
 
     include SetTimeKeyMixin
     include SetTagKeyMixin
 
+    # For fluentd v0.12.16 or earlier
+    class << self
+      unless method_defined?(:desc)
+        def desc(description)
+        end
+      end
+    end
+
+    desc 'RDBMS host'
     config_param :host, :string
+    desc 'RDBMS port'
     config_param :port, :integer, :default => nil
+    desc 'RDBMS driver name.'
     config_param :adapter, :string
+    desc 'RDBMS login user name'
     config_param :username, :string, :default => nil
+    desc 'RDBMS login password'
     config_param :password, :string, :default => nil, :secret => true
+    desc 'RDBMS database name'
     config_param :database, :string
+    desc 'RDBMS socket path'
     config_param :socket, :string, :default => nil
+    desc 'remove the given prefix from the events'
     config_param :remove_tag_prefix, :string, :default => nil
 
     attr_accessor :tables
 
     unless method_defined?(:log)
@@ -62,11 +78,11 @@
         base_model.const_set(class_name, @model)
         model_name = ActiveModel::Name.new(@model, nil, class_name)
         @model.define_singleton_method(:model_name) { model_name }
 
         # TODO: check column_names and table schema
-        columns = @model.columns.map { |column| column.name }.sort
+        # @model.column_names
       end
 
       def import(chunk)
         records = []
         chunk.msgpack_each { |tag, time, data|
@@ -197,9 +213,12 @@
     def format(tag, time, record)
       [tag, time, record].to_msgpack
     end
 
     def write(chunk)
+      conn = @base_model.connection
+      conn.active? || conn.reconnect!
+
       @tables.each { |table|
         if table.pattern.match(chunk.key)
           return table.import(chunk)
         end
       }