Tests.

.gitignore

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

.travis.yml

@@ -5,4 +5,10 @@ rvm:
 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
+  - sudo apt-get install mysql-server  -qq -y
+  - echo "create database logstash_output_jdbc_test;" | mysql -u root
+  - wget http://search.maven.org/remotecontent?filepath=mysql/mysql-connector-java/5.1.38/mysql-connector-java-5.1.38.jar -O /tmp/mysql.jar
+  - export JDBC_MYSQL_JAR=/tmp/mysql.jar
+  - wget http://search.maven.org/remotecontent?filepath=org/xerial/sqlite-jdbc/3.8.11.2/sqlite-jdbc-3.8.11.2.jar -O /tmp/sqlite.jar
+  - export JDBC_SQLITE_JAR=/tmp/sqlite.jar
 script: bundle exec rspec

Vagrantfile

@@ -0,0 +1,15 @@
+# -*- mode: ruby -*-
+# vi: set ft=ruby :
+
+Vagrant.configure(2) do |config|
+  config.vm.box = "debian/jessie64"
+  config.vm.synced_folder ".", "/vagrant", type: :virtualbox
+
+  config.vm.provision "shell", inline: <<-EOP
+    gpg --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
+    curl -sSL https://get.rvm.io | bash -s stable --ruby=jruby-1.7
+    usermod -a -G rvm vagrant
+    gem install bundler
+    cd /vagrant && bundle install
+  EOP
+end

examples/mysql.md

@@ -9,8 +9,9 @@ input
 }
 output {
 	jdbc {
+        driver_class => "com.mysql.jdbc.Driver",
 		connection_string => "jdbc:mysql://HOSTNAME/DATABASE?user=USER&password=PASSWORD"
-		statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, CAST (? AS timestamp), ?)", "host", "@timestamp", "message" ]
+		statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, CAST(? AS timestamp), ?)", "host", "@timestamp", "message" ]
 	}
 }
 ```

lib/logstash-output-jdbc_ring-buffer.rb

@@ -14,4 +14,8 @@ class RingBuffer < Array
       self.push(el)
     end
   end
+
+  def not_nil_length
+    reject { |i| i.nil? }.count
+  end
 end

lib/logstash/outputs/jdbc.rb

@@ -10,6 +10,8 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
   # Adds buffer support
   include Stud::Buffer
 
+  STRFTIME_FMT = "%Y-%m-%d %T.%L".freeze
+
   config_name "jdbc"
 
   # Driver class - Reintroduced for https://github.com/theangryangel/logstash-output-jdbc/issues/26
@@ -126,7 +128,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
 
     @exceptions_tracker << e.class
 
-    if @exceptions_tracker.reject { |i| i.nil? }.count >= @max_flush_exceptions
+    if @exceptions_tracker.not_nil_length >= @max_flush_exceptions
       @logger.error("JDBC - max_flush_exceptions has been reached")
       log_jdbc_exception(e)
       raise LogStash::ShutdownSignal.new
@@ -250,11 +252,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

spec/jdbc_spec_helper.rb

@@ -0,0 +1,98 @@
+require "logstash/devutils/rspec/spec_helper"
+require "logstash/outputs/jdbc"
+require "stud/temporary"
+require "java"
+require 'securerandom'
+
+RSpec.shared_context 'rspec setup' do
+
+  it 'ensure jar is available' do
+    expect(ENV[jdbc_jar_env]).not_to be_nil, "#{jdbc_jar_env} not defined, required to run tests"
+    expect(File.exists?(ENV[jdbc_jar_env])).to eq(true), "#{jdbc_jar_env} defined, but not valid"
+  end
+
+end
+
+RSpec.shared_context 'when initializing' do
+
+  it 'shouldn\'t register with a missing jar file' do
+    jdbc_settings['driver_jar_path'] = nil
+    plugin = LogStash::Plugin.lookup("output", "jdbc").new(jdbc_settings)
+    expect { plugin.register }.to raise_error
+  end
+
+end
+
+RSpec.shared_context 'when outputting messages' do
+
+  let (:jdbc_test_table) do
+    'logstash_output_jdbc_test'
+  end
+
+  let(:jdbc_drop_table) do
+    "DROP TABLE IF EXISTS #{jdbc_test_table}"
+  end
+
+  let(:jdbc_create_table) do
+    "CREATE table #{jdbc_test_table} (created_at datetime, message varchar(512))"
+  end
+
+  let(:event_fields) do
+    { message: "test-message #{SecureRandom.uuid}" }
+  end
+
+  let(:event) { LogStash::Event.new(event_fields) }
+
+  let(:plugin) {
+    # Setup plugin
+    output = LogStash::Plugin.lookup("output", "jdbc").new(jdbc_settings)
+    output.register
+    if ENV['JDBC_DEBUG'] == '1'
+      output.logger.subscribe(STDOUT)
+    end
+
+    # Setup table
+    c = output.instance_variable_get(:@pool).getConnection()
+
+    unless jdbc_drop_table.nil?
+      stmt = c.createStatement()
+      stmt.executeUpdate(jdbc_drop_table)
+      stmt.close()
+    end
+
+    stmt = c.createStatement()
+    stmt.executeUpdate(jdbc_create_table)
+    stmt.close()
+    c.close()
+
+    output
+  }
+
+  it 'should save a event' do
+
+    previous_exceptions_count = plugin.instance_variable_get(:@exceptions_tracker).not_nil_length
+
+    expect { plugin.receive(event) }.to_not raise_error
+
+    # Force flush the buffer, so that we know anything in the buffer
+    # has been sent
+    plugin.buffer_flush(force: true)
+
+    expect(plugin.instance_variable_get(:@exceptions_tracker).not_nil_length).to eq(previous_exceptions_count)
+
+    # Verify the number of items in the output table
+    c = plugin.instance_variable_get(:@pool).getConnection()
+    stmt = c.prepareStatement("select count(*) as total from #{jdbc_test_table} where message = ?")
+    stmt.setString(1, event['message'])
+    rs = stmt.executeQuery()
+    count = 0
+    while rs.next()
+      count = rs.getInt("total")
+    end
+    stmt.close()
+    c.close()
+
+    expect(count).to eq(1)
+  end
+
+end

spec/outputs/jdbc_mysql_spec.rb

@@ -0,0 +1,21 @@
+require_relative "../jdbc_spec_helper"
+
+describe "logstash-output-jdbc: mysql", if: ENV['JDBC_MYSQL_JAR'] do
+
+  include_context "when initializing"
+  include_context "when outputting messages"
+
+  let(:jdbc_jar_env) do
+    'JDBC_MYSQL_JAR'
+  end
+
+  let(:jdbc_settings) do
+    { 
+      "driver_class" => "com.mysql.jdbc.Driver",
+      "connection_string" => "jdbc:mysql://localhost/logstash_output_jdbc_test?user=root",
+      "driver_jar_path" => ENV[jdbc_jar_env],
+      "statement" => [ "insert into #{jdbc_test_table} (created_at, message) values(?, ?)", "@timestamp", "message" ]
+    }
+  end
+
+end

spec/outputs/jdbc_spec.rb

@@ -1,32 +1,32 @@
-require "logstash/devutils/rspec/spec_helper"
+require_relative '../jdbc_spec_helper'
-require "logstash/outputs/jdbc"
-require "stud/temporary"
-require "java"
 
 describe LogStash::Outputs::Jdbc do
 
-  let(:derby_settings) do
+  include_context "rspec setup"
+  include_context "when initializing"
+  include_context "when outputting messages"
+
+  let(:jdbc_jar_env) do
+    'JDBC_DERBY_JAR'
+  end
+
+  let(:jdbc_drop_table) do
+    nil
+  end
+
+  let(:jdbc_create_table) do
+    "CREATE table #{jdbc_test_table} (created_at timestamp, message varchar(512))"
+  end
+
+  let(:jdbc_settings) do
     { 
       "driver_class" => "org.apache.derby.jdbc.EmbeddedDriver",
       "connection_string" => "jdbc:derby:memory:testdb;create=true",
-      "driver_jar_path" => ENV['JDBC_DERBY_JAR'],
+      "driver_jar_path" => ENV[jdbc_jar_env],
-      # Grumble. Grumble.
+      "statement" => [ "insert into logstash_output_jdbc_test (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
 
-  context 'rspec setup' do
-
-    it 'ensure derby is available' do
-      j = ENV['JDBC_DERBY_JAR']
-      expect(j).not_to be_nil, "JDBC_DERBY_JAR not defined, required to run tests"
-      expect(File.exists?(j)).to eq(true), "JDBC_DERBY_JAR defined, but not valid"
-    end
-    
-  end
-
   context 'when initializing' do
 
     it 'shouldn\'t register without a config' do
@@ -35,63 +35,6 @@ describe LogStash::Outputs::Jdbc do
       }.to raise_error(LogStash::ConfigurationError)
     end
 
-    it 'shouldn\'t register with a missing jar file' do
-      derby_settings['driver_jar_path'] = nil
-      plugin = LogStash::Plugin.lookup("output", "jdbc").new(derby_settings)
-      expect { plugin.register }.to raise_error
-    end
-
-    it 'shouldn\'t register with a missing jar file' do
-      derby_settings['connection_string'] = nil
-      plugin = LogStash::Plugin.lookup("output", "jdbc").new(derby_settings)
-      expect { plugin.register }.to raise_error
-    end
-
-  end
-
-  context 'when outputting messages' do
-
-    let(:event_fields) do
-      { message: 'test-message' }
-    end
-    let(:event) { LogStash::Event.new(event_fields) }
-    let(:plugin) {
-      # Setup plugin
-      output = LogStash::Plugin.lookup("output", "jdbc").new(derby_settings)
-      output.register
-      if ENV['JDBC_DEBUG'] == '1'
-        output.logger.subscribe(STDOUT)
-      end
-
-      # Setup table
-      c = output.instance_variable_get(:@pool).getConnection()
-      stmt = c.createStatement()
-      stmt.executeUpdate("CREATE table log (created_at timestamp, message varchar(512))")
-      stmt.close()
-      c.close()
-
-      output
-    }
-
-    it 'should save a event' do
-      expect { plugin.receive(event) }.to_not raise_error
-      
-      # Wait for 1 second, for the buffer to flush
-      sleep 1
-
-      c = plugin.instance_variable_get(:@pool).getConnection()
-      stmt = c.createStatement()
-      rs = stmt.executeQuery("select count(*) as total from log")
-      count = 0
-      while rs.next()
-        count = rs.getInt("total")
-      end
-      stmt.close()
-      c.close()
-
-      expect(count).to be > 0
-    end
-
   end
 
 end

spec/outputs/jdbc_sqlite_spec.rb

@@ -0,0 +1,27 @@
+require_relative "../jdbc_spec_helper"
+
+describe "logstash-output-jdbc: sqlite", if: ENV['JDBC_SQLITE_JAR'] do
+
+  JDBC_SQLITE_FILE = "/tmp/logstash_output_jdbc_test.db"
+
+  before(:context) do
+    File.delete(JDBC_SQLITE_FILE) if File.exists? JDBC_SQLITE_FILE
+  end
+
+  include_context "when initializing"
+  include_context "when outputting messages"
+
+  let(:jdbc_jar_env) do
+    'JDBC_SQLITE_JAR'
+  end
+
+  let(:jdbc_settings) do
+    { 
+      "driver_class" => "org.sqlite.JDBC",
+      "connection_string" => "jdbc:sqlite:#{JDBC_SQLITE_FILE}",
+      "driver_jar_path" => ENV[jdbc_jar_env],
+      "statement" => [ "insert into #{jdbc_test_table} (created_at, message) values(?, ?)", "@timestamp", "message" ]
+    }
+  end
+
+end