Code cleanup and add comments

Signed-off-by: srigovs <srigovs@amazon.com>

data-prepper-pipeline-parser/build.gradle

@@ -19,7 +19,6 @@ dependencies {
     implementation 'org.projectlombok:lombok:1.18.22'
     implementation 'com.jayway.jsonpath:json-path:2.6.0'
     implementation 'javax.inject:javax.inject:1'
-//    implementation 'org.testng:testng:7.1.0'
     implementation(libs.spring.core) {
         exclude group: 'commons-logging', module: 'commons-logging'
     }
@@ -28,6 +27,8 @@ dependencies {
     }
     implementation 'software.amazon.cloudwatchlogs:aws-embedded-metrics:2.0.0-beta-1'
     testImplementation 'org.apache.logging.log4j:log4j-jpl:2.23.0'
+    testImplementation 'org.assertj:assertj-core:3.20.2'
+
     compileOnly 'org.projectlombok:lombok:1.18.20'
     annotationProcessor 'org.projectlombok:lombok:1.18.20'
 }

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/PipelinesDataflowModelParser.java

@@ -35,14 +35,9 @@ public PipelinesDataflowModelParser(final PipelineConfigurationReader pipelineCo
     }
 
 
-    //TODO
-    // move transformation code after user pipeline validation in pipelineTransformer.java
     public PipelinesDataFlowModel parseConfiguration() {
         final List<PipelinesDataFlowModel> pipelinesDataFlowModels = parseStreamsToPipelinesDataFlowModel();
         PipelinesDataFlowModel pipelinesDataFlowModel = mergePipelinesDataModels(pipelinesDataFlowModels);
-
-//        performPipelineConfigurationTransformationIfNeeded(pipelinesDataFlowModel);
-
         return pipelinesDataFlowModel;
     }
 
@@ -60,21 +55,6 @@ private List<PipelinesDataFlowModel> parseStreamsToPipelinesDataFlowModel() {
                 .collect(Collectors.toList());
     }
 
-    //TODO
-//    private PipelinesDataFlowModel performPipelineConfigurationTransformationIfNeeded(PipelinesDataFlowModel pipelinesDataFlowModel) {
-//
-//        //check if transformation is required based on rules present in yaml file
-//        RuleEvaluator ruleEvaluator = new RuleEvaluator(pipelinesDataFlowModel);
-//        PipelineConfigurationTransformer configurationTransformer = new DynamicYamlTransformer();
-//
-//        if (ruleEvaluator.isTransformationNeeded()) {
-//            PipelineTemplateModel templateModel = ruleEvaluator.getTemplateModel();
-//            PipelinesDataFlowModel transformedPipelineDataModel = configurationTransformer.transformConfiguration(pipelinesDataFlowModel, templateModel);
-//            return transformedPipelineDataModel;
-//        }
-//        return pipelinesDataFlowModel;
-//    }
-
     private PipelinesDataFlowModel parseStreamToPipelineDataFlowModel(final InputStream configurationInputStream) {
         try (final InputStream pipelineConfigurationInputStream = configurationInputStream) {
             final PipelinesDataFlowModel pipelinesDataFlowModel = OBJECT_MAPPER.readValue(pipelineConfigurationInputStream,

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/README.md

@@ -8,20 +8,68 @@ User provided configuration passes through rules, if the rules are valid,
 the template for the transformations are dynamically chosen and applied.
 
 **User config**
+```aidl
+simple-pipeline:
+  source:
+    someSource:
+      hostname: "database.example.com"
+      port: "27017"
+  sink:   
+    - opensearch:
+       hosts: ["https://search-service.example.com"]
+       index: "my_index"
+
+```
 
 **Template**
+```aidl
+"<<pipeline-name>>-transformed":
+  source: "<<$.*.someSource>>"
+  sink:
+    - opensearch:
+       hosts: "<<$.*.sink[?(@.opensearch)].opensearch.hosts>>"
+       port: "<<$.*.someSource.documentdb.port>>"
+       index: "<<$.*.sink[0].opensearch.index>>"
+       aws:
+        sts_role_arn: "arn123"
+        region: "us-test-1"
+       dlq:
+          s3:
+              bucket: "test-bucket"
+```
 
 **Rule**
+```
+apply_when:
+  - "$..source.someSource"
+  ```
 
 **Expected Transformed Config**
-
+```aidl
+simple-pipeline-transformed:
+  source:
+    someSource:
+      hostname: "database.example.com"
+      port: "27017"
+  sink:
+    - opensearch:
+        hosts: ["https://search-service.example.com"]
+        port: "27017"
+        index: "my_index"
+        aws:
+          sts_role_arn: "arn123"
+          region: "us-test-1"
+        dlq:
+          s3:
+            bucket: "test-bucket"
+```
 
 ### Assumptions
-1. Deep scan or recursive expressions like`$..` is NOT supported. Always use a more specific expression.
+1. In the template definition, Deep scan or recursive expressions like`$..` is NOT supported. Always use a more specific expression.
 In the event specific variables in a path are not known, use wildcards.
 2. User could provide multiple pipelines in their user config but 
 there can be only one pipeline that can support transformation.
 3. There cannot be multiple transformations in a single pipeline.
-4. `{{$ .. }}` is the placeholder in the template. //TODO
-`{{ pipeline-name }}` is handled differently as compared to other placeholders
+4. `<<$ .. >>` is the placeholder in the template. //TODO
+`<< pipeline-name >>` is handled differently as compared to other placeholders
 as other placeholders are jsonPaths.

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/rule/RuleEvaluator.java

@@ -19,6 +19,8 @@
 import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.pipeline.parser.transformer.PipelineTemplateModel;
 import org.opensearch.dataprepper.pipeline.parser.transformer.TransformersFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -27,6 +29,7 @@
 
 public class RuleEvaluator {
 
+    private static final Logger LOG = LoggerFactory.getLogger(RuleEvaluator.class);
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
     private static final ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
     private final TransformersFactory transformersFactory;
@@ -41,34 +44,36 @@ public RuleEvaluatorResult isTransformationNeeded(PipelinesDataFlowModel pipelin
     }
 
     /**
-     *  Evaluates based on rules if it is docDB and needs transformation.
-     *  Assumption: only one pipeline can have transformation.
+     * Evaluates model based on pre defined rules and
+     * result contains the name of the pipeline that will need transformation,
+     * evaluated boolean result and the corresponding template model
+     * Assumption: only one pipeline can have transformation.
      * @param pipelinesModel
-     * @return
+     * @return RuleEvaluatorResult
      */
     private RuleEvaluatorResult isDocDBSource(PipelinesDataFlowModel pipelinesModel) {
-        //TODO
-        //dynamically find pluginName Needed for transformation based on rule that applies
         PLUGIN_NAME = "documentdb";
         String pluginRulesPath = transformersFactory.getPluginRuleFileLocation(PLUGIN_NAME);
         Map<String, PipelineModel> pipelines = pipelinesModel.getPipelines();
 
         for (Map.Entry<String, PipelineModel> entry : pipelines.entrySet()) {
             try {
                 String pipelineJson = OBJECT_MAPPER.writeValueAsString(entry);
-                if(evaluate(pipelineJson, pluginRulesPath)) {
+                if (evaluate(pipelineJson, pluginRulesPath)) {
                     String templateFilePath = transformersFactory.getPluginTemplateFileLocation(PLUGIN_NAME);
                     PipelineTemplateModel templateModel = yamlMapper.readValue(new File(templateFilePath),
                             PipelineTemplateModel.class);
                     return RuleEvaluatorResult.builder()
                             .withEvaluatedResult(true)
                             .withPipelineTemplateModel(templateModel)
-                            .withPipelineName(entry.getKey()) // TODO naming
+                            .withPipelineName(entry.getKey())
                             .build();
                 }
             } catch (JsonProcessingException e) {
+                LOG.error("Error processing json");
                 throw new RuntimeException(e);
             } catch (IOException e) {
+                LOG.error("Error reading file");
                 throw new RuntimeException(e);
             }
         }
@@ -82,7 +87,6 @@ private RuleEvaluatorResult isDocDBSource(PipelinesDataFlowModel pipelinesModel)
     private Boolean evaluate(String pipelinesJson,
                              String rulePath) {
 
-//        ReadContext readPathContext = null;
         Configuration parseConfig = Configuration.builder()
                 .jsonProvider(new JacksonJsonProvider())
                 .mappingProvider(new JacksonMappingProvider())
@@ -98,12 +102,10 @@ private Boolean evaluate(String pipelinesJson,
                 Object result = readPathContext.read(rule);
             }
         } catch (IOException e) {
-            //TODO --> this is failing the integ tests in core
-            // reason - transformer not injected correctly
-//            throw new RuntimeException(format("error reading file %s.",rulePath));
-
+            LOG.warn("Error reading file {}", rulePath);
             return false;
         } catch (PathNotFoundException e) {
+            LOG.warn("Path not found {}", rulePath);
             return false;
         }
         return true;