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

Compare commits

base: will:master
Branches Tags
will:master
will:patch-1
will:v2.x
will:v1.4
will:v1.5
will:v5.3.0
will:v5.2.1
will:v5.2.0
will:v5.1.0
will:v5.0.0
will:v0.3.2
will:v0.3.1
will:v0.3.0
will:v0.2.10
will:v0.2.9
will:v0.2.8
will:v0.2.7
will:v0.2.5
will:v0.2.2
will:v0.2.2.rc1
will:v0.2.1
..
compare: will:v5.2.1
Branches Tags
will:patch-1
will:master
will:v2.x
will:v1.4
will:v1.5
will:v5.3.0
will:v5.2.1
will:v5.2.0
will:v5.1.0
will:v5.0.0
will:v0.3.2
will:v0.3.1
will:v0.3.0
will:v0.2.10
will:v0.2.9
will:v0.2.8
will:v0.2.7
will:v0.2.5
will:v0.2.2
will:v0.2.2.rc1
will:v0.2.1

No commits in common. "master" and "v5.2.1" have entirely different histories.
master ... v5.2.1

14 changed files with 61 additions and 214 deletions
Show Stats Expand all files Collapse all files

5
CHANGELOG.md
View File

@ -1,11 +1,6 @@
# Change Log # Change Log
All notable changes to this project will be documented in this file, from 0.2.0. All notable changes to this project will be documented in this file, from 0.2.0.
## [5.3.0] - 2017-11-08
- Adds configuration options `enable_event_as_json_keyword` and `event_as_json_keyword`
- Adds BigDecimal support
- Adds additional logging for debugging purposes (with thanks to @mlkmhd's work)
## [5.2.1] - 2017-04-09 ## [5.2.1] - 2017-04-09
- Adds Array and Hash to_json support for non-sprintf syntax - Adds Array and Hash to_json support for non-sprintf syntax

40
README.md
View File

@ -37,27 +37,25 @@ For development:
## Configuration options ## Configuration options
| Option | Type | Description | Required? | Default | | Option | Type | Description | Required? | Default |
| ------ | ---- | ----------- | --------- | ------- | | ------ | ---- | ----------- | --------- | ------- |
| driver_class | String | Specify a driver class if autoloading fails | No | | | driver_class | String | Specify a driver class if autoloading fails | No | |
| driver_auto_commit | Boolean | If the driver does not support auto commit, you should set this to false | No | True | | driver_auto_commit | Boolean | If the driver does not support auto commit, you should set this to false | No | True |
| driver_jar_path | String | File path to jar file containing your JDBC driver. This is optional, and all JDBC jars may be placed in $LOGSTASH_HOME/vendor/jar/jdbc instead. | No | | | driver_jar_path | String | File path to jar file containing your JDBC driver. This is optional, and all JDBC jars may be placed in $LOGSTASH_HOME/vendor/jar/jdbc instead. | No | |
| connection_string | String | JDBC connection URL | Yes | | | connection_string | String | JDBC connection URL | Yes | |
| connection_test | Boolean | Run a JDBC connection test. Some drivers do not function correctly, and you may need to disable the connection test to supress an error. Cockroach with the postgres JDBC driver is such an example. | No | Yes | | connection_test | Boolean | Run a JDBC connection test. Some drivers do not function correctly, and you may need to disable the connection test to supress an error. Cockroach with the postgres JDBC driver is such an example. | No | Yes |
| connection_test_query | String | Connection test and init query string, required for some JDBC drivers that don't support isValid(). Typically you'd set to this "SELECT 1" | No | | | connection_test_query | String | Connection test and init query string, required for some JDBC drivers that don't support isValid(). Typically you'd set to this "SELECT 1" | No | |
| username | String | JDBC username - this is optional as it may be included in the connection string, for many drivers | No | | | username | String | JDBC username - this is optional as it may be included in the connection string, for many drivers | No | |
| password | String | JDBC password - this is optional as it may be included in the connection string, for many drivers | No | | | password | String | JDBC password - this is optional as it may be included in the connection string, for many drivers | No | |
| statement | Array | An array of strings representing the SQL statement to run. Index 0 is the SQL statement that is prepared, all other array entries are passed in as parameters (in order). A parameter may either be a property of the event (i.e. "@timestamp", or "host") or a formatted string (i.e. "%{host} - %{message}" or "%{message}"). If a key is passed then it will be automatically converted as required for insertion into SQL. If it's a formatted string then it will be passed in verbatim. | Yes | | | statement | Array | An array of strings representing the SQL statement to run. Index 0 is the SQL statement that is prepared, all other array entries are passed in as parameters (in order). A parameter may either be a property of the event (i.e. "@timestamp", or "host") or a formatted string (i.e. "%{host} - %{message}" or "%{message}"). If a key is passed then it will be automatically converted as required for insertion into SQL. If it's a formatted string then it will be passed in verbatim. | Yes | |
| unsafe_statement | Boolean | If yes, the statement is evaluated for event fields - this allows you to use dynamic table names, etc. **This is highly dangerous** and you should **not** use this unless you are 100% sure that the field(s) you are passing in are 100% safe. Failure to do so will result in possible SQL injections. Example statement: [ "insert into %{table_name_field} (column) values(?)", "fieldname" ] | No | False | | unsafe_statement | Boolean | If yes, the statement is evaluated for event fields - this allows you to use dynamic table names, etc. **This is highly dangerous** and you should **not** use this unless you are 100% sure that the field(s) you are passing in are 100% safe. Failure to do so will result in possible SQL injections. Example statement: [ "insert into %{table_name_field} (column) values(?)", "fieldname" ] | No | False |
| max_pool_size | Number | Maximum number of connections to open to the SQL server at any 1 time | No | 5 | | max_pool_size | Number | Maximum number of connections to open to the SQL server at any 1 time | No | 5 |
| connection_timeout | Number | Number of seconds before a SQL connection is closed | No | 2800 | | connection_timeout | Number | Number of seconds before a SQL connection is closed | No | 2800 |
| flush_size | Number | Maximum number of entries to buffer before sending to SQL - if this is reached before idle_flush_time | No | 1000 | | flush_size | Number | Maximum number of entries to buffer before sending to SQL - if this is reached before idle_flush_time | No | 1000 |
| max_flush_exceptions | Number | Number of sequential flushes which cause an exception, before the set of events are discarded. Set to a value less than 1 if you never want it to stop. This should be carefully configured with respect to retry_initial_interval and retry_max_interval, if your SQL server is not highly available | No | 10 | | max_flush_exceptions | Number | Number of sequential flushes which cause an exception, before the set of events are discarded. Set to a value less than 1 if you never want it to stop. This should be carefully configured with respect to retry_initial_interval and retry_max_interval, if your SQL server is not highly available | No | 10 |
| retry_initial_interval | Number | Number of seconds before the initial retry in the event of a failure. On each failure it will be doubled until it reaches retry_max_interval | No | 2 | | retry_initial_interval | Number | Number of seconds before the initial retry in the event of a failure. On each failure it will be doubled until it reaches retry_max_interval | No | 2 |
| retry_max_interval | Number | Maximum number of seconds between each retry | No | 128 | | retry_max_interval | Number | Maximum number of seconds between each retry | No | 128 |
| retry_sql_states | Array of strings | An array of custom SQL state codes you wish to retry until `max_flush_exceptions`. Useful if you're using a JDBC driver which returns retry-able, but non-standard SQL state codes in it's exceptions. | No | [] | | retry_sql_states | Array of strings | An array of custom SQL state codes you wish to retry until `max_flush_exceptions`. Useful if you're using a JDBC driver which returns retry-able, but non-standard SQL state codes in it's exceptions. | No | [] |
| event_as_json_keyword | String | The magic key word that the plugin looks for to convert the entire event into a JSON object. As Logstash does not support this out of the box with it's `sprintf` implementation, you can use whatever this field is set to in the statement parameters | No | @event |
| enable_event_as_json_keyword | Boolean | Enables the magic keyword set in the configuration option `event_as_json_keyword`. Without this enabled the plugin will not convert the `event_as_json_keyword` into JSON encoding of the entire event. | No | False |
## Example configurations ## Example configurations
Example logstash configurations, can now be found in the examples directory. Where possible we try to link every configuration with a tested jar. Example logstash configurations, can now be found in the examples directory. Where possible we try to link every configuration with a tested jar.

18
THANKS.md
View File

@ -1,18 +0,0 @@
logstash-output-jdbc is a project originally created by Karl Southern
(the_angry_angel), but there are a number of people that have contributed
or implemented key features over time. We do our best to keep this list
up-to-date, but you can also have a look at the nice contributor graphs
produced by GitHub: https://github.com/theangryangel/logstash-output-jdbc/graphs/contributors
* [hordijk](https://github.com/hordijk)
* [dmitryakadiamond](https://github.com/dmitryakadiamond)
* [MassimoSporchia](https://github.com/MassimoSporchia)
* [ebuildy](https://github.com/ebuildy)
* [kushtrimjunuzi](https://github.com/kushtrimjunuzi)
* [josemazo](https://github.com/josemazo)
* [aceoliver](https://github.com/aceoliver)
* [roflmao](https://github.com/roflmao)
* [onesuper](https://github.com/onesuper)
* [phr0gz](https://github.com/phr0gz)
* [jMonsinjon](https://github.com/jMonsinjon)
* [mlkmhd](https://github.com/mlkmhd)

17
Vagrantfile vendored
View File

@ -1,19 +1,20 @@
# -*- mode: ruby -*- # -*- mode: ruby -*-
# vi: set ft=ruby : # vi: set ft=ruby :
JRUBY_VERSION = "jruby-1.7"
Vagrant.configure(2) do |config| Vagrant.configure(2) do |config|
config.vm.define "debian" do |deb| config.vm.define "debian" do |deb|
deb.vm.box = 'debian/stretch64' deb.vm.box = 'debian/jessie64'
deb.vm.synced_folder '.', '/vagrant', type: :virtualbox deb.vm.synced_folder '.', '/vagrant', type: :virtualbox
deb.vm.provision 'shell', inline: <<-EOP deb.vm.provision 'shell', inline: <<-EOP
echo "deb http://ftp.debian.org/debian jessie-backports main" | tee --append /etc/apt/sources.list > /dev/null
sed -i 's/main/main contrib non-free/g' /etc/apt/sources.list
apt-get update apt-get update
apt-get install openjdk-8-jre ca-certificates-java git curl -y -q apt-get remove openjdk-7-jre-headless -y -q
curl -sSL https://rvm.io/mpapis.asc | sudo gpg --import - apt-get install git openjdk-8-jre curl -y -q
curl -sSL https://get.rvm.io | bash -s stable --ruby=#{JRUBY_VERSION} 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 usermod -a -G rvm vagrant
EOP EOP
end end
@ -26,8 +27,8 @@ Vagrant.configure(2) do |config|
centos.vm.provision 'shell', inline: <<-EOP centos.vm.provision 'shell', inline: <<-EOP
yum update yum update
yum install java-1.7.0-openjdk yum install java-1.7.0-openjdk
curl -sSL https://rvm.io/mpapis.asc | sudo gpg --import - gpg2 --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
curl -sSL https://get.rvm.io | bash -s stable --ruby=#{JRUBY_VERSION} curl -sSL https://get.rvm.io | bash -s stable --ruby=jruby-1.7
usermod -a -G rvm vagrant usermod -a -G rvm vagrant
EOP EOP
end end

17
examples/sql-server.md
View File

@ -8,25 +8,8 @@ input
} }
output { output {
jdbc { jdbc {
driver_jar_path => '/opt/sqljdbc42.jar'
connection_string => "jdbc:sqlserver://server:1433;databaseName=databasename;user=username;password=password" connection_string => "jdbc:sqlserver://server:1433;databaseName=databasename;user=username;password=password"
statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, ?, ?)", "host", "@timestamp", "message" ] statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, ?, ?)", "host", "@timestamp", "message" ]
} }
}
```
Another example, with mixed static strings and parameters, with thanks to [@MassimoSporchia](https://github.com/MassimoSporchia)
```
input
{
stdin { }
}
output {
jdbc {
driver_jar_path => '/opt/sqljdbc42.jar'
connection_string => "jdbc:sqlserver://server:1433;databaseName=databasename;user=username;password=password"
statement => [ "INSERT INTO log (host, timestamp, message, comment) VALUES(?, ?, ?, 'static string')", "host", "@timestamp", "message" ]
}
} }
``` ```

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

@ -6,7 +6,6 @@ require 'stud/interval'
require 'java' require 'java'
require 'logstash-output-jdbc_jars' require 'logstash-output-jdbc_jars'
require 'json' require 'json'
require 'bigdecimal'
# Write events to a SQL engine, using JDBC. # Write events to a SQL engine, using JDBC.
# #
@ -65,7 +64,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
config :unsafe_statement, validate: :boolean, default: false config :unsafe_statement, validate: :boolean, default: false
# Number of connections in the pool to maintain # Number of connections in the pool to maintain
config :max_pool_size, validate: :number, default: 5 config :max_pool_size, validate: :number, default: 24
# Connection timeout # Connection timeout
config :connection_timeout, validate: :number, default: 10000 config :connection_timeout, validate: :number, default: 10000
@ -100,12 +99,6 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
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, 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 :max_repeat_exceptions_time, obsolete: 'This is no longer required'
config :idle_flush_time, obsolete: 'No longer necessary under Logstash v5' config :idle_flush_time, obsolete: 'No longer necessary under Logstash v5'
# Allows the whole event to be converted to JSON
config :enable_event_as_json_keyword, validate: :boolean, default: false
# The magic key used to convert the whole event to JSON. If you need this, and you have the default in your events, you can use this to change your magic keyword.
config :event_as_json_keyword, validate: :string, default: '@event'
def register def register
@logger.info('JDBC - Starting up') @logger.info('JDBC - Starting up')
@ -208,7 +201,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
begin begin
connection = @pool.getConnection connection = @pool.getConnection
rescue => e rescue => e
log_jdbc_exception(e, true, nil) log_jdbc_exception(e, true)
# If a connection is not available, then the server has gone away # If a connection is not available, then the server has gone away
# We're not counting that towards our retry count. # We're not counting that towards our retry count.
return events, false return events, false
@ -222,7 +215,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
statement = add_statement_event_params(statement, event) if @statement.length > 1 statement = add_statement_event_params(statement, event) if @statement.length > 1
statement.execute statement.execute
rescue => e rescue => e
if retry_exception?(e, event.to_json()) if retry_exception?(e)
events_to_retry.push(event) events_to_retry.push(event)
end end
ensure ensure
@ -269,9 +262,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
def add_statement_event_params(statement, event) def add_statement_event_params(statement, event)
@statement[1..-1].each_with_index do |i, idx| @statement[1..-1].each_with_index do |i, idx|
if @enable_event_as_json_keyword == true and i.is_a? String and i == @event_as_json_keyword if i.is_a? String
value = event.to_json
elsif i.is_a? String
value = event.get(i) value = event.get(i)
if value.nil? and i =~ /%\{/ if value.nil? and i =~ /%\{/
value = event.sprintf(i) value = event.sprintf(i)
@ -299,9 +290,6 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
else else
statement.setInt(idx + 1, value) statement.setInt(idx + 1, value)
end end
when BigDecimal
# TODO: There has to be a better way than this. Find it.
statement.setBigDecimal(idx + 1, java.math.BigDecimal.new(value.to_s))
when Float when Float
statement.setFloat(idx + 1, value) statement.setFloat(idx + 1, value)
when String when String
@ -318,23 +306,20 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
statement statement
end end
def retry_exception?(exception, event) def retry_exception?(exception)
retrying = (exception.respond_to? 'getSQLState' and (RETRYABLE_SQLSTATE_CLASSES.include?(exception.getSQLState.to_s[0,2]) or @retry_sql_states.include?(exception.getSQLState))) retrying = (exception.respond_to? 'getSQLState' and (RETRYABLE_SQLSTATE_CLASSES.include?(exception.getSQLState.to_s[0,2]) or @retry_sql_states.include?(exception.getSQLState)))
log_jdbc_exception(exception, retrying, event) log_jdbc_exception(exception, retrying)
retrying retrying
end end
def log_jdbc_exception(exception, retrying, event) def log_jdbc_exception(exception, retrying)
current_exception = exception current_exception = exception
log_text = 'JDBC - Exception. ' + (retrying ? 'Retrying' : 'Not retrying') log_text = 'JDBC - Exception. ' + (retrying ? 'Retrying' : 'Not retrying') + '.'
log_method = (retrying ? 'warn' : 'error') log_method = (retrying ? 'warn' : 'error')
loop do loop do
# TODO reformat event output so that it only shows the fields necessary. @logger.send(log_method, log_text, :exception => current_exception)
@logger.send(log_method, log_text, :exception => current_exception, :statement => @statement[0], :event => event)
if current_exception.respond_to? 'getNextException' if current_exception.respond_to? 'getNextException'
current_exception = current_exception.getNextException() current_exception = current_exception.getNextException()

6
logstash-output-jdbc.gemspec
View File

@ -1,6 +1,6 @@
Gem::Specification.new do |s| Gem::Specification.new do |s|
s.name = 'logstash-output-jdbc' s.name = 'logstash-output-jdbc'
s.version = '5.3.0' s.version = '5.2.0'
s.licenses = ['Apache License (2.0)'] s.licenses = ['Apache License (2.0)']
s.summary = 'This plugin allows you to output to SQL, via JDBC' 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.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"
@ -26,13 +26,13 @@ Gem::Specification.new do |s|
s.add_runtime_dependency 'stud' s.add_runtime_dependency 'stud'
s.add_runtime_dependency 'logstash-codec-plain' s.add_runtime_dependency 'logstash-codec-plain'
s.requirements << "jar 'com.zaxxer:HikariCP', '2.7.2'" s.requirements << "jar 'com.zaxxer:HikariCP', '2.6.1'"
s.requirements << "jar 'org.apache.logging.log4j:log4j-slf4j-impl', '2.6.2'" s.requirements << "jar 'org.apache.logging.log4j:log4j-slf4j-impl', '2.6.2'"
s.add_development_dependency 'jar-dependencies' s.add_development_dependency 'jar-dependencies'
s.add_development_dependency 'ruby-maven', '~> 3.3' s.add_development_dependency 'ruby-maven', '~> 3.3'
s.add_development_dependency "logstash-devutils", "~> 1.3", ">= 1.3.1" s.add_development_dependency 'logstash-devutils'
s.add_development_dependency 'rubocop', '0.41.2' s.add_development_dependency 'rubocop', '0.41.2'
end end

6
scripts/travis-before_script.sh
View File

@ -1,10 +1,8 @@
#!/bin/bash #!/bin/bash
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 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
sudo apt-get install mysql-server postgresql-client postgresql -qq -y sudo apt-get install mysql-server -qq -y
echo "create database logstash; grant all privileges on logstash.* to 'logstash'@'localhost' identified by 'logstash'; flush privileges;" | sudo -u root mysql echo "create database logstash_output_jdbc_test;" | mysql -u root
echo "create user logstash PASSWORD 'logstash'; create database logstash; grant all privileges on database logstash to logstash;" | sudo -u postgres psql
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 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
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 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
wget http://central.maven.org/maven2/org/postgresql/postgresql/42.1.4/postgresql-42.1.4.jar -O /tmp/postgres.jar

2
scripts/travis-variables.sh
View File

@ -1,5 +1,3 @@
export JDBC_DERBY_JAR=/tmp/derby.jar export JDBC_DERBY_JAR=/tmp/derby.jar
export JDBC_MYSQL_JAR=/tmp/mysql.jar export JDBC_MYSQL_JAR=/tmp/mysql.jar
export JDBC_SQLITE_JAR=/tmp/sqlite.jar export JDBC_SQLITE_JAR=/tmp/sqlite.jar
export JDBC_POSTGRES_JAR=/tmp/postgres.jar

70
spec/jdbc_spec_helper.rb
View File

@ -4,8 +4,6 @@ require 'stud/temporary'
require 'java' require 'java'
require 'securerandom' require 'securerandom'
RSpec::Support::ObjectFormatter.default_instance.max_formatted_output_length = 80000
RSpec.configure do |c| RSpec.configure do |c|
def start_service(name) def start_service(name)
@ -60,57 +58,28 @@ RSpec.shared_context 'when outputting messages' do
"DROP TABLE #{jdbc_test_table}" "DROP TABLE #{jdbc_test_table}"
end end
let(:jdbc_statement_fields) do
[
{db_field: "created_at", db_type: "datetime", db_value: '?', event_field: '@timestamp'},
{db_field: "message", db_type: "varchar(512)", db_value: '?', event_field: 'message'},
{db_field: "message_sprintf", db_type: "varchar(512)", db_value: '?', event_field: 'sprintf-%{message}'},
{db_field: "static_int", db_type: "int", db_value: '?', event_field: 'int'},
{db_field: "static_bigint", db_type: "bigint", db_value: '?', event_field: 'bigint'},
{db_field: "static_float", db_type: "float", db_value: '?', event_field: 'float'},
{db_field: "static_bool", db_type: "boolean", db_value: '?', event_field: 'bool'},
{db_field: "static_bigdec", db_type: "decimal", db_value: '?', event_field: 'bigdec'}
]
end
let(:jdbc_create_table) do let(:jdbc_create_table) do
fields = jdbc_statement_fields.collect { |entry| "#{entry[:db_field]} #{entry[:db_type]} not null" }.join(", ") "CREATE table #{jdbc_test_table} (created_at datetime not null, message varchar(512) not null, message_sprintf varchar(512) not null, static_int int not null, static_bit bit not null, static_bigint bigint not null)"
"CREATE table #{jdbc_test_table} (#{fields})"
end
let(:jdbc_drop_table) do
"DROP table #{jdbc_test_table}"
end end
let(:jdbc_statement) do let(:jdbc_statement) do
fields = jdbc_statement_fields.collect { |entry| "#{entry[:db_field]}" }.join(", ") ["insert into #{jdbc_test_table} (created_at, message, message_sprintf, static_int, static_bit, static_bigint) values(?, ?, ?, ?, ?, ?)", '@timestamp', 'message', 'sprintf-%{message}', 1, true, 4000881632477184]
values = jdbc_statement_fields.collect { |entry| "#{entry[:db_value]}" }.join(", ")
statement = jdbc_statement_fields.collect { |entry| entry[:event_field] }
statement.insert(0, "insert into #{jdbc_test_table} (#{fields}) values(#{values})")
end end
let(:systemd_database_service) do let(:systemd_database_service) do
nil nil
end end
let(:event) do let(:event_fields) do
# TODO: Auto generate fields from jdbc_statement_fields { message: "test-message #{SecureRandom.uuid}" }
LogStash::Event.new({
message: "test-message #{SecureRandom.uuid}",
float: 12.1,
bigint: 4000881632477184,
bool: true,
int: 1,
bigdec: BigDecimal.new("123.123")
})
end end
let(:event) { LogStash::Event.new(event_fields) }
let(:plugin) do let(:plugin) do
# Setup logger # Setup logger
allow(LogStash::Outputs::Jdbc).to receive(:logger).and_return(logger) allow(LogStash::Outputs::Jdbc).to receive(:logger).and_return(logger)
# XXX: Suppress reflection logging. There has to be a better way around this. # XXX: Suppress reflection logging. There has to be a better way around this.
allow(logger).to receive(:debug).with(/config LogStash::/) allow(logger).to receive(:debug).with(/config LogStash::/)
@ -124,12 +93,8 @@ RSpec.shared_context 'when outputting messages' do
output = LogStash::Plugin.lookup('output', 'jdbc').new(jdbc_settings) output = LogStash::Plugin.lookup('output', 'jdbc').new(jdbc_settings)
output.register output.register
output
end
before :each do
# Setup table # Setup table
c = plugin.instance_variable_get(:@pool).getConnection c = output.instance_variable_get(:@pool).getConnection
# Derby doesn't support IF EXISTS. # Derby doesn't support IF EXISTS.
# Seems like the quickest solution. Bleurgh. # Seems like the quickest solution. Bleurgh.
@ -146,16 +111,8 @@ RSpec.shared_context 'when outputting messages' do
stmt.close stmt.close
c.close c.close
end end
end
# Delete table after each output
after :each do
c = plugin.instance_variable_get(:@pool).getConnection
stmt = c.createStatement
stmt.executeUpdate(jdbc_drop_table)
stmt.close
c.close
end end
it 'should save a event' do it 'should save a event' do
@ -163,9 +120,6 @@ RSpec.shared_context 'when outputting messages' do
# Verify the number of items in the output table # Verify the number of items in the output table
c = plugin.instance_variable_get(:@pool).getConnection c = plugin.instance_variable_get(:@pool).getConnection
# TODO replace this simple count with a check of the actual contents
stmt = c.prepareStatement("select count(*) as total from #{jdbc_test_table} where message = ?") stmt = c.prepareStatement("select count(*) as total from #{jdbc_test_table} where message = ?")
stmt.setString(1, event.get('message')) stmt.setString(1, event.get('message'))
rs = stmt.executeQuery rs = stmt.executeQuery
@ -189,7 +143,7 @@ RSpec.shared_context 'when outputting messages' do
end end
it 'it should retry after a connection loss, and log a warning' do it 'it should retry after a connection loss, and log a warning' do
skip "does not run as a service, or known issue with test" if systemd_database_service.nil? skip "does not run as a service" if systemd_database_service.nil?
p = plugin p = plugin
@ -201,12 +155,12 @@ RSpec.shared_context 'when outputting messages' do
# Start a thread to restart the service after the fact. # Start a thread to restart the service after the fact.
t = Thread.new(systemd_database_service) { |systemd_database_service| t = Thread.new(systemd_database_service) { |systemd_database_service|
sleep 20 sleep 20
start_service(systemd_database_service) start_service(systemd_database_service)
} }
t.run t.run
expect(logger).to receive(:warn).at_least(:once).with(/JDBC - Exception. Retrying/, Hash) expect(logger).to receive(:warn).at_least(:once).with(/JDBC - Exception. Retrying/, Hash)
expect { p.multi_receive([event]) }.to_not raise_error expect { p.multi_receive([event]) }.to_not raise_error

16
spec/outputs/jdbc_derby_spec.rb
View File

@ -2,25 +2,17 @@ require_relative '../jdbc_spec_helper'
describe 'logstash-output-jdbc: derby', if: ENV['JDBC_DERBY_JAR'] do describe 'logstash-output-jdbc: derby', if: ENV['JDBC_DERBY_JAR'] do
include_context 'rspec setup' include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages' include_context 'when outputting messages'
let(:jdbc_jar_env) do let(:jdbc_jar_env) do
'JDBC_DERBY_JAR' 'JDBC_DERBY_JAR'
end end
let(:jdbc_statement_fields) do let(:jdbc_create_table) do
[ "CREATE table #{jdbc_test_table} (created_at timestamp not null, message varchar(512) not null, message_sprintf varchar(512) not null, static_int int not null, static_bit boolean not null, static_bigint bigint not null)"
{db_field: "created_at", db_type: "timestamp", db_value: 'CAST(? as timestamp)', event_field: '@timestamp'},
{db_field: "message", db_type: "varchar(512)", db_value: '?', event_field: 'message'},
{db_field: "message_sprintf", db_type: "varchar(512)", db_value: '?', event_field: 'sprintf-%{message}'},
{db_field: "static_int", db_type: "int", db_value: '?', event_field: 'int'},
{db_field: "static_bigint", db_type: "bigint", db_value: '?', event_field: 'bigint'},
{db_field: "static_float", db_type: "float", db_value: '?', event_field: 'float'},
{db_field: "static_bool", db_type: "boolean", db_value: '?', event_field: 'bool'},
{db_field: "static_bigdec", db_type: "decimal", db_value: '?', event_field: 'bigdec'}
]
end end
let(:jdbc_settings) do let(:jdbc_settings) do
{ {
'driver_class' => 'org.apache.derby.jdbc.EmbeddedDriver', 'driver_class' => 'org.apache.derby.jdbc.EmbeddedDriver',

3
spec/outputs/jdbc_mysql_spec.rb
View File

@ -2,6 +2,7 @@ require_relative '../jdbc_spec_helper'
describe 'logstash-output-jdbc: mysql', if: ENV['JDBC_MYSQL_JAR'] do describe 'logstash-output-jdbc: mysql', if: ENV['JDBC_MYSQL_JAR'] do
include_context 'rspec setup' include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages' include_context 'when outputting messages'
let(:jdbc_jar_env) do let(:jdbc_jar_env) do
@ -15,7 +16,7 @@ describe 'logstash-output-jdbc: mysql', if: ENV['JDBC_MYSQL_JAR'] do
let(:jdbc_settings) do let(:jdbc_settings) do
{ {
'driver_class' => 'com.mysql.jdbc.Driver', 'driver_class' => 'com.mysql.jdbc.Driver',
'connection_string' => 'jdbc:mysql://localhost/logstash?user=logstash&password=logstash', 'connection_string' => 'jdbc:mysql://localhost/logstash_output_jdbc_test?user=root',
'driver_jar_path' => ENV[jdbc_jar_env], 'driver_jar_path' => ENV[jdbc_jar_env],
'statement' => jdbc_statement, 'statement' => jdbc_statement,
'max_flush_exceptions' => 1 'max_flush_exceptions' => 1

41
spec/outputs/jdbc_postgres_spec.rb
View File

@ -1,41 +0,0 @@
require_relative '../jdbc_spec_helper'
describe 'logstash-output-jdbc: postgres', if: ENV['JDBC_POSTGRES_JAR'] do
include_context 'rspec setup'
include_context 'when outputting messages'
let(:jdbc_jar_env) do
'JDBC_POSTGRES_JAR'
end
# TODO: Postgres doesnt kill connections fast enough for the test to pass
# Investigate options.
#let(:systemd_database_service) do
# 'postgresql'
#end
let(:jdbc_statement_fields) do
[
{db_field: "created_at", db_type: "timestamp", db_value: 'CAST(? as timestamp)', event_field: '@timestamp'},
{db_field: "message", db_type: "varchar(512)", db_value: '?', event_field: 'message'},
{db_field: "message_sprintf", db_type: "varchar(512)", db_value: '?', event_field: 'sprintf-%{message}'},
{db_field: "static_int", db_type: "int", db_value: '?', event_field: 'int'},
{db_field: "static_bigint", db_type: "bigint", db_value: '?', event_field: 'bigint'},
{db_field: "static_float", db_type: "float", db_value: '?', event_field: 'float'},
{db_field: "static_bool", db_type: "boolean", db_value: '?', event_field: 'bool'},
{db_field: "static_bigdec", db_type: "decimal", db_value: '?', event_field: 'bigdec'}
]
end
let(:jdbc_settings) do
{
'driver_class' => 'org.postgresql.Driver',
'connection_string' => 'jdbc:postgresql://localhost/logstash?user=logstash&password=logstash',
'driver_jar_path' => ENV[jdbc_jar_env],
'statement' => jdbc_statement,
'max_flush_exceptions' => 1
}
end
end

1
spec/outputs/jdbc_sqlite_spec.rb
View File

@ -8,6 +8,7 @@ describe 'logstash-output-jdbc: sqlite', if: ENV['JDBC_SQLITE_JAR'] do
end end
include_context 'rspec setup' include_context 'rspec setup'
include_context 'when initializing'
include_context 'when outputting messages' include_context 'when outputting messages'
let(:jdbc_jar_env) do let(:jdbc_jar_env) do