ingest: correctly measure chained pipeline stats (#33912)

Prior to this change when a pipeline processor called another
pipeline, only the stats for the first processor were recorded.
The stats for the subsequent pipelines were ignored. This change
properly accounts for pipelines irregardless if they are the first
or subsequently called pipelines.

This change moves the state of the stats from the IngestService
to the pipeline itself. Cluster updates are safe since the pipelines
map is atomically swapped, and if a cluster update happens
while iterating over stats (now read directly from the pipeline)
a slightly stale view of stats may be shown.

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

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

@@ -23,16 +23,16 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
@@ -50,8 +50,6 @@
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.metrics.CounterMetric;
-import org.elasticsearch.common.metrics.MeanMetric;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
@@ -80,8 +78,7 @@ public class IngestService implements ClusterStateApplier {
     // are loaded, so in the cluster state we just save the pipeline config and here we keep the actual pipelines around.
     private volatile Map<String, Pipeline> pipelines = new HashMap<>();
     private final ThreadPool threadPool;
-    private final StatsHolder totalStats = new StatsHolder();
-    private volatile Map<String, StatsHolder> statsHolderPerPipeline = Collections.emptyMap();
+    private final IngestMetric totalMetrics = new IngestMetric();
 
     public IngestService(ClusterService clusterService, ThreadPool threadPool,
                          Environment env, ScriptService scriptService, AnalysisRegistry analysisRegistry,
@@ -258,10 +255,16 @@ Map<String, Pipeline> pipelines() {
     @Override
     public void applyClusterState(final ClusterChangedEvent event) {
         ClusterState state = event.state();
+        Map<String, Pipeline> originalPipelines = pipelines;
         innerUpdatePipelines(event.previousState(), state);
-        IngestMetadata ingestMetadata = state.getMetaData().custom(IngestMetadata.TYPE);
-        if (ingestMetadata != null) {
-            updatePipelineStats(ingestMetadata);
+        //pipelines changed, so add the old metrics to the new metrics
+        if (originalPipelines != pipelines) {
+            pipelines.forEach((id, pipeline) -> {
+                Pipeline originalPipeline = originalPipelines.get(id);
+                if (originalPipeline != null) {
+                    pipeline.getMetrics().add(originalPipeline.getMetrics());
+                }
+            });
         }
     }
 
@@ -326,6 +329,7 @@ void validatePipeline(Map<DiscoveryNode, IngestInfo> ingestInfos, PutPipelineReq
     public void executeBulkRequest(Iterable<DocWriteRequest<?>> actionRequests,
         BiConsumer<IndexRequest, Exception> itemFailureHandler, Consumer<Exception> completionHandler,
         Consumer<IndexRequest> itemDroppedHandler) {
+
         threadPool.executor(ThreadPool.Names.WRITE).execute(new AbstractRunnable() {
 
             @Override
@@ -368,37 +372,11 @@ protected void doRun() {
     }
 
     public IngestStats stats() {
-        Map<String, StatsHolder> statsHolderPerPipeline = this.statsHolderPerPipeline;
 
-        Map<String, IngestStats.Stats> statsPerPipeline = new HashMap<>(statsHolderPerPipeline.size());
-        for (Map.Entry<String, StatsHolder> entry : statsHolderPerPipeline.entrySet()) {
-            statsPerPipeline.put(entry.getKey(), entry.getValue().createStats());
-        }
+        Map<String, IngestStats.Stats> statsPerPipeline =
+            pipelines.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> v.getValue().getMetrics().createStats()));
 
-        return new IngestStats(totalStats.createStats(), statsPerPipeline);
-    }
-
-    void updatePipelineStats(IngestMetadata ingestMetadata) {
-        boolean changed = false;
-        Map<String, StatsHolder> newStatsPerPipeline = new HashMap<>(statsHolderPerPipeline);
-        Iterator<String> iterator = newStatsPerPipeline.keySet().iterator();
-        while (iterator.hasNext()) {
-            String pipeline = iterator.next();
-            if (ingestMetadata.getPipelines().containsKey(pipeline) == false) {
-                iterator.remove();
-                changed = true;
-            }
-        }
-        for (String pipeline : ingestMetadata.getPipelines().keySet()) {
-            if (newStatsPerPipeline.containsKey(pipeline) == false) {
-                newStatsPerPipeline.put(pipeline, new StatsHolder());
-                changed = true;
-            }
-        }
-
-        if (changed) {
-            statsHolderPerPipeline = Collections.unmodifiableMap(newStatsPerPipeline);
-        }
+        return new IngestStats(totalMetrics.createStats(), statsPerPipeline);
     }
 
     private void innerExecute(IndexRequest indexRequest, Pipeline pipeline, Consumer<IndexRequest> itemDroppedHandler) throws Exception {
@@ -409,10 +387,8 @@ private void innerExecute(IndexRequest indexRequest, Pipeline pipeline, Consumer
         long startTimeInNanos = System.nanoTime();
         // the pipeline specific stat holder may not exist and that is fine:
         // (e.g. the pipeline may have been removed while we're ingesting a document
-        Optional<StatsHolder> pipelineStats = Optional.ofNullable(statsHolderPerPipeline.get(pipeline.getId()));
         try {
-            totalStats.preIngest();
-            pipelineStats.ifPresent(StatsHolder::preIngest);
+            totalMetrics.preIngest();
             String index = indexRequest.index();
             String type = indexRequest.type();
             String id = indexRequest.id();
@@ -438,13 +414,11 @@ private void innerExecute(IndexRequest indexRequest, Pipeline pipeline, Consumer
                 indexRequest.source(ingestDocument.getSourceAndMetadata());
             }
         } catch (Exception e) {
-            totalStats.ingestFailed();
-            pipelineStats.ifPresent(StatsHolder::ingestFailed);
+            totalMetrics.ingestFailed();
             throw e;
         } finally {
             long ingestTimeInMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeInNanos);
-            totalStats.postIngest(ingestTimeInMillis);
-            pipelineStats.ifPresent(statsHolder -> statsHolder.postIngest(ingestTimeInMillis));
+            totalMetrics.postIngest(ingestTimeInMillis);
         }
     }
 
@@ -481,27 +455,4 @@ private void innerUpdatePipelines(ClusterState previousState, ClusterState state
         ExceptionsHelper.rethrowAndSuppress(exceptions);
     }
 
-    private static class StatsHolder {
-
-        private final MeanMetric ingestMetric = new MeanMetric();
-        private final CounterMetric ingestCurrent = new CounterMetric();
-        private final CounterMetric ingestFailed = new CounterMetric();
-
-        void preIngest() {
-            ingestCurrent.inc();
-        }
-
-        void postIngest(long ingestTimeInMillis) {
-            ingestCurrent.dec();
-            ingestMetric.inc(ingestTimeInMillis);
-        }
-
-        void ingestFailed() {
-            ingestFailed.inc();
-        }
-
-        IngestStats.Stats createStats() {
-            return new IngestStats.Stats(ingestMetric.count(), ingestMetric.sum(), ingestCurrent.count(), ingestFailed.count());
-        }
-    }
 }

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

@@ -22,10 +22,12 @@
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Nullable;
 
+import java.time.Clock;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+
 import org.elasticsearch.script.ScriptService;
 
 /**
@@ -44,12 +46,21 @@ public final class Pipeline {
     @Nullable
     private final Integer version;
     private final CompoundProcessor compoundProcessor;
+    private final IngestMetric metrics;
+    private final Clock clock;
 
     public Pipeline(String id, @Nullable String description, @Nullable Integer version, CompoundProcessor compoundProcessor) {
+        this(id, description, version, compoundProcessor, Clock.systemUTC());
+    }
+
+    //package private for testing
+    Pipeline(String id, @Nullable String description, @Nullable Integer version, CompoundProcessor compoundProcessor, Clock clock) {
         this.id = id;
         this.description = description;
         this.compoundProcessor = compoundProcessor;
         this.version = version;
+        this.metrics = new IngestMetric();
+        this.clock = clock;
     }
 
     public static Pipeline create(String id, Map<String, Object> config,
@@ -78,7 +89,17 @@ public static Pipeline create(String id, Map<String, Object> config,
      * Modifies the data of a document to be indexed based on the processor this pipeline holds
      */
     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
-        return compoundProcessor.execute(ingestDocument);
+        long startTimeInMillis = clock.millis();
+        try {
+            metrics.preIngest();
+            return compoundProcessor.execute(ingestDocument);
+        } catch (Exception e) {
+            metrics.ingestFailed();
+            throw e;
+        } finally {
+            long ingestTimeInMillis = clock.millis() - startTimeInMillis;
+            metrics.postIngest(ingestTimeInMillis);
+        }
     }
 
     /**
@@ -135,4 +156,11 @@ public List<Processor> getOnFailureProcessors() {
     public List<Processor> flattenAllProcessors() {
         return compoundProcessor.flattenProcessors();
     }
+
+    /**
+     * The metrics associated with this pipeline.
+     */
+    public IngestMetric getMetrics() {
+        return metrics;
+    }
 }