ingest: better support for conditionals with simulate?verbose

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

@@ -641,7 +641,6 @@ teardown:
 - match: { acknowledged: true }
 
 - do:
-    catch: /illegal_state_exception/
     ingest.simulate:
       verbose: true
       body: >
@@ -667,8 +666,10 @@ teardown:
           }
           ]
         }
-- match: { error.root_cause.0.type: "illegal_state_exception" }
-- match: { error.root_cause.0.reason: "Cycle detected for pipeline: inner" }
+- length: { docs: 1 }
+- length: { docs.0.processor_results: 1 }
+- match: { docs.0.processor_results.0.error.reason: "java.lang.IllegalArgumentException: java.lang.IllegalStateException: Cycle detected for pipeline: outer" }
+- match: { docs.0.processor_results.0.error.caused_by.caused_by.reason: "Cycle detected for pipeline: outer" }
 
 ---
 "Test verbose simulate with Pipeline Processor with Multiple Pipelines":

server/src/main/java/org/elasticsearch/action/ingest/SimulateExecutionService.java

@@ -21,17 +21,13 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
+import org.elasticsearch.ingest.CompoundProcessor;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Pipeline;
-import org.elasticsearch.ingest.CompoundProcessor;
-import org.elasticsearch.ingest.PipelineProcessor;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.IdentityHashMap;
 import java.util.List;
-import java.util.Set;
 
 import static org.elasticsearch.ingest.TrackingResultProcessor.decorate;
 
@@ -46,11 +42,9 @@ class SimulateExecutionService {
     }
 
     SimulateDocumentResult executeDocument(Pipeline pipeline, IngestDocument ingestDocument, boolean verbose) {
-        // Prevent cycles in pipeline decoration
-        final Set<PipelineProcessor> pipelinesSeen = Collections.newSetFromMap(new IdentityHashMap<>());
         if (verbose) {
             List<SimulateProcessorResult> processorResultList = new ArrayList<>();
-            CompoundProcessor verbosePipelineProcessor = decorate(pipeline.getCompoundProcessor(), processorResultList, pipelinesSeen);
+            CompoundProcessor verbosePipelineProcessor = decorate(pipeline.getCompoundProcessor(), processorResultList);
             try {
                 Pipeline verbosePipeline = new Pipeline(pipeline.getId(), pipeline.getDescription(), pipeline.getVersion(),
                     verbosePipelineProcessor);

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

@@ -62,10 +62,7 @@ public class ConditionalProcessor extends AbstractProcessor {
 
     @Override
     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
-        IngestConditionalScript script =
-            scriptService.compile(condition, IngestConditionalScript.CONTEXT).newInstance(condition.getParams());
-        if (script.execute(new UnmodifiableIngestData(ingestDocument.getSourceAndMetadata()))) {
-            // Only record metric if the script evaluates to true
+        if (evaluate(ingestDocument)) {
             long startTimeInNanos = relativeTimeProvider.getAsLong();
             try {
                 metric.preIngest();
@@ -81,6 +78,12 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
         return ingestDocument;
     }
 
+    boolean evaluate(IngestDocument ingestDocument) {
+        IngestConditionalScript script =
+            scriptService.compile(condition, IngestConditionalScript.CONTEXT).newInstance(condition.getParams());
+        return script.execute(new UnmodifiableIngestData(ingestDocument.getSourceAndMetadata()));
+    }
+
     Processor getProcessor() {
         return processor;
     }

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

@@ -19,11 +19,11 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ingest.SimulateProcessorResult;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Processor to be used within Simulate API to keep track of processors executed in pipeline.
@@ -42,14 +42,46 @@ public final class TrackingResultProcessor implements Processor {
 
     @Override
     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
+        Processor processor = actualProcessor;
         try {
-            actualProcessor.execute(ingestDocument);
-            processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument)));
+            if (processor instanceof ConditionalProcessor) {
+                ConditionalProcessor conditionalProcessor = (ConditionalProcessor) processor;
+                if (conditionalProcessor.evaluate(ingestDocument) == false) {
+                    return ingestDocument;
+                }
+                if (conditionalProcessor.getProcessor() instanceof PipelineProcessor) {
+                    processor = conditionalProcessor.getProcessor();
+                }
+            }
+            if (processor instanceof PipelineProcessor) {
+                PipelineProcessor pipelineProcessor = ((PipelineProcessor) processor);
+                Pipeline pipeline = pipelineProcessor.getPipeline();
+                //runtime check for cycles against a copy of the document. This is needed to properly handle conditionals around pipelines
+                try {
+                    IngestDocument ingestDocumentCopy = new IngestDocument(ingestDocument);
+                    ingestDocumentCopy.executePipeline(pipelineProcessor.getPipeline());
+                } catch (ElasticsearchException elasticsearchException) {
+                    if (elasticsearchException.getCause().getCause() instanceof IllegalStateException) {
+                        throw elasticsearchException;
+                    }
+                    //else do nothing, let the tracking processors throw the exception while recording the path up to the failure
+                } catch (Exception e) {
+                    // do nothing, let the tracking processors throw the exception while recording the path up to the failure
+                }
+                //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(), processorResultList);
+                Pipeline verbosePipeline = new Pipeline(pipeline.getId(), pipeline.getDescription(), pipeline.getVersion(),
+                    verbosePipelineProcessor);
+                ingestDocument.executePipeline(verbosePipeline);
+            } else {
+                processor.execute(ingestDocument);
+                processorResultList.add(new SimulateProcessorResult(processor.getTag(), new IngestDocument(ingestDocument)));
+            }
         } catch (Exception e) {
             if (ignoreFailure) {
-                processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument), e));
+                processorResultList.add(new SimulateProcessorResult(processor.getTag(), new IngestDocument(ingestDocument), e));
             } else {
-                processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), e));
+                processorResultList.add(new SimulateProcessorResult(processor.getTag(), e));
             }
             throw e;
         }
@@ -66,35 +98,19 @@ public String getTag() {
         return actualProcessor.getTag();
     }
 
-    public static CompoundProcessor decorate(CompoundProcessor compoundProcessor, List<SimulateProcessorResult> processorResultList,
-                                             Set<PipelineProcessor> pipelinesSeen) {
+    public static CompoundProcessor decorate(CompoundProcessor compoundProcessor, List<SimulateProcessorResult> processorResultList) {
         List<Processor> processors = new ArrayList<>(compoundProcessor.getProcessors().size());
         for (Processor processor : compoundProcessor.getProcessors()) {
-            if (processor instanceof PipelineProcessor) {
-                PipelineProcessor pipelineProcessor = ((PipelineProcessor) processor);
-                if (pipelinesSeen.add(pipelineProcessor) == false) {
-                    throw new IllegalStateException("Cycle detected for pipeline: " + pipelineProcessor.getPipeline().getId());
-                }
-                processors.add(decorate(pipelineProcessor.getPipeline().getCompoundProcessor(), processorResultList, pipelinesSeen));
-                pipelinesSeen.remove(pipelineProcessor);
-            } else if (processor instanceof CompoundProcessor) {
-                processors.add(decorate((CompoundProcessor) processor, processorResultList, pipelinesSeen));
+            if (processor instanceof CompoundProcessor) {
+                processors.add(decorate((CompoundProcessor) processor, processorResultList));
             } else {
                 processors.add(new TrackingResultProcessor(compoundProcessor.isIgnoreFailure(), processor, processorResultList));
             }
         }
         List<Processor> onFailureProcessors = new ArrayList<>(compoundProcessor.getProcessors().size());
         for (Processor processor : compoundProcessor.getOnFailureProcessors()) {
-            if (processor instanceof PipelineProcessor) {
-                PipelineProcessor pipelineProcessor = ((PipelineProcessor) processor);
-                if (pipelinesSeen.add(pipelineProcessor) == false) {
-                    throw new IllegalStateException("Cycle detected for pipeline: " + pipelineProcessor.getPipeline().getId());
-                }
-                onFailureProcessors.add(decorate(pipelineProcessor.getPipeline().getCompoundProcessor(), processorResultList,
-                    pipelinesSeen));
-                pipelinesSeen.remove(pipelineProcessor);
-            } else if (processor instanceof CompoundProcessor) {
-                onFailureProcessors.add(decorate((CompoundProcessor) processor, processorResultList, pipelinesSeen));
+            if (processor instanceof CompoundProcessor) {
+                onFailureProcessors.add(decorate((CompoundProcessor) processor, processorResultList));
             } else {
                 onFailureProcessors.add(new TrackingResultProcessor(compoundProcessor.isIgnoreFailure(), processor, processorResultList));
             }