Feat: ECS support + review dependencies (#20)

The plugin is setting `host` on each event (as host's name).
In ECS mode we expect to use `host.hostname` and we're also set `event.original` to be a good ECS citizen.

Besides concurrent dependency was dropped (hasn't been used),

Author: kares

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.3.0
+  - Feat: ECS support + review dependencies [#20](https://github.com/logstash-plugins/logstash-input-stdin/pull/20)
+
 ## 3.2.6
   - Docs: Set the default_codec doc attribute.
 

docs/index.asciidoc

@@ -29,10 +29,38 @@ want to join lines, you'll want to use the multiline codec.
 [id="plugins-{type}s-{plugin}-options"]
 ==== Stdin Input Configuration Options
 
-There are no special configuration options for this plugin,
-but it does support the <<plugins-{type}s-{plugin}-common-options>>.
+This plugin supports the following configuration options.
+
+[cols="<,<,<",options="header",]
+|=======================================================================
+|Setting |Input type|Required
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
+|=======================================================================
+
+Also see <<plugins-{type}s-{plugin}-common-options>> for a list of options supported by all
+input plugins.
+
+
+&nbsp;
+
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+* Value type is <<string,string>>
+* Supported values are:
+** `disabled`: does not use ECS-compatible field names (using `host` field to store host name)
+** `v1`: uses fields that are compatible with Elastic Common Schema (using `[host][hostname]`)
+* Default value depends on which version of Logstash is running:
+** When Logstash provides a `pipeline.ecs_compatibility` setting, its value is used as the default
+** Otherwise, the default value is `disabled`.
+
+Controls this plugin's compatibility with the
+{ecs-ref}[Elastic Common Schema (ECS)].
+
+
+
 
 [id="plugins-{type}s-{plugin}-common-options"]
 include::{include_path}/{type}.asciidoc[]
 
-:default_codec!:
+:default_codec!:

lib/logstash/inputs/stdin.rb

@@ -1,7 +1,7 @@
 # encoding: utf-8
 require "logstash/inputs/base"
 require "logstash/namespace"
-require "concurrent/atomics"
+require 'logstash/plugin_mixins/ecs_compatibility_support'
 require "socket" # for Socket.gethostname
 require "jruby-stdin-channel"
 
@@ -10,12 +10,27 @@
 # By default, each event is assumed to be one line. If you
 # want to join lines, you'll want to use the multiline codec.
 class LogStash::Inputs::Stdin < LogStash::Inputs::Base
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1)
+
   config_name "stdin"
 
   default :codec, "line"
 
   READ_SIZE = 16384
 
+  # When a configuration is using this plugin
+  # We are defining a blocking pipeline which cannot be reloaded
+  def self.reloadable?
+    false
+  end
+
+  def initialize(*params)
+    super
+
+    @host_key = ecs_select[disabled: 'host', v1: '[host][hostname]']
+    @event_original_key = ecs_select[disabled: nil, v1: '[event][original]']
+  end
+
   def register
     begin
       @stdin = StdinChannel::Reader.new
@@ -35,23 +50,24 @@ def run(queue)
     puts "The stdin plugin is now waiting for input:" if $stdin.tty?
     while !stop?
       if data = stdin_read
-        @codec.decode(data) do |event|
-          decorate(event)
-          event.set("host", @host) if !event.include?("host")
-          queue << event
-        end
+        process(data, queue)
       end
     end
   end
 
-  # When a configuration is using this plugin
-  # We are defining a blocking pipeline which cannot be reloaded
-  def self.reloadable?
-    false
-  end
-
   private
 
+  def process(data, queue)
+    @codec.decode(data) do |event|
+      decorate(event)
+      if @event_original_key && !event.include?(@event_original_key)
+        event.set(@event_original_key, data)
+      end
+      event.set(@host_key, @host) if !event.include?(@host_key)
+      queue << event
+    end
+  end
+
   def default_stop
     $stdin.close rescue nil
   end

logstash-input-stdin.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-stdin'
-  s.version         = '3.2.6'
+  s.version         = '3.3.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads events from standard input"
   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,13 +21,12 @@ 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.1'
   s.add_runtime_dependency "logstash-codec-line"
-  s.add_runtime_dependency "concurrent-ruby"
   s.add_runtime_dependency "jruby-stdin-channel"
 
   s.add_development_dependency "logstash-codec-plain"
   s.add_development_dependency "logstash-codec-json"
   s.add_development_dependency "logstash-codec-json_lines"
   s.add_development_dependency "logstash-devutils"
-  s.add_development_dependency "insist"
 end

spec/inputs/stdin_spec.rb

@@ -1,7 +1,6 @@
 # encoding: utf-8
 require "logstash/devutils/rspec/spec_helper"
-require "insist"
-require "socket"
+require 'logstash/plugin_mixins/ecs_compatibility_support/spec_helper'
 require "logstash/inputs/stdin"
 
 describe LogStash::Inputs::Stdin do
@@ -19,15 +18,85 @@
       require "logstash/codecs/line"
       plugin = LogStash::Inputs::Stdin.new("codec" => LogStash::Codecs::Plain.new)
       plugin.register
-      insist { plugin.codec }.is_a?(LogStash::Codecs::Line)
+      expect( plugin.codec ).is_a?(LogStash::Codecs::Line)
     end
 
     it "switches from json to json_lines" do
       require "logstash/codecs/json"
       require "logstash/codecs/json_lines"
       plugin = LogStash::Inputs::Stdin.new("codec" => LogStash::Codecs::JSON.new)
       plugin.register
-      insist { plugin.codec }.is_a?(LogStash::Codecs::JSONLines)
+      expect( plugin.codec ).is_a?(LogStash::Codecs::JSONLines)
     end
   end
+
+  context 'ECS behavior', :ecs_compatibility_support do
+
+    subject { LogStash::Inputs::Stdin.new }
+
+    ecs_compatibility_matrix(:v1) do
+
+      before(:each) do
+        allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
+
+        subject.register
+
+        subject.send :process, stdin_data, queue
+
+        expect( queue.size ).to eql 1
+      end
+
+      let(:queue) { Queue.new }
+
+      let(:stdin_data) { "a foo bar\n" }
+
+      after { subject.close }
+
+      it "sets message" do
+        event = queue.pop
+        expect( event.get('message') ).to eql 'a foo bar'
+      end
+
+      it "sets hostname" do
+        event = queue.pop
+        expect( event.get('host') ).to eql 'hostname' => `hostname`.strip
+      end
+
+      it "sets event.original" do
+        event = queue.pop
+        expect( event.get('event') ).to eql 'original' => stdin_data
+      end
+
+    end
+  end
+
+  context 'ECS disabled' do
+
+    subject { LogStash::Inputs::Stdin.new('ecs_compatibility' => 'disabled') }
+
+    before(:each) do
+      subject.register
+
+      subject.send :process, stdin_data, queue
+
+      expect( queue.size ).to eql 1
+    end
+
+    let(:queue) { Queue.new }
+
+    let(:stdin_data) { "a bar foo\n" }
+
+    after { subject.close }
+
+    it "sets message" do
+      event = queue.pop
+      expect( event.get('message') ).to eql 'a bar foo'
+    end
+
+    it "sets hostname" do
+      event = queue.pop
+      expect( event.get('host') ).to eql `hostname`.strip
+    end
+
+  end
 end