Fix handling of mandatory meta plugins

server/src/main/java/org/elasticsearch/plugins/PluginsService.java

@@ -56,15 +56,16 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.elasticsearch.common.io.FileSystemUtils.isAccessibleDirectory;
 
@@ -102,6 +103,8 @@ public PluginsService(Settings settings, Path configPath, Path modulesDirectory,
 
         List<Tuple<PluginInfo, Plugin>> pluginsLoaded = new ArrayList<>();
         List<PluginInfo> pluginsList = new ArrayList<>();
+        // we need to build a List of plugins for checking mandatory plugins
+        final List<String> pluginsNames = new ArrayList<>();
         // first we load plugins that are on the classpath. this is for tests and transport clients
         for (Class<? extends Plugin> pluginClass : classpathPlugins) {
             Plugin plugin = loadPlugin(pluginClass, settings, configPath);
@@ -112,6 +115,7 @@ public PluginsService(Settings settings, Path configPath, Path modulesDirectory,
             }
             pluginsLoaded.add(new Tuple<>(pluginInfo, plugin));
             pluginsList.add(pluginInfo);
+            pluginsNames.add(pluginInfo.getName());
         }
 
         Set<Bundle> seenBundles = new LinkedHashSet<>();
@@ -135,11 +139,15 @@ public PluginsService(Settings settings, Path configPath, Path modulesDirectory,
                 // TODO: remove this leniency, but tests bogusly rely on it
                 if (isAccessibleDirectory(pluginsDirectory, logger)) {
                     checkForFailedPluginRemovals(pluginsDirectory);
-                    Set<Bundle> plugins = getPluginBundles(pluginsDirectory);
-                    for (Bundle bundle : plugins) {
-                        pluginsList.add(bundle.plugin);
+                    List<BundleCollection> plugins = getPluginBundleCollections(pluginsDirectory);
+                    for (final BundleCollection plugin : plugins) {
+                        final Collection<Bundle> bundles = plugin.bundles();
+                        for (final Bundle bundle : bundles) {
+                            pluginsList.add(bundle.plugin);
+                        }
+                        seenBundles.addAll(bundles);
+                        pluginsNames.add(plugin.name());
                     }
-                    seenBundles.addAll(plugins);
                 }
             } catch (IOException ex) {
                 throw new IllegalStateException("Unable to initialize plugins", ex);
@@ -152,12 +160,6 @@ public PluginsService(Settings settings, Path configPath, Path modulesDirectory,
         this.info = new PluginsAndModules(pluginsList, modulesList);
         this.plugins = Collections.unmodifiableList(pluginsLoaded);
 
-        // We need to build a List of plugins for checking mandatory plugins
-        Set<String> pluginsNames = new HashSet<>();
-        for (Tuple<PluginInfo, Plugin> tuple : this.plugins) {
-            pluginsNames.add(tuple.v1().getName());
-        }
-
         // Checking expected plugins
         List<String> mandatoryPlugins = MANDATORY_SETTING.get(settings);
         if (mandatoryPlugins.isEmpty() == false) {
@@ -168,7 +170,11 @@ public PluginsService(Settings settings, Path configPath, Path modulesDirectory,
                 }
             }
             if (!missingPlugins.isEmpty()) {
-                throw new ElasticsearchException("Missing mandatory plugins [" + Strings.collectionToDelimitedString(missingPlugins, ", ") + "]");
+                final String message = String.format(
+                        Locale.ROOT,
+                        "missing mandatory plugins [%s]",
+                        Strings.collectionToDelimitedString(missingPlugins, ", "));
+                throw new IllegalStateException(message);
             }
         }
 
@@ -244,9 +250,17 @@ public PluginsAndModules info() {
         return info;
     }
 
+    /**
+     * An abstraction over a single plugin and meta-plugins.
+     */
+    interface BundleCollection {
+        String name();
+        Collection<Bundle> bundles();
+    }
+
     // a "bundle" is a group of plugins in a single classloader
     // really should be 1-1, but we are not so fortunate
-    static class Bundle {
+    static class Bundle implements BundleCollection {
         final PluginInfo plugin;
         final Set<URL> urls;
 
@@ -266,6 +280,16 @@ static class Bundle {
             this.urls = Objects.requireNonNull(urls);
         }
 
+        @Override
+        public String name() {
+            return plugin.getName();
+        }
+
+        @Override
+        public Collection<Bundle> bundles() {
+            return Collections.singletonList(this);
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
@@ -281,35 +305,78 @@ public int hashCode() {
     }
 
     /**
-     * Extracts all installed plugin directories from the provided {@code rootPath} expanding meta plugins if needed.
+     * Represents a meta-plugin and the {@link Bundle}s corresponding to its constituents.
+     */
+    static class MetaBundle implements BundleCollection {
+        private final String name;
+        private final List<Bundle> bundles;
+
+        MetaBundle(final String name, final List<Bundle> bundles) {
+            this.name = name;
+            this.bundles = bundles;
+        }
+
+        @Override
+        public String name() {
+            return name;
+        }
+
+        @Override
+        public Collection<Bundle> bundles() {
+            return bundles;
+        }
+
+    }
+
+    /**
+     * Extracts all installed plugin directories from the provided {@code rootPath} expanding meta-plugins if needed.
+     *
      * @param rootPath the path where the plugins are installed
-     * @return A list of all plugin paths installed in the {@code rootPath}
+     * @return a list of all plugin paths installed in the {@code rootPath}
      * @throws IOException if an I/O exception occurred reading the directories
      */
     public static List<Path> findPluginDirs(final Path rootPath) throws IOException {
+        final Tuple<List<Path>, Map<String, List<Path>>> groupedPluginDirs = findGroupedPluginDirs(rootPath);
+        return Stream.concat(
+                groupedPluginDirs.v1().stream(),
+                groupedPluginDirs.v2().values().stream().flatMap(Collection::stream))
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * Extracts all installed plugin directories from the provided {@code rootPath} expanding meta-plugins if needed. The plugins are
+     * grouped into plugins and meta-plugins. The meta-plugins are keyed by the meta-plugin name.
+     *
+     * @param rootPath the path where the plugins are installed
+     * @return a tuple of plugins as the first component and meta-plugins keyed by meta-plugin name as the second component
+     * @throws IOException if an I/O exception occurred reading the directories
+     */
+    private static Tuple<List<Path>, Map<String, List<Path>>> findGroupedPluginDirs(final Path rootPath) throws IOException {
         final List<Path> plugins = new ArrayList<>();
+        final Map<String, List<Path>> metaPlugins = new LinkedHashMap<>();
         final Set<String> seen = new HashSet<>();
         if (Files.exists(rootPath)) {
             try (DirectoryStream<Path> stream = Files.newDirectoryStream(rootPath)) {
                 for (Path plugin : stream) {
                     if (FileSystemUtils.isDesktopServicesStore(plugin) ||
-                        plugin.getFileName().toString().startsWith(".removing-")) {
+                            plugin.getFileName().toString().startsWith(".removing-")) {
                         continue;
                     }
                     if (seen.add(plugin.getFileName().toString()) == false) {
                         throw new IllegalStateException("duplicate plugin: " + plugin);
                     }
                     if (MetaPluginInfo.isMetaPlugin(plugin)) {
+                        final String name = plugin.getFileName().toString();
                         try (DirectoryStream<Path> subStream = Files.newDirectoryStream(plugin)) {
                             for (Path subPlugin : subStream) {
                                 if (MetaPluginInfo.isPropertiesFile(subPlugin) ||
-                                    FileSystemUtils.isDesktopServicesStore(subPlugin)) {
+                                        FileSystemUtils.isDesktopServicesStore(subPlugin)) {
                                     continue;
                                 }
                                 if (seen.add(subPlugin.getFileName().toString()) == false) {
                                     throw new IllegalStateException("duplicate plugin: " + subPlugin);
                                 }
-                                plugins.add(subPlugin);
+                                metaPlugins.computeIfAbsent(name, n -> new ArrayList<>()).add(subPlugin);
                             }
                         }
                     } else {
@@ -318,7 +385,7 @@ public static List<Path> findPluginDirs(final Path rootPath) throws IOException
                 }
             }
         }
-        return plugins;
+        return Tuple.tuple(plugins, metaPlugins);
     }
 
     /**
@@ -380,26 +447,46 @@ static void checkForFailedPluginRemovals(final Path pluginsDirectory) throws IOE
      * @throws IOException if an I/O exception occurs reading the plugin bundles
      */
     static Set<Bundle> getPluginBundles(final Path pluginsDirectory) throws IOException {
-        Logger logger = Loggers.getLogger(PluginsService.class);
-        Set<Bundle> bundles = new LinkedHashSet<>();
+        return getPluginBundleCollections(pluginsDirectory).stream().flatMap(b -> b.bundles().stream()).collect(Collectors.toSet());
+    }
 
-        List<Path> infos = findPluginDirs(pluginsDirectory);
-        for (Path plugin : infos) {
-            logger.trace("--- adding plugin [{}]", plugin.toAbsolutePath());
-            final PluginInfo info;
-            try {
-                info = PluginInfo.readFromProperties(plugin);
-            } catch (IOException e) {
-                throw new IllegalStateException("Could not load plugin descriptor for existing plugin ["
-                        + plugin.getFileName() + "]. Was the plugin built before 2.0?", e);
-            }
-            if (bundles.add(new Bundle(info, plugin)) == false) {
-                throw new IllegalStateException("duplicate plugin: " + info);
+    private static List<BundleCollection> getPluginBundleCollections(final Path pluginsDirectory) throws IOException {
+        final List<BundleCollection> bundles = new ArrayList<>();
+        final Set<Bundle> seenBundles = new HashSet<>();
+        final Tuple<List<Path>, Map<String, List<Path>>> groupedPluginDirs = findGroupedPluginDirs(pluginsDirectory);
+        for (final Path plugin : groupedPluginDirs.v1()) {
+            final Bundle bundle = bundle(seenBundles, plugin);
+            bundles.add(bundle);
+        }
+        for (final Map.Entry<String, List<Path>> metaPlugin : groupedPluginDirs.v2().entrySet()) {
+            final List<Bundle> metaPluginBundles = new ArrayList<>();
+            for (final Path metaPluginPlugin : metaPlugin.getValue()) {
+                final Bundle bundle = bundle(seenBundles, metaPluginPlugin);
+                metaPluginBundles.add(bundle);
             }
+            final MetaBundle metaBundle = new MetaBundle(metaPlugin.getKey(), metaPluginBundles);
+            bundles.add(metaBundle);
         }
+
         return bundles;
     }
 
+    private static Bundle bundle(final Set<Bundle> bundles, final Path plugin) throws IOException {
+        Loggers.getLogger(PluginsService.class).trace("--- adding plugin [{}]", plugin.toAbsolutePath());
+        final PluginInfo info;
+        try {
+            info = PluginInfo.readFromProperties(plugin);
+        } catch (final IOException e) {
+            throw new IllegalStateException("Could not load plugin descriptor for existing plugin ["
+                    + plugin.getFileName() + "]. Was the plugin built before 2.0?", e);
+        }
+        final Bundle bundle = new Bundle(info, plugin);
+        if (bundles.add(bundle) == false) {
+            throw new IllegalStateException("duplicate plugin: " + info);
+        }
+        return bundle;
+    }
+
     /**
      * Return the given bundles, sorted in dependency loading order.
      *

server/src/test/java/org/elasticsearch/plugins/PluginsServiceTests.java

@@ -604,4 +604,123 @@ public void testIncompatibleJavaVersion() throws Exception {
         IllegalStateException e = expectThrows(IllegalStateException.class, () -> PluginsService.verifyCompatibility(info));
         assertThat(e.getMessage(), containsString("my_plugin requires Java"));
     }
+
+    public void testFindPluginDirs() throws IOException {
+        final Path plugins = createTempDir();
+
+        final Path fake = plugins.resolve("fake");
+
+        PluginTestUtil.writePluginProperties(
+                fake,
+                "description", "description",
+                "name", "fake",
+                "version", "1.0.0",
+                "elasticsearch.version", Version.CURRENT.toString(),
+                "java.version", System.getProperty("java.specification.version"),
+                "classname", "test.DummyPlugin");
+
+        try (InputStream jar = PluginsServiceTests.class.getResourceAsStream("dummy-plugin.jar")) {
+            Files.copy(jar, fake.resolve("plugin.jar"));
+        }
+
+        final Path fakeMeta = plugins.resolve("fake-meta");
+
+        PluginTestUtil.writeMetaPluginProperties(fakeMeta, "description", "description", "name", "fake-meta");
+
+        final Path fakeMetaCore = fakeMeta.resolve("fake-meta-core");
+        PluginTestUtil.writePluginProperties(
+                fakeMetaCore,
+                "description", "description",
+                "name", "fake-meta-core",
+                "version", "1.0.0",
+                "elasticsearch.version", Version.CURRENT.toString(),
+                "java.version", System.getProperty("java.specification.version"),
+                "classname", "test.DummyPlugin");
+        try (InputStream jar = PluginsServiceTests.class.getResourceAsStream("dummy-plugin.jar")) {
+            Files.copy(jar, fakeMetaCore.resolve("plugin.jar"));
+        }
+
+        assertThat(PluginsService.findPluginDirs(plugins), containsInAnyOrder(fake, fakeMetaCore));
+    }
+
+    public void testMissingMandatoryPlugin() {
+        final Settings settings =
+                Settings.builder()
+                        .put("path.home", createTempDir())
+                        .put("plugin.mandatory", "fake")
+                        .build();
+        final IllegalStateException e = expectThrows(IllegalStateException.class, () -> newPluginsService(settings));
+        assertThat(e, hasToString(containsString("missing mandatory plugins [fake]")));
+    }
+
+    public void testExistingMandatoryClasspathPlugin() {
+        final Settings settings =
+                Settings.builder()
+                        .put("path.home", createTempDir())
+                        .put("plugin.mandatory", "org.elasticsearch.plugins.PluginsServiceTests$FakePlugin")
+                        .build();
+        newPluginsService(settings, FakePlugin.class);
+    }
+
+    public static class FakePlugin extends Plugin {
+
+        public FakePlugin() {
+
+        }
+
+    }
+
+    public void testExistingMandatoryInstalledPlugin() throws IOException {
+        final Path pathHome = createTempDir();
+        final Path plugins = pathHome.resolve("plugins");
+        final Path fake = plugins.resolve("fake");
+
+        PluginTestUtil.writePluginProperties(
+                fake,
+                "description", "description",
+                "name", "fake",
+                "version", "1.0.0",
+                "elasticsearch.version", Version.CURRENT.toString(),
+                "java.version", System.getProperty("java.specification.version"),
+                "classname", "test.DummyPlugin");
+        try (InputStream jar = PluginsServiceTests.class.getResourceAsStream("dummy-plugin.jar")) {
+            Files.copy(jar, fake.resolve("plugin.jar"));
+        }
+
+        final Settings settings =
+                Settings.builder()
+                        .put("path.home", pathHome)
+                        .put("plugin.mandatory", "fake")
+                        .build();
+        newPluginsService(settings);
+    }
+
+    public void testExistingMandatoryMetaPlugin() throws IOException {
+        final Path pathHome = createTempDir();
+        final Path plugins = pathHome.resolve("plugins");
+        final Path fakeMeta = plugins.resolve("fake-meta");
+
+        PluginTestUtil.writeMetaPluginProperties(fakeMeta, "description", "description", "name", "fake-meta");
+
+        final Path fake = fakeMeta.resolve("fake");
+        PluginTestUtil.writePluginProperties(
+                fake,
+                "description", "description",
+                "name", "fake",
+                "version", "1.0.0",
+                "elasticsearch.version", Version.CURRENT.toString(),
+                "java.version", System.getProperty("java.specification.version"),
+                "classname", "test.DummyPlugin");
+        try (InputStream jar = PluginsServiceTests.class.getResourceAsStream("dummy-plugin.jar")) {
+            Files.copy(jar, fake.resolve("plugin.jar"));
+        }
+
+        final Settings settings =
+                Settings.builder()
+                        .put("path.home", pathHome)
+                        .put("plugin.mandatory", "fake-meta")
+                        .build();
+        newPluginsService(settings);
+    }
+
 }