Add extra safety to local data retrieval (#78)

* Add extra safety to local data retrieval

Under certain circumstances, retrieving local data from the embedded derby database
can result in an exception being thrown from the Sequel library that is not wrapped in
a `SequelError`. Prior to this commit, that would result in the exception "escaping" the
plugin, causing the plugin (and the pipeline) to crash.

This commit broadens the scope of the rescue to catch any errors that might result from
that call, enabling the appropriate logging and tagging to take place without crashing
the plugin.

Fixes #77

Author: robbavey

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 5.1.3
+  - Improve robustness when handling errors from `sequel` library in jdbc static and streaming
+    filters[#78](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/78)
+
 ## 5.1.2
   -  Fix `prepared_statement_bind_values` in streaming filter to resolve nested event's fields[#76](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/76)
 

lib/logstash/filters/jdbc/lookup.rb

@@ -162,8 +162,11 @@ def retrieve_local_data(local, event, &proc)
       begin
         logger.debug? && logger.debug("Executing Jdbc query", :lookup_id => @id, :statement => query, :parameters => params)
         proc.call(local, query, params, result)
-      rescue ::Sequel::Error => e
-        # all sequel errors are a subclass of this, let all other standard or runtime errors bubble up
+      rescue => e
+        # In theory all exceptions in Sequel should be wrapped in Sequel::Error
+        # However, there are cases where other errors can occur - a `SQLTransactionRollbackException`
+        # may be thrown during `prepareStatement`. Let's handle these cases here, where we can tag and warn
+        # appropriately rather than bubble up and potentially crash the plugin.
         result.failed!
         logger.warn? && logger.warn("Exception when executing Jdbc query", :lookup_id => @id, :exception => e.message, :backtrace => e.backtrace.take(8))
       end

lib/logstash/plugin_mixins/jdbc/jdbc.rb

@@ -119,8 +119,7 @@ def jdbc_connect
           else
             @logger.error("Failed to connect to database. #{@jdbc_pool_timeout} second timeout exceeded. Trying again.")
           end
-        # rescue Java::JavaSql::SQLException, ::Sequel::Error => e
-        rescue ::Sequel::Error => e
+        rescue Java::JavaSql::SQLException, ::Sequel::Error => e
           if retry_attempts <= 0
             log_java_exception(e.cause)
             @logger.error("Unable to connect to database. Tried #{@connection_retry_attempts} times", error_details(e, trace: true))

lib/logstash/plugin_mixins/jdbc_streaming/statement_handler.rb

@@ -38,8 +38,11 @@ def common_cache_lookup(db, event)
         begin
           logger.debug? && logger.debug("Executing JDBC query", :statement => statement, :parameters => params)
           execute_extract_records(db, params, result)
-        rescue ::Sequel::Error => e
-          # all sequel errors are a subclass of this, let all other standard or runtime errors bubble up
+        rescue => e
+          # In theory all exceptions in Sequel should be wrapped in Sequel::Error
+          # However, there are cases where other errors can occur - a `SQLException`may be thrown
+          # during `prepareStatement`. Let's handle these cases here, where we can tag and warn
+          # appropriately rather than bubble up and potentially crash the plugin.
           result.failed!
           logger.warn? && logger.warn("Exception when executing JDBC query", :statement => statement, :parameters => params, :exception => e)
         end

logstash-integration-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-integration-jdbc'
-  s.version         = '5.1.2'
+  s.version         = '5.1.3'
   s.licenses = ['Apache License (2.0)']
   s.summary         = "Integration with JDBC - input and filter plugins"
   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 gemname. This gem is not a stand-alone program"

spec/filters/jdbc/lookup_spec.rb

@@ -249,6 +249,41 @@ module LogStash module Filters module Jdbc
       end
     end
 
+    describe "lookup operations when prepareStatement throws" do
+      let(:local_db) { double("local_db") }
+      let(:lookup_hash) do
+        {
+            "query" => "select * from servers WHERE ip LIKE ? AND os LIKE ?",
+            "prepared_parameters" => ["%%{[ip]}"],
+            "target" => "server",
+            "tag_on_failure" => ["_jdbcstaticfailure_server"]
+        }
+      end
+      let(:event) { LogStash::Event.new()}
+      let(:records) { [{"name" => "ldn-1-23", "rack" => "2:1:6"}] }
+      let(:prepared_statement) { double("prepared_statement")}
+
+      subject(:lookup) { described_class.new(lookup_hash, {}, "lookup-1") }
+
+      before(:each) do
+        allow(local_db).to receive(:prepare).once.and_return(prepared_statement)
+        allow(prepared_statement).to receive(:call).once.and_raise(Java::JavaSql::SQLTransactionRollbackException.new)
+      end
+
+      it "must not be valid" do
+        expect(subject.valid?).to be_falsey
+      end
+
+      it "should tag event as failed" do
+        event.set("ip", "20.20")
+        event.set("os", "MacOS")
+        subject.prepare(local_db)
+        subject.enhance(local_db, event)
+        expect(event.get("tags")).to eq(["_jdbcstaticfailure_server"])
+        expect(event.get("server")).to be_nil
+      end
+    end
+
     describe "validation of target option" do
       let(:lookup_hash) do
         {