class LogStash::Filters::JdbcStatic

Public Class Methods

old_validate_value(value, validator)
Alias for: validate_value
validate_value(value, validator) click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 133
def validate_value(value, validator)
  if validator.is_a?(Array) && validator.first.respond_to?(:find_validation_errors)
    validation_errors = validator.first.find_validation_errors(value)
    unless validation_errors.nil?
      return false, validation_errors
    end
  elsif validator.respond_to?(:find_validation_errors)
    validation_errors = validator.find_validation_errors(value)
    unless validation_errors.nil?
      return false, validation_errors
    end
  else
    return old_validate_value(value, validator)
  end
  [true, value]
end
Also aliased as: old_validate_value

Public Instance Methods

close() click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 163
def close
  @scheduler.stop if @scheduler
  @parsed_loaders.each(&:close)
  @processor.close
end
filter(event) click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 158
def filter(event)
  enhancement_states = @processor.enhance(event)
  filter_matched(event) if enhancement_states.all?
end
loader_runner() click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 169
def loader_runner
  # use for test verification
  @loader_runner
end
register() click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 153
def register
  prepare_data_dir
  prepare_runner
end

Private Instance Methods

add_plugin_configs(options) click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 225
def add_plugin_configs(options)
  if @jdbc_driver_library
    options["jdbc_driver_library"] = @jdbc_driver_library
  end
  if @jdbc_driver_class
    options["jdbc_driver_class"] = @jdbc_driver_class
  end
  if @jdbc_connection_string
    options["jdbc_connection_string"] = @jdbc_connection_string
  end
  if @jdbc_user
    options["jdbc_user"] = @jdbc_user
  end
  if @jdbc_password
    options["jdbc_password"] = @jdbc_password
  end
  if @staging_directory
    options["staging_directory"] = @staging_directory
  end
end
global_lookup_options(options = Hash.new) click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 211
def global_lookup_options(options = Hash.new)
  if @tag_on_failure && !@tag_on_failure.empty? && !options.key?("tag_on_failure")
    options["tag_on_failure"] = @tag_on_failure
  end
  if @tag_on_default_use && !@tag_on_default_use.empty? && !options.key?("tag_on_default_use")
    options["tag_on_default_use"] = @tag_on_default_use
  end
  options["lookup_jdbc_driver_class"] = @lookup_jdbc_driver_class
  options["lookup_jdbc_driver_library"] = @lookup_jdbc_driver_library
  options["lookup_jdbc_connection_string"] = @lookup_jdbc_connection_string
  options["ecs_compatibility"] = ecs_compatibility
  options
end
prepare_data_dir() click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 176
def prepare_data_dir
  # later, when local persistent databases are allowed set this property to LS_HOME/data/jdbc-static/
  # must take multi-pipelines into account and more than one config using the same jdbc-static settings
  java.lang.System.setProperty("derby.system.home", ENV["HOME"])
  logger.info("derby.system.home is: #{java.lang.System.getProperty("derby.system.home")}")
end
prepare_runner() click to toggle source
# File lib/logstash/filters/jdbc_static.rb, line 183
def prepare_runner
  @parsed_loaders = @loaders.map do |options|
    add_plugin_configs(options)
    loader = Jdbc::Loader.new(options)
    loader.build_remote_db
    loader
  end
  runner_args = [@parsed_loaders, @local_db_objects]
  @processor = Jdbc::LookupProcessor.new(@local_lookups, global_lookup_options)
  runner_args.unshift(@processor.local)
  if @loader_schedule
    args = []
    @loader_runner = Jdbc::RepeatingLoadRunner.new(*runner_args)
    @loader_runner.initial_load
    cronline = Jdbc::LoaderSchedule.new(@loader_schedule)
    cronline.to_log_string.tap do |msg|
      logger.info("Scheduler operations: #{msg}") unless msg.empty?
    end
    logger.info("Scheduler scan for work frequency is: #{cronline.schedule_frequency}")
    rufus_args = {:max_work_threads => 1, :frequency => cronline.schedule_frequency}
    @scheduler = Rufus::Scheduler.new(rufus_args)
    @scheduler.cron(cronline.loader_schedule, @loader_runner)
  else
    @loader_runner = Jdbc::SingleLoadRunner.new(*runner_args)
    @loader_runner.initial_load
  end
end