ENH: Plugin errors consolidator

Signed-off-by: George Chen <qchea@amazon.com>
opensearch-project · Aug 22, 2024 · d587b54 · d587b54
1 parent 0387808
commit d587b54
Show file tree

Hide file tree

Showing 12 changed files with 93 additions and 36 deletions.
diff --git a/...re/src/integrationTest/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryIT.java b/...re/src/integrationTest/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryIT.java
@@ -18,6 +18,7 @@
 import org.opensearch.dataprepper.plugins.TestObjectPlugin;
 import org.opensearch.dataprepper.plugins.test.TestPlugin;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
 import java.util.HashMap;
@@ -63,6 +64,7 @@ private DefaultPluginFactory createObjectUnderTest() {
         coreContext.scan(DefaultPluginFactory.class.getPackage().getName());
         coreContext.register(PluginBeanFactoryProvider.class);
         coreContext.registerBean(PluginErrorCollector.class, PluginErrorCollector::new);
+        coreContext.registerBean(PluginErrorsConsolidator.class, PluginErrorsConsolidator::new);
         coreContext.registerBean(ExtensionsConfiguration.class, () -> extensionsConfiguration);
         coreContext.registerBean(PipelinesDataFlowModel.class, () -> pipelinesDataFlowModel);
         coreContext.refresh();

diff --git a/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/ExtensionsIT.java b/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/ExtensionsIT.java
@@ -32,6 +32,7 @@
 import org.opensearch.dataprepper.plugins.test.TestExtension;
 import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
 import java.util.ArrayList;
@@ -70,6 +71,7 @@ public class ExtensionsIT {
     private AnnotationConfigApplicationContext coreContext;
     private PluginFactory pluginFactory;
     private PluginErrorCollector pluginErrorCollector;
+    private PluginErrorsConsolidator pluginErrorsConsolidator;
     private String pluginName;
     private String pipelineName;
 
@@ -78,6 +80,7 @@ void setUp() {
         pluginName = "test_plugin_using_extension";
         pipelineName = UUID.randomUUID().toString();
         pluginErrorCollector = new PluginErrorCollector();
+        pluginErrorsConsolidator = new PluginErrorsConsolidator();
         publicContext = new AnnotationConfigApplicationContext();
         publicContext.refresh();
 
@@ -108,6 +111,7 @@ void setUp() {
         coreContext.registerBean(ObjectMapper.class, () -> new ObjectMapper(new YAMLFactory()));
         coreContext.register(PipelineParserConfiguration.class);
         coreContext.registerBean(PluginErrorCollector.class, () -> pluginErrorCollector);
+        coreContext.registerBean(PluginErrorsConsolidator.class, () -> pluginErrorsConsolidator);
         coreContext.refresh();
 
         pluginFactory = coreContext.getBean(DefaultPluginFactory.class);

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/parser/PipelineTransformer.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/parser/PipelineTransformer.java
@@ -34,6 +34,7 @@
 import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.validation.PluginError;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,6 +66,7 @@ public class PipelineTransformer {
     private final AcknowledgementSetManager acknowledgementSetManager;
     private final SourceCoordinatorFactory sourceCoordinatorFactory;
     private final PluginErrorCollector pluginErrorCollector;
+    private final PluginErrorsConsolidator pluginErrorsConsolidator;
 
     public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
                                final PluginFactory pluginFactory,
@@ -75,7 +77,8 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
                                final EventFactory eventFactory,
                                final AcknowledgementSetManager acknowledgementSetManager,
                                final SourceCoordinatorFactory sourceCoordinatorFactory,
-                               final PluginErrorCollector pluginErrorCollector) {
+                               final PluginErrorCollector pluginErrorCollector,
+                               final PluginErrorsConsolidator pluginErrorsConsolidator) {
         this.pipelinesDataFlowModel = pipelinesDataFlowModel;
         this.pluginFactory = Objects.requireNonNull(pluginFactory);
         this.peerForwarderProvider = Objects.requireNonNull(peerForwarderProvider);
@@ -86,6 +89,7 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
         this.acknowledgementSetManager = acknowledgementSetManager;
         this.sourceCoordinatorFactory = sourceCoordinatorFactory;
         this.pluginErrorCollector = pluginErrorCollector;
+        this.pluginErrorsConsolidator = pluginErrorsConsolidator;
     }
 
     public Map<String, Pipeline> transformConfiguration() {
@@ -167,7 +171,8 @@ private void buildPipelineFromConfiguration(
             if (!subPipelinePluginErrors.isEmpty()) {
                 throw new InvalidPluginConfigurationException(
                         String.format("One or more plugins are not configured correctly in the pipeline: %s.\n",
-                                pipelineName) + pluginErrorCollector.getConsolidatedErrorMessage());
+                                pipelineName) + pluginErrorsConsolidator.consolidatedErrorMessage(
+                                        subPipelinePluginErrors));
             }
 
             final List<List<Processor>> decoratedProcessorSets = processorSets.stream()

diff --git a/...e/src/main/java/org/opensearch/dataprepper/parser/config/PipelineParserConfiguration.java b/...e/src/main/java/org/opensearch/dataprepper/parser/config/PipelineParserConfiguration.java
@@ -22,6 +22,7 @@
 import org.opensearch.dataprepper.pipeline.router.RouterFactory;
 import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -46,7 +47,8 @@ public PipelineTransformer pipelineParser(
             final EventFactory eventFactory,
             final AcknowledgementSetManager acknowledgementSetManager,
             final SourceCoordinatorFactory sourceCoordinatorFactory,
-            final PluginErrorCollector pluginErrorCollector
+            final PluginErrorCollector pluginErrorCollector,
+            final PluginErrorsConsolidator pluginErrorsConsolidator
             ) {
         return new PipelineTransformer(pipelinesDataFlowModel,
                 pluginFactory,
@@ -57,7 +59,8 @@ public PipelineTransformer pipelineParser(
                 eventFactory,
                 acknowledgementSetManager,
                 sourceCoordinatorFactory,
-                pluginErrorCollector);
+                pluginErrorCollector,
+                pluginErrorsConsolidator);
     }
 
     @Bean

diff --git a/...repper-core/src/test/java/org/opensearch/dataprepper/parser/PipelineTransformerTests.java b/...repper-core/src/test/java/org/opensearch/dataprepper/parser/PipelineTransformerTests.java
@@ -39,6 +39,7 @@
 import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.validation.PluginError;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
 import java.time.Duration;
@@ -87,6 +88,8 @@ class PipelineTransformerTests {
 
     private PluginErrorCollector pluginErrorCollector;
 
+    private PluginErrorsConsolidator pluginErrorsConsolidator;
+
     private EventFactory eventFactory;
 
     private DefaultAcknowledgementSetManager acknowledgementSetManager;
@@ -99,6 +102,7 @@ void setUp() {
         eventFactory = mock(EventFactory.class);
         acknowledgementSetManager = mock(DefaultAcknowledgementSetManager.class);
         pluginErrorCollector = new PluginErrorCollector();
+        pluginErrorsConsolidator = new PluginErrorsConsolidator();
         final AnnotationConfigApplicationContext publicContext = new AnnotationConfigApplicationContext();
         publicContext.refresh();
 
@@ -126,9 +130,11 @@ private PipelineTransformer createObjectUnderTest(final String pipelineConfigura
 
         final PipelinesDataFlowModel pipelinesDataFlowModel = new PipelinesDataflowModelParser(
                 new PipelineConfigurationFileReader(pipelineConfigurationFileLocation)).parseConfiguration();
-        return new PipelineTransformer(pipelinesDataFlowModel, pluginFactory, peerForwarderProvider,
-                                  routerFactory, dataPrepperConfiguration, circuitBreakerManager, eventFactory,
-                                  acknowledgementSetManager, sourceCoordinatorFactory, pluginErrorCollector);
+        return new PipelineTransformer(
+                pipelinesDataFlowModel, pluginFactory, peerForwarderProvider,
+                routerFactory, dataPrepperConfiguration, circuitBreakerManager, eventFactory,
+                acknowledgementSetManager, sourceCoordinatorFactory, pluginErrorCollector,
+                pluginErrorsConsolidator);
     }
 
     @Test

diff --git a/...c/test/java/org/opensearch/dataprepper/parser/config/PipelineParserConfigurationTest.java b/...c/test/java/org/opensearch/dataprepper/parser/config/PipelineParserConfigurationTest.java
@@ -20,6 +20,7 @@
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
@@ -59,12 +60,15 @@ class PipelineParserConfigurationTest {
     @Mock
     private PluginErrorCollector pluginErrorCollector;
 
+    @Mock
+    private PluginErrorsConsolidator pluginErrorsConsolidator;
+
     @Test
     void pipelineParser() {
         final PipelineTransformer pipelineTransformer = pipelineParserConfiguration.pipelineParser(
                 pipelinesDataFlowModel, pluginFactory, peerForwarderProvider, routerFactory,
                 dataPrepperConfiguration, circuitBreakerManager, eventFactory, acknowledgementSetManager,
-                sourceCoordinatorFactory, pluginErrorCollector);
+                sourceCoordinatorFactory, pluginErrorCollector, pluginErrorsConsolidator);
 
         assertThat(pipelineTransformer, is(notNullValue()));
     }

diff --git a/...line-parser/src/main/java/org/opensearch/dataprepper/validation/PluginErrorCollector.java b/...line-parser/src/main/java/org/opensearch/dataprepper/validation/PluginErrorCollector.java
@@ -5,8 +5,6 @@
 import javax.inject.Named;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 @Named
 @Getter
@@ -16,16 +14,4 @@ public class PluginErrorCollector {
     public void collectPluginError(final PluginError pluginError) {
         pluginErrors.add(pluginError);
     }
-
-    public List<String> getAllErrorMessages() {
-        return pluginErrors.stream().map(PluginError::getErrorMessage).collect(Collectors.toList());
-    }
-
-    public String getConsolidatedErrorMessage() {
-        final List<String> allErrorMessages = getAllErrorMessages();
-
-        return IntStream.range(0, allErrorMessages.size())
-                .mapToObj(i -> (i + 1) + ". " + allErrorMessages.get(i))
-                .collect(Collectors.joining("\n"));
-    }
 }
diff --git a/...-parser/src/main/java/org/opensearch/dataprepper/validation/PluginErrorsConsolidator.java b/...-parser/src/main/java/org/opensearch/dataprepper/validation/PluginErrorsConsolidator.java
@@ -0,0 +1,18 @@
+package org.opensearch.dataprepper.validation;
+
+import javax.inject.Named;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+@Named
+public class PluginErrorsConsolidator {
+    public String consolidatedErrorMessage(final List<PluginError> pluginErrors) {
+        final List<String> allErrorMessages = pluginErrors.stream()
+                .map(PluginError::getErrorMessage)
+                .collect(Collectors.toList());
+        return IntStream.range(0, allErrorMessages.size())
+                .mapToObj(i -> (i + 1) + ". " + allErrorMessages.get(i))
+                .collect(Collectors.joining("\n"));
+    }
+}
diff --git a/...-parser/src/test/java/org/opensearch/dataprepper/validation/PluginErrorCollectorTest.java b/...-parser/src/test/java/org/opensearch/dataprepper/validation/PluginErrorCollectorTest.java
@@ -4,27 +4,17 @@
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.contains;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 class PluginErrorCollectorTest {
 
     @Test
     void testWithPluginErrors() {
         final PluginErrorCollector objectUnderTest = new PluginErrorCollector();
-        final String testErrorMessage1 = "test error message 1";
-        final String testErrorMessage2 = "test error message 2";
         final PluginError pluginError1 = mock(PluginError.class);
-        when(pluginError1.getErrorMessage()).thenReturn(testErrorMessage1);
         final PluginError pluginError2 = mock(PluginError.class);
-        when(pluginError2.getErrorMessage()).thenReturn(testErrorMessage2);
         objectUnderTest.collectPluginError(pluginError1);
         objectUnderTest.collectPluginError(pluginError2);
         assertThat(objectUnderTest.getPluginErrors().size(), equalTo(2));
-        assertThat(objectUnderTest.getAllErrorMessages().size(), equalTo(2));
-        assertThat(objectUnderTest.getAllErrorMessages(), contains(testErrorMessage1, testErrorMessage2));
-        assertThat(objectUnderTest.getConsolidatedErrorMessage(), equalTo(
-                String.format("1. %s\n2. %s", testErrorMessage1, testErrorMessage2)));
     }
 }
diff --git a/...ser/src/test/java/org/opensearch/dataprepper/validation/PluginErrorsConsolidatorTest.java b/...ser/src/test/java/org/opensearch/dataprepper/validation/PluginErrorsConsolidatorTest.java
@@ -0,0 +1,28 @@
+package org.opensearch.dataprepper.validation;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class PluginErrorsConsolidatorTest {
+
+    @Test
+    void testWithPluginErrors() {
+        final PluginErrorsConsolidator objectUnderTest = new PluginErrorsConsolidator();
+        final String testErrorMessage1 = "test error message 1";
+        final String testErrorMessage2 = "test error message 2";
+        final PluginError pluginError1 = mock(PluginError.class);
+        when(pluginError1.getErrorMessage()).thenReturn(testErrorMessage1);
+        final PluginError pluginError2 = mock(PluginError.class);
+        when(pluginError2.getErrorMessage()).thenReturn(testErrorMessage2);
+        final String consolidatedErrorMessage = objectUnderTest.consolidatedErrorMessage(
+                List.of(pluginError1, pluginError2));
+        assertThat(consolidatedErrorMessage, equalTo(
+                String.format("1. %s\n2. %s", testErrorMessage1, testErrorMessage2)));
+    }
+}
diff --git a/...per-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/ExtensionLoader.java b/...per-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/ExtensionLoader.java
@@ -12,6 +12,7 @@
 import org.opensearch.dataprepper.model.plugin.InvalidPluginDefinitionException;
 import org.opensearch.dataprepper.validation.PluginError;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -26,17 +27,20 @@ public class ExtensionLoader {
     private final ExtensionClassProvider extensionClassProvider;
     private final PluginCreator extensionPluginCreator;
     private final PluginErrorCollector pluginErrorCollector;
+    private final PluginErrorsConsolidator pluginErrorsConsolidator;
 
     @Inject
     ExtensionLoader(
             final ExtensionPluginConfigurationConverter extensionPluginConfigurationConverter,
             final ExtensionClassProvider extensionClassProvider,
             @Named("extensionPluginCreator") final PluginCreator extensionPluginCreator,
-            final PluginErrorCollector pluginErrorCollector) {
+            final PluginErrorCollector pluginErrorCollector,
+            final PluginErrorsConsolidator pluginErrorsConsolidator) {
         this.extensionPluginConfigurationConverter = extensionPluginConfigurationConverter;
         this.extensionClassProvider = extensionClassProvider;
         this.extensionPluginCreator = extensionPluginCreator;
         this.pluginErrorCollector = pluginErrorCollector;
+        this.pluginErrorsConsolidator = pluginErrorsConsolidator;
     }
 
     public List<? extends ExtensionPlugin> loadExtensions() {
@@ -59,10 +63,14 @@ public List<? extends ExtensionPlugin> loadExtensions() {
                     }
                 })
                 .collect(Collectors.toList());
-        if (!pluginErrorCollector.getPluginErrors().isEmpty()) {
+        final List<PluginError> extensionPluginErrors = pluginErrorCollector.getPluginErrors()
+                .stream().filter(pluginError -> PipelinesDataFlowModel.EXTENSION_PLUGIN_TYPE
+                        .equals(pluginError.getComponentType()))
+                .collect(Collectors.toList());
+        if (!extensionPluginErrors.isEmpty()) {
             throw new InvalidPluginConfigurationException(
                     "One or more extension plugins are not configured correctly.\n"
-                            + pluginErrorCollector.getConsolidatedErrorMessage());
+                            + pluginErrorsConsolidator.consolidatedErrorMessage(extensionPluginErrors));
         } else {
             return result;
         }

diff --git a/...plugin-framework/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java b/...plugin-framework/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java
@@ -25,6 +25,7 @@
 import org.opensearch.dataprepper.plugins.test.TestExtensionWithConfig;
 import org.opensearch.dataprepper.validation.PluginError;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsConsolidator;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -60,15 +61,17 @@ class ExtensionLoaderTest {
     @Captor
     private ArgumentCaptor<PluginArgumentsContext> pluginArgumentsContextArgumentCaptor;
     private PluginErrorCollector pluginErrorCollector;
+    private PluginErrorsConsolidator pluginErrorsConsolidator;
 
     private ExtensionLoader createObjectUnderTest() {
         return new ExtensionLoader(extensionPluginConfigurationConverter, extensionClassProvider,
-                extensionPluginCreator, pluginErrorCollector);
+                extensionPluginCreator, pluginErrorCollector, pluginErrorsConsolidator);
     }
 
     @BeforeEach
     void setUp() {
         pluginErrorCollector = new PluginErrorCollector();
+        pluginErrorsConsolidator = new PluginErrorsConsolidator();
     }
 
     @Test