will/logstash-output-jdbc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rubocop, in preparation for pre-release rake task that sanity checks for release

Browse Source
This commit is contained in:
Karl Southern 2016-05-17 16:21:37 +01:00
parent 5f0f897114
commit d362e791e5
12 changed files with 191 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
.rubocop.yml Normal file
View File

@ -0,0 +1,25 @@
# I don't care for underscores in numbers.
Style/NumericLiterals:
Enabled: false
Style/ClassAndModuleChildren:
Enabled: false
Metrics/AbcSize:
Enabled: false
Metrics/CyclomaticComplexity:
Max: 9
Metrics/PerceivedComplexity:
Max: 10
Metrics/LineLength:
Enabled: false
Metrics/MethodLength:
Max: 50
Style/FileName:
Exclude:
- 'lib/logstash-output-jdbc_jars.rb'

8
Rakefile
View File

@ -1,9 +1,9 @@
require "logstash/devutils/rake"
require 'logstash/devutils/rake'
require 'jars/installer'
desc "Fetch any jars required for this plugin"
desc 'Fetch any jars required for this plugin'
task :install_jars do
ENV['JARS_HOME'] = Dir.pwd + "/vendor/jar-dependencies/runtime-jars"
ENV['JARS_VENDOR'] = "false"
ENV['JARS_HOME'] = Dir.pwd + '/vendor/jar-dependencies/runtime-jars'
ENV['JARS_VENDOR'] = 'false'
Jars::Installer.new.vendor_jars!(false)
end

6
Vagrantfile vendored
View File

@ -2,10 +2,10 @@
# vi: set ft=ruby :
Vagrant.configure(2) do |config|
config.vm.box = "debian/jessie64"
config.vm.synced_folder ".", "/vagrant", type: :virtualbox
config.vm.box = 'debian/jessie64'
config.vm.synced_folder '.', '/vagrant', type: :virtualbox
config.vm.provision "shell", inline: <<-EOP
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

4
lib/logstash-output-jdbc_jars.rb
View File

@ -1,5 +1,5 @@
# encoding: utf-8
require 'logstash/environment'
root_dir = File.expand_path(File.join(File.dirname(__FILE__), ".."))
LogStash::Environment.load_runtime_jars! File.join(root_dir, "vendor")
root_dir = File.expand_path(File.join(File.dirname(__FILE__), '..'))
LogStash::Environment.load_runtime_jars! File.join(root_dir, 'vendor')

136
lib/logstash/outputs/jdbc.rb
View File

@ -1,13 +1,18 @@
# encoding: utf-8
require "logstash/outputs/base"
require "logstash/namespace"
require "concurrent"
require "stud/interval"
require "java"
require "logstash-output-jdbc_jars"
require 'logstash/outputs/base'
require 'logstash/namespace'
require 'concurrent'
require 'stud/interval'
require 'java'
require 'logstash-output-jdbc_jars'
# 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
STRFTIME_FMT = "%Y-%m-%d %T.%L".freeze
STRFTIME_FMT = '%Y-%m-%d %T.%L'.freeze
# Will never work, but only because it duplicates data (i.e. duplicate keys)
# Will log a warning, but not retry.
@ -21,7 +26,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
23000,
# A violation of the constraint imposed by a unique index or a unique constraint occurred.
23505
]
].freeze
# Will log an error, but not retry.
SQL_STATES_FATAL = [
@ -44,82 +49,79 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
23503,
# The range of values for the identity column or sequence is exhausted.
23522
]
].freeze
config_name "jdbc"
config_name 'jdbc'
# Driver class - Reintroduced for https://github.com/theangryangel/logstash-output-jdbc/issues/26
config :driver_class, :validate => :string
config :driver_class, validate: :string
# Does the JDBC driver support autocommit?
config :driver_auto_commit, :validate => :boolean, :default => true, :required => true
config :driver_auto_commit, validate: :boolean, default: true, required: true
# Where to find the jar
# Defaults to not required, and to the original behaviour
config :driver_jar_path, :validate => :string, :required => false
config :driver_jar_path, validate: :string, required: false
# jdbc connection string
config :connection_string, :validate => :string, :required => true
config :connection_string, validate: :string, required: true
# jdbc username - optional, maybe in the connection string
config :username, :validate => :string, :required => false
config :username, validate: :string, required: false
# jdbc password - optional, maybe in the connection string
config :password, :validate => :string, :required => false
config :password, validate: :string, required: false
# [ "insert into table (message) values(?)", "%{message}" ]
config :statement, :validate => :array, :required => true
config :statement, validate: :array, required: true
# If this is an unsafe statement, use event.sprintf
# This also has potential performance penalties due to having to create a
# new statement for each event, rather than adding to the batch and issuing
# multiple inserts in 1 go
config :unsafe_statement, :validate => :boolean, :default => false
config :unsafe_statement, validate: :boolean, default: false
# Number of connections in the pool to maintain
config :max_pool_size, :validate => :number, :default => 5
config :max_pool_size, validate: :number, default: 5
# Connection timeout
config :connection_timeout, :validate => :number, :default => 10000
config :connection_timeout, validate: :number, default: 10000
# 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.
config :flush_size, :validate => :number, :default => 1000
config :flush_size, validate: :number, default: 1000
# Set initial interval in seconds between retries. Doubled on each retry up to `retry_max_interval`
config :retry_initial_interval, :validate => :number, :default => 2
config :retry_initial_interval, validate: :number, default: 2
# Maximum time between retries, in seconds
config :retry_max_interval, :validate => :number, :default => 128
config :retry_max_interval, validate: :number, default: 128
# Maximum number of sequential failed attempts, before we stop retrying.
# If set to < 1, then it will infinitely retry.
# At the default values this is a little over 10 minutes
config :max_flush_exceptions, :validate => :number, :default => 10
config :max_flush_exceptions, validate: :number, default: 10
config :max_repeat_exceptions, :obsolete => "This has been replaced by max_flush_exceptions - which behaves slightly differently. Please check the documentation."
config :max_repeat_exceptions_time, :obsolete => "This is no longer required"
config :idle_flush_time, :obsolete => "No longer necessary under Logstash v5"
config :max_repeat_exceptions, obsolete: 'This has been replaced by max_flush_exceptions - which behaves slightly differently. Please check the documentation.'
config :max_repeat_exceptions_time, obsolete: 'This is no longer required'
config :idle_flush_time, obsolete: 'No longer necessary under Logstash v5'
public
def register
@logger.info("JDBC - Starting up")
@logger.info('JDBC - Starting up')
LogStash::Logger.setup_log4j(@logger)
load_jar_files!
@stopping = Concurrent::AtomicBoolean.new(false)
if (@flush_size > 1000)
@logger.warn("JDBC - Flush size is set to > 1000")
@logger.warn('JDBC - Flush size is set to > 1000') if @flush_size > 1000
if @statement.empty?
@logger.error('JDBC - No statement provided. Configuration error.')
end
if @statement.length < 1
@logger.error("JDBC - No statement provided. Configuration error.")
end
if (!@unsafe_statement and @statement.length < 2)
if !@unsafe_statement && @statement.length < 2
@logger.error("JDBC - Statement has no parameters. No events will be inserted into SQL as you're not passing any event data. Likely configuration error.")
end
@ -133,7 +135,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
end
def teardown
@pool.close()
@pool.close
super
end
@ -161,36 +163,36 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
validate_connection_timeout = (@connection_timeout / 1000) / 2
# Test connection
test_connection = @pool.getConnection()
test_connection = @pool.getConnection
unless test_connection.isValid(validate_connection_timeout)
@logger.error("JDBC - Connection is not valid. Please check connection string or that your JDBC endpoint is available.")
@logger.error('JDBC - Connection is not valid. Please check connection string or that your JDBC endpoint is available.')
end
test_connection.close()
test_connection.close
end
def load_jar_files!
# Load jar from driver path
unless @driver_jar_path.nil?
raise Exception.new("JDBC - Could not find jar file at given path. Check config.") unless File.exists? @driver_jar_path
raise Exception, 'JDBC - Could not find jar file at given path. Check config.' unless File.exist? @driver_jar_path
require @driver_jar_path
return
end
# Revert original behaviour of loading from vendor directory
# if no path given
if ENV['LOGSTASH_HOME']
jarpath = File.join(ENV['LOGSTASH_HOME'], "/vendor/jar/jdbc/*.jar")
jarpath = if ENV['LOGSTASH_HOME']
File.join(ENV['LOGSTASH_HOME'], '/vendor/jar/jdbc/*.jar')
else
jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/jdbc/*.jar")
File.join(File.dirname(__FILE__), '../../../vendor/jar/jdbc/*.jar')
end
@logger.debug("JDBC - jarpath", path: jarpath)
@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?
raise Exception, 'JDBC - No jars found. Have you read the README?' if jars.empty?
jars.each do |jar|
@logger.debug("JDBC - Loaded jar", :jar => jar)
@logger.debug('JDBC - Loaded jar', jar: jar)
require jar
end
end
@ -201,7 +203,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
events_to_retry = []
begin
connection = @pool.getConnection()
connection = @pool.getConnection
rescue => e
log_jdbc_exception(e)
return events
@ -209,19 +211,15 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
events.each do |event|
begin
if @unsafe_statement == true
statement = connection.prepareStatement(event.sprintf(@statement[0]))
else
statement = connection.prepareStatement(@statement[0])
end
statement = connection.prepareStatement(
(@unsafe_statement == true) ? event.sprintf(@statement[0]) : @statement[0]
)
statement = add_statement_event_params(statement, event) if @statement.length > 1
statement.execute()
statement.execute
rescue java.sql.SQLException => e
if SQL_STATES_IGNORE.include? e.getSQLState()
if SQL_STATES_IGNORE.include? e.getSQLState
@logger.warn('JDBC - Dropping event. Ignore-able exception (duplicate key most likely)', exception: e, event: event)
elsif SQL_STATES_FATAL.include? e.getSQLState()
elsif SQL_STATES_FATAL.include? e.getSQLState
@logger.error('JDBC - Fatal SQL exception. Can never succeed. Dropping event.', exception: e, event: event)
else
log_jdbc_exception(e)
@ -232,13 +230,13 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
log_jdbc_exception(e)
events_to_retry.push(event)
ensure
statement.close() unless statement.nil?
statement.close unless statement.nil?
end
end
connection.close() unless connection.nil?
connection.close unless connection.nil?
return events_to_retry
events_to_retry
end
def retrying_submit(actions)
@ -248,7 +246,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
attempts = 0
sleep_interval = @retry_initial_interval
while submit_actions && submit_actions.length > 0
while submit_actions && !submit_actions.empty?
return if !submit_actions || submit_actions.empty? # If everything's a success we move along
# We retry whatever didn't succeed
submit_actions = submit(submit_actions)
@ -293,12 +291,10 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
statement.setFloat(idx + 1, event.get(i))
when String
statement.setString(idx + 1, event.get(i))
when true
statement.setBoolean(idx + 1, true)
when false
statement.setBoolean(idx + 1, false)
when true, false
statement.setBoolean(idx + 1, event.get(i))
else
if event.get(i).nil? and i =~ /%\{/
if event.get(i).nil? && i =~ /%\{/
statement.setString(idx + 1, event.sprintf(i))
else
statement.setString(idx + 1, nil)
@ -312,9 +308,9 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
def log_jdbc_exception(exception)
current_exception = exception
loop do
@logger.warn("JDBC Exception encountered: Will automatically retry.", :exception => current_exception)
current_exception = current_exception.getNextException()
break if current_exception == nil
@logger.warn('JDBC Exception encountered: Will automatically retry.', exception: current_exception)
current_exception = current_exception.getNextException
break if current_exception.nil?
end
end

28
logstash-output-jdbc.gemspec
View File

@ -1,38 +1,40 @@
Gem::Specification.new do |s|
s.name = 'logstash-output-jdbc'
s.version = "1.0.0.pre"
s.licenses = [ "Apache License (2.0)" ]
s.summary = "This plugin allows you to output to SQL, via JDBC"
s.version = '1.0.0.pre'
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/logstash-plugin install 'logstash-output-jdbc'. This gem is not a stand-alone program"
s.authors = ["the_angry_angel"]
s.email = "karl+github@theangryangel.co.uk"
s.homepage = "https://github.com/theangryangel/logstash-output-jdbc"
s.require_paths = [ "lib" ]
s.authors = ['the_angry_angel']
s.email = 'karl+github@theangryangel.co.uk'
s.homepage = 'https://github.com/theangryangel/logstash-output-jdbc'
s.require_paths = ['lib']
# Java only
s.platform = 'java'
# Files
s.files = Dir.glob("{lib,spec}/**/*.rb") + Dir.glob("vendor/**/*") + %w(LICENSE.txt README.md)
s.files = Dir.glob('{lib,spec}/**/*.rb') + Dir.glob('vendor/**/*') + %w(LICENSE.txt README.md)
# Tests
s.test_files = s.files.grep(%r{^(test|spec|features)/})
# Special flag to let us know this is actually a logstash plugin
s.metadata = { "logstash_plugin" => "true", "logstash_group" => "output" }
s.metadata = { 'logstash_plugin' => 'true', 'logstash_group' => 'output' }
# Gem dependencies
s.add_runtime_dependency "logstash-core-plugin-api", "~> 2.0"
s.add_runtime_dependency 'logstash-core-plugin-api', '~> 2.0'
s.add_runtime_dependency 'stud'
s.add_runtime_dependency "logstash-codec-plain"
s.add_runtime_dependency 'logstash-codec-plain'
# Stuck on HikariCP 2.4.2 until https://github.com/brettwooldridge/HikariCP/issues/600
# is resolved.
s.requirements << "jar 'com.zaxxer:HikariCP', '2.4.2'"
s.requirements << "jar 'org.slf4j:slf4j-log4j12', '1.7.13'"
s.add_development_dependency "jar-dependencies"
s.add_development_dependency 'jar-dependencies'
s.add_development_dependency 'ruby-maven', '~> 3.3'
s.add_development_dependency "logstash-devutils"
s.add_development_dependency 'logstash-devutils'
s.add_development_dependency 'rubocop'
end

1
scripts/minutes_to_retries.rb
View File

@ -1,7 +1,6 @@
#!/usr/bin/env ruby -w
seconds_to_reach = 10 * 60
default_interval = 2
retry_max_interval = 128
current_interval = 2

52
spec/jdbc_spec_helper.rb
View File

@ -1,30 +1,25 @@
require "logstash/devutils/rspec/spec_helper"
require "logstash/outputs/jdbc"
require "stud/temporary"
require "java"
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"
expect(File.exist?(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)
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
@ -43,30 +38,28 @@ RSpec.shared_context 'when outputting messages' do
let(:event) { LogStash::Event.new(event_fields) }
let(:plugin) {
let(:plugin) do
# Setup plugin
output = LogStash::Plugin.lookup("output", "jdbc").new(jdbc_settings)
output = LogStash::Plugin.lookup('output', 'jdbc').new(jdbc_settings)
output.register
if ENV['JDBC_DEBUG'] == '1'
output.logger.subscribe(STDOUT)
end
output.logger.subscribe(STDOUT) if ENV['JDBC_DEBUG'] == '1'
# Setup table
c = output.instance_variable_get(:@pool).getConnection()
c = output.instance_variable_get(:@pool).getConnection
unless jdbc_drop_table.nil?
stmt = c.createStatement()
stmt = c.createStatement
stmt.executeUpdate(jdbc_drop_table)
stmt.close()
stmt.close
end
stmt = c.createStatement()
stmt = c.createStatement
stmt.executeUpdate(jdbc_create_table)
stmt.close()
c.close()
stmt.close
c.close
output
}
end
it 'should save a event' do
expect { plugin.multi_receive([event]) }.to_not raise_error
@ -74,18 +67,15 @@ RSpec.shared_context 'when outputting messages' do
sleep 5
# Verify the number of items in the output table
c = plugin.instance_variable_get(:@pool).getConnection()
c = plugin.instance_variable_get(:@pool).getConnection
stmt = c.prepareStatement("select count(*) as total from #{jdbc_test_table} where message = ?")
stmt.setString(1, event.get('message'))
rs = stmt.executeQuery()
rs = stmt.executeQuery
count = 0
while rs.next()
count = rs.getInt("total")
end
stmt.close()
c.close()
count = rs.getInt('total') while rs.next
stmt.close
c.close
expect(count).to eq(1)
end
end

18
spec/outputs/jdbc_derby_spec.rb
View File

@ -1,10 +1,9 @@
require_relative '../jdbc_spec_helper'
describe "logstash-output-jdbc: derby", if: ENV['JDBC_DERBY_JAR'] do
include_context "rspec setup"
include_context "when initializing"
include_context "when outputting messages"
describe 'logstash-output-jdbc: derby', if: ENV['JDBC_DERBY_JAR'] do
include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages'
let(:jdbc_jar_env) do
'JDBC_DERBY_JAR'
@ -20,11 +19,10 @@ describe "logstash-output-jdbc: derby", if: ENV['JDBC_DERBY_JAR'] do
let(:jdbc_settings) do
{
"driver_class" => "org.apache.derby.jdbc.EmbeddedDriver",
"connection_string" => "jdbc:derby:memory:testdb;create=true",
"driver_jar_path" => ENV[jdbc_jar_env],
"statement" => [ "insert into logstash_output_jdbc_test (created_at, message) values(?, ?)", "@timestamp", "message" ]
'driver_class' => 'org.apache.derby.jdbc.EmbeddedDriver',
'connection_string' => 'jdbc:derby:memory:testdb;create=true',
'driver_jar_path' => ENV[jdbc_jar_env],
'statement' => ['insert into logstash_output_jdbc_test (created_at, message) values(?, ?)', '@timestamp', 'message']
}
end
end

20
spec/outputs/jdbc_mysql_spec.rb
View File

@ -1,10 +1,9 @@
require_relative "../jdbc_spec_helper"
require_relative '../jdbc_spec_helper'
describe "logstash-output-jdbc: mysql", if: ENV['JDBC_MYSQL_JAR'] do
include_context "rspec setup"
include_context "when initializing"
include_context "when outputting messages"
describe 'logstash-output-jdbc: mysql', if: ENV['JDBC_MYSQL_JAR'] do
include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages'
let(:jdbc_jar_env) do
'JDBC_MYSQL_JAR'
@ -12,11 +11,10 @@ describe "logstash-output-jdbc: mysql", if: ENV['JDBC_MYSQL_JAR'] do
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" ]
'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

12
spec/outputs/jdbc_spec.rb
View File

@ -1,13 +1,11 @@
require_relative "../jdbc_spec_helper"
require_relative '../jdbc_spec_helper'
describe LogStash::Outputs::Jdbc do
context 'when initializing' do
it 'shouldn\'t register without a config' do
expect {
LogStash::Plugin.lookup("output", "jdbc").new()
}.to raise_error(LogStash::ConfigurationError)
end
expect do
LogStash::Plugin.lookup('output', 'jdbc').new
end.to raise_error(LogStash::ConfigurationError)
end
end
end

24
spec/outputs/jdbc_sqlite_spec.rb
View File

@ -1,16 +1,15 @@
require_relative "../jdbc_spec_helper"
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"
describe 'logstash-output-jdbc: sqlite', if: ENV['JDBC_SQLITE_JAR'] do
JDBC_SQLITE_FILE = '/tmp/logstash_output_jdbc_test.db'.freeze
before(:context) do
File.delete(JDBC_SQLITE_FILE) if File.exists? JDBC_SQLITE_FILE
File.delete(JDBC_SQLITE_FILE) if File.exist? JDBC_SQLITE_FILE
end
include_context "rspec setup"
include_context "when initializing"
include_context "when outputting messages"
include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages'
let(:jdbc_jar_env) do
'JDBC_SQLITE_JAR'
@ -18,11 +17,10 @@ describe "logstash-output-jdbc: sqlite", if: ENV['JDBC_SQLITE_JAR'] do
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" ]
'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