Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c5aeae1b02 | ||
|
|
a7d5a2e623 | ||
|
|
3a64a22ac4 | ||
|
|
c4b62769b9 | ||
|
|
b9e5f64d40 | ||
|
|
c0e358aafb | ||
|
|
442ddf16eb | ||
|
|
4e7985dafd |
21
CHANGELOG.md
Normal file
21
CHANGELOG.md
Normal file
@@ -0,0 +1,21 @@
|
||||
# Change Log
|
||||
All notable changes to this project will be documented in this file, from 0.2.0.
|
||||
|
||||
## [0.2.5] - 2016-04-11
|
||||
### Added
|
||||
- Basic tests running against DerbyDB
|
||||
- Fix for converting Logstash::Timestamp to iso8601 from @hordijk
|
||||
|
||||
## [0.2.4] - 2016-04-07
|
||||
- Documentation fixes from @hordijk
|
||||
|
||||
## [0.2.3] - 2016-02-16
|
||||
- Bug fixes
|
||||
|
||||
## [0.2.2] - 2015-12-30
|
||||
- Bug fixes
|
||||
|
||||
## [0.2.1] - 2015-12-22
|
||||
- Support for connection pooling support added through HikariCP
|
||||
- Support for unsafe statement handling (allowing dynamic queries)
|
||||
- Altered exception handling to now count sequential flushes with exceptions thrown
|
||||
23
README.md
23
README.md
@@ -11,34 +11,35 @@ If you do find this works for a JDBC driver not listed, let me know and provide
|
||||
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.
|
||||
|
||||
## Headlines
|
||||
- Support for connection pooling added in 0.2.0
|
||||
- Support for unsafe statement handling (allowing dynamic queries) in 0.2.0
|
||||
- Altered exception handling to now count sequential flushes with exceptions thrown in 0.2.0
|
||||
## ChangeLog
|
||||
See CHANGELOG.md
|
||||
|
||||
## Versions
|
||||
Released versions are are tagged as of v0.2.1, and available via rubygems.
|
||||
|
||||
For development:
|
||||
- See master branch for logstash v2+
|
||||
- See master branch for logstash v2
|
||||
- See v1.5 branch for logstash v1.5
|
||||
- See v1.4 branch for logstash 1.4
|
||||
|
||||
## Installation
|
||||
- Run `bin/plugin install logstash-output-jdbc` in your logstash installation directory
|
||||
- Now either:
|
||||
- Use driver_path in your configuraton to specify a path to your jar file
|
||||
- Use driver_jar_path in your configuraton to specify a path to your jar file
|
||||
- Or:
|
||||
- Create the directory vendor/jar/jdbc in your logstash installation (`mkdir -p vendor/jar/jdbc/`)
|
||||
- Add JDBC jar files to vendor/jar/jdbc in your logstash installation
|
||||
- And then configure (examples below)
|
||||
|
||||
## Running tests
|
||||
Assuming valid JDBC jar, and jruby is setup and installed, and you have issued `jruby -S bundle install` in the development directory
|
||||
- `SQL_JAR=path/to/your.jar jruby -S bundle exec rspec`
|
||||
If you need to provide username and password you may do this via the environment variables `SQL_USERNAME` and `SQL_PASSWORD`.
|
||||
At this time tests only run against Derby, in an in-memory database.
|
||||
Acceptance tests for individual database engines will be added over time.
|
||||
|
||||
Tests are not yet 100% complete.
|
||||
Assuming valid jruby is installed
|
||||
- First time, issue `jruby -S bundle install` to install dependencies
|
||||
- Next, download Derby jar from https://db.apache.org/derby/
|
||||
- Run the tests `JDBC_DERBY_JAR=path/to/derby.jar jruby -S rspec`
|
||||
- Optionally add the `JDBC_DEBUG=1` env variable to add logging to stdout
|
||||
|
||||
## Configuration options
|
||||
|
||||
@@ -46,7 +47,7 @@ Tests are not yet 100% complete.
|
||||
| ------ | ---- | ----------- | --------- | ------- |
|
||||
| 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_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 | |
|
||||
| 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 | |
|
||||
|
||||
@@ -44,7 +44,7 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
|
||||
config :max_pool_size, :validate => :number, :default => 5
|
||||
|
||||
# Connection timeout
|
||||
config :connection_timeout, :validate => :number, :default => 2800
|
||||
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
|
||||
@@ -156,9 +156,11 @@ class LogStash::Outputs::Jdbc < LogStash::Outputs::Base
|
||||
@pool.setMaximumPoolSize(@max_pool_size)
|
||||
@pool.setConnectionTimeout(@connection_timeout)
|
||||
|
||||
validate_connection_timeout = (@connection_timeout / 1000) / 2
|
||||
|
||||
# Test connection
|
||||
test_connection = @pool.getConnection()
|
||||
unless test_connection.isValid(10)
|
||||
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.")
|
||||
end
|
||||
test_connection.close()
|
||||
@@ -248,9 +250,12 @@ 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]
|
||||
when Time, LogStash::Timestamp
|
||||
when Time
|
||||
# Most reliable solution, cross JDBC driver
|
||||
statement.setString(idx + 1, event[i].iso8601())
|
||||
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
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
Gem::Specification.new do |s|
|
||||
s.name = 'logstash-output-jdbc'
|
||||
s.version = "0.2.2"
|
||||
s.version = "0.2.5"
|
||||
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"
|
||||
|
||||
@@ -4,92 +4,93 @@ require "stud/temporary"
|
||||
require "java"
|
||||
|
||||
describe LogStash::Outputs::Jdbc do
|
||||
def fetch_log_table_rowcount
|
||||
# sleep for a second to let the flush happen
|
||||
sleep 1
|
||||
|
||||
stmt = @sql.createStatement()
|
||||
rs = stmt.executeQuery("select count(*) as total from log")
|
||||
count = 0
|
||||
while rs.next()
|
||||
count = rs.getInt("total")
|
||||
end
|
||||
stmt.close()
|
||||
|
||||
return count
|
||||
end
|
||||
|
||||
let(:base_settings) { {
|
||||
"driver_jar_path" => @driver_jar_path,
|
||||
"connection_string" => @test_connection_string,
|
||||
"username" => ENV['SQL_USERNAME'],
|
||||
"password" => ENV['SQL_PASSWORD'],
|
||||
"statement" => [ "insert into log (message) values(?)", "message" ],
|
||||
"max_pool_size" => 1,
|
||||
"flush_size" => 1,
|
||||
"max_flush_exceptions" => 1
|
||||
} }
|
||||
let(:test_settings) { {} }
|
||||
let(:plugin) { LogStash::Outputs::Jdbc.new(base_settings.merge(test_settings)) }
|
||||
let(:event_fields) { { "message" => "This is a message!" } }
|
||||
let(:event) { LogStash::Event.new(event_fields) }
|
||||
|
||||
before(:all) do
|
||||
@driver_jar_path = File.absolute_path(ENV['SQL_JAR'])
|
||||
@test_db_path = File.join(Stud::Temporary.directory, "test.db")
|
||||
@test_connection_string = "jdbc:sqlite:#{@test_db_path}"
|
||||
|
||||
require @driver_jar_path
|
||||
|
||||
@sql = java.sql.DriverManager.get_connection(@test_connection_string, ENV['SQL_USERNAME'].to_s, ENV['SQL_PASSWORD'].to_s)
|
||||
stmt = @sql.createStatement()
|
||||
stmt.executeUpdate("CREATE table log (host text, timestamp datetime, message text);")
|
||||
stmt.close()
|
||||
end
|
||||
|
||||
before(:each) do
|
||||
stmt = @sql.createStatement()
|
||||
stmt.executeUpdate("delete from log")
|
||||
stmt.close()
|
||||
end
|
||||
|
||||
after(:all) do
|
||||
File.unlink(@test_db_path)
|
||||
Dir.rmdir(File.dirname(@test_db_path))
|
||||
end
|
||||
|
||||
describe "safe statement" do
|
||||
it "should register without errors" do
|
||||
expect { plugin.register }.to_not raise_error
|
||||
end
|
||||
|
||||
it "receive event, without error" do
|
||||
plugin.register
|
||||
expect { plugin.receive(event) }.to_not raise_error
|
||||
|
||||
expect(fetch_log_table_rowcount).to eq(1)
|
||||
end
|
||||
|
||||
end
|
||||
|
||||
describe "unsafe statement" do
|
||||
let(:event_fields) {
|
||||
{ "message" => "This is a message!", "table" => "log" }
|
||||
let(:derby_settings) 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.
|
||||
# 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" ]
|
||||
}
|
||||
let(:test_settings) { {
|
||||
"statement" => [ "insert into %{table} (message) values(?)", "message" ],
|
||||
"unsafe_statement" => true
|
||||
} }
|
||||
|
||||
it "should register without errors" do
|
||||
expect { plugin.register }.to_not raise_error
|
||||
end
|
||||
end
|
||||
|
||||
it "receive event, without error" do
|
||||
plugin.register
|
||||
plugin.receive(event)
|
||||
expect(fetch_log_table_rowcount).to eq(1)
|
||||
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
|
||||
expect {
|
||||
LogStash::Plugin.lookup("output", "jdbc").new()
|
||||
}.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
|
||||
|
||||
Reference in New Issue
Block a user