Feat: adjust fields for ECS compatibility

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 3.4.0
+  - Feat: adjust fields for ECS compatibility [#28](https://github.com/logstash-plugins/logstash-input-exec/pull/28)
+  - Plugin will no longer override fields if they exist in the decoded payload.
+    (It no longer sets the `host` field if decoded from the command's output.)
+
 ## 3.3.3
   - Docs: improved doc on memory usage [#27](https://github.com/logstash-plugins/logstash-input-exec/pull/27)
 

docs/index.asciidoc

@@ -41,24 +41,44 @@ Example:
 ----------------------------------
 input {
   exec {
-    command => "ls"
+    command => "echo 'hi!'"
     interval => 30
   }
 }
 ----------------------------------
 
-This will execute `ls` command every 30 seconds.
+This will execute `echo` command every 30 seconds.
+
+[id="plugins-{type}s-{plugin}-ecs"]
+==== Compatibility with the Elastic Common Schema (ECS)
+
+This plugin adds metadata about the 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 not overwrite existing values.
+
+|========
+| ECS Disabled | ECS v1 | Description
+
+| `host`  | `[host][name]` | The name of the {ls} host that processed the event
+| `command` | `[process][command_line]` | The command run by the plugin
+| `[@metadata][exit_status]` | `[process][exit_code]` | The exit code of the process
+
+| `[@metadata][input][exec][process][elapsed_time]` | -- | The elapsed time the command took to run in nanoseconds
+| `[@metadata][duration]` | -- | Command duration in seconds as a floating point number (deprecated)
+|========
 
 
 [id="plugins-{type}s-{plugin}-options"]
-==== Exec Input Configuration Options
+==== Exec Input configuration options
 
 This plugin supports the following configuration options plus the <<plugins-{type}s-{plugin}-common-options>> described later.
 
 [cols="<,<,<",options="header",]
 |=======================================================================
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-command>> |<<string,string>>|Yes
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-interval>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-schedule>> |<<string,string>>|No
 |=======================================================================
@@ -77,6 +97,58 @@ input plugins.
 
 Command to run. For example, `uptime`
 
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+  * Value type is <<string,string>>
+  * Supported values are:
+    ** `disabled`: uses backwards compatible field names, such as `[host]`
+    ** `v1`, `v8`: uses fields that are compatible with ECS, such as `[host][name]`
+
+Controls this plugin's compatibility with the {ecs-ref}[Elastic Common Schema (ECS)].
+See <<plugins-{type}s-{plugin}-ecs>> for detailed information.
+
+
+**Sample output: ECS enabled**
+[source,ruby]
+-----
+{
+    "message" => "hi!\n",
+    "process" => {
+        "command_line" => "echo 'hi!'",
+        "exit_code" => 0
+    },
+    "host" => {
+        "name" => "deus-ex-machina"
+    },
+
+    "@metadata" => {
+        "input" => {
+            "exec" => {
+                "process" => {
+                    "elapsed_time"=>3042
+                }
+            }
+        }
+    }
+}
+-----
+
+**Sample output: ECS disabled**
+[source,ruby]
+-----
+{
+    "message" => "hi!\n",
+    "command" => "echo 'hi!'",
+    "host" => "deus-ex-machina",
+
+    "@metadata" => {
+        "exit_status" => 0,
+        "duration" => 0.004388
+    }
+}
+-----
+
 [id="plugins-{type}s-{plugin}-interval"]
 ===== `interval` 
 

lib/logstash/inputs/exec.rb

@@ -5,6 +5,8 @@
 require "stud/interval"
 require "rufus/scheduler"
 
+require 'logstash/plugin_mixins/ecs_compatibility_support'
+
 # Periodically run a shell command and capture the whole output as an event.
 #
 # Notes:
@@ -14,6 +16,8 @@
 #
 class LogStash::Inputs::Exec < LogStash::Inputs::Base
 
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
+
   config_name "exec"
 
   default :codec, "plain"
@@ -31,13 +35,18 @@ class LogStash::Inputs::Exec < LogStash::Inputs::Base
   config :schedule, :validate => :string
 
   def register
-    @logger.info("Registering Exec Input", :type => @type, :command => @command, :interval => @interval, :schedule => @schedule)
-    @hostname = Socket.gethostname
+    @hostname = Socket.gethostname.freeze
     @io       = nil
 
     if (@interval.nil? && @schedule.nil?) || (@interval && @schedule)
       raise LogStash::ConfigurationError, "exec input: either 'interval' or 'schedule' option must be defined."
     end
+
+    @host_name_field = ecs_select[disabled: 'host', v1: '[host][name]']
+    @process_command_line_field = ecs_select[disabled: 'command', v1: '[process][command_line]']
+    @process_exit_code_field = ecs_select[disabled: '[@metadata][exit_status]', v1: '[process][exit_code]']
+    @process_elapsed_time_field = ecs_select[disabled: nil, v1: '[@metadata][input][exec][process][elapsed_time]'] # in nanos
+    @legacy_duration_field = ecs_select[disabled: '[@metadata][duration]', v1: nil] # in seconds
   end # def register
 
   def run(queue)
@@ -71,20 +80,21 @@ def execute(queue)
       output, exit_status = run_command()
     rescue StandardError => e
       @logger.error("Error while running command",
-        :command => @command, :e => e, :backtrace => e.backtrace)
+        :command => @command, :exception => e, :backtrace => e.backtrace)
     rescue Exception => e
       @logger.error("Exception while running command",
-        :command => @command, :e => e, :backtrace => e.backtrace)
+        :command => @command, :exception => e, :backtrace => e.backtrace)
     end
     duration = Time.now - start
     @logger.debug? && @logger.debug("Command completed", :command => @command, :duration => duration)
     if output
       @codec.decode(output) do |event|
         decorate(event)
-        event.set("host", @hostname)
-        event.set("command", @command)
-        event.set("[@metadata][duration]", duration)
-        event.set("[@metadata][exit_status]", exit_status)
+        event.set(@host_name_field, @hostname) unless event.include?(@host_name_field)
+        event.set(@process_command_line_field, @command) unless event.include?(@process_command_line_field)
+        event.set(@process_exit_code_field, exit_status) unless event.include?(@process_exit_code_field)
+        event.set(@process_elapsed_time_field, to_nanos(duration)) if @process_elapsed_time_field
+        event.set(@legacy_duration_field, duration) if @legacy_duration_field
         queue << event
       end
     end
@@ -97,7 +107,7 @@ def run_command
     @io = IO.popen(@command)
     output = @io.read
     @io.close # required in order to read $?
-    exit_status = $?.exitstatus # should be threadsafe as per rb_thread_save_context
+    exit_status = $?.exitstatus
     [output, exit_status]
   ensure
     close_io()
@@ -124,5 +134,9 @@ def wait_until_end_of_interval(duration)
     end
   end
 
+  # convert seconds (float) to nanoseconds
+  def to_nanos(time_diff)
+    (time_diff * 1_000_000).to_i
+  end
 
 end # class LogStash::Inputs::Exec

logstash-input-exec.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-exec'
-  s.version         = '3.3.3'
+  s.version         = '3.4.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Captures the output of a shell command as an event"
   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"
@@ -21,6 +21,8 @@ Gem::Specification.new do |s|
 
   # 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 'stud', '~> 0.0.22'
   s.add_runtime_dependency 'logstash-codec-plain'
   s.add_runtime_dependency 'rufus-scheduler'

spec/inputs/exec_spec.rb

@@ -3,75 +3,118 @@
 require "time"
 require_relative "../spec_helper"
 require "logstash/devutils/rspec/shared_examples"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
 
-describe LogStash::Inputs::Exec do
+describe LogStash::Inputs::Exec, :ecs_compatibility_support do
 
-  context "when register" do
-    let(:input) { described_class.new("command" => "ls", "interval" => 0) }
-    it "should not raise error if config is valid" do
-      # register will try to load jars and raise if it cannot find jars or if org.apache.log4j.spi.LoggingEvent class is not present
-      expect {input.register}.to_not raise_error
+  ecs_compatibility_matrix(:disabled, :v1, :v8) do |ecs_select|
+
+    before(:each) do
+      allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
     end
-    context "with an invalid config" do
-      let(:input) { described_class.new("command" => "ls") }
-      it "should raise error" do
-        expect {input.register}.to raise_error(LogStash::ConfigurationError)
+
+    context "when register" do
+      let(:input) { described_class.new("command" => "ls", "interval" => 0) }
+
+      it "should not raise error if config is valid" do
+        # register will try to load jars and raise if it cannot find jars or if org.apache.log4j.spi.LoggingEvent class is not present
+        expect { input.register }.to_not raise_error
       end
-    end
-  end
 
-  context "when operating normally" do
-    let(:input) { described_class.new("command" => "ls", "interval" => 0) }
-    let(:queue) { [] }
-    let(:loggr) { double('loggr') }
-
-    before :each do
-      expect(described_class).to receive(:logger).and_return(loggr).exactly(7).times
-      allow(loggr).to receive(:info)
-      allow(loggr).to receive(:info?)
-      allow(loggr).to receive(:warn)
-      allow(loggr).to receive(:warn?)
-      allow(loggr).to receive(:debug)
-      allow(loggr).to receive(:debug?)
+      context "with an invalid config" do
+        let(:input) { described_class.new("command" => "ls") }
+        it "should raise error" do
+          expect { input.register }.to raise_error(LogStash::ConfigurationError)
+        end
+      end
     end
 
-    it "enqueues some events" do
-      input.register
-      expect(loggr).not_to receive(:error)
+    context "when operating normally" do
+      let(:input) { described_class.new("command" => "echo 'hi!'", "interval" => 0) }
+      let(:queue) { [] }
 
-      input.execute(queue)
+      before :each do
+        input.register
+      end
 
-      expect(queue.size).not_to be_zero
-    end
-  end
+      it "enqueues some events" do
+        expect(input.logger).not_to receive(:error)
 
-  context "when a command runs normally" do
-    let(:input) { described_class.new("command" => "/bin/sh -c 'sleep 1; /bin/echo -n two; exit 3'", "interval" => 0) }
-    let(:queue) { [] }
+        input.execute(queue)
 
-    before do
-      input.register
-      input.execute(queue)
+        expect(queue.size).not_to be_zero
+      end
     end
 
-    after do
-      input.stop
-    end
+    context "when command fails" do
+      let(:input) { described_class.new("command" => "invalid_command 1 2 3", "interval" => 0) }
+      let(:queue) { [] }
+
+      before :each do
+        input.register
+      end
+
+      it "does not enqueue an event (in a non-Docker env)" do
+        expect(input.logger).to receive(:error).and_call_original
+
+        input.execute(queue)
+
+        expect(queue.map(&:to_hash)).to be_empty
+      end
+    end if ENV['CI'] != 'true' # in Docker the behavior differs - missing command files are not raised
+
+    context "when a command runs normally" do
+      let(:command) { "/bin/sh -c 'sleep 1; /bin/echo -n two; exit 3'" }
+      let(:input) { described_class.new("command" => command, "interval" => 0) }
+      let(:queue) { [] }
+
+      before do
+        input.register
+        input.execute(queue)
+      end
+
+      after do
+        input.stop
+      end
+
+      it "has duration (in seconds)" do
+        duration = queue.pop.get('[@metadata][duration]')
+        expect(duration).to be > 1
+        expect(duration).to be < 3
+      end if ecs_select.active_mode == :disabled
+
+      it "reports process elapsed time (in nanos)" do
+        elapsed_time = queue.pop.get('[@metadata][input][exec][process][elapsed_time]')
+        expect(elapsed_time).to be > 1 * 1_000_000
+        expect(elapsed_time).to be < 3 * 1_000_000
+      end if ecs_select.active_mode != :disabled
+
+      it "has output as expected" do
+        expect(queue.pop.get('message')).to eq "two"
+      end
+
+      it "reports process command_line  " do
+        if ecs_select.active_mode == :disabled
+          expect(queue.pop.get('command')).to eql command
+        else
+          expect(queue.pop.get('[process][command_line]')).to eql command
+        end
+      end
+
+      it "reports process exit_code" do
+        if ecs_select.active_mode == :disabled
+          expect(queue.pop.get('[@metadata][exit_status]')).to eq 3
+        else
+          expect(queue.pop.get('[process][exit_code]')).to eq 3
+        end
+      end
 
-    it "has duration tracked" do
-      expect(queue.pop.get('[@metadata][duration]')).to be > 1
-    end
-    it "has output as expected" do
-      expect(queue.pop.get('message')).to eq "two"
-    end
-    it "has exit_status tracked" do
-      expect(queue.pop.get('[@metadata][exit_status]')).to eq 3
     end
 
   end
 
   context "when scheduling" do
-    let(:input) { described_class.new("command" => "ls", "schedule" => "* * * * * UTC") }
+    let(:input) { described_class.new("command" => "ls --help", "schedule" => "* * * * * UTC") }
     let(:queue) { [] }
 
     before do