0.2.9

.gitignore

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

.travis.yml

@@ -0,0 +1,8 @@
+language: ruby
+cache: bundler
+rvm:
+  - jruby
+before_script:
+  - wget http://search.maven.org/remotecontent?filepath=org/apache/derby/derby/10.12.1.1/derby-10.12.1.1.jar -O /tmp/derby.jar
+  - export JDBC_DERBY_JAR=/tmp/derby.jar
+script: bundle exec rspec

CHANGELOG.md

@@ -1,6 +1,17 @@
 # Change Log
 All notable changes to this project will be documented in this file, from 0.2.0.
 
+## [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
+
 ## [0.2.5] - 2016-04-11
 ### Added
   - Basic tests running against DerbyDB

README.md

@@ -1,4 +1,7 @@
 # logstash-output-jdbc 
+
+[![Build Status](https://travis-ci.org/theangryangel/logstash-output-jdbc.svg?branch=master)](https://travis-ci.org/theangryangel/logstash-output-jdbc)
+
 This plugin is provided as an external plugin and is not part of the Logstash project.
 
 This plugin allows you to output to SQL databases, using JDBC adapters.
@@ -6,7 +9,7 @@ See below for tested adapters, and example configurations.
 
 This has not yet been extensively tested with all JDBC drivers and may not yet work for you.
 
-If you do find this works for a JDBC driver not listed, let me know and provide a small example configuration.
+If you do find this works for a JDBC driver without an example, let me know and provide a small example configuration if you can.
 
 This plugin does not bundle any JDBC jar files, and does expect them to be in a
 particular location. Please ensure you read the 4 installation lines below.
@@ -15,10 +18,11 @@ particular location. Please ensure you read the 4 installation lines below.
 See CHANGELOG.md
 
 ## Versions
-Released versions are are tagged as of v0.2.1, and available via rubygems.
+Released versions are available via rubygems, and typically tagged.
 
 For development:
-  - See master branch for logstash v2
+  - See master branch for logstash v5
+  - See v2.x branch for logstash v2
   - See v1.5 branch for logstash v1.5 
   - See v1.4 branch for logstash 1.4
 

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
@@ -194,56 +216,71 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
   end
 
   def safe_flush(events, teardown=false)
-    connection = @pool.getConnection()
+    connection = nil
-    statement = connection.prepareStatement(@statement[0])
+    statement = nil
     
-    events.each do |event|
+    begin
-      next if event.cancelled?
+      connection = @pool.getConnection()
-      next if @statement.length < 2
+    rescue => e
-      statement = add_statement_event_params(statement, event)
+      log_jdbc_exception(e, true)
-
+      raise
-      statement.addBatch()
     end
 
     begin
+      statement = connection.prepareStatement(@statement[0])
+
+      events.each do |event|
+        next if event.cancelled?
+        next if @statement.length < 2
+        statement = add_statement_event_params(statement, event)
+
+        statement.addBatch()
+      end
+
       statement.executeBatch()
       statement.close()
       @exceptions_tracker << nil
-      
     rescue => e
-      # Raising an exception will incur a retry from Stud::Buffer.
+      if retry_exception?(e)
-      # Since the exceutebatch failed this should mean any events failed to be
+        raise
-      # inserted will be re-run. We're going to log it for the lols anyway.
+      end
-      log_jdbc_exception(e)
     ensure
-      connection.close();
+      statement.close() unless statement.nil?
+      connection.close() unless connection.nil?
     end
   end
 
   def unsafe_flush(events, teardown=false)
-    connection = @pool.getConnection()
+    connection = nil
+    statement = nil
+    begin
+      connection = @pool.getConnection()
+    rescue => e
+      log_jdbc_exception(e, true)
+      raise
+    end
 
-    events.each do |event|
+    begin
-      next if event.cancelled?
+      events.each do |event|
+        next if event.cancelled?
 
-      statement = connection.prepareStatement(event.sprintf(@statement[0]))
+        statement = connection.prepareStatement(event.sprintf(@statement[0]))
-      statement = add_statement_event_params(statement, event) if @statement.length > 1
+        statement = add_statement_event_params(statement, event) if @statement.length > 1
 
-      begin
         statement.execute()
 
         # cancel the event, since we may end up outputting the same event multiple times
         # if an exception happens later down the line
         event.cancel
         @exceptions_tracker << nil
-      rescue => e
-        # Raising an exception will incur a retry from Stud::Buffer.
-        # We log for the lols.
-        log_jdbc_exception(e)
-      ensure
-        statement.close()
-        connection.close()
       end
+    rescue => e
+      if retry_exception?(e)
+        raise
+      end
+    ensure
+      statement.close() unless statement.nil?
+      connection.close() unless connection.nil?
     end
   end
 
@@ -251,11 +288,17 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
     @statement[1..-1].each_with_index do |i, idx|
       case event[i]
       when Time
-        # Most reliable solution, cross JDBC driver
+        # See LogStash::Timestamp, below, for the why behind strftime.
-        statement.setString(idx + 1, event[i].iso8601())
+        statement.setString(idx + 1, event[i].strftime(STRFTIME_FMT))
       when LogStash::Timestamp
-        # Most reliable solution, cross JDBC driver
+        # XXX: Using setString as opposed to setTimestamp, because setTimestamp
-        statement.setString(idx + 1, event[i].to_iso8601())
+        # 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, event[i].time.strftime(STRFTIME_FMT))
       when Fixnum, Integer
         statement.setInt(idx + 1, event[i])
       when Float
@@ -278,12 +321,23 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
     statement
   end
 
-  def log_jdbc_exception(e)
+  def log_jdbc_exception(exception, retrying)
-    ce = e
+    current_exception = exception
     loop do
-      @logger.error("JDBC Exception encountered: Will automatically retry.", :exception => ce)
+      if retrying
-      ce = e.getNextException()
+        @logger.error("JDBC Exception. Retrying.", :exception => current_exception)
-      break if ce == nil
+      else
+        @logger.error("JDBC Exception. No retry.", :exception => current_exception)
+      end
+      current_exception = current_exception.getNextException()
+      break if current_exception == nil
     end
   end
+
+  def retry_exception?(exception)
+    retrying = (exception.respond_to? 'getSQLState' and RETRYABLE_SQLSTATE_CLASSES.include?(exception.getSQLState[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.5"
+  s.version = "0.2.9"
   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"

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
 
@@ -88,6 +85,7 @@ describe LogStash::Outputs::Jdbc do
       end
       stmt.close()
       c.close()
+
       expect(count).to be > 0
     end