Added LogstashPluginAttributesMapper for custom mapping (#568)

Refactored the DefaultPluginMapper to use LogstashPluginAttributesMapper as the mechanism for mapping plugins. This can be configure in the mapping YAML to allow detailed mapping configurations. #466

Signed-off-by: David Venable <[email protected]>

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import org.opensearch.dataprepper.logstash.exception.LogstashMappingException;
+
+import java.lang.reflect.Constructor;
+
+class AttributesMapperCreator {
+
+    LogstashPluginAttributesMapper createMapperClass(final String attributesMapperClassName) {
+        final Class<?> attributesMapperClass;
+        try {
+            attributesMapperClass = Class.forName(attributesMapperClassName);
+        } catch (final ClassNotFoundException ex) {
+            throw new LogstashMappingException("Unable to find Mapper class with name of " + attributesMapperClassName, ex);
+        }
+
+        if(!LogstashPluginAttributesMapper.class.isAssignableFrom(attributesMapperClass)) {
+            throw new LogstashMappingException("The provided mapping class does not implement " + LogstashPluginAttributesMapper.class);
+        }
+
+        try {
+            final Constructor<?> defaultConstructor = attributesMapperClass.getConstructor();
+            final Object instance = defaultConstructor.newInstance();
+            return  (LogstashPluginAttributesMapper) instance;
+        } catch (final Exception ex) {
+            throw new LogstashMappingException("Unable to create Mapper class with name of " + attributesMapperClassName, ex);
+        }
+    }
+}

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

@@ -0,0 +1,29 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+class AttributesMapperProvider {
+    private final AttributesMapperCreator attributesMapperCreator;
+    private final DefaultLogstashPluginAttributesMapper defaultLogstashPluginAttributesMapper;
+
+    AttributesMapperProvider() {
+        this(new AttributesMapperCreator());
+    }
+
+    AttributesMapperProvider(final AttributesMapperCreator attributesMapperCreator) {
+        this.attributesMapperCreator = attributesMapperCreator;
+        defaultLogstashPluginAttributesMapper = new DefaultLogstashPluginAttributesMapper();
+    }
+
+    LogstashPluginAttributesMapper getAttributesMapper(final LogstashMappingModel mappingModel) {
+        final String attributesMapperClassName = mappingModel.getAttributesMapperClass();
+        if(attributesMapperClassName == null) {
+            return defaultLogstashPluginAttributesMapper;
+        }
+
+        return attributesMapperCreator.createMapperClass(attributesMapperClassName);
+    }
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import org.opensearch.dataprepper.logstash.model.LogstashAttribute;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+class DefaultLogstashPluginAttributesMapper implements LogstashPluginAttributesMapper {
+    private static final Logger LOG = LoggerFactory.getLogger(DefaultLogstashPluginAttributesMapper.class);
+
+    @Override
+    public Map<String, Object> mapAttributes(final List<LogstashAttribute> logstashAttributes, final LogstashAttributesMappings logstashAttributesMappings) {
+        final Map<String, Object> pluginSettings = new LinkedHashMap<>(logstashAttributesMappings.getAdditionalAttributes());
+
+        logstashAttributes.forEach(logstashAttribute -> {
+            if (logstashAttributesMappings.getMappedAttributeNames().containsKey(logstashAttribute.getAttributeName())) {
+                pluginSettings.put(
+                        logstashAttributesMappings.getMappedAttributeNames().get(logstashAttribute.getAttributeName()),
+                        logstashAttribute.getAttributeValue().getValue()
+                );
+            }
+            else {
+                LOG.warn("Attribute name {} is not found in mapping file.", logstashAttribute.getAttributeName());
+            }
+        });
+
+        return pluginSettings;
+    }
+}

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

@@ -5,12 +5,9 @@
 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;
 
 /**
@@ -19,8 +16,16 @@
  * @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());
+    private final AttributesMapperProvider attributesMapperProvider;
+
+    public DefaultPluginMapper() {
+        this(new AttributesMapperProvider());
+    }
+
+    DefaultPluginMapper(final AttributesMapperProvider attributesMapperProvider) {
+        this.attributesMapperProvider = attributesMapperProvider;
+    }
 
     @Override
     public PluginModel mapPlugin(LogstashPlugin logstashPlugin) {
@@ -43,19 +48,10 @@ public PluginModel mapPlugin(LogstashPlugin logstashPlugin) {
         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());
-            }
-        });
+
+        final LogstashPluginAttributesMapper pluginAttributesMapper = attributesMapperProvider.getAttributesMapper(logstashMappingModel);
+
+        final Map<String, Object> pluginSettings = pluginAttributesMapper.mapAttributes(logstashPlugin.getAttributes(), logstashMappingModel);
 
         return new PluginModel(logstashMappingModel.getPluginName(), pluginSettings);
     }

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

@@ -0,0 +1,19 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import java.util.Map;
+
+/**
+ * Represents attribute mappings from Logstash into Data Prepper.
+ *
+ * @since 1.2
+ */
+public interface LogstashAttributesMappings {
+    Map<String, String> getMappedAttributeNames();
+
+    Map<String, Object> getAdditionalAttributes();
+}

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

@@ -1,31 +1,41 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
 package org.opensearch.dataprepper.logstash.mapping;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 import java.util.Map;
 
 /**
  * Mapping Model class for Logstash plugins and attributes
  *
  * @since 1.2
  */
-public class LogstashMappingModel {
+class LogstashMappingModel implements LogstashAttributesMappings {
 
+    @JsonProperty
     private String pluginName;
-    private Map<String, String> mappedAttributeNames;
-    private Map<String, Object> additionalAttributes;
 
-    public LogstashMappingModel(final String pluginName, final Map<String, String> mappedAttributeNames,
-                                final Map<String, Object> additionalAttributes) {
-        this.pluginName = pluginName;
-        this.mappedAttributeNames = mappedAttributeNames;
-        this.additionalAttributes = additionalAttributes;
-    }
+    @JsonProperty
+    private String attributesMapperClass;
 
-    public LogstashMappingModel() {}
+    @JsonProperty
+    private Map<String, String> mappedAttributeNames;
+
+    @JsonProperty
+    private Map<String, Object> additionalAttributes;
 
     public String getPluginName() {
         return pluginName;
     }
 
+    public String getAttributesMapperClass() {
+        return attributesMapperClass;
+    }
+
     public Map<String, String> getMappedAttributeNames() {
         return mappedAttributeNames;
     }

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

@@ -0,0 +1,29 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import org.opensearch.dataprepper.logstash.model.LogstashAttribute;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An interface for a class which is responsible for mapping attributes
+ * from a Logstash plugin into Data Prepper plugin settings.
+ *
+ * @since 1.2
+ */
+public interface LogstashPluginAttributesMapper {
+    /**
+     * Map all logstashAttributes from a Logstash plugin.
+     *
+     * @param logstashAttributes All the Logstash logstashAttributes for the plugin
+     * @param logstashAttributesMappings The mappings for this Logstash plugin
+     * @return A map of Data Prepper plugin settings.
+     * @since 1.2
+     */
+    Map<String, Object> mapAttributes(List<LogstashAttribute> logstashAttributes, LogstashAttributesMappings logstashAttributesMappings);
+}

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

@@ -0,0 +1,78 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.opensearch.dataprepper.logstash.exception.LogstashMappingException;
+import org.opensearch.dataprepper.logstash.model.LogstashAttribute;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class AttributesMapperCreatorTest {
+    static class DoesNotImplement {
+    }
+
+    public static class NoDefaultConstructor implements LogstashPluginAttributesMapper {
+        public NoDefaultConstructor(final String ignored) { }
+        @Override
+        public Map<String, Object> mapAttributes(List<LogstashAttribute> logstashAttributes, LogstashAttributesMappings logstashAttributesMappings) {
+            return null;
+        }
+    }
+
+    public static class ThrowingConstructor implements LogstashPluginAttributesMapper {
+        public ThrowingConstructor() {
+            throw new RuntimeException("Intentional exception for testing.");
+        }
+        @Override
+        public Map<String, Object> mapAttributes(List<LogstashAttribute> logstashAttributes, LogstashAttributesMappings logstashAttributesMappings) {
+            return null;
+        }
+    }
+
+    public static class ValidMapper implements LogstashPluginAttributesMapper {
+        @Override
+        public Map<String, Object> mapAttributes(final List<LogstashAttribute> logstashAttributes, final LogstashAttributesMappings logstashAttributesMappings) {
+            return null;
+        }
+    }
+
+    private AttributesMapperCreator createObjectUnderTest() {
+        return new AttributesMapperCreator();
+    }
+
+    @Test
+    void createMapperClass_should_throw_if_class_does_not_exist() {
+        final AttributesMapperCreator objectUnderTest = createObjectUnderTest();
+
+        assertThrows(LogstashMappingException.class,
+                () -> objectUnderTest.createMapperClass("org.opensearch.dataprepper.logstash.DoesNotExist"));
+    }
+
+    @ParameterizedTest
+    @ValueSource(classes = {
+            DoesNotImplement.class, NoDefaultConstructor.class, ThrowingConstructor.class
+    })
+    void createMapperClass_should_throw_for_classes_which_cannot_be_constructor(final Class<?> invalidClass) {
+        final AttributesMapperCreator objectUnderTest = createObjectUnderTest();
+
+        assertThrows(LogstashMappingException.class,
+                () -> objectUnderTest.createMapperClass(invalidClass.getName()));
+    }
+
+    @Test
+    void createMapperClass_returns_new_instance() {
+        assertThat(createObjectUnderTest().createMapperClass(ValidMapper.class.getName()),
+            instanceOf(LogstashPluginAttributesMapper.class));
+    }
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.logstash.mapping;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class AttributesMapperProviderTest {
+
+    private AttributesMapperCreator attributesMapperCreator;
+    private LogstashMappingModel logstashMappingModel;
+
+    @BeforeEach
+    void setUp() {
+        attributesMapperCreator = mock(AttributesMapperCreator.class);
+        logstashMappingModel = mock(LogstashMappingModel.class);
+    }
+
+    private AttributesMapperProvider createObjectUnderTest() {
+        return new AttributesMapperProvider(attributesMapperCreator);
+    }
+
+    @Test
+    void getAttributesMapper_should_return_Default_when_model_has_no_AttributesMapperClass() {
+        final LogstashPluginAttributesMapper attributesMapper = createObjectUnderTest().getAttributesMapper(logstashMappingModel);
+
+        assertThat(attributesMapper, notNullValue());
+        assertThat(attributesMapper, instanceOf(DefaultLogstashPluginAttributesMapper.class));
+    }
+
+    @Test
+    void getAttributesMapper_should_return_new_instance_for_AttributesMapperClass() {
+        final String className = UUID.randomUUID().toString();
+        when(logstashMappingModel.getAttributesMapperClass())
+                .thenReturn(className);
+        final LogstashPluginAttributesMapper expectedMapper = mock(LogstashPluginAttributesMapper.class);
+        when(attributesMapperCreator.createMapperClass(className))
+                .thenReturn(expectedMapper);
+
+        assertThat(createObjectUnderTest().getAttributesMapper(logstashMappingModel), equalTo(expectedMapper));
+    }
+}