Exclude substitution refinement on pipelines.yml (elastic#16375)

* Exclude substitution refinement on pipelines.yml (applies on ENV vars and logstash.yml where env2yaml saves vars)

* Safety integration test for pipeline config.string contains ENV .

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

@@ -93,7 +93,7 @@ def config_init(params)
 
     # Resolve environment variables references
     params.each do |name, value|
-      params[name.to_s] = deep_replace(value)
+      params[name.to_s] = deep_replace(value, true)
     end
 
     # Intercept codecs that have not been instantiated

logstash-core/lib/logstash/settings.rb

@@ -187,7 +187,7 @@ def reset
 
     def from_yaml(yaml_path, file_name = "logstash.yml")
       settings = read_yaml(::File.join(yaml_path, file_name))
-      self.merge(deep_replace(flatten_hash(settings)), true)
+      self.merge(deep_replace(flatten_hash(settings), true), true)
       self
     end
 

logstash-core/lib/logstash/util/substitution_variables.rb

@@ -28,19 +28,19 @@ module ::LogStash::Util::SubstitutionVariables
   SECRET_STORE = ::LogStash::Util::LazySingleton.new { load_secret_store }
   private_constant :SECRET_STORE
 
-  # Recursive method to replace substitution variable references in parameters
-  def deep_replace(value)
+  # Recursive method to replace substitution variable references in parameters and refine if required
+  def deep_replace(value, refine = false)
     if value.is_a?(Hash)
       value.each do |valueHashKey, valueHashValue|
-        value[valueHashKey.to_s] = deep_replace(valueHashValue)
+        value[valueHashKey.to_s] = deep_replace(valueHashValue, refine)
       end
     else
       if value.is_a?(Array)
         value.each_with_index do |single_value, i|
-          value[i] = deep_replace(single_value)
+          value[i] = deep_replace(single_value, refine)
         end
       else
-        return replace_placeholders(value)
+        return replace_placeholders(value, refine)
       end
     end
   end
@@ -49,9 +49,11 @@ def deep_replace(value)
   # Process following patterns : ${VAR}, ${VAR:defaultValue}
   # If value matches the pattern, returns the following precedence : Secret store value, Environment entry value, default value as provided in the pattern
   # If the value does not match the pattern, the 'value' param returns as-is
-  def replace_placeholders(value)
+  # When setting refine to true, substituted value will be cleaned against escaped single/double quotes
+  #   and generates array if resolved substituted value is array string
+  def replace_placeholders(value, refine)
     if value.kind_of?(::LogStash::Util::Password)
-      interpolated = replace_placeholders(value.value)
+      interpolated = replace_placeholders(value.value, refine)
       return ::LogStash::Util::Password.new(interpolated)
     end
     return value unless value.is_a?(String)
@@ -80,8 +82,8 @@ def replace_placeholders(value)
       replacement.to_s
     end
 
-    # no further action need if substitution didn't happen
-    return placeholder_value unless is_placeholder_found
+    # no further action need if substitution didn't happen or refine isn't required
+    return placeholder_value unless is_placeholder_found && refine
 
     # ENV ${var} value may carry single quote or escaped double quote
     # or single/double quoted entries in array string, needs to be refined

qa/integration/specs/multiple_pipeline_spec.rb

@@ -34,6 +34,7 @@
 
   let(:temporary_out_file_1) { Stud::Temporary.pathname }
   let(:temporary_out_file_2) { Stud::Temporary.pathname }
+  let(:temporary_out_file_3) { Stud::Temporary.pathname }
 
   let(:pipelines) {[
     {
@@ -47,6 +48,12 @@
       "pipeline.workers" => 1,
       "pipeline.batch.size" => 1,
       "config.string" => "input { generator { count => 1 } } output { file { path => \"#{temporary_out_file_2}\" } }"
+    },
+    {
+      "pipeline.id" => "config-string-with-env-var-pipeline",
+      "pipeline.workers" => 1,
+      "pipeline.batch.size" => 1,
+      "config.string" => "input { generator { count => 1 } } output { file { path => \"${TEMP_FILE_PATH}\" } }"
     }
   ]}
 
@@ -65,6 +72,7 @@
 
   it "executes the multiple pipelines" do
     logstash_service = @fixture.get_service("logstash")
+    logstash_service.env_variables = {'TEMP_FILE_PATH' => temporary_out_file_3}
     logstash_service.spawn_logstash("--path.settings", settings_dir, "--log.level=debug")
     try(retry_attempts) do
       expect(logstash_service.exited?).to be(true)
@@ -74,6 +82,8 @@
     expect(IO.readlines(temporary_out_file_1).size).to eq(1)
     expect(File.exist?(temporary_out_file_2)).to be(true)
     expect(IO.readlines(temporary_out_file_2).size).to eq(1)
+    expect(File.exist?(temporary_out_file_3)).to be(true)
+    expect(IO.readlines(temporary_out_file_3).size).to eq(1)
   end
 
   context 'effectively-empty pipelines.yml file' do