Module: LogStash::PluginMixins::Jdbc

Included in:
Inputs::Jdbc
Defined in:
lib/logstash/plugin_mixins/jdbc.rb

Overview

Tentative of abstracting JDBC logic to a mixin for potential reuse in other plugins (input/output)

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(base) ⇒ Object

This method is called when someone includes this module



12
13
14
15
16
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 12

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 



196
197
198
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 196

def close_jdbc_connection
  @database.disconnect if @database
end

#execute_statement(statement, parameters) ⇒ Object 



201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 201

def execute_statement(statement, parameters)
  success = false
  begin
    parameters = symbolized_params(parameters)
    query = @database[statement, parameters]
    sql_last_value = @use_column_value ? @sql_last_value : Time.now.utc
    @tracking_column_warning_sent = false
    @logger.debug? and @logger.debug("Executing JDBC query", :statement => statement, :parameters => parameters, :count => query.count)

    if @jdbc_paging_enabled
      query.each_page(@jdbc_page_size) do |paged_dataset|
        paged_dataset.each do |row|
          sql_last_value = get_column_value(row) if @use_column_value
          if @tracking_column_type=="timestamp" and @use_column_value and sql_last_value.is_a?(DateTime)
            sql_last_value=Time.parse(sql_last_value.to_s) # Coerce the timestamp to a `Time`
          end
          yield extract_values_from(row)
        end
      end
    else
      query.each do |row|
        sql_last_value = get_column_value(row) if @use_column_value
        if @tracking_column_type=="timestamp" and @use_column_value and sql_last_value.is_a?(DateTime)
          sql_last_value=Time.parse(sql_last_value.to_s) # Coerce the timestamp to a `Time`
        end
        yield extract_values_from(row)
      end
    end
    success = true
  rescue Sequel::DatabaseConnectionError, Sequel::DatabaseError => e
    @logger.warn("Exception when executing JDBC query", :exception => e)
  else
    @sql_last_value = sql_last_value
  end
  return success
end

#get_column_value(row) ⇒ Object 



239
240
241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 239

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
    @sql_last_value
  else
    # Otherwise send the updated tracking column
    row[@tracking_column.to_sym]
  end
end

#prepare_jdbc_connection ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 142

def prepare_jdbc_connection
  require "java"
  require "sequel"
  require "sequel/adapters/jdbc"
  load_drivers(@jdbc_driver_library.split(",")) if @jdbc_driver_library

  begin
    Sequel::JDBC.load_driver(@jdbc_driver_class)
  rescue Sequel::AdapterNotFound => e
    message = if @jdbc_driver_library.nil?
                ":jdbc_driver_library is not set, are you sure you included
                the proper driver client libraries in your classpath?"
              else
                "Are you sure you've included the correct jdbc driver in :jdbc_driver_library?"
              end
    raise LogStash::ConfigurationError, "#{e}. #{message}"
  end
  @database = jdbc_connect()
  @database.extension(:pagination)
  if @jdbc_default_timezone
    @database.extension(:named_timezones)
    @database.timezone = @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 Sequel::DatabaseConnectionError => 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
  if @lowercase_column_names
    @database.identifier_output_method = :downcase
  else
    @database.identifier_output_method = :to_s
  end
  if @use_column_value
    case @tracking_column_type
      when "numeric"
        @sql_last_value = 0
      when "timestamp"
        @sql_last_value = Time.at(0).utc
    end
  else
    @sql_last_value = Time.at(0).utc
  end
end

#setup_jdbc_config ⇒ Object 



20
21
22
23
24
25
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
 
# File 'lib/logstash/plugin_mixins/jdbc.rb', line 20

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
end