Geoip database update implementation #4105

asifsmohammed · 2024-02-11T20:55:28Z

Description

This PR adds implementation for geoip processor.

Refactoring old processor logic
CDN download support
Database update support
Missing Unit tests (Will add more unit tests to improve test coverage)

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Asif Sohail Mohammed <[email protected]>

chenqi0805 · 2024-02-12T18:45:36Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java


      for (final EntryConfig entry : geoIPProcessorConfig.getEntries()) {
        final String source = entry.getSource();
-        final List<String> attributes = entry.getFields();
+        final List<GeoIPField> fields = entry.getFields();
+        final Set<GeoIPDatabase> databases = entry.getGeoIPDatabases();
        final String ipAddress = event.get(source, String.class);

        //Lookup from DB


Could be in a separate method for readability

This is just a getter in EntryConfig. It will be calculated only once and subsequent call will just return the object.

Not a blocker but would still be more readable to wrap the L131-151 into a private method.

chenqi0805 · 2024-02-12T19:00:04Z

...n/java/org/opensearch/dataprepper/plugins/processor/extension/databasedownload/DBSource.java

-    public String tarFolderPath = tempFolderPath + "/tar";
-    public String downloadTarFilepath = tarFolderPath + "/out.tar.gz";
-    public void initiateDownload(List<String> config) throws Exception;
+    Logger LOG = LoggerFactory.getLogger(DBSource.class);


Does this need to be at interface level?

I will try to refactor this. There is a LOG in line 44, I will try to remove this.

Loggers in interfaces is probably unnecessary.

dlvenable

Thanks @asifsmohammed . This is making a lot of improvements! I made a number of high-level comments.

dlvenable · 2024-02-13T03:05:27Z

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

+    def gradleName = enterpriseDatabaseName.replaceAll('-', '')
+    def downloadEnterpriseTask = tasks.register("download${gradleName}", Download) {
+        src(url)
+        dest "build/resources/test/mmdb-files/geo-ip2/" + enterpriseDatabaseName + ".mmdb"


You can improve this through Groovy's GStrings:

dest "build/resources/test/mmdb-files/geo-ip2/${enterpriseDatabaseName}.mmdb"

dlvenable · 2024-02-13T03:05:35Z

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

 }

 jacocoTestCoverageVerification {
    dependsOn jacocoTestReport
    violationRules {
        rule {
            limit {
-                minimum = 0.1 // temporarily reduce coverage for the builds to pass
+                minimum = 0.8 // temporarily reduce coverage for the builds to pass


dlvenable · 2024-02-13T03:08:28Z

...or/src/main/java/org/opensearch/dataprepper/plugins/processor/configuration/EntryConfig.java

+    private List<String> includeFields = new ArrayList<>();
+
+    @JsonProperty("exclude_fields")
+    private List<String> excludeFields = new ArrayList<>();


You should be able to use @JsonCreator in the enum class to make this serializable. Then you can change this to: private List<String> excludeFields.

I didn't want to deserialize to enum directly because processor doesn't have control of Object Mapper to ignore invalid enum option. It throws an exception. So I kept it as List without any default value. If it's empty no fields will be excluded and all fields are added.

dlvenable · 2024-02-13T03:09:27Z

...or/src/main/java/org/opensearch/dataprepper/plugins/processor/configuration/EntryConfig.java

+        return geoIPFields;
+    }
+
+    public Set<GeoIPDatabase> getGeoIPDatabases() {


I think this should be moved to a higher-level class that determines the databases from the fields. The database list is not directly linked to the configuration.

I agree, I had it here because it simplifies getting this list for each entry, instead of calculation it in the processor. But I can move it to processor initialization.

I'd recommend a specific class for database determination. It will have a specific set of concerns and be more testable.

dlvenable · 2024-02-13T03:10:26Z

...or/src/main/java/org/opensearch/dataprepper/plugins/processor/configuration/EntryConfig.java

-    public List<String> getFields() {
-        return fields;
+
+    public List<GeoIPField> getFields() {


I think you could consider moving this to an abstraction class. But, I don't feel as strongly about that as with the databases. These are more closely related to the configuration than the databases.

dlvenable · 2024-02-13T03:20:25Z

...or/src/main/java/org/opensearch/dataprepper/plugins/processor/configuration/EntryConfig.java

    @JsonProperty("source")
    @NotEmpty
    private String source;

    @JsonProperty("target")
    private String target = DEFAULT_TARGET;

-    @JsonProperty("fields")
-    private List<String> fields;
+    @JsonProperty("include_fields")


I tend to think we should move these up a level. Part of the reason to combine the processing in a single processor is to simplify the configurations. If we move it to the top, then users only have to configure once.

Let's discuss it. I think if we are moving this there is no use of entries, and user can configure multiple instances of geoip processor as we are only downloading data once.

dlvenable · 2024-02-13T03:22:48Z

...pensearch/dataprepper/plugins/processor/exception/DatabaseReaderInitializationException.java

+
+package org.opensearch.dataprepper.plugins.processor.exception;
+
+public class DatabaseReaderInitializationException extends RuntimeException {


Per my comments about grouping errors for the purpose of tagging (and I think we can extend that to metrics), let's create a few base exceptions that represent the important main groups.

e.g.

public class EngineFailureException extends RuntimeException

public class DatabaseReaderInitializationException extends EngineFailureException

This comment applies to all the exception types.

dlvenable · 2024-02-13T03:25:32Z

...org/opensearch/dataprepper/plugins/processor/extension/databasedownload/DBSourceOptions.java

@@ -17,7 +17,8 @@
 public enum DBSourceOptions {
    PATH("path"),
    URL("url"),
-    S3("s3");
+    S3("s3"),
+    CDN("cdn");


Generically for the user, this should be configured as an HTTP endpoint.

I think you can make this:

HTTP("http")

The fact that the implementation is a CDN is not relevant for the user. That is a detail of the server itself, not the client (ie. Data Prepper).

dlvenable · 2024-02-13T03:26:02Z

...n/java/org/opensearch/dataprepper/plugins/processor/extension/databasedownload/DBSource.java

-    public String tarFolderPath = tempFolderPath + "/tar";
-    public String downloadTarFilepath = tarFolderPath + "/out.tar.gz";
-    public void initiateDownload(List<String> config) throws Exception;
+    Logger LOG = LoggerFactory.getLogger(DBSource.class);


Loggers in interfaces is probably unnecessary.

dlvenable · 2024-02-13T03:26:28Z

.../opensearch/dataprepper/plugins/processor/extension/databasedownload/CDNDownloadService.java

+import static org.opensearch.dataprepper.plugins.processor.databaseenrich.GeoIPDatabaseReader.MAXMIND_DATABASE_EXTENSION;
+
+
+public class CDNDownloadService implements DBSource {


Make this HttpDownloadService.

Also, please consider how we can make the project support two different scenarios:

An HTTP endpoint using the manifest (what you have now)

An HTTP endpoint without a manifest. Say a user just wants to host these files in their own HTTP server.

Signed-off-by: Asif Sohail Mohammed <[email protected]>

graytaylor0 · 2024-02-14T21:10:01Z

...rationTest/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessorUrlServiceIT.java

@@ -93,7 +93,7 @@ void verify_enrichment_of_data_from_maxmind_url() throws UnknownHostException {
        if (IPValidationCheck.isPublicIpAddress(ipAddress)) {
            InetAddress inetAddress = InetAddress.getByName(ipAddress);
            //All attributes are considered by default with the null value
-            geoData = geoIPProcessorService.getGeoData(inetAddress, null);
+//            geoData = geoIPProcessorService.getGeoData(inetAddress, null);


Is this meant to be commented out?

I haven't fixed these integ tests yet, I need to change this because the constructor changed.

graytaylor0 · 2024-02-14T21:12:39Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

      final String whenCondition = geoIPProcessorConfig.getWhenCondition();

      if (whenCondition != null && !expressionEvaluator.evaluateConditional(whenCondition, event)) {
        continue;
      }
+      geoIpEventsProcessed.increment();
+
+      // TODO: Need to decide the behaviour, right now if all databases are expired we don't enrich the data.


I think just having this processor be a no-op on Events is ok when the database is expired, but we should be sure to tag these Events so that they can be routed appropriately in downstream sink and processors. I would say we could even default the tag to be _expired_geoip_database, and could make it configurable with tags_on_expired_database or something along those lines, given that this is one of the 2 special cases of errors

Yeah there will be tags added if database expired. The default tags are not being added right now and we will consider a mechanism on how we can add tags in plugins and get back to it. Right now it will be configurable

graytaylor0 · 2024-02-14T21:15:56Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

+        final Set<GeoIPDatabase> databases = entryDatabaseMap.get(entry);
+        String ipAddress = null;
+        try {
+          ipAddress = event.get(source, String.class);


This is just a placeholder for future implementation?

No, but as you mentioned above I will try to put all doExecute logic in try/catch in upcoming PR's

graytaylor0 · 2024-02-14T21:23:14Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

+    final GeoIPDatabaseReader geoIPDatabaseReader = geoIPProcessorService.getGeoIPDatabaseReader();
+    final boolean databasesExpired = geoIPDatabaseReader.isExpired();
+


Let's wrap the entirety of the doExecute block in a generic try/catch that will ensure geoip never throws a runtime exception. We should also have a try/catch in the looping of Events, where we can catch any exceptions for one Event, log them, and then continue to the next Event.

graytaylor0 · 2024-02-14T21:23:48Z

settings.gradle

@@ -117,7 +117,7 @@ include 'data-prepper-expression'
 include 'data-prepper-plugins:mutate-string-processors'
 include 'data-prepper-plugins:s3-source'
 include 'data-prepper-plugins:s3-sink'
-include 'data-prepper-plugins:rss-source'
+//include 'data-prepper-plugins:rss-source'


Any reason to remove this in this PR?

I should not have removed it, but the build has failed multiple times in RSS. I will add it back

chenqi0805 · 2024-02-16T23:35:15Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java


      for (final EntryConfig entry : geoIPProcessorConfig.getEntries()) {
        final String source = entry.getSource();
-        final List<String> attributes = entry.getFields();
+        final List<GeoIPField> fields = entry.getFields();
+        final Set<GeoIPDatabase> databases = entry.getGeoIPDatabases();
        final String ipAddress = event.get(source, String.class);

        //Lookup from DB


Not a blocker but would still be more readable to wrap the L131-151 into a private method.

dlvenable · 2024-02-19T16:10:58Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

@@ -119,21 +160,65 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
        event.getMetadata().addTags(tagsOnFailure);
      }
    }
+    geoIPDatabaseReader.close();


This could be missed if you have an exception that isn't caught. I recommend you use try-with-resources.

try(final GeoIPDatabaseReader geoIPDatabaseReader = geoIPProcessorService.getGeoIPDatabaseReader()) { ... }

The try block will call .close(), so you won't even need this line.

I will address this in a follow-on PR.

dlvenable · 2024-02-19T16:13:12Z

...n/java/org/opensearch/dataprepper/plugins/processor/databaseenrich/GeoIP2DatabaseReader.java

+    public void close() {
+        final int count = closeCount.decrementAndGet();
+        if (count == 0) {
+            LOG.info("Closing old geoip database readers");


Maybe this should be the DEBUG level.

dlvenable · 2024-02-19T16:16:51Z

...n/java/org/opensearch/dataprepper/plugins/processor/databaseenrich/GeoIP2DatabaseReader.java

+
+    @Override
+    public void close() {
+        final int count = closeCount.decrementAndGet();


This counting logic is common between both the GeoIp Lite and Enterprise DatabaseReader implementations. I'd consider a decorator which handles the counting.

AutoCountingDatabaseReader implements DatabaseReader, AutoCloseable { private final DatabaseReader delegateDatabaseReader; // pass in through constructor @Override public Map<String, Object> getGeoData(final InetAddress inetAddress, final List<GeoIPField> fields, final Set<GeoIPDatabase> geoIPDatabases) { return delegateDatabaseReader.getGeoData(inetAddress, fields, geoIpDatabases); } @Override public void close() { final int count = closeCount.decrementAndGet(); if (count == 0) { LOG.info("Closing old geoip database readers"); delegateReader.close(); } }

This could come as a follow-on PR.

I will address this in a follow-on PR.

dlvenable · 2024-02-19T16:17:46Z

...n/java/org/opensearch/dataprepper/plugins/processor/databaseenrich/GeoIP2DatabaseReader.java

+    }
+
+    @Override
+    public void retain() {


Consider finding a way to automatically retain - perhaps on creation. This is not necessary and could be a follow-on PR.

dlvenable · 2024-02-19T16:19:25Z

.../java/org/opensearch/dataprepper/plugins/processor/extension/DefaultGeoIpConfigSupplier.java

    }

    @Override
    public GeoIPProcessorService getGeoIPProcessorService() {
-        return new GeoIPProcessorService(geoIpServiceConfig);
+        if (geoIpServiceConfig != null)


Consider making this an Optional. Then the code that calls it can look somewhat like:

this.service = supplier.getGeoIpProcessorService().orElseThrow(() -> ... existing exception);

dlvenable · 2024-02-19T16:21:21Z

...ssor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/MaxMindConfig.java

@@ -57,6 +68,19 @@ boolean isAwsAuthenticationOptionsRequired() {
        return true;
    }

+    @AssertTrue(message = "database_paths should be https endpoint if using URL and if insecure is set to false")
+    public boolean isSecureEndpoint() throws URISyntaxException {


Please consider renaming this method. The logic below looks incorrect based on the name. Perhaps isHttpsEndpointOrInsecure?

dlvenable · 2024-02-19T16:21:46Z

...ssor/src/main/java/org/opensearch/dataprepper/plugins/processor/extension/MaxMindConfig.java

+     * @since 2.7
+     */
+    public String getDatabaseDestination() {
+        return databaseDestination + File.separator + File.separator + "geoip";


Why do you have two File.separators appended?

dlvenable · 2024-02-19T16:27:30Z

...search/dataprepper/plugins/processor/extension/databasedownload/ManifestDownloadService.java

+        try {
+            final URL url = new URL(CDNEndpoint);
+            httpURLConnection = (HttpURLConnection) url.openConnection();
+            httpURLConnection.addRequestProperty("User-Agent", "Custom-User-Agent");


Why do we add these? Should they be configurable?

Signed-off-by: Asif Sohail Mohammed <[email protected]>

dlvenable · 2024-02-19T20:02:50Z

...oip-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/GeoIPProcessor.java

@@ -69,8 +68,10 @@ public GeoIPProcessor(final PluginMetrics pluginMetrics,
                        final GeoIpConfigSupplier geoIpConfigSupplier,
                        final ExpressionEvaluator expressionEvaluator) {
    super(pluginMetrics);
-    this.geoIPProcessorService = geoIpConfigSupplier.getGeoIPProcessorService();
-    Objects.requireNonNull(geoIPProcessorService, "geoip_service configuration is required when using geoip processor.");
+    if (geoIpConfigSupplier.getGeoIPProcessorService().isEmpty()) {


You can simplify this code:

this.geoIPProcessorService = geoIpConfigSupplier.getGeoIPProcessorService().orElseThrow(() -> new IlegalStateException("geoip_service configuration is required when using geoip processor."));

Perhaps do that in the follow-up PRs.

asifsmohammed requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, dlvenable and oeyh as code owners February 11, 2024 20:55

asifsmohammed added 9 commits February 12, 2024 10:34

Add when condition to each entry

8cdb559

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Addressed feedback by updating metric names

d2b991f

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Updated metric logic and Added new test

063505c

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Reverted when condition to processor level

6c2d2a7

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Update database updater logic and added some boilerplate code

fcfb3bd

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Geoip database readers update

3b6dadf

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Geoip processor implementation

9d5ae91

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Add tests to CDN download service

2064cca

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Rebased from main

5bc1419

Signed-off-by: Asif Sohail Mohammed <[email protected]>

asifsmohammed force-pushed the geoip-database-update branch from 5c864af to 5bc1419 Compare February 12, 2024 17:01

chenqi0805 reviewed Feb 12, 2024

View reviewed changes

dlvenable requested changes Feb 13, 2024

View reviewed changes

asifsmohammed added 2 commits February 12, 2024 21:29

Fixed geoip filed to database relation

9776fa2

Signed-off-by: Asif Sohail Mohammed <[email protected]>

Addressed feedback

0cdbdad

Signed-off-by: Asif Sohail Mohammed <[email protected]>

graytaylor0 reviewed Feb 14, 2024

View reviewed changes

chenqi0805 previously approved these changes Feb 16, 2024

View reviewed changes

dlvenable requested changes Feb 19, 2024

View reviewed changes

Addressed feedback

3493565

Signed-off-by: Asif Sohail Mohammed <[email protected]>

asifsmohammed dismissed chenqi0805’s stale review via 3493565 February 19, 2024 19:22

dlvenable approved these changes Feb 19, 2024

View reviewed changes

chenqi0805 approved these changes Feb 19, 2024

View reviewed changes

asifsmohammed merged commit bb494de into opensearch-project:main Feb 21, 2024
46 of 47 checks passed


		package org.opensearch.dataprepper.plugins.processor.exception;

		public class DatabaseReaderInitializationException extends RuntimeException {

		import static org.opensearch.dataprepper.plugins.processor.databaseenrich.GeoIPDatabaseReader.MAXMIND_DATABASE_EXTENSION;


		public class CDNDownloadService implements DBSource {

		final GeoIPDatabaseReader geoIPDatabaseReader = geoIPProcessorService.getGeoIPDatabaseReader();
		final boolean databasesExpired = geoIPDatabaseReader.isExpired();

Geoip database update implementation #4105

Geoip database update implementation #4105

Conversation

asifsmohammed commented Feb 11, 2024 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asifsmohammed commented Feb 11, 2024 •

edited

Loading