Feat: event target => namespace support (ECS) (#41)

We're introducing a `target => ...` configuration option for the codec, to aid ECS support.
When `target` isn't set in ECS mode, we do the usual info log message.

Author: kares

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 3.1.0
+  - Feat: event `target => namespace` support (ECS) [#41](https://github.com/logstash-plugins/logstash-codec-json_lines/pull/41)
+  - Refactor: dropped support for old Logstash versions (< 6.0)
+
 ## 3.0.6
   - Support flush method, see https://github.com/logstash-plugins/logstash-codec-json_lines/pull/35
 

Gemfile

@@ -9,7 +9,3 @@ if Dir.exist?(logstash_path) && use_logstash_source
   gem 'logstash-core', :path => "#{logstash_path}/logstash-core"
   gem 'logstash-core-plugin-api', :path => "#{logstash_path}/logstash-core-plugin-api"
 end
-
-if RUBY_VERSION == "1.9.3"
-  gem 'rake', '12.2.1'
-end

docs/index.asciidoc

@@ -36,6 +36,8 @@ Therefore this codec cannot work with line oriented inputs.
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-charset>> |<<string,string>>, one of `["ASCII-8BIT", "UTF-8", "US-ASCII", "Big5", "Big5-HKSCS", "Big5-UAO", "CP949", "Emacs-Mule", "EUC-JP", "EUC-KR", "EUC-TW", "GB2312", "GB18030", "GBK", "ISO-8859-1", "ISO-8859-2", "ISO-8859-3", "ISO-8859-4", "ISO-8859-5", "ISO-8859-6", "ISO-8859-7", "ISO-8859-8", "ISO-8859-9", "ISO-8859-10", "ISO-8859-11", "ISO-8859-13", "ISO-8859-14", "ISO-8859-15", "ISO-8859-16", "KOI8-R", "KOI8-U", "Shift_JIS", "UTF-16BE", "UTF-16LE", "UTF-32BE", "UTF-32LE", "Windows-31J", "Windows-1250", "Windows-1251", "Windows-1252", "IBM437", "IBM737", "IBM775", "CP850", "IBM852", "CP852", "IBM855", "CP855", "IBM857", "IBM860", "IBM861", "IBM862", "IBM863", "IBM864", "IBM865", "IBM866", "IBM869", "Windows-1258", "GB1988", "macCentEuro", "macCroatian", "macCyrillic", "macGreek", "macIceland", "macRoman", "macRomania", "macThai", "macTurkish", "macUkraine", "CP950", "CP951", "IBM037", "stateless-ISO-2022-JP", "eucJP-ms", "CP51932", "EUC-JIS-2004", "GB12345", "ISO-2022-JP", "ISO-2022-JP-2", "CP50220", "CP50221", "Windows-1256", "Windows-1253", "Windows-1255", "Windows-1254", "TIS-620", "Windows-874", "Windows-1257", "MacJapanese", "UTF-7", "UTF8-MAC", "UTF-16", "UTF-32", "UTF8-DoCoMo", "SJIS-DoCoMo", "UTF8-KDDI", "SJIS-KDDI", "ISO-2022-JP-KDDI", "stateless-ISO-2022-JP-KDDI", "UTF8-SoftBank", "SJIS-SoftBank", "BINARY", "CP437", "CP737", "CP775", "IBM850", "CP857", "CP860", "CP861", "CP862", "CP863", "CP864", "CP865", "CP866", "CP869", "CP1258", "Big5-HKSCS:2008", "ebcdic-cp-us", "eucJP", "euc-jp-ms", "EUC-JISX0213", "eucKR", "eucTW", "EUC-CN", "eucCN", "CP936", "ISO2022-JP", "ISO2022-JP2", "ISO8859-1", "ISO8859-2", "ISO8859-3", "ISO8859-4", "ISO8859-5", "ISO8859-6", "CP1256", "ISO8859-7", "CP1253", "ISO8859-8", "CP1255", "ISO8859-9", "CP1254", "ISO8859-10", "ISO8859-11", "CP874", "ISO8859-13", "CP1257", "ISO8859-14", "ISO8859-15", "ISO8859-16", "CP878", "MacJapan", "ASCII", "ANSI_X3.4-1968", "646", "CP65000", "CP65001", "UTF-8-MAC", "UTF-8-HFS", "UCS-2BE", "UCS-4BE", "UCS-4LE", "CP932", "csWindows31J", "SJIS", "PCK", "CP1250", "CP1251", "CP1252", "external", "locale"]`|No
 | <<plugins-{type}s-{plugin}-delimiter>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-target>> |<<string,string>>|No
 |=======================================================================
 
 &nbsp;
@@ -64,4 +66,35 @@ For nxlog users, you'll want to set this to `CP1252`
 
 Change the delimiter that separates lines
 
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+  * Value type is <<string,string>>
+  * Supported values are:
+    ** `disabled`: does not use ECS-compatible field names
+    ** `v1`, `v8`: Elastic Common Schema compliant behavior (warns when `target` isn't set)
+  * Default value depends on which version of Logstash is running:
+    ** When Logstash provides a `pipeline.ecs_compatibility` setting, its value is used as the default
+    ** Otherwise, the default value is `disabled`
+
+Controls this plugin's compatibility with the {ecs-ref}[Elastic Common Schema (ECS)].
+
+[id="plugins-{type}s-{plugin}-target"]
+===== `target`
+
+  * Value type is <<string,string>>
+  * There is no default value for this setting.
+
+Define the target field for placing the parsed data. If this setting is not
+set, the JSON data will be stored at the root (top level) of the event.
+
+For example, if you want data to be put under the `document` field:
+[source,ruby]
+    input {
+      http {
+        codec => json_lines {
+          target => "[document]"
+        }
+      }
+    }
 

lib/logstash/codecs/json_lines.rb

@@ -3,6 +3,11 @@
 require "logstash/util/charset"
 require "logstash/util/buftok"
 require "logstash/json"
+require 'logstash/plugin_mixins/ecs_compatibility_support'
+require 'logstash/plugin_mixins/ecs_compatibility_support/target_check'
+require 'logstash/plugin_mixins/validator_support/field_reference_validation_adapter'
+require 'logstash/plugin_mixins/event_support/event_factory_adapter'
+require 'logstash/plugin_mixins/event_support/from_json_helper'
 
 # This codec will decode streamed JSON that is newline delimited.
 # Encoding will emit a single JSON string ending in a `@delimiter`
@@ -12,6 +17,15 @@
 # terminated lines. The file input will produce a line string without a newline.
 # Therefore this codec cannot work with line oriented inputs.
 class LogStash::Codecs::JSONLines < LogStash::Codecs::Base
+
+  include LogStash::PluginMixins::ECSCompatibilitySupport
+  include LogStash::PluginMixins::ECSCompatibilitySupport::TargetCheck
+
+  extend LogStash::PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
+
+  include LogStash::PluginMixins::EventSupport::EventFactoryAdapter
+  include LogStash::PluginMixins::EventSupport::FromJsonHelper
+
   config_name "json_lines"
 
   # The character encoding used in this codec. Examples include `UTF-8` and
@@ -28,6 +42,11 @@ class LogStash::Codecs::JSONLines < LogStash::Codecs::Base
   # Change the delimiter that separates lines
   config :delimiter, :validate => :string, :default => "\n"
 
+  # Defines a target field for placing decoded fields.
+  # If this setting is omitted, data gets stored at the root (top level) of the event.
+  # The target is only relevant while decoding data into a new event.
+  config :target, :validate => :field_reference
+
   public
 
   def register
@@ -38,7 +57,7 @@ def register
 
   def decode(data, &block)
     @buffer.extract(data).each do |line|
-      parse(@converter.convert(line), &block)
+      parse_json(@converter.convert(line), &block)
     end
   end
 
@@ -51,31 +70,21 @@ def encode(event)
   def flush(&block)
     remainder = @buffer.flush
     if !remainder.empty?
-      parse(@converter.convert(remainder), &block)
+      parse_json(@converter.convert(remainder), &block)
     end
   end
 
   private
 
-  # from_json_parse uses the Event#from_json method to deserialize and directly produce events
-  def from_json_parse(json, &block)
-    LogStash::Event.from_json(json).each { |event| yield event }
-  rescue LogStash::Json::ParserError => e
-    @logger.warn("JSON parse error, original data now in message field", :error => e, :data => json)
-    yield LogStash::Event.new("message" => json, "tags" => ["_jsonparsefailure"])
+  def parse_json(json)
+    events_from_json(json, targeted_event_factory).each { |event| yield event }
+  rescue => e
+    @logger.warn("JSON parse error, original data now in message field", message: e.message, exception: e.class, data: json)
+    yield parse_json_error_event(json)
   end
 
-  # legacy_parse uses the LogStash::Json class to deserialize json
-  def legacy_parse(json, &block)
-    # ignore empty/blank lines which LogStash::Json#load returns as nil
-    o = LogStash::Json.load(json)
-    yield(LogStash::Event.new(o)) if o
-  rescue LogStash::Json::ParserError => e
-    @logger.warn("JSON parse error, original data now in message field", :error => e, :data => json)
-    yield LogStash::Event.new("message" => json, "tags" => ["_jsonparsefailure"])
+  def parse_json_error_event(json)
+    event_factory.new_event("message" => json, "tags" => ["_jsonparsefailure"])
   end
 
-  # keep compatibility with all v2.x distributions. only in 2.3 will the Event#from_json method be introduced
-  # and we need to keep compatibility for all v2 releases.
-  alias_method :parse, LogStash::Event.respond_to?(:from_json) ? :from_json_parse : :legacy_parse
 end

logstash-codec-json_lines.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-codec-json_lines'
-  s.version         = '3.0.6'
+  s.version         = '3.1.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads and writes newline-delimited JSON"
   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"
@@ -19,11 +19,15 @@ Gem::Specification.new do |s|
   # Special flag to let us know this is actually a logstash plugin
   s.metadata = { "logstash_plugin" => "true", "logstash_group" => "codec" }
 
+  s.required_ruby_version = '>= 2.3' # Event.from_json exists at least since LS 5.6
+
   # Gem dependencies
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~> 1.3'
+  s.add_runtime_dependency 'logstash-mixin-event_support', '~> 1.0', '>= 1.0.1'
+  s.add_runtime_dependency 'logstash-mixin-validator_support', '~> 1.0'
 
-  s.add_runtime_dependency 'logstash-codec-line', '>= 2.1.0'
-
+  s.add_development_dependency 'logstash-codec-line', '>= 2.1.0'
   s.add_development_dependency 'logstash-devutils'
   s.add_development_dependency 'insist'
 end

spec/codecs/json_lines_spec.rb

@@ -4,8 +4,9 @@
 require "logstash/event"
 require "logstash/json"
 require "insist"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
 
-describe LogStash::Codecs::JSONLines do
+describe LogStash::Codecs::JSONLines, :ecs_compatibility_support do
 
   let(:codec_options) { {} }
 
@@ -185,29 +186,40 @@
     end
   end
 
-  end
+  ecs_compatibility_matrix(:disabled, :v1, :v8 => :v1) do
 
-  context "forcing legacy parsing" do
-    it_behaves_like :codec do
-      subject do
-        # register method is called in the constructor
-        LogStash::Codecs::JSONLines.new(codec_options)
+    before(:each) do
+      allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
+    end
+
+    context 'with target' do
+      let(:input) do
+        %{{"field": "value1"}
+{"field": 2.0}
+}
       end
 
-      before(:each) do
-        # stub codec parse method to force use of the legacy parser.
-        # this is very implementation specific but I am not sure how
-        # this can be tested otherwise.
-        allow(subject).to receive(:parse) do |line, &block|
-          subject.send(:legacy_parse, line, &block)
+      let(:codec_options) { super().merge "target" => 'foo' }
+
+      let(:collector) { Array.new }
+
+      it 'should generate two events' do
+        subject.decode(input) do |event|
+          collector.push(event)
         end
+        expect(collector.size).to eq(2)
+        expect(collector[0].include?('field')).to be false
+        expect(collector[0].get('foo')).to eql 'field' => 'value1'
+        expect(collector[1].include?('field')).to be false
+        expect(collector[1].get('foo')).to eql 'field' => 2.0
       end
     end
+
+  end
+
   end
 
   context "default parser choice" do
-    # here we cannot force the use of the Event#from_json since if this test is run in the
-    # legacy context (no Java Event) it will fail but if in the new context, it will be picked up.
     it_behaves_like :codec do
       subject do
         # register method is called in the constructor