Add geoip service extension

data-prepper-plugins/geoip-processor/build.gradle

@@ -26,6 +26,7 @@ dependencies {
     implementation 'com.maxmind.db:maxmind-db:3.0.0'
 
     implementation libs.commons.lang3
+    testImplementation project(':data-prepper-core')
     testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
     testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     testImplementation project(':data-prepper-test-common')

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

@@ -15,6 +15,8 @@
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.plugins.processor.configuration.KeysConfig;
 import org.opensearch.dataprepper.plugins.processor.databaseenrich.EnrichFailedException;
+import org.opensearch.dataprepper.plugins.processor.extension.GeoIpConfigSupplier;
+import org.opensearch.dataprepper.plugins.processor.extension.GeoIpServiceConfig;
 import org.opensearch.dataprepper.plugins.processor.utils.IPValidationcheck;
 import org.opensearch.dataprepper.logging.DataPrepperMarkers;
 import org.slf4j.Logger;
@@ -42,6 +44,7 @@ public class GeoIPProcessor extends AbstractProcessor<Record<Event>, Record<Even
   private final GeoIPProcessorConfig geoIPProcessorConfig;
   private final String tempPath;
   private final List<String> tagsOnSourceNotFoundFailure;
+  private final GeoIpServiceConfig geoIpServiceConfig;
   private GeoIPProcessorService geoIPProcessorService;
   private static final String TEMP_PATH_FOLDER = "GeoIP";
 
@@ -52,14 +55,17 @@ public class GeoIPProcessor extends AbstractProcessor<Record<Event>, Record<Even
    */
   @DataPrepperPluginConstructor
   public GeoIPProcessor(PluginSetting pluginSetting,
-                        final GeoIPProcessorConfig geoCodingProcessorConfig) {
+                        final GeoIPProcessorConfig geoCodingProcessorConfig,
+                        final GeoIpConfigSupplier geoIpConfigSupplier) {
     super(pluginSetting);
     this.geoIPProcessorConfig = geoCodingProcessorConfig;
     this.tempPath = System.getProperty("java.io.tmpdir")+ File.separator + TEMP_PATH_FOLDER;
     geoIPProcessorService = new GeoIPProcessorService(geoCodingProcessorConfig,tempPath);
     tagsOnSourceNotFoundFailure = geoCodingProcessorConfig.getTagsOnSourceNotFoundFailure();
     this.geoIpProcessingMatchCounter = pluginMetrics.counter(GEO_IP_PROCESSING_MATCH);
     this.geoIpProcessingMismatchCounter = pluginMetrics.counter(GEO_IP_PROCESSING_MISMATCH);
+    // TODO: use this config and clean up MaxMind service config from pipeline.yaml
+    geoIpServiceConfig = geoIpConfigSupplier.getGeoIpServiceConfig();
   }
 
   /**

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/DefaultGeoIpConfigSupplier.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+public class DefaultGeoIpConfigSupplier implements GeoIpConfigSupplier {
+    private final GeoIpServiceConfig geoIpServiceConfig;
+
+    public DefaultGeoIpConfigSupplier(final GeoIpServiceConfig geoIpServiceConfig) {
+        this.geoIpServiceConfig = geoIpServiceConfig;
+    }
+
+    @Override
+    public GeoIpServiceConfig getGeoIpServiceConfig() {
+        return this.geoIpServiceConfig;
+    }
+}

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/GeoIpConfigExtension.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import org.opensearch.dataprepper.model.annotations.DataPrepperExtensionPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.plugin.ExtensionPlugin;
+import org.opensearch.dataprepper.model.plugin.ExtensionPoints;
+
+@DataPrepperExtensionPlugin(modelType = GeoIpServiceConfig.class, rootKeyJsonPath = "/geoip_service")
+public class GeoIpConfigExtension implements ExtensionPlugin {
+    private final DefaultGeoIpConfigSupplier defaultGeoIpConfigSupplier;
+
+    @DataPrepperPluginConstructor
+    public GeoIpConfigExtension(final GeoIpServiceConfig geoIpServiceConfig) {
+        this.defaultGeoIpConfigSupplier = new DefaultGeoIpConfigSupplier(geoIpServiceConfig != null ? geoIpServiceConfig : new GeoIpServiceConfig());
+    }
+
+    @Override
+    public void apply(final ExtensionPoints extensionPoints) {
+        extensionPoints.addExtensionProvider(new GeoIpConfigProvider(this.defaultGeoIpConfigSupplier));
+    }
+}

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/GeoIpConfigProvider.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import org.opensearch.dataprepper.model.plugin.ExtensionProvider;
+
+import java.util.Optional;
+
+public class GeoIpConfigProvider implements ExtensionProvider<GeoIpConfigSupplier> {
+    private final GeoIpConfigSupplier geoIpConfigSupplier;
+
+    public GeoIpConfigProvider(final GeoIpConfigSupplier geoIpConfigSupplier) {
+        this.geoIpConfigSupplier = geoIpConfigSupplier;
+    }
+
+    @Override
+    public Optional<GeoIpConfigSupplier> provideInstance(Context context) {
+        return Optional.of(this.geoIpConfigSupplier);
+    }
+
+    @Override
+    public Class<GeoIpConfigSupplier> supportedClass() {
+        return GeoIpConfigSupplier.class;
+    }
+}

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/GeoIpConfigSupplier.java

@@ -0,0 +1,10 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+public interface GeoIpConfigSupplier {
+    GeoIpServiceConfig getGeoIpServiceConfig();
+}

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/GeoIpServiceConfig.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.Valid;
+
+public class GeoIpServiceConfig {
+    private static final MaxMindConfig DEFAULT_MAXMIND_CONFIG = new MaxMindConfig();
+
+    public GeoIpServiceConfig() {
+        // This default constructor is used if geoip_service is not configured
+    }
+
+    @JsonProperty("maxmind")
+    @Valid
+    private MaxMindConfig maxMindConfig = DEFAULT_MAXMIND_CONFIG;
+
+    /**
+     * Gets the configuration for MaxMind.
+     *
+     * @return The MaxMind configuration
+     * @since 2.7
+     */
+    public MaxMindConfig getMaxMindConfig() {
+        return maxMindConfig;
+    }
+}

data-prepper-plugins/geoip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/MaxMindConfig.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.Min;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MaxMindConfig {
+    //TODO: Add validations to database paths
+    //TODO: Make default path to be a public CDN endpoint
+    private static final List<String> DEFAULT_DATABASE_PATHS = new ArrayList<>();
+    private static final Duration DEFAULT_DATABASE_REFRESH_INTERVAL = Duration.ofDays(7);
+    private static final int DEFAULT_CACHE_SIZE = 4096;
+
+    @JsonProperty("database_paths")
+    private List<String> databasePaths = DEFAULT_DATABASE_PATHS;
+    @JsonProperty("database_refresh_interval")
+    private Duration databaseRefreshInterval = DEFAULT_DATABASE_REFRESH_INTERVAL;
+    @JsonProperty("cache_size")
+    @Min(1)
+    //TODO:  Add a Max limit on cache size
+    private int cacheSize = DEFAULT_CACHE_SIZE;
+    //TODO: Add a destination path to store database files
+
+    public MaxMindConfig() {
+        // This default constructor is used if maxmind is not configured
+    }
+
+    @AssertTrue(message = "database_refresh_interval should be between 1 and 30 days.")
+    public boolean isValidDatabaseRefreshInterval() {
+        return databaseRefreshInterval.toDays() >= 1 && databaseRefreshInterval.toDays() <= 30;
+    }
+
+    /**
+     * Gets the MaxMind database paths
+     *
+     * @return The MaxMind database paths
+     * @since 2.7
+     */
+    public List<String> getDatabasePaths() {
+        return databasePaths;
+    }
+
+    /**
+     * Gets the database refresh duration. This loads the database from the paths into memory again in case if there are any updates.
+     *
+     * @return The refresh duration
+     * @since 2.7
+     */
+    public Duration getDatabaseRefreshInterval() {
+        return databaseRefreshInterval;
+    }
+
+    /**
+     * Gets the cache size used in CHM cache for MaxMind DatabaseReader.
+     *
+     * @return The cache size
+     * @since 2.7
+     */
+    public int getCacheSize() {
+        return cacheSize;
+    }
+}

data-prepper-plugins/geoip-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessorTest.java

@@ -22,6 +22,7 @@
 import org.opensearch.dataprepper.plugins.processor.configuration.MaxMindServiceConfig;
 import org.opensearch.dataprepper.plugins.processor.configuration.ServiceTypeOptions;
 import org.opensearch.dataprepper.plugins.processor.databaseenrich.EnrichFailedException;
+import org.opensearch.dataprepper.plugins.processor.extension.GeoIpConfigSupplier;
 import org.opensearch.dataprepper.plugins.processor.loadtype.LoadTypeOptions;
 import org.opensearch.dataprepper.test.helper.ReflectivelySetField;
 
@@ -40,7 +41,6 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
@@ -56,6 +56,8 @@ class GeoIPProcessorTest {
     @Mock
     private GeoIPProcessorConfig geoCodingProcessorConfig;
     @Mock
+    private GeoIpConfigSupplier geoIpConfigSupplier;
+    @Mock
     private PluginSetting pluginSetting;
     @Mock
     private ServiceTypeOptions serviceTypeOptions;
@@ -164,7 +166,7 @@ private Map<String, Object> prepareGeoData() {
     }
 
     private GeoIPProcessor createObjectUnderTest() throws MalformedURLException {
-        return new GeoIPProcessor(pluginSetting, geoCodingProcessorConfig);
+        return new GeoIPProcessor(pluginSetting, geoCodingProcessorConfig, geoIpConfigSupplier);
     }
 
     private List<String> setAttributes() {

data-prepper-plugins/geoip-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/extension/DefaultGeoIpConfigSupplierTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class DefaultGeoIpConfigSupplierTest {
+    @Mock
+    private GeoIpServiceConfig geoIpServiceConfig;
+
+    private DefaultGeoIpConfigSupplier createObjectUnderTest() {
+        return new DefaultGeoIpConfigSupplier(geoIpServiceConfig);
+    }
+
+    @Test
+    void testGetGeoIpServiceConfig() {
+        assertThat(createObjectUnderTest().getGeoIpServiceConfig(), equalTo(geoIpServiceConfig));
+    }
+
+}

data-prepper-plugins/geoip-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/extension/GeoIpConfigExtensionTest.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright OpenSearch Contributors
+ *  PDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.extension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.model.plugin.ExtensionPoints;
+import org.opensearch.dataprepper.model.plugin.ExtensionProvider;
+
+import java.time.Duration;
+
+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.verify;
+
+@ExtendWith(MockitoExtension.class)
+class GeoIpConfigExtensionTest {
+    @Mock
+    private ExtensionPoints extensionPoints;
+
+    @Mock
+    private GeoIpServiceConfig geoIpServiceConfig;
+
+    @Mock
+    private ExtensionProvider.Context context;
+
+    private GeoIpConfigExtension createObjectUnderTest() {
+        return new GeoIpConfigExtension(geoIpServiceConfig);
+    }
+
+    @Test
+    void apply_should_addExtensionProvider() {
+        createObjectUnderTest().apply(extensionPoints);
+        final ArgumentCaptor<ExtensionProvider> extensionProviderArgumentCaptor =
+                ArgumentCaptor.forClass(ExtensionProvider.class);
+
+        verify(extensionPoints).addExtensionProvider(extensionProviderArgumentCaptor.capture());
+
+        final ExtensionProvider actualExtensionProvider = extensionProviderArgumentCaptor.getValue();
+
+        assertThat(actualExtensionProvider, instanceOf(GeoIpConfigProvider.class));
+    }
+
+    @Test
+    void apply_should_addExtensionProvider_and_supplier_should_use_default_config_if_not_configured() {
+        final GeoIpConfigExtension geoIpConfigExtension = new GeoIpConfigExtension(null);
+
+        geoIpConfigExtension.apply(extensionPoints);
+        final ArgumentCaptor<ExtensionProvider> extensionProviderArgumentCaptor =
+                ArgumentCaptor.forClass(ExtensionProvider.class);
+
+        verify(extensionPoints).addExtensionProvider(extensionProviderArgumentCaptor.capture());
+
+        final ExtensionProvider actualExtensionProvider = extensionProviderArgumentCaptor.getValue();
+
+        assertThat(actualExtensionProvider, instanceOf(GeoIpConfigProvider.class));
+
+        final GeoIpConfigSupplier geoIpConfigSupplier = (GeoIpConfigSupplier) actualExtensionProvider.provideInstance(context).get();
+
+        final GeoIpServiceConfig geoIpServiceConfig = geoIpConfigSupplier.getGeoIpServiceConfig();
+        assertThat(geoIpServiceConfig, notNullValue());
+
+        final MaxMindConfig maxMindConfig = geoIpServiceConfig.getMaxMindConfig();
+        assertThat(maxMindConfig, notNullValue());
+        assertThat(maxMindConfig.getCacheSize(), equalTo(4096));
+        assertThat(maxMindConfig.getDatabasePaths().size(), equalTo(0));
+        assertThat(maxMindConfig.getDatabaseRefreshInterval(), equalTo(Duration.ofDays(7)));
+    }
+
+}