Added line and columns to plugin name

logstash-core/lib/logstash/codecs/delegator.rb

@@ -1,7 +1,9 @@
 module LogStash::Codecs
   class Delegator < SimpleDelegator
-    def initialize(obj)
+    def initialize(obj, parent_plugin_name)
       super(obj)
+      @parent_plugin_name = parent_plugin_name
+      @parent_config_reference = nil
       @encode_metric = LogStash::Instrument::NamespacedNullMetric.new
       @decode_metric = LogStash::Instrument::NamespacedNullMetric.new
     end
@@ -14,6 +16,8 @@ def metric=(metric)
       __getobj__.metric = metric
 
       __getobj__.metric.gauge(:name, __getobj__.class.config_name)
+      __getobj__.metric.gauge(:"parent-config-ref", @parent_config_reference)
+      __getobj__.metric.gauge(:"parent-name", @parent_plugin_name)
 
       @encode_metric = __getobj__.metric.namespace(:encode)
       @encode_metric.counter(:writes_in)
@@ -25,6 +29,10 @@ def metric=(metric)
       @decode_metric.report_time(:duration_in_millis, 0)
     end
 
+    def parent_config_reference=(parent_config_reference)
+      @parent_config_reference = parent_config_reference
+    end
+
     def encode(event)
       @encode_metric.increment(:writes_in)
       @encode_metric.time(:duration_in_millis) do

logstash-core/lib/logstash/compiler.rb

@@ -6,9 +6,10 @@
 module LogStash; class Compiler
   include ::LogStash::Util::Loggable
 
-  def self.compile_sources(sources_with_metadata, support_escapes)
+  def self.compile_sources(pipeline_config, support_escapes)
+   sources_with_metadata = [org.logstash.common.SourceWithMetadata.new("str", "pipeline", 0, 0, pipeline_config.config_string)]
     graph_sections = sources_with_metadata.map do |swm|
-      self.compile_graph(swm, support_escapes)
+      self.compile_graph(pipeline_config, swm, support_escapes)
     end
 
     input_graph = Graph.combine(*graph_sections.map {|s| s[:input] }).graph
@@ -29,7 +30,7 @@ def self.compile_sources(sources_with_metadata, support_escapes)
     PipelineIR.new(input_graph, filter_graph, output_graph, original_source)
   end
 
-  def self.compile_imperative(source_with_metadata, support_escapes)
+  def self.compile_imperative(pipeline_config, source_with_metadata, support_escapes)
     if !source_with_metadata.is_a?(org.logstash.common.SourceWithMetadata)
       raise ArgumentError, "Expected 'org.logstash.common.SourceWithMetadata', got #{source_with_metadata.class}"
     end
@@ -41,11 +42,23 @@ def self.compile_imperative(source_with_metadata, support_escapes)
       raise ConfigurationError, grammar.failure_reason
     end
 
+    plugins = config.recursive_select(LogStashCompilerLSCLGrammar::LogStash::Compiler::LSCL::AST::Plugin)
+    added_source_line_column(pipeline_config, plugins)
+
     config.process_escape_sequences = support_escapes
     config.compile(source_with_metadata)
   end
 
-  def self.compile_graph(source_with_metadata, support_escapes)
-    Hash[compile_imperative(source_with_metadata, support_escapes).map {|section,icompiled| [section, icompiled.toGraph]}]
+  def self.compile_graph(pipeline_config, source_with_metadata, support_escapes)
+    Hash[compile_imperative(pipeline_config, source_with_metadata, support_escapes).map {|section,icompiled| [section, icompiled.toGraph]}]
   end
+
+  def self.added_source_line_column(pipeline_config, plugins)
+    plugins.each do |plugin|
+      remapped_line = pipeline_config.lookup_source_and_line(plugin.line_and_column[0])
+      plugin.source_file = remapped_line[0]
+      plugin.source_line = remapped_line[1]
+    end
+  end
+
 end; end

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

@@ -72,8 +72,11 @@ def expr
 
   class Plugins < Node; end
   class Plugin < Node
+
+    attr_accessor :source_file, :source_line
+
     def expr
-      jdsl.iPlugin(source_meta, plugin_type_enum, self.plugin_name, self.expr_attributes)
+      jdsl.iPlugin(source_meta, plugin_type_enum, self.plugin_name, source_file, source_line, self.expr_attributes)
     end
 
     def plugin_type_enum
@@ -95,7 +98,11 @@ def plugin_name
 
     def expr_attributes
       # Turn attributes into a hash map
-      self.attributes.recursive_select(Attribute).map(&:expr).map {|k,v|
+      self.attributes.recursive_select(Attribute).map {|attribute|
+        attribute.source_file = source_file
+        attribute.source_line = source_line
+        attribute
+      }.map(&:expr).map {|k,v|
         if v.java_kind_of?(Java::OrgLogstashConfigIrExpression::ValueExpression)
           [k, v.get]
         else
@@ -130,7 +137,12 @@ def expr
   end
 
   class Attribute < Node
+    attr_accessor :source_file, :source_line
     def expr
+      if value.is_a?(Plugin)
+        value.source_file = source_file
+        value.source_line = source_line
+      end
       [name.text_value, value.expr]
     end
   end

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

@@ -213,6 +213,8 @@ def generate_variables
 
   class Plugins < Node; end
   class Plugin < Node
+    attr_accessor :source_file, :source_line
+
     def plugin_type
       if recursive_select_parent(Plugin).any?
         return "codec"
@@ -231,14 +233,18 @@ def variable_name
 
     def compile_initializer
       # If any parent is a Plugin, this must be a codec.
-
       if attributes.elements.nil?
-        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column})" << (plugin_type == "codec" ? "" : "\n")
+        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column}, #{source_file.inspect}, #{source_line.inspect})" << (plugin_type == "codec" ? "" : "\n")
       else
-        settings = attributes.recursive_select(Attribute).collect(&:compile).reject(&:empty?)
+        attrs = attributes.recursive_select(Attribute)
+        attrs.each do |attribute|
+          attribute.source_file = source_file
+          attribute.source_line = source_line
+        end
+        settings = attrs.collect(&:compile).reject(&:empty?)
 
         attributes_code = "LogStash::Util.hash_merge_many(#{settings.map { |c| "{ #{c} }" }.join(", ")})"
-        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column}, #{attributes_code})" << (plugin_type == "codec" ? "" : "\n")
+        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column}, #{source_file.inspect}, #{source_line.inspect}, #{attributes_code})" << (plugin_type == "codec" ? "" : "\n")
       end
     end
 
@@ -255,7 +261,7 @@ def compile
       when "codec"
         settings = attributes.recursive_select(Attribute).collect(&:compile).reject(&:empty?)
         attributes_code = "LogStash::Util.hash_merge_many(#{settings.map { |c| "{ #{c} }" }.join(", ")})"
-        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column}, #{attributes_code})"
+        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{source_meta.line}, #{source_meta.column}, #{source_file.inspect}, #{source_line.inspect}, #{attributes_code})"
       end
     end
 
@@ -309,7 +315,12 @@ def compile
     end
   end
   class Attribute < Node
+    attr_accessor :source_file, :source_line
     def compile
+      if value.is_a?(Plugin)
+        value.source_file = source_file
+        value.source_line = source_line
+      end
       return %Q(#{name.compile} => #{value.compile})
     end
   end

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

@@ -411,7 +411,9 @@ def validate_value(value, validator)
         case validator
           when :codec
             if value.first.is_a?(String)
-              value = LogStash::Codecs::Delegator.new LogStash::Plugin.lookup("codec", value.first).new
+              parent_plugin_name = self.config_name
+              codec = LogStash::Plugin.lookup("codec", value.first).new
+              value = LogStash::Codecs::Delegator.new(codec, parent_plugin_name)
               return true, value
             else
               value = value.first

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

@@ -1,7 +1,10 @@
 # encoding: utf-8
 require "digest"
 
+java_import org.logstash.config.ir.ConfigSourceSegment
+
 module LogStash module Config
+
   class PipelineConfig
     include LogStash::Util::Loggable
 
@@ -44,5 +47,27 @@ def display_debug_information
       logger.debug("Merged config")
       logger.debug("\n\n#{config_string}")
     end
+
+    def lookup_source_and_line(merged_line_number)
+      res = source_map.find { |source_segment| source_segment.contains(merged_line_number) }
+      if res == nil
+        raise IndexError
+      end
+      [res.getSource(), res.rebase(merged_line_number)]
+    end
+
+    private
+    def source_map
+      @source_map ||= begin
+        offset = 0
+        source_map = []
+        config_parts.each do |config_part|
+          source_segment = ConfigSourceSegment.new config_part.id, offset, config_part.getLinesCount()
+          source_map << source_segment
+          offset += source_segment.getLength()
+        end
+        source_map.freeze
+      end
+    end
   end
 end end

logstash-core/lib/logstash/inputs/base.rb

@@ -99,6 +99,15 @@ def clone
     cloned
   end
 
+  #used only in tests
+  def codec_attribute
+    @codec
+  end
+
+  def codec
+    params["codec"]
+  end
+
   def metric=(metric)
     super
     # Hack to create a new metric namespace using 'plugins' as the root

logstash-core/lib/logstash/pipeline.rb

@@ -38,6 +38,9 @@ def initialize(pipeline_config, namespaced_metric = nil, agent = nil)
     parsed_config = grammar.parse(config_str)
     raise(ConfigurationError, grammar.failure_reason) if parsed_config.nil?
 
+    plugins = parsed_config.recursive_select(LogStash::Config::AST::Plugin)
+    added_source_line_column(pipeline_config, plugins)
+
     parsed_config.process_escape_sequences = settings.get_value("config.support_escapes")
     config_code = parsed_config.compile
 
@@ -68,9 +71,16 @@ def non_reloadable_plugins
 
   private
 
+  def added_source_line_column(pipeline_config, plugins)
+    plugins.each do |plugin|
+      remapped_line = pipeline_config.lookup_source_and_line(plugin.line_and_column[0])
+      plugin.source_file = remapped_line[0]
+      plugin.source_line = remapped_line[1]
+    end
+  end
 
-  def plugin(plugin_type, name, line, column, *args)
-    @plugin_factory.plugin(plugin_type, name, line, column, *args)
+  def plugin(plugin_type, name, line, column, source_file, source_line, *args)
+    @plugin_factory.plugin(plugin_type, name, line, column, source_file, source_line, *args)
   end
 
   def default_logging_keys(other_keys = {})

logstash-core/spec/logstash/codecs/delegator_spec.rb

@@ -23,7 +23,7 @@ def decode(e)
   let(:codec) { LogStash::Codecs::MockCodec.new }
 
   subject do
-    delegator = described_class.new(codec)
+    delegator = described_class.new(codec, "MockCodec")
     delegator.metric = metric.namespace([:stats, :pipelines, :main, :plugins, :codecs, :my_id])
     delegator
   end

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

@@ -1,6 +1,7 @@
 require "spec_helper"
 require "logstash/compiler"
 require "support/helpers"
+require "logstash/config/pipeline_config"
 java_import Java::OrgLogstashConfigIr::DSL
 
 describe LogStash::Compiler do
@@ -50,8 +51,9 @@ def rand_meta
           org.logstash.common.SourceWithMetadata.new("#{source_protocol}_#{idx}", "#{source_id}_#{idx}", 0, 0, source)
         end
       end
+      let(:pipeline_config) {LogStash::Config::PipelineConfig.new LogStash::Config::Source::Local, "test_pipeline", sources_with_metadata, LogStash::SETTINGS}
 
-      subject(:pipeline) { described_class.compile_sources(sources_with_metadata, false) }
+      subject(:pipeline) { described_class.compile_sources(pipeline_config, false) }
 
       it "should generate a hash" do
         expect(pipeline.unique_hash).to be_a(String)
@@ -64,20 +66,6 @@ def rand_meta
       it "should provide the original source" do
         expect(pipeline.original_source).to eq(sources.join("\n"))
       end
-
-      describe "applying protocol and id metadata" do
-        it "should apply the correct source metadata to all components" do
-          # TODO: seems to be a jruby regression we cannot currently call each on a stream
-          pipeline.get_plugin_vertices.each do |pv|
-            name_idx = pv.plugin_definition.name.split("_").last
-            source_protocol_idx = pv.source_with_metadata.protocol.split("_").last
-            source_id_idx = pv.source_with_metadata.id.split("_").last
-
-            expect(name_idx).to eq(source_protocol_idx)
-            expect(name_idx).to eq(source_id_idx)
-          end
-        end
-      end
     end
 
     describe "complex configs" do
@@ -104,7 +92,8 @@ def rand_meta
   describe "compiling imperative" do
     let(:source_id) { "fake_sourcefile" }
     let(:source_with_metadata) { org.logstash.common.SourceWithMetadata.new(source_protocol, source_id, 0, 0, source) }
-    subject(:compiled) { described_class.compile_imperative(source_with_metadata, settings.get_value("config.support_escapes")) }
+    let(:pipeline_config) {LogStash::Config::PipelineConfig.new LogStash::Config::Source::Local, "test_pipeline", [source_with_metadata], LogStash::SETTINGS}
+    subject(:compiled) { described_class.compile_imperative(pipeline_config, source_with_metadata, settings.get_value("config.support_escapes")) }
 
     context "when config.support_escapes" do
       let(:parser) { LogStashCompilerLSCLGrammarParser.new }

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

@@ -72,4 +72,58 @@
       end
     end
   end
+
+  describe "source and line remapping" do
+    context "when pipeline is constructed from single file single line" do
+      let (:pipeline_conf_string) { 'input { generator1 }' }
+      subject { described_class.new(source, pipeline_id, [org.logstash.common.SourceWithMetadata.new("file", "/tmp/1", 0, 0, pipeline_conf_string)], settings) }
+      it "return the same line of the queried" do
+        expect(subject.lookup_source_and_line(1)[1]).to eq(1)
+      end
+    end
+
+    context "when pipeline is constructed from single file" do
+      let (:pipeline_conf_string) { 'input {
+                                       generator1
+                                     }' }
+      subject { described_class.new(source, pipeline_id, [org.logstash.common.SourceWithMetadata.new("file", "/tmp/1", 0, 0, pipeline_conf_string)], settings) }
+
+      it "return the same line of the queried" do
+        expect(subject.lookup_source_and_line(1)[1]).to eq(1)
+        expect(subject.lookup_source_and_line(2)[1]).to eq(2)
+      end
+
+      it "throw exception if line is out of bound" do
+        expect { subject.lookup_source_and_line(100) }.to raise_exception(IndexError)
+      end
+    end
+
+    context "when pipeline is constructed from multiple files" do
+      let (:pipeline_conf_string_part1) { 'input {
+                                             generator1
+                                           }' }
+      let (:pipeline_conf_string_part2) { 'output {
+                                             stdout
+                                           }' }
+      let(:merged_config_parts) do
+        [
+          org.logstash.common.SourceWithMetadata.new("file", "/tmp/input", 0, 0, pipeline_conf_string_part1),
+          org.logstash.common.SourceWithMetadata.new("file", "/tmp/output", 0, 0, pipeline_conf_string_part2)
+        ]
+      end
+      subject { described_class.new(source, pipeline_id, merged_config_parts, settings) }
+
+      it "return the line of first segment" do
+        expect(subject.lookup_source_and_line(2)).to eq(["/tmp/input", 2])
+      end
+
+      it "return the line of second segment" do
+        expect(subject.lookup_source_and_line(4)).to eq(["/tmp/output", 1])
+      end
+
+      it "throw exception if line is out of bound" do
+        expect { subject.lookup_source_and_line(100) }.to raise_exception(IndexError)
+      end
+    end
+  end
 end

logstash-core/spec/logstash/filter_delegator_spec.rb

@@ -11,7 +11,7 @@
 
   let(:filter_id) { "my-filter" }
   let(:config) do
-    { "host" => "127.0.0.1", "id" => filter_id }
+    { "host" => "127.0.0.1", "id" => filter_id, "config-ref" => "S: <filter_delegator_spec.rb> L: 14" }
   end
   let(:collector) {LogStash::Instrument::Collector.new}
   let(:metric) { LogStash::Instrument::Metric.new(collector).namespace(:null) }