Replace ES classes with their Bridge Stable API pairs.

Author: mashhurs

build.gradle

@@ -358,14 +358,14 @@ task shadeElasticsearchLogstashBridge(type: com.github.jengelman.gradle.plugins.
     description "Shades the Elasticsearch logstash-bridge jar"
 
     dependsOn buildElasticsearchLogstashBridge
-    
+
     from("${buildDir}/elasticsearch-source/libs/logstash-bridge/build/distributions") {
         include "elasticsearch-logstash-bridge-*.jar"
     }
-    
+
     archiveFileName = "elasticsearch-logstash-bridge-shaded.jar"
     destinationDirectory = file("${buildDir}/shaded")
-    
+
     exclude '**/module-info.class'
 }
 

src/main/java/co/elastic/logstash/filters/elasticintegration/ElasticsearchPipelineConfigurationResolver.java

@@ -15,7 +15,7 @@
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
-import org.elasticsearch.ingest.PipelineConfiguration;
+import org.elasticsearch.logstashbridge.ingest.PipelineConfigurationBridge;
 
 import java.util.Optional;
 
@@ -24,7 +24,7 @@
  * that retrieves pipelines from Elasticsearch.
  */
 public class ElasticsearchPipelineConfigurationResolver
-        extends AbstractSimpleResolver<String,PipelineConfiguration>
+        extends AbstractSimpleResolver<String, PipelineConfigurationBridge>
         implements PipelineConfigurationResolver {
     private final RestClient elasticsearchRestClient;
     private final PipelineConfigurationFactory pipelineConfigurationFactory;
@@ -37,13 +37,13 @@ public ElasticsearchPipelineConfigurationResolver(final RestClient elasticsearch
     }
 
     @Override
-    public Optional<PipelineConfiguration> resolveSafely(String pipelineName) throws Exception {
+    public Optional<PipelineConfigurationBridge> resolveSafely(String pipelineName) throws Exception {
         final Response response;
         try {
             final Request request = new Request("GET", URLEncodedUtils.formatSegments("_ingest", "pipeline", pipelineName));
             response = elasticsearchRestClient.performRequest(request);
             final String jsonEncodedPayload = EntityUtils.toString(response.getEntity());
-            final PipelineConfiguration pipelineConfiguration = pipelineConfigurationFactory.parseNamedObject(jsonEncodedPayload);
+            final PipelineConfigurationBridge pipelineConfiguration = pipelineConfigurationFactory.parseNamedObject(jsonEncodedPayload);
             return Optional.of(pipelineConfiguration);
         } catch (ResponseException re) {
             if (re.getResponse().getStatusLine().getStatusCode() == 404) {

src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessor.java

@@ -14,15 +14,15 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.support.RefCountingRunnable;
-import org.elasticsearch.core.IOUtils;
-import org.elasticsearch.ingest.IngestDocument;
-import org.elasticsearch.ingest.LogstashInternalBridge;
 import org.elasticsearch.ingest.common.FailProcessorException;
+import org.elasticsearch.logstashbridge.core.IOUtilsBridge;
+import org.elasticsearch.logstashbridge.ingest.IngestDocumentBridge;
 
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -32,7 +32,6 @@
 
 import static co.elastic.logstash.filters.elasticintegration.util.EventUtil.eventAsMap;
 import static co.elastic.logstash.filters.elasticintegration.util.EventUtil.serializeEventForLog;
-import static org.elasticsearch.core.Strings.format;
 
 /**
  * An {@link EventProcessor} processes {@link Event}s by:
@@ -151,7 +150,7 @@ void processRequest(final IntegrationRequest request) {
 
             final IngestPipeline ingestPipeline = loadedPipeline.get();
             LOGGER.trace(() -> String.format("Using loaded pipeline `%s` (%s)", pipelineName, System.identityHashCode(ingestPipeline)));
-            final IngestDocument ingestDocument = eventMarshaller.toIngestDocument(request.event());
+            final IngestDocumentBridge ingestDocument = eventMarshaller.toIngestDocument(request.event());
 
             resolvedIndexName.ifPresent(indexName -> {
                 ingestDocument.getMetadata().setIndex(indexName);
@@ -170,7 +169,7 @@ void processRequest(final IntegrationRequest request) {
         }
     }
 
-    private void executePipeline(final IngestDocument ingestDocument, final IngestPipeline ingestPipeline, final IntegrationRequest request) {
+    private void executePipeline(final IngestDocumentBridge ingestDocument, final IngestPipeline ingestPipeline, final IntegrationRequest request) {
         final String pipelineName = ingestPipeline.getId();
         final String originalIndex = ingestDocument.getMetadata().getIndex();
         ingestPipeline.execute(ingestDocument, (resultIngestDocument, ingestPipelineException) -> {
@@ -193,17 +192,17 @@ private void executePipeline(final IngestDocument ingestDocument, final IngestPi
             } else {
 
                 final String newIndex = resultIngestDocument.getMetadata().getIndex();
-                if (!Objects.equals(originalIndex, newIndex) && LogstashInternalBridge.isReroute(resultIngestDocument)) {
-                    LogstashInternalBridge.resetReroute(resultIngestDocument);
+                if (!Objects.equals(originalIndex, newIndex) && ingestDocument.isReroute()) {
+                    ingestDocument.resetReroute();
                     boolean cycle = !resultIngestDocument.updateIndexHistory(newIndex);
                     if (cycle) {
                         request.complete(incomingEvent -> {
-                            annotateIngestPipelineFailure(incomingEvent, pipelineName, Map.of("message", format(
-                                    "index cycle detected while processing pipeline [%s]: %s + %s",
-                                    pipelineName,
-                                    resultIngestDocument.getIndexHistory(),
-                                    newIndex
-                            )));
+                            annotateIngestPipelineFailure(incomingEvent, pipelineName, Map.of("message",
+                                    String.format(Locale.ROOT, "index cycle detected while processing pipeline [%s]: %s + %s",
+                                        pipelineName,
+                                        resultIngestDocument.getIndexHistory(),
+                                        newIndex)
+                            ));
                         });
                         return;
                     }
@@ -214,12 +213,14 @@ private void executePipeline(final IngestDocument ingestDocument, final IngestPi
                         final Optional<IngestPipeline> reroutePipeline = resolve(reroutePipelineName.get(), internalPipelineProvider);
                         if (reroutePipeline.isEmpty()) {
                             request.complete(incomingEvent -> {
-                                annotateIngestPipelineFailure(incomingEvent, pipelineName, Map.of("message", format(
-                                        "reroute failed to load next pipeline [%s]: %s -> %s",
+                                annotateIngestPipelineFailure(
+                                        incomingEvent,
                                         pipelineName,
-                                        resultIngestDocument.getIndexHistory(),
-                                        reroutePipelineName.get()
-                                )));
+                                        Map.of("message",
+                                                String.format(Locale.ROOT, "reroute failed to load next pipeline [%s]: %s -> %s",
+                                                        pipelineName,
+                                                        resultIngestDocument.getIndexHistory(),
+                                                        reroutePipelineName.get())));
                             });
                         } else {
                             executePipeline(resultIngestDocument, reroutePipeline.get(), request);
@@ -277,6 +278,6 @@ static private <T,R> Optional<R> resolve(T resolvable, Resolver<T,R> resolver) {
 
     @Override
     public void close() throws IOException {
-        IOUtils.closeWhileHandlingException(this.resourcesToClose);
+        IOUtilsBridge.closeWhileHandlingException(this.resourcesToClose);
     }
 }

src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java

@@ -17,41 +17,20 @@
 import co.elastic.logstash.filters.elasticintegration.resolver.ResolverCache;
 import co.elastic.logstash.filters.elasticintegration.util.Exceptions;
 import co.elastic.logstash.filters.elasticintegration.util.PluginContext;
-import co.elastic.logstash.filters.elasticintegration.util.PluginProjectResolver;
 import com.google.common.util.concurrent.Service;
 import com.google.common.util.concurrent.ServiceManager;
 import org.elasticsearch.client.RestClient;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.IOUtils;
-import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.env.Environment;
-import org.elasticsearch.ingest.IngestService;
-import org.elasticsearch.ingest.LogstashInternalBridge;
-import org.elasticsearch.ingest.Processor;
-import org.elasticsearch.ingest.common.IngestCommonPlugin;
-import org.elasticsearch.ingest.common.ProcessorsWhitelistExtension;
-import org.elasticsearch.ingest.useragent.IngestUserAgentPlugin;
-import org.elasticsearch.painless.PainlessPlugin;
-import org.elasticsearch.painless.PainlessScriptEngine;
-import org.elasticsearch.painless.spi.PainlessExtension;
-import org.elasticsearch.plugins.ExtensiblePlugin;
-import org.elasticsearch.plugins.IngestPlugin;
-import org.elasticsearch.script.IngestConditionalScript;
-import org.elasticsearch.script.IngestScript;
-import org.elasticsearch.script.ScriptEngine;
-import org.elasticsearch.script.ScriptModule;
-import org.elasticsearch.script.ScriptService;
-import org.elasticsearch.script.mustache.MustacheScriptEngine;
-import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.xpack.constantkeyword.ConstantKeywordPainlessExtension;
-import org.elasticsearch.xpack.spatial.SpatialPainlessExtension;
-import org.elasticsearch.xpack.wildcard.WildcardPainlessExtension;
+import org.elasticsearch.logstashbridge.common.SettingsBridge;
+import org.elasticsearch.logstashbridge.core.IOUtilsBridge;
+import org.elasticsearch.logstashbridge.env.EnvironmentBridge;
+import org.elasticsearch.logstashbridge.ingest.ProcessorBridge;
+import org.elasticsearch.logstashbridge.plugins.IngestPluginBridge;
+import org.elasticsearch.logstashbridge.script.ScriptServiceBridge;
+import org.elasticsearch.logstashbridge.threadpool.ThreadPoolBridge;
 
 import java.io.Closeable;
-import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -91,7 +70,7 @@ public static EventProcessorBuilder fromElasticsearch(final RestClient elasticse
     }
 
     public EventProcessorBuilder() {
-        this.addProcessorsFromPlugin(IngestCommonPlugin::new, Set.of(
+        this.addProcessorsFromPlugin(() -> IngestPluginBridge.wrap(new org.elasticsearch.ingest.common.IngestCommonPlugin()), Set.of(
                 org.elasticsearch.ingest.common.AppendProcessor.TYPE,
                 org.elasticsearch.ingest.common.BytesProcessor.TYPE,
                 org.elasticsearch.ingest.common.CommunityIdProcessor.TYPE,
@@ -127,7 +106,7 @@ public EventProcessorBuilder() {
                 org.elasticsearch.ingest.common.URLDecodeProcessor.TYPE,
                 org.elasticsearch.ingest.common.UppercaseProcessor.TYPE,
                 org.elasticsearch.ingest.common.UriPartsProcessor.TYPE));
-        this.addProcessorsFromPlugin(IngestUserAgentPlugin::new);
+        this.addProcessorsFromPlugin(() -> IngestPluginBridge.wrap(new org.elasticsearch.ingest.useragent.IngestUserAgentPlugin()));
         this.addProcessorsFromPlugin(RedactPlugin::new);
         this.addProcessor(SetSecurityUserProcessor.TYPE, SetSecurityUserProcessor.Factory::new);
     }
@@ -150,7 +129,7 @@ public EventProcessorBuilder() {
     // filer match listener
     private FilterMatchListener filterMatchListener;
 
-    private final List<Supplier<IngestPlugin>> ingestPlugins = new ArrayList<>();
+    private final List<Supplier<IngestPluginBridge>> ingestPlugins = new ArrayList<>();
 
     public synchronized EventProcessorBuilder setPipelineConfigurationResolver(final PipelineConfigurationResolver pipelineConfigurationResolver) {
         if (Objects.nonNull(this.pipelineConfigurationResolver)) {
@@ -215,15 +194,15 @@ private synchronized EventProcessorBuilder setFilterMatchListener(final FilterMa
         return this;
     }
 
-    public EventProcessorBuilder addProcessor(final String type, final Supplier<Processor.Factory> processorFactorySupplier) {
+    public EventProcessorBuilder addProcessor(final String type, final Supplier<ProcessorBridge.Factory> processorFactorySupplier) {
         return this.addProcessorsFromPlugin(SingleProcessorIngestPlugin.of(type, processorFactorySupplier));
     }
 
-    public EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPlugin> pluginSupplier, Set<String> requiredProcessors) {
+    public EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPluginBridge> pluginSupplier, Set<String> requiredProcessors) {
         return this.addProcessorsFromPlugin(safeSubset(pluginSupplier, requiredProcessors));
     }
 
-    public synchronized EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPlugin> pluginSupplier) {
+    public synchronized EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPluginBridge> pluginSupplier) {
         this.ingestPlugins.add(pluginSupplier);
         return this;
     }
@@ -233,7 +212,7 @@ public synchronized EventProcessor build(final PluginContext pluginContext) {
         Objects.requireNonNull(this.eventToIndexNameResolver, "event index name resolver is REQUIRED");
         Objects.requireNonNull(this.indexNameToPipelineNameResolver, "pipeline name resolver is REQUIRED");
 
-        final Settings settings = Settings.builder()
+        final SettingsBridge settings = SettingsBridge.builder()
                 .put("path.home", "/")
                 .put("node.name", "logstash.filter.elastic_integration." + pluginContext.pluginId())
                 .put("ingest.grok.watchdog.interval", "1s")
@@ -245,33 +224,22 @@ public synchronized EventProcessor build(final PluginContext pluginContext) {
         try {
             final ArrayList<Service> services = new ArrayList<>();
 
-            final ThreadPool threadPool = LogstashInternalBridge.createThreadPool(settings);
-            resourcesToClose.add(() -> ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS));
+            final ThreadPoolBridge threadPool = new ThreadPoolBridge(settings);
+            resourcesToClose.add(() -> ThreadPoolBridge.terminate(threadPool, 10, TimeUnit.SECONDS));
 
-            final ScriptService scriptService = initScriptService(settings, threadPool);
+            final ScriptServiceBridge scriptService = new ScriptServiceBridge(settings, threadPool::absoluteTimeInMillis);
             resourcesToClose.add(scriptService);
 
-            final Environment env = new Environment(settings, null);
-            final Processor.Parameters processorParameters = new Processor.Parameters(
-                    env,
-                    scriptService,
-                    null,
-                    threadPool.getThreadContext(),
-                    threadPool::relativeTimeInMillis,
-                    (delay, command) -> threadPool.schedule(command, TimeValue.timeValueMillis(delay), threadPool.generic()),
-                    null,
-                    null,
-                    threadPool.generic()::execute,
-                    IngestService.createGrokThreadWatchdog(env, threadPool)
-            );
+            final EnvironmentBridge env = new EnvironmentBridge(settings, null);
+            final ProcessorBridge.Parameters processorParameters = new ProcessorBridge.Parameters(env, scriptService, threadPool);
 
             IngestPipelineFactory ingestPipelineFactory = new IngestPipelineFactory(scriptService);
-            for (Supplier<IngestPlugin> ingestPluginSupplier : ingestPlugins) {
-                final IngestPlugin ingestPlugin = ingestPluginSupplier.get();
+            for (Supplier<IngestPluginBridge> ingestPluginSupplier : ingestPlugins) {
+                final IngestPluginBridge ingestPlugin = ingestPluginSupplier.get();
                 if (ingestPlugin instanceof Closeable closeableIngestPlugin) {
                     resourcesToClose.add(closeableIngestPlugin);
                 }
-                final Map<String, Processor.Factory> processorFactories = ingestPlugin.getProcessors(processorParameters);
+                final Map<String, ProcessorBridge.Factory> processorFactories = ingestPlugin.getProcessors(processorParameters);
                 ingestPipelineFactory = ingestPipelineFactory.withProcessors(processorFactories);
             }
 
@@ -309,53 +277,8 @@ public synchronized EventProcessor build(final PluginContext pluginContext) {
                                       indexNameToPipelineNameResolver,
                                       resourcesToClose);
         } catch (Exception e) {
-            IOUtils.closeWhileHandlingException(resourcesToClose);
+            IOUtilsBridge.closeWhileHandlingException(resourcesToClose);
             throw Exceptions.wrap(e, "Failed to build EventProcessor");
         }
     }
-
-    private static ScriptService initScriptService(final Settings settings, final ThreadPool threadPool) throws IOException {
-        Map<String, ScriptEngine> engines = new HashMap<>();
-        engines.put(PainlessScriptEngine.NAME, getPainlessScriptEngine(settings));
-        engines.put(MustacheScriptEngine.NAME, new MustacheScriptEngine(settings));
-
-        return new ScriptService(
-                settings,
-                engines,
-                ScriptModule.CORE_CONTEXTS,
-                threadPool::absoluteTimeInMillis,
-                new PluginProjectResolver());
-    }
-
-    /**
-     * @param settings the Elasticsearch settings object
-     * @return a {@link ScriptEngine} for painless scripts for use in {@link IngestScript} and
-     *         {@link IngestConditionalScript} contexts, including all available {@link PainlessExtension}s.
-     * @throws IOException when the underlying script engine cannot be created
-     */
-    private static ScriptEngine getPainlessScriptEngine(final Settings settings) throws IOException {
-        try (final PainlessPlugin painlessPlugin = new PainlessPlugin()) {
-
-            painlessPlugin.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
-                @Override
-                @SuppressWarnings("unchecked")
-                public <T> List<T> loadExtensions(Class<T> extensionPointType) {
-                    if (extensionPointType.isAssignableFrom(PainlessExtension.class)) {
-                        final List<PainlessExtension> extensions = new ArrayList<>();
-
-                        extensions.add(new ConstantKeywordPainlessExtension()); // module: constant-keyword
-                        extensions.add(new ProcessorsWhitelistExtension());     // module: ingest-common
-                        extensions.add(new SpatialPainlessExtension());         // module: spatial
-                        extensions.add(new WildcardPainlessExtension());        // module: wildcard
-
-                        return (List<T>) extensions;
-                    } else {
-                        return List.of();
-                    }
-                }
-            });
-
-            return painlessPlugin.getScriptEngine(settings, Set.of(IngestScript.CONTEXT, IngestConditionalScript.CONTEXT));
-        }
-    }
 }