Replace ES classes with their Bridge Stable API pairs.

Author: mashhurs

build.gradle

@@ -257,6 +257,45 @@ task buildElasticsearchLocalDistro(dependsOn: unzipDownloadedElasticsearchSource
     }
 }
 
+task buildElasticsearchLogstashBridge(type: Exec) {
+    description "builds logstash-bridge lib module"
+
+    dependsOn buildElasticsearchLocalDistro
+    mustRunAfter buildElasticsearchLocalDistro
+
+    def logFile = project.file("${buildDir}/logstash-bridge-build.log")
+    doFirst {
+        def funneler = new OutputStreamFunneler(new LazyFileOutputStream(logFile))
+        standardOutput = funneler.funnelInstance
+        errorOutput = funneler.funnelInstance
+    }
+
+    def esSource = "${buildDir}/elasticsearch-source/"
+    def esBuildDir = "${esSource}/build"
+
+    inputs.dir esSource
+    outputs.dir "${esBuildDir}/libs/logstash-bridge"
+
+    ext.buildRoot = esBuildDir
+    workingDir esSource
+    commandLine "./gradlew", ":lib:logstash-bridge:build"
+
+    ignoreExitValue true // handled in doLast
+    doLast {
+        def exitValue = executionResult.get().exitValue
+        if (exitValue != 0) {
+            if (logFile.exists()) {
+                println "\n===== Elasticsearch logstash-bridge build Log ====="
+                println logFile.text
+                println "===== End of Elasticsearch logstash-bridge build Log =====\n"
+            } else {
+                "Elasticsearch logstash-bridge build failed and ${logFile.path} log does not exist"
+            }
+            throw new GradleException("Elasticsearch logstash-bridge build failed, see the logs for details.")
+        }
+    }
+}
+
 task shadeElasticsearchIngestGeoIpModule(type: com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar) {
     description "Shades embedded dependencies of the Elasticsearch Ingest GeoIP module"
 
@@ -316,13 +355,30 @@ task shadeElasticsearchRedactPlugin(type: com.github.jengelman.gradle.plugins.sh
     exclude '**/module-info.class'
 }
 
+task shadeElasticsearchLogstashBridge(type: com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar) {
+    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'
+}
+
 task importMinimalElasticsearch() {
     description "Imports minimal portions of Elasticsearch localDistro"
 
     dependsOn buildElasticsearchLocalDistro
+    dependsOn buildElasticsearchLogstashBridge
     dependsOn shadeElasticsearchIngestGeoIpModule
     dependsOn shadeElasticsearchGrokImplementation
     dependsOn shadeElasticsearchRedactPlugin
+    dependsOn shadeElasticsearchLogstashBridge
 
     ext.jars = "${buildDir}/elasticsearch-minimal-jars"
 
@@ -341,6 +397,7 @@ task importMinimalElasticsearch() {
                 include jarPackageNamed("lucene-core")
                 include jarPackageNamed("lucene-analysis-common")
             }
+            from(shadeElasticsearchLogstashBridge)
             from(shadeElasticsearchGrokImplementation)
             from(buildElasticsearchLocalDistro.module("x-pack-core"))
 

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