-
Notifications
You must be signed in to change notification settings - Fork 36
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Stabilize adaptive rate limit by considering current rate (#1027)
* Stabilize adaptive rate limit by considering current rate Signed-off-by: Tomoyuki Morita <[email protected]> * Fix bulk retry condition Signed-off-by: Tomoyuki Morita <[email protected]> * Fix FlintSparkConfSuite Signed-off-by: Tomoyuki Morita <[email protected]> * Fix FlintSparkConfSuite Signed-off-by: Tomoyuki Morita <[email protected]> * Reformat Signed-off-by: Tomoyuki Morita <[email protected]> * Use Queue interface instead of List Signed-off-by: Tomoyuki Morita <[email protected]> * Fix doc for BULK_INITIAL_BACKOFF Signed-off-by: Tomoyuki Morita <[email protected]> * Add retryable http status code Signed-off-by: Tomoyuki Morita <[email protected]> * Fix test Signed-off-by: Tomoyuki Morita <[email protected]> * Fix RequestRateMeter.addDataPoint Signed-off-by: Tomoyuki Morita <[email protected]> --------- Signed-off-by: Tomoyuki Morita <[email protected]>
- Loading branch information
Showing
9 changed files
with
204 additions
and
20 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
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
48 changes: 48 additions & 0 deletions
48
flint-core/src/main/scala/org/opensearch/flint/core/storage/RequestRateMeter.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,48 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.storage; | ||
|
|
||
| import java.util.LinkedList; | ||
| import java.util.List; | ||
| import java.util.Queue; | ||
|
|
||
| /** | ||
| * Track the current request rate based on the past requests within ESTIMATE_RANGE_DURATION_MSEC | ||
| * milliseconds period. | ||
| */ | ||
| public class RequestRateMeter { | ||
| private static final long ESTIMATE_RANGE_DURATION_MSEC = 3000; | ||
|
|
||
| private static class DataPoint { | ||
| long timestamp; | ||
| long requestCount; | ||
| public DataPoint(long timestamp, long requestCount) { | ||
| this.timestamp = timestamp; | ||
| this.requestCount = requestCount; | ||
| } | ||
| } | ||
|
|
||
| private Queue<DataPoint> dataPoints = new LinkedList<>(); | ||
| private long currentSum = 0; | ||
|
|
||
| public synchronized void addDataPoint(long timestamp, long requestCount) { | ||
| dataPoints.add(new DataPoint(timestamp, requestCount)); | ||
| currentSum += requestCount; | ||
| removeOldDataPoints(); | ||
| } | ||
|
|
||
| public synchronized long getCurrentEstimatedRate() { | ||
| removeOldDataPoints(); | ||
| return currentSum * 1000 / ESTIMATE_RANGE_DURATION_MSEC; | ||
| } | ||
|
|
||
| private synchronized void removeOldDataPoints() { | ||
| long curr = System.currentTimeMillis(); | ||
| while (!dataPoints.isEmpty() && dataPoints.peek().timestamp < curr - ESTIMATE_RANGE_DURATION_MSEC) { | ||
| currentSum -= dataPoints.remove().requestCount; | ||
| } | ||
| } | ||
| } |
66 changes: 66 additions & 0 deletions
66
flint-core/src/test/java/org/opensearch/flint/core/storage/RequestRateMeterTest.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,66 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.storage; | ||
|
|
||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
| import static org.junit.jupiter.api.Assertions.*; | ||
|
|
||
| class RequestRateMeterTest { | ||
|
|
||
| private RequestRateMeter requestRateMeter; | ||
|
|
||
| @BeforeEach | ||
| void setUp() { | ||
| requestRateMeter = new RequestRateMeter(); | ||
| } | ||
|
|
||
| @Test | ||
| void testAddDataPoint() { | ||
| long timestamp = System.currentTimeMillis(); | ||
| requestRateMeter.addDataPoint(timestamp, 30); | ||
| assertEquals(10, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
|
|
||
| @Test | ||
| void testAddDataPointRemoveOldDataPoint() { | ||
| long timestamp = System.currentTimeMillis(); | ||
| requestRateMeter.addDataPoint(timestamp - 4000, 30); | ||
| requestRateMeter.addDataPoint(timestamp, 90); | ||
| assertEquals(90 / 3, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
|
|
||
| @Test | ||
| void testRemoveOldDataPoints() { | ||
| long currentTime = System.currentTimeMillis(); | ||
| requestRateMeter.addDataPoint(currentTime - 4000, 30); | ||
| requestRateMeter.addDataPoint(currentTime - 2000, 60); | ||
| requestRateMeter.addDataPoint(currentTime, 90); | ||
|
|
||
| assertEquals((60 + 90)/3, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
|
|
||
| @Test | ||
| void testGetCurrentEstimatedRate() { | ||
| long currentTime = System.currentTimeMillis(); | ||
| requestRateMeter.addDataPoint(currentTime - 2500, 30); | ||
| requestRateMeter.addDataPoint(currentTime - 1500, 60); | ||
| requestRateMeter.addDataPoint(currentTime - 500, 90); | ||
|
|
||
| assertEquals((30 + 60 + 90)/3, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
|
|
||
| @Test | ||
| void testEmptyRateMeter() { | ||
| assertEquals(0, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
|
|
||
| @Test | ||
| void testSingleDataPoint() { | ||
| requestRateMeter.addDataPoint(System.currentTimeMillis(), 30); | ||
| assertEquals(30 / 3, requestRateMeter.getCurrentEstimatedRate()); | ||
| } | ||
| } |
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