[logstash-bridge] Introduce GeoIp interfaces

libs/logstash-bridge/README.md

@@ -6,3 +6,8 @@ other Elasticsearch internals.
 
 If a change is introduced in a separate Elasticsearch project that causes this project to fail,
 please consult with members of @elastic/logstash to chart a path forward.
+
+## How to build the module?
+```shell
+./gradlew :lib:logstash-bridge:build
+```

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/core/FailProcessorExceptionBridge.java

@@ -0,0 +1,23 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.ingest.common.FailProcessorException;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+public class FailProcessorExceptionBridge extends StableBridgeAPI.ProxyInternal<FailProcessorException> {
+    protected FailProcessorExceptionBridge(FailProcessorException internalDelegate) {
+        super(internalDelegate);
+    }
+
+    public static boolean isInstanceOf(Throwable exception) {
+        return exception instanceof FailProcessorException;
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/core/RefCountingRunnableBridge.java

@@ -0,0 +1,37 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.action.support.RefCountingRunnable;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+public class RefCountingRunnableBridge extends StableBridgeAPI.ProxyInternal<RefCountingRunnable> {
+
+    private RefCountingRunnableBridge(final RefCountingRunnable delegate) {
+        super(delegate);
+    }
+
+    public RefCountingRunnableBridge(final Runnable delegate) {
+        super(new RefCountingRunnable(delegate));
+    }
+
+    public void close() {
+        toInternal().close();
+    }
+
+    public ReleasableBridge acquire() {
+        return new ReleasableBridge.ProxyInternal(toInternal().acquire());
+    }
+
+    @Override
+    public RefCountingRunnable toInternal() {
+        return this.internalDelegate;
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/core/ReleasableBridge.java

@@ -0,0 +1,30 @@
+/*
+ * 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.core;
+
+import org.elasticsearch.core.Releasable;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+public interface ReleasableBridge extends StableBridgeAPI<Releasable> {
+
+    void close();
+
+    class ProxyInternal extends StableBridgeAPI.ProxyInternal<Releasable> implements ReleasableBridge {
+
+        public ProxyInternal(final Releasable delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public void close() {
+            toInternal().close();
+        }
+    }
+}

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);
+        }
+    }
+}