Forward port #85000 and #84838 to 8.1 branch (#85040)

Forward port #85000 (Revert enrich cache lookup optimisation ) and #84838 (CompoundProcessor should also catch exceptions when executing a processor) to 8.1 branch.

Revert enrich cache lookup optimisation (#85028)

Forwardporting #85000 to master branch.
This PR reverts the optimisation that was added via #77259.
This optimisation cleverly ensures no duplicate searches happen if multiple threads concurrently execute the same search. However there are issues with the implementation that cause issues like #84781. The optimisation make use of CompletableFuture and in this case we don't check whether the result has completed exceptionally. Which causes the callback not being invoked and this leads to bulk request not being completed and hanging around.

The ingest framework due to its asynchronous nature is already complex and adding CompletableFuture into the mix makes debugging these issues very time consuming. This is the main reason why we like to revert this commit.

* CompoundProcessor should also catch exceptions when executing a processor (#84838) (#85035)

Currently, CompoundProcessor does not catch Exception and
if a processor throws an error and a method higher in the
call stack doesn't catch the exception then pipeline execution
stalls and bulk requests may not complete.

Usually these exceptions are caught by IngestService#executePipelines(...) method,
but when a processor executes async (for example: enrich processor) and the thread
that executes enrich is no longer the original write thread then there is no logic
that deals with failing pipeline execution and cleaning resources up. This then leads
to memory leaks.

Closes #84781

Also change how 'pipeline doesn't exist' error is thrown in TrackingResultProcessor.

With the change to CompoundProcessor thrown exceptions are caught and delegated to handler.
SimulateExecutionService in verbose mode ignores exceptions delegated to its handler, since
it assumes that processorResultList contains the result (successful or not successful) of every
processor in the pipeline.

In case TrackingResultProcessor for PipelineProcessor couldn't find the mentioned pipeline then
it just throws an error without updating the processorResultList. This commit addresses that.

Author: martijnvg

docs/changelog/84838.yaml

@@ -0,0 +1,6 @@
+pr: 84838
+summary: '`CompoundProcessor` should also catch exceptions when executing a processor'
+area: Ingest
+type: bug
+issues:
+ - 84781

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/90_simulate.yml

@@ -953,7 +953,6 @@ teardown:
 ---
 "Test simulate with provided pipeline that does not exist":
   - do:
-      catch: bad_request
       ingest.simulate:
         verbose: true
         body: >
@@ -974,5 +973,6 @@ teardown:
               }
             ]
           }
-  - match: { error.root_cause.0.type: "illegal_argument_exception" }
-  - match: { error.root_cause.0.reason: "Pipeline processor configured for non-existent pipeline [____pipeline_doesnot_exist___]" }
+  - match: { docs.0.processor_results.0.status: "error" }
+  - match: { docs.0.processor_results.0.error.root_cause.0.type: "illegal_argument_exception" }
+  - match: { docs.0.processor_results.0.error.root_cause.0.reason: "Pipeline processor configured for non-existent pipeline [____pipeline_doesnot_exist___]" }

server/src/main/java/org/elasticsearch/ingest/CompoundProcessor.java

@@ -133,30 +133,47 @@ void innerExecute(int currentProcessor, IngestDocument ingestDocument, BiConsume
         final IngestMetric metric = processorWithMetric.v2();
         final long startTimeInNanos = relativeTimeProvider.getAsLong();
         metric.preIngest();
-        processor.execute(ingestDocument, (result, e) -> {
-            long ingestTimeInNanos = relativeTimeProvider.getAsLong() - startTimeInNanos;
-            metric.postIngest(ingestTimeInNanos);
+        try {
+            processor.execute(ingestDocument, (result, e) -> {
+                long ingestTimeInNanos = relativeTimeProvider.getAsLong() - startTimeInNanos;
+                metric.postIngest(ingestTimeInNanos);
 
-            if (e != null) {
-                metric.ingestFailed();
-                if (ignoreFailure) {
-                    innerExecute(currentProcessor + 1, ingestDocument, handler);
+                if (e != null) {
+                    executeOnFailure(currentProcessor, ingestDocument, handler, processor, metric, e);
                 } else {
-                    IngestProcessorException compoundProcessorException = newCompoundProcessorException(e, processor, ingestDocument);
-                    if (onFailureProcessors.isEmpty()) {
-                        handler.accept(null, compoundProcessorException);
+                    if (result != null) {
+                        innerExecute(currentProcessor + 1, result, handler);
                     } else {
-                        executeOnFailureAsync(0, ingestDocument, compoundProcessorException, handler);
+                        handler.accept(null, null);
                     }
                 }
+            });
+        } catch (Exception e) {
+            long ingestTimeInNanos = relativeTimeProvider.getAsLong() - startTimeInNanos;
+            metric.postIngest(ingestTimeInNanos);
+            executeOnFailure(currentProcessor, ingestDocument, handler, processor, metric, e);
+        }
+    }
+
+    private void executeOnFailure(
+        int currentProcessor,
+        IngestDocument ingestDocument,
+        BiConsumer<IngestDocument, Exception> handler,
+        Processor processor,
+        IngestMetric metric,
+        Exception e
+    ) {
+        metric.ingestFailed();
+        if (ignoreFailure) {
+            innerExecute(currentProcessor + 1, ingestDocument, handler);
+        } else {
+            IngestProcessorException compoundProcessorException = newCompoundProcessorException(e, processor, ingestDocument);
+            if (onFailureProcessors.isEmpty()) {
+                handler.accept(null, compoundProcessorException);
             } else {
-                if (result != null) {
-                    innerExecute(currentProcessor + 1, result, handler);
-                } else {
-                    handler.accept(null, null);
-                }
+                executeOnFailureAsync(0, ingestDocument, compoundProcessorException, handler);
             }
-        });
+        }
     }
 
     void executeOnFailureAsync(

server/src/main/java/org/elasticsearch/ingest/TrackingResultProcessor.java

@@ -69,11 +69,23 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
             IngestDocument ingestDocumentCopy = new IngestDocument(ingestDocument);
             Pipeline pipelineToCall = pipelineProcessor.getPipeline(ingestDocument);
             if (pipelineToCall == null) {
-                throw new IllegalArgumentException(
+                IllegalArgumentException e = new IllegalArgumentException(
                     "Pipeline processor configured for non-existent pipeline ["
                         + pipelineProcessor.getPipelineToCallName(ingestDocument)
                         + ']'
                 );
+                // Add error as processor result, otherwise this gets lost in SimulateExecutionService#execute(...) and
+                // an empty response gets returned by the ingest simulate api.
+                processorResultList.add(
+                    new SimulateProcessorResult(
+                        pipelineProcessor.getType(),
+                        pipelineProcessor.getTag(),
+                        pipelineProcessor.getDescription(),
+                        e,
+                        conditionalWithResult
+                    )
+                );
+                throw e;
             }
             ingestDocumentCopy.executePipeline(pipelineToCall, (result, e) -> {
                 // special handling for pipeline cycle errors

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichCache.java

@@ -7,7 +7,6 @@
 
 package org.elasticsearch.xpack.enrich;
 
-import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
@@ -24,11 +23,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.function.BiConsumer;
-
-import static org.elasticsearch.action.ActionListener.wrap;
 
 /**
  * A simple cache for enrich that uses {@link Cache}. There is one instance of this cache and
@@ -48,24 +42,32 @@
  * current enrich index the enrich alias of an policy refers to. It would require checking
  * all cached entries on each cluster state update)
  */
-public class EnrichCache {
+public final class EnrichCache {
 
-    protected final Cache<CacheKey, CompletableFuture<List<Map<?, ?>>>> cache;
+    private final Cache<CacheKey, List<Map<?, ?>>> cache;
     private volatile Metadata metadata;
 
     EnrichCache(long maxSize) {
-        this.cache = CacheBuilder.<CacheKey, CompletableFuture<List<Map<?, ?>>>>builder().setMaximumWeight(maxSize).build();
+        this.cache = CacheBuilder.<CacheKey, List<Map<?, ?>>>builder().setMaximumWeight(maxSize).build();
     }
 
-    /**
-     * Get the value from the cache if present. Returns immediately.
-     * See {@link #resolveOrDispatchSearch(SearchRequest, BiConsumer, BiConsumer)} to implement a read-through, possibly async interaction.
-     * @param searchRequest the key
-     * @return the cached value or null
-     */
-    CompletableFuture<List<Map<?, ?>>> get(SearchRequest searchRequest) {
-        CacheKey cacheKey = toKey(searchRequest);
-        return cache.get(cacheKey);
+    List<Map<?, ?>> get(SearchRequest searchRequest) {
+        String enrichIndex = getEnrichIndexKey(searchRequest);
+        CacheKey cacheKey = new CacheKey(enrichIndex, searchRequest);
+
+        List<Map<?, ?>> response = cache.get(cacheKey);
+        if (response != null) {
+            return deepCopy(response, false);
+        } else {
+            return null;
+        }
+    }
+
+    void put(SearchRequest searchRequest, List<Map<?, ?>> response) {
+        String enrichIndex = getEnrichIndexKey(searchRequest);
+        CacheKey cacheKey = new CacheKey(enrichIndex, searchRequest);
+
+        cache.put(cacheKey, response);
     }
 
     void setMetadata(Metadata metadata) {
@@ -83,61 +85,13 @@ public EnrichStatsAction.Response.CacheStats getStats(String localNodeId) {
         );
     }
 
-    /**
-     * resolves the entry from the cache and provides reports the result to the `callBack` This method does not dispatch any logic
-     * to another thread. Under contention the searchDispatcher is only called once when the value is not in the cache. The
-     * searchDispatcher should schedule the search / callback _asynchronously_ because if the searchDispatcher blocks, then this
-     * method will block. The callback is call on the thread calling this method or under cache miss and contention, the thread running
-     * the part of the searchDispatcher that calls the callback.
-     * @param searchRequest the cache key and input for the search dispatcher
-     * @param searchDispatcher the logical block to be called on cache miss
-     * @param callBack the callback which gets the value asynchronously, which could be a searchResponse or exception (negative lookup)
-     */
-    public void resolveOrDispatchSearch(
-        SearchRequest searchRequest,
-        BiConsumer<SearchRequest, ActionListener<SearchResponse>> searchDispatcher,
-        BiConsumer<List<Map<?, ?>>, Exception> callBack
-    ) {
-        CacheKey cacheKey = toKey(searchRequest);
-        try {
-            CompletableFuture<List<Map<?, ?>>> cacheEntry = cache.computeIfAbsent(cacheKey, request -> {
-                CompletableFuture<List<Map<?, ?>>> completableFuture = new CompletableFuture<>();
-                searchDispatcher.accept(
-                    searchRequest,
-                    wrap(response -> completableFuture.complete(toCacheValue(response)), completableFuture::completeExceptionally)
-                );
-                return completableFuture;
-            });
-            cacheEntry.whenComplete((response, throwable) -> {
-                if (throwable != null) {
-                    // Don't cache failures
-                    cache.invalidate(cacheKey, cacheEntry);
-                    if (throwable instanceof Exception e) {
-                        callBack.accept(null, e);
-                        return;
-                    }
-                    // Let ElasticsearchUncaughtExceptionHandler handle this, which should halt Elasticsearch
-                    throw (Error) throwable;
-                }
-                callBack.accept(deepCopy(response, false), null);
-            });
-        } catch (ExecutionException e) {
-            callBack.accept(null, e);
-        }
-    }
-
-    protected CacheKey toKey(SearchRequest searchRequest) {
-        String enrichIndex = getEnrichIndexKey(searchRequest);
-        return new CacheKey(enrichIndex, searchRequest);
-    }
-
     private String getEnrichIndexKey(SearchRequest searchRequest) {
         String alias = searchRequest.indices()[0];
         IndexAbstraction ia = metadata.getIndicesLookup().get(alias);
         return ia.getIndices().get(0).getName();
     }
 
-    private List<Map<?, ?>> toCacheValue(SearchResponse response) {
+    List<Map<?, ?>> toCacheValue(SearchResponse response) {
         List<Map<?, ?>> result = new ArrayList<>(response.getHits().getHits().length);
         for (SearchHit hit : response.getHits()) {
             result.add(deepCopy(hit.getSourceAsMap(), true));

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichProcessorFactory.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.enrich;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
@@ -128,10 +129,18 @@ public void accept(ClusterState state) {
         EnrichCache enrichCache
     ) {
         Client originClient = new OriginSettingClient(client, ENRICH_ORIGIN);
-        return (req, handler) -> enrichCache.resolveOrDispatchSearch(
-            req,
-            (searchRequest, listener) -> originClient.execute(EnrichCoordinatorProxyAction.INSTANCE, searchRequest, listener),
-            handler
-        );
+        return (req, handler) -> {
+            // intentionally non-locking for simplicity...it's OK if we re-put the same key/value in the cache during a race condition.
+            List<Map<?, ?>> response = enrichCache.get(req);
+            if (response != null) {
+                handler.accept(response, null);
+            } else {
+                originClient.execute(EnrichCoordinatorProxyAction.INSTANCE, req, ActionListener.wrap(resp -> {
+                    List<Map<?, ?>> value = enrichCache.toCacheValue(resp);
+                    enrichCache.put(req, value);
+                    handler.accept(EnrichCache.deepCopy(value, false), null);
+                }, e -> { handler.accept(null, e); }));
+            }
+        };
     }
 }

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/BasicEnrichTests.java

@@ -371,6 +371,39 @@ public void testTemplating() throws Exception {
         assertThat(keys.contains(userEntry.get(MATCH_FIELD)), is(true));
     }
 
+    public void testFailureAfterEnrich() throws Exception {
+        List<String> keys = createSourceMatchIndex(1, 1);
+        String policyName = "my-policy";
+        EnrichPolicy enrichPolicy = new EnrichPolicy(
+            EnrichPolicy.MATCH_TYPE,
+            null,
+            List.of(SOURCE_INDEX_NAME),
+            MATCH_FIELD,
+            Arrays.asList(DECORATE_FIELDS)
+        );
+        PutEnrichPolicyAction.Request request = new PutEnrichPolicyAction.Request(policyName, enrichPolicy);
+        client().execute(PutEnrichPolicyAction.INSTANCE, request).actionGet();
+        client().execute(ExecuteEnrichPolicyAction.INSTANCE, new ExecuteEnrichPolicyAction.Request(policyName)).actionGet();
+
+        // A pipeline with a foreach that uses a non existing field that is specified after enrich has run:
+        String pipelineName = "my-pipeline";
+        String pipelineBody = "{\"processors\": [{\"enrich\": {\"policy_name\":\""
+            + policyName
+            + "\", \"field\": \"email\", \"target_field\": \"users\"}},"
+            + "{ \"foreach\": {\"field\":\"users\", \"processor\":{\"append\":{\"field\":\"matched2\",\"value\":\"{{_ingest._value}}\"}}}}"
+            + "]}";
+        PutPipelineRequest putPipelineRequest = new PutPipelineRequest(pipelineName, new BytesArray(pipelineBody), XContentType.JSON);
+        client().admin().cluster().putPipeline(putPipelineRequest).actionGet();
+
+        for (int i = 0; i < 5; i++) {
+            IndexRequest indexRequest = new IndexRequest("my-index").id("1")
+                .setPipeline(pipelineName)
+                .source(Map.of(MATCH_FIELD, "non_existing"));
+            Exception e = expectThrows(IllegalArgumentException.class, () -> client().index(indexRequest).actionGet());
+            assertThat(e.getMessage(), equalTo("field [users] not present as part of path [users]"));
+        }
+    }
+
     private List<String> createSourceMatchIndex(int numKeys, int numDocsPerKey) {
         Set<String> keys = new HashSet<>();
         for (int id = 0; id < numKeys; id++) {