Load a plugin by alias name. (#12796)

Introcuce the concept of alias for a plugin.

Creates an AliasRegistry to map plugin aliases to original plugins.
If a real plugin with same name of the an alias is present in the system, then the real plugin take precedence during the
instantiation of the pipeline.

Simplified the error handling in class lookup

Co-authored-by: Ry Biesemeyer <[email protected]>

Author: andsel

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

@@ -113,7 +113,7 @@ def register(hooks, settings)
 
     attr_reader :hooks
 
-    def initialize
+    def initialize(alias_registry = nil)
       @mutex = Mutex.new
       # We need a threadsafe class here because we may perform
       # get/set operations concurrently despite the fact we don't use
@@ -123,6 +123,7 @@ def initialize
       @registry = java.util.concurrent.ConcurrentHashMap.new
       @java_plugins = java.util.concurrent.ConcurrentHashMap.new
       @hooks = HooksRegistry.new
+      @alias_registry = alias_registry || Java::org.logstash.plugins.AliasRegistry.new
     end
 
     def setup!
@@ -196,41 +197,47 @@ def lookup(type, plugin_name, &block)
     # a plugin and will do a lookup on the namespace of the required class to find a matching
     # plugin with the appropriate type.
     def legacy_lookup(type, plugin_name)
-      begin
-        path = "logstash/#{type}s/#{plugin_name}"
+      klass = load_plugin_class(type, plugin_name)
 
-        klass = begin
-          namespace_lookup(type, plugin_name)
-        rescue UnknownPlugin => e
-          # Plugin not registered. Try to load it.
-          begin
-            require path
-            namespace_lookup(type, plugin_name)
-          rescue LoadError => e
-            logger.error("Tried to load a plugin's code, but failed.", :exception => e, :path => path, :type => type, :name => plugin_name)
-            raise
-          end
-        end
+      if !klass && @alias_registry.alias?(type.to_java, plugin_name)
+        resolved_plugin_name = @alias_registry.original_from_alias(type.to_java, plugin_name)
+        logger.debug("Loading #{type} plugin #{resolved_plugin_name} via its alias #{plugin_name}...")
+        klass = load_plugin_class(type, resolved_plugin_name)
+        lazy_add(type, resolved_plugin_name, klass) if klass
+      end
 
-        plugin = lazy_add(type, plugin_name, klass)
-      rescue => e
-        logger.error("Problems loading a plugin with",
-                    :type => type,
-                    :name => plugin_name,
-                    :path => path,
-                    :error_message => e.message,
-                    :error_class => e.class,
-                    :error_backtrace => e.backtrace)
-
-        raise LoadError, "Problems loading the requested plugin named #{plugin_name} of type #{type}. Error: #{e.class} #{e.message}"
+      unless klass
+        logger.error("Unable to load plugin.", :type => type, :name => plugin_name)
+        raise LoadError, "Unable to load the requested plugin named #{plugin_name} of type #{type}. The plugin is not installed."
       end
 
-      plugin
+      lazy_add(type, plugin_name, klass)
+    end
+
+    # load a plugin's class, or return nil if the plugin cannot be loaded.
+    # attempts to load the class purely through namespace lookup,
+    # and falls back to requiring the path of the expected plugin.
+    # @param type [String]: plugin type, such as "input", "output", "filter", "codec"
+    # @param plugin_name [String]: plugin name, such as "grok", "elasticsearch"
+    # @return [Class,nil] the plugin class, or nil
+    private
+    def load_plugin_class(type, plugin_name)
+      klass = namespace_lookup(type, plugin_name)
+
+      unless klass
+        require("logstash/#{type}s/#{plugin_name}")
+        klass = namespace_lookup(type, plugin_name)
+      end
+      klass
+    rescue LoadError => e
+      logger.debug("Tried to load a plugin's code, but failed.", :exception => e, :path => e.path, :type => type, :name => plugin_name)
+      nil
     end
 
+    public
     def lookup_pipeline_plugin(type, name)
       LogStash::PLUGIN_REGISTRY.lookup(type, name) do |plugin_klass, plugin_name|
-        is_a_plugin?(plugin_klass, plugin_name)
+        is_a_plugin_or_alias?(plugin_klass, type.to_java, plugin_name)
       end
     rescue LoadError, NameError => e
       logger.debug("Problems loading the plugin with", :type => type, :name => name)
@@ -268,19 +275,15 @@ def size
     # ex.: namespace_lookup("filter", "grok") looks for LogStash::Filters::Grok
     # @param type [String] plugin type, "input", "output", "filter"
     # @param name [String] plugin name, ex.: "grok"
-    # @return [Class] the plugin class or raises NameError
-    # @raise NameError if plugin class does not exist or is invalid
+    # @return [Class,nil] the plugin class or nil
     def namespace_lookup(type, name)
       type_const = "#{type.capitalize}s"
       namespace = LogStash.const_get(type_const)
       # the namespace can contain constants which are not for plugins classes (do not respond to :config_name)
       # namespace.constants is the shallow collection of all constants symbols in namespace
       # note that below namespace.const_get(c) should never result in a NameError since c is from the constants collection
       klass_sym = namespace.constants.find { |c| is_a_plugin?(namespace.const_get(c), name) }
-      klass = klass_sym && namespace.const_get(klass_sym)
-
-      raise(UnknownPlugin) unless klass
-      klass
+      klass_sym && namespace.const_get(klass_sym)
     end
 
     # check if klass is a valid plugin for name
@@ -290,7 +293,19 @@ def namespace_lookup(type, name)
     def is_a_plugin?(klass, name)
       (klass.class == Java::JavaLang::Class && klass.simple_name.downcase == name.gsub('_','')) ||
       (klass.class == Java::JavaClass && klass.simple_name.downcase == name.gsub('_','')) ||
-      (klass.ancestors.include?(LogStash::Plugin) && klass.respond_to?(:config_name) && klass.config_name == name)
+      (klass.ancestors.include?(LogStash::Plugin) && klass.respond_to?(:config_name) &&
+        klass.config_name == name)
+    end
+
+    # check if klass is a valid plugin for name,
+    # including alias resolution
+    def is_a_plugin_or_alias?(klass, type, plugin_name)
+      return true if is_a_plugin?(klass, plugin_name)
+
+      resolved_plugin_name = @alias_registry.resolve_alias(type, plugin_name)
+      return true if is_a_plugin?(klass, resolved_plugin_name)
+
+      false
     end
 
     def add_plugin(type, name, klass)

logstash-core/spec/logstash/java_pipeline_spec.rb

@@ -199,6 +199,15 @@ def flush(options)
     end
   end
 
+  describe "aliased plugin instantiation" do
+    it "should create the pipeline as if it's using the original plugin" do
+      alias_registry = Java::org.logstash.plugins.AliasRegistry.new({["input", "alias"] => "generator"})
+      LogStash::PLUGIN_REGISTRY = LogStash::Plugins::Registry.new alias_registry
+      pipeline = mock_java_pipeline_from_string("input { alias { count => 1 } } output { null {} }")
+      expect(pipeline.ephemeral_id).to_not be_nil
+      pipeline.close
+    end
+  end
 
   describe "event cancellation" do
     # test harness for https://github.com/elastic/logstash/issues/6055

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

@@ -34,16 +34,23 @@ def register; end
 end
 
 describe LogStash::Plugins::Registry do
-  let(:registry) { described_class.new }
+  let(:alias_registry) { nil }
+  let(:registry) { described_class.new alias_registry }
 
   context "when loading installed plugins" do
+    let(:alias_registry) { Java::org.logstash.plugins.AliasRegistry.new({["input", "alias_std_input"] => "stdin"}) }
     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 load an aliased ruby plugin" do
+      klass = registry.lookup("input", "alias_std_input")
+      expect(klass).to eq(LogStash::Inputs::Stdin)
+    end
+
     it "should raise an error if can not find the plugin class" do
       expect { registry.lookup("input", "do-not-exist-elastic") }.to raise_error(LoadError)
     end
@@ -53,13 +60,34 @@ def register; end
       expect { registry.lookup("input", "new_plugin") }.to change { registry.size }.by(1)
       expect { registry.lookup("input", "new_plugin") }.not_to change { registry.size }
     end
+
+    context "when loading installed plugin that overrides an alias" do
+      let(:alias_registry) { Java::org.logstash.plugins.AliasRegistry.new({["input", "dummy"] => "new_plugin"}) }
+
+      it 'should load the concrete implementation instead of resolving the alias' do
+        klass = registry.lookup("input", "dummy")
+        expect(klass).to eq(LogStash::Inputs::Dummy)
+      end
+    end
   end
 
   context "when loading code defined plugins" do
+    let(:alias_registry) { Java::org.logstash.plugins.AliasRegistry.new({["input", "alias_input"] => "new_plugin"}) }
+
     it "should return the expected class" do
       klass = registry.lookup("input", "dummy")
       expect(klass).to eq(LogStash::Inputs::Dummy)
     end
+
+    it "should return the expected class also for aliased plugins" do
+      klass = registry.lookup("input", "alias_input")
+      expect(klass).to eq(LogStash::Inputs::NewPlugin)
+    end
+
+    it "should return the expected class also for alias-targeted plugins" do
+      klass = registry.lookup("input", "new_plugin")
+      expect(klass).to eq(LogStash::Inputs::NewPlugin)
+    end
   end
 
   context "when plugin is not installed and not defined" do

logstash-core/src/main/java/org/logstash/plugins/AliasRegistry.java

@@ -0,0 +1,117 @@
+package org.logstash.plugins;
+
+import org.logstash.plugins.PluginLookup.PluginType;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+public class AliasRegistry {
+
+    private final static class PluginCoordinate {
+        private final PluginType type;
+        private final String name;
+
+        public PluginCoordinate(PluginType type, String name) {
+            this.type = type;
+            this.name = name;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            PluginCoordinate that = (PluginCoordinate) o;
+            return type == that.type && Objects.equals(name, that.name);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(type, name);
+        }
+
+        PluginCoordinate withName(String name) {
+            return new PluginCoordinate(this.type, name);
+        }
+    }
+
+
+    private final Map<PluginCoordinate, String> aliases = new HashMap<>();
+    private final Map<PluginCoordinate, String> reversedAliases = new HashMap<>();
+
+    public AliasRegistry() {
+        Map<PluginCoordinate, String> defaultDefinitions = new HashMap<>();
+        defaultDefinitions.put(new PluginCoordinate(PluginType.INPUT, "elastic_agent"), "beats");
+        configurePluginAliases(defaultDefinitions);
+    }
+
+    /**
+     * Constructor used in tests to customize the plugins renames.
+     * The input map's key are tuples of (type, name)
+     * */
+    public AliasRegistry(Map<List<String>, String> aliasDefinitions) {
+        Map<PluginCoordinate, String> aliases = new HashMap<>();
+
+        // transform the (tye, name) into PluginCoordinate
+        for (Map.Entry<List<String>, String> e : aliasDefinitions.entrySet()) {
+            final List<String> tuple = e.getKey();
+            final PluginCoordinate key = mapTupleToCoordinate(tuple);
+            aliases.put(key, e.getValue());
+        }
+
+        configurePluginAliases(aliases);
+    }
+
+    private PluginCoordinate mapTupleToCoordinate(List<String> tuple) {
+        if (tuple.size() != 2) {
+            throw new IllegalArgumentException("Expected a tuple of 2 elements, but found: " + tuple);
+        }
+        final PluginType type = PluginType.valueOf(tuple.get(0).toUpperCase());
+        final String name = tuple.get(1);
+        final PluginCoordinate key = new PluginCoordinate(type, name);
+        return key;
+    }
+
+    private void configurePluginAliases(Map<PluginCoordinate, String> aliases) {
+        this.aliases.putAll(aliases);
+        for (Map.Entry<PluginCoordinate, String> e : this.aliases.entrySet()) {
+            final PluginCoordinate reversedAlias = e.getKey().withName(e.getValue());
+            if (reversedAliases.containsKey(reversedAlias)) {
+                throw new IllegalStateException("Found plugin " + e.getValue() + " aliased more than one time");
+            }
+            reversedAliases.put(reversedAlias, e.getKey().name);
+        }
+    }
+
+    public boolean isAlias(String type, String pluginName) {
+        final PluginType pluginType = PluginType.valueOf(type.toUpperCase());
+
+        return isAlias(pluginType, pluginName);
+    }
+
+    public boolean isAlias(PluginType type, String pluginName) {
+        return aliases.containsKey(new PluginCoordinate(type, pluginName));
+    }
+
+    public String originalFromAlias(PluginType type, String alias) {
+        return aliases.get(new PluginCoordinate(type, alias));
+    }
+
+    public String originalFromAlias(String type, String alias) {
+        return originalFromAlias(PluginType.valueOf(type.toUpperCase()), alias);
+    }
+
+    public Optional<String> aliasFromOriginal(PluginType type, String realPluginName) {
+        return Optional.ofNullable(reversedAliases.get(new PluginCoordinate(type, realPluginName)));
+    }
+
+    /**
+     * if pluginName is an alias then return the real plugin name else return it unchanged
+     */
+    public String resolveAlias(String type, String pluginName) {
+        final PluginCoordinate pluginCoord = new PluginCoordinate(PluginType.valueOf(type.toUpperCase()), pluginName);
+        return aliases.getOrDefault(pluginCoord, pluginName);
+    }
+}