ingest: support simulate with verbose for pipeline processor #33839

jakelandis · 2018-09-18T21:58:53Z

This change better supports the use of simulate?verbose with the
pipeline processor. Prior to this change any pipeline processors
executed with simulate?verbose would not show the full steps
that include the intermediate processors for the inner pipelines.

This changes also moves the PipelineProcess and TrackingResultProcessor
classes to enable instance checks and to avoid overly public classes.

Prior to this change:

PUT _ingest/pipeline/mypipeline1
{
  "processors": [
    {
      "set": {
        "if" : "true",
        "tag": "set_from_pipeline1",
        "field": "a",
        "value": "foo"
      }
    },
    {
      "pipeline": {
        "pipeline": "mypipeline2"
      }
    }
  ]
}

PUT _ingest/pipeline/mypipeline2
{
  "processors": [
    {
      "rename": {
        "tag": "rename_from_pipeline2",
        "field": "a",
        "target_field": "a1"
      }
    },
    {
      "pipeline": {
        "pipeline": "mypipeline3"
      }
    }
  ]
}

PUT _ingest/pipeline/mypipeline3
{
  "processors": [
    {
      "rename": {
        "tag": "rename_from_pipeline3",
        "field": "a1",
        "target_field": "a2"
      }
    }
  ]
}


POST _ingest/pipeline/mypipeline1/_simulate?verbose
{
  "docs": [
    {
      "_source": {
        "a" : "b"
      }
    }
  ]
}

Results in

{
  "docs": [
    {
      "processor_results": [
        {
          "tag": "set_from_pipeline1",
          "doc": {
            "_index": "_index",
            "_type": "_type",
            "_id": "_id",
            "_source": {
              "a": "foo"
            },
            "_ingest": {
              "timestamp": "2018-09-18T21:49:01.881075Z"
            }
          }
        },
        {
          "doc": {
            "_index": "_index",
            "_type": "_type",
            "_id": "_id",
            "_source": {
              "a2": "foo"
            },
            "_ingest": {
              "timestamp": "2018-09-18T21:49:01.881075Z"
            }
          }
        }
      ]
    }
  ]
}

^^ Note pipeline2 is missing from the result steps.

After this change:

{
  "docs": [
    {
      "processor_results": [
        {
          "tag": "set_from_pipeline1",
          "doc": {
            "_index": "_index",
            "_type": "_type",
            "_id": "_id",
            "_source": {
              "a": "foo"
            },
            "_ingest": {
              "timestamp": "2018-09-18T21:49:32.52323Z"
            }
          }
        },
        {
          "tag": "rename_from_pipeline2",
          "doc": {
            "_index": "_index",
            "_type": "_type",
            "_id": "_id",
            "_source": {
              "a1": "foo"
            },
            "_ingest": {
              "timestamp": "2018-09-18T21:49:32.52323Z"
            }
          }
        },
        {
          "tag": "rename_from_pipeline3",
          "doc": {
            "_index": "_index",
            "_type": "_type",
            "_id": "_id",
            "_source": {
              "a2": "foo"
            },
            "_ingest": {
              "timestamp": "2018-09-18T21:49:32.52323Z"
            }
          }
        }
      ]
    }
  ]
}

This change better supports the use of simulate?verbose with the pipeline processor. Prior to this change any pipeline processors executed with simulate?verbose would not show the full steps that include the intermediate processors for the inner pipelines. This changes also moves the PipelineProcess and TrackingResultProcessor classes to enable instance checks and to avoid overly public classes.

elasticmachine · 2018-09-18T21:58:55Z

Pinging @elastic/es-core-infra

original-brownbear

The recursion check is wrong I think + a few nits on formatting noise.

Also, maybe it would be nice to provide some visual indication for what processor is part of what pipeline automatically instead of having to manually set tags for that? (e.g. we would just automatically prefix the tags for nested pipeline calls?)

original-brownbear · 2018-09-19T11:30:32Z

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

+            if (processor instanceof PipelineProcessor) {
+                PipelineProcessor pipelineProcessor = ((PipelineProcessor) processor);
+                if (pipelinesSeen.add(pipelineProcessor) == false) {
+                    throw new IllegalStateException("Recursive invocation of pipeline ["


I think this is not correct because it would also throw for repeated invocation of the same pipeline right? (Since PipelineProcessor doesn't implement equals you will probably only see this in some crazier scenarios (A calls B calls C and then another step of just B which would call C again making it throw on seeing C twice even though it's not a recursive invocation)) See #33419 for more here. I think the easiest fix to get out of that problem is to do the same thing that the pipeline processor does and simply track the current stack of pipelines in pipelinesSeen instead of all the pipelines ever seen (in this case this means just removing a pipeline proc from the pipelinesSeen after it has been unwrapped)

good catch. fixed and updated in latest commit.

thanks looks good :)

original-brownbear · 2018-09-19T11:31:31Z

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

    }

    public void testActualProcessor() throws Exception {
-        TestProcessor actualProcessor = new TestProcessor(ingestDocument -> {});
+        TestProcessor actualProcessor = new TestProcessor(ingestDocument -> {


NIT: noisy reformat :)

fixed. sorry for the noise.
Note to self:
Intellij - Simple lambdas in one line to avoid this kind of noise.

original-brownbear · 2018-09-19T11:31:48Z

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

@@ -69,9 +79,11 @@ public void testActualProcessor() throws Exception {

    public void testActualCompoundProcessorWithoutOnFailure() throws Exception {
        RuntimeException exception = new RuntimeException("processor failed");
-        TestProcessor testProcessor = new TestProcessor(ingestDocument -> {  throw exception; });
+        TestProcessor testProcessor = new TestProcessor(ingestDocument -> {


NIT: noisy reformat :)

original-brownbear · 2018-09-19T11:32:06Z

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

@@ -90,14 +102,17 @@ public void testActualCompoundProcessorWithoutOnFailure() throws Exception {

    public void testActualCompoundProcessorWithOnFailure() throws Exception {
        RuntimeException exception = new RuntimeException("fail");
-        TestProcessor failProcessor = new TestProcessor("fail", "test", ingestDocument -> {  throw exception; });
-        TestProcessor onFailureProcessor = new TestProcessor("success", "test", ingestDocument -> {});
+        TestProcessor failProcessor = new TestProcessor("fail", "test", ingestDocument -> {


NIT: Noisy reformatting

jakelandis · 2018-09-19T15:16:32Z

@original-brownbear - thanks for the review. changes implemented.

Also, maybe it would be nice to provide some visual indication for what processor is part of what pipeline automatically instead of having to manually set tags for that? (e.g. we would just automatically prefix the tags for nested pipeline calls?)

Agreed it would be nice, and will consider for future enhancement. (but not part of this PR)

jakelandis · 2018-09-19T15:20:11Z

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

+        assertThat(resultList.get(0).getFailure(), sameInstance(exception));
+        assertThat(resultList.get(0).getProcessorTag(), equalTo(expectedResult.getProcessorTag()));
+    }
+


Hmm... not sure why github didn't preserve the location change in the diff here. Everything below this comment is new, nothing above (except imports) have changed.

original-brownbear

LGTM but probably best to wait for @rjernst :)

rjernst

LGTM too. One nit on naming: it's not necessarily recursion (self invocation), but just any form of a cycle across pipeline processors right? I would tweak the error message to something like "Cycle found between processors A and B"

jakelandis · 2018-09-19T21:35:10Z

@rjernst thanks! updated the wording to "Cycle detected for pipeline: <pipeline>" . I also updated the non-simulate error to match.

will merge when green.

…#33839) This change better supports the use of simulate?verbose with the pipeline processor. Prior to this change any pipeline processors executed with simulate?verbose would not show all intermediate processors for the inner pipelines. This changes also moves the PipelineProcess and TrackingResultProcessor classes to enable instance checks and to avoid overly public classes. As well this updates the error message for when cycles are detected in pipelines calling other pipelines.

…#33839) * ingest: support simulate with verbose for pipeline processor This change better supports the use of simulate?verbose with the pipeline processor. Prior to this change any pipeline processors executed with simulate?verbose would not show all intermediate processors for the inner pipelines. This changes also moves the PipelineProcess and TrackingResultProcessor classes to enable instance checks and to avoid overly public classes. As well this updates the error message for when cycles are detected in pipelines calling other pipelines.

* ingest: support simulate with verbose for pipeline processor This change better supports the use of simulate?verbose with the pipeline processor. Prior to this change any pipeline processors executed with simulate?verbose would not show all intermediate processors for the inner pipelines. This changes also moves the PipelineProcess and TrackingResultProcessor classes to enable instance checks and to avoid overly public classes. As well this updates the error message for when cycles are detected in pipelines calling other pipelines.

jakelandis · 2018-10-23T16:21:06Z

6.5 backport: 2d24192

* ingest: support simulate with verbose for pipeline processor This change better supports the use of simulate?verbose with the pipeline processor. Prior to this change any pipeline processors executed with simulate?verbose would not show all intermediate processors for the inner pipelines. This changes also moves the PipelineProcess and TrackingResultProcessor classes to enable instance checks and to avoid overly public classes. As well this updates the error message for when cycles are detected in pipelines calling other pipelines.

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

jakelandis requested review from rjernst and original-brownbear September 18, 2018 21:58

original-brownbear requested changes Sep 19, 2018

View reviewed changes

jakelandis added 2 commits September 19, 2018 08:39

fix noisy re-formatting

7f0d154

support repeated invocations and clean up variable names in tests

36eafc9

jakelandis commented Sep 19, 2018

View reviewed changes

original-brownbear approved these changes Sep 19, 2018

View reviewed changes

Merge branch 'master' into simulate_with_pipeline_processor

a2145f6

rjernst approved these changes Sep 19, 2018

View reviewed changes

re-word 'recursive' error case to 'cycle'

db4d53e

jakelandis merged commit e37e5df into elastic:master Sep 20, 2018

jakelandis added the backport pending label Sep 20, 2018

jakelandis mentioned this pull request Oct 19, 2018

6.x Ingest node back ports #34653

Closed

jakelandis removed the backport pending label Oct 23, 2018

colings86 added the >enhancement label Oct 25, 2018

Mpdreamz mentioned this pull request Dec 13, 2018

[meta] 6.5.0 Release elastic/elasticsearch-net#3457

Closed

colings86 added v7.0.0-beta1 and removed v7.0.0 labels Feb 7, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ingest: support simulate with verbose for pipeline processor #33839

ingest: support simulate with verbose for pipeline processor #33839

jakelandis commented Sep 18, 2018

elasticmachine commented Sep 18, 2018

original-brownbear left a comment

original-brownbear Sep 19, 2018

jakelandis Sep 19, 2018

original-brownbear Sep 19, 2018

original-brownbear Sep 19, 2018

jakelandis Sep 19, 2018

original-brownbear Sep 19, 2018

original-brownbear Sep 19, 2018

jakelandis commented Sep 19, 2018

jakelandis Sep 19, 2018

original-brownbear left a comment

rjernst left a comment

jakelandis commented Sep 19, 2018 •

edited

Loading

jakelandis commented Oct 23, 2018

ingest: support simulate with verbose for pipeline processor #33839

ingest: support simulate with verbose for pipeline processor #33839

Conversation

jakelandis commented Sep 18, 2018

elasticmachine commented Sep 18, 2018

original-brownbear 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

jakelandis commented Sep 19, 2018

Choose a reason for hiding this comment

original-brownbear left a comment

Choose a reason for hiding this comment

rjernst left a comment

Choose a reason for hiding this comment

jakelandis commented Sep 19, 2018 • edited Loading

jakelandis commented Oct 23, 2018

jakelandis commented Sep 19, 2018 •

edited

Loading