Feat: adjust fields for ECS compatibility (#28)

Previously plugin was setting the top level command and host fields for every event (which aren't ECS compliant). Also changes behavior not to override fields if they exists in the decoded payload (e.g. no longer force the host field if such a field is decoded from the command's output). Co-authored-by: Ry Biesemeyer <yaauie@users.noreply.github.com> Co-authored-by: Karen Metts <35154725+karenzone@users.noreply.github.com>
logstash-plugins · Nov 16, 2021 · 9cbb494 · 9cbb494
1 parent 78215e4
commit 9cbb494
Show file tree

Hide file tree

Showing 5 changed files with 195 additions and 73 deletions.
diff --git a/CHANGELOG.md b/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)
 

diff --git a/docs/index.asciidoc b/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` 
 

diff --git a/lib/logstash/inputs/exec.rb b/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
diff --git a/logstash-input-exec.gemspec b/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'