Allow events to directly encode the downstream pipeline to use.

[yaauie]

When pipeline parameter for this plugin is unset and an event has a [@metadata][target_ingest_pipeline] field containing a string value, use that value in the event tuple's pipeline.

Note: we cannot implement this as a default value for the pipeline parameter because the field's absence would cause us to send the uninterpolated %{[@metadata][target_ingest_pipeline]} as the pipeline name.

Instead, we must interpolate the result of a non-nil (`@pipeline || event.get("[@metadata][target_ingest_pipeline]")&.to_s)

Likely solution:

diff --git a/lib/logstash/outputs/elasticsearch.rb b/lib/logstash/outputs/elasticsearch.rb
index 0da3c90..93e486e 100644
--- a/lib/logstash/outputs/elasticsearch.rb
+++ b/lib/logstash/outputs/elasticsearch.rb
@@ -516,19 +516,23 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
         routing_field_name => @routing ? event.sprintf(@routing) : nil
     }
 
-    if @pipeline
-      value = event.sprintf(@pipeline)
-      # convention: empty string equates to not using a pipeline
-      # this is useful when using a field reference in the pipeline setting, e.g.
-      #      elasticsearch {
-      #        pipeline => "%{[@metadata][pipeline]}"
-      #      }
-      params[:pipeline] = value unless value.empty?
-    end
+    target_pipeline = resolve_pipeline(event)
+    # convention: empty string equates to not using a pipeline
+    # this is useful when using a field reference in the pipeline setting, e.g.
+    #      elasticsearch {
+    #        pipeline => "%{[@metadata][pipeline]}"
+    #      }
+    params[:pipeline] = target_pipeline unless target_pipeline.empty?
 
     params
   end
 
+  def resolve_pipeline(event)
+    pipeline_template = @pipeline || event.get("[@metadata][target_ingest_pipeline]")&.to_s
+
+    pipeline_template && event.sprintf(pipeline_template)
+  end
+
   @@plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-elasticsearch-/ }
 
   @@plugins.each do |plugin|