Logstash bridge refinement

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/common/ProjectIdBridge.java

@@ -0,0 +1,50 @@
+/*
+ * 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.common;
+
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+public interface ProjectIdBridge extends StableBridgeAPI<ProjectId> {
+    String id();
+
+    static ProjectIdBridge fromInternal(final ProjectId projectId) {
+        return new ProxyInternal(projectId);
+    }
+
+    static ProjectIdBridge fromId(final String id) {
+        final ProjectId internal = ProjectId.fromId(id);
+        return new ProxyInternal(internal);
+    }
+
+    static ProjectIdBridge getDefault() {
+        return ProxyInternal.DEFAULT;
+    }
+
+    class ProxyInternal implements ProjectIdBridge {
+        private final ProjectId internalDelegate;
+
+        static final ProjectIdBridge.ProxyInternal DEFAULT = new ProjectIdBridge.ProxyInternal(ProjectId.DEFAULT);
+
+        public ProxyInternal(ProjectId internalDelegate) {
+            this.internalDelegate = internalDelegate;
+        }
+
+        @Override
+        public String id() {
+            return toInternal().id();
+        }
+
+        @Override
+        public ProjectId toInternal() {
+            return this.internalDelegate;
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/core/FailProcessorExceptionBridge.java → libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/core/CheckedBiFunctionBridge.java

@@ -9,15 +9,14 @@
 
 package org.elasticsearch.logstashbridge.core;
 
-import org.elasticsearch.ingest.common.FailProcessorException;
+import org.elasticsearch.common.CheckedBiFunction;
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
 
-public class FailProcessorExceptionBridge extends StableBridgeAPI.ProxyInternal<FailProcessorException> {
-    protected FailProcessorExceptionBridge(FailProcessorException internalDelegate) {
-        super(internalDelegate);
-    }
+public interface CheckedBiFunctionBridge<T, U, R, E extends Exception> extends StableBridgeAPI<CheckedBiFunction<T, U, R, E>> {
+    R apply(T t, U u) throws E;
 
-    public static boolean isInstanceOf(Throwable exception) {
-        return exception instanceof FailProcessorException;
+    @Override
+    default CheckedBiFunction<T, U, R, E> toInternal() {
+        return this::apply;
     }
 }

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

@@ -9,18 +9,25 @@
 package org.elasticsearch.logstashbridge.geoip;
 
 import org.elasticsearch.ingest.geoip.GeoIpProcessor;
-import org.elasticsearch.ingest.geoip.IpDatabaseProvider;
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
+import org.elasticsearch.logstashbridge.ingest.ProcessorBridge;
 
 /**
  * An external bridge for {@link GeoIpProcessor}
  */
-public interface GeoIpProcessorBridge {
+public interface GeoIpProcessorBridge extends StableBridgeAPI<GeoIpProcessor> {
 
-    class Factory extends StableBridgeAPI.ProxyInternal<GeoIpProcessor.Factory> {
+    static Factory newFactory(final IpDatabaseProviderBridge ipDatabaseProviderBridge) {
+        return new Factory(ipDatabaseProviderBridge);
+    }
+
+    /**
+     * A {@link ProcessorBridge.Factory} implementation for the {@link GeoIpProcessor}
+     */
+    class Factory extends ProcessorBridge.Factory.ProxyInternal {
 
-        public Factory(String type, IpDatabaseProvider ipDatabaseProvider) {
-            super(new GeoIpProcessor.Factory(type, ipDatabaseProvider));
+        Factory(final IpDatabaseProviderBridge ipDatabaseProviderBridge) {
+            super(new GeoIpProcessor.Factory("geoip", ipDatabaseProviderBridge.toInternal()));
         }
     }
 }

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

@@ -10,11 +10,11 @@
 
 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 org.elasticsearch.logstashbridge.core.CheckedBiFunctionBridge;
 
 import java.io.IOException;
 
@@ -25,31 +25,16 @@ 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);
-        }
-    }
+    <RESPONSE> RESPONSE getResponse(String ipAddress, CheckedBiFunctionBridge<Reader, String, RESPONSE, Exception> responseProvider);
 
     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.
+     * that proxies calls to the external implementation.
      */
     abstract class AbstractExternal implements IpDatabaseBridge {
         private ProxyExternal internalDatabase;
@@ -64,10 +49,6 @@ public IpDatabase toInternal() {
 
         private class ProxyExternal implements IpDatabase {
 
-            private AbstractExternal getIpDatabaseBridge() {
-                return AbstractExternal.this;
-            }
-
             @Override
             public String getDatabaseType() throws IOException {
                 return AbstractExternal.this.getDatabaseType();
@@ -78,7 +59,7 @@ public <RESPONSE> RESPONSE getResponse(
                 String ipAddress,
                 CheckedBiFunction<Reader, String, RESPONSE, Exception> responseProvider
             ) {
-                return AbstractExternal.this.getResponse(ipAddress, responseProvider);
+                return AbstractExternal.this.getResponse(ipAddress, responseProvider::apply);
             }
 
             @Override
@@ -87,34 +68,4 @@ public void close() throws IOException {
             }
         }
     }
-
-    /**
-     * 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

@@ -14,7 +14,7 @@
 import org.elasticsearch.ingest.geoip.IpDatabaseProvider;
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
 
-import java.util.Objects;
+import static org.elasticsearch.logstashbridge.StableBridgeAPI.toInternalNullable;
 
 /**
  * An external bridge for {@link Processor}
@@ -25,18 +25,11 @@ public interface IpDatabaseProviderBridge extends StableBridgeAPI<IpDatabaseProv
 
     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.
+     * that proxies calls to the external implementation.
      */
     abstract class AbstractExternal implements IpDatabaseProviderBridge {
         private AbstractExternal.ProxyExternal internalProcessor;
@@ -56,34 +49,13 @@ private AbstractExternal getIpDatabaseProviderBridge() {
 
             @Override
             public Boolean isValid(ProjectId projectId, String name) {
-                return IpDatabaseProviderBridge.AbstractExternal.this.isValid(name);
+                return 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();
+                return toInternalNullable(AbstractExternal.this.getDatabase(name));
             }
         }
     }
-
-    /**
-     * 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/ingest/PipelineBridge.java

@@ -8,7 +8,7 @@
  */
 package org.elasticsearch.logstashbridge.ingest;
 
-import org.elasticsearch.core.FixForMultiProject;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.ingest.Pipeline;
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
 import org.elasticsearch.logstashbridge.script.ScriptServiceBridge;
@@ -24,7 +24,6 @@ public static PipelineBridge fromInternal(final Pipeline pipeline) {
         return new PipelineBridge(pipeline);
     }
 
-    @FixForMultiProject(description = "should we pass a non-null project ID here?")
     public static PipelineBridge create(
         String id,
         Map<String, Object> config,
@@ -37,7 +36,7 @@ public static PipelineBridge create(
                 config,
                 StableBridgeAPI.toInternal(processorFactories),
                 StableBridgeAPI.toInternalNullable(scriptServiceBridge),
-                null
+                ProjectId.DEFAULT
             )
         );
     }