Feat: adjust fields for ECS compatibility (#28)

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

Author: kares

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.1.0
+  - Feat: adjust fields for ECS compatibility [#28](https://github.com/logstash-plugins/logstash-input-unix/pull/28) 
+
 ## 3.0.7
   - Docs: Set the default_codec doc attribute.
 

docs/index.asciidoc

@@ -28,6 +28,23 @@ Like `stdin` and `file` inputs, each event is assumed to be one line of text.
 Can either accept connections from clients or connect to a server,
 depending on `mode`.
 
+[id="plugins-{type}s-{plugin}-ecs"]
+==== Compatibility with the Elastic Common Schema (ECS)
+
+This plugin adds extra fields about the event's source. 
+Configure the <<plugins-{type}s-{plugin}-ecs_compatibility>> option if you want
+to ensure that these fields are compatible with {ecs-ref}[ECS].
+
+These fields are 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
+| `path`       | `[file][path]`   | The socket path configured in the plugin
+|========
+
 [id="plugins-{type}s-{plugin}-options"]
 ==== Unix Input Configuration Options
 
@@ -37,6 +54,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 |=======================================================================
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-data_timeout>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-force_unlink>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-mode>> |<<string,string>>, one of `["server", "client"]`|No
 | <<plugins-{type}s-{plugin}-path>> |<<string,string>>|Yes
@@ -59,6 +77,44 @@ more than this timeout period, we will assume it is dead and close it.
 
 If you never want to timeout, use -1.
 
+[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]
+-----
+{
+    "@timestamp" => 2021-11-16T13:20:06.308Z,
+    "file" => {
+      "path" => "/tmp/sock41299"
+    },
+    "host" => {
+      "name" => "deus-ex-machina"
+    },
+    "message" => "foo"
+}
+-----
+
+**Sample output: ECS disabled**
+[source,ruby]
+-----
+{
+    "@timestamp" => 2021-11-16T13:20:06.308Z,
+    "path" => "/tmp/sock41299",
+    "host" => "deus-ex-machina",
+    "message" => "foo"
+}
+-----
+
 [id="plugins-{type}s-{plugin}-force_unlink"]
 ===== `force_unlink` 
 

lib/logstash/inputs/unix.rb

@@ -3,14 +3,20 @@
 require "logstash/namespace"
 require "logstash/util/socket_peer"
 
+require 'logstash/plugin_mixins/ecs_compatibility_support'
+
 # Read events over a UNIX socket.
 #
 # Like `stdin` and `file` inputs, each event is assumed to be one line of text.
 #
 # Can either accept connections from clients or connect to a server,
 # depending on `mode`.
 class LogStash::Inputs::Unix < LogStash::Inputs::Base
+
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
+
   class Interrupted < StandardError; end
+
   config_name "unix"
 
   default :codec, "line"
@@ -38,8 +44,11 @@ class Interrupted < StandardError; end
   # This setting is only used if `mode` is `client`.
   config :socket_not_present_retry_interval_seconds, :validate => :number, :required => true, :default => 5
 
-  def initialize(*args)
-    super(*args)
+  def initialize(*params)
+    super
+
+    @host_name_field = ecs_select[disabled: 'host', v1: '[host][name]']
+    @file_path_field = ecs_select[disabled: 'path', v1: '[file][path]']
   end # def initialize
 
   public
@@ -48,7 +57,7 @@ def register
     require "timeout"
 
     if server?
-      @logger.info("Starting unix input listener", :address => "#{@path}", :force_unlink => "#{@force_unlink}")
+      @logger.info("Starting unix input listener", :address => @path, :force_unlink => @force_unlink)
       begin
         @server_socket = UNIXServer.new(@path)
       rescue Errno::EADDRINUSE, IOError
@@ -58,18 +67,16 @@ def register
             @server_socket = UNIXServer.new(@path)
             return
           rescue Errno::EADDRINUSE, IOError
-            @logger.error("!!!Could not start UNIX server: Address in use",
-                          :path => @path)
+            @logger.error("Could not start UNIX server: address in use", :path => @path)
             raise
           end
         end
-        @logger.error("Could not start UNIX server: Address in use",
-                      :path => @path)
+        @logger.error("Could not start UNIX server: address in use", :path => @path)
         raise
       end
     else # client
-      if @socket_not_present_retry_interval_seconds < 0
-        @logger.warn("Value #{@socket_not_present_retry_interval_seconds} for socket_not_present_retry_interval_seconds is not valid, using default value of 5 instead")
+      if socket_not_present_retry_interval_seconds < 0
+        @logger.warn("Value #{socket_not_present_retry_interval_seconds} for socket_not_present_retry_interval_seconds is not valid, using default value of 5 instead")
         @socket_not_present_retry_interval_seconds = 5
       end
     end
@@ -93,16 +100,20 @@ def handle_socket(socket, output_queue)
         end
         @codec.decode(buf) do |event|
           decorate(event)
-          event.set("host", hostname) unless event.include?("host")
-          event.set("path", @path) unless event.include?("path")
+          event.set(@host_name_field, hostname) unless event.include?(@host_name_field)
+          event.set(@file_path_field, @path) unless event.include?(@file_path_field)
           output_queue << event
         end
       end
-    rescue => e
-      @logger.debug("Closing connection", :path => @path, :exception => e, :backtrace => e.backtrace)
     rescue Timeout::Error
-      @logger.debug("Closing connection after read timeout", :path => @path)
-    end # begin
+      @logger.info("Closing connection after read timeout", :path => @path)
+    rescue => e
+      if @logger.debug?
+        @logger.debug("Closing connection", :path => @path, :exception => e, :backtrace => e.backtrace)
+      else
+        @logger.info("Closing connection", :path => @path, :exception => e)
+      end
+    end
 
   ensure
     begin
@@ -124,7 +135,7 @@ def run(output_queue)
       while !stop?
         # Start a new thread for each connection.
         @client_threads << Thread.start(@server_socket.accept) do |s|
-          @logger.debug("Accepted connection", :server => "#{@path}")
+          @logger.debug("Accepted connection", :server => @path)
           handle_socket(s, output_queue)
         end
       end
@@ -136,8 +147,8 @@ def run(output_queue)
           @logger.debug("Opened connection", :client => @path)
           handle_socket(@client_socket, output_queue)
         else
-          @logger.warn("Socket not present, wait for #{@subscription_retry_interval_seconds} seconds for socket to appear", :client => @path)
-          sleep @socket_not_present_retry_interval_seconds
+          @logger.warn("Socket not present, wait for #{socket_not_present_retry_interval_seconds} seconds for socket to appear", :client => @path)
+          sleep socket_not_present_retry_interval_seconds
         end
       end
     end
@@ -158,6 +169,6 @@ def stop
   rescue IOError
     # if socket with @mode == client was closed by the client, an other call to @client_socket.close
     # will raise an IOError. We catch IOError here and do nothing, just let logstash terminate
-    @logger.warn("Cloud not close socket while Logstash is shutting down. Socket already closed by the other party?", :path => @path)
+    @logger.warn("Could not close socket while Logstash is shutting down. Socket already closed by the other party?", :path => @path)
   end # def stop
 end # class LogStash::Inputs::Unix

logstash-input-unix.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-unix'
-  s.version         = '3.0.7'
+  s.version         = '3.1.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads events over a UNIX socket"
   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,8 +21,9 @@ 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 'logstash-codec-line'
+
   s.add_development_dependency 'logstash-devutils'
 end
 

spec/inputs/unix_spec.rb

@@ -1,65 +1,95 @@
 # encoding: utf-8
 require_relative "../spec_helper"
 require "logstash/devutils/rspec/shared_examples"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
 require "stud/temporary"
 require "tempfile"
 
 describe LogStash::Inputs::Unix do
 
-  let(:tempfile)   { Tempfile.new("/tmp/foo") }
+  let(:config) { { 'path' => tempfile.path, 'socket_not_present_retry_interval_seconds' => 1, 'force_unlink' => true } }
+  let(:tempfile) { Tempfile.new("unix-input-test") }
+
+  subject(:input) { described_class.new(config) }
 
   it "should register without errors" do
-    plugin = LogStash::Plugin.lookup("input", "unix").new({ "path" => tempfile.path, "force_unlink" => true })
-    expect { plugin.register }.to_not raise_error
+    expect { subject.register }.to_not raise_error
   end
 
   describe "when mode is client" do
 
-    let(:mode) { "client" }
+    let(:config) { super().merge("mode" => 'client', "socket_not_present_retry_interval_seconds" => -1) }
 
     context "if socket_not_present_retry_interval_seconds is out of bounds" do
       it "should fallback to default value" do
-        plugin = LogStash::Plugin.lookup("input", "unix").new({ "path" => tempfile.path, "force_unlink" => true, "mode" => mode, "socket_not_present_retry_interval_seconds" => -1 })
-        plugin.register
-        expect(plugin.instance_variable_get(:@socket_not_present_retry_interval_seconds)).to be 5
+        subject.register
+        expect( subject.socket_not_present_retry_interval_seconds ).to eql 5
       end
     end
   end
 
-  describe "when interrupting the plugin" do
+  context "#server" do
+    it_behaves_like "an interruptible input plugin" do
+      let(:config) { super().merge "mode" => 'server' }
+    end
+  end
 
-    context "#server" do
-      it_behaves_like "an interruptible input plugin" do
-        let(:config) { { "path" => tempfile.path, "force_unlink" => true } }
-      end
+  context "#client", :ecs_compatibility_support do
+    let(:temp_path)    { "/tmp/sock#{rand(65532)}" }
+    let(:config)      { super().merge "path" => temp_path, "mode" => "client" }
+    let(:unix_socket) { UnixSocketHelper.new('foo').new_socket(temp_path) }
+    let(:run_forever) { true }
+
+    before(:each) do
+      unix_socket.loop(run_forever)
     end
 
-    context "#client" do
-      let(:tempfile)    { "/tmp/sock#{rand(65532)}" }
-      let(:config)      { { "path" => tempfile, "mode" => "client" } }
-      let(:unix_socket) { UnixSocketHelper.new.new_socket(tempfile) }
-      let(:run_forever) { true }
+    after(:each) do
+      unix_socket.close
+    end
 
-      before(:each) do
-        unix_socket.loop(run_forever)
+    context "when the unix socket has data to be read" do
+      it_behaves_like "an interruptible input plugin" do
+        let(:run_forever) { true }
       end
+    end
 
-      after(:each) do
-        unix_socket.close
+    context "when the unix socket has no data to be read" do
+      it_behaves_like "an interruptible input plugin" do
+        let(:run_forever) { false }
       end
+    end
+
+    ecs_compatibility_matrix(:disabled, :v1, :v8) do |ecs_select|
+
+      let(:config) { super().merge 'ecs_compatibility' => ecs_compatibility }
 
-      context "when the unix socket has data to be read" do
-        it_behaves_like "an interruptible input plugin" do
-          let(:run_forever) { true }
+      let(:queue) { java.util.Vector.new }
+
+      it 'generates events with host, path and message set' do
+        subject.register
+        Thread.new(subject, queue) { |subject, queue| subject.run(queue) }
+        try(10) do
+          expect( queue.size ).to_not eql 0
         end
-      end
+        subject.do_stop # stop the plugin
 
-      context "when the unix socket has no data to be read" do
-        it_behaves_like "an interruptible input plugin" do
-          let(:run_forever) { false }
+        event = queue.first
+
+        if ecs_select.active_mode == :disabled
+          expect( event.get('host') ).to be_a String
+          expect( event.get('path') ).to eql temp_path
+        else
+          expect( event.get('[host][name]') ).to be_a String
+          expect( event.get('[file][path]') ).to eql temp_path
+          expect( event.include?('path') ).to be false
         end
+
+        expect( event.get('message') ).to eql 'foo'
       end
+
     end
 
   end
+
 end

spec/spec_helper.rb

@@ -6,8 +6,9 @@ class UnixSocketHelper
 
   attr_reader :path
 
-  def initialize
+  def initialize(line = 'hi!')
     @socket = nil
+    @line = line
   end
 
   def new_socket(path)
@@ -21,9 +22,9 @@ def loop(forever=false)
     @thread = Thread.new do
       begin
         s = @socket.accept
-        s.puts "hi" while forever
-      rescue Errno::EPIPE, Errno::ECONNRESET
-        # ...
+        s.puts @line while forever
+      rescue Errno::EPIPE, Errno::ECONNRESET => e
+        warn e.inspect if $VERBOSE
       end
     end
     self