[logstash bridge]: stableapi bootstrap (elastic#108171)

* [logstash bridge]: stableapi bootstrap

Adds a new `logstash-bridge` project in `/libs` that exports api-stable
wrappers for the elasticsearch-internal types that Logstash's Elastic
Integration Filter relies on to provide ingest pipeline execution inside
of Logstash.

These bridge classes prevent Elasticsearch-internal refactorings from
breaking the Logstash-owned project.

* Update docs/changelog/108171.yaml

* rename StableAPI -> StableBridgeAPI

Author: yaauie

.github/CODEOWNERS

@@ -20,6 +20,9 @@ x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/Monito
 x-pack/plugin/fleet/src/main/java/org/elasticsearch/xpack/fleet @elastic/fleet
 x-pack/plugin/core/src/main/resources/fleet-* @elastic/fleet
 
+# Logstash
+libs/logstash-bridge @elastic/logstash
+
 # Kibana Security
 x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/store/KibanaOwnedReservedRoleDescriptors.java @elastic/kibana-security
 

docs/changelog/108171.yaml

@@ -0,0 +1,5 @@
+pr: 108171
+summary: "add Elastic-internal stable bridge api for use by Logstash"
+area: Infra/Core
+type: enhancement
+issues: []

libs/logstash-bridge/README.md

@@ -0,0 +1,8 @@
+## Logstash Bridge
+
+This package contains bridge functionality to ensure that Logstash's Elastic Integration plugin
+has access to the minimal subset of Elasticsearch to perform its functions without relying on
+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.

libs/logstash-bridge/build.gradle

@@ -0,0 +1,24 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+apply plugin: 'elasticsearch.build'
+
+dependencies {
+  compileOnly project(':server')
+  compileOnly project(':libs:elasticsearch-core')
+  compileOnly project(':libs:elasticsearch-plugin-api')
+  compileOnly project(':libs:elasticsearch-x-content')
+  compileOnly project(':modules:lang-painless')
+  compileOnly project(':modules:lang-painless:spi')
+  compileOnly project(':modules:lang-mustache')
+  compileOnly project(':modules:ingest-common')
+//  compileOnly project(':modules:ingest-geoip')
+}
+
+tasks.named('forbiddenApisMain').configure {
+  replaceSignatureFiles 'jdk-signatures'
+}

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

@@ -0,0 +1,27 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+/** Elasticsearch Logstash Bridge. */
+module org.elasticsearch.logstashbridge {
+    requires org.elasticsearch.base;
+    requires org.elasticsearch.grok;
+    requires org.elasticsearch.server;
+    requires org.elasticsearch.painless;
+    requires org.elasticsearch.painless.spi;
+    requires org.elasticsearch.mustache;
+    requires org.elasticsearch.xcontent;
+
+    exports org.elasticsearch.logstashbridge;
+    exports org.elasticsearch.logstashbridge.common;
+    exports org.elasticsearch.logstashbridge.core;
+    exports org.elasticsearch.logstashbridge.env;
+    exports org.elasticsearch.logstashbridge.ingest;
+    exports org.elasticsearch.logstashbridge.plugins;
+    exports org.elasticsearch.logstashbridge.script;
+    exports org.elasticsearch.logstashbridge.threadpool;
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * A {@code StableBridgeAPI} is the stable bridge to an Elasticsearch API, and can produce instances
+ * from the actual API that they mirror. As part of the LogstashBridge project, these classes are relied
+ * upon by the "Elastic Integration Filter Plugin" for Logstash and their external shapes mut not change
+ * without coordination with the maintainers of that project.
+ *
+ * @param <T> the actual type of the Elasticsearch API being mirrored
+ */
+public interface StableBridgeAPI<T> {
+    T unwrap();
+
+    static <T> T unwrapNullable(final StableBridgeAPI<T> nullableStableBridgeAPI) {
+        if (Objects.isNull(nullableStableBridgeAPI)) {
+            return null;
+        }
+        return nullableStableBridgeAPI.unwrap();
+    }
+
+    static <K, T> Map<K, T> unwrap(final Map<K, ? extends StableBridgeAPI<T>> bridgeMap) {
+        return bridgeMap.entrySet().stream().collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> e.getValue().unwrap()));
+    }
+
+    static <K, T, B extends StableBridgeAPI<T>> Map<K, B> wrap(final Map<K, T> rawMap, final Function<T, B> wrapFunction) {
+        return rawMap.entrySet().stream().collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> wrapFunction.apply(e.getValue())));
+    }
+
+    static <T, B extends StableBridgeAPI<T>> B wrap(final T delegate, final Function<T, B> wrapFunction) {
+        if (Objects.isNull(delegate)) {
+            return null;
+        }
+        return wrapFunction.apply(delegate);
+    }
+
+    abstract class Proxy<T> implements StableBridgeAPI<T> {
+        protected final T delegate;
+
+        protected Proxy(final T delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public T unwrap() {
+            return delegate;
+        }
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/common/SettingsBridge.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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge.common;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+
+public class SettingsBridge extends StableBridgeAPI.Proxy<Settings> {
+
+    public static SettingsBridge wrap(final Settings delegate) {
+        return new SettingsBridge(delegate);
+    }
+
+    public static Builder builder() {
+        return Builder.wrap(Settings.builder());
+    }
+
+    public SettingsBridge(final Settings delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public Settings unwrap() {
+        return this.delegate;
+    }
+
+    public static class Builder extends StableBridgeAPI.Proxy<Settings.Builder> {
+        static Builder wrap(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.delegate.put(key, value);
+            return this;
+        }
+
+        public SettingsBridge build() {
+            return new SettingsBridge(this.delegate.build());
+        }
+    }
+}

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

@@ -0,0 +1,22 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge.core;
+
+import org.elasticsearch.core.IOUtils;
+
+import java.io.Closeable;
+
+public class IOUtilsBridge {
+    public static void closeWhileHandlingException(final Iterable<? extends Closeable> objects) {
+        IOUtils.closeWhileHandlingException(objects);
+    }
+
+    public static void closeWhileHandlingException(final Closeable closeable) {
+        IOUtils.closeWhileHandlingException(closeable);
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/env/EnvironmentBridge.java

@@ -0,0 +1,33 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge.env;
+
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+import org.elasticsearch.logstashbridge.common.SettingsBridge;
+
+import java.nio.file.Path;
+
+public class EnvironmentBridge extends StableBridgeAPI.Proxy<Environment> {
+    public static EnvironmentBridge wrap(final Environment delegate) {
+        return new EnvironmentBridge(delegate);
+    }
+
+    public EnvironmentBridge(final SettingsBridge settingsBridge, final Path configPath) {
+        this(new Environment(settingsBridge.unwrap(), configPath));
+    }
+
+    private EnvironmentBridge(final Environment delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public Environment unwrap() {
+        return this.delegate;
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/ingest/ConfigurationUtilsBridge.java

@@ -0,0 +1,47 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge.ingest;
+
+import org.elasticsearch.ingest.ConfigurationUtils;
+import org.elasticsearch.logstashbridge.script.ScriptServiceBridge;
+import org.elasticsearch.logstashbridge.script.TemplateScriptBridge;
+
+import java.util.Map;
+
+public class ConfigurationUtilsBridge {
+    public static TemplateScriptBridge.Factory compileTemplate(
+        final String processorType,
+        final String processorTag,
+        final String propertyName,
+        final String propertyValue,
+        final ScriptServiceBridge scriptServiceBridge
+    ) {
+        return new TemplateScriptBridge.Factory(
+            ConfigurationUtils.compileTemplate(processorType, processorTag, propertyName, propertyValue, scriptServiceBridge.unwrap())
+        );
+    }
+
+    public static String readStringProperty(
+        final String processorType,
+        final String processorTag,
+        final Map<String, Object> configuration,
+        final String propertyName
+    ) {
+        return ConfigurationUtils.readStringProperty(processorType, processorTag, configuration, propertyName);
+    }
+
+    public static Boolean readBooleanProperty(
+        final String processorType,
+        final String processorTag,
+        final Map<String, Object> configuration,
+        final String propertyName,
+        final boolean defaultValue
+    ) {
+        return ConfigurationUtils.readBooleanProperty(processorType, processorTag, configuration, propertyName, defaultValue);
+    }
+}

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/ingest/IngestDocumentBridge.java

@@ -0,0 +1,91 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.logstashbridge.ingest;
+
+import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.LogstashInternalBridge;
+import org.elasticsearch.logstashbridge.StableBridgeAPI;
+import org.elasticsearch.logstashbridge.script.MetadataBridge;
+import org.elasticsearch.logstashbridge.script.TemplateScriptBridge;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiConsumer;
+
+public class IngestDocumentBridge extends StableBridgeAPI.Proxy<IngestDocument> {
+
+    public static String INGEST_KEY = IngestDocument.INGEST_KEY;
+
+    public static IngestDocumentBridge wrap(final IngestDocument ingestDocument) {
+        if (ingestDocument == null) {
+            return null;
+        }
+        return new IngestDocumentBridge(ingestDocument);
+    }
+
+    public IngestDocumentBridge(final Map<String, Object> sourceAndMetadata, final Map<String, Object> ingestMetadata) {
+        this(new IngestDocument(sourceAndMetadata, ingestMetadata));
+    }
+
+    private IngestDocumentBridge(IngestDocument inner) {
+        super(inner);
+    }
+
+    public MetadataBridge getMetadata() {
+        return new MetadataBridge(delegate.getMetadata());
+    }
+
+    public Map<String, Object> getSource() {
+        return delegate.getSource();
+    }
+
+    public boolean updateIndexHistory(final String index) {
+        return delegate.updateIndexHistory(index);
+    }
+
+    public Set<String> getIndexHistory() {
+        return Set.copyOf(delegate.getIndexHistory());
+    }
+
+    public boolean isReroute() {
+        return LogstashInternalBridge.isReroute(delegate);
+    }
+
+    public void resetReroute() {
+        LogstashInternalBridge.resetReroute(delegate);
+    }
+
+    public Map<String, Object> getIngestMetadata() {
+        return Map.copyOf(delegate.getIngestMetadata());
+    }
+
+    public <T> T getFieldValue(final String fieldName, final Class<T> type) {
+        return delegate.getFieldValue(fieldName, type);
+    }
+
+    public <T> T getFieldValue(final String fieldName, final Class<T> type, final boolean ignoreMissing) {
+        return delegate.getFieldValue(fieldName, type, ignoreMissing);
+    }
+
+    public String renderTemplate(final TemplateScriptBridge.Factory templateScriptFactory) {
+        return delegate.renderTemplate(templateScriptFactory.unwrap());
+    }
+
+    public void setFieldValue(final String path, final Object value) {
+        delegate.setFieldValue(path, value);
+    }
+
+    public void removeField(final String path) {
+        delegate.removeField(path);
+    }
+
+    // public void executePipeline(Pipeline pipeline, BiConsumer<IngestDocument, Exception> handler) {
+    public void executePipeline(final PipelineBridge pipelineBridge, final BiConsumer<IngestDocumentBridge, Exception> handler) {
+        this.delegate.executePipeline(pipelineBridge.unwrap(), (unwrapped, e) -> handler.accept(IngestDocumentBridge.wrap(unwrapped), e));
+    }
+}