Decoupling PipelineDataFlowModel dependency from PipelineTransformer (#5809)

* decoupled PipelineDataModel dependency from PipelineTransformer. It will now get the pipeline data model as an argument instead of constructor argument

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

Author: san81

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/DataPrepper.java

@@ -5,7 +5,7 @@
 
 package org.opensearch.dataprepper.core;
 
-import io.micrometer.core.instrument.util.StringUtils;
+
 import org.opensearch.dataprepper.DataPrepperShutdownListener;
 import org.opensearch.dataprepper.DataPrepperShutdownOptions;
 import org.opensearch.dataprepper.core.parser.PipelineTransformer;
@@ -38,7 +38,6 @@ public class DataPrepper implements PipelinesProvider {
     private static final Logger LOG = LoggerFactory.getLogger(DataPrepper.class);
     private static final String DATAPREPPER_SERVICE_NAME = "DATAPREPPER_SERVICE_NAME";
     private static final String DEFAULT_SERVICE_NAME = "dataprepper";
-    private static final int MAX_RETRIES = 100;
 
     private final PluginFactory pluginFactory;
     private final PeerForwarderServer peerForwarderServer;
@@ -51,27 +50,28 @@ public class DataPrepper implements PipelinesProvider {
     @Inject
     @Lazy
     private DataPrepperServer dataPrepperServer;
-    private List<DataPrepperShutdownListener> shutdownListeners = new LinkedList<>();
+    private final List<DataPrepperShutdownListener> shutdownListeners = new LinkedList<>();
 
     /**
      * returns serviceName if exists or default serviceName
      * @return serviceName for data-prepper
      */
     public static String getServiceNameForMetrics() {
         final String serviceName = System.getenv(DATAPREPPER_SERVICE_NAME);
-        return StringUtils.isNotBlank(serviceName) ? serviceName : DEFAULT_SERVICE_NAME;
+        return serviceName != null && !serviceName.trim().isEmpty() ? serviceName : DEFAULT_SERVICE_NAME;
     }
 
     @Inject
     public DataPrepper(
+            final PipelinesDataFlowModel pipelinesDataFlowModel,
             final PipelineTransformer pipelineTransformer,
             final PluginFactory pluginFactory,
             final PeerForwarderServer peerForwarderServer,
             final Predicate<Map<String, Pipeline>> shouldShutdownOnPipelineFailurePredicate) {
         this.pluginFactory = pluginFactory;
 
-        transformationPipelines = pipelineTransformer.transformConfiguration();
-        pipelinesDataFlowModel = pipelineTransformer.getPipelinesDataFlowModel();
+        this.pipelinesDataFlowModel = pipelinesDataFlowModel;
+        transformationPipelines = pipelineTransformer.transformConfiguration(pipelinesDataFlowModel);
         this.shouldShutdownOnPipelineFailurePredicate = shouldShutdownOnPipelineFailurePredicate;
         if (transformationPipelines.isEmpty()) {
             throw new RuntimeException("No valid pipeline is available for execution, exiting");

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java

@@ -60,7 +60,6 @@ public class PipelineTransformer {
     private static final Logger LOG = LoggerFactory.getLogger(PipelineTransformer.class);
     private static final String PIPELINE_TYPE = "pipeline";
     private static final String ATTRIBUTE_NAME = "name";
-    private final PipelinesDataFlowModel pipelinesDataFlowModel;
     private final RouterFactory routerFactory;
     private final DataPrepperConfiguration dataPrepperConfiguration;
     private final CircuitBreakerManager circuitBreakerManager;
@@ -75,7 +74,7 @@ public class PipelineTransformer {
 
     private final ExpressionEvaluator expressionEvaluator;
 
-    public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
+    public PipelineTransformer(
                                final PluginFactory pluginFactory,
                                final PeerForwarderProvider peerForwarderProvider,
                                final RouterFactory routerFactory,
@@ -87,7 +86,6 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
                                final PluginErrorCollector pluginErrorCollector,
                                final PluginErrorsHandler pluginErrorsHandler,
                                final ExpressionEvaluator expressionEvaluator) {
-        this.pipelinesDataFlowModel = pipelinesDataFlowModel;
         this.pluginFactory = Objects.requireNonNull(pluginFactory);
         this.peerForwarderProvider = Objects.requireNonNull(peerForwarderProvider);
         this.routerFactory = routerFactory;
@@ -101,7 +99,7 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel,
         this.expressionEvaluator = expressionEvaluator;
     }
 
-    public Map<String, Pipeline> transformConfiguration() {
+    public Map<String, Pipeline> transformConfiguration(final PipelinesDataFlowModel pipelinesDataFlowModel) {
         final Map<String, PipelineConfiguration> pipelineConfigurationMap = pipelinesDataFlowModel.getPipelines().entrySet()
                 .stream()
                 .collect(Collectors.toMap(
@@ -422,10 +420,6 @@ private Buffer applyCircuitBreakerToBuffer(final Source source, final Buffer buf
                 .orElseGet(() -> buffer);
     }
 
-    public PipelinesDataFlowModel getPipelinesDataFlowModel() {
-        return pipelinesDataFlowModel;
-    }
-
     private static class IdentifiedComponent<T> {
         private final T component;
         private final String name;

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/config/PipelineParserConfiguration.java

@@ -5,15 +5,14 @@
 
 package org.opensearch.dataprepper.core.parser.config;
 
-import org.opensearch.dataprepper.expression.ExpressionEvaluator;
 import org.opensearch.dataprepper.core.breaker.CircuitBreakerManager;
 import org.opensearch.dataprepper.core.parser.PipelineTransformer;
 import org.opensearch.dataprepper.core.parser.model.DataPrepperConfiguration;
 import org.opensearch.dataprepper.core.peerforwarder.PeerForwarderProvider;
 import org.opensearch.dataprepper.core.pipeline.router.RouterFactory;
 import org.opensearch.dataprepper.core.sourcecoordination.SourceCoordinatorFactory;
 import org.opensearch.dataprepper.core.validation.PluginErrorCollector;
-import org.opensearch.dataprepper.validation.PluginErrorsHandler;
+import org.opensearch.dataprepper.expression.ExpressionEvaluator;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.model.event.EventFactory;
@@ -24,6 +23,7 @@
 import org.opensearch.dataprepper.pipeline.parser.rule.RuleEvaluator;
 import org.opensearch.dataprepper.pipeline.parser.transformer.DynamicConfigTransformer;
 import org.opensearch.dataprepper.pipeline.parser.transformer.PipelineConfigurationTransformer;
+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;
@@ -39,7 +39,6 @@ public class PipelineParserConfiguration {
 
     @Bean
     public PipelineTransformer pipelineParser(
-            final PipelinesDataFlowModel pipelinesDataFlowModel,
             final PluginFactory pluginFactory,
             final PeerForwarderProvider peerForwarderProvider,
             final RouterFactory routerFactory,
@@ -52,8 +51,7 @@ public PipelineTransformer pipelineParser(
             final PluginErrorsHandler pluginErrorsHandler,
             final ExpressionEvaluator expressionEvaluator
             ) {
-        return new PipelineTransformer(pipelinesDataFlowModel,
-                pluginFactory,
+        return new PipelineTransformer(pluginFactory,
                 peerForwarderProvider,
                 routerFactory,
                 dataPrepperConfiguration,

data-prepper-core/src/test/java/org/opensearch/dataprepper/core/DataPrepperTests.java

@@ -18,6 +18,7 @@
 import org.opensearch.dataprepper.core.pipeline.Pipeline;
 import org.opensearch.dataprepper.core.pipeline.PipelineObserver;
 import org.opensearch.dataprepper.core.pipeline.server.DataPrepperServer;
+import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
 
 import java.lang.reflect.Field;
@@ -40,6 +41,8 @@
 public class DataPrepperTests {
     private Map<String, Pipeline> parseConfigurationFixture;
     @Mock
+    private PipelinesDataFlowModel pipelinesDataFlowModel;
+    @Mock
     private PipelineTransformer pipelineTransformer;
     @Mock
     private Pipeline pipeline;
@@ -59,12 +62,13 @@ public void before() {
         lenient().when(pipeline.getName()).thenReturn("testKey");
         lenient().when(pipeline.isReady()).thenReturn(true);
 
-        lenient().when(pipelineTransformer.transformConfiguration())
+        lenient().when(pipelineTransformer.transformConfiguration(this.pipelinesDataFlowModel))
                 .thenReturn(parseConfigurationFixture);
     }
 
     private DataPrepper createObjectUnderTest() throws NoSuchFieldException, IllegalAccessException {
-        final DataPrepper dataPrepper = new DataPrepper(pipelineTransformer, pluginFactory, peerForwarderServer, shouldShutdownOnPipelineFailurePredicate);
+        final DataPrepper dataPrepper = new DataPrepper(pipelinesDataFlowModel, pipelineTransformer, pluginFactory,
+                peerForwarderServer, shouldShutdownOnPipelineFailurePredicate);
         final Field dataPrepperServerField = dataPrepper.getClass().getDeclaredField("dataPrepperServer");
         dataPrepperServerField.setAccessible(true);
         dataPrepperServerField.set(dataPrepper, dataPrepperServer);
@@ -87,7 +91,8 @@ public void testGivenInvalidInputThenExceptionThrown() {
 
         assertThrows(
                 RuntimeException.class,
-                () -> new DataPrepper(pipelineTransformer, pluginFactory, peerForwarderServer, shouldShutdownOnPipelineFailurePredicate),
+                () -> new DataPrepper(pipelinesDataFlowModel, pipelineTransformer, pluginFactory,
+                        peerForwarderServer, shouldShutdownOnPipelineFailurePredicate),
                 "Exception should be thrown if pipeline parser has no pipeline configuration");
     }