[7.9] Handle error conditions when simulating ingest pipelines with verbosity enabled (#63518)

Author: danhermann

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

@@ -417,6 +417,58 @@ teardown:
   - is_true: docs.1.processor_results.1.doc._ingest.timestamp
   - is_true: docs.1.processor_results.1.doc._ingest.pipeline
 
+---
+"Test verbose simulate with error in pipeline":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "rename" : {
+                  "field" : "does_not_exist",
+                  "target_field" : "_value"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      ingest.simulate:
+        verbose: true
+        body: >
+          {
+            "pipeline": {
+              "description": "_description",
+              "processors": [
+                {
+                  "pipeline" : {
+                    "name" : "my_pipeline"
+                  }
+                }
+              ]
+            },
+            "docs": [
+              {
+                "_index": "index",
+                "_id": "id",
+                "_source": {
+                  "foo": "bar",
+                  "bar": "hello"
+                }
+              }
+            ]
+          }
+  - length: { docs: 1 }
+  - length: { docs.0.processor_results: 1 }
+  - 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: "field [does_not_exist] doesn't exist" }
+  - match: { docs.0.processor_results.0.error.type: "illegal_argument_exception" }
+  - match: { docs.0.processor_results.0.error.reason: "field [does_not_exist] doesn't exist" }
+
 ---
 "Test verbose simulate with on_failure":
   - do:

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

@@ -51,6 +51,7 @@
 public final class IngestDocument {
 
     public static final String INGEST_KEY = "_ingest";
+    public static final String PIPELINE_CYCLE_ERROR_MESSAGE = "Cycle detected for pipeline: ";
     private static final String INGEST_KEY_PREFIX = INGEST_KEY + ".";
     private static final String SOURCE_PREFIX = SourceFieldMapper.NAME + ".";
 
@@ -685,7 +686,7 @@ public void executePipeline(Pipeline pipeline, BiConsumer<IngestDocument, Except
                 handler.accept(result, e);
             });
         } else {
-            handler.accept(null, new IllegalStateException("Cycle detected for pipeline: " + pipeline.getId()));
+            handler.accept(null, new IllegalStateException(PIPELINE_CYCLE_ERROR_MESSAGE + pipeline.getId()));
         }
     }
 

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

@@ -26,6 +26,8 @@
 import java.util.List;
 import java.util.function.BiConsumer;
 
+import static org.elasticsearch.ingest.IngestDocument.PIPELINE_CYCLE_ERROR_MESSAGE;
+
 /**
  * Processor to be used within Simulate API to keep track of processors executed in pipeline.
  */
@@ -59,19 +61,17 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
             //runtime check for cycles against a copy of the document. This is needed to properly handle conditionals around pipelines
             IngestDocument ingestDocumentCopy = new IngestDocument(ingestDocument);
             ingestDocumentCopy.executePipeline(pipelineProcessor.getPipeline(ingestDocument), (result, e) -> {
-                // do nothing, let the tracking processors throw the exception while recording the path up to the failure
-                if (e instanceof ElasticsearchException) {
-                    ElasticsearchException elasticsearchException = (ElasticsearchException) e;
-                    //else do nothing, let the tracking processors throw the exception while recording the path up to the failure
-                    if (elasticsearchException.getCause() instanceof IllegalStateException) {
-                        if (ignoreFailure) {
-                            processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
-                                new IngestDocument(ingestDocument), e));
-                        } else {
-                            processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(), e));
-                        }
-                        handler.accept(null, elasticsearchException);
+                // special handling for pipeline cycle errors
+                if (e instanceof ElasticsearchException &&
+                    e.getCause() instanceof IllegalStateException &&
+                    e.getCause().getMessage().startsWith(PIPELINE_CYCLE_ERROR_MESSAGE)) {
+                    if (ignoreFailure) {
+                        processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
+                            new IngestDocument(ingestDocument), e));
+                    } else {
+                        processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(), e));
                     }
+                    handler.accept(null, e);
                 } else {
                     //now that we know that there are no cycles between pipelines, decorate the processors for this pipeline and execute it
                     CompoundProcessor verbosePipelineProcessor = decorate(pipeline.getCompoundProcessor(), null, processorResultList);

server/src/test/java/org/elasticsearch/ingest/TrackingResultProcessorTests.java

@@ -476,6 +476,40 @@ pipelineId, null, null, new CompoundProcessor(
         assertThat(resultList.get(3).getProcessorTag(), nullValue());
     }
 
+    public void testActualPipelineProcessorWithUnhandledFailure() throws Exception {
+        String pipelineId = "pipeline1";
+        IngestService ingestService = createIngestService();
+        Map<String, Object> pipelineConfig = new HashMap<>();
+        pipelineConfig.put("name", pipelineId);
+        PipelineProcessor.Factory factory = new PipelineProcessor.Factory(ingestService);
+
+        String key1 = randomAlphaOfLength(10);
+        IllegalStateException exception = new IllegalStateException("Not a pipeline cycle error");
+
+        Pipeline pipeline = new Pipeline(
+            pipelineId, null, null, new CompoundProcessor(
+            new TestProcessor(ingestDocument -> ingestDocument.setFieldValue(key1, randomInt())),
+            new TestProcessor(ingestDocument -> { throw exception; }))
+        );
+        when(ingestService.getPipeline(pipelineId)).thenReturn(pipeline);
+
+        PipelineProcessor pipelineProcessor = factory.create(Collections.emptyMap(), null, null, pipelineConfig);
+        CompoundProcessor actualProcessor = new CompoundProcessor(pipelineProcessor);
+
+        CompoundProcessor trackingProcessor = decorate(actualProcessor, null, resultList);
+
+        trackingProcessor.execute(ingestDocument, (result, e) -> {});
+
+        SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
+        expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);
+
+        verify(ingestService, Mockito.atLeast(1)).getPipeline(pipelineId);
+
+        assertThat(resultList.size(), equalTo(2));
+        assertTrue(resultList.get(0).getIngestDocument().hasField(key1));
+        assertThat(resultList.get(1).getFailure(), equalTo(exception));
+    }
+
     public void testActualPipelineProcessorWithCycle() throws Exception {
         String pipelineId1 = "pipeline1";
         String pipelineId2 = "pipeline2";