Fix addEmptyBuckets from creating too many buckets when given big extended bounds

[harshavamsi]

Description

Most of the description is in #17702, this PR adds checks before we can create empty buckets.

Before we create empty buckets, we check how many potential buckets would be created and add those to the CircuitBreaker which could either trip or cause max_buckets_exception.

Related Issues

Resolves #17702

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

❌ Gradle check result for 60c7b21: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[jainankitk]

@harshavamsi - Are you still working on this PR?

[sandeshkr419]

Do we need to get rid of this constructor and force the consumers to provide CircuitBreaker as a safety mechanism?
(if this is a breaking change - then we can atleast mark this for deprecation)

[github-actions]

❌ Gradle check result for eaadff0: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for eaadff0: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 88c6733: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 0dc14fa: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 666d610: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

✅ Gradle check result for 666d610: SUCCESS

[harshavamsi]

@jainankitk @bowenlan-amzn could you take a look again, I've addressed the concerns from before

[bowenlan-amzn]

@harshavamsi could add REST tests based on this #17718 (comment)
Without the extended bounds in the query.

[bowenlan-amzn]

I can see we are now checking list.getFirst and getLast key values. However, it seems to me the if condition bounds != null still limit the applicable scenario to when user provide extended bound

[bowenlan-amzn]

I remember we can account for negative value here.

OpenSearch/server/src/main/java/org/opensearch/search/aggregations/MultiBucketConsumerService.java

Lines 165 to 203 in f7b31b0

	public void accept(int value) {
	if (value != 0) {
	count += value;
	if (count > limit) {
	throw new TooManyBucketsException(
	"Trying to create too many buckets. Must be less than or equal to: ["
	+ limit
	+ "] but was ["
	+ count
	+ "]. This limit can be set by changing the ["
	+ MAX_BUCKET_SETTING.getKey()
	+ "] cluster level setting.",
	limit
	);
	}
	}
	callCount.increment();
	// tripping the circuit breaker for other threads in case of concurrent search
	// if the circuit breaker has tripped for one of the threads already, more info
	// can be found on: https://github.com/opensearch-project/OpenSearch/issues/7785
	if (circuitBreakerTripped) {
	throw new CircuitBreakingException(
	"Circuit breaker for this consumer has already been tripped by previous invocations. "
	+ "This can happen in case of concurrent segment search when multiple threads are "
	+ "executing the request and one of the thread has already tripped the circuit breaker",
	breaker.getDurability()
	);
	}
	// check parent circuit breaker every 1024 to (1024 + available processors) calls
	long sum = callCount.sum();
	if ((sum >= 1024) && (sum & 0x3FF) <= availProcessors) {
	try {
	breaker.addEstimateBytesAndMaybeBreak(0, "allocated_buckets");
	} catch (CircuitBreakingException e) {
	circuitBreakerTripped = true;
	throw e;
	}
	}
	}

[bowenlan-amzn]

I realize the multiBucketConsumer of reduce context already has breaker in it. Should we just use that?

OpenSearch/server/src/main/java/org/opensearch/search/aggregations/MultiBucketConsumerService.java

Line 132 in f7b31b0

private final CircuitBreaker breaker;