Plugin: implement ecs_compatibility for all plugins

 - Default value supplied by `pipeline.ecs_compatibility`
 - Warns when default used (default will change in next major)
 - Warns when value other than `disabled` is used (BETA: by opting in, users
   can accidentally consume breaking changes in a minor release of Logstash,
   such as one that includes an upgrade to a plugin that introduces ECS
   compatibility logic)

docker/data/logstash/env2yaml/env2yaml.go

@@ -57,6 +57,7 @@ func normalizeSetting(setting string) (string, error) {
 		"pipeline.batch.delay",
 		"pipeline.unsafe_shutdown",
 		"pipeline.java_execution",
+		"pipeline.ecs_compatibility"
 		"pipeline.plugin_classloaders",
 		"path.config",
 		"config.string",

logstash-core/lib/logstash/agent.rb

@@ -75,6 +75,14 @@ def initialize(settings = LogStash::SETTINGS, source_loader = nil)
       logger.warn("deprecated setting `config.field_reference.parser` set; field reference parsing is strict by default")
     end
 
+    if @settings.set?('pipeline.ecs_compatibility')
+      ecs_compatibility_value = settings.get('pipeline.ecs_compatibility')
+      if ecs_compatibility_value != 'disabled'
+        logger.warn("Setting `pipeline.ecs_compatibility` given as `#{ecs_compatibility_value}`; " +
+                    "values other than `disabled` are currently considered BETA and may have unintended consequences when upgrading minor versions of Logstash.")
+      end
+    end
+
     # This is for backward compatibility in the tests
     if source_loader.nil?
       @source_loader = LogStash::Config::SourceLoader.new

logstash-core/lib/logstash/config/mixin.rb

@@ -201,7 +201,7 @@ def config(name, opts={})
       name = name.to_s if name.is_a?(Symbol)
       @config[name] = opts  # ok if this is empty
 
-      if name.is_a?(String)
+      if name.is_a?(String) && opts.fetch(:attr_accessor, true)
         define_method(name) { instance_variable_get("@#{name}") }
         define_method("#{name}=") { |v| instance_variable_set("@#{name}", v) }
       end

logstash-core/lib/logstash/environment.rb

@@ -62,6 +62,7 @@ module Environment
            Setting::Boolean.new("pipeline.plugin_classloaders", false),
            Setting::Boolean.new("pipeline.separate_logs", false),
    Setting::CoercibleString.new("pipeline.ordered", "auto", true, ["auto", "true", "false"]),
+   Setting::CoercibleString.new("pipeline.ecs_compatibility", "disabled", true, %w(disabled v1 v2)),
                     Setting.new("path.plugins", Array, []),
     Setting::NullableString.new("interactive", nil, false),
            Setting::Boolean.new("config.debug", false),

logstash-core/lib/logstash/plugin.rb

@@ -16,6 +16,7 @@
 # under the License.
 
 require "logstash/config/mixin"
+require "logstash/plugins/ecs_compatibility_support"
 require "concurrent"
 require "securerandom"
 
@@ -29,6 +30,7 @@ class LogStash::Plugin
   NL = "\n"
 
   include LogStash::Config::Mixin
+  include LogStash::Plugins::ECSCompatibilitySupport
 
   # Disable or enable metric logging for this specific plugin instance
   # by default we record all the metrics we can, but you can disable metrics collection

logstash-core/lib/logstash/plugins/ecs_compatibility_support.rb

@@ -0,0 +1,53 @@
+module LogStash
+  module Plugins
+    module ECSCompatibilitySupport
+      def self.included(base)
+        base.extend(ArgumentValidator)
+        base.config(:ecs_compatibility, :validate => :ecs_compatibility_argument,
+                                        :attr_accessor => false)
+      end
+
+      MUTEX = Mutex.new
+      private_constant :MUTEX
+
+      def ecs_compatibility
+        @_ecs_compatibility || MUTEX.synchronize do
+          @_ecs_compatibility ||= begin
+            # use config_init-set value if present
+            break @ecs_compatibility unless @ecs_compatibility.nil?
+
+            pipeline = execution_context.pipeline
+            pipeline_settings = pipeline && pipeline.settings
+            pipeline_settings ||= LogStash::SETTINGS
+
+            if !pipeline_settings.set?('pipeline.ecs_compatibility')
+              deprecation_logger.deprecated("Relying on default value of `pipeline.ecs_compatibility`, which may change in a future major release of Logstash. " +
+                                            "To avoid unexpected changes when upgrading Logstash, please explicitly declare your desired ECS Compatibility mode.")
+            end
+
+            pipeline_settings.get_value('pipeline.ecs_compatibility').to_sym
+          end
+        end
+      end
+
+      module ArgumentValidator
+        V_PREFIXED_INTEGER_PATTERN = %r(\Av[1-9][0-9]?\Z).freeze
+        private_constant :V_PREFIXED_INTEGER_PATTERN
+
+        def validate_value(value, validator)
+          return super unless validator == :ecs_compatibility_argument
+
+          value = deep_replace(value)
+          value = hash_or_array(value)
+
+          if value.size == 1
+            return true, :disabled if value.first.to_s == 'disabled'
+            return true, value.first.to_sym if value.first.to_s =~ V_PREFIXED_INTEGER_PATTERN
+          end
+
+          return false, "Expected a v-prefixed integer major-version number (e.g., `v1`) or the literal `disabled`, got #{value.inspect}"
+        end
+      end
+    end
+  end
+end

logstash-core/lib/logstash/runner.rb

@@ -154,6 +154,11 @@ class LogStash::Runner < Clamp::StrictCommand
     :attribute_name => "pipeline.unsafe_shutdown",
     :default => LogStash::SETTINGS.get_default("pipeline.unsafe_shutdown")
 
+  option ["--pipeline.ecs_compatibility"], "STRING",
+    I18n.t("logstash.runner.flag.ecs_compatibility"),
+    :attribute_name => "pipeline.ecs_compatibility",
+    :default => LogStash::SETTINGS.get_default('pipeline.ecs_compatibility')
+
   # Data Path Setting
   option ["--path.data"] , "PATH",
     I18n.t("logstash.runner.flag.datapath"),

logstash-core/lib/logstash/settings.rb

@@ -48,6 +48,7 @@ class Settings
       "pipeline.system",
       "pipeline.workers",
       "pipeline.ordered",
+      "pipeline.ecs_compatibility",
       "queue.checkpoint.acks",
       "queue.checkpoint.interval",
       "queue.checkpoint.writes",

logstash-core/locales/en.yml

@@ -362,6 +362,21 @@ en:
           if there are still inflight events in memory.
           By default, logstash will refuse to quit until all
           received events have been pushed to the outputs.
+        ecs_compatibility: |+
+          Sets the pipeline's default value for `ecs_compatibility`,
+          a setting that is available to plugins that implement
+          an ECS Compatibility mode for use with the Elastic Common
+          Schema.
+          Possible values are:
+           - disabled (default)
+           - v1
+           - v2
+          This option allows the early opt-in (or preemptive opt-out)
+          of ECS Compatibility modes in plugins, which is scheduled to
+          be on-by-default in a future major release of Logstash.
+
+          Values other than `disabled` are currently considered BETA,
+          and may produce unintended consequences when upgrading Logstash.
         rubyshell: |+
           Drop to shell instead of running as normal.
           Valid shells are "irb" and "pry"

logstash-core/spec/logstash/plugin_spec.rb

@@ -417,6 +417,66 @@ def register; end
     end
   end
 
+  describe "#ecs_compatibility" do
+    let(:plugin_class) do
+      Class.new(LogStash::Filters::Base) do
+        config_name "ecs_validator_sample"
+        def register; end
+      end
+    end
+    let(:config) { Hash.new }
+    let(:instance) { plugin_class.new(config) }
+
+    let(:deprecation_logger_stub) { double('DeprecationLogger').as_null_object }
+    before(:each) do
+      allow(plugin_class).to receive(:deprecation_logger).and_return(deprecation_logger_stub)
+    end
+
+    context 'when plugin initialized with explicit value' do
+      let(:config) { super().merge("ecs_compatibility" => "v17") }
+      it 'returns the explicitly-given value' do
+        expect(instance.ecs_compatibility).to eq(:v17)
+      end
+    end
+
+    context 'when plugin is not initialized with an explicit value' do
+      let(:settings_stub) { LogStash::SETTINGS.clone }
+
+      before(:each) do
+        allow(settings_stub).to receive(:get_value).with(anything).and_call_original # allow spies
+        stub_const('LogStash::SETTINGS', settings_stub)
+      end
+
+      context 'and pipeline-level setting is explicitly `v1`' do
+        let(:settings_stub) do
+          super().tap do |settings|
+            settings.set_value('pipeline.ecs_compatibility', 'v1')
+          end
+        end
+        it 'reads the setting' do
+          expect(instance.ecs_compatibility).to eq(:v1)
+
+          expect(settings_stub).to have_received(:get_value)
+        end
+      end
+
+      context 'and pipeline-level setting is not specified' do
+        it 'emits a deprecation warning about using the default which may change' do
+          instance.ecs_compatibility
+
+          expect(deprecation_logger_stub).to have_received(:deprecated) do |message|
+            expect(message).to include("Relying on default value of `pipeline.ecs_compatibility`")
+          end
+        end
+        it 'returns `disabled`' do
+          # Default value of `pipeline.ecs_compatibility`
+          expect(instance.ecs_compatibility).to eq(:disabled)
+        end
+      end
+    end
+
+  end
+
   describe "deprecation logger" do
     let(:config) do
       {