-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added in-flight cancellation of SearchShardTask based on resource con…
…sumption This feature aims to identify and cancel resource intensive SearchShardTasks if they have breached certain thresholds. This will help in terminating problematic queries which can put nodes in duress and degrade the cluster performance. Signed-off-by: Ketan Verma <ketan9495@gmail.com>
- Loading branch information
Showing
25 changed files
with
1,649 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.common; | ||
|
|
||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * Streak is a data structure that keeps track of the number of consecutive successful events. | ||
| */ | ||
| public class Streak { | ||
| private final AtomicInteger consecutiveSuccessfulEvents = new AtomicInteger(); | ||
|
|
||
| public int record(boolean isSuccessful) { | ||
| if (isSuccessful) { | ||
| return consecutiveSuccessfulEvents.incrementAndGet(); | ||
| } else { | ||
| consecutiveSuccessfulEvents.set(0); | ||
| return 0; | ||
| } | ||
| } | ||
|
|
||
| public int length() { | ||
| return consecutiveSuccessfulEvents.get(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
server/src/main/java/org/opensearch/common/util/MovingAverage.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.common.util; | ||
|
|
||
| /** | ||
| * MovingAverage is used to calculate the moving average of last 'n' observations. | ||
| */ | ||
| public class MovingAverage { | ||
| private final int windowSize; | ||
| private final long[] observations; | ||
|
|
||
| private long count = 0; | ||
| private long sum = 0; | ||
| private double average = 0; | ||
|
|
||
| public MovingAverage(int windowSize) { | ||
| if (windowSize <= 0) { | ||
| throw new IllegalArgumentException("window size must be greater than zero"); | ||
| } | ||
|
|
||
| this.windowSize = windowSize; | ||
| this.observations = new long[windowSize]; | ||
| } | ||
|
|
||
| /** | ||
| * Records a new observation and evicts the n-th last observation. | ||
| */ | ||
| public synchronized double record(long value) { | ||
| long delta = value - observations[(int) (count % observations.length)]; | ||
| observations[(int) (count % observations.length)] = value; | ||
|
|
||
| count++; | ||
| sum += delta; | ||
| average = (double) sum / Math.min(count, observations.length); | ||
| return average; | ||
| } | ||
|
|
||
| public double getAverage() { | ||
| return average; | ||
| } | ||
|
|
||
| public long getCount() { | ||
| return count; | ||
| } | ||
|
|
||
| public boolean isReady() { | ||
| return count >= windowSize; | ||
| } | ||
| } |
95 changes: 95 additions & 0 deletions
95
server/src/main/java/org/opensearch/common/util/TokenBucket.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.common.util; | ||
|
|
||
| import java.util.function.LongSupplier; | ||
|
|
||
| /** | ||
| * TokenBucket is used to limit the number of operations at a constant rate while allowing for short bursts. | ||
| */ | ||
| public class TokenBucket { | ||
| /** | ||
| * Defines a monotonically increasing counter. | ||
| * | ||
| * Usage examples: | ||
| * 1. clock = System::nanoTime can be used to perform rate-limiting per unit time | ||
| * 2. clock = AtomicLong::get can be used to perform rate-limiting per unit number of operations | ||
| */ | ||
| private final LongSupplier clock; | ||
|
|
||
| /** | ||
| * Defines the number of tokens added to the bucket per clock cycle. | ||
| */ | ||
| private final double rate; | ||
|
|
||
| /** | ||
| * Defines the capacity and the maximum number of operations that can be performed per clock cycle before | ||
| * the bucket runs out of tokens. | ||
| */ | ||
| private final double burst; | ||
|
|
||
| private double tokens; | ||
|
|
||
| private long lastRefilledAt; | ||
|
|
||
| public TokenBucket(LongSupplier clock, double rate, double burst) { | ||
| this(clock, rate, burst, burst); | ||
| } | ||
|
|
||
| public TokenBucket(LongSupplier clock, double rate, double burst, double initialTokens) { | ||
| if (rate <= 0.0) { | ||
| throw new IllegalArgumentException("rate must be greater than zero"); | ||
| } | ||
|
|
||
| if (burst <= 0.0) { | ||
| throw new IllegalArgumentException("burst must be greater than zero"); | ||
| } | ||
|
|
||
| this.clock = clock; | ||
| this.rate = rate; | ||
| this.burst = burst; | ||
| this.tokens = initialTokens; | ||
| this.lastRefilledAt = clock.getAsLong(); | ||
| } | ||
|
|
||
| /** | ||
| * Refills the token bucket. | ||
| */ | ||
| private void refill() { | ||
| long now = clock.getAsLong(); | ||
| double incr = (now - lastRefilledAt) * rate; | ||
| tokens = Math.min(tokens + incr, burst); | ||
| lastRefilledAt = now; | ||
| } | ||
|
|
||
| /** | ||
| * If there are enough tokens in the bucket, it requests/deducts 'n' tokens and returns true. | ||
| * Otherwise, returns false and leaves the bucket untouched. | ||
| */ | ||
| public boolean request(double n) { | ||
| if (n <= 0) { | ||
| throw new IllegalArgumentException("requested tokens must be greater than zero"); | ||
| } | ||
|
|
||
| synchronized (this) { | ||
| refill(); | ||
|
|
||
| if (tokens >= n) { | ||
| tokens -= n; | ||
| return true; | ||
| } | ||
|
|
||
| return false; | ||
| } | ||
| } | ||
|
|
||
| public boolean request() { | ||
| return request(1.0); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.