Add check in pipeline transformation to make sure there is atleast one pipeline with a valid external source

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Author: kkondaka

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

@@ -36,6 +36,7 @@
 import org.opensearch.dataprepper.model.sink.SinkContext;
 import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.pipeline.parser.PipelineConfigurationValidator;
+import org.opensearch.dataprepper.pipeline.parser.InvalidPipelineConfigurationException;
 import org.opensearch.dataprepper.pipeline.parser.model.PipelineConfiguration;
 import org.opensearch.dataprepper.pipeline.parser.model.SinkContextPluginSetting;
 import org.opensearch.dataprepper.validation.PluginError;
@@ -123,11 +124,18 @@ public Map<String, Pipeline> transformConfiguration(final PipelinesDataFlowModel
             }
         }
         Map<String, HeadlessPipeline> headlessPipelines = new LinkedHashMap<>();
+        boolean validSourcePresent = false;
         for (Map.Entry<String, Pipeline> pipeline: pipelineMap.entrySet()){
             if (pipeline.getValue().getSource() instanceof HeadlessPipelineSource) {
                 headlessPipelines.put(pipeline.getKey(), pipeline.getValue());
+            } else if (!(pipeline.getValue().getSource() instanceof PipelineConnector)) {
+                validSourcePresent = true;
             }
         }
+        if (pipelineMap.size() > 0 && !validSourcePresent) {
+            throw new InvalidPipelineConfigurationException(
+                    String.format("No valid source present in the configuration\n"));
+        }
 
         for (Map.Entry<String, Pipeline> pipelineEntry: pipelineMap.entrySet()) {
             final Pipeline pipeline = pipelineEntry.getValue();

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

@@ -79,4 +79,7 @@ public class TestDataProvider {
     public static final String INVALID_ZERO_BUFFER_MULTIPLE_THREADS_NO_SINGLE_THREAD_PROCESSORS_CONFIG_FILE = "src/test/resources/invalid_zero_buffer_multiple_threads_no_single_thread_processors.yml";
     public static Set<String> VALID_MULTIPLE_PIPELINE_NAMES = new HashSet<>(Arrays.asList("test-pipeline-1",
             "test-pipeline-2", "test-pipeline-3"));
+    public static final String INVALID_ONLY_HEADLESS_AND_SUBPIPELINES_CONFIG_FILE = "src/test/resources/invalid_only_headless_and_subpipelines_config.yml";
+    public static final String INVALID_ONLY_HEADLESS_PIPELINES_CONFIG_FILE = "src/test/resources/invalid_only_headless_pipelines_config.yml";
+
 }

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

@@ -39,6 +39,7 @@
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.EventFactory;
 import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
+import org.opensearch.dataprepper.pipeline.parser.InvalidPipelineConfigurationException;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.pipeline.parser.DataPrepperDeserializationProblemHandler;
@@ -395,6 +396,23 @@ void testMultipleSinksWithFailurePipeline() {
         assertThat(((HeadlessPipelineSource)failurePipeline.getSource()).getAcknowledgementsEnabled(), equalTo(false));
     }
 
+    @Test
+    void testPipelineConfigWithOnlyHeadlessPipelines() {
+        mockDataPrepperConfigurationAccesses();
+        final String configFile = TestDataProvider.INVALID_ONLY_HEADLESS_PIPELINES_CONFIG_FILE;
+        final PipelineTransformer pipelineTransformer = createObjectUnderTest(configFile);
+        assertThrows(InvalidPipelineConfigurationException.class, () -> pipelineTransformer.transformConfiguration(this.pipelinesDataFlowModel));
+    }
+
+    @Test
+    void testPipelineConfigWithHeadlessPipelinesAndSubPipelines() {
+        mockDataPrepperConfigurationAccesses();
+        final String configFile = TestDataProvider.INVALID_ONLY_HEADLESS_AND_SUBPIPELINES_CONFIG_FILE;
+        final PipelineTransformer pipelineTransformer = createObjectUnderTest(configFile);
+        assertThrows(InvalidPipelineConfigurationException.class, () -> pipelineTransformer.transformConfiguration(this.pipelinesDataFlowModel));
+    }
+
+
     @Test
     void testMultipleProcessors() {
         mockDataPrepperConfigurationAccesses();

data-prepper-core/src/test/resources/invalid_only_headless_and_subpipelines_config.yml

@@ -0,0 +1,12 @@
+headless-pipeline1:
+  sink:
+    - stdout:
+    - pipeline:
+        name : "sub-pipeline1"
+sub-pipeline1:
+  source:
+    pipeline:
+        name : "headless-pipeline1"
+  sink:
+    - stdout:
+

data-prepper-core/src/test/resources/invalid_only_headless_pipelines_config.yml

@@ -0,0 +1,6 @@
+headless-pipeline1:
+  sink:
+    - stdout:
+headless-pipeline2:
+  sink:
+    - stdout: