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

@@ -23,14 +23,13 @@ include::{include_path}/plugin_header.asciidoc[]
 
 Periodically run a shell command and capture the whole output as an event.
 
-Notes:
-
+[NOTE]
+========
 * The `command` field of this event will be the command run.
 * The `message` field of this event will be the entire stdout of the command.
+========
 
-===== IMPORTANT
-
-The exec input ultimately uses `fork` to spawn a child process.
+IMPORTANT: The exec input ultimately uses `fork` to spawn a child process.
 Using fork duplicates the parent process address space (in our case, **logstash and the JVM**); this is mitigated with OS copy-on-write but ultimately you can end up allocating lots of memory just for a "simple" executable.
 If the exec input fails with errors like `ENOMEM: Cannot allocate memory` it is an indication that there is not enough non-JVM-heap physical memory to perform the fork.
 
@@ -41,24 +40,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 , v8               | 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 +96,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,20 @@ 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]']
+
+    # migrate elapsed time tracking to whole nanos, from legacy floating-point fractional seconds
+    @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)
@@ -61,8 +72,7 @@ def stop
   end
 
   # Execute a given command
-  # @param [String] A command string
-  # @param [Array or Queue] A queue to append events to
+  # @param queue the LS queue to append events to
   def execute(queue)
     start = Time.now
     output = exit_status = nil
@@ -71,20 +81,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)
+    duration = Time.now.to_r - start.to_r
+    @logger.debug? && @logger.debug("Command completed", :command => @command, :duration => duration.to_f)
     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.to_f) if @legacy_duration_field
         queue << event
       end
     end
@@ -97,7 +108,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()
@@ -111,7 +122,7 @@ def close_io
   end
 
   # Wait until the end of the interval
-  # @param [Integer] the duration of the last command executed
+  # @param duration [Integer] the duration of the last command executed
   def wait_until_end_of_interval(duration)
     # Sleep for the remainder of the interval, or 0 if the duration ran
     # longer than the interval.
@@ -124,5 +135,10 @@ def wait_until_end_of_interval(duration)
     end
   end
 
+  # convert seconds to nanoseconds
+  # @param time_diff [Numeric] the (rational value) difference to convert
+  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'