ingest: processor stats #34202

jakelandis · 2018-10-01T20:38:21Z

This change introduces stats per processors. Total, time, failed,
current are currently supported. All pipelines will now show all
top level processors that belong to it. Failure processors are not
displayed, however, the time taken to execute the failure chain is part
of the stats for the top level processor.

The processor name is the type of the processor, ordered as defined in
the pipeline. If a tag for the processor is found, then the tag is
appended to the type (colon separated).

Pipeline processors will have the pipeline name appended (colon separated)
to the name of the pipeline (before the tag if one exists).
If more then one pipeline is used to process the document, then each pipeline
will carry its own stats. The outer most pipeline will also include the
inner most pipeline stats.

Conditional processors will only be included in the stats if the condition
evaluates to true.

Best attempts are made to carry forward processor metrics between
cluster state changes. If no changes are made to the pipeline, metrics
will be carried forward. However, if the pipeline changes and the
number of processors or the order of the processors (determined by type)
changes, the processor metrics will be reset to zero.

Closes #33387

An example pipeline:

"test_pipeline": {
  "count": 0,
  "time_in_millis": 0,
  "current": 0,
  "failed": 0,
  "processors": [
    {
      "grok": {
        "count": 0,
        "time_in_millis": 0,
        "current": 0,
        "failed": 0
      }
    },
    {
      "date": {
        "count": 0,
        "time_in_millis": 0,
        "current": 0,
        "failed": 0
      }
    },
    {
      "remove": {
        "count": 0,
        "time_in_millis": 0,
        "current": 0,
        "failed": 0
      }
    }
  ]
}

^^ The "processors": [ is new.

An example with a tag and a pipeline processor:

"mypipeline1": {
  "count": 0,
  "time_in_millis": 0,
  "current": 0,
  "failed": 0,
  "processors": [
    {
      "set:sets the thing to true": {
        "count": 0,
        "time_in_millis": 0,
        "current": 0,
        "failed": 0
      }
    },
    {
      "pipeline:mypipeline2": {
        "count": 0,
        "time_in_millis": 0,
        "current": 0,
        "failed": 0
      }
    }
  ]
}

Rally with the http_logs with grok was run against the code prior and with this change. No performance impact was noticed.

Also, note this PR does not make any changes to the output of simulate or simulate?verbose. (that will be a different PR).

This change introduces stats per processors. Total, time, failed, current are currently supported. All pipelines will now show all top level processors that belong to it. Failure processors are not displayed, however, the time taken to execute the failure chain is part of the stats for the top level processor. The processor name is the type of the processor, ordered as defined in the pipeline. If a tag for the processor is found, then the tag is appended to the type. Pipeline processors will have the pipeline name appended to the name of the name of the processors (before the tag if one exists). If more then one pipeline is used to process the document, then each pipeline will carry its own stats. The outer most pipeline will also include the inner most pipeline stats. Conditional processors will only included in the stats if the condition evaluates to true.

elasticmachine · 2018-10-01T20:38:23Z

Pinging @elastic/es-core-infra

rjernst

A few comments

rjernst · 2018-10-03T23:12:18Z

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

        super();
        this.ignoreFailure = ignoreFailure;
        this.processors = processors;
        this.onFailureProcessors = onFailureProcessors;
+        this.clock = clock;
+        processorsWithMetrics = new ArrayList<>(processors.size());


Please use consistent style, setting with this.

rjernst · 2018-10-03T23:15:34Z

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

+        for (Tuple<Processor, IngestMetric> processorWithMetric : processorsWithMetrics) {
+            Processor processor = processorWithMetric.v1();
+            IngestMetric metric = processorWithMetric.v2();
+            long startTimeInMillis = clock.millis();


This is backed by System.currentTimeMillis(). In other areas of the system, we avoid this and use a wrapper on ThreadPool which avoids excessive calls to that method, caching the result across threads. Either we should be using a LongSupplier like we do in other areas, or have a Clock implementation backed by ThreadPool. Additionally, I'm not sure we need absolute time, so probably the other method for relative time would be better.

@rjernst I removed the clock in favor a LongSupplier backed by System:nanoTime 2714338. Can you please confirm that this is the correct fix here ? I made the same mistake in a different PR I will need fix too.

@rjernst - I also fixed this for other places where I made the same mistake: ff98a82

rjernst · 2018-10-03T23:16:55Z

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


-    ConditionalProcessor(String tag, Script script, ScriptService scriptService, Processor processor) {
+   ConditionalProcessor(String tag, Script script, ScriptService scriptService, Processor processor) {


nit: spacing is off here

rjernst · 2018-10-03T23:17:31Z

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

        }
        return ingestDocument;
    }

+    Processor getProcessor() {


If this is just going to be used by tests, why not make the member package protected instead of a getter?

This is used here: https://github.com/elastic/elasticsearch/pull/34202/files#diff-579dffc1e22e3db13c41f685046b2891R439 to get the actual processor name

rjernst · 2018-10-03T23:17:50Z

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

+                    //Best attempt to populate new processor metrics using a parallel array of the old metrics. This is not ideal since
+                    //the per processor metrics may get reset when the arrays don't match. However, to get to an ideal model, unique and
+                    //consistent id's per processor and/or semantic equals for each processor will be needed.
+                    if(newPerProcessMetrics.size() == oldPerProcessMetrics.size()) {


nit: space after if

rjernst · 2018-10-03T23:20:10Z

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

+                    List<Tuple<Processor, IngestMetric>> newPerProcessMetrics = new ArrayList<>();
+                    getProcessorMetrics(originalPipeline.getCompoundProcessor(), oldPerProcessMetrics);
+                    getProcessorMetrics(pipeline.getCompoundProcessor(), newPerProcessMetrics);
+                    //Best attempt to populate new processor metrics using a parallel array of the old metrics. This is not ideal since


Why try to transfer metrics at all? Could we just say when a pipeline's configuration is updated, the metrics are reset?

When 1 pipeline changes, then all pipelines are rebuilt and without this code we would loose all metrics on any pipeline change. For example we have 2 pipelines, and 1 one of them is deleted, we don't want to loose the metrics for the pipeline that had no changes. We don't have easy access to exactly which pipeline changed, and the heuristic to carry forward the metrics is if the pipeline still exists, the count of processors and the types of processor (in order) don't change, then carry forward the metrics.

rjernst · 2018-10-03T23:20:41Z

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

+            pipelineStats.writeTo(out);
+            List<Tuple<String, Stats>> processorStats = entry.getValue().v2();
+            out.writeVInt(processorStats.size());
+            for(Tuple<String, Stats> processorTuple : processorStats){


nit: space after for

rjernst · 2018-10-03T23:24:11Z

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

     */
-    public Map<String, Stats> getStatsPerPipeline() {
+    public Map<String, Tuple<IngestStats.Stats, List<Tuple<String, IngestStats.Stats>>>> getStatsPerPipeline() {


I think the type here would be much easier to understand as a dedicated class, rather than a very nested set of Tuple/List/Map

Fixed in 8067758. IngestStats now accepts totalStats, List, and Map<String, ProcessorStat> (keyed by pipelineId), and a builder to help build from the Metrics representation. I hope this makes the code more readable.

rjernst · 2018-10-03T23:24:59Z

server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java

+                        List<Tuple<String, IngestStats.Stats>> deserializedProcessorStats =
+                            deserializedIngestStats.getProcessorStatsForPipeline(pipelineName);
+                        Iterator<Tuple<String, IngestStats.Stats>> it = deserializedProcessorStats.iterator();
+                        for(Tuple<String, IngestStats.Stats> processorTuple : processorStats){


nit: space after for

jakelandis · 2018-10-15T21:12:39Z

@rjernst - All initial comments have been addressed. Mind to take another look ?

rjernst

LGTM

* master: Use trial license in docs tests (elastic#34673) Scripting: Convert script fields to use script context (elastic#34164) TEST: Mute testDedupByPrimaryTerm ingest: processor stats (elastic#34202)

This reverts commit 6567729.

jasontedor · 2018-10-21T17:18:19Z

I reverted this from master in 0577703 due to failing tests in the mixed cluster tests.

* master: Revert "ingest: processor stats (elastic#34202)"

This change introduces stats per processors. Total, time, failed, current are currently supported. All pipelines will now show all top level processors that belong to it. Failure processors are not displayed, however, the time taken to execute the failure chain is part of the stats for the top level processor. The processor name is the type of the processor, ordered as defined in the pipeline. If a tag for the processor is found, then the tag is appended to the type. Pipeline processors will have the pipeline name appended to the name of the name of the processors (before the tag if one exists). If more then one pipeline is used to process the document, then each pipeline will carry its own stats. The outer most pipeline will also include the inner most pipeline stats. Conditional processors will only included in the stats if the condition evaluates to true.

jakelandis · 2018-10-22T21:18:59Z

Re-introduce this change with test fix on #34724

ruflin · 2018-10-29T11:21:18Z

@yaronp68 FYI

This change introduces stats per processors. Total, time, failed, current are currently supported. All pipelines will now show all top level processors that belong to it. Failure processors are not displayed, however, the time taken to execute the failure chain is part of the stats for the top level processor. The processor name is the type of the processor, ordered as defined in the pipeline. If a tag for the processor is found, then the tag is appended to the type. Pipeline processors will have the pipeline name appended to the name of the name of the processors (before the tag if one exists). If more then one pipeline is used to process the document, then each pipeline will carry its own stats. The outer most pipeline will also include the inner most pipeline stats. Conditional processors will only included in the stats if the condition evaluates to true.

This reverts commit 6567729.

jakelandis added 2 commits October 1, 2018 13:43

add missing test

3337458

jakelandis added :Data Management/Ingest Node Execution or management of Ingest Pipelines including GeoIP v7.0.0 v6.5.0 labels Oct 1, 2018

jakelandis requested a review from rjernst October 1, 2018 20:38

rjernst reviewed Oct 3, 2018

View reviewed changes

jakelandis added 3 commits October 4, 2018 09:23

cosmetic changes

4588782

remove clock infavor of a LongSupplier to System::nanoTime

2714338

simplify the IngestStats pipeline->per-processor object model

8067758

jakelandis added 2 commits October 17, 2018 12:10

Merge branch 'master' into processor_stats

34af7e9

fix other incorrect usage of Clock

ff98a82

rjernst approved these changes Oct 19, 2018

View reviewed changes

jakelandis merged commit 6567729 into elastic:master Oct 20, 2018

jakelandis added the backport pending label Oct 20, 2018

jakelandis mentioned this pull request Oct 20, 2018

6.x Ingest node back ports #34653

Closed

jasontedor added a commit that referenced this pull request Oct 21, 2018

Revert "ingest: processor stats (#34202)"

0577703

This reverts commit 6567729.

jasontedor added a commit to jasontedor/elasticsearch that referenced this pull request Oct 21, 2018

Merge branch 'master' into allow-set-section-in-setup

f4b68da

* master: Revert "ingest: processor stats (elastic#34202)"

jasontedor added a commit to jasontedor/elasticsearch that referenced this pull request Oct 21, 2018

Merge branch 'master' into wait-for-pending-tasks

9b81204

* master: Revert "ingest: processor stats (elastic#34202)"

jakelandis mentioned this pull request Oct 22, 2018

ingest: better support for conditionals with simulate?verbose #34155

Merged

jakelandis mentioned this pull request Oct 22, 2018

ingest: processor stats #34724

Merged

jakelandis removed backport pending v6.5.0 v7.0.0 labels Oct 22, 2018

original-brownbear mentioned this pull request Oct 29, 2018

Initial start on parsing PAM messages elastic/beats#8756

Closed

kcm pushed a commit that referenced this pull request Oct 30, 2018

Revert "ingest: processor stats (#34202)"

0d6199b

This reverts commit 6567729.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ingest: processor stats #34202

ingest: processor stats #34202

jakelandis commented Oct 1, 2018 •

edited

Loading

elasticmachine commented Oct 1, 2018

rjernst left a comment

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

jakelandis Oct 17, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

rjernst Oct 3, 2018

jakelandis Oct 15, 2018

rjernst Oct 3, 2018

jakelandis Oct 4, 2018

jakelandis commented Oct 15, 2018

rjernst left a comment

jasontedor commented Oct 21, 2018

jakelandis commented Oct 22, 2018

ruflin commented Oct 29, 2018


		ConditionalProcessor(String tag, Script script, ScriptService scriptService, Processor processor) {
		ConditionalProcessor(String tag, Script script, ScriptService scriptService, Processor processor) {

ingest: processor stats #34202

ingest: processor stats #34202

Conversation

jakelandis commented Oct 1, 2018 • edited Loading

elasticmachine commented Oct 1, 2018

rjernst left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakelandis commented Oct 15, 2018

rjernst left a comment

Choose a reason for hiding this comment

jasontedor commented Oct 21, 2018

jakelandis commented Oct 22, 2018

ruflin commented Oct 29, 2018

jakelandis commented Oct 1, 2018 •

edited

Loading