0.2.10

.gitignore

@@ -2,3 +2,4 @@
 Gemfile.lock
 Gemfile.bak
 .bundle
+.vagrant

CHANGELOG.md

@@ -1,6 +1,18 @@
 # Change Log
 All notable changes to this project will be documented in this file, from 0.2.0.
 
+## [0.2.10] - 2016-07-07
+  - Support non-string entries in statement array
+  - Adds backtrace to exception logging
+
+## [0.2.9] - 2016-06-29
+  - Fix NameError exception. 
+  - Moved log_jdbc_exception calls
+
+## [0.2.7] - 2016-05-29
+  - Backport retry exception logic from v5 branch
+  - Backport improved timestamp compatibility from v5 branch
+
 ## [0.2.6] - 2016-05-02
   - Fix for exception infinite loop
 

lib/logstash/outputs/jdbc.rb

@@ -6,10 +6,33 @@ require "java"
 require "logstash-output-jdbc_jars"
 require "logstash-output-jdbc_ring-buffer"
 
+# Write events to a SQL engine, using JDBC.
+#
+# It is upto the user of the plugin to correctly configure the plugin. This
+# includes correctly crafting the SQL statement, and matching the number of
+# parameters correctly.
 class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
   # Adds buffer support
   include Stud::Buffer
 
+  STRFTIME_FMT = '%Y-%m-%d %T.%L'.freeze
+
+  RETRYABLE_SQLSTATE_CLASSES = [
+    # Classes of retryable SQLSTATE codes
+    # Not all in the class will be retryable. However, this is the best that 
+    # we've got right now.
+    # If a custom state code is required, set it in retry_sql_states.
+    '08', # Connection Exception
+    '24', # Invalid Cursor State (Maybe retry-able in some circumstances)
+    '25', # Invalid Transaction State 
+    '40', # Transaction Rollback 
+    '53', # Insufficient Resources
+    '54', # Program Limit Exceeded (MAYBE)
+    '55', # Object Not In Prerequisite State
+    '57', # Operator Intervention
+    '58', # System Error
+  ].freeze
+
   config_name "jdbc"
 
   # Driver class - Reintroduced for https://github.com/theangryangel/logstash-output-jdbc/issues/26
@@ -128,7 +151,6 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
 
     if @exceptions_tracker.reject { |i| i.nil? }.count >= @max_flush_exceptions
       @logger.error("JDBC - max_flush_exceptions has been reached")
-      log_jdbc_exception(e)
       raise LogStash::ShutdownSignal.new
     end
   end
@@ -196,8 +218,15 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
   def safe_flush(events, teardown=false)
     connection = nil
     statement = nil
+    
     begin
       connection = @pool.getConnection()
+    rescue => e
+      log_jdbc_exception(e, true)
+      raise
+    end
+
+    begin
       statement = connection.prepareStatement(@statement[0])
 
       events.each do |event|
@@ -212,7 +241,9 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
       statement.close()
       @exceptions_tracker << nil
     rescue => e
-      log_jdbc_exception(e)
+      if retry_exception?(e)
+        raise
+      end
     ensure
       statement.close() unless statement.nil?
       connection.close() unless connection.nil?
@@ -224,7 +255,12 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
     statement = nil
     begin
       connection = @pool.getConnection()
+    rescue => e
+      log_jdbc_exception(e, true)
+      raise
+    end
 
+    begin
       events.each do |event|
         next if event.cancelled?
 
@@ -239,7 +275,9 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
         @exceptions_tracker << nil
       end
     rescue => e
-      log_jdbc_exception(e)
+      if retry_exception?(e)
+        raise
+      end
     ensure
       statement.close() unless statement.nil?
       connection.close() unless connection.nil?
@@ -248,41 +286,67 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
 
   def add_statement_event_params(statement, event)
     @statement[1..-1].each_with_index do |i, idx|
-      case event[i]
+      if i.is_a? String 
-      when Time
+        value = event[i]
-        # Most reliable solution, cross JDBC driver
+        if value.nil? and i =~ /%\{/
-        statement.setString(idx + 1, event[i].iso8601())
+          value = event.sprintf(i)
-      when LogStash::Timestamp
-        # Most reliable solution, cross JDBC driver
-        statement.setString(idx + 1, event[i].to_iso8601())
-      when Fixnum, Integer
-        statement.setInt(idx + 1, event[i])
-      when Float
-        statement.setFloat(idx + 1, event[i])
-      when String
-        statement.setString(idx + 1, event[i])
-      when true
-        statement.setBoolean(idx + 1, true)
-      when false
-        statement.setBoolean(idx + 1, false)
-      else
-        if event[i].nil? and i =~ /%\{/
-          statement.setString(idx + 1, event.sprintf(i))
-        else
-          statement.setString(idx + 1, nil)
         end
+      else
+        value = i
+      end
+      
+      case value
+      when Time
+        # See LogStash::Timestamp, below, for the why behind strftime.
+        statement.setString(idx + 1, value.strftime(STRFTIME_FMT))
+      when LogStash::Timestamp
+        # XXX: Using setString as opposed to setTimestamp, because setTimestamp
+        # doesn't behave correctly in some drivers (Known: sqlite)
+        #
+        # Additionally this does not use `to_iso8601`, since some SQL databases
+        # choke on the 'T' in the string (Known: Derby).
+        #
+        # strftime appears to be the most reliable across drivers.
+        statement.setString(idx + 1, value.time.strftime(STRFTIME_FMT))
+      when Fixnum, Integer
+        statement.setInt(idx + 1, value)
+      when Float
+        statement.setFloat(idx + 1, value)
+      when String
+        statement.setString(idx + 1, value)
+      when true, false
+        statement.setBoolean(idx + 1, value)
+      else
+        statement.setString(idx + 1, nil)
       end
     end
 
     statement
   end
 
-  def log_jdbc_exception(exception)
+
+  def log_jdbc_exception(exception, retrying)
     current_exception = exception
+    log_text = 'JDBC - Exception. ' + (retrying ? 'Retrying' : 'Not retrying') + '.'
+    log_method = (retrying ? 'warn' : 'error')
+
     loop do
-      @logger.error("JDBC Exception encountered: Will automatically retry.", :exception => current_exception)
+      @logger.send(log_method, log_text, :exception => current_exception, :backtrace => current_exception.backtrace)
-      current_exception = current_exception.getNextException()
+
+      if current_exception.respond_to? 'getNextException'
+        current_exception = current_exception.getNextException()
+      else
+        current_exception = nil
+      end
+
       break if current_exception == nil
     end
   end
+
+  def retry_exception?(exception)
+    retrying = (exception.respond_to? 'getSQLState' and RETRYABLE_SQLSTATE_CLASSES.include?(exception.getSQLState.to_s[0,2]))
+    log_jdbc_exception(exception, retrying)
+
+    retrying
+  end
 end # class LogStash::Outputs::jdbc

logstash-output-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-output-jdbc'
-  s.version = "0.2.6"
+  s.version = "0.2.10"
   s.licenses = [ "Apache License (2.0)" ]
   s.summary = "This plugin allows you to output to SQL, via JDBC"
   s.description = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"
@@ -19,11 +19,11 @@ Gem::Specification.new do |s|
   s.metadata = { "logstash_plugin" => "true", "logstash_group" => "output" }
 
   # Gem dependencies
-  s.add_runtime_dependency "logstash-core", ">= 2.0.0.beta2", "< 3.0.0"
   s.add_runtime_dependency 'stud'
+
+  s.add_runtime_dependency "logstash-core", ">= 2.0.0", "< 3.0.0"
   s.add_runtime_dependency "logstash-codec-plain"
 
-  s.add_development_dependency "logstash-devutils"
+  # https://github.com/elastic/logstash-devutils/issues/48
-
+  s.add_development_dependency "logstash-devutils", '0.0.18'
-  s.post_install_message = "logstash-output-jdbc 0.2.0 introduces several new features - please ensure you check the documentation in the README file"
 end

spec/outputs/jdbc_spec.rb

@@ -10,10 +10,7 @@ describe LogStash::Outputs::Jdbc do
       "driver_class" => "org.apache.derby.jdbc.EmbeddedDriver",
       "connection_string" => "jdbc:derby:memory:testdb;create=true",
       "driver_jar_path" => ENV['JDBC_DERBY_JAR'],
-      # Grumble. Grumble.
+      "statement" => [ "insert into log (created_at, message) values(?, ?)", "@timestamp" "message" ]
-      # Derby doesn't like 'T' in timestamps as of current writing, so for now
-      # we'll just use CURRENT_TIMESTAMP as opposed to the event @timestamp
-      "statement" => [ "insert into log (created_at, message) values(CURRENT_TIMESTAMP, ?)", "message" ]
     }
   end