Enabling experimental plugins specifically by plugin type and name (opensearch-project#5676)

Support enabling experimental plugins specifically by plugin type and name. This also includes a change to the core plugin classes to allow them to define themselves as a plugin component type along with a name that is used to create the mapping in the YAML file.

Resolves opensearch-project#5675

Signed-off-by: David Venable <dlv@amazon.com>

Author: dlvenable

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/Buffer.java

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.model.buffer;
 
 import org.opensearch.dataprepper.model.CheckpointState;
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.time.Duration;
@@ -18,6 +19,7 @@
  * Buffer queues the records between TI components and acts as a layer between source and processor/sink. Buffer can
  * be in-memory, disk based or other a standalone implementation.
  */
+@PluginComponentType("buffer")
 public interface Buffer<T extends Record<?>> {
     /**
      * writes the record to the buffer

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/PluginComponentType.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.model.plugin;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation for Data Prepper plugin type components.
+ * Intended for processor, sink, source, buffer.
+ *
+ * @since 2.12
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
+public @interface PluginComponentType {
+    /**
+     * Gets the name of the plugin component type.
+     *
+     * @return The name of the plugin component type.
+     */
+    String value();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/processor/Processor.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.model.processor;
 
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.util.Collection;
@@ -14,6 +15,7 @@
  * Processor interface. These are intermediary processing units using which users can filter,
  * transform and enrich the records into desired format before publishing to the sink.
  */
+@PluginComponentType("processor")
 public interface Processor<InputRecord extends Record<?>, OutputRecord extends Record<?>> {
 
     /**

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/Sink.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.model.sink;
 
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.util.Collection;
@@ -13,6 +14,7 @@
  * Data Prepper sink interface. Sink may publish records to a disk, a file,
  * to OpenSearch, other pipelines, or other external systems.
  */
+@PluginComponentType("sink")
 public interface Sink<T extends Record<?>> {
 
     /**

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/source/Source.java

@@ -6,13 +6,15 @@
 package org.opensearch.dataprepper.model.source;
 
 import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.codec.HasByteDecoder;
 
 /**
  * Data Prepper source interface. Source acts as receiver of the events that flow
  * through the transformation pipeline.
  */
+@PluginComponentType("source")
 public interface Source<T extends Record<?>> extends HasByteDecoder {
 
     /**

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginFactory.java

@@ -144,12 +144,14 @@ private <T> Class<? extends T> getPluginClass(final Class<T> baseClass, final St
                 .orElseThrow(() -> new NoPluginFoundException(
                         "Unable to find a plugin named '" + pluginName + "'. Please ensure that plugin is annotated with appropriate values."));
 
-        handleDefinedPlugins(pluginClass, pluginName);
+        handleDefinedPlugins(pluginClass, baseClass, pluginName);
         return pluginClass;
     }
 
-    private <T> void handleDefinedPlugins(final Class<? extends T> pluginClass, final String pluginName) {
-        final DefinedPlugin<? extends T> definedPlugin = new DefinedPlugin<>(pluginClass, pluginName);
+    private <T> void handleDefinedPlugins(final Class<? extends T> pluginClass,
+                                          final Class<? extends T> pluginTypeClass,
+                                          final String pluginName) {
+        final DefinedPlugin<? extends T> definedPlugin = new DefinedPlugin<>(pluginClass, pluginTypeClass, pluginName);
 
         definedPluginConsumers.forEach(definedPluginConsumer -> definedPluginConsumer.accept(definedPlugin));
     }

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/DefinedPlugin.java

@@ -9,14 +9,20 @@
 
 package org.opensearch.dataprepper.plugin;
 
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
+
 import java.util.Objects;
 
 class DefinedPlugin<T> {
     private final Class<? extends T> pluginClass;
+    private final Class<? extends T> pluginTypeClass;
     private final String pluginName;
 
-    public DefinedPlugin(final Class<? extends T> pluginClass, final String pluginName) {
+    public DefinedPlugin(final Class<? extends T> pluginClass,
+                         final Class<? extends T> pluginTypeClass,
+                         final String pluginName) {
         this.pluginClass = Objects.requireNonNull(pluginClass);
+        this.pluginTypeClass = Objects.requireNonNull(pluginTypeClass);
         this.pluginName = Objects.requireNonNull(pluginName);
     }
 
@@ -27,4 +33,12 @@ public Class<? extends T> getPluginClass() {
     public String getPluginName() {
         return pluginName;
     }
+
+    public String getPluginTypeName() {
+        if(pluginTypeClass.isAnnotationPresent(PluginComponentType.class)) {
+            return pluginTypeClass.getAnnotation(PluginComponentType.class).value();
+        }
+
+        return null;
+    }
 }

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/ExperimentalConfiguration.java

@@ -11,6 +11,10 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * Data Prepper configurations for experimental features.
  *
@@ -20,6 +24,9 @@ public class ExperimentalConfiguration {
     @JsonProperty("enable_all")
     private boolean enableAll = false;
 
+    @JsonProperty("enabled_plugins")
+    private Map<String, Set<String>> enabledPlugins;
+
     public static ExperimentalConfiguration defaultConfiguration() {
         return new ExperimentalConfiguration();
     }
@@ -32,4 +39,14 @@ public static ExperimentalConfiguration defaultConfiguration() {
     public boolean isEnableAll() {
         return enableAll;
     }
+
+    /**
+     * Gets enabled plugins by plugin type.
+     *
+     * @return A map of plugin types to list of allowed plugins by name.
+     * @since 2.12
+     */
+    public Map<String, Set<String>> getEnabledPlugins() {
+        return enabledPlugins != null ? enabledPlugins : Collections.emptyMap();
+    }
 }

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/ExperimentalPluginValidator.java

@@ -13,6 +13,8 @@
 import org.opensearch.dataprepper.model.plugin.NoPluginFoundException;
 
 import javax.inject.Named;
+import java.util.Collections;
+import java.util.Set;
 import java.util.function.Consumer;
 
 @Named
@@ -25,13 +27,22 @@ class ExperimentalPluginValidator implements Consumer<DefinedPlugin<?>> {
 
     @Override
     public void accept(final DefinedPlugin<?> definedPlugin) {
-        if(isPluginDisallowedAsExperimental(definedPlugin.getPluginClass())) {
+        if(isPluginDisallowedAsExperimental(definedPlugin)) {
             throw new NoPluginFoundException("Unable to create experimental plugin " + definedPlugin.getPluginName() +
                     ". You must enable experimental plugins in data-prepper-config.yaml in order to use them.");
         }
     }
 
-    private boolean isPluginDisallowedAsExperimental(final Class<?> pluginClass) {
-        return pluginClass.isAnnotationPresent(Experimental.class) && !experimentalConfiguration.isEnableAll();
+    private boolean isPluginDisallowedAsExperimental(final DefinedPlugin<?> definedPlugin) {
+        final Class<?> pluginClass = definedPlugin.getPluginClass();
+        if(!pluginClass.isAnnotationPresent(Experimental.class))
+            return false;
+        if(experimentalConfiguration.isEnableAll())
+            return false;
+
+        final Set<String> enabledPluginsForType =
+                experimentalConfiguration.getEnabledPlugins()
+                        .getOrDefault(definedPlugin.getPluginTypeName(), Collections.emptySet());
+        return !enabledPluginsForType.contains(definedPlugin.getPluginName());
     }
 }

data-prepper-plugin-framework/src/test/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryTest.java

@@ -248,6 +248,7 @@ void loadPlugin_should_call_all_definedPluginConsumers() {
 
                 final DefinedPlugin<?> actualDefinedPlugin = definedPluginArgumentCaptor.getValue();
                 assertThat(actualDefinedPlugin.getPluginClass(), equalTo(expectedPluginClass));
+                assertThat(actualDefinedPlugin.getPluginTypeName(), equalTo("sink"));
                 assertThat(actualDefinedPlugin.getPluginName(), equalTo(pluginName));
             }
         }
@@ -357,6 +358,7 @@ void loadPlugin_with_varargs_should_call_all_definedPluginConsumers() {
 
                 final DefinedPlugin<?> actualDefinedPlugin = definedPluginArgumentCaptor.getValue();
                 assertThat(actualDefinedPlugin.getPluginClass(), equalTo(expectedPluginClass));
+                assertThat(actualDefinedPlugin.getPluginTypeName(), equalTo("sink"));
                 assertThat(actualDefinedPlugin.getPluginName(), equalTo(pluginName));
             }
         }