From 2ef60259e9098e64e8e17dbd80f912dcc615ecb0 Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Tue, 25 May 2021 18:53:03 +0200
Subject: [PATCH 1/6] add ECS support

---
 CHANGELOG.md                |   3 +
 VERSION                     |   2 +-
 docs/index.asciidoc         |  23 ++
 lib/logstash/filters/csv.rb |  12 +
 logstash-filter-csv.gemspec |   2 +-
 spec/filters/csv_spec.rb    | 575 ++++++++++++++++++------------------
 6 files changed, 330 insertions(+), 287 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6329393..8af4a59 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 3.1.0
+  - Add ECS support 
+
 ## 3.0.11
   - [DOC] Fixed formatting to improve readability [#84](https://github.com/logstash-plugins/logstash-filter-csv/pull/84)
 
diff --git a/VERSION b/VERSION
index 778bf95..fd2a018 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-3.0.11
+3.1.0
diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 5c37b97..724fca0 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -24,6 +24,11 @@ The CSV filter takes an event field containing CSV data, parses it,
 and stores it as individual fields with optionally-specified field names.
 This filter can parse data with any separator, not just commas.
 
+[id="plugins-{type}s-{plugin}-ecs_metadata"]
+==== Event Metadata and the Elastic Common Schema (ECS)
+When ECS compatibility is disabled, the value is stored in the root level.
+When ECS is enabled, the value is stored in the `target` field.
+
 [id="plugins-{type}s-{plugin}-options"]
 ==== Csv Filter Configuration Options
 
@@ -36,6 +41,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-autogenerate_column_names>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-columns>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-convert>> |<<hash,hash>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
 | <<plugins-{type}s-{plugin}-quote_char>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-separator>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-skip_empty_columns>> |<<boolean,boolean>>|No
@@ -104,6 +110,19 @@ Example:
       }
     }
 
+[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`: uses the value in `target` as field name
+
+Controls this plugin's compatibility with the
+{ecs-ref}[Elastic Common Schema (ECS)].
+See <<plugins-{type}s-{plugin}-ecs_metadata>> for detailed information.
+The value of this setting affects the _default_ value of <<plugins-{type}s-{plugin}-???>>
+
 [id="plugins-{type}s-{plugin}-quote_char"]
 ===== `quote_char` 
 
@@ -179,6 +198,10 @@ data structure.
 Define target field for placing the data.
 Defaults to writing to the root of the event.
 
+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}-common-options"]
diff --git a/lib/logstash/filters/csv.rb b/lib/logstash/filters/csv.rb
index cc5f785..5a6d2c3 100644
--- a/lib/logstash/filters/csv.rb
+++ b/lib/logstash/filters/csv.rb
@@ -1,6 +1,7 @@
 # encoding: utf-8
 require "logstash/filters/base"
 require "logstash/namespace"
+require "logstash/plugin_mixins/ecs_compatibility_support"
 
 require "csv"
 
@@ -8,6 +9,8 @@
 # and stores it as individual fields (can optionally specify the names).
 # This filter can also parse data with any separator, not just commas.
 class LogStash::Filters::CSV < LogStash::Filters::Base
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
+
   config_name "csv"
 
   # The CSV data in the value of the `source` field will be expanded into a
@@ -102,6 +105,15 @@ class LogStash::Filters::CSV < LogStash::Filters::Base
   CONVERTERS.default = lambda {|v| v}
   CONVERTERS.freeze
 
+  def initialize(params)
+    super
+    if ecs_compatibility != :disabled && @target.nil?
+      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
     # validate conversion types to be the valid ones.
     bad_types = @convert.values.select do |type|
diff --git a/logstash-filter-csv.gemspec b/logstash-filter-csv.gemspec
index 2f410b2..6f934d7 100644
--- a/logstash-filter-csv.gemspec
+++ b/logstash-filter-csv.gemspec
@@ -23,7 +23,7 @@ Gem::Specification.new do |s|
 
   # Gem dependencies
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
-
   s.add_development_dependency 'logstash-devutils'
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~>1.2'
 end
 
diff --git a/spec/filters/csv_spec.rb b/spec/filters/csv_spec.rb
index 2fa0450..aef153e 100644
--- a/spec/filters/csv_spec.rb
+++ b/spec/filters/csv_spec.rb
@@ -1,272 +1,260 @@
 # encoding: utf-8
 require "logstash/devutils/rspec/spec_helper"
 require "logstash/filters/csv"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
 
-describe LogStash::Filters::CSV do
+describe LogStash::Filters::CSV, :ecs_compatibility_support, :aggregate_failures do
+  ecs_compatibility_matrix(:disabled, :v1, :v8 => :v1) do |ecs_select|
 
-  subject(:plugin) { LogStash::Filters::CSV.new(config) }
-  let(:config) { Hash.new }
+    subject(:plugin) { LogStash::Filters::CSV.new(config) }
+    let(:config) { Hash.new }
 
-  let(:doc)   { "" }
-  let(:event) { LogStash::Event.new("message" => doc) }
+    let(:doc)   { "" }
+    let(:event) { LogStash::Event.new("message" => doc) }
 
-  describe "registration" do
+    describe "registration" do
 
-    context "when using invalid data types" do
-      let(:config) do
-        { "convert" => { "custom1" => "integer", "custom3" => "wrong_type" },
-          "columns" => ["custom1", "custom2", "custom3"] }
-      end
-
-      it "should register" do
-        input = LogStash::Plugin.lookup("filter", "csv").new(config)
-        expect {input.register}.to raise_error(LogStash::ConfigurationError)
-      end
-    end
-  end
-
-  describe "receive" do
-
-    before(:each) do
-      plugin.register
-    end
-
-    describe "all defaults" do
-
-      let(:config) { Hash.new }
-
-      let(:doc) { "big,bird,sesame street" }
-
-      it "extract all the values" do
-        plugin.filter(event)
-        expect(event.get("column1")).to eq("big")
-        expect(event.get("column2")).to eq("bird")
-        expect(event.get("column3")).to eq("sesame street")
-      end
-
-      it "should not mutate the source field" do
-        plugin.filter(event)
-        expect(event.get("message")).to be_kind_of(String)
-      end
-    end
-
-    describe "empty message" do
-      let(:doc) { "" }
-
-      let(:config) do
-        { "skip_empty_rows" => true }
-      end
-
-      it "skips empty rows" do
-        plugin.filter(event)
-        expect(event.get("tags")).to include("_csvskippedemptyfield") 
-        expect(event).not_to be_cancelled 
-      end
-    end
-
-    describe "custom separator" do
-      let(:doc) { "big,bird;sesame street" }
+      context "when using invalid data types" do
+        let(:config) do
+          { "convert" => { "custom1" => "integer", "custom3" => "wrong_type" },
+            "columns" => ["custom1", "custom2", "custom3"] }
+        end
 
-      let(:config) do
-        { "separator" => ";" }
-      end
-      it "extract all the values" do
-        plugin.filter(event)
-        expect(event.get("column1")).to eq("big,bird")
-        expect(event.get("column2")).to eq("sesame street")
+        it "should register" do
+          input = LogStash::Plugin.lookup("filter", "csv").new(config)
+          expect {input.register}.to raise_error(LogStash::ConfigurationError)
+        end
       end
     end
 
-    describe "quote char" do
-      let(:doc) { "big,bird,'sesame street'" }
+    describe "receive" do
 
-      let(:config) do
-        { "quote_char" => "'"}
+      before(:each) do
+        allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
+        plugin.register
       end
 
-      it "extract all the values" do
-        plugin.filter(event)
-        expect(event.get("column1")).to eq("big")
-        expect(event.get("column2")).to eq("bird")
-        expect(event.get("column3")).to eq("sesame street")
-      end
+      describe "all defaults" do
 
-      context "using the default one" do
-        let(:doc) { 'big,bird,"sesame, street"' }
         let(:config) { Hash.new }
 
+        let(:doc) { "big,bird,sesame street" }
+
         it "extract all the values" do
           plugin.filter(event)
           expect(event.get("column1")).to eq("big")
           expect(event.get("column2")).to eq("bird")
-          expect(event.get("column3")).to eq("sesame, street")
-        end
-      end
-
-      context "using a null" do
-        let(:doc) { 'big,bird,"sesame" street' }
-        let(:config) do
-          { "quote_char" => "\x00" }
+          expect(event.get("column3")).to eq("sesame street")
         end
 
-        it "extract all the values" do
+        it "should not mutate the source field" do
           plugin.filter(event)
-          expect(event.get("column1")).to eq("big")
-          expect(event.get("column2")).to eq("bird")
-          expect(event.get("column3")).to eq('"sesame" street')
+          expect(event.get("message")).to be_kind_of(String)
         end
       end
-      
-      context "using a null as read from config" do
-        let(:doc) { 'big,bird,"sesame" street' }
+
+      describe "empty message" do
+        let(:doc) { "" }
+
         let(:config) do
-          { "quote_char" => "\\x00" }
+          { "skip_empty_rows" => true }
         end
 
-        it "extract all the values" do
+        it "skips empty rows" do
           plugin.filter(event)
-          expect(event.get("column1")).to eq("big")
-          expect(event.get("column2")).to eq("bird")
-          expect(event.get("column3")).to eq('"sesame" street')
+          expect(event.get("tags")).to include("_csvskippedemptyfield")
+          expect(event).not_to be_cancelled
         end
       end
-    end
-
-    describe "given column names" do
-      let(:doc)    { "big,bird,sesame street" }
-      let(:config) do
-        { "columns" => ["first", "last", "address" ] }
-      end
-
-      it "extract all the values" do
-        plugin.filter(event)
-        expect(event.get("first")).to eq("big")
-        expect(event.get("last")).to eq("bird")
-        expect(event.get("address")).to eq("sesame street")
-      end
 
-      context "parse csv without autogeneration of names" do
+      describe "custom separator" do
+        let(:doc) { "big,bird;sesame street" }
 
-        let(:doc)    { "val1,val2,val3" }
         let(:config) do
-          {  "autogenerate_column_names" => false,
-             "columns" => ["custom1", "custom2"] }
+          { "separator" => ";" }
         end
-
         it "extract all the values" do
           plugin.filter(event)
-          expect(event.get("custom1")).to eq("val1")
-          expect(event.get("custom2")).to eq("val2")
-          expect(event.get("column3")).to be_falsey
+          expect(event.get("column1")).to eq("big,bird")
+          expect(event.get("column2")).to eq("sesame street")
         end
       end
 
+      describe "quote char" do
+        let(:doc) { "big,bird,'sesame street'" }
 
-      context "parse csv and skip the header" do
-        
-        let(:doc)    { "first_column,second_column,third_column" }
         let(:config) do
-          { "skip_header" => true,
-            "columns" => ["first_column", "second_column", "third_column"] }
+          { "quote_char" => "'"}
         end
 
-        it "expects the event to be cancelled" do
+        it "extract all the values" do
           plugin.filter(event)
-          expect(event).to be_cancelled
+          expect(event.get("column1")).to eq("big")
+          expect(event.get("column2")).to eq("bird")
+          expect(event.get("column3")).to eq("sesame street")
         end
-      end
 
-      context "parse csv skipping empty columns" do
+        context "using the default one" do
+          let(:doc) { 'big,bird,"sesame, street"' }
+          let(:config) { Hash.new }
 
-        let(:doc)    { "val1,,val3" }
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("column1")).to eq("big")
+            expect(event.get("column2")).to eq("bird")
+            expect(event.get("column3")).to eq("sesame, street")
+          end
+        end
 
-        let(:config) do
-          { "skip_empty_columns" => true,
-            "source" => "datafield",
-            "columns" => ["custom1", "custom2", "custom3"] }
+        context "using a null" do
+          let(:doc) { 'big,bird,"sesame" street' }
+          let(:config) do
+            { "quote_char" => "\x00" }
+          end
+
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("column1")).to eq("big")
+            expect(event.get("column2")).to eq("bird")
+            expect(event.get("column3")).to eq('"sesame" street')
+          end
         end
 
-        let(:event) { LogStash::Event.new("datafield" => doc) }
+        context "using a null as read from config" do
+          let(:doc) { 'big,bird,"sesame" street' }
+          let(:config) do
+            { "quote_char" => "\\x00" }
+          end
 
-        it "extract all the values" do
-          plugin.filter(event)
-          expect(event.get("custom1")).to eq("val1")
-          expect(event.get("custom2")).to be_falsey
-          expect(event.get("custom3")).to eq("val3")
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("column1")).to eq("big")
+            expect(event.get("column2")).to eq("bird")
+            expect(event.get("column3")).to eq('"sesame" street')
+          end
         end
       end
 
-      context "parse csv with more data than defined" do
-        let(:doc)    { "val1,val2,val3" }
+      describe "given column names" do
+        let(:doc)    { "big,bird,sesame street" }
         let(:config) do
-          { "columns" => ["custom1", "custom2"] }
+          { "columns" => ["first", "last", "address" ] }
         end
 
         it "extract all the values" do
           plugin.filter(event)
-          expect(event.get("custom1")).to eq("val1")
-          expect(event.get("custom2")).to eq("val2")
-          expect(event.get("column3")).to eq("val3")
+          expect(event.get("first")).to eq("big")
+          expect(event.get("last")).to eq("bird")
+          expect(event.get("address")).to eq("sesame street")
         end
-      end
 
-      context "parse csv from a given source" do
-        let(:doc)    { "val1,val2,val3" }
-        let(:config) do
-          { "source"  => "datafield",
-            "columns" => ["custom1", "custom2", "custom3"] }
+        context "parse csv without autogeneration of names" do
+
+          let(:doc)    { "val1,val2,val3" }
+          let(:config) do
+            {  "autogenerate_column_names" => false,
+               "columns" => ["custom1", "custom2"] }
+          end
+
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("custom1")).to eq("val1")
+            expect(event.get("custom2")).to eq("val2")
+            expect(event.get("column3")).to be_falsey
+          end
         end
-        let(:event) { LogStash::Event.new("datafield" => doc) }
 
-        it "extract all the values" do
-          plugin.filter(event)
-          expect(event.get("custom1")).to eq("val1")
-          expect(event.get("custom2")).to eq("val2")
-          expect(event.get("custom3")).to eq("val3")
+
+        context "parse csv and skip the header" do
+
+          let(:doc)    { "first_column,second_column,third_column" }
+          let(:config) do
+            { "skip_header" => true,
+              "columns" => ["first_column", "second_column", "third_column"] }
+          end
+
+          it "expects the event to be cancelled" do
+            plugin.filter(event)
+            expect(event).to be_cancelled
+          end
         end
-      end
 
-      context "that use [@metadata]" do
-        let(:metadata_field) { "[@metadata][one]" }
-        let(:config) do
-          {
-            "columns" => [ metadata_field, "foo" ]
-          }
+        context "parse csv skipping empty columns" do
+
+          let(:doc)    { "val1,,val3" }
+
+          let(:config) do
+            { "skip_empty_columns" => true,
+              "source" => "datafield",
+              "columns" => ["custom1", "custom2", "custom3"] }
+          end
+
+          let(:event) { LogStash::Event.new("datafield" => doc) }
+
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("custom1")).to eq("val1")
+            expect(event.get("custom2")).to be_falsey
+            expect(event.get("custom3")).to eq("val3")
+          end
         end
 
-        let(:event) { LogStash::Event.new("message" => "hello,world") }
+        context "parse csv with more data than defined" do
+          let(:doc)    { "val1,val2,val3" }
+          let(:config) do
+            { "columns" => ["custom1", "custom2"] }
+          end
 
-        before do
-          plugin.filter(event)
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("custom1")).to eq("val1")
+            expect(event.get("custom2")).to eq("val2")
+            expect(event.get("column3")).to eq("val3")
+          end
         end
 
-        it "should work correctly" do
-          expect(event.get(metadata_field)).to eq("hello")
+        context "parse csv from a given source" do
+          let(:doc)    { "val1,val2,val3" }
+          let(:config) do
+            { "source"  => "datafield",
+              "columns" => ["custom1", "custom2", "custom3"] }
+          end
+          let(:event) { LogStash::Event.new("datafield" => doc) }
+
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("custom1")).to eq("val1")
+            expect(event.get("custom2")).to eq("val2")
+            expect(event.get("custom3")).to eq("val3")
+          end
         end
-      end
-    end
 
-    describe "givin target" do
-      let(:config) do
-        { "target" => "data" }
-      end
-      let(:doc)   { "big,bird,sesame street" }
-      let(:event) { LogStash::Event.new("message" => doc) }
-
-      it "extract all the values" do
-        plugin.filter(event)
-        expect(event.get("data")["column1"]).to eq("big")
-        expect(event.get("data")["column2"]).to eq("bird")
-        expect(event.get("data")["column3"]).to eq("sesame street")
+        context "that use [@metadata]" do
+          let(:metadata_field) { "[@metadata][one]" }
+          let(:config) do
+            {
+              "columns" => [ metadata_field, "foo" ]
+            }
+          end
+
+          let(:event) { LogStash::Event.new("message" => "hello,world") }
+
+          before do
+            plugin.filter(event)
+          end
+
+          it "should work correctly" do
+            expect(event.get(metadata_field)).to eq("hello")
+          end
+        end
       end
 
-      context "when having also source" do
+      describe "givin target" do
         let(:config) do
-          {  "source" => "datain",
-             "target" => "data" }
+          { "target" => "data" }
         end
-        let(:event) { LogStash::Event.new("datain" => doc) }
         let(:doc)   { "big,bird,sesame street" }
+        let(:event) { LogStash::Event.new("message" => doc) }
 
         it "extract all the values" do
           plugin.filter(event)
@@ -274,135 +262,152 @@
           expect(event.get("data")["column2"]).to eq("bird")
           expect(event.get("data")["column3"]).to eq("sesame street")
         end
-      end
-
-      context "which uses [nested][fieldref] syntax" do
-        let(:target) { "[foo][bar]" }
-        let(:config) do
-          {
-            "target" => target
-          }
-        end
-
-        let(:event) { LogStash::Event.new("message" => "hello,world") }
-
-        before do
-          plugin.filter(event)
-        end
 
-        it "should set fields correctly in the target" do
-          expect(event.get("#{target}[column1]")).to eq("hello")
-          expect(event.get("#{target}[column2]")).to eq("world")
+        context "when having also source" do
+          let(:config) do
+            {  "source" => "datain",
+               "target" => "data" }
+          end
+          let(:event) { LogStash::Event.new("datain" => doc) }
+          let(:doc)   { "big,bird,sesame street" }
+
+          it "extract all the values" do
+            plugin.filter(event)
+            expect(event.get("data")["column1"]).to eq("big")
+            expect(event.get("data")["column2"]).to eq("bird")
+            expect(event.get("data")["column3"]).to eq("sesame street")
+          end
         end
 
-        context "with nested fieldrefs as columns" do
+        context "which uses [nested][fieldref] syntax" do
+          let(:target) { "[foo][bar]" }
           let(:config) do
             {
-              "target" => target,
-              "columns" => [ "[test][one]", "[test][two]" ]
+              "target" => target
             }
           end
 
+          let(:event) { LogStash::Event.new("message" => "hello,world") }
+
+          before do
+            plugin.filter(event)
+          end
+
           it "should set fields correctly in the target" do
-            expect(event.get("#{target}[test][one]")).to eq("hello")
-            expect(event.get("#{target}[test][two]")).to eq("world")
+            expect(event.get("#{target}[column1]")).to eq("hello")
+            expect(event.get("#{target}[column2]")).to eq("world")
+          end
+
+          context "with nested fieldrefs as columns" do
+            let(:config) do
+              {
+                "target" => target,
+                "columns" => [ "[test][one]", "[test][two]" ]
+              }
+            end
+
+            it "should set fields correctly in the target" do
+              expect(event.get("#{target}[test][one]")).to eq("hello")
+              expect(event.get("#{target}[test][two]")).to eq("world")
+            end
           end
-        end
 
+        end
       end
-    end
 
-    describe "using field convertion" do
+      describe "using field convertion" do
 
-      let(:config) do
-        {
+        let(:config) do
+          {
             "convert" => {
-                "column1" => "integer",
-                "column3" => "boolean",
-                "column4" => "float",
-                "column5" => "date",
-                "column6" => "date_time",
-                "column7" => "date",
-                "column8" => "date_time",
+              "column1" => "integer",
+              "column3" => "boolean",
+              "column4" => "float",
+              "column5" => "date",
+              "column6" => "date_time",
+              "column7" => "date",
+              "column8" => "date_time",
             }
-        }
-      end
-      # 2017-06-01,2001-02-03T04:05:06+07:00
-      let(:doc)   { "1234,bird,false,3.14159265359,2017-06-01,2001-02-03 04:05:06,invalid_date,invalid_date_time" }
-      let(:event) { LogStash::Event.new("message" => doc) }
+          }
+        end
+        # 2017-06-01,2001-02-03T04:05:06+07:00
+        let(:doc)   { "1234,bird,false,3.14159265359,2017-06-01,2001-02-03 04:05:06,invalid_date,invalid_date_time" }
+        let(:event) { LogStash::Event.new("message" => doc) }
 
-      it "converts to integer" do
-        plugin.filter(event)
-        expect(event.get("column1")).to eq(1234)
-      end
+        it "converts to integer" do
+          plugin.filter(event)
+          expect(event.get("column1")).to eq(1234)
+        end
 
-      it "does not convert without converter" do
-        plugin.filter(event)
-        expect(event.get("column2")).to eq("bird")
-      end
+        it "does not convert without converter" do
+          plugin.filter(event)
+          expect(event.get("column2")).to eq("bird")
+        end
 
-      it "converts to boolean" do
-        plugin.filter(event)
-        expect(event.get("column3")).to eq(false)
-      end
+        it "converts to boolean" do
+          plugin.filter(event)
+          expect(event.get("column3")).to eq(false)
+        end
 
-      it "converts to float" do
-        plugin.filter(event)
-        expect(event.get("column4")).to eq(3.14159265359)
-      end
+        it "converts to float" do
+          plugin.filter(event)
+          expect(event.get("column4")).to eq(3.14159265359)
+        end
 
-      it "converts to date" do
-        plugin.filter(event)
-        expect(event.get("column5")).to be_a(LogStash::Timestamp)
-        expect(event.get("column5").to_s).to eq(LogStash::Timestamp.new(Date.parse("2017-06-01").to_time).to_s)
-      end
+        it "converts to date" do
+          plugin.filter(event)
+          expect(event.get("column5")).to be_a(LogStash::Timestamp)
+          expect(event.get("column5").to_s).to eq(LogStash::Timestamp.new(Date.parse("2017-06-01").to_time).to_s)
+        end
 
-      it "converts to date_time" do
-        plugin.filter(event)
-        expect(event.get("column6")).to be_a(LogStash::Timestamp)
-        expect(event.get("column6").to_s).to eq(LogStash::Timestamp.new(DateTime.parse("2001-02-03 04:05:06").to_time).to_s)
-      end
+        it "converts to date_time" do
+          plugin.filter(event)
+          expect(event.get("column6")).to be_a(LogStash::Timestamp)
+          expect(event.get("column6").to_s).to eq(LogStash::Timestamp.new(DateTime.parse("2001-02-03 04:05:06").to_time).to_s)
+        end
 
-      it "tries to converts to date but return original" do
-        plugin.filter(event)
-        expect(event.get("column7")).to eq("invalid_date")
-      end
+        it "tries to converts to date but return original" do
+          plugin.filter(event)
+          expect(event.get("column7")).to eq("invalid_date")
+        end
 
-      it "tries to converts to date_time but return original" do
-        plugin.filter(event)
-        expect(event.get("column8")).to eq("invalid_date_time")
-      end
+        it "tries to converts to date_time but return original" do
+          plugin.filter(event)
+          expect(event.get("column8")).to eq("invalid_date_time")
+        end
+
+        context "when using column names" do
 
-      context "when using column names" do
+          let(:config) do
+            { "convert" => { "custom1" => "integer", "custom3" => "boolean" },
+              "columns" => ["custom1", "custom2", "custom3"] }
+          end
+
+          it "get converted values to the expected type" do
+            plugin.filter(event)
+            expect(event.get("custom1")).to eq(1234)
+            expect(event.get("custom2")).to eq("bird")
+            expect(event.get("custom3")).to eq(false)
+          end
+        end
+      end
 
+      describe "given autodetect option" do
+        let(:header) { LogStash::Event.new("message" => "first,last,address") }
+        let(:doc)    { "big,bird,sesame street" }
         let(:config) do
-          { "convert" => { "custom1" => "integer", "custom3" => "boolean" },
-            "columns" => ["custom1", "custom2", "custom3"] }
+          { "autodetect_column_names" => true }
         end
 
-        it "get converted values to the expected type" do
+        it "extract all the values with the autodetected header" do
+          plugin.filter(header)
           plugin.filter(event)
-          expect(event.get("custom1")).to eq(1234)
-          expect(event.get("custom2")).to eq("bird")
-          expect(event.get("custom3")).to eq(false)
+          expect(event.get("first")).to eq("big")
+          expect(event.get("last")).to eq("bird")
+          expect(event.get("address")).to eq("sesame street")
         end
       end
     end
 
-    describe "given autodetect option" do
-      let(:header) { LogStash::Event.new("message" => "first,last,address") }
-      let(:doc)    { "big,bird,sesame street" }
-      let(:config) do
-        { "autodetect_column_names" => true }
-      end
-
-      it "extract all the values with the autodetected header" do
-        plugin.filter(header)
-        plugin.filter(event)
-        expect(event.get("first")).to eq("big")
-        expect(event.get("last")).to eq("bird")
-        expect(event.get("address")).to eq("sesame street")
-      end
-    end
   end
 end

From 25fff27c131b2338a8243ba5edcf75606b0d7ead Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Tue, 25 May 2021 19:01:15 +0200
Subject: [PATCH 2/6] dump version

---
 CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8af4a59..681ae71 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,5 @@
 ## 3.1.0
-  - Add ECS support 
+  - Add ECS support [#85](https://github.com/logstash-plugins/logstash-filter-csv/pull/85)
 
 ## 3.0.11
   - [DOC] Fixed formatting to improve readability [#84](https://github.com/logstash-plugins/logstash-filter-csv/pull/84)

From 2b42918c3825e39d288f068a3065faaac1185aef Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Tue, 25 May 2021 19:05:20 +0200
Subject: [PATCH 3/6] update doc

---
 docs/index.asciidoc | 1 -
 1 file changed, 1 deletion(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 724fca0..12e1638 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -121,7 +121,6 @@ Example:
 Controls this plugin's compatibility with the
 {ecs-ref}[Elastic Common Schema (ECS)].
 See <<plugins-{type}s-{plugin}-ecs_metadata>> for detailed information.
-The value of this setting affects the _default_ value of <<plugins-{type}s-{plugin}-???>>
 
 [id="plugins-{type}s-{plugin}-quote_char"]
 ===== `quote_char` 

From b982e05ec48b298059cd01a1b9bdb46e5e759541 Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Wed, 26 May 2021 11:53:35 +0200
Subject: [PATCH 4/6] update doc

---
 docs/index.asciidoc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 12e1638..9edd0a2 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -26,8 +26,8 @@ This filter can parse data with any separator, not just commas.
 
 [id="plugins-{type}s-{plugin}-ecs_metadata"]
 ==== Event Metadata and the Elastic Common Schema (ECS)
-When ECS compatibility is disabled, the value is stored in the root level.
-When ECS is enabled, the value is stored in the `target` field.
+The plugin behaves the same except giving a warning when ECS is enabled and `target` isn't set.
+It is recommended to set the `target` option to avoid potential schema conflicts.
 
 [id="plugins-{type}s-{plugin}-options"]
 ==== Csv Filter Configuration Options

From d973877f3e8aa66d0b7e370a61bd0f503ce7de9f Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Sat, 29 May 2021 01:26:26 +0200
Subject: [PATCH 5/6] update doc

---
 docs/index.asciidoc | 5 -----
 1 file changed, 5 deletions(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 9edd0a2..1479efc 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -197,11 +197,6 @@ data structure.
 Define target field for placing the data.
 Defaults to writing to the root of the event.
 
-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}-common-options"]
 include::{include_path}/{type}.asciidoc[]

From 80f807055810c3c79961be4d9e694b97b005a1da Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Sat, 29 May 2021 17:21:12 +0200
Subject: [PATCH 6/6] update doc

---
 docs/index.asciidoc | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 1479efc..6b197e9 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -26,8 +26,9 @@ This filter can parse data with any separator, not just commas.
 
 [id="plugins-{type}s-{plugin}-ecs_metadata"]
 ==== Event Metadata and the Elastic Common Schema (ECS)
-The plugin behaves the same except giving a warning when ECS is enabled and `target` isn't set.
-It is recommended to set the `target` option to avoid potential schema conflicts.
+The plugin behaves the same regardless of ECS compatibility, except giving a warning when ECS is enabled and `target` isn't set.
+
+TIP: Set the `target` option to avoid potential schema conflicts.
 
 [id="plugins-{type}s-{plugin}-options"]
 ==== Csv Filter Configuration Options