Request-level circuit breaker support on coordinating nodes

[markharwood]

Currently we do not have circuit breaker support for search requests executed on the coordinating node. We have multi-phase reduction which should help avoid OOMs but it is still possible to have abusive queries taking a node down.
A recent example OOM was caused by date histograms with 5 minute intervals executed across many time-based indices. Each of the data nodes failed to trip a circuit breaker because they were only seeing a small part of the final result. The multi-phase reduction did nothing to reduce the final number of buckets required and the final OOM occurred while rendering results in toXContent. This scenario was exacerbated by the fact there was a top-level terms agg for hostname under which there were the date histograms.

[elasticmachine]

Pinging @elastic/es-analytics-geo

[elasticmachine]

Pinging @elastic/es-core-infra

[markharwood]

@jimczi Since opening this ticket I've been made aware of #27581
Was that the final solution or is there more work on circuit breakers to be done on coordinating nodes?

[jimczi]

Was that the final solution or is there more work on circuit breakers to be done on coordinating nodes?

I think so yes, the issue you described here should not happen on master where the maximum number of bucket is set to 10,000 by default.

[markharwood]

While #27581 should address issues with aggregations like DateHistogram with many small buckets we don't currently account for internal memory consumed by more expensive aggregations like cardinality.
For this reason I'm keeping this ticket open to track the problem of accounting for these more complex aggregations.

[howardhuanghua]

@jimczi The maximum number of bucket only limit reduction phase bucket count. So how about if coordinate node received lots of first-phase aggregation bucket result from other data node before going to reduction phase and exploded the coordinate node memory? How do we limit the received temporary data memory?

[howardhuanghua]

Follow "results.consumeResult(result);" may consume lots of memory and how to avoid OOM in huge number of nodes case aggregation request?

    @Override
    public final void onShardSuccess(Result result) {
        successfulOps.incrementAndGet();
        results.consumeResult(result);
        if (logger.isTraceEnabled()) {
            logger.trace("got first-phase result from {}", result != null ? result.getSearchShardTarget() : null);
        }
        // clean a previous error on this shard group (note, this code will be serialized on the same shardIndex value level
        // so its ok concurrency wise to miss potentially the shard failures being created because of another failure
        // in the #addShardFailure, because by definition, it will happen on *another* shardIndex
        AtomicArray<ShardSearchFailure> shardFailures = this.shardFailures.get();
        if (shardFailures != null) {
            shardFailures.set(result.getShardIndex(), null);
        }
    }

[ppf2]

@markharwood @jimczi

we don't currently account for internal memory consumed by more expensive aggregations like cardinality.

Will the new real memory breaker on 7.0 resolve this since it should break based on real heap usage threshold even on coordinating nodes correct? Or do we still plan on implementing a coordinating node specific request breaker in addition to the real memory breaker?

[danielmitterdorfer]

@ppf2 I don't think the real memory circuit breaker is able to reliably prevent this. Circuit breakers are not actively observing system state but rather need to be invoked explicitly at certain points during request handling. Only then they will check current resource usage and potentially reject a request. If a request is past that check and then allocates a lot of memory, also the real memory circuit breaker cannot prevent this situation (although it might detect too high memory usage and rejects other requests that are sent concurrently). Therefore, a circuit breaker that checks (expected) memory usage of aggregations on the coordinating node can make sense.

[nik9000]

This is sort of linked in with a few things that @jimczi and I are working on slowly, in between a bunch of other things. In particular, we now have an accurate accounting of the memory usage of buffered aggregation results which we expect to be the bulk of the "big" stuff in a request on the coordinating node. We're working on refactoring the partial reductions (in #58461). From there we hope to trigger these partial reductions based on memory usage. That isn't quite the same thing as this issue, but it is fairly close.

[jimczi]

From there we hope to trigger these partial reductions based on memory usage. That isn't quite the same thing as this issue, but it is fairly close.

In my mind the follow up of #58461 was to account the memory used by buffered aggs in the circuit breaker so that's closer to what this issue is about.