GeoIP interfaces for the Logstash bridge where elastic_integration plugin utilizes.

Author: mashhurs

libs/logstash-bridge/build.gradle

@@ -24,6 +24,8 @@ dependencies {
   compileOnly project(':x-pack:plugin:redact')
   compileOnly project(':x-pack:plugin:spatial')
   compileOnly project(':x-pack:plugin:wildcard')
+
+  compileOnly('com.maxmind.db:maxmind-db:3.1.1')
 }
 
 tasks.named('forbiddenApisMain').configure {

libs/logstash-bridge/src/main/java/module-info.java

@@ -23,6 +23,8 @@
     requires org.elasticsearch.redact;
     requires org.elasticsearch.spatial;
     requires org.elasticsearch.wildcard;
+    requires org.elasticsearch.ingest.geoip;
+    requires com.maxmind.db;
 
     exports org.elasticsearch.logstashbridge;
     exports org.elasticsearch.logstashbridge.common;

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/geoip/GeoIpProcessorBridge.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.logstashbridge.geoip;
+
+import org.elasticsearch.ingest.geoip.GeoIpProcessor;
+import org.elasticsearch.ingest.geoip.IpDatabaseProvider;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+/**
+ * An external bridge for {@link GeoIpProcessor}
+ */
+public interface GeoIpProcessorBridge {
+
+    class Factory extends StableBridgeAPI.ProxyInternal<GeoIpProcessor.Factory> {
+
+        public Factory(String type, IpDatabaseProvider ipDatabaseProvider) {
+            super(new GeoIpProcessor.Factory(type, ipDatabaseProvider));
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/geoip/IpDatabaseBridge.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.logstashbridge.geoip;
+
+import com.maxmind.db.Reader;
+
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.common.CheckedBiFunction;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.ingest.geoip.IpDatabase;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+import java.io.IOException;
+
+/**
+ * An external bridge for {@link IpDatabase}
+ */
+public interface IpDatabaseBridge extends StableBridgeAPI<IpDatabase> {
+
+    String getDatabaseType() throws IOException;
+
+    MaxMindDbBridge.Reader getDatabaseReader() throws IOException;
+
+    @Nullable
+    default <RESPONSE> RESPONSE getResponse(String ipAddress, CheckedBiFunction<Reader, String, RESPONSE, Exception> responseProvider) {
+        try {
+            return responseProvider.apply(this.getDatabaseReader().toInternal(), ipAddress);
+        } catch (Exception e) {
+            throw ExceptionsHelper.convertToRuntime(e);
+        }
+    }
+
+    void close() throws IOException;
+
+    static IpDatabaseBridge fromInternal(final IpDatabase internalDatabase) {
+        if (internalDatabase instanceof AbstractExternal.ProxyExternal externalProxy) {
+            return externalProxy.getIpDatabaseBridge();
+        }
+        return new ProxyInternal(internalDatabase);
+    }
+
+    /**
+     * The {@code IpDatabaseBridge.AbstractExternal} is an abstract base class for implementing
+     * the {@link IpDatabaseBridge} externally to the Elasticsearch code-base. It takes care of
+     * the details of maintaining a singular internal-form implementation of {@link IpDatabase}
+     * that proxies calls through the external implementation.
+     */
+    abstract class AbstractExternal implements IpDatabaseBridge {
+        private ProxyExternal internalDatabase;
+
+        @Override
+        public IpDatabase toInternal() {
+            if (internalDatabase == null) {
+                internalDatabase = new ProxyExternal();
+            }
+            return internalDatabase;
+        }
+
+        private class ProxyExternal implements IpDatabase {
+
+            private AbstractExternal getIpDatabaseBridge() {
+                return AbstractExternal.this;
+            }
+
+            @Override
+            public String getDatabaseType() throws IOException {
+                return AbstractExternal.this.getDatabaseType();
+            }
+
+            @Override
+            public <RESPONSE> RESPONSE getResponse(
+                String ipAddress,
+                CheckedBiFunction<Reader, String, RESPONSE, Exception> responseProvider
+            ) {
+                return AbstractExternal.this.getResponse(ipAddress, responseProvider);
+            }
+
+            @Override
+            public void close() throws IOException {
+                AbstractExternal.this.close();
+            }
+        }
+    }
+
+    /**
+     * An implementation of {@link IpDatabaseBridge} that proxies to an internal {@link IpDatabase}
+     */
+    class ProxyInternal extends StableBridgeAPI.ProxyInternal<IpDatabase> implements IpDatabaseBridge {
+
+        public ProxyInternal(final IpDatabase delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public String getDatabaseType() throws IOException {
+            return toInternal().getDatabaseType();
+        }
+
+        @Override
+        public MaxMindDbBridge.Reader getDatabaseReader() throws IOException {
+            return null;
+        }
+
+        @Override
+        public <RESPONSE> RESPONSE getResponse(String ipAddress, CheckedBiFunction<Reader, String, RESPONSE, Exception> responseProvider) {
+            return toInternal().getResponse(ipAddress, responseProvider);
+        }
+
+        @Override
+        public void close() throws IOException {
+            toInternal().close();
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/geoip/IpDatabaseProviderBridge.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.logstashbridge.geoip;
+
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.ingest.geoip.IpDatabase;
+import org.elasticsearch.ingest.geoip.IpDatabaseProvider;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+import java.util.Objects;
+
+/**
+ * An external bridge for {@link Processor}
+ */
+public interface IpDatabaseProviderBridge extends StableBridgeAPI<IpDatabaseProvider> {
+
+    Boolean isValid(String name);
+
+    IpDatabaseBridge getDatabase(String name);
+
+    static IpDatabaseProviderBridge fromInternal(final IpDatabaseProvider internalProvider) {
+        if (internalProvider instanceof IpDatabaseProviderBridge.AbstractExternal.ProxyExternal externalProxy) {
+            return externalProxy.getIpDatabaseProviderBridge();
+        }
+        return new IpDatabaseProviderBridge.ProxyInternal(internalProvider);
+    }
+
+    /**
+     * The {@code IpDatabaseProviderBridge.AbstractExternal} is an abstract base class for implementing
+     * the {@link IpDatabaseProviderBridge} externally to the Elasticsearch code-base. It takes care of
+     * the details of maintaining a singular internal-form implementation of {@link IpDatabaseProvider}
+     * that proxies calls through the external implementation.
+     */
+    abstract class AbstractExternal implements IpDatabaseProviderBridge {
+        private AbstractExternal.ProxyExternal internalProcessor;
+
+        public IpDatabaseProvider toInternal() {
+            if (internalProcessor == null) {
+                internalProcessor = new AbstractExternal.ProxyExternal();
+            }
+            return internalProcessor;
+        }
+
+        private class ProxyExternal implements IpDatabaseProvider {
+
+            private AbstractExternal getIpDatabaseProviderBridge() {
+                return AbstractExternal.this;
+            }
+
+            @Override
+            public Boolean isValid(ProjectId projectId, String name) {
+                return IpDatabaseProviderBridge.AbstractExternal.this.isValid(name);
+            }
+
+            @Override
+            public IpDatabase getDatabase(ProjectId projectId, String name) {
+                IpDatabaseBridge bridge = IpDatabaseProviderBridge.AbstractExternal.this.getDatabase(name);
+                return Objects.isNull(bridge) ? null : bridge.toInternal();
+            }
+        }
+    }
+
+    /**
+     * An implementation of {@link IpDatabaseProviderBridge} that proxies to an internal {@link IpDatabaseProvider}
+     */
+    class ProxyInternal extends StableBridgeAPI.ProxyInternal<IpDatabaseProvider> implements IpDatabaseProviderBridge {
+        public ProxyInternal(final IpDatabaseProvider delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public Boolean isValid(String name) {
+            return toInternal().isValid(ProjectId.DEFAULT, name);
+        }
+
+        @Override
+        public IpDatabaseBridge getDatabase(String name) {
+            IpDatabase ipDatabase = toInternal().getDatabase(ProjectId.DEFAULT, name);
+            return new IpDatabaseBridge.ProxyInternal(ipDatabase);
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/geoip/MaxMindDbBridge.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.logstashbridge.geoip;
+
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+import java.io.File;
+import java.io.IOException;
+
+public interface MaxMindDbBridge {
+
+    class Reader extends StableBridgeAPI.ProxyInternal<com.maxmind.db.Reader> {
+
+        @SuppressForbidden(reason = "Maxmind Reader constructor requires java.io.File")
+        public Reader(final File databasePath, final NodeCache nodeCache) throws IOException {
+            super(new com.maxmind.db.Reader(databasePath, nodeCache.toInternal()));
+        }
+
+        protected Reader(final com.maxmind.db.Reader internalDelegate) {
+            super(internalDelegate);
+        }
+
+        @Override
+        public com.maxmind.db.Reader toInternal() {
+            return internalDelegate;
+        }
+
+        public String getDatabaseType() {
+            return toInternal().getMetadata().getDatabaseType();
+        }
+
+        public void close() throws IOException {
+            toInternal().close();
+        }
+    }
+
+    class NodeCache extends StableBridgeAPI.ProxyInternal<com.maxmind.db.NodeCache> {
+
+        protected NodeCache(final com.maxmind.db.NodeCache internalDelegate) {
+            super(internalDelegate);
+        }
+
+        public static NodeCache get(final int capacity) {
+            return new NodeCache(new com.maxmind.db.CHMCache(capacity));
+        }
+
+        public static NodeCache getInstance() {
+            return new NodeCache(com.maxmind.db.NoCache.getInstance());
+        }
+    }
+
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/script/ScriptServiceBridge.java

@@ -117,7 +117,7 @@ public void close() throws IOException {
     @FixForMultiProject
     // Logstash resolves and runs ingest pipelines based on the datastream.
     // How should ProjectIdResolverBridge behave in this case?
-    //  In other words, it looks we need to find a way to figure out which ingest pipeline belongs to which project.
+    // In other words, it looks we need to find a way to figure out which ingest pipeline belongs to which project.
     static class ProjectIdResolverBridge implements ProjectResolver {
 
         public static final ProjectIdResolverBridge INSTANCE = new ProjectIdResolverBridge();

modules/ingest-geoip/src/main/java/module-info.java

@@ -19,5 +19,5 @@
     exports org.elasticsearch.ingest.geoip.direct to org.elasticsearch.server;
     exports org.elasticsearch.ingest.geoip.stats to org.elasticsearch.server;
 
-    exports org.elasticsearch.ingest.geoip to com.maxmind.db;
+    exports org.elasticsearch.ingest.geoip to com.maxmind.db, org.elasticsearch.logstashbridge;
 }