Add ECS Compatibility mode (#291)

* ecs: add ECS Compatibility mode

* Apply suggestions from code review

Co-authored-by: Karen Metts <[email protected]>

Co-authored-by: Karen Metts <[email protected]>

Author: yaauie

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.3.0
+  - Add ECS Compatibility Mode [#291](https://github.com/logstash-plugins/logstash-input-file/pull/291)
+
 ## 4.2.4
   - Fix: sincedb_write issue on Windows machines [#283](https://github.com/logstash-plugins/logstash-input-file/pull/283)
 

docs/index.asciidoc

@@ -78,6 +78,21 @@ Read mode also allows for an action to take place after processing the file comp
 In the past attempts to simulate a Read mode while still assuming infinite streams
 was not ideal and a dedicated Read mode is an improvement.
 
+[id="plugins-{type}s-{plugin}-ecs"]
+==== Compatibility with the Elastic Common Schema (ECS)
+
+This plugin adds metadata about event's source, and can be configured to do so
+in an {ecs-ref}[ECS-compatible] way with <<plugins-{type}s-{plugin}-ecs_compatibility>>.
+This metadata is added after the event has been decoded by the appropriate codec,
+and will never overwrite existing values.
+
+|========
+| ECS Disabled | ECS v1 | Description
+
+| `host` | `[host][name]` | The name of the {ls} host that processed the event
+| `path` | `[log][file][path]` | The full path to the log file from which the event originates
+|========
+
 ==== Tracking of current position in watched files
 
 The plugin keeps track of the current position in each file by
@@ -168,6 +183,7 @@ see <<plugins-{type}s-{plugin}-string_duration,string_duration>> for the details
 | <<plugins-{type}s-{plugin}-close_older>> |<<number,number>> or <<plugins-{type}s-{plugin}-string_duration,string_duration>>|No
 | <<plugins-{type}s-{plugin}-delimiter>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-discover_interval>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-exclude>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-exit_after_read>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-file_chunk_count>> |<<number,number>>|No
@@ -242,6 +258,20 @@ This value is a multiple to `stat_interval`, e.g. if `stat_interval` is "500 ms"
 files could be discovered every 15 X 500 milliseconds - 7.5 seconds.
 In practice, this will be the best case because the time taken to read new content needs to be factored in.
 
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+* Value type is <<string,string>>
+* Supported values are:
+** `disabled`: sets non-ECS metadata on event (such as top-level `host`, `path`)
+** `v1`: sets ECS-compatible metadata on event (such as `[host][name]`, `[log][file][path]`)
+* 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}-exclude"]
 ===== `exclude`
 

lib/logstash/inputs/file.rb

@@ -2,6 +2,7 @@
 require "logstash/namespace"
 require "logstash/inputs/base"
 require "logstash/codecs/identity_map_codec"
+require 'logstash/plugin_mixins/ecs_compatibility_support'
 
 require "pathname"
 require "socket" # for Socket.gethostname
@@ -88,6 +89,8 @@ module LogStash module Inputs
 class File < LogStash::Inputs::Base
   config_name "file"
 
+  include PluginMixins::ECSCompatibilitySupport(:disabled, :v1)
+
   # The path(s) to the file(s) to use as an input.
   # You can use filename patterns here, such as `/var/log/*.log`.
   # If you use a pattern like `/var/log/**/*.log`, a recursive search
@@ -325,6 +328,9 @@ def register
     @codec = LogStash::Codecs::IdentityMapCodec.new(@codec)
     @completely_stopped = Concurrent::AtomicBoolean.new
     @queue = Concurrent::AtomicReference.new
+
+    @source_host_field = ecs_select[disabled: 'host', v1:'[host][name]']
+    @source_path_field = ecs_select[disabled: 'path', v1:'[log][file][path]']
   end # def register
 
   def completely_stopped?
@@ -369,7 +375,11 @@ def run(queue)
 
   def post_process_this(event)
     event.set("[@metadata][host]", @host)
-    event.set("host", @host) unless event.include?("host")
+    attempt_set(event, @source_host_field, @host)
+
+    source_path = event.get('[@metadata][path]') and
+      attempt_set(event, @source_path_field, source_path)
+
     decorate(event)
     @queue.get << event
   end
@@ -407,6 +417,17 @@ def build_sincedb_base_from_settings(settings)
     end
   end
 
+  # Attempt to set an event's field to the provided value
+  # without overwriting an existing value or producing an error
+  def attempt_set(event, field_reference, value)
+    return false if event.include?(field_reference)
+
+    event.set(field_reference, value)
+  rescue => e
+    logger.trace("failed to set #{field_reference} to `#{value}`", :exception => e.message)
+    false
+  end
+
   def build_sincedb_base_from_env
     # This section is going to be deprecated eventually, as path.data will be
     # the default, not an environment variable (SINCEDB_DIR or LOGSTASH_HOME)

lib/logstash/inputs/file_listener.rb

@@ -41,7 +41,6 @@ def accept(data)
 
     def process_event(event)
       event.set("[@metadata][path]", path)
-      event.set("path", path) unless event.include?("path")
       input.post_process_this(event)
     end
 

logstash-input-file.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-file'
-  s.version         = '4.2.4'
+  s.version         = '4.3.0'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Streams events from files"
   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"
@@ -33,6 +33,7 @@ Gem::Specification.new do |s|
 
   s.add_runtime_dependency 'concurrent-ruby', '~> 1.0'
   s.add_runtime_dependency 'logstash-codec-multiline', ['~> 3.0']
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~>1.1'
 
   s.add_development_dependency 'stud', ['~> 0.0.19']
   s.add_development_dependency 'logstash-devutils'

spec/inputs/file_tail_spec.rb

@@ -3,7 +3,9 @@
 require "helpers/spec_helper"
 require "logstash/devutils/rspec/shared_examples"
 require "logstash/inputs/file"
+require "logstash/plugin_mixins/ecs_compatibility_support/spec_helper"
 
+require "json"
 require "tempfile"
 require "stud/temporary"
 require "logstash/codecs/multiline"
@@ -99,41 +101,59 @@
       end
     end
 
-    context "when path and host fields exist" do
-      let(:name) { "C" }
-      it "should not overwrite them" do
-        conf = <<-CONFIG
-          input {
-            file {
-              type => "blah"
-              path => "#{path_path}"
-              start_position => "beginning"
-              sincedb_path => "#{sincedb_path}"
-              delimiter => "#{TEST_FILE_DELIMITER}"
-              codec => "json"
-            }
-          }
-        CONFIG
 
-        File.open(tmpfile_path, "w") do |fd|
-          fd.puts('{"path": "my_path", "host": "my_host"}')
-          fd.puts('{"my_field": "my_val"}')
-          fd.fsync
+    context "when path and host fields exist", :ecs_compatibility_support do
+      ecs_compatibility_matrix(:disabled, :v1) do |ecs_select|
+
+        before(:each) do
+          allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
         end
 
-        events = input(conf) do |pipeline, queue|
-          2.times.collect { queue.pop }
+        let(:file_path_target_field  ) { ecs_select[disabled: "path", v1: '[log][file][path]'] }
+        let(:source_host_target_field) { ecs_select[disabled: "host", v1: '[host][name]'] }
+
+        let(:event_with_existing) do
+          LogStash::Event.new.tap do |e|
+            e.set(file_path_target_field, 'my_path')
+            e.set(source_host_target_field, 'my_host')
+          end.to_hash
         end
 
-        existing_path_index, added_path_index  = "my_val" == events[0].get("my_field") ? [1,0] : [0,1]
+        let(:name) { "C" }
+        it "should not overwrite them" do
+          conf = <<-CONFIG
+            input {
+              file {
+                type => "blah"
+                path => "#{path_path}"
+                start_position => "beginning"
+                sincedb_path => "#{sincedb_path}"
+                delimiter => "#{TEST_FILE_DELIMITER}"
+                codec => "json"
+              }
+            }
+          CONFIG
 
-        expect(events[existing_path_index].get("path")).to eq "my_path"
-        expect(events[existing_path_index].get("host")).to eq "my_host"
-        expect(events[existing_path_index].get("[@metadata][host]")).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
+          File.open(tmpfile_path, "w") do |fd|
+            fd.puts(event_with_existing.to_json)
+            fd.puts('{"my_field": "my_val"}')
+            fd.fsync
+          end
 
-        expect(events[added_path_index].get("path")).to eq "#{tmpfile_path}"
-        expect(events[added_path_index].get("host")).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
-        expect(events[added_path_index].get("[@metadata][host]")).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
+          events = input(conf) do |pipeline, queue|
+            2.times.collect { queue.pop }
+          end
+
+          existing_path_index, added_path_index  = "my_val" == events[0].get("my_field") ? [1,0] : [0,1]
+
+          expect(events[existing_path_index].get(file_path_target_field)).to eq "my_path"
+          expect(events[existing_path_index].get(source_host_target_field)).to eq "my_host"
+          expect(events[existing_path_index].get("[@metadata][host]")).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
+
+          expect(events[added_path_index].get(file_path_target_field)).to eq "#{tmpfile_path}"
+          expect(events[added_path_index].get(source_host_target_field)).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
+          expect(events[added_path_index].get("[@metadata][host]")).to eq "#{Socket.gethostname.force_encoding(Encoding::UTF_8)}"
+        end
       end
     end