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 "
# 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
2014-06-01 12:16:33 +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
2015-06-26 11:45:45 +00:00
2014-08-09 12:56:11 +00:00
@logger . info ( " JDBC - Starting up " )
2014-04-15 11:32:41 +00:00
2015-06-26 11:45:45 +00:00
if ENV [ 'LOGSTASH_HOME' ]
jarpath = File . join ( ENV [ 'LOGSTASH_HOME' ] , " /vendor/jar/jdbc/*.jar " )
else
jarpath = File . join ( File . dirname ( __FILE__ ) , " ../../../vendor/jar/jdbc/*.jar " )
end
@logger . debug ( " JDBC - jarpath " , path : jarpath )
jars = Dir [ jarpath ]
raise Exception . new ( " JDBC - No jars found in jarpath. Have you read the README? " ) if jars . empty?
jars . 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 ]
2015-06-26 11:45:45 +00:00
when Time , LogStash :: Timestamp
2014-08-09 12:56:11 +00:00
# Most reliable solution, cross JDBC driver
statement . setString ( idx + 1 , event [ i ] . iso8601 ( ) )
when Fixnum , Integer
2014-10-09 07:53:21 +00:00
statement . setInt ( idx + 1 , event [ i ] )
2014-08-09 12:56:11 +00:00
when Float
2014-10-09 07:53:21 +00:00
statement . setFloat ( idx + 1 , event [ i ] )
2014-08-09 12:56:11 +00:00
when String
statement . setString ( idx + 1 , event [ i ] )
2015-07-22 15:30:37 +00:00
when true
statement . setBoolean ( idx + 1 , true )
when false
statement . setBoolean ( idx + 1 , false )
when nil
statement . setString ( idx + 1 , nil )
else
2014-08-09 12:56:11 +00:00
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 ( )
2014-06-01 12:16:33 +00:00
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
2014-06-01 12:16:33 +00:00
# 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 )
2015-07-22 15:30:37 +00:00
if e . getNextException ( ) != nil
@logger . warn ( " JDBC - Exception. Will automatically retry " , :exception = > e . getNextException ( ) )
end
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