Module: LogStash::PluginMixins::Jdbc::Jdbc
- Included in:
- Inputs::Jdbc
- Defined in:
- lib/logstash/plugin_mixins/jdbc/jdbc.rb
Class Method Summary collapse
-
.included(base) ⇒ Object
This method is called when someone includes this module.
Instance Method Summary collapse
- #close_jdbc_connection ⇒ Object
- #execute_statement ⇒ Object
- #get_column_value(row) ⇒ Object
- #prepare_jdbc_connection ⇒ Object
- #setup_jdbc_config ⇒ Object
Class Method Details
.included(base) ⇒ Object
This method is called when someone includes this module
18 19 20 21 22 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 18 def self.included(base) # Add these methods to the 'base' given. base.extend(self) base.setup_jdbc_config end |
Instance Method Details
#close_jdbc_connection ⇒ Object
236 237 238 239 240 241 242 243 244 245 246 247 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 236 def close_jdbc_connection begin # 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 @connection_lock.lock @database.disconnect if @database rescue => e @logger.warn("Failed to close connection", :exception => e) ensure @connection_lock.unlock end end |
#execute_statement ⇒ Object
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 250 def execute_statement success = false @connection_lock.lock open_jdbc_connection begin 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::DatabaseConnectionError, Sequel::DatabaseError, Java::JavaSql::SQLException => e @logger.warn("Exception when executing JDBC query", :exception => e) else @value_tracker.set_value(sql_last_value) ensure close_jdbc_connection @connection_lock.unlock end return success end |
#get_column_value(row) ⇒ Object
274 275 276 277 278 279 280 281 282 283 284 285 286 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 274 def get_column_value(row) if !row.has_key?(@tracking_column.to_sym) if !@tracking_column_warning_sent @logger.warn("tracking_column not found in dataset.", :tracking_column => @tracking_column) @tracking_column_warning_sent = true end # If we can't find the tracking column, return the current value in the ivar @value_tracker.value else # Otherwise send the updated tracking column row[@tracking_column.to_sym] end end |
#prepare_jdbc_connection ⇒ Object
230 231 232 233 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 230 def prepare_jdbc_connection @connection_lock = ReentrantLock.new @drivers_loaded = Concurrent::AtomicBoolean.new end |
#setup_jdbc_config ⇒ Object
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
# File 'lib/logstash/plugin_mixins/jdbc/jdbc.rb', line 26 def setup_jdbc_config # JDBC driver library path to third party driver library. In case of multiple libraries being # required you can pass them separated by a comma. # # If not provided, Plugin will look for the driver class in the Logstash Java classpath. config :jdbc_driver_library, :validate => :string # JDBC driver class to load, for exmaple, "org.apache.derby.jdbc.ClientDriver" # NB per https://github.com/logstash-plugins/logstash-input-jdbc/issues/43 if you are using # the Oracle JDBC driver (ojdbc6.jar) the correct `jdbc_driver_class` is `"Java::oracle.jdbc.driver.OracleDriver"` config :jdbc_driver_class, :validate => :string, :required => true # JDBC connection string config :jdbc_connection_string, :validate => :string, :required => true # JDBC user config :jdbc_user, :validate => :string, :required => true # JDBC password config :jdbc_password, :validate => :password # JDBC password filename config :jdbc_password_filepath, :validate => :path # JDBC enable paging # # This will cause a sql statement to be broken up into multiple queries. # Each query will use limits and offsets to collectively retrieve the full # result-set. The limit size is set with `jdbc_page_size`. # # Be aware that ordering is not guaranteed between queries. config :jdbc_paging_enabled, :validate => :boolean, :default => false # JDBC page size config :jdbc_page_size, :validate => :number, :default => 100000 # JDBC fetch size. if not provided, respective driver's default will be used config :jdbc_fetch_size, :validate => :number # Connection pool configuration. # Validate connection before use. config :jdbc_validate_connection, :validate => :boolean, :default => false # Connection pool configuration. # How often to validate a connection (in seconds) config :jdbc_validation_timeout, :validate => :number, :default => 3600 # Connection pool configuration. # The amount of seconds to wait to acquire a connection before raising a PoolTimeoutError (default 5) config :jdbc_pool_timeout, :validate => :number, :default => 5 # Timezone conversion. # SQL does not allow for timezone data in timestamp fields. This plugin will automatically # convert your SQL timestamp fields to Logstash timestamps, in relative UTC time in ISO8601 format. # # Using this setting will manually assign a specified timezone offset, instead # of using the timezone setting of the local machine. You must use a canonical # timezone, *America/Denver*, for example. config :jdbc_default_timezone, :validate => :string # General/Vendor-specific Sequel configuration options. # # An example of an optional connection pool configuration # max_connections - The maximum number of connections the connection pool # # examples of vendor-specific options can be found in this # documentation page: https://github.com/jeremyevans/sequel/blob/master/doc/opening_databases.rdoc config :sequel_opts, :validate => :hash, :default => {} # Log level at which to log SQL queries, the accepted values are the common ones fatal, error, warn, # info and debug. The default value is info. config :sql_log_level, :validate => [ "fatal", "error", "warn", "info", "debug" ], :default => "info" # Maximum number of times to try connecting to database config :connection_retry_attempts, :validate => :number, :default => 1 # Number of seconds to sleep between connection attempts config :connection_retry_attempts_wait_time, :validate => :number, :default => 0.5 # give users the ability to force Sequel application side into using local timezone config :plugin_timezone, :validate => ["local", "utc"], :default => "utc" end |