logstash-output-jdbc/lib/logstash/outputs/jdbc.rb

157 lines
5.0 KiB
Ruby
Raw Normal View History

2014-04-15 11:32:41 +00:00
# encoding: utf-8
require "logstash/outputs/base"
require "logstash/namespace"
2014-05-31 13:33:59 +00:00
require "stud/buffer"
2014-08-09 12:56:11 +00:00
require "java"
2014-04-15 11:32:41 +00:00
class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
2014-05-31 13:33:59 +00:00
# Adds buffer support
include Stud::Buffer
2014-04-15 11:32:41 +00:00
config_name "jdbc"
milestone 1
# Driver class
config :driver_class, :validate => :string
# connection string
config :connection_string, :validate => :string, :required => true
2014-08-09 12:56:11 +00:00
# [ "insert into table (message) values(?)", "%{message}" ]
2014-04-15 11:32:41 +00:00
config :statement, :validate => :array, :required => true
2014-05-31 13:33:59 +00:00
# We buffer a certain number of events before flushing that out to SQL.
# This setting controls how many events will be buffered before sending a
# batch of events.
2014-05-31 13:33:59 +00:00
config :flush_size, :validate => :number, :default => 1000
# The amount of time since last flush before a flush is forced.
#
# This setting helps ensure slow event rates don't get stuck in Logstash.
# For example, if your `flush_size` is 100, and you have received 10 events,
# and it has been more than `idle_flush_time` seconds since the last flush,
# Logstash will flush those 10 events automatically.
#
# This helps keep both fast and slow log streams moving along in
# a timely manner.
2014-08-09 12:56:11 +00:00
#
# If you change this value please ensure that you change
# max_repeat_exceptions_time accordingly.
2014-05-31 13:33:59 +00:00
config :idle_flush_time, :validate => :number, :default => 1
2014-08-09 12:56:11 +00:00
# Maximum number of repeating (sequential) exceptions, before we stop retrying
# If set to < 1, then it will infinitely retry.
config :max_repeat_exceptions, :validate => :number, :default => 5
# The max number of seconds since the last exception, before we consider it
# a different cause.
# This value should be carefully considered in respect to idle_flush_time.
config :max_repeat_exceptions_time, :validate => :number, :default => 30
2014-04-15 11:32:41 +00:00
public
def register
2014-08-09 12:56:11 +00:00
@logger.info("JDBC - Starting up")
2014-04-15 11:32:41 +00:00
jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/jdbc/*.jar")
@logger.info(jarpath)
Dir[jarpath].each do |jar|
2014-08-09 12:56:11 +00:00
@logger.debug("JDBC - Loaded jar", :jar => jar)
2014-04-15 11:32:41 +00:00
require jar
end
import @driver_class
driver = Object.const_get(@driver_class[@driver_class.rindex('.') + 1, @driver_class.length]).new
@connection = driver.connect(@connection_string, java.util.Properties.new)
2014-08-09 12:56:11 +00:00
@logger.debug("JDBC - Created connection", :driver => driver, :connection => @connection)
2014-05-31 13:33:59 +00:00
if (@flush_size > 1000)
2014-08-09 12:56:11 +00:00
@logger.warn("JDBC - Flush size is set to > 1000. May have performance penalties, depending on your SQL engine.")
end
@repeat_exception_count = 0
@last_exception_time = Time.now
if (@max_repeat_exceptions > 0) and ((@idle_flush_time * @max_repeat_exceptions) > @max_repeat_exceptions_time)
@logger.warn("JDBC - max_repeat_exceptions_time is set such that it may still permit a looping exception. You probably changed idle_flush_time. Considering increasing max_repeat_exceptions_time.")
2014-05-31 13:33:59 +00:00
end
buffer_initialize(
:max_items => @flush_size,
:max_interval => @idle_flush_time,
:logger => @logger
)
2014-04-15 11:32:41 +00:00
end
def receive(event)
return unless output?(event)
2014-04-16 15:41:45 +00:00
return unless @statement.length > 0
2014-05-31 13:33:59 +00:00
buffer_receive(event)
end
2014-04-16 15:41:45 +00:00
2014-05-31 13:33:59 +00:00
def flush(events, teardown=false)
statement = @connection.prepareStatement(@statement[0])
2014-08-09 12:56:11 +00:00
2014-05-31 13:33:59 +00:00
events.each do |event|
2014-08-09 12:56:11 +00:00
next if @statement.length < 2
@statement[1..-1].each_with_index do |i, idx|
case event[i]
when Time
# Most reliable solution, cross JDBC driver
statement.setString(idx + 1, event[i].iso8601())
when Fixnum, Integer
statement.setInt(idx + 1, event[i])
2014-08-09 12:56:11 +00:00
when Float
statement.setFloat(idx + 1, event[i])
2014-08-09 12:56:11 +00:00
when String
statement.setString(idx + 1, event[i])
else
statement.setString(idx + 1, event.sprintf(i))
end
end
2014-05-31 13:33:59 +00:00
statement.addBatch()
end
2014-04-16 15:41:45 +00:00
2014-04-22 09:01:41 +00:00
begin
2014-08-09 12:56:11 +00:00
@logger.debug("JDBC - Sending SQL", :sql => statement.toString())
2014-05-31 13:33:59 +00:00
statement.executeBatch()
rescue => e
2014-05-31 13:33:59 +00:00
# Raising an exception will incur a retry from Stud::Buffer.
# Since the exceutebatch failed this should mean any events failed to be
# inserted will be re-run. We're going to log it for the lols anyway.
2014-08-09 12:56:11 +00:00
@logger.warn("JDBC - Exception. Will automatically retry", :exception => e)
2014-04-22 09:01:41 +00:00
end
2014-05-31 13:33:59 +00:00
2014-04-16 15:41:45 +00:00
statement.close()
end
2014-08-09 12:56:11 +00:00
def on_flush_error(e)
return if @max_repeat_exceptions < 1
if @last_exception == e.to_s
@repeat_exception_count += 1
else
@repeat_exception_count = 0
end
if (@repeat_exception_count >= @max_repeat_exceptions) and (Time.now - @last_exception_time) < @max_repeat_exceptions_time
@logger.error("JDBC - Exception repeated more than the maximum configured", :exception => e, :max_repeat_exceptions => @max_repeat_exceptions, :max_repeat_exceptions_time => @max_repeat_exceptions_time)
raise e
end
@last_exception_time = Time.now
@last_exception = e.to_s
end
2014-04-16 15:41:45 +00:00
def teardown
2014-05-31 13:33:59 +00:00
buffer_flush(:final => true)
2014-04-16 15:41:45 +00:00
@connection.close()
super
2014-04-15 11:32:41 +00:00
end
end # class LogStash::Outputs::jdbc