Add target option to jdbc input

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 5.1.0
+  - Added `target` option to JDBC input, allowing the row columns to target a specific field instead of being expanded 
+    at the root of the event. This allows the input to play nicer with the Elastic Common Schema when 
+    the input does not follow the schema. [#69](https://github.com/logstash-plugins/logstash-integration-jdbc/issues/69)
+
 ## 5.0.7
   - Feat: try hard to log Java cause (chain) [#62](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/62)
 

docs/input-jdbc.asciidoc

@@ -211,6 +211,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-sql_log_level>> |<<string,string>>, one of `["fatal", "error", "warn", "info", "debug"]`|No
 | <<plugins-{type}s-{plugin}-statement>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-statement_filepath>> |a valid filesystem path|No
+| <<plugins-{type}s-{plugin}-target>> | {logstash-ref}/field-references-deepdive.html[field reference] | No
 | <<plugins-{type}s-{plugin}-tracking_column>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-tracking_column_type>> |<<string,string>>, one of `["numeric", "timestamp"]`|No
 | <<plugins-{type}s-{plugin}-use_column_value>> |<<boolean,boolean>>|No
@@ -535,6 +536,17 @@ with the `parameters` setting.
 
 Path of file containing statement to execute
 
+[id="plugins-{type}s-{plugin}-target"]
+===== `target`
+
+* Value type is {logstash-ref}/field-references-deepdive.html[field reference]
+* There is no default value for this setting.
+
+Without a `target`, events are created from each row column at the root level.
+When the `target` is set to a field reference, the column of each row is placed in the target field instead.
+
+This option can be useful to avoid populating unknown fields when a downstream schema such as ECS is enforced.
+
 [id="plugins-{type}s-{plugin}-tracking_column"]
 ===== `tracking_column`
 

lib/logstash/inputs/jdbc.rb

@@ -3,6 +3,8 @@
 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"
@@ -129,6 +131,11 @@
 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.
@@ -209,6 +216,9 @@ module LogStash module Inputs class Jdbc < LogStash::Inputs::Base
 
   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
@@ -260,6 +270,11 @@ def register
         converters[encoding] = converter
       end
     end
+
+    # target must be populated if ecs_compatibility is not :disabled
+    if @target.nil? && ecs_compatibility != :disabled
+      logger.warn("When ECS compatibility is enabled also target option must be valued")
+    end
   end # def register
 
   # test injection points
@@ -318,7 +333,12 @@ def execute_query(queue)
         ## 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

logstash-integration-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-integration-jdbc'
-  s.version         = '5.0.7'
+  s.version         = '5.1.0'
   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"
@@ -36,6 +36,8 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'tzinfo-data'
   # 3.5 limitation is required for jdbc-static loading schedule
   s.add_runtime_dependency 'rufus-scheduler', '< 3.5'
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~>1.2'
+  s.add_runtime_dependency "logstash-mixin-validator_support", '~> 1.0'
 
   s.add_development_dependency "childprocess"
   s.add_development_dependency 'logstash-devutils'

spec/inputs/jdbc_spec.rb

@@ -328,6 +328,48 @@
 
   end
 
+  context "when using target option" do
+    let(:settings) do
+      {
+        "statement" => "SELECT * from test_table FETCH FIRST 1 ROWS ONLY",
+        "target" => "sub_field"
+      }
+    end
+
+    before do
+      plugin.register
+    end
+
+    after do
+      plugin.stop
+    end
+
+    it "should put all columns under sub-field" do
+      db[:test_table].insert(:num => 1, :custom_time => Time.now.utc, :created_at => Time.now.utc, :string => "Test target option")
+
+      plugin.run(queue)
+
+      expect(queue.size).to eq(1)
+      event = queue.pop
+      expect(event.get("[sub_field][string]")).to eq("Test target option")
+    end
+  end
+
+  context "when using target option is not set and ecs_compatibility is enabled" do
+    let(:settings) do
+      {
+        "statement" => "SELECT * from test_table FETCH FIRST 1 ROWS ONLY",
+        "ecs_compatibility" => :v1
+      }
+    end
+
+    it "should log a warn of missed target usage" do
+      expect(plugin.logger).to receive(:warn).once.with("When ECS compatibility is enabled also target option must be valued")
+
+      plugin.register
+    end
+  end
+
   context "when fetching time data" do
 
     let(:settings) do