Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ENH: Plugin errors consolidator #4863

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@
import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
import org.opensearch.dataprepper.plugins.TestObjectPlugin;
import org.opensearch.dataprepper.plugins.test.TestPlugin;
import org.opensearch.dataprepper.validation.LoggingPluginErrorsHandler;
import org.opensearch.dataprepper.validation.PluginErrorCollector;
import org.opensearch.dataprepper.validation.PluginErrorsHandler;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.util.HashMap;
Expand Down Expand Up @@ -63,6 +65,7 @@ private DefaultPluginFactory createObjectUnderTest() {
coreContext.scan(DefaultPluginFactory.class.getPackage().getName());
coreContext.register(PluginBeanFactoryProvider.class);
coreContext.registerBean(PluginErrorCollector.class, PluginErrorCollector::new);
coreContext.registerBean(PluginErrorsHandler.class, LoggingPluginErrorsHandler::new);
coreContext.registerBean(ExtensionsConfiguration.class, () -> extensionsConfiguration);
coreContext.registerBean(PipelinesDataFlowModel.class, () -> pipelinesDataFlowModel);
coreContext.refresh();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,9 @@
import org.opensearch.dataprepper.plugin.TestPluggableInterface;
import org.opensearch.dataprepper.plugins.test.TestExtension;
import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
import org.opensearch.dataprepper.validation.LoggingPluginErrorsHandler;
import org.opensearch.dataprepper.validation.PluginErrorCollector;
import org.opensearch.dataprepper.validation.PluginErrorsHandler;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.util.ArrayList;
Expand Down Expand Up @@ -70,6 +72,7 @@ public class ExtensionsIT {
private AnnotationConfigApplicationContext coreContext;
private PluginFactory pluginFactory;
private PluginErrorCollector pluginErrorCollector;
private PluginErrorsHandler pluginErrorsHandler;
private String pluginName;
private String pipelineName;

Expand All @@ -78,6 +81,7 @@ void setUp() {
pluginName = "test_plugin_using_extension";
pipelineName = UUID.randomUUID().toString();
pluginErrorCollector = new PluginErrorCollector();
pluginErrorsHandler = new LoggingPluginErrorsHandler();
publicContext = new AnnotationConfigApplicationContext();
publicContext.refresh();

Expand Down Expand Up @@ -108,6 +112,7 @@ void setUp() {
coreContext.registerBean(ObjectMapper.class, () -> new ObjectMapper(new YAMLFactory()));
coreContext.register(PipelineParserConfiguration.class);
coreContext.registerBean(PluginErrorCollector.class, () -> pluginErrorCollector);
coreContext.registerBean(PluginErrorsHandler.class, () -> pluginErrorsHandler);
coreContext.refresh();

pluginFactory = coreContext.getBean(DefaultPluginFactory.class);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.PluginErrorsHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -65,6 +66,7 @@ public class PipelineTransformer {
private final AcknowledgementSetManager acknowledgementSetManager;
private final SourceCoordinatorFactory sourceCoordinatorFactory;
private final PluginErrorCollector pluginErrorCollector;
private final PluginErrorsHandler pluginErrorsHandler;

public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
final PluginFactory pluginFactory,
Expand All @@ -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 PluginErrorsHandler pluginErrorsHandler) {
this.pipelinesDataFlowModel = pipelinesDataFlowModel;
this.pluginFactory = Objects.requireNonNull(pluginFactory);
this.peerForwarderProvider = Objects.requireNonNull(peerForwarderProvider);
Expand All @@ -86,6 +89,7 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
this.acknowledgementSetManager = acknowledgementSetManager;
this.sourceCoordinatorFactory = sourceCoordinatorFactory;
this.pluginErrorCollector = pluginErrorCollector;
this.pluginErrorsHandler = pluginErrorsHandler;
}

public Map<String, Pipeline> transformConfiguration() {
Expand Down Expand Up @@ -165,9 +169,10 @@ private void buildPipelineFromConfiguration(
.stream().filter(pluginError -> pipelineName.equals(pluginError.getPipelineName()))
.collect(Collectors.toList());
if (!subPipelinePluginErrors.isEmpty()) {
pluginErrorsHandler.handleErrors(subPipelinePluginErrors);
throw new InvalidPluginConfigurationException(
String.format("One or more plugins are not configured correctly in the pipeline: %s.\n",
pipelineName) + pluginErrorCollector.getConsolidatedErrorMessage());
pipelineName));
}

final List<List<Processor>> decoratedProcessorSets = processorSets.stream()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.PluginErrorsHandler;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
Expand All @@ -46,7 +47,8 @@ public PipelineTransformer pipelineParser(
final EventFactory eventFactory,
final AcknowledgementSetManager acknowledgementSetManager,
final SourceCoordinatorFactory sourceCoordinatorFactory,
final PluginErrorCollector pluginErrorCollector
final PluginErrorCollector pluginErrorCollector,
final PluginErrorsHandler pluginErrorsHandler
) {
return new PipelineTransformer(pipelinesDataFlowModel,
pluginFactory,
Expand All @@ -57,7 +59,8 @@ public PipelineTransformer pipelineParser(
eventFactory,
acknowledgementSetManager,
sourceCoordinatorFactory,
pluginErrorCollector);
pluginErrorCollector,
pluginErrorsHandler);
}

@Bean
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.opensearch.dataprepper.TestDataProvider;
Expand All @@ -22,6 +24,7 @@
import org.opensearch.dataprepper.core.event.EventFactoryApplicationContextMarker;
import org.opensearch.dataprepper.model.breaker.CircuitBreaker;
import org.opensearch.dataprepper.model.buffer.Buffer;
import org.opensearch.dataprepper.model.configuration.PipelineModel;
import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
import org.opensearch.dataprepper.model.event.Event;
import org.opensearch.dataprepper.model.event.EventFactory;
Expand All @@ -39,14 +42,18 @@
import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
import org.opensearch.dataprepper.validation.PluginError;
import org.opensearch.dataprepper.validation.PluginErrorCollector;
import org.opensearch.dataprepper.validation.PluginErrorsHandler;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Stream;

Expand Down Expand Up @@ -84,6 +91,10 @@ class PipelineTransformerTests {
private SourceCoordinatorFactory sourceCoordinatorFactory;
@Mock
private CircuitBreakerManager circuitBreakerManager;
@Mock
private PluginErrorsHandler pluginErrorsHandler;
@Captor
private ArgumentCaptor<Collection<PluginError>> pluginErrorsArgumentCaptor;

private PluginErrorCollector pluginErrorCollector;

Expand Down Expand Up @@ -111,6 +122,7 @@ void setUp() {

coreContext.scan(DefaultPluginFactory.class.getPackage().getName());
coreContext.registerBean(PluginErrorCollector.class, () -> pluginErrorCollector);
coreContext.registerBean(PluginErrorsHandler.class, () -> pluginErrorsHandler);
coreContext.registerBean(DataPrepperConfiguration.class, () -> dataPrepperConfiguration);
coreContext.registerBean(PipelinesDataFlowModel.class, () -> pipelinesDataFlowModel);
coreContext.refresh();
Expand All @@ -126,9 +138,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,
pluginErrorsHandler);
}

@Test
Expand Down Expand Up @@ -214,17 +228,29 @@ void parseConfiguration_with_invalid_root_pipeline_creates_empty_pipelinesMap()
})
void parseConfiguration_with_incorrect_child_pipeline_returns_empty_pipelinesMap(
final String pipelineConfigurationFileLocation) {
pluginErrorCollector.collectPluginError(
PluginError.builder()
.componentType("non-pipeline-plugin")
.pluginName("preexisting-plugin")
.exception(new RuntimeException())
.build());
mockDataPrepperConfigurationAccesses();
final PipelineTransformer pipelineTransformer = createObjectUnderTest(pipelineConfigurationFileLocation);
final Map<String, Pipeline> connectedPipelines = pipelineTransformer.transformConfiguration();
assertThat(connectedPipelines.size(), equalTo(0));
verifyDataPrepperConfigurationAccesses();
verify(dataPrepperConfiguration).getPipelineExtensions();
assertThat(pluginErrorCollector.getPluginErrors().size(), equalTo(1));
final PluginError pluginError = pluginErrorCollector.getPluginErrors().get(0);
assertThat(pluginErrorCollector.getPluginErrors().size(), equalTo(2));
final PluginError pluginError = pluginErrorCollector.getPluginErrors().get(1);
assertThat(pluginError.getPipelineName(), equalTo("test-pipeline-2"));
assertThat(pluginError.getPluginName(), notNullValue());
assertThat(pluginError.getException(), notNullValue());
verify(pluginErrorsHandler).handleErrors(pluginErrorsArgumentCaptor.capture());
final Collection<PluginError> pluginErrorCollection = pluginErrorsArgumentCaptor.getValue();
assertThat(pluginErrorCollection.size(), equalTo(1));
final PluginError capturedPluginError = new ArrayList<>(pluginErrorCollection).get(0);
assertThat(Set.of(PipelineModel.PROCESSOR_PLUGIN_TYPE, PipelineModel.SINK_PLUGIN_TYPE)
.contains(capturedPluginError.getComponentType()), is(true));
}

@Test
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.PluginErrorsHandler;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
Expand Down Expand Up @@ -59,12 +60,15 @@ class PipelineParserConfigurationTest {
@Mock
private PluginErrorCollector pluginErrorCollector;

@Mock
private PluginErrorsHandler pluginErrorsHandler;

@Test
void pipelineParser() {
final PipelineTransformer pipelineTransformer = pipelineParserConfiguration.pipelineParser(
pipelinesDataFlowModel, pluginFactory, peerForwarderProvider, routerFactory,
dataPrepperConfiguration, circuitBreakerManager, eventFactory, acknowledgementSetManager,
sourceCoordinatorFactory, pluginErrorCollector);
sourceCoordinatorFactory, pluginErrorCollector, pluginErrorsHandler);

assertThat(pipelineTransformer, is(notNullValue()));
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
package org.opensearch.dataprepper.validation;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Named;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Named
public class LoggingPluginErrorsHandler implements PluginErrorsHandler {
private static final Logger LOG = LoggerFactory.getLogger(LoggingPluginErrorsHandler.class);

@Override
public void handleErrors(final Collection<PluginError> pluginErrors) {
final List<String> allErrorMessages = pluginErrors.stream()
.map(PluginError::getErrorMessage)
.collect(Collectors.toList());
final String consolidatedErrorMessage = IntStream.range(0, allErrorMessages.size())
.mapToObj(i -> (i + 1) + ". " + allErrorMessages.get(i))
.collect(Collectors.joining("\n"));
LOG.error(consolidatedErrorMessage);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
package org.opensearch.dataprepper.validation;

import java.util.Collection;

public interface PluginErrorsHandler {

public void handleErrors(final Collection<PluginError> pluginErrors);
}

This file was deleted.

Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
package org.opensearch.dataprepper.validation;

import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Collection;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

class LoggingPluginErrorsHandlerTest {
@Test
void testHandleErrors() {
final Logger mockLogger = mock(Logger.class);

try (final MockedStatic<LoggerFactory> mockedLoggerFactory = mockStatic(LoggerFactory.class)) {
mockedLoggerFactory.when(() -> LoggerFactory.getLogger(LoggingPluginErrorsHandler.class))
.thenReturn(mockLogger);
final LoggingPluginErrorsHandler loggingPluginErrorsHandler = new LoggingPluginErrorsHandler();
final PluginError error1 = mock(PluginError.class);
final PluginError error2 = mock(PluginError.class);
when(error1.getErrorMessage()).thenReturn("Error 1 occurred");
when(error2.getErrorMessage()).thenReturn("Error 2 occurred");
final Collection<PluginError> pluginErrors = Arrays.asList(error1, error2);
loggingPluginErrorsHandler.handleErrors(pluginErrors);
final String expectedMessage = "1. Error 1 occurred\n2. Error 2 occurred";
verify(mockLogger).error(expectedMessage);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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)));
}
}
Loading
Loading