Iteratively execute synchronous ingest processors

[danhermann]

Iteratively executes ingest processors that are not asynchronous. This resolves the issue of stack overflows that could occur in large ingest pipelines. As a secondary effect, it dramatically reduces the depth of the stack during ingest pipeline execution which reduces the performance cost of gathering a stack trace for any exceptions instantiated during pipeline execution. The aggregate performance effect of these changes has been observed to be a 10-15% improvement in some larger pipelines.

The primary changes required were adding iterative execution loops to CompoundProcessor and ForEachProcessor. Most of the other changes were consequences of changing those two processors.

Fixes #84274.

[elasticsearchmachine]

Hi @danhermann, I've created a changelog YAML for you.

[elasticmachine]

Pinging @elastic/es-data-management (Team:Data Management)

[elasticsearchmachine]

Hi @danhermann, I've updated the changelog YAML for you.

[danhermann]

@elasticmachine update branch

[masseyke]

Adding a brief description of how CompoundProcessor works since I was initially thrown off:
It iterates through all non-async processors, executing them until it comes across an async processor. It then asynchronously executes that one and recurses a stack frame, where it again iterates through all non-async processors, executing them until it comes across an async processor. And so on. So we still have recursion, but only one level per async processor. The synchronous processors won't all run at the same level in the same for loop, but we're at least no longer going additional levels for each synchronous processor.

[masseyke]

@elasticmachine update branch

[martijnvg]

@elasticmachine update branch

[masseyke]

@elasticmachine run elasticsearch-ci/bwc

[martijnvg]

LGTM

[masseyke]

I ran a couple of rally tests to make sure that we don't see a performance regression with this. For all tests I used a one-node Elasticsearch cluster and a separate node running rally.
First I ran the ingest_pipeline:baseline track that comes with rally. This doesn't have many processors, but it was a good check that we hadn't accidentally messed up performance:

esrally race --pipeline=benchmark-only --target-hosts="http://10.10.26.75:9200/" --track="http_logs" --challenge="append-index-only-with-ingest-pipeline" --track-params="ingest_pipeline:baseline" --client-options=~/cluster2-client-options.json --on-error=abort --track-params="bulk_indexing_clients:16,number_of_shards:1"

The median throughput improved by about 4.5%.
Next I ran the solutions/logs track in rally-internal-tracks, after configuring it to use apache access logs for 100% of the data.

esrally race --pipeline=benchmark-only --target-hosts="http://10.10.26.75:9200"  --track="solutions/logs" --client-options=~/cluster2-client-options.json --on-error=abort  --track-repository=rally-internal-tracks --track-params="bulk_indexing_clients:16,number_of_shards:1" --kill-running-processes

I ran that multiple times (since the first run on a fresh ES was always slowest), and on the non-first runs I'm consistently seeing the Total Ingest Pipeline time before the change around 190s, and after the change around 170s, so an ~11% improvement.

[DJRickyB]

Thanks @masseyke for pushing this across the finish line. It had a positive impact on the notoriously hard-to-pin-down conditional processor as well as script and geoip in our nightly benchmarks. I have annotated the charts: https://elasticsearch-benchmarks.elastic.co/#tracks/logging/nightly/default/30d