Unify ConfigPart and SourceMeta into SourceWithMetadata

This unifies the two different config classes that represented mainly
the same data. While this does expose a plain java class into ruby
this works fine because ruby only needs to access and set values, not
work with ruby return types.

Fixes elastic#7003

Fixes elastic#7004

logstash-core/lib/logstash/agent.rb

@@ -134,10 +134,10 @@ def converge_state_and_update
 
     unless results.success?
       if auto_reload?
-        logger.debug("Count not fetch the configuration to converge, will retry", :message => results.error, :retrying_in => @reload_interval)
+        logger.debug("Could not fetch the configuration to converge, will retry", :message => results.error, :retrying_in => @reload_interval)
         return
       else
-        raise "Count not fetch the configuration, message: #{results.error}"
+        raise "Could not fetch the configuration, message: #{results.error}"
       end
     end
 

logstash-core/lib/logstash/compiler/lscl.rb

@@ -3,14 +3,14 @@
 require "treetop"
 require "logstash/compiler/treetop_monkeypatches"
 java_import org.logstash.config.ir.DSL
-java_import org.logstash.config.ir.SourceMetadata
+java_import org.logstash.common.SourceWithMetadata
 
 module LogStashCompilerLSCLGrammar; module LogStash; module Compiler; module LSCL; module AST
   # Helpers for parsing LSCL files
   module Helpers
     def source_meta
       line, column = line_and_column
-      org.logstash.config.ir.SourceMetadata.new(source_file, line, column, self.text_value)
+      org.logstash.common.SourceWithMetadata.new(source_file, line, column, self.text_value)
     end
 
     def source_file=(value)
@@ -39,7 +39,7 @@ def line_and_column
     end
 
     def empty_source_meta()
-      org.logstash.config.ir.SourceMetadata.new()
+      org.logstash.common.SourceWithMetadata.new()
     end
 
     def jdsl

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

@@ -10,7 +10,9 @@ class PipelineConfig
     def initialize(source, pipeline_id, config_parts, settings)
       @source = source
       @pipeline_id = pipeline_id
-      @config_parts = Array(config_parts).sort_by { |config_part| [config_part.reader.to_s, config_part.source_id] }
+      # We can't use Array() since config_parts may be a java object!
+      config_parts_array = config_parts.is_a?(Array) ? config_parts : [config_parts]
+      @config_parts = config_parts_array.sort_by { |config_part| [config_part.protocol.to_s, config_part.id] }
       @settings = settings
       @read_at = Time.now
     end
@@ -20,7 +22,7 @@ def config_hash
     end
 
     def config_string
-      @config_string = config_parts.collect(&:config_string).join("\n")
+      @config_string = config_parts.collect(&:text).join("\n")
     end
 
     def system?
@@ -36,8 +38,8 @@ def display_debug_information
       logger.debug("Config from source", :source => source, :pipeline_id => pipeline_id)
 
       config_parts.each do |config_part|
-        logger.debug("Config string", :reader => config_part.reader, :source_id => config_part.source_id)
-        logger.debug("\n\n#{config_part.config_string}")
+        logger.debug("Config string", :protocol => config_part.protocol, :id => config_part.id)
+        logger.debug("\n\n#{config_part.text}")
       end
       logger.debug("Merged config")
       logger.debug("\n\n#{config_string}")

logstash-core/lib/logstash/config/source/local.rb

@@ -1,6 +1,5 @@
 # encoding: utf-8
 require "logstash/config/source/base"
-require "logstash/config/config_part"
 require "logstash/config/pipeline_config"
 require "logstash/util/loggable"
 require "logstash/errors"
@@ -20,7 +19,7 @@ module LogStash module Config module Source
   class Local < Base
     class ConfigStringLoader
       def self.read(config_string)
-        [ConfigPart.new(self.name, "config_string", config_string)]
+        [org.logstash.common.SourceWithMetadata.new("string", "config_string", config_string)]
       end
     end
 
@@ -51,7 +50,8 @@ def read
           config_string = ::File.read(file)
 
           if valid_encoding?(config_string)
-            config_parts << ConfigPart.new(self.class.name, file, config_string)
+            part = org.logstash.common.SourceWithMetadata.new("file", file, config_string)
+            config_parts << part
           else
             encoding_issue_files << file
           end
@@ -110,7 +110,7 @@ def self.read(uri)
           # since we have fetching config we wont follow any redirection.
           case response.code.to_i
           when 200
-            [ConfigPart.new(self.name, uri.to_s, response.body)]
+            [org.logstash.common.SourceWithMetadata.new(uri.scheme, uri.to_s, response.body)]
           when 302
             raise LogStash::ConfigLoadingError, I18n.t("logstash.runner.configuration.fetch-failed", :path => uri.to_s, :message => "We don't follow redirection for remote configuration")
           when 404
@@ -149,7 +149,7 @@ def pipeline_configs
 
       add_missing_default_inputs_or_outputs(config_parts)
 
-      PipelineConfig.new(self.class, PIPELINE_ID, config_parts, @settings)
+      [PipelineConfig.new(self.class, PIPELINE_ID, config_parts, @settings)]
     end
 
     def match?
@@ -161,13 +161,13 @@ def match?
     # if its not the case we will add stdin and stdout
     # this is for backward compatibility reason
     def add_missing_default_inputs_or_outputs(config_parts)
-      if !config_parts.any? { |part| INPUT_BLOCK_RE.match(part.config_string) }
-        config_parts << LogStash::Config::ConfigPart.new(self.class.name, "default input", LogStash::Config::Defaults.input)
+      if !config_parts.any? { |part| INPUT_BLOCK_RE.match(part.text) }
+        config_parts << org.logstash.common.SourceWithMetadata.new(self.class.name, "default input", LogStash::Config::Defaults.input)
       end
 
       # include a default stdout output if no outputs given
-      if !config_parts.any? { |part| OUTPUT_BLOCK_RE.match(part.config_string) }
-        config_parts << LogStash::Config::ConfigPart.new(self.class.name, "default output", LogStash::Config::Defaults.output)
+      if !config_parts.any? { |part| OUTPUT_BLOCK_RE.match(part.text) }
+        config_parts << org.logstash.common.SourceWithMetadata.new(self.class.name, "default output", LogStash::Config::Defaults.output)
       end
     end
 

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

@@ -54,7 +54,7 @@ def fetch
       else
         begin
           pipeline_configs = sources_loaders
-            .collect { |source| Array(source.pipeline_configs) }
+            .collect { |source| source.pipeline_configs }
             .compact
             .flatten
 
@@ -68,7 +68,7 @@ def fetch
 
           SuccessfulFetch.new(pipeline_configs)
         rescue => e
-          logger.error("Could not fetch all the sources", :exception => e.class, :message => e.message)
+          logger.error("Could not fetch all the sources", :exception => e.class, :message => e.message, :backtrace => e.backtrace)
           FailedFetch.new(e.message)
         end
       end

logstash-core/spec/logstash/agent_spec.rb

@@ -3,7 +3,6 @@
 require "stud/temporary"
 require "logstash/inputs/generator"
 require "logstash/config/pipeline_config"
-require "logstash/config/config_part"
 require "logstash/config/source/local"
 require_relative "../support/mocks_classes"
 require "fileutils"

logstash-core/spec/logstash/compiler/compiler_spec.rb

@@ -1,5 +1,6 @@
 require "spec_helper"
 require "logstash/compiler"
+require "support/helpers"
 java_import Java::OrgLogstashConfigIr::DSL
 
 describe LogStash::Compiler do
@@ -62,7 +63,7 @@ def j
       let(:source) { "input { generator {} } output { }" }
 
       it "should attach correct source text for components" do
-        expect(compiled[:input].get_meta.getSourceText).to eql("generator {}")
+        expect(compiled[:input].get_meta.getText).to eql("generator {}")
       end
     end
 
@@ -188,12 +189,12 @@ def compose(*statements)
                                       ))
         end
 
-        it "should attach source_metadata with correct info to the statements" do
+        it "should attach source_with_metadata with correct info to the statements" do
           meta = compiled_section.statements.first.meta
-          expect(meta.getSourceText).to eql("aplugin { count => 1 }")
-          expect(meta.getSourceLine).to eql(2)
-          expect(meta.getSourceColumn).to eql(13)
-          expect(meta.getSourceFile).to eql(source_file)
+          expect(meta.text).to eql("aplugin { count => 1 }")
+          expect(meta.line).to eql(2)
+          expect(meta.column).to eql(13)
+          expect(meta.id).to eql(source_file)
           expect(compiled_section.statements.first.meta)
           expect(compiled_section)
         end

logstash-core/spec/logstash/config/pipeline_config_spec.rb

@@ -1,20 +1,19 @@
 # encoding: utf-8
 require "logstash/config/pipeline_config"
-require "logstash/config/config_part"
 require "logstash/config/source/local"
 
 describe LogStash::Config::PipelineConfig do
   let(:source) { LogStash::Config::Source::Local }
   let(:pipeline_id) { :main }
   let(:ordered_config_parts) do
     [
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/1", "input { generator1 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/2", "input { generator2 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/3", "input { generator3 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/4", "input { generator4 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/5", "input { generator5 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigPathLoader, "/tmp/6", "input { generator6 }"),
-      LogStash::Config::ConfigPart.new(LogStash::Config::Source::Local::ConfigStringLoader, "config_string", "input { generator1 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/1", "input { generator1 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/2", "input { generator2 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/3", "input { generator3 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/4", "input { generator4 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/5", "input { generator5 }"),
+      org.logstash.common.SourceWithMetadata.new("file", "/tmp/6", "input { generator6 }"),
+      org.logstash.common.SourceWithMetadata.new("string", "config_string", "input { generator1 }"),
     ]
   end
 
@@ -40,7 +39,7 @@
   end
 
   it "returns the merged `ConfigPart#config_string`" do
-    expect(subject.config_string).to eq(ordered_config_parts.collect(&:config_string).join("\n"))
+    expect(subject.config_string).to eq(ordered_config_parts.collect(&:text).join("\n"))
   end
 
   it "records when the config was read" do

logstash-core/spec/logstash/config/source/local_spec.rb

@@ -19,7 +19,7 @@
 
   it "returns a valid config part" do
     config_part = subject.read(config_string).first
-    expect(config_part).to be_a_config_part(described_class.to_s, "config_string", config_string)
+    expect(config_part).to be_a_source_with_metadata("string", "config_string", config_string)
   end
 end
 
@@ -71,17 +71,17 @@
 
       it "returns alphabetically sorted parts" do
         parts = subject.read(reader_config)
-        expect(parts.collect { |part| ::File.basename(part.source_id) }).to eq(files.keys.sort)
+        expect(parts.collect { |part| ::File.basename(part.id) }).to eq(files.keys.sort)
       end
 
       it "returns valid `config_parts`" do
         parts = subject.read(reader_config)
 
         parts.each do |part|
-          basename = ::File.basename(part.source_id)
+          basename = ::File.basename(part.id)
           file_path = ::File.join(directory, basename)
           content = files[basename]
-          expect(part).to be_a_config_part(described_class.to_s, file_path, content)
+          expect(part).to be_a_source_with_metadata("file", file_path, content)
         end
       end
     end
@@ -256,7 +256,7 @@
 
     it "returns a valid config part" do
       config_part = subject.read(remote_url).first
-      expect(config_part).to be_a_config_part(described_class.to_s, remote_url, config_string)
+      expect(config_part).to be_a_source_with_metadata("http", remote_url, config_string)
     end
   end
 
@@ -293,7 +293,7 @@
     end
 
     it "returns a merged config" do
-      expect(subject.pipeline_configs.config_string).to include(input_block, output_block, filter_block)
+      expect(subject.pipeline_configs.first.config_string).to include(input_block, output_block, filter_block)
     end
   end
 
@@ -303,7 +303,7 @@
     end
 
     it "returns a config" do
-      expect(subject.pipeline_configs.config_string).to include(filter_block)
+      expect(subject.pipeline_configs.first.config_string).to include(filter_block)
     end
   end
 
@@ -314,7 +314,7 @@
     end
 
     it "returns a config" do
-      expect(subject.pipeline_configs.config_string).to include(input_block)
+      expect(subject.pipeline_configs.first.config_string).to include(input_block)
     end
   end
 
@@ -342,7 +342,7 @@
     end
 
     it "returns a config" do
-      expect(subject.pipeline_configs.config_string).to include(input_block)
+      expect(subject.pipeline_configs.first.config_string).to include(input_block)
     end
 
     context "when `config.string` is set" do
@@ -354,7 +354,7 @@
       end
 
       it "returns a merged config" do
-        expect(subject.pipeline_configs.config_string).to include(input_block, filter_block)
+        expect(subject.pipeline_configs.first.config_string).to include(input_block, filter_block)
       end
     end
   end
@@ -364,31 +364,31 @@
       let(:settings) { mock_settings( "config.string" => "#{filter_block} #{output_block}") }
 
       it "add stdin input" do
-        expect(subject.pipeline_configs.config_string).to include(LogStash::Config::Defaults.input)
+        expect(subject.pipeline_configs.first.config_string).to include(LogStash::Config::Defaults.input)
       end
     end
 
     context "when the output block is missing" do
       let(:settings) { mock_settings( "config.string" => "#{input_block} #{filter_block}") }
 
       it "add stdout output" do
-        expect(subject.pipeline_configs.config_string).to include(LogStash::Config::Defaults.output)
+        expect(subject.pipeline_configs.first.config_string).to include(LogStash::Config::Defaults.output)
       end
     end
 
     context "when both the output block and input block are missing" do
       let(:settings) { mock_settings( "config.string" => "#{filter_block}") }
 
       it "add stdin and output" do
-        expect(subject.pipeline_configs.config_string).to include(LogStash::Config::Defaults.output, LogStash::Config::Defaults.input)
+        expect(subject.pipeline_configs.first.config_string).to include(LogStash::Config::Defaults.output, LogStash::Config::Defaults.input)
       end
     end
 
     context "when it has an input and an output" do
       let(:settings) { mock_settings( "config.string" => "#{input_block} #{filter_block} #{output_block}") }
 
       it "doesn't add anything" do
-        expect(subject.pipeline_configs.config_string).not_to include(LogStash::Config::Defaults.output, LogStash::Config::Defaults.input)
+        expect(subject.pipeline_configs.first.config_string).not_to include(LogStash::Config::Defaults.output, LogStash::Config::Defaults.input)
       end
     end
   end

logstash-core/spec/logstash/state_resolver_spec.rb

@@ -3,7 +3,6 @@
 require_relative "../support/helpers"
 require_relative "../support/matchers"
 require "logstash/state_resolver"
-require "logstash/config/config_part"
 require "logstash/config/pipeline_config"
 require "logstash/instrument/null_metric"
 require "logstash/pipeline"

logstash-core/spec/support/helpers.rb

@@ -50,7 +50,7 @@ def mock_pipeline_config(pipeline_id, config_string = nil, settings = {})
     settings = mock_settings(settings)
   end
 
-  config_part = LogStash::Config::ConfigPart.new(:config_string, "config_string", config_string)
+  config_part = org.logstash.common.SourceWithMetadata.new("config_string", "config_string", config_string)
 
   LogStash::Config::PipelineConfig.new(LogStash::Config::Source::Local, pipeline_id, config_part, settings)
 end