added default mapper for mapping logstash config models to data prepper plugin model

data-prepper-logstash-configuration/build.gradle

@@ -11,6 +11,9 @@ repositories {
 dependencies {
     antlr "org.antlr:antlr4:4.9.2"
     implementation project(':data-prepper-api')
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    testImplementation 'org.slf4j:slf4j-simple:1.7.32'
     testImplementation "org.hamcrest:hamcrest:2.2"
     testImplementation "org.mockito:mockito-inline:${versionMap.mockito}"
     testImplementation platform("org.junit:junit-bom:${versionMap.junitJupiter}")

data-prepper-logstash-configuration/src/main/java/org/opensearch/dataprepper/logstash/exception/LogstashConfigurationException.java

@@ -10,4 +10,9 @@ public class LogstashConfigurationException extends RuntimeException {
     public LogstashConfigurationException(String errorMessage) {
         super(errorMessage);
     }
+
+    public LogstashConfigurationException(String errorMessage, Exception inner) {
+        super(errorMessage, inner);
+    }
+
 }

data-prepper-logstash-configuration/src/main/java/org/opensearch/dataprepper/logstash/exception/LogstashMappingException.java

@@ -0,0 +1,18 @@
+package org.opensearch.dataprepper.logstash.exception;
+
+/**
+ * Exception thrown when attempting to map between Logstash configuration model and Data Prepper model
+ *
+ * @since 1.2
+ */
+public class LogstashMappingException extends LogstashConfigurationException{
+
+    public LogstashMappingException(String errorMessage) {
+        super(errorMessage);
+    }
+
+    public LogstashMappingException(String errorMessage, Exception inner) {
+        super(errorMessage, inner);
+    }
+
+}

data-prepper-logstash-configuration/src/main/java/org/opensearch/dataprepper/logstash/mapping/DefaultPluginMapper.java

@@ -0,0 +1,62 @@
+package org.opensearch.dataprepper.logstash.mapping;
+
+import com.amazon.dataprepper.model.configuration.PluginModel;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import org.opensearch.dataprepper.logstash.exception.LogstashMappingException;
+import org.opensearch.dataprepper.logstash.model.LogstashPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Converts Logstash plugin model to Data Prepper plugin model using mapping file
+ *
+ * @since 1.2
+ */
+public class DefaultPluginMapper implements LogstashPluginMapper {
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultPluginMapper.class);
+    private final ObjectMapper objectMapper = new ObjectMapper(new YAMLFactory());
+
+    @Override
+    public PluginModel mapPlugin(LogstashPlugin logstashPlugin) {
+
+        String mappingResourceName = logstashPlugin.getPluginName() + ".mapping.yaml";
+
+        final InputStream inputStream = this.getClass().getResourceAsStream(mappingResourceName);
+        if (inputStream == null) {
+            throw new LogstashMappingException("Unable to find mapping resource " + mappingResourceName);
+        }
+
+        LogstashMappingModel logstashMappingModel;
+        try {
+            logstashMappingModel = objectMapper.readValue(inputStream, LogstashMappingModel.class);
+        }
+        catch(IOException ex) {
+            throw new LogstashMappingException("Unable to parse mapping file " + mappingResourceName, ex);
+        }
+
+        if (logstashMappingModel.getPluginName() == null) {
+            throw new LogstashMappingException("The mapping file " + mappingResourceName + " has a null value for 'pluginName'.");
+        }
+        Map<String, Object> pluginSettings = new LinkedHashMap<>(logstashMappingModel.getAdditionalAttributes());
+
+        logstashPlugin.getAttributes().forEach(logstashAttribute -> {
+            if (logstashMappingModel.getMappedAttributeNames().containsKey(logstashAttribute.getAttributeName())) {
+                pluginSettings.put(
+                        logstashMappingModel.getMappedAttributeNames().get(logstashAttribute.getAttributeName()),
+                        logstashAttribute.getAttributeValue().getValue()
+                );
+            }
+            else {
+                LOG.warn("Attribute name {} is not found in mapping file.", logstashAttribute.getAttributeName());
+            }
+        });
+
+        return new PluginModel(logstashMappingModel.getPluginName(), pluginSettings);
+    }
+}

data-prepper-logstash-configuration/src/main/java/org/opensearch/dataprepper/logstash/mapping/LogstashMappingModel.java

@@ -9,23 +9,25 @@
  */
 public class LogstashMappingModel {
 
-    private final String pluginName;
-    private final Map<String, Object> mappedAttributes;
-    private final Map<String, Object> additionalAttributes;
+    private String pluginName;
+    private Map<String, String> mappedAttributeNames;
+    private Map<String, Object> additionalAttributes;
 
-    public LogstashMappingModel(final String pluginName, final Map<String, Object> mappedAttributes,
+    public LogstashMappingModel(final String pluginName, final Map<String, String> mappedAttributeNames,
                                 final Map<String, Object> additionalAttributes) {
         this.pluginName = pluginName;
-        this.mappedAttributes = mappedAttributes;
+        this.mappedAttributeNames = mappedAttributeNames;
         this.additionalAttributes = additionalAttributes;
     }
 
+    public LogstashMappingModel() {}
+
     public String getPluginName() {
         return pluginName;
     }
 
-    public Map<String, Object> getMappedAttributes() {
-        return mappedAttributes;
+    public Map<String, String> getMappedAttributeNames() {
+        return mappedAttributeNames;
     }
 
     public Map<String, Object> getAdditionalAttributes() {

data-prepper-logstash-configuration/src/test/java/org/opensearch/dataprepper/logstash/mapping/DefaultPluginMapperTest.java

@@ -0,0 +1,75 @@
+package org.opensearch.dataprepper.logstash.mapping;
+
+import com.amazon.dataprepper.model.configuration.PluginModel;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.logstash.exception.LogstashMappingException;
+import org.opensearch.dataprepper.logstash.model.LogstashPlugin;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertThrows;
+
+class DefaultPluginMapperTest {
+
+    private DefaultPluginMapper defaultPluginMapper;
+
+    @BeforeEach
+    void createObjectUnderTest() {
+        defaultPluginMapper = new DefaultPluginMapper();
+    }
+
+    @Test
+    void mapPlugin_without_mapping_file_throws_logstash_mapping_exception_Test() {
+        LogstashPlugin logstashPlugin = TestDataProvider.invalidMappingResourceNameData();
+        String mappingResourceName = logstashPlugin.getPluginName() + ".mapping.yaml";
+
+        Exception exception = assertThrows(LogstashMappingException.class, () ->
+                defaultPluginMapper.mapPlugin(logstashPlugin));
+
+        String expectedMessage = "Unable to find mapping resource " + mappingResourceName;
+        String actualMessage = exception.getMessage();
+
+        assertThat(expectedMessage, equalTo(actualMessage));
+    }
+
+    @Test
+    void mapPlugin_with_incorrect_mapping_file_throws_logstash_mapping_exception_Test() {
+        LogstashPlugin logstashPlugin = TestDataProvider.invalidMappingResourceData();
+        String mappingResourceName = logstashPlugin.getPluginName() + ".mapping.yaml";
+
+        Exception exception = assertThrows(LogstashMappingException.class, () ->
+                defaultPluginMapper.mapPlugin(logstashPlugin));
+
+        String expectedMessage = "Unable to parse mapping file " + mappingResourceName;
+        String actualMessage = exception.getMessage();
+
+        assertThat(expectedMessage, equalTo(actualMessage));
+    }
+
+    @Test
+    void mapPlugin_without_plugin_name_in_mapping_file_throws_logstash_mapping_exception_Test() {
+        LogstashPlugin logstashPlugin = TestDataProvider.noPluginNameMappingResourceData();
+        String mappingResourceName = logstashPlugin.getPluginName() + ".mapping.yaml";
+
+        Exception exception = assertThrows(LogstashMappingException.class, () ->
+                defaultPluginMapper.mapPlugin(logstashPlugin));
+
+        String expectedMessage = "The mapping file " + mappingResourceName + " has a null value for 'pluginName'.";
+        String actualMessage = exception.getMessage();
+
+        assertThat(expectedMessage, equalTo(actualMessage));
+    }
+
+    @Test
+    void mapPlugin_returns_plugin_model_Test() {
+        LogstashPlugin logstashPlugin = TestDataProvider.pluginData();
+
+        PluginModel actualPluginModel = defaultPluginMapper.mapPlugin(logstashPlugin);
+        PluginModel expectedPluginModel = TestDataProvider.getSamplePluginModel();
+
+        assertThat(expectedPluginModel.getPluginName(), equalTo(actualPluginModel.getPluginName()));
+        assertThat(expectedPluginModel.getPluginSettings(), equalTo(actualPluginModel.getPluginSettings()));
+    }
+
+}

data-prepper-logstash-configuration/src/test/java/org/opensearch/dataprepper/logstash/mapping/TestDataProvider.java

@@ -0,0 +1,71 @@
+package org.opensearch.dataprepper.logstash.mapping;
+
+import com.amazon.dataprepper.model.configuration.PluginModel;
+import org.opensearch.dataprepper.logstash.model.LogstashAttribute;
+import org.opensearch.dataprepper.logstash.model.LogstashAttributeValue;
+import org.opensearch.dataprepper.logstash.model.LogstashPlugin;
+import org.opensearch.dataprepper.logstash.model.LogstashValueType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class TestDataProvider {
+
+    public static LogstashPlugin invalidMappingResourceNameData() {
+        return LogstashPlugin.builder()
+                .pluginName("amazon_elasticsearch")
+                .attributes(Collections.singletonList(getArrayTypeAttribute())).build();
+    }
+
+    public static LogstashPlugin invalidMappingResourceData() {
+        return LogstashPlugin.builder()
+                .pluginName("invalid")
+                .attributes(Collections.singletonList(getArrayTypeAttribute())).build();
+    }
+
+    public static LogstashPlugin noPluginNameMappingResourceData() {
+        return LogstashPlugin.builder()
+                .pluginName("no_plugin_name")
+                .attributes(Collections.singletonList(getArrayTypeAttribute())).build();
+    }
+
+    public static LogstashPlugin pluginData() {
+        return LogstashPlugin.builder()
+                .pluginName("amazon_es")
+                .attributes(Arrays.asList(getArrayTypeAttribute(), getStringTypeAttribute())).build();
+    }
+
+    public static PluginModel getSamplePluginModel() {
+        Map<String, Object> attributes = new LinkedHashMap<>();
+        attributes.put("aws_sigv4", true);
+        attributes.put("insecure", false);
+        attributes.put("aws_region", "us-west-2");
+        attributes.put("hosts", Collections.singletonList("https://localhost:9200"));
+        return new PluginModel("opensearch", attributes);
+    }
+
+    private static LogstashAttribute getArrayTypeAttribute() {
+        LogstashAttributeValue logstashAttributeValue = LogstashAttributeValue.builder()
+                .attributeValueType(LogstashValueType.ARRAY)
+                .value(Collections.singletonList("https://localhost:9200"))
+                .build();
+        return LogstashAttribute.builder()
+                .attributeName("hosts")
+                .attributeValue(logstashAttributeValue)
+                .build();
+    }
+
+    private static LogstashAttribute getStringTypeAttribute() {
+        LogstashAttributeValue logstashAttributeValue = LogstashAttributeValue.builder()
+                .attributeValueType(LogstashValueType.STRING)
+                .value("us-west-2")
+                .build();
+        return LogstashAttribute.builder()
+                .attributeName("region")
+                .attributeValue(logstashAttributeValue)
+                .build();
+    }
+
+}

data-prepper-logstash-configuration/src/test/resources/org/opensearch/dataprepper/logstash/mapping/amazon_es.mapping.yaml

@@ -0,0 +1,8 @@
+pluginName: opensearch
+mappedAttributeNames:
+  hosts: hosts
+  index: index
+  region: aws_region
+additionalAttributes:
+  aws_sigv4: true
+  insecure: false

data-prepper-logstash-configuration/src/test/resources/org/opensearch/dataprepper/logstash/mapping/invalid.mapping.yaml

@@ -0,0 +1,8 @@
+pluginName:
+  mappedAttributeNames:
+    hosts: hosts
+    cacert: cert
+    user: username
+    password: password
+    index: index
+    proxy: proxy

data-prepper-logstash-configuration/src/test/resources/org/opensearch/dataprepper/logstash/mapping/no_plugin_name.mapping.yaml

@@ -0,0 +1,7 @@
+mappedAttributeNames:
+  hosts: hosts
+  index: index
+  region: aws_region
+additionalAttributes:
+  aws_sigv4: true
+  insecure: false