jdbc.rb in logstash-integration-jdbc-5.1.0

- old
+ new

@@ -1,10 +1,12 @@
 # encoding: utf-8
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "logstash/plugin_mixins/jdbc/common"
 require "logstash/plugin_mixins/jdbc/jdbc"
+require "logstash/plugin_mixins/ecs_compatibility_support"
+require "logstash/plugin_mixins/validator_support/field_reference_validation_adapter"
 
 # this require_relative returns early unless the JRuby version is between 9.2.0.0 and 9.2.8.0
 require_relative "tzinfo_jruby_patch"
 
 # This plugin was created as a way to ingest data from any database
@@ -127,10 +129,15 @@
 # ---------------------------------------------------------------------------------------------------
 #
 module LogStash module Inputs class Jdbc < LogStash::Inputs::Base
   include LogStash::PluginMixins::Jdbc::Common
   include LogStash::PluginMixins::Jdbc::Jdbc
+  # adds ecs_compatibility config which could be :disabled or :v1
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled,:v1,:v8 => :v1)
+  # adds :field_reference validator adapter
+  extend LogStash::PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
+
   config_name "jdbc"
 
   # If undefined, Logstash will complain, even if codec is unused.
   default :codec, "plain"
 
@@ -207,10 +214,13 @@
 
   config :prepared_statement_name, :validate => :string, :default => ""
 
   config :prepared_statement_bind_values, :validate => :array, :default => []
 
+  # Define the target field to store the loaded columns
+  config :target, :validate => :field_reference, :required => false
+
   attr_reader :database # for test mocking/stubbing
 
   public
 
   def register
@@ -258,10 +268,17 @@
         converter = LogStash::Util::Charset.new(encoding)
         converter.logger = self.logger
         converters[encoding] = converter
       end
     end
+
+    # target must be populated if ecs_compatibility is not :disabled
+    if @target.nil? && ecs_compatibility != :disabled
+      logger.info('ECS compatibility is enabled but no ``target`` option was specified, it is recommended'\
+                  ' to set the option to avoid potential schema conflicts (if your data is ECS compliant or'\
+                  ' non-conflicting feel free to ignore this message)')
+    end
   end # def register
 
   # test injection points
   def set_statement_logger(instance)
     @statement_handler = LogStash::PluginMixins::Jdbc::StatementHandler.build_statement_handler(self, instance)
@@ -316,10 +333,15 @@
     execute_statement do |row|
       if enable_encoding?
         ## do the necessary conversions to string elements
         row = Hash[row.map { |k, v| [k.to_s, convert(k, v)] }]
       end
-      event = LogStash::Event.new(row)
+      if @target
+        event = LogStash::Event.new
+        event.set(@target, row)
+      else
+        event = LogStash::Event.new(row)
+      end
       decorate(event)
       queue << event
     end
     @value_tracker.write
   end