Use logstash-bridge GeoIP interfaces.

Author: mashhurs

build.gradle

@@ -221,7 +221,7 @@ task buildElasticsearchLocalDistro(dependsOn: unzipDownloadedElasticsearchSource
         errorOutput = funneler.funnelInstance
     }
 
-    def esSource = "${buildDir}/elasticsearch-source/"
+    def esSource = "/Users/mashhur/Dev/elastic/elasticsearch/"
     def esBuildDir = "${esSource}/build"
 
     inputs.dir esSource
@@ -261,6 +261,7 @@ task buildElasticsearchLogstashBridge(type: Exec) {
     description "builds logstash-bridge lib module"
 
     dependsOn buildElasticsearchLocalDistro
+    mustRunAfter buildElasticsearchLocalDistro
 
     def logFile = project.file("${buildDir}/logstash-bridge-build.log")
     doFirst {
@@ -269,7 +270,7 @@ task buildElasticsearchLogstashBridge(type: Exec) {
         errorOutput = funneler.funnelInstance
     }
 
-    def esSource = "${buildDir}/elasticsearch-source/"
+    def esSource = "/Users/mashhur/Dev/elastic/elasticsearch/"
     def esBuildDir = "${esSource}/build"
 
     inputs.dir esSource
@@ -358,14 +359,14 @@ task shadeElasticsearchLogstashBridge(type: com.github.jengelman.gradle.plugins.
     description "Shades the Elasticsearch logstash-bridge jar"
 
     dependsOn buildElasticsearchLogstashBridge
-
-    from("${buildDir}/elasticsearch-source/libs/logstash-bridge/build/distributions") {
+    
+    from("/Users/mashhur/Dev/elastic/elasticsearch/libs/logstash-bridge/build/distributions") {
         include "elasticsearch-logstash-bridge-*.jar"
     }
-
+    
     archiveFileName = "elasticsearch-logstash-bridge-shaded.jar"
     destinationDirectory = file("${buildDir}/shaded")
-
+    
     exclude '**/module-info.class'
 }
 

gradle.properties

@@ -1,3 +1,3 @@
 LOGSTASH_PATH=../../logstash
 ELASTICSEARCH_REPO=mashhurs/elasticsearch
-ELASTICSEARCH_TREEISH=move-to-bridge-stable-api-investigation
+ELASTICSEARCH_TREEISH=logstash-bridge-geoip-interfaces

src/main/java/co/elastic/logstash/filters/elasticintegration/geoip/ConstantIpDatabaseHolder.java

@@ -1,7 +1,5 @@
 package co.elastic.logstash.filters.elasticintegration.geoip;
 
-import org.elasticsearch.ingest.geoip.IpDatabase;
-
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Objects;

src/main/java/co/elastic/logstash/filters/elasticintegration/geoip/GeoIpProcessorFactory.java

@@ -6,8 +6,8 @@
  */
 package co.elastic.logstash.filters.elasticintegration.geoip;
 
-import org.elasticsearch.ingest.geoip.GeoIpProcessor;
 import org.elasticsearch.logstashbridge.ingest.ProcessorBridge;
+import org.elasticsearch.logstashbridge.geoip.GeoIpProcessorBridge;
 
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -24,7 +24,8 @@ public ProcessorBridge create(Map<String, ProcessorBridge.Factory> processorFact
                             String tag,
                             String description,
                             Map<String, Object> config) throws Exception {
-        return ProcessorBridge.fromInternal(new GeoIpProcessor.Factory("geoip", this.ipDatabaseProvider)
+        return ProcessorBridge.fromInternal(new GeoIpProcessorBridge.Factory("geoip", this.ipDatabaseProvider.toInternal())
+                .toInternal()
                 .create(processorFactories.entrySet()
                                 .stream()
                                 .collect(Collectors.toMap(Map.Entry::getKey,e -> e.getValue().toInternal())),

src/main/java/co/elastic/logstash/filters/elasticintegration/geoip/IpDatabaseAdapter.java

@@ -8,53 +8,32 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.ExceptionsHelper;
-import org.elasticsearch.common.CheckedBiFunction;
-import org.elasticsearch.ingest.geoip.IpDatabase;
-import org.elasticsearch.ingest.geoip.shaded.com.maxmind.db.CHMCache;
-import org.elasticsearch.ingest.geoip.shaded.com.maxmind.db.NoCache;
-import org.elasticsearch.ingest.geoip.shaded.com.maxmind.db.NodeCache;
-import org.elasticsearch.ingest.geoip.shaded.com.maxmind.db.Reader;
+import org.elasticsearch.logstashbridge.geoip.IpDatabaseBridge;
+import org.elasticsearch.logstashbridge.geoip.MaxMindDbBridge;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Optional;
 
-public class IpDatabaseAdapter implements IpDatabase {
+public class IpDatabaseAdapter implements IpDatabaseBridge {
     private static final Logger LOGGER = LogManager.getLogger(IpDatabaseAdapter.class);
 
-    private final Reader databaseReader;
+    private final MaxMindDbBridge.Reader databaseReader;
     private final String databaseType;
 
     private volatile boolean isReaderClosed = false;
 
-    public IpDatabaseAdapter(final Reader databaseReader) {
+    public IpDatabaseAdapter(final MaxMindDbBridge.Reader databaseReader) {
         this.databaseReader = databaseReader;
-        this.databaseType = databaseReader.getMetadata().getDatabaseType();
+        this.databaseType = databaseReader.getDatabaseType();
     }
 
     @Override
     public String getDatabaseType() {
         return this.databaseType;
     }
 
-    @Override
-    public <RESPONSE> RESPONSE getResponse(String ipAddress, CheckedBiFunction<Reader, String, RESPONSE, Exception> responseProvider) {
-        try {
-            return responseProvider.apply(this.databaseReader, ipAddress);
-        } catch (Exception e) {
-            throw ExceptionsHelper.convertToRuntime(e);
-        }
-    }
-
-    @Override
-    public void close() throws IOException {
-        // no-op
-        // IpDatabase is an AutoCloseable class which is not our intention to close after try-with-resource operations.
-        // use closeReader() instead
-    }
-
     public void closeReader() throws IOException {
         LOGGER.debug("Closing the database adapter");
         this.databaseReader.close();
@@ -67,25 +46,25 @@ boolean isReaderClosed() {
     }
 
     public static IpDatabaseAdapter defaultForPath(final Path database) throws IOException {
-        return new Builder(database.toFile()).setCache(new CHMCache(10_000)).build();
+        return new Builder(database.toFile()).setCache(MaxMindDbBridge.NodeCache.get(10_000)).build();
     }
 
     public static class Builder {
         private File databasePath;
-        private NodeCache nodeCache;
+        private MaxMindDbBridge.NodeCache nodeCache;
 
-        public Builder(File databasePath) {
+        public Builder(final File databasePath) {
             this.databasePath = databasePath;
         }
 
-        public Builder setCache(final NodeCache nodeCache) {
+        public Builder setCache(final MaxMindDbBridge.NodeCache nodeCache) {
             this.nodeCache = nodeCache;
             return this;
         }
 
         public IpDatabaseAdapter build() throws IOException {
-            final NodeCache nodeCache = Optional.ofNullable(this.nodeCache).orElseGet(NoCache::getInstance);
-            final Reader databaseReader = new Reader(this.databasePath, nodeCache);
+            final MaxMindDbBridge.NodeCache nodeCache = Optional.ofNullable(this.nodeCache).orElseGet(MaxMindDbBridge.NodeCache::getInstance);
+            final MaxMindDbBridge.Reader databaseReader = new MaxMindDbBridge.Reader(this.databasePath, nodeCache);
             return new IpDatabaseAdapter(databaseReader);
         }
     }

src/main/java/co/elastic/logstash/filters/elasticintegration/geoip/IpDatabaseHolder.java

@@ -1,7 +1,5 @@
 package co.elastic.logstash.filters.elasticintegration.geoip;
 
-import org.elasticsearch.ingest.geoip.IpDatabase;
-
 interface IpDatabaseHolder {
     boolean isValid();
 

src/main/java/co/elastic/logstash/filters/elasticintegration/geoip/IpDatabaseProvider.java

@@ -8,9 +8,9 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.cluster.metadata.ProjectId;
-import org.elasticsearch.ingest.geoip.IpDatabase;
 import org.elasticsearch.logstashbridge.core.IOUtilsBridge;
+import org.elasticsearch.logstashbridge.geoip.IpDatabaseBridge;
+import org.elasticsearch.logstashbridge.geoip.IpDatabaseProviderBridge;
 
 import java.io.Closeable;
 import java.io.File;
@@ -23,24 +23,24 @@
 import java.util.Map;
 import java.util.Objects;
 
-public class IpDatabaseProvider implements org.elasticsearch.ingest.geoip.IpDatabaseProvider, Closeable {
+public class IpDatabaseProvider extends IpDatabaseProviderBridge.AbstractExternal implements Closeable {
 
     private static final Logger LOGGER = LogManager.getLogger(IpDatabaseProvider.class);
 
     private final Map<String, IpDatabaseHolder> databaseMap;
 
-    IpDatabaseProvider(Map<String, IpDatabaseHolder> databaseMap) {
+    public IpDatabaseProvider(Map<String, IpDatabaseHolder> databaseMap) {
         this.databaseMap = Map.copyOf(databaseMap);
     }
 
     @Override
-    public Boolean isValid(ProjectId projectId, String databaseIdentifierFileName) {
+    public Boolean isValid(String databaseIdentifierFileName) {
         final IpDatabaseHolder holder = getDatabaseHolder(databaseIdentifierFileName);
         return Objects.nonNull(holder) && holder.isValid();
     }
 
     @Override
-    public IpDatabase getDatabase(ProjectId projectId, String databaseIdentifierFileName) {
+    public IpDatabaseBridge getDatabase(String databaseIdentifierFileName) {
         final IpDatabaseHolder holder = getDatabaseHolder(databaseIdentifierFileName);
         if (Objects.isNull(holder)) {
             return null;