Session handling reprise (#130)

* Revert "Switch pattern of usage Sequel to leverage the DB connection pool (#119)"

This reverts functionality from commit deeb4b2.

* input: avoid caching db connections in statement handler

* input: fix issue where retried query could prevent pipeline shutdown

The use of the `retry` keyword when a lock is acquired in the `begin...` block
and released in the `ensure` block will cause the lock to be acquired more
times than released whenever the block is retried. Because only one thread
ever executes the queries, this isn't a problem at runtime, but can cause the
plugin to stall during shutdown when a separate thread attempts to acquire
the lock for cleanup.

* input: retain fail-in-register if driver cannot be loaded

Author: yaauie

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.5.2
+  - FIX: the input plugin's prior behaviour of opening a new database connection for each scheduled run (removed in `v5.4.1`) is restored, ensuring that infrequently-run schedules do not hold open connections to their databases indefinitely, _without_ reintroducing the leak [#130](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/130)
+
 ## 5.5.1
   - Document `statement_retry_attempts` and `statement_retry_attempts_wait_time` options
 

lib/logstash/inputs/jdbc.rb

@@ -261,6 +261,8 @@ def register
       end
     end
 
+    prepare_jdbc_connection
+
     if @use_column_value
       # Raise an error if @use_column_value is true, but no @tracking_column is set
       if @tracking_column.nil?
@@ -303,20 +305,7 @@ def register
         converters[encoding] = converter
       end
     end
-
     load_driver
-    begin
-      open_jdbc_connection
-    rescue Sequel::DatabaseConnectionError,
-           Sequel::DatabaseError,
-           Sequel::InvalidValue,
-           Java::JavaSql::SQLException => e
-      details = { exception: e.class, message: e.message }
-      details[:cause] = e.cause.inspect if e.cause
-      details[:backtrace] = e.backtrace if @logger.debug?
-      @logger.warn("Exception when executing JDBC query", details)
-      raise(LogStash::ConfigurationError, "Can't create a connection pool to the database")
-    end
   end # def register
 
   # test injection points

lib/logstash/plugin_mixins/jdbc/jdbc.rb

@@ -3,15 +3,17 @@
 require "logstash/config/mixin"
 require "time"
 require "date"
+require "thread" # Monitor
 require_relative "value_tracking"
 require_relative "timezone_proxy"
 require_relative "statement_handler"
 require_relative "value_handler"
 
-java_import java.util.concurrent.locks.ReentrantLock
-
 # Tentative of abstracting JDBC logic to a mixin
 # for potential reuse in other plugins (input/output)
+#
+# CAUTION: implementation of this "potential reuse" module is
+#          VERY tightly-coupled with the JDBC Input's implementation.
 module LogStash  module PluginMixins module Jdbc
   module Jdbc
     include LogStash::PluginMixins::Jdbc::ValueHandler
@@ -159,82 +161,94 @@ def log_java_exception(e)
     end
 
     def open_jdbc_connection
-      # at this point driver is already loaded
-      Sequel.application_timezone = @plugin_timezone.to_sym
-
-      @database = jdbc_connect()
-      @database.extension(:pagination)
-      if @jdbc_default_timezone
-        @database.extension(:named_timezones)
-        @database.timezone = TimezoneProxy.load(@jdbc_default_timezone)
-      end
-      if @jdbc_validate_connection
-        @database.extension(:connection_validator)
-        @database.pool.connection_validation_timeout = @jdbc_validation_timeout
-      end
-      @database.fetch_size = @jdbc_fetch_size unless @jdbc_fetch_size.nil?
-      begin
-        @database.test_connection
-      rescue Java::JavaSql::SQLException => e
-        @logger.warn("Failed test_connection with java.sql.SQLException.", :exception => e)
-      rescue Sequel::DatabaseConnectionError => e
-        @logger.warn("Failed test_connection.", :exception => e)
-        #TODO return false and let the plugin raise a LogStash::ConfigurationError
-        raise e
-      end
+      @connection_lock.synchronize do
+        # at this point driver is already loaded
+        Sequel.application_timezone = @plugin_timezone.to_sym
+
+        @database = jdbc_connect()
+        @database.extension(:pagination)
+        if @jdbc_default_timezone
+          @database.extension(:named_timezones)
+          @database.timezone = TimezoneProxy.load(@jdbc_default_timezone)
+        end
+        if @jdbc_validate_connection
+          @database.extension(:connection_validator)
+          @database.pool.connection_validation_timeout = @jdbc_validation_timeout
+        end
+        @database.fetch_size = @jdbc_fetch_size unless @jdbc_fetch_size.nil?
+        begin
+          @database.test_connection
+        rescue Java::JavaSql::SQLException => e
+          @logger.warn("Failed test_connection with java.sql.SQLException.", :exception => e)
+        rescue Sequel::DatabaseConnectionError => e
+          @logger.warn("Failed test_connection.", :exception => e)
+          #TODO return false and let the plugin raise a LogStash::ConfigurationError
+          raise e
+        end
 
-      @database.sql_log_level = @sql_log_level.to_sym
-      @database.logger = @logger
+        @database.sql_log_level = @sql_log_level.to_sym
+        @database.logger = @logger
 
-      @database.extension :identifier_mangling
+        @database.extension :identifier_mangling
 
-      if @lowercase_column_names
-        @database.identifier_output_method = :downcase
-      else
-        @database.identifier_output_method = :to_s
+        if @lowercase_column_names
+          @database.identifier_output_method = :downcase
+        else
+          @database.identifier_output_method = :to_s
+        end
       end
     end
 
+    public
+    def prepare_jdbc_connection
+      @connection_lock = Monitor.new # aka ReentrantLock
+    end
+
     public
     def close_jdbc_connection
-      begin
+      @connection_lock.synchronize do
         # pipeline restarts can also close the jdbc connection, block until the current executing statement is finished to avoid leaking connections
         # connections in use won't really get closed
         @database.disconnect if @database
-      rescue => e
-        @logger.warn("Failed to close connection", :exception => e)
       end
+    rescue => e
+      @logger.warn("Failed to close connection", :exception => e)
     end
 
     public
-    def execute_statement
+    def execute_statement(&result_handler)
       success = false
       retry_attempts = @statement_retry_attempts
 
-      begin
-        retry_attempts -= 1
-        sql_last_value = @use_column_value ? @value_tracker.value : Time.now.utc
-        @tracking_column_warning_sent = false
-        @statement_handler.perform_query(@database, @value_tracker.value) do |row|
-          sql_last_value = get_column_value(row) if @use_column_value
-          yield extract_values_from(row)
-        end
-        success = true
-      rescue Sequel::Error, Java::JavaSql::SQLException => e
-        details = { exception: e.class, message: e.message }
-        details[:cause] = e.cause.inspect if e.cause
-        details[:backtrace] = e.backtrace if @logger.debug?
-        @logger.warn("Exception when executing JDBC query", details)
-
-        if retry_attempts == 0
-          @logger.error("Unable to execute statement. Tried #{@statement_retry_attempts} times.")
+      @connection_lock.synchronize do
+        begin
+          retry_attempts -= 1
+          open_jdbc_connection
+          sql_last_value = @use_column_value ? @value_tracker.value : Time.now.utc
+          @tracking_column_warning_sent = false
+          @statement_handler.perform_query(@database, @value_tracker.value) do |row|
+            sql_last_value = get_column_value(row) if @use_column_value
+            yield extract_values_from(row)
+          end
+          success = true
+        rescue Sequel::Error, Java::JavaSql::SQLException => e
+          details = { exception: e.class, message: e.message }
+          details[:cause] = e.cause.inspect if e.cause
+          details[:backtrace] = e.backtrace if @logger.debug?
+          @logger.warn("Exception when executing JDBC query", details)
+
+          if retry_attempts == 0
+            @logger.error("Unable to execute statement. Tried #{@statement_retry_attempts} times.")
+          else
+            @logger.error("Unable to execute statement. Trying again.")
+            sleep(@statement_retry_attempts_wait_time)
+            retry
+          end
         else
-          @logger.error("Unable to execute statement. Trying again.")
-          sleep(@statement_retry_attempts_wait_time)
-          retry
+          @value_tracker.set_value(sql_last_value)
+        ensure
+          close_jdbc_connection
         end
-      else
-        @value_tracker.set_value(sql_last_value)
       end
 
       return success

lib/logstash/plugin_mixins/jdbc/statement_handler.rb

@@ -124,14 +124,14 @@ def perform_query(db, sql_last_value)
   end
 
   class PreparedStatementHandler < StatementHandler
-    attr_reader :name, :bind_values_array, :statement_prepared, :prepared, :parameters
+    attr_reader :name
 
     def initialize(plugin)
       super(plugin)
       @name = plugin.prepared_statement_name.to_sym
-      @bind_values_array = plugin.prepared_statement_bind_values
-      @parameters = plugin.parameters
-      @statement_prepared = Concurrent::AtomicBoolean.new(false)
+
+      @positional_bind_mapping =  create_positional_bind_mapping(plugin.prepared_statement_bind_values).freeze
+      @positional_bind_placeholders = @positional_bind_mapping.keys.map { |v| :"$#{v}" }.freeze
     end
 
     # Performs the query, ignoring our pagination settings, yielding once per row of data
@@ -148,41 +148,28 @@ def perform_query(db, sql_last_value)
     private
 
     def build_query(db, sql_last_value)
-      @parameters = create_bind_values_hash
-      if statement_prepared.false?
-        prepended = parameters.keys.map{|v| v.to_s.prepend("$").to_sym}
-        @prepared = db[statement, *prepended].prepare(:select, name)
-        statement_prepared.make_true
-      end
       # under the scheduler the Sequel database instance is recreated each time
       # so the previous prepared statements are lost, add back
-      if db.prepared_statement(name).nil?
-        db.set_prepared_statement(name, prepared)
-      end
-      bind_value_sql_last_value(sql_last_value)
-      begin
-        db.call(name, parameters)
-      rescue => e
-        # clear the statement prepared flag - the statement may be closed by this
-        # time.
-        statement_prepared.make_false
-        raise e
-      end
+      prepared = db.prepared_statement(name)
+      prepared ||= db[statement, *positional_bind_placeholders].prepare(:select, name)
+
+      prepared.call(positional_bind_mapping(sql_last_value))
     end
 
-    def create_bind_values_hash
+    def create_positional_bind_mapping(bind_values_array)
       hash = {}
       bind_values_array.each_with_index {|v,i| hash[:"p#{i}"] = v}
       hash
     end
 
-    def bind_value_sql_last_value(sql_last_value)
-      parameters.keys.each do |key|
-        value = parameters[key]
-        if value == ":sql_last_value"
-          parameters[key] = sql_last_value
-        end
+    def positional_bind_mapping(sql_last_value)
+      @positional_bind_mapping.transform_values do |value|
+        value == ":sql_last_value" ? sql_last_value : value
       end
     end
+
+    def positional_bind_placeholders
+      @positional_bind_mapping.keys.map { |v| :"$#{v}" }.freeze
+    end
   end
 end end end

logstash-integration-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-integration-jdbc'
-  s.version         = '5.5.1'
+  s.version         = '5.5.2'
   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/inputs/integration/integ_spec.rb

@@ -113,21 +113,25 @@
     end
 
     it "log warning msg when plugin run" do
+      plugin.register
       expect( plugin ).to receive(:log_java_exception)
       expect(plugin.logger).to receive(:warn).once.with("Exception when executing JDBC query",
                                                         hash_including(:message => instance_of(String)))
-      expect{ plugin.register }.to raise_error(::LogStash::ConfigurationError)
+      q = Queue.new
+      expect{ plugin.run(q) }.not_to raise_error
     end
 
     it "should log (native) Java driver error" do
+      plugin.register
       expect( org.apache.logging.log4j.LogManager ).to receive(:getLogger).and_wrap_original do |m, *args|
         logger = m.call(*args)
         expect( logger ).to receive(:error) do |_, e|
           expect( e ).to be_a org.postgresql.util.PSQLException
         end.and_call_original
         logger
       end
-      expect{ plugin.register }.to raise_error(::LogStash::ConfigurationError)
+      q = Queue.new
+      expect{ plugin.run(q) }.not_to raise_error
     end
   end
 end

spec/inputs/jdbc_spec.rb

@@ -1283,6 +1283,7 @@
       plugin.register
       plugin.run(queue)
       db = plugin.instance_variable_get(:@database)
+      expect(db.pool).to be_a_kind_of(::Sequel::ThreadedConnectionPool) # pries into internal details
       expect(db.pool.instance_variable_get(:@timeout)).to eq(0)
       expect(db.pool.instance_variable_get(:@max_size)).to eq(1)
 
@@ -1296,11 +1297,12 @@
 
     it "should log error message" do
       allow(Sequel).to receive(:connect).and_raise(Sequel::PoolTimeout)
-      expect(plugin.logger).to receive(:error).with("Failed to connect to database. 0 second timeout exceeded. Tried 1 times.")
-      expect do
-        plugin.register
-        plugin.run(queue)
-      end.to raise_error(Sequel::PoolTimeout)
+      allow(plugin.logger).to receive(:error)
+
+      plugin.register
+      plugin.run(queue)
+
+      expect(plugin.logger).to have_received(:error).with("Failed to connect to database. 0 second timeout exceeded. Tried 1 times.")
     end
   end
 
@@ -1376,12 +1378,13 @@
       mixin_settings['connection_retry_attempts'] = 2
       mixin_settings['jdbc_pool_timeout'] = 0
       allow(Sequel).to receive(:connect).and_raise(Sequel::PoolTimeout)
-      expect(plugin.logger).to receive(:error).with("Failed to connect to database. 0 second timeout exceeded. Trying again.")
-      expect(plugin.logger).to receive(:error).with("Failed to connect to database. 0 second timeout exceeded. Tried 2 times.")
-      expect do
-        plugin.register
-        plugin.run(queue)
-      end.to raise_error(Sequel::PoolTimeout)
+      allow(plugin.logger).to receive(:error)
+
+      plugin.register
+      plugin.run(queue)
+
+      expect(plugin.logger).to have_received(:error).with("Failed to connect to database. 0 second timeout exceeded. Trying again.")
+      expect(plugin.logger).to have_received(:error).with("Failed to connect to database. 0 second timeout exceeded. Tried 2 times.")
     end
 
     it "should not fail when passed a non-positive value" do
@@ -1642,16 +1645,12 @@ def load_derby_version
       { "statement" => "SELECT * from types_table", "jdbc_driver_library" => invalid_driver_jar_path }
     end
 
-    after do
-      plugin.stop
-    end
-
-    it "raise a loading error" do
+    it "raise a loading error during #register" do
       expect(File.exists?(invalid_driver_jar_path)).to be true
       expect(FileTest.readable?(invalid_driver_jar_path)).to be false
 
-      expect { plugin.register }.
-          to raise_error(LogStash::PluginLoadingError, /unable to load .*? from :jdbc_driver_library, file not readable/)
+      expect { plugin.register }
+        .to raise_error(LogStash::PluginLoadingError, /unable to load .*? from :jdbc_driver_library, file not readable/)
     end
   end