Add ECS support (#129)

Restructure metadata and create ECS failure event

Author: kaisecheng

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.1.0
+  - Add ECS support [#129](https://github.com/logstash-plugins/logstash-input-http_poller/pull/129)
+
 ## 5.0.2
  - [DOC]Expanded url option to include Manticore keys [#119](https://github.com/logstash-plugins/logstash-input-http_poller/pull/119)
 

docs/index.asciidoc

@@ -87,6 +87,36 @@ The above snippet will create two files `downloaded_cert.pem` and `downloaded_tr
 ----------------------------------
 
 
+[id="plugins-{type}s-{plugin}-ecs_metadata"]
+==== Event Metadata and the Elastic Common Schema (ECS)
+
+This input will add metadata about the HTTP connection itself to each event.
+
+When ECS compatibility is disabled, metadata was added to a variety of non-standard top-level fields, which has the potential to create confusion and schema conflicts downstream.
+
+With ECS Compatibility Mode, we can ensure a pipeline maintains access to this metadata throughout the event's lifecycle without polluting the top-level namespace.
+
+Here’s how ECS compatibility mode affects output.
+[cols="<l,<l,e,<e"]
+|=======================================================================
+| ECS disabled | ECS v1 | Availability | Description
+
+| [@metadata][host] | [@metadata][input][http_poller][request][host][hostname] | Always | Hostname
+| [@metadata][code] | [@metadata][input][http_poller][response][status_code] | When server responds a valid status code | HTTP response code
+| [@metadata][response_headers] | [@metadata][input][http_poller][response][headers] | When server responds with headers | HTTP headers of the response
+| [@metadata][response_message] | [@metadata][input][http_poller][response][status_message] | When server responds with status line | message of status line of HTTP headers
+| [@metadata][runtime_seconds] | [@metadata][input][http_poller][response][elapsed_time_ns] | When server responds a valid status code | elapsed time of calling endpoint. ECS v1 shows in nanoseconds.
+| [http_request_failure][runtime_seconds] | [event][duration] | When server throws exception | elapsed time of calling endpoint. ECS v1 shows in nanoseconds.
+| [@metadata][times_retried] | [@metadata][input][http_poller][request][retry_count] | When the poller calls server successfully | retry count from http client library
+| [@metadata][name] / [http_request_failure][name] | [@metadata][input][http_poller][request][name] | Always | The key of `urls` from poller config
+| [@metadata][request] / [http_request_failure][request]| [@metadata][input][http_poller][request][original] | Always | The whole object of `urls` from poller config
+| [http_request_failure][error] | [error][message] | When server throws exception | Error message
+| [http_request_failure][backtrace] | [error][stack_trace] | When server throws exception | Stack trace of error
+| -- | [url][full] | When server throws exception | The URL of the endpoint
+| -- | [http][request][method] | When server throws exception | HTTP request method
+| -- | [host][hostname] | When server throws exception | Hostname
+|=======================================================================
+
 [id="plugins-{type}s-{plugin}-options"]
 ==== Http_poller Input Configuration Options
 
@@ -101,6 +131,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-client_key>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-connect_timeout>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-cookies>> |<<boolean,boolean>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
 | <<plugins-{type}s-{plugin}-follow_redirects>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-keepalive>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-keystore>> |a valid filesystem path|No
@@ -180,6 +211,81 @@ Timeout (in seconds) to wait for a connection to be established. Default is `10s
 Enable cookie support. With this enabled the client will persist cookies
 across requests as a normal web browser would. Enabled by default
 
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+* Value type is <<string,string>>
+* Supported values are:
+** `disabled`: unstructured data added at root level
+** `v1`: uses `error`, `url` and `http` fields that are compatible with Elastic Common Schema
+
+Controls this plugin's compatibility with the
+{ecs-ref}[Elastic Common Schema (ECS)].
+See <<plugins-{type}s-{plugin}-ecs_metadata>> for detailed information.
+
+Example output:
+
+**Sample output: ECS disabled**
+[source,text]
+-----
+{
+    "http_poller_data" => {
+        "@version" => "1",
+        "@timestamp" => 2021-01-01T00:43:22.388Z,
+        "status" => "UP"
+    },
+    "@version" => "1",
+    "@timestamp" => 2021-01-01T00:43:22.389Z,
+}
+-----
+
+**Sample output: ECS enabled**
+[source,text]
+-----
+{
+    "http_poller_data" => {
+        "status" => "UP",
+        "@version" => "1",
+        "event" => {
+            "original" => "{\"status\":\"UP\"}"
+        },
+        "@timestamp" => 2021-01-01T00:40:59.558Z
+    },
+    "@version" => "1",
+    "@timestamp" => 2021-01-01T00:40:59.559Z
+}
+-----
+
+**Sample error output: ECS enabled**
+[source,text]
+----
+{
+    "@timestamp" => 2021-07-09T09:53:48.721Z,
+    "@version" => "1",
+    "host" => {
+        "hostname" => "MacBook-Pro"
+    },
+    "http" => {
+        "request" => {
+            "method" => "get"
+        }
+    },
+    "event" => {
+        "duration" => 259019
+    },
+    "error" => {
+        "stack_trace" => nil,
+        "message" => "Connection refused (Connection refused)"
+    },
+    "url" => {
+        "full" => "http://localhost:8080/actuator/health"
+    },
+    "tags" => [
+        [0] "_http_request_failure"
+    ]
+}
+----
+
 [id="plugins-{type}s-{plugin}-follow_redirects"]
 ===== `follow_redirects` 
 
@@ -315,6 +421,10 @@ Timeout (in seconds) to wait for data on the socket. Default is `10s`
 
 Define the target field for placing the received data. If this setting is omitted, the data will be stored at the root (top level) of the event.
 
+TIP: When ECS is enabled, set `target` in the codec (if the codec has a `target` option).
+Example: `codec => json { target => "TARGET_FIELD_NAME" }`
+
+
 [id="plugins-{type}s-{plugin}-truststore"]
 ===== `truststore` 
 

lib/logstash/inputs/http_poller.rb

@@ -5,9 +5,18 @@
 require "socket" # for Socket.gethostname
 require "manticore"
 require "rufus/scheduler"
+require "logstash/plugin_mixins/ecs_compatibility_support"
+require 'logstash/plugin_mixins/ecs_compatibility_support/target_check'
+require 'logstash/plugin_mixins/validator_support/field_reference_validation_adapter'
+require 'logstash/plugin_mixins/event_support/event_factory_adapter'
 
 class LogStash::Inputs::HTTP_Poller < LogStash::Inputs::Base
   include LogStash::PluginMixins::HttpClient
+  include LogStash::PluginMixins::ECSCompatibilitySupport(:disabled, :v1, :v8 => :v1)
+  include LogStash::PluginMixins::ECSCompatibilitySupport::TargetCheck
+  include LogStash::PluginMixins::EventSupport::EventFactoryAdapter
+
+  extend LogStash::PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
 
   config_name "http_poller"
 
@@ -28,7 +37,7 @@ class LogStash::Inputs::HTTP_Poller < LogStash::Inputs::Base
   config :schedule, :validate => :hash, :required => true
 
   # Define the target field for placing the received data. If this setting is omitted, the data will be stored at the root (top level) of the event.
-  config :target, :validate => :string
+  config :target, :validate => :field_reference
 
   # If you'd like to work with the request/response metadata.
   # Set this value to the name of the field you'd like to store a nested
@@ -42,6 +51,7 @@ def register
 
     @logger.info("Registering http_poller Input", :type => @type, :schedule => @schedule, :timeout => @timeout)
 
+    setup_ecs_field!
     setup_requests!
   end
 
@@ -55,6 +65,35 @@ def setup_requests!
     @requests = Hash[@urls.map {|name, url| [name, normalize_request(url)] }]
   end
 
+  private
+  # In the context of ECS, there are two type of events in this plugin, valid HTTP response and failure
+  # For a valid HTTP response, `url`, `request_method` and `host` are metadata of request.
+  #   The call could retrieve event which contain `[url]`, `[http][request][method]`, `[host][hostname]` data
+  #   Therefore, metadata should not write to those fields
+  # For a failure, `url`, `request_method` and `host` are primary data of the event because the plugin owns this event,
+  #   so it writes to url.*, http.*, host.*
+  def setup_ecs_field!
+    @request_host_field = ecs_select[disabled: "[#{metadata_target}][host]", v1: "[#{metadata_target}][input][http_poller][request][host][hostname]"]
+    @response_code_field = ecs_select[disabled: "[#{metadata_target}][code]", v1: "[#{metadata_target}][input][http_poller][response][status_code]"]
+    @response_headers_field = ecs_select[disabled: "[#{metadata_target}][response_headers]", v1: "[#{metadata_target}][input][http_poller][response][headers]"]
+    @response_message_field = ecs_select[disabled: "[#{metadata_target}][response_message]", v1: "[#{metadata_target}][input][http_poller][response][status_message]"]
+    @response_time_s_field = ecs_select[disabled: "[#{metadata_target}][runtime_seconds]", v1: nil]
+    @response_time_ns_field = ecs_select[disabled: nil, v1: "[#{metadata_target}][input][http_poller][response][elapsed_time_ns]"]
+    @request_retry_count_field = ecs_select[disabled: "[#{metadata_target}][times_retried]", v1: "[#{metadata_target}][input][http_poller][request][retry_count]"]
+    @request_name_field = ecs_select[disabled: "[#{metadata_target}][name]", v1: "[#{metadata_target}][input][http_poller][request][name]"]
+    @original_request_field = ecs_select[disabled: "[#{metadata_target}][request]", v1: "[#{metadata_target}][input][http_poller][request][original]"]
+
+    @error_msg_field = ecs_select[disabled: "[http_request_failure][error]", v1: "[error][message]"]
+    @stack_trace_field = ecs_select[disabled: "[http_request_failure][backtrace]", v1: "[error][stack_trace]"]
+    @fail_original_request_field = ecs_select[disabled: "[http_request_failure][request]", v1: nil]
+    @fail_request_name_field = ecs_select[disabled: "[http_request_failure][name]", v1: nil]
+    @fail_response_time_s_field = ecs_select[disabled: "[http_request_failure][runtime_seconds]", v1: nil]
+    @fail_response_time_ns_field = ecs_select[disabled: nil, v1: "[event][duration]"]
+    @fail_request_url_field = ecs_select[disabled: nil, v1: "[url][full]"]
+    @fail_request_method_field = ecs_select[disabled: nil, v1: "[http][request][method]"]
+    @fail_request_host_field = ecs_select[disabled: nil, v1: "[host][hostname]"]
+  end
+
   private
   def normalize_request(url_or_spec)
     if url_or_spec.is_a?(String)
@@ -151,10 +190,14 @@ def request_async(queue, name, request)
 
     method, *request_opts = request
     client.async.send(method, *request_opts).
-      on_success {|response| handle_success(queue, name, request, response, Time.now - started)}.
-      on_failure {|exception|
-      handle_failure(queue, name, request, exception, Time.now - started)
-    }
+      on_success {|response| handle_success(queue, name, request, response, Time.now - started) }.
+      on_failure {|exception| handle_failure(queue, name, request, exception, Time.now - started) }
+  end
+
+  private
+  # time diff in float to nanoseconds
+  def to_nanoseconds(time_diff)
+    (time_diff * 1000000).to_i
   end
 
   private
@@ -164,11 +207,11 @@ def handle_success(queue, name, request, response, execution_time)
     # responses come up as "" which will cause the codec to not yield anything
     if body && body.size > 0
       decode_and_flush(@codec, body) do |decoded|
-        event = @target ? LogStash::Event.new(@target => decoded.to_hash) : decoded
+        event = @target ? targeted_event_factory.new_event(decoded.to_hash) : decoded
         handle_decoded_event(queue, name, request, response, event, execution_time)
       end
     else
-      event = ::LogStash::Event.new
+      event = event_factory.new_event
       handle_decoded_event(queue, name, request, response, event, execution_time)
     end
   end
@@ -197,20 +240,10 @@ def handle_decoded_event(queue, name, request, response, event, execution_time)
   private
   # Beware, on old versions of manticore some uncommon failures are not handled
   def handle_failure(queue, name, request, exception, execution_time)
-    event = LogStash::Event.new
-    apply_metadata(event, name, request)
-
+    event = event_factory.new_event
     event.tag("_http_request_failure")
-
-    # This is also in the metadata, but we send it anyone because we want this
-    # persisted by default, whereas metadata isn't. People don't like mysterious errors
-    event.set("http_request_failure", {
-      "request" => structure_request(request),
-      "name" => name,
-      "error" => exception.to_s,
-      "backtrace" => exception.backtrace,
-      "runtime_seconds" => execution_time
-   })
+    apply_metadata(event, name, request, nil, execution_time)
+    apply_failure_fields(event, name, request, exception, execution_time)
 
     queue << event
   rescue StandardError, java.lang.Exception => e
@@ -231,29 +264,39 @@ def handle_failure(queue, name, request, exception, execution_time)
   end
 
   private
-  def apply_metadata(event, name, request, response=nil, execution_time=nil)
+  def apply_metadata(event, name, request, response, execution_time)
     return unless @metadata_target
-    event.set(@metadata_target, event_metadata(name, request, response, execution_time))
-  end
-
-  private
-  def event_metadata(name, request, response=nil, execution_time=nil)
-    m = {
-        "name" => name,
-        "host" => @host,
-        "request" => structure_request(request),
-      }
 
-    m["runtime_seconds"] = execution_time
+    event.set(@request_host_field, @host)
+    event.set(@response_time_s_field, execution_time) if @response_time_s_field
+    event.set(@response_time_ns_field, to_nanoseconds(execution_time)) if @response_time_ns_field
+    event.set(@request_name_field, name)
+    event.set(@original_request_field, structure_request(request))
 
     if response
-      m["code"] = response.code
-      m["response_headers"] = response.headers
-      m["response_message"] = response.message
-      m["times_retried"] = response.times_retried
+      event.set(@response_code_field, response.code)
+      event.set(@response_headers_field, response.headers)
+      event.set(@response_message_field, response.message)
+      event.set(@request_retry_count_field, response.times_retried)
     end
+  end
 
-    m
+  private
+  def apply_failure_fields(event, name, request, exception, execution_time)
+    # This is also in the metadata, but we send it anyone because we want this
+    # persisted by default, whereas metadata isn't. People don't like mysterious errors
+    event.set(@fail_original_request_field, structure_request(request)) if @fail_original_request_field
+    event.set(@fail_request_name_field, name) if @fail_request_name_field
+
+    method, url, _ = request
+    event.set(@fail_request_url_field, url) if @fail_request_url_field
+    event.set(@fail_request_method_field, method.to_s) if @fail_request_method_field
+    event.set(@fail_request_host_field, @host) if @fail_request_host_field
+
+    event.set(@fail_response_time_s_field, execution_time) if @fail_response_time_s_field
+    event.set(@fail_response_time_ns_field, to_nanoseconds(execution_time)) if @fail_response_time_ns_field
+    event.set(@error_msg_field, exception.to_s)
+    event.set(@stack_trace_field, exception.backtrace)
   end
 
   private

logstash-input-http_poller.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name        = 'logstash-input-http_poller'
-  s.version         = '5.0.2'
+  s.version         = '5.1.0'
   s.licenses    = ['Apache License (2.0)']
   s.summary     = "Decodes the output of an HTTP API into events"
   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"
@@ -23,6 +23,9 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'logstash-mixin-http_client', "~> 7"
   s.add_runtime_dependency 'stud', "~> 0.0.22"
   s.add_runtime_dependency 'rufus-scheduler', "~>3.0.9"
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~>1.3'
+  s.add_runtime_dependency 'logstash-mixin-event_support', '~> 1.0', '>= 1.0.1'
+  s.add_runtime_dependency 'logstash-mixin-validator_support', '~> 1.0'
 
   s.add_development_dependency 'logstash-codec-json'
   s.add_development_dependency 'logstash-codec-line'