Continue registering pipelines after one pipeline parse failure.

[talevy]

Ingest has been failing to apply existing pipelines from cluster-state
into the in-memory representation that are no longer valid. One example of
this is a pipeline with a script processor. If a cluster starts up with scripting
disabled, these pipelines will not be loaded. Even though GETing a pipeline worked,
indexing operations claimed that this pipeline did not exist. This is because one
gets information from cluster-state and the other is from an in-memory data-structure.

Now, two things happen

1. suppress the exceptions until after other successfull pipelines are loaded
2. replace failed pipelines with a placeholder pipeline that carries the exception message
   in its description

If the pipeline execution service encounters a pipeline stub, it is known that
something went wrong at the time of pipeline creation and an exception was thrown to
the user at some point at start-up.

closes #28269.

[martijnvg]

This looks good @talevy! I left a suggestion on how to improve handeling these pipeline construction errors.

[martijnvg]

I wonder if we can create a pipeline with a single fail processor. This fail processor can then contain the reason why the pipeline in question couldn't be loaded. This way there is no need to check the logs.

[talevy]

I like this, but I will try to do this in such a way that the pipeline isn't actually executed and is only used to store the exception. If I can't find a way to cleanly do that, then I will just let the pipeline run and fail

[talevy]

ah, also, fail processor is inside ingest-common so that makes it slightly more annoying

[talevy]

@martijnvg I've updated to carry the exception's message and pass it on to the user at runtime. good call — way better than telling them to look at logs.

[martijnvg]

@talevy Cool! I left an additional suggestion on how to improve the error handling.

[martijnvg]

what if we create a pipeline that throws the exception for us (fail processor is in a ingest-common, but it very simple to have that behaviour here too):

pipelines.put(pipeline.getId(), substitutePipeline(pipeline.getId(), e));

private Pipeline substitutePipeline(String id, ElasticsearchParseException e) {
        String tag = e.getHeaderKeys().contains("processor_tag") ? e.getHeader("processor_tag").get(0) : null;
        String type = e.getHeaderKeys().contains("processor_type") ? e.getHeader("processor_type").get(0) : "unknown";
        String errorMessage = "pipeline with id [" + id + "] could not be loaded, caused by [" + e.getDetailedMessage() + "]";
        Processor failureProcessor = new AbstractProcessor(tag) {
            @Override
            public void execute(IngestDocument ingestDocument) throws Exception {
                throw new IllegalStateException(errorMessage);
            }

            @Override
            public String getType() {
                return type;
            }
        };
        String description = "this is a place holder pipeline, because pipeline with id [" +  id + "] could not be loaded";
        return new Pipeline(id, description, null, new CompoundProcessor(failureProcessor));
    }

This way the if statement added PipelineExecutionService.java is not needed and this failure to load logic is in one place.

[martijnvg]

Note, I updated the above comment.

Also I think it is important to throw an IllegalStateException instead of a IllegalArgumentException, because the fact that we failed to load the pipeline at this stage is a real issue and this is the best way we can deal with it?

[talevy]

The problem with this approach is that we will be put these substituted pipelines in PUT requests.

this will return true: https://github.com/elastic/elasticsearch/pull/28752/files#diff-fd6036d9f6ce14fecb8751a3adb9fc59R280 (testPutWithPipelineFactoryError)

[talevy]

hmm. nevermind.. for some reason I saw that test fail. I might have done something wrong or a different test failed

[martijnvg]

s/ArrayList/List ?

[talevy]

updated

[talevy]

@martijnvg I've updated the PR to reflect your suggestion around inserting a placeholder pipeline that fails during execution

[martijnvg]

LGTM!

(The build only failed, because of a checkstyle violation)

Also I think we should backport this to the 5.6 branch. wdyt?

[talevy]

@martijnvg Once I merge into 6.x and master, I'll see how the backport looks for 5.6