INGEST: Add Pipeline Processor

[original-brownbear]

• Adds Processor capable of invoking other pipelines
• Closes [FEATURE REQUEST] - ingest nodes - run a pipeline as a processor in another pipeline #31842

Example:

Put inner and outer pipeline (with new processor):

PUT _ingest/pipeline/inner-pipeline
{
  "description" : "inner pipeline",
  "processors" : [
    {
      "set" : {
        "field": "foo",
        "value": "bar"
      }
    },
    {
      "set" : {
        "field": "baz",
        "value": "blub"
      }
    }
  ]
}

PUT _ingest/pipeline/outer-pipeline
{
  "description" : "outer pipeline",
  "processors" : [
    {
      "pipeline" : {
        "pipeline": "inner-pipeline"
      }
    }
  ]
}

Add document:

POST /myindex/1?pipeline=outer-pipeline
{
  "field": "value"
}

Works:

POST /myindex/_search
{}

=>

{
  "took": 30,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 1,
    "hits": [
      {
        "_index": "myindex",
        "_type": "1",
        "_id": "LSGj7GQBWAIRG1UBRP2D",
        "_score": 1,
        "_source": {
          "field": "value",
          "foo": "bar",
          "baz": "blub"
        }
      }
    ]
  }
}

[elasticmachine]

Pinging @elastic/es-core-infra

[original-brownbear]

@rjernst @tsg @talevy @jakelandis

just a suggestion for the inter pipeline communication we talked about a while back. Maybe take a look if this kind of API is what we're looking for here, then I'd add tests and clean things up a bit. This version already works fine functionally at least I think :)

[rjernst]

I think there are a couple things needed before this could be merged:

• Validation - we should have validation when PUTting the outer pipeline that the name of the inner pipeline exists. Similar validation should exist when deleting a pipeline, to ensure a pipeline in use by a pipeline processor cannot be deleted.
• Ingest services consolidation - I don't like that this adds essentially a volatile global to get access to the pipelines. It has always confused me that we have IngestService being essentially a dummy service, and the real logic being split between PipelineStore and PipelineExecutionService. I think these could all be consolidate together. Then, part of the Processor.Parameters could be the IngestService, which would have a public method for executing a pipeline.

Additionally, we need to think how this can be exposed as a method in painless. I've started discussions on tweaking how SPI works to possibly allow functors to be bound to a method in painless through the whitelist, so that the method call to say runPipeline("mypipeline", ctx) could be backed by a call to the ingest service.

[original-brownbear]

@rjernst

Validation - we should have validation when PUTting the outer pipeline that the name of the inner pipeline exists. Similar validation should exist when deleting a pipeline, to ensure a pipeline in use by a pipeline processor cannot be deleted.

Sure that's a trivial if to add to the code :)

Ingest services consolidation - I don't like that this adds essentially a volatile global to get access to the pipelines. It has always confused me that we have IngestService being essentially a dummy service, and the real logic being split between PipelineStore and PipelineExecutionService. I think these could all be consolidate together. Then, part of the Processor.Parameters could be the IngestService, which would have a public method for executing a pipeline.

Def, that sounds good but would be a much larger refactoring. I just went with the most straightforward approach here for the POC. I can look into making that refactoring that if this is something we want now?

[jakelandis]

I like the approach, it is pretty intuitive.

I assume that if you have a pipeline processor in the middle of your (outer) pipeline that control will eventually come back to the original (outer) pipeline, and finish out that pipeline's execution ?

Also, we should probably detect loops, preferably at time of creation... however that will be much more complicated when we get the if conditional and ability to call other pipelines from painless.

[original-brownbear]

@jakelandis

I assume that if you have a pipeline processor in the middle of your (outer) pipeline that control will eventually come back to the original (outer) pipeline, and finish out that pipeline's execution ?

yea that should work just fine and transparently :)

Also, we should probably detect loops, preferably at time of creation... however that will be much more complicated when we get the if conditional and ability to call other pipelines from painless.

Yea this is a good one: I guess detecting loops if we also want to factor in conditionals is impossible (that's hypercomputation imo :P). I think theoretically we have the choice between not detecting this and allowing some crazy recursive pipelines or simply walking the pipeline graph and not allowing circles even if conditionals are present.
I'm strongly in favour of the latter. There's almost no reasonable use cases for the former that you couldn't also handle via Painless/scripting and allowing recursion would really allow for a lot of "shooting yourself in the foot" .

[jakelandis]

simply walking the pipeline graph and not allowing circles even if conditionals are present
+1

We could, in-addition to walking the pipeline at creation, provide runtime checks such that no single processor can execute more then N times in the context of a single document (a circuit breaker of sorts). This would however, require maintaining some additional per document state accessible to all processors in the pipeline. I think this may be warranted, since "shooting yourself in the foot" here could mean a full stop of all ingest.

[original-brownbear]

@jakelandis

provide runtime checks such that no single processor can execute more then N times in the context of a single document (a circuit breaker of sorts)

Right now that you mention it ... we have to have this imo. You can update pipelines ... so some up front check isn't sufficient here in preventing infinite loops.
I think tracking the "stacktrace" of executed pipelines isn't too hard though :) We're passing the same ingest document instance from pipeline to pipeline.
The most low-tech solution here could simply be tracking the "stack depth" of pipeline calls in an int and putting some limit on it. => Runtime checking should easy :)

[rjernst]

I don't think we should have or need stack depth limits like that. We should simply not allow recursion at all? If we use IngestDocument to track the stack, and do not allow any recursion, then an identity hash set of the processors being executed should work. When a processor starts, it adds itself to the hash set, erroring if it already existed, then removes itself when it is done.

[original-brownbear]

@rjernst

We should simply not allow recursion at all?

I'm fine either way. I don't see a good use case for recursion.
Tracking by processor (I guess it's fine to just track pipeline processor calls themselves here) should work out :) Will add that to the PR.

[tsg]

From the user perspective, I think this is a wonderful enhancement, enabling for example code reuse between Beats modules. It also fits nicely with the per-processor conditionals. I guess it's going to be possible to add a condition to the pipeline processor, once both features land, right?

+1 on forbidding recursion, that seems like the right decision to me.

[original-brownbear]

@rjernst alright, now that #32617 (consolidated IngestService) got merged I updated this and added the recursive invocation check.
Can you maybe take a look and let me know if the approach is ok now, then I'd add some UTs (ITs are there already) and we should be good here?

[rjernst]

A couple comments, and yes, unit tests would be good.

[rjernst]

Was this accidentally added back? It shouldn't be necessary now right?

[original-brownbear]

@rjernst yea I apparently suck at merging :) Will remove

[rjernst]

Instead of using a set, I think there a call stack could be passed through, that doesn't need to be a member variable? I don't know ingest document coming more mutable than it already is. This method signature is also odd, as I would expect this to be an exception, but it looks like you are avoiding that because it would collide with exceptions that could be thrown from the pipeline itself? But this should fail the pipeline anyways, so I think it is ok to use an exception?

[original-brownbear]

@rjernst I can't really pass the callstack with the document can I? I only have the execute method available to pass things (because any called pipeline itself could contain additional pipeline processors).

The reason I made this return boolean instead of throwing right away was more of a style thing to make it clear that the exception was triggered by the pipeline processor. But in hindsight this may be a little needlessly complex :) Moving it in here.

[original-brownbear]

@rjernst questions answered, unit tests added :)

[rjernst]

I was originally suggesting changing the signature of execute on the ingest service to take in the stack, but I don't feel that strongly about it. LGTM.

[original-brownbear]

@rjernst thanks! Will merge once CI passes :)