[logstash-bridge] Introduce GeoIp interfaces (#132595)

* Use default project resolver instead null which caused NPE when running elastic_integration integration tests.

* Introduce BridgeProjectIdResolver to applying null for project ID.

* Put a note to the ProjectIdResolverBridge about resolving project ID for multi-project.

* GeoIP interfaces for the Logstash bridge where elastic_integration plugin utilizes.

* Add a note to maxmind depenedency of the ingest-geoip module that what upgrading the dependency, apply same to the logstash-bridge.

* A safeguard in case if bridge becomes null, might happen while cyclic pipeline execution.

* Introduce bridges for FailProcessorException, Releasable and RefCountingRunnable ES core components.

* hamfisted cleanup

* add static GeoIpProcessorBridge#newFactory(IpDatabaseProviderBridge)

* provide ProcessorBridge.Factory.AbstractExternal

* add ProjectIdBridge

* add synchronous ProcessorBridge#execute

* migrate RefCountingRunnableBridge to interface pattern

* migrate SettingsBridge to interface pattern

* migrate PipelineConfigurationBridge to interface pattern

* migrate EnvironmentBridge to interface pattern

* migrate IngestDocumentBridge to interface pattern

* migrate PipelineBridge to interface pattern

* reduce unnecessary proxying in ProcessorBridge

* migrate MetadataBridge to interface pattern

* migrate ScriptServiceBridge to interface pattern

* migrate ThreadPoolBridge to interface pattern

* visibility cleanup, move AbstractExternal-s

* Remove accidental change where ingest-geoip module was opened to maxmind.db pkg

---------

Co-authored-by: Ry Biesemeyer <[email protected]>

Author: mashhurs

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/StableBridgeAPI.java

@@ -50,6 +50,11 @@ static <T, B extends StableBridgeAPI<T>> B fromInternal(final T delegate, final
      * An {@code ProxyInternal<INTERNAL>} is an implementation of {@code StableBridgeAPI<INTERNAL>} that
      * proxies calls to a delegate that is an actual {@code INTERNAL}.
      *
+     * <p>
+     *     implementations are intended to be <em>opaque</em> to consumers of this library,
+     *     and should <em>NOT</em> have public constructors.
+     * </p>
+     *
      * @param <INTERNAL>
      */
     abstract class ProxyInternal<INTERNAL> implements StableBridgeAPI<INTERNAL> {

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

@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+/**
+ * A {@link StableBridgeAPI} for {@link ProjectId}
+ */
+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;
+    }
+
+    /**
+     * An implementation of {@link ProjectIdBridge} that proxies calls to
+     * an internal {@link ProjectId} instance.
+     *
+     * @see StableBridgeAPI.ProxyInternal
+     */
+    final class ProxyInternal extends StableBridgeAPI.ProxyInternal<ProjectId> implements ProjectIdBridge {
+        private static final ProjectIdBridge.ProxyInternal DEFAULT = new ProjectIdBridge.ProxyInternal(ProjectId.DEFAULT);
+
+        ProxyInternal(ProjectId internalDelegate) {
+            super(internalDelegate);
+        }
+
+        @Override
+        public String id() {
+            return this.internalDelegate.id();
+        }
+
+        @Override
+        public ProjectId toInternal() {
+            return this.internalDelegate;
+        }
+    }
+}

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

@@ -12,46 +12,27 @@
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
 
 /**
- * An external bridge for {@link Settings}
+ * A {@link StableBridgeAPI} for {@link Settings}
  */
-public class SettingsBridge extends StableBridgeAPI.ProxyInternal<Settings> {
+public interface SettingsBridge extends StableBridgeAPI<Settings> {
 
-    public static SettingsBridge fromInternal(final Settings delegate) {
-        return new SettingsBridge(delegate);
+    static SettingsBridge fromInternal(final Settings delegate) {
+        return new ProxyInternal(delegate);
     }
 
-    public static Builder builder() {
-        return Builder.fromInternal(Settings.builder());
-    }
-
-    public SettingsBridge(final Settings delegate) {
-        super(delegate);
-    }
-
-    @Override
-    public Settings toInternal() {
-        return this.internalDelegate;
+    static SettingsBuilderBridge builder() {
+        return SettingsBuilderBridge.fromInternal(Settings.builder());
     }
 
     /**
-     * An external bridge for {@link Settings.Builder} that proxies calls to a real {@link Settings.Builder}
+     * An implementation of {@link SettingsBridge} that proxies calls to
+     * an internal {@link Settings} instance.
+     *
+     * @see StableBridgeAPI.ProxyInternal
      */
-    public static class Builder extends StableBridgeAPI.ProxyInternal<Settings.Builder> {
-        static Builder fromInternal(final Settings.Builder delegate) {
-            return new Builder(delegate);
-        }
-
-        private Builder(final Settings.Builder delegate) {
-            super(delegate);
-        }
-
-        public Builder put(final String key, final String value) {
-            this.internalDelegate.put(key, value);
-            return this;
-        }
-
-        public SettingsBridge build() {
-            return new SettingsBridge(this.internalDelegate.build());
+    final class ProxyInternal extends StableBridgeAPI.ProxyInternal<Settings> implements SettingsBridge {
+        ProxyInternal(Settings internalDelegate) {
+            super(internalDelegate);
         }
     }
 }

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

@@ -0,0 +1,51 @@
+/*
+ * 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.common.settings.Settings;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+/**
+ * A {@link StableBridgeAPI} for {@link Settings.Builder}.
+ */
+public interface SettingsBuilderBridge extends StableBridgeAPI<Settings.Builder> {
+
+    SettingsBuilderBridge put(String key, String value);
+
+    SettingsBridge build();
+
+    static SettingsBuilderBridge fromInternal(final Settings.Builder builder) {
+        return new ProxyInternal(builder);
+    }
+
+    /**
+     * An implementation of {@link SettingsBuilderBridge} that proxies calls to
+     * an internal {@link Settings.Builder} instance.
+     *
+     * @see StableBridgeAPI.ProxyInternal
+     */
+    final class ProxyInternal extends StableBridgeAPI.ProxyInternal<Settings.Builder> implements SettingsBuilderBridge {
+        ProxyInternal(Settings.Builder internalDelegate) {
+            super(internalDelegate);
+        }
+
+        @Override
+        public SettingsBuilderBridge put(String key, String value) {
+            internalDelegate.put(key, value);
+            return this;
+        }
+
+        @Override
+        public SettingsBridge build() {
+            final Settings delegate = internalDelegate.build();
+            return SettingsBridge.fromInternal(delegate);
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/core/CheckedBiFunctionBridge.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.common.CheckedBiFunction;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+/**
+ * A stable interface on top of {@link CheckedBiFunction}.
+ * @param <T> type of lhs parameter
+ * @param <U> type of rhs parameter
+ * @param <R> type of return value
+ * @param <E> type of anticipated exception
+ */
+@FunctionalInterface
+public interface CheckedBiFunctionBridge<T, U, R, E extends Exception> extends StableBridgeAPI<CheckedBiFunction<T, U, R, E>> {
+    R apply(T t, U u) throws E;
+
+    @Override
+    default CheckedBiFunction<T, U, R, E> toInternal() {
+        return this::apply;
+    }
+}

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

@@ -16,6 +16,8 @@
  * An external bridge for {@link IOUtils}
  */
 public class IOUtilsBridge {
+    private IOUtilsBridge() {}
+
     public static void closeWhileHandlingException(final Iterable<? extends Closeable> objects) {
         IOUtils.closeWhileHandlingException(objects);
     }

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

@@ -0,0 +1,61 @@
+/*
+ * 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.core.Releasable;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+import java.io.Closeable;
+
+/**
+ * A {@link StableBridgeAPI} for {@link RefCountingRunnable}
+ */
+public interface RefCountingRunnableBridge extends StableBridgeAPI<RefCountingRunnable>, Closeable {
+
+    @Override // only RuntimeException
+    void close();
+
+    ReleasableBridge acquire();
+
+    /**
+     * An API-stable factory method for {@link RefCountingRunnableBridge}
+     * @param delegate the {@link Runnable} to execute when all refs are closed
+     * @return a {@link RefCountingRunnableBridge} that will execute the provided
+     * block when all refs are closed
+     */
+    static RefCountingRunnableBridge create(final Runnable delegate) {
+        final RefCountingRunnable refCountingRunnable = new RefCountingRunnable(delegate);
+        return new ProxyInternal(refCountingRunnable);
+    }
+
+    /**
+     * An implementation of {@link RefCountingRunnableBridge} that proxies calls through
+     * to an internal {@link RefCountingRunnable}.
+     * @see StableBridgeAPI.ProxyInternal
+     */
+    final class ProxyInternal extends StableBridgeAPI.ProxyInternal<RefCountingRunnable> implements RefCountingRunnableBridge {
+        private ProxyInternal(final RefCountingRunnable delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public void close() {
+            toInternal().close();
+        }
+
+        @Override
+        public ReleasableBridge acquire() {
+            @SuppressWarnings("resource")
+            final Releasable releasable = toInternal().acquire();
+            return ReleasableBridge.fromInternal(releasable);
+        }
+    }
+}