Feat: adjust fields for ECS compatibility (#28)

Co-authored-by: Karen Metts <35154725+karenzone@users.noreply.github.com>

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