Add generic code used to load any kind of plugins across logstash

Introduce the idea of a registry plugin placeholder where all necessary
interface to plugins is mantained, also simplified the internal registry
calls to be more generic.

Add a way to handle registrations for plugins explicitly

introduced the idea of self.plugin_type method to fetch plugin type from base clase, also removed the former plugins:mixin used to annotate defined plugins

make the config_name method also handle the registration to the plugin registry, that way old plugins get registration out of the box and we can simply incoming plugin registry without automatic loading

simplify the plugin registry by removing former need to load classes, now they all get registered automatically when using the config_name method

cleanup unnecessary former changes

updated typo in comments for the plugins registry and also removed internal attr_reader for the same class

renamed plugin annotate to declare_plugin to have a more meaningful name

change Registry::Plugin.gem_name -> cannonic_gem_name to reflect the idea of having probably also other non cannonic gem names

Fixes #4535

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

@@ -7,8 +7,13 @@ require "logstash/logging"
 # This is the base class for logstash codecs.
 module LogStash::Codecs; class Base < LogStash::Plugin
   include LogStash::Config::Mixin
+
   config_name "codec"
 
+  def self.plugin_type
+    "codec"
+  end
+
   def initialize(params={})
     super
     config_init(@params)

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

@@ -1,6 +1,7 @@
 # encoding: utf-8
 require "logstash/namespace"
 require "logstash/config/registry"
+require "logstash/plugins/registry"
 require "logstash/logging"
 require "logstash/util/password"
 require "logstash/version"
@@ -191,8 +192,12 @@ module LogStash::Config::Mixin
     def config_name(name = nil)
       @config_name = name if !name.nil?
       LogStash::Config::Registry.registry[@config_name] = self
+      if self.respond_to?("plugin_type")
+        declare_plugin(self.plugin_type, @config_name)
+      end
       return @config_name
     end
+    alias_method :config_plugin, :config_name
 
     # Deprecated: Declare the version of the plugin
     # inside the gemspec.

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

@@ -117,6 +117,10 @@ class LogStash::Filters::Base < LogStash::Plugin
   # Optional.
   config :periodic_flush, :validate => :boolean, :default => false
 
+  def self.plugin_type
+    "filter"
+  end
+
   public
   def initialize(params)
     super

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

@@ -10,6 +10,7 @@ require "logstash/util/decorators"
 # This is the base class for Logstash inputs.
 class LogStash::Inputs::Base < LogStash::Plugin
   include LogStash::Config::Mixin
+
   config_name "input"
 
   # Add a `type` field to all events handled by this input.
@@ -48,6 +49,10 @@ class LogStash::Inputs::Base < LogStash::Plugin
   attr_accessor :params
   attr_accessor :threadable
 
+  def self.plugin_type
+    "input"
+  end
+
   public
   def initialize(params={})
     super

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

@@ -57,6 +57,10 @@ class LogStash::Outputs::Base < LogStash::Plugin
     self.class.declare_workers_not_supported!(message)
   end
 
+  def self.plugin_type
+    "output"
+  end
+
   public
   def initialize(params={})
     super

logstash-core/lib/logstash/plugin.rb

@@ -6,6 +6,7 @@ require "logstash/instrument/null_metric"
 require "cabin"
 require "concurrent"
 require "securerandom"
+require "logstash/plugins/registry"
 
 class LogStash::Plugin
   attr_accessor :params
@@ -117,23 +118,21 @@ class LogStash::Plugin
   # Look up a plugin by type and name.
   def self.lookup(type, name)
     path = "logstash/#{type}s/#{name}"
-
-    # first check if plugin already exists in namespace and continue to next step if not
-    begin
-      return namespace_lookup(type, name)
-    rescue NameError
-      logger.debug("Plugin not defined in namespace, checking for plugin file", :type => type, :name => name, :path => path)
+    LogStash::Registry.instance.lookup(type ,name) do |plugin_klass, plugin_name|
+      is_a_plugin?(plugin_klass, plugin_name)
     end
-
-    # try to load the plugin file. ex.: lookup("filter", "grok") will require logstash/filters/grok
-    require(path)
-
-    # check again if plugin is now defined in namespace after the require
-    namespace_lookup(type, name)
   rescue LoadError, NameError => e
+    logger.debug("Problems loading the plugin with", :type => type, :name => name, :path => path)
     raise(LogStash::PluginLoadingError, I18n.t("logstash.pipeline.plugin-loading-error", :type => type, :name => name, :path => path, :error => e.to_s))
   end
 
+  public
+  def self.declare_plugin(type, name)
+    path = "logstash/#{type}s/#{name}"
+    registry = LogStash::Registry.instance
+    registry.register(path, self)
+  end
+
   private
   # lookup a plugin by type and name in the existing LogStash module namespace
   # ex.: namespace_lookup("filter", "grok") looks for LogStash::Filters::Grok
@@ -165,4 +164,5 @@ class LogStash::Plugin
   def self.logger
     @logger ||= Cabin::Channel.get(LogStash)
   end
+
 end # class LogStash::Plugin

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

@@ -0,0 +1,83 @@
+# encoding: utf-8
+require 'singleton'
+require "rubygems/package"
+
+module LogStash
+  class Registry
+
+    ##
+    # Placeholder class for registered plugins
+    ##
+    class Plugin
+      attr_reader :type, :name
+
+      def initialize(type, name)
+        @type  = type
+        @name  = name
+      end
+
+      def path
+        "logstash/#{type}s/#{name}"
+      end
+
+      def cannonic_gem_name
+        "logstash-#{type}-#{name}"
+      end
+
+      def installed?
+        find_plugin_spec(cannonic_gem_name).any?
+      end
+
+      private
+
+      def find_plugin_spec(name)
+        specs = ::Gem::Specification.find_all_by_name(name)
+        specs.select{|spec| logstash_plugin_spec?(spec)}
+      end
+
+      def logstash_plugin_spec?(spec)
+        spec.metadata && spec.metadata["logstash_plugin"] == "true"
+      end
+
+    end
+
+    include Singleton
+
+    def initialize
+      @registry = {}
+      @logger = Cabin::Channel.get(LogStash)
+    end
+
+    def lookup(type, plugin_name, &block)
+
+      plugin = Plugin.new(type, plugin_name)
+
+      if plugin.installed?
+        return @registry[plugin.path] if registered?(plugin.path)
+        require plugin.path
+        klass = @registry[plugin.path]
+        if block_given? # if provided pass a block to do validation
+          raise LoadError unless block.call(klass, plugin_name)
+        end
+        return klass
+      else
+        # The plugin was defined directly in the code, so there is no need to use the
+        # require way of loading classes
+        return @registry[plugin.path] if registered?(plugin.path)
+        raise LoadError
+      end
+    rescue => e
+      @logger.debug("Problems loading a plugin with", :type => type, :name => plugin, :path => plugin.path, :error => e) if @logger.debug?
+      raise LoadError, "Problems loading the requested plugin named #{plugin_name} of type #{type}."
+    end
+
+    def register(path, klass)
+      @registry[path] = klass
+    end
+
+    def registered?(path)
+      @registry.has_key?(path)
+    end
+
+  end
+end

logstash-core/spec/logstash/plugin_spec.rb

@@ -34,6 +34,19 @@ describe LogStash::Plugin do
     expect(LogStash::Plugin.lookup("filter", "lady_gaga")).to eq(LogStash::Filters::LadyGaga)
   end
 
+  describe "plugin signup in the registry" do
+
+    let(:registry) { LogStash::Registry.instance }
+
+    it "should be present in the registry" do
+      class LogStash::Filters::MyPlugin < LogStash::Filters::Base
+        config_name "my_plugin"
+      end
+      path     = "logstash/filters/my_plugin"
+      expect(registry.registered?(path)).to eq(true)
+    end
+  end
+
   describe "#inspect" do
     class LogStash::Filters::MyTestFilter < LogStash::Filters::Base
       config_name "param1"

logstash-core/spec/logstash/plugins/registry_spec.rb

@@ -0,0 +1,57 @@
+# encoding: utf-8
+require "spec_helper"
+require "logstash/plugins/registry"
+require "logstash/inputs/base"
+
+# use a dummy NOOP input to test plugin registry
+class LogStash::Inputs::Dummy < LogStash::Inputs::Base
+  config_name "dummy"
+
+  def register; end
+
+end
+
+describe LogStash::Registry do
+
+  let(:registry) { described_class.instance }
+
+  context "when loading installed plugins" do
+
+    let(:plugin) { double("plugin") }
+
+    it "should return the expected class" do
+      klass = registry.lookup("input", "stdin")
+      expect(klass).to eq(LogStash::Inputs::Stdin)
+    end
+
+    it "should raise an error if can not find the plugin class" do
+      expect(LogStash::Registry::Plugin).to receive(:new).with("input", "elasticsearch").and_return(plugin)
+      expect(plugin).to receive(:path).and_return("logstash/input/elasticsearch").twice
+      expect(plugin).to receive(:installed?).and_return(true)
+      expect { registry.lookup("input", "elasticsearch") }.to raise_error(LoadError)
+    end
+
+    it "should load from registry is already load" do
+      registry.lookup("input", "stdin")
+      expect(registry).to receive(:registered?).and_return(true).once
+      registry.lookup("input", "stdin")
+      internal_registry = registry.instance_variable_get("@registry")
+      expect(internal_registry).to include("logstash/inputs/stdin" => LogStash::Inputs::Stdin)
+    end
+  end
+
+  context "when loading code defined plugins" do
+    it "should return the expected class" do
+      klass = registry.lookup("input", "dummy")
+      expect(klass).to eq(LogStash::Inputs::Dummy)
+    end
+  end
+
+  context "when plugin is not installed and not defined" do
+    it "should raise an error" do
+      expect { registry.lookup("input", "elasticsearch") }.to raise_error(LoadError)
+    end
+  end
+
+end
+