Implement ECS Compatibility Mode

.gitignore

@@ -12,3 +12,4 @@ target/
 vendor/
 /spec/fixtures/server.key
 /spec/fixtures/server.crt
+/lib/logstash/outputs/elasticsearch/templates/ecs-v*

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 10.6.0
+ - Added `ecs_compatiblity` mode, for managing ECS-compatable templates [#952](https://github.com/logstash-plugins/logstash-output-elasticsearch/issue/952)
+
 ## 10.5.1
   - [DOC] Removed outdated compatibility notices, reworked cloud notice, and fixed formatting for `hosts` examples [#938](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/938)
 

Rakefile

@@ -1 +1,37 @@
 require "logstash/devutils/rake"
+
+ECS_VERSIONS = {
+    v1: 'v1.5.0'
+}
+
+ECS_LOGSTASH_INDEX_PATTERNS = %w(
+  ecs-logstash-*
+)
+
+task :'vendor-ecs-schemata' do
+  download_ecs_schema(:v1, 6)
+  download_ecs_schema(:v1, 7)
+end
+task :vendor => :'vendor-ecs-schemata'
+
+def download_ecs_schema(ecs_major_version, es_major)
+  $stderr.puts("Vendoring ECS #{ecs_major_version} template for Elasticsearch #{es_major}")
+  require 'net/http'
+  require 'json'
+  Net::HTTP.start('raw.githubusercontent.com', :use_ssl => true) do |http|
+    ecs_release_tag = ECS_VERSIONS.fetch(ecs_major_version)
+    response = http.get("/elastic/ecs/#{ecs_release_tag}/generated/elasticsearch/#{es_major}/template.json")
+    fail "#{response.code} #{response.message}" unless (200...300).cover?(response.code.to_i)
+    template_directory = File.expand_path("../lib/logstash/outputs/elasticsearch/templates/ecs-#{ecs_major_version}", __FILE__)
+    Dir.mkdir(template_directory) unless File.exists?(template_directory)
+    File.open(File.join(template_directory, "/elasticsearch-#{es_major}x.json"), "w") do |handle|
+      handle.write(replace_index_patterns(response.body, ECS_LOGSTASH_INDEX_PATTERNS))
+    end
+  end
+end
+
+def replace_index_patterns(template_json, replacement_index_patterns)
+  template_obj = JSON.load(template_json)
+  template_obj.update('index_patterns' => replacement_index_patterns)
+  JSON.pretty_generate(template_obj)
+end

docs/index.asciidoc

@@ -58,6 +58,20 @@ https://www.elastic.co/cloud/elasticsearch-service[hosted {es} Service] on
 Elastic Cloud. The Elasticsearch Service is available on AWS, Google Cloud
 Platform, and Microsoft Azure. {ess-trial}[Try the {es} Service for free].
 
+==== Compatibility with the Elastic Common Schema (ECS)
+
+This plugin will persist events to Elasticsearch in the shape produced by
+your pipeline, and _cannot_ be used to re-shape the event structure into a
+shape that complies with ECS. To produce events that fully comply with ECS,
+you will need to populate ECS-defined fields throughout your pipeline
+definition.
+
+However, the Elasticsearch Index Templates it manages can be configured to
+be ECS-compatible by setting <<plugins-{type}s-{plugin}-ecs_compatibility>>.
+By having an ECS-compatible template in place, we can ensure that Elasticsearch
+is prepared to create and index fields in a way that is compatible with ECS,
+and will correctly reject events with fields that conflict and cannot be coerced.
+
 ==== Writing to different indices: best practices
 
 [NOTE]
@@ -234,6 +248,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-doc_as_upsert>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-document_id>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-document_type>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
 | <<plugins-{type}s-{plugin}-failure_type_logging_whitelist>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-healthcheck_path>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-hosts>> |<<uri,uri>>|No
@@ -393,6 +408,25 @@ If you don't set a value for this option:
 - for elasticsearch clusters 6.x: the value of 'doc' will be used;
 - for elasticsearch clusters 5.x and below: the event's 'type' field will be used, if the field is not present the value of 'doc' will be used.
 
+[id="plugins-{type}s-{plugin}-ecs_compatibility"]
+===== `ecs_compatibility`
+
+* Value type is <<string,string>>
+* Supported values are:
+** `disabled`: does not provide ECS-compatible templates
+** `v1`: provides defaults that are compatible with v1 of the Elastic Common Schema
+* 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)],
+including the installation of ECS-compatible index templates.
+The value of this setting affects the _default_ values of:
+
+* <<plugins-{type}s-{plugin}-index>>
+* <<plugins-{type}s-{plugin}-template_name>>
+* <<plugins-{type}s-{plugin}-ilm_rollover_alias>>
+
 [id="plugins-{type}s-{plugin}-failure_type_logging_whitelist"]
 ===== `failure_type_logging_whitelist` 
 
@@ -500,7 +534,9 @@ NOTE: If this setting is specified, the policy must already exist in Elasticsear
 ===== `ilm_rollover_alias`
 
   * Value type is <<string,string>>
-  * Default value is `logstash`
+  * Default value depends on whether <<plugins-{type}s-{plugin}-ecs_compatibility>> is enabled:
+  ** ECS Compatibility disabled: `logstash`
+  ** ECS Compatibility enabled: `ecs-logstash`
 
 The rollover alias is the alias where indices managed using Index Lifecycle Management will be written to.
 
@@ -514,7 +550,9 @@ NOTE: `ilm_rollover_alias` does NOT support dynamic variable substitution as `in
 ===== `index` 
 
   * Value type is <<string,string>>
-  * Default value is `"logstash-%{+yyyy.MM.dd}"`
+  * Default value depends on whether <<plugins-{type}s-{plugin}-ecs_compatibility>> is enabled:
+  ** ECS Compatibility disabled: `"logstash-%{+yyyy.MM.dd}"`
+  ** ECS Compatibility enabled: `"ecs-logstash-%{+yyyy.MM.dd}"`
 
 The index to write events to. This can be dynamic using the `%{foo}` syntax.
 The default value will partition your indices by day so you can more easily
@@ -548,7 +586,8 @@ Set the keystore password
   * Default value is `true`
 
 From Logstash 1.3 onwards, a template is applied to Elasticsearch during
-Logstash's startup if one with the name `template_name` does not already exist.
+Logstash's startup if one with the name <<plugins-{type}s-{plugin}-template_name>>
+does not already exist.
 By default, the contents of this template is the default template for
 `logstash-%{+YYYY.MM.dd}` which always matches indices based on the pattern
 `logstash-*`.  Should you require support for other index names, or would like
@@ -799,7 +838,10 @@ If not set, the included template will be used.
 ===== `template_name` 
 
   * Value type is <<string,string>>
-  * Default value is `"logstash"`
+  * Default value depends on whether <<plugins-{type}s-{plugin}-ecs_compatibility>> is enabled:
+  ** ECS Compatibility disabled: `logstash`
+  ** ECS Compatibility enabled: `ecs-logstash`
+
 
 This configuration option defines how the template is named inside Elasticsearch.
 Note that if you have used the template management features and subsequently

lib/logstash/outputs/elasticsearch.rb

@@ -92,6 +92,8 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   require "logstash/outputs/elasticsearch/common"
   require "logstash/outputs/elasticsearch/ilm"
 
+  require 'logstash/plugin_mixins/ecs_compatibility_support'
+
   # Protocol agnostic (i.e. non-http, non-java specific) configs go here
   include(LogStash::Outputs::ElasticSearch::CommonConfigs)
 
@@ -101,6 +103,9 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # Methods for ILM support
   include(LogStash::Outputs::ElasticSearch::Ilm)
 
+  # ecs_compatibility option, provided by Logstash core or the support adapter.
+  include(LogStash::PluginMixins::ECSCompatibilitySupport)
+
   config_name "elasticsearch"
 
   # The Elasticsearch action to perform. Valid actions are:
@@ -242,6 +247,34 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # Custom Headers to send on each request to elasticsearch nodes
   config :custom_headers, :validate => :hash, :default => {}
 
+  def initialize(*params)
+    super
+    setup_ecs_compatibility_related_defaults
+  end
+
+  def setup_ecs_compatibility_related_defaults
+    case ecs_compatibility
+    when :disabled
+      @default_index = "logstash-%{+yyyy.MM.dd}"
+      @default_ilm_rollover_alias = "logstash"
+      @default_template_name = 'logstash'
+    when :v1
+      @default_index = "ecs-logstash-%{+yyyy.MM.dd}"
+      @default_ilm_rollover_alias = "ecs-logstash"
+      @default_template_name = 'ecs-logstash'
+    else
+      fail("unsupported ECS Compatibility `#{ecs_compatibility}`")
+    end
+
+    @index ||= default_index
+    @ilm_rollover_alias ||= default_ilm_rollover_alias
+    @template_name ||= default_template_name
+  end
+
+  attr_reader :default_index
+  attr_reader :default_ilm_rollover_alias
+  attr_reader :default_template_name
+
   # @override to handle proxy => '' as if none was set
   def config_init(params)
     proxy = params['proxy']

lib/logstash/outputs/elasticsearch/common.rb

@@ -60,8 +60,8 @@ def successful_connection?
       !!maximum_seen_major_version
     end
 
-    def use_event_type?(client)
-      client.maximum_seen_major_version < 8
+    def use_event_type?
+      maximum_seen_major_version < 8
     end
 
     # Convert the event into a 3-tuple of action, params, and event
@@ -74,7 +74,7 @@ def event_action_tuple(event)
         routing_field_name => @routing ? event.sprintf(@routing) : nil
       }
 
-      params[:_type] = get_event_type(event) if use_event_type?(client)
+      params[:_type] = get_event_type(event) if use_event_type?
 
       if @pipeline
         params[:pipeline] = event.sprintf(@pipeline)
@@ -347,11 +347,11 @@ def get_event_type(event)
       type = if @document_type
                event.sprintf(@document_type)
              else
-               if client.maximum_seen_major_version < 6
+               if maximum_seen_major_version < 6
                  event.get("type") || DEFAULT_EVENT_TYPE_ES6
-               elsif client.maximum_seen_major_version == 6
+               elsif maximum_seen_major_version == 6
                  DEFAULT_EVENT_TYPE_ES6
-               elsif client.maximum_seen_major_version == 7
+               elsif maximum_seen_major_version == 7
                  DEFAULT_EVENT_TYPE_ES7
                else
                  nil
@@ -436,7 +436,7 @@ def safe_bulk(actions)
     end
 
     def default_index?(index)
-      @index == LogStash::Outputs::ElasticSearch::CommonConfigs::DEFAULT_INDEX_NAME
+      @index == @default_index
     end
 
     def dlq_enabled?

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -17,7 +17,7 @@ def self.included(mod)
       # For weekly indexes ISO 8601 format is recommended, eg. logstash-%{+xxxx.ww}.
       # LS uses Joda to format the index pattern from event timestamp.
       # Joda formats are defined http://www.joda.org/joda-time/apidocs/org/joda/time/format/DateTimeFormat.html[here].
-      mod.config :index, :validate => :string, :default => DEFAULT_INDEX_NAME
+      mod.config :index, :validate => :string
 
       mod.config :document_type, 
         :validate => :string, 
@@ -44,7 +44,7 @@ def self.included(mod)
       # `curl -XDELETE <http://localhost:9200/_template/OldTemplateName?pretty>`
       #
       # where `OldTemplateName` is whatever the former setting was.
-      mod.config :template_name, :validate => :string, :default => "logstash"
+      mod.config :template_name, :validate => :string
 
       # You can set the path to your own template here, if you so desire.
       # If not set, the included template will be used.
@@ -153,7 +153,7 @@ def self.included(mod)
       mod.config :ilm_enabled, :validate => [true, false, 'true', 'false', 'auto'], :default => 'auto'
 
       # Rollover alias used for indexing data. If rollover alias doesn't exist, Logstash will create it and map it to the relevant index
-      mod.config :ilm_rollover_alias, :validate => :string, :default => DEFAULT_ROLLOVER_ALIAS
+      mod.config :ilm_rollover_alias, :validate => :string
 
       # appends “{now/d}-000001” by default for new index creation, subsequent rollover indices will increment based on this pattern i.e. “000002”
       # {now/d} is date math, and will insert the appropriate value automatically.

lib/logstash/outputs/elasticsearch/ilm.rb

@@ -12,7 +12,7 @@ def setup_ilm
     end
 
     def default_rollover_alias?(rollover_alias)
-      rollover_alias == LogStash::Outputs::ElasticSearch::DEFAULT_ROLLOVER_ALIAS
+      rollover_alias == default_ilm_rollover_alias
     end
 
     def ilm_alias_set?

lib/logstash/outputs/elasticsearch/template_manager.rb

@@ -3,14 +3,15 @@ class TemplateManager
     # To be mixed into the elasticsearch plugin base
     def self.install_template(plugin)
       return unless plugin.manage_template
-      if plugin.template.nil?
-        plugin.logger.info("Using default mapping template")
-      else
+      if plugin.template
         plugin.logger.info("Using mapping template from", :path => plugin.template)
+        template = read_template_file(plugin.template)
+      else
+        plugin.logger.info("Using a default mapping template", :es_version => plugin.maximum_seen_major_version,
+                                                               :ecs_compatibility => plugin.ecs_compatibility)
+        template = load_default_template(plugin.maximum_seen_major_version, plugin.ecs_compatibility)
       end
 
-
-      template = get_template(plugin.template, plugin.maximum_seen_major_version)
       add_ilm_settings_to_template(plugin, template) if plugin.ilm_in_use?
       plugin.logger.info("Attempting to install template", :manage_template => template)
       install(plugin.client, template_name(plugin), template, plugin.template_overwrite)
@@ -19,9 +20,11 @@ def self.install_template(plugin)
     end
 
     private
-    def self.get_template(path, es_major_version)
-      template_path = path || default_template_path(es_major_version)
+    def self.load_default_template(es_major_version, ecs_compatibility)
+      template_path = default_template_path(es_major_version, ecs_compatibility)
       read_template_file(template_path)
+    rescue => e
+      fail "Failed to load default template for Elasticsearch v#{es_major_version} with ECS #{ecs_compatibility}; caused by: #{e.inspect}"
     end
 
     def self.install(client, template_name, template, template_overwrite)
@@ -46,9 +49,9 @@ def self.template_name(plugin)
       plugin.ilm_in_use? && !plugin.original_params.key?('template_name') ? plugin.ilm_rollover_alias : plugin.template_name
     end
 
-    def self.default_template_path(es_major_version)
+    def self.default_template_path(es_major_version, ecs_compatibility=:disabled)
       template_version = es_major_version == 1 ? 2 : es_major_version
-      default_template_name = "elasticsearch-template-es#{template_version}x.json"
+      default_template_name = "templates/ecs-#{ecs_compatibility}/elasticsearch-#{template_version}x.json"
       ::File.expand_path(default_template_name, ::File.dirname(__FILE__))
     end
 

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '10.5.1'
+  s.version         = '10.6.0'
 
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"
@@ -25,6 +25,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'stud', ['>= 0.0.17', '~> 0.0']
   s.add_runtime_dependency 'cabin', ['~> 0.6']
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
+  s.add_runtime_dependency 'logstash-mixin-ecs_compatibility_support', '~>1.0'
 
   s.add_development_dependency 'logstash-codec-plain'
   s.add_development_dependency 'logstash-devutils'

spec/integration/outputs/ilm_spec.rb

@@ -249,9 +249,9 @@
     }
     let (:small_max_doc_policy) { max_docs_policy(3) }
     let (:large_max_doc_policy) { max_docs_policy(1000000) }
-    let (:expected_index) { LogStash::Outputs::ElasticSearch::DEFAULT_ROLLOVER_ALIAS }
+    let (:expected_index) { elasticsearch_output_plugin.default_ilm_rollover_alias }
 
-    subject { LogStash::Outputs::ElasticSearch.new(settings) }
+    subject(:elasticsearch_output_plugin) { LogStash::Outputs::ElasticSearch.new(settings) }
 
     before :each do
       # Delete all templates first.

spec/unit/outputs/elasticsearch/template_manager_spec.rb

@@ -8,18 +8,24 @@
   describe ".default_template_path" do
     context "elasticsearch 1.x" do
       it "chooses the 2x template" do
-        expect(described_class.default_template_path(1)).to match(/elasticsearch-template-es2x.json/)
+        expect(described_class.default_template_path(1)).to end_with("/templates/ecs-disabled/elasticsearch-2x.json")
       end
     end
     context "elasticsearch 2.x" do
       it "chooses the 2x template" do
-        expect(described_class.default_template_path(2)).to match(/elasticsearch-template-es2x.json/)
+        expect(described_class.default_template_path(2)).to end_with("/templates/ecs-disabled/elasticsearch-2x.json")
       end
     end
     context "elasticsearch 5.x" do
       it "chooses the 5x template" do
-        expect(described_class.default_template_path(5)).to match(/elasticsearch-template-es5x.json/)
+        expect(described_class.default_template_path(5)).to end_with("/templates/ecs-disabled/elasticsearch-5x.json")
       end
     end
   end
+
+  context 'when ECS v1 is requested' do
+    it 'resolves' do
+      expect(described_class.default_template_path(7, :v1)).to end_with("/templates/ecs-v1/elasticsearch-7x.json")
+    end
+  end
 end

spec/unit/outputs/elasticsearch_spec.rb

@@ -6,7 +6,7 @@
 describe LogStash::Outputs::ElasticSearch do
   subject { described_class.new(options) }
   let(:options) { {} }
-  let(:maximum_seen_major_version) { rand(100) }
+  let(:maximum_seen_major_version) { [1,2,5,6,7,8].sample }
 
   let(:do_register) { true }