6.x Ingest node back ports

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.grok.Grok;
 import org.elasticsearch.grok.ThreadWatchdog;
+import org.elasticsearch.ingest.PipelineProcessor;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.IngestPlugin;

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/210_pipeline_processor.yml

@@ -110,4 +110,4 @@ teardown:
       pipeline: "outer"
       body: {}
 - match: { error.root_cause.0.type: "exception" }
-- match: { error.root_cause.0.reason: "java.lang.IllegalArgumentException: java.lang.IllegalStateException: Recursive invocation of pipeline [inner] detected." }
+- match: { error.root_cause.0.reason: "java.lang.IllegalArgumentException: java.lang.IllegalStateException: Cycle detected for pipeline: inner" }

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

@@ -605,3 +605,150 @@ teardown:
   - length: { docs.0.processor_results.1: 2 }
   - match: { docs.0.processor_results.1.tag: "rename-1" }
   - match: { docs.0.processor_results.1.doc._source.new_status: 200 }
+
+---
+"Test verbose simulate with Pipeline Processor with Circular Pipelines":
+- do:
+    ingest.put_pipeline:
+      id: "outer"
+      body:  >
+        {
+          "description" : "outer pipeline",
+          "processors" : [
+            {
+              "pipeline" : {
+                "pipeline": "inner"
+              }
+            }
+          ]
+        }
+- match: { acknowledged: true }
+
+- do:
+    ingest.put_pipeline:
+      id: "inner"
+      body:  >
+        {
+          "description" : "inner pipeline",
+          "processors" : [
+            {
+              "pipeline" : {
+                "pipeline": "outer"
+              }
+            }
+          ]
+        }
+- match: { acknowledged: true }
+
+- do:
+    catch: /illegal_state_exception/
+    ingest.simulate:
+      verbose: true
+      body: >
+        {
+          "pipeline": {
+            "processors" : [
+            {
+              "pipeline" : {
+                "pipeline": "outer"
+              }
+            }
+            ]
+          }
+          ,
+          "docs": [
+          {
+            "_index": "index",
+            "_type": "type",
+            "_id": "id",
+            "_source": {
+              "field1": "123.42 400 <foo>"
+            }
+          }
+          ]
+        }
+- match: { error.root_cause.0.type: "illegal_state_exception" }
+- match: { error.root_cause.0.reason: "Cycle detected for pipeline: inner" }
+
+---
+"Test verbose simulate with Pipeline Processor with Multiple Pipelines":
+- do:
+    ingest.put_pipeline:
+      id: "pipeline1"
+      body:  >
+        {
+          "processors": [
+          {
+            "set": {
+              "field": "pipeline1",
+              "value": true
+            }
+          },
+          {
+            "pipeline": {
+              "pipeline": "pipeline2"
+            }
+          }
+          ]
+        }
+- match: { acknowledged: true }
+
+- do:
+    ingest.put_pipeline:
+      id: "pipeline2"
+      body:  >
+        {
+          "processors": [
+          {
+            "set": {
+              "field": "pipeline2",
+              "value": true
+            }
+          }
+          ]
+        }
+- match: { acknowledged: true }
+
+- do:
+    ingest.simulate:
+      verbose: true
+      body: >
+        {
+          "pipeline": {
+            "processors": [
+            {
+              "set": {
+                "field": "pipeline0",
+                "value": true
+              }
+            },
+            {
+              "pipeline": {
+                "pipeline": "pipeline1"
+              }
+            }
+            ]
+          },
+          "docs": [
+          {
+            "_index": "index",
+            "_type": "type",
+            "_id": "id",
+            "_source": {
+              "field1": "123.42 400 <foo>"
+            }
+          }
+          ]
+        }
+- length: { docs: 1 }
+- length: { docs.0.processor_results: 3 }
+- match: { docs.0.processor_results.0.doc._source.pipeline0: true }
+- is_false: docs.0.processor_results.0.doc._source.pipeline1
+- is_false: docs.0.processor_results.0.doc._source.pipeline2
+- match: { docs.0.processor_results.1.doc._source.pipeline0: true }
+- match: { docs.0.processor_results.1.doc._source.pipeline1: true }
+- is_false: docs.0.processor_results.1.doc._source.pipeline2
+- match: { docs.0.processor_results.2.doc._source.pipeline0: true }
+- match: { docs.0.processor_results.2.doc._source.pipeline1: true }
+- match: { docs.0.processor_results.2.doc._source.pipeline2: true }
+

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

@@ -24,12 +24,16 @@
 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.action.ingest.TrackingResultProcessor.decorate;
+import static org.elasticsearch.ingest.TrackingResultProcessor.decorate;
 
 class SimulateExecutionService {
 
@@ -42,11 +46,15 @@ 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);
+            CompoundProcessor verbosePipelineProcessor = decorate(pipeline.getCompoundProcessor(), processorResultList, pipelinesSeen);
             try {
-                verbosePipelineProcessor.execute(ingestDocument);
+                Pipeline verbosePipeline = new Pipeline(pipeline.getId(), pipeline.getDescription(), pipeline.getVersion(),
+                    verbosePipelineProcessor);
+                ingestDocument.executePipeline(verbosePipeline);
                 return new SimulateDocumentVerboseResult(processorResultList);
             } catch (Exception e) {
                 return new SimulateDocumentVerboseResult(processorResultList);

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

@@ -649,10 +649,14 @@ private static Object deepCopy(Object value) {
      * @throws Exception On exception in pipeline execution
      */
     public IngestDocument executePipeline(Pipeline pipeline) throws Exception {
-        if (this.executedPipelines.add(pipeline) == false) {
-            throw new IllegalStateException("Recursive invocation of pipeline [" + pipeline.getId() + "] detected.");
+        try {
+            if (this.executedPipelines.add(pipeline) == false) {
+                throw new IllegalStateException("Cycle detected for pipeline: " + pipeline.getId());
+            }
+            return pipeline.execute(this);
+        } finally {
+            executedPipelines.remove(pipeline);
         }
-        return pipeline.execute(this);
     }
 
     @Override

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

@@ -0,0 +1,95 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest;
+
+import org.elasticsearch.common.metrics.CounterMetric;
+import org.elasticsearch.common.metrics.MeanMetric;
+
+/**
+ * <p>Metrics to measure ingest actions.
+ * <p>This counts measure documents and timings for a given scope.
+ * The scope is determined by the calling code. For example you can use this class to count all documents across all pipeline,
+ * or you can use this class to count documents for a given pipeline or a specific processor.
+ * This class does not make assumptions about it's given scope.
+ */
+class IngestMetric {
+
+    /**
+     * The time it takes to complete the measured item.
+     */
+    private final MeanMetric ingestTime = new MeanMetric();
+    /**
+     * The current count of things being measure. Should most likely ever be 0 or 1.
+     * Useful when aggregating multiple metrics to see how many things are in flight.
+     */
+    private final CounterMetric ingestCurrent = new CounterMetric();
+    /**
+     * The ever increasing count of things being measured
+     */
+    private final CounterMetric ingestCount = new CounterMetric();
+    /**
+     * The only increasing count of failures
+     */
+    private final CounterMetric ingestFailed = new CounterMetric();
+
+    /**
+     * Call this prior to the ingest action.
+     */
+    void preIngest() {
+        ingestCurrent.inc();
+    }
+
+    /**
+     * Call this after the performing the ingest action, even if the action failed.
+     * @param ingestTimeInMillis The time it took to perform the action.
+     */
+    void postIngest(long ingestTimeInMillis) {
+        ingestCurrent.dec();
+        ingestTime.inc(ingestTimeInMillis);
+        ingestCount.inc();
+    }
+
+    /**
+     * Call this if the ingest action failed.
+     */
+    void ingestFailed() {
+        ingestFailed.inc();
+    }
+
+    /**
+     * <p>Add two sets of metrics together.
+     * <p><strong>Note -</strong> this method does <strong>not</strong> add the current count values.
+     * The current count value is ephemeral and requires a increase/decrease operation pairs to keep the value correct.
+     *
+     * @param metrics The metric to add.
+     */
+    void add(IngestMetric metrics) {
+        ingestCount.inc(metrics.ingestCount.count());
+        ingestTime.inc(metrics.ingestTime.sum());
+        ingestFailed.inc(metrics.ingestFailed.count());
+    }
+
+    /**
+     * Creates a serializable representation for these metrics.
+     */
+    IngestStats.Stats createStats() {
+        return new IngestStats.Stats(ingestCount.count(), ingestTime.sum(), ingestCurrent.count(), ingestFailed.count());
+    }
+}