-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Backport 2.x] Cluster Manager task throttling (#5041)
* Add basic thorttler/exponential backoff policy for retry/Defination of throttling exception (#3856) * Corrected Java doc for Throttler * Changed the default behaviour of Throttler to return Optional * Removed generics from Throttler and used String as key * Ignore backport / autocut / dependabot branches for gradle checks on push * Master node changes for master task throttling (#3882) * Data node changes for master task throttling (#4204) * Onboarding of few task types to throttling (#4542) * Fix timeout exception and Add Integ test for Master task throttling (#4588) * Complete TODO for version change and removed unused classes(Throttler and Semaphore) (#4846) * Remove V1 version from throttling testcase Signed-off-by: Dhwanil Patel <[email protected]>
- Loading branch information
1 parent
802e693
commit d10bc9f
Showing
49 changed files
with
2,058 additions
and
93 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
214 changes: 214 additions & 0 deletions
214
...nternalClusterTest/java/org/opensearch/clustermanager/ClusterManagerTaskThrottlingIT.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,214 @@ | ||
| /* | ||
| * 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.clustermanager; | ||
|
|
||
| import org.opensearch.action.ActionListener; | ||
| import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest; | ||
| import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest; | ||
| import org.opensearch.cluster.metadata.ProcessClusterEventTimeoutException; | ||
| import org.opensearch.cluster.service.ClusterManagerThrottlingException; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.test.OpenSearchIntegTestCase; | ||
| import org.opensearch.transport.TransportService; | ||
| import org.opensearch.transport.TransportMessageListener; | ||
|
|
||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; | ||
|
|
||
| @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, numDataNodes = 0) | ||
| public class ClusterManagerTaskThrottlingIT extends OpenSearchIntegTestCase { | ||
|
|
||
| /* | ||
| * This integ test will test end-end cluster manager throttling feature for | ||
| * remote cluster manager. | ||
| * | ||
| * It will check the number of request coming to cluster manager node | ||
| * should be total number of requests + throttled requests from cluster manager. | ||
| * This will ensure the end-end feature is working as cluster manager is throwing | ||
| * Throttling exception and data node is performing retries on it. | ||
| * | ||
| */ | ||
| public void testThrottlingForRemoteClusterManager() throws Exception { | ||
| try { | ||
| internalCluster().beforeTest(random()); | ||
| String clusterManagerNode = internalCluster().startClusterManagerOnlyNode(); | ||
| String dataNode = internalCluster().startDataOnlyNode(); | ||
| int throttlingLimit = randomIntBetween(1, 5); | ||
| createIndex("test"); | ||
| setPutMappingThrottlingLimit(throttlingLimit); | ||
|
|
||
| TransportService clusterManagerTransportService = (internalCluster().getInstance(TransportService.class, clusterManagerNode)); | ||
| AtomicInteger requestCountOnClusterManager = new AtomicInteger(); | ||
| AtomicInteger throttledRequest = new AtomicInteger(); | ||
| int totalRequest = randomIntBetween(throttlingLimit, 5 * throttlingLimit); | ||
| CountDownLatch latch = new CountDownLatch(totalRequest); | ||
|
|
||
| clusterManagerTransportService.addMessageListener(new TransportMessageListener() { | ||
| @Override | ||
| public void onRequestReceived(long requestId, String action) { | ||
| if (action.contains("mapping")) { | ||
| requestCountOnClusterManager.incrementAndGet(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onResponseSent(long requestId, String action, Exception error) { | ||
| if (action.contains("mapping")) { | ||
| throttledRequest.incrementAndGet(); | ||
| assertEquals(ClusterManagerThrottlingException.class, error.getClass()); | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| ActionListener listener = new ActionListener() { | ||
| @Override | ||
| public void onResponse(Object o) { | ||
| latch.countDown(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| latch.countDown(); | ||
| throw new AssertionError(e); | ||
| } | ||
| }; | ||
|
|
||
| executePutMappingRequests(totalRequest, dataNode, listener); | ||
| latch.await(); | ||
|
|
||
| assertEquals(totalRequest + throttledRequest.get(), requestCountOnClusterManager.get()); | ||
| assertBusy( | ||
| () -> { assertEquals(clusterService().getMasterService().numberOfThrottledPendingTasks(), throttledRequest.get()); } | ||
| ); | ||
| } finally { | ||
| clusterSettingCleanUp(); | ||
| } | ||
| } | ||
|
|
||
| /* | ||
| * This will test the throttling feature for single node. | ||
| * | ||
| * Here we will assert the client behaviour that client's request is not | ||
| * failed, i.e. Throttling exception is not passed to the client. | ||
| * Data node will internally do the retry and request should pass. | ||
| * | ||
| */ | ||
| public void testThrottlingForSingleNode() throws Exception { | ||
| try { | ||
| internalCluster().beforeTest(random()); | ||
| String node = internalCluster().startNode(); | ||
| int throttlingLimit = randomIntBetween(1, 5); | ||
| createIndex("test"); | ||
| setPutMappingThrottlingLimit(throttlingLimit); | ||
|
|
||
| AtomicInteger successfulRequest = new AtomicInteger(); | ||
| int totalRequest = randomIntBetween(throttlingLimit, 3 * throttlingLimit); | ||
| CountDownLatch latch = new CountDownLatch(totalRequest); | ||
|
|
||
| ActionListener listener = new ActionListener() { | ||
| @Override | ||
| public void onResponse(Object o) { | ||
| latch.countDown(); | ||
| successfulRequest.incrementAndGet(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| latch.countDown(); | ||
| throw new AssertionError(e); | ||
| } | ||
| }; | ||
| executePutMappingRequests(totalRequest, node, listener); | ||
|
|
||
| latch.await(); | ||
| assertEquals(totalRequest, successfulRequest.get()); | ||
| } finally { | ||
| clusterSettingCleanUp(); | ||
| } | ||
| } | ||
|
|
||
| /* | ||
| * This will test the timeout of tasks during throttling. | ||
| * | ||
| * Here we will assert the client behaviour that client's request is not | ||
| * failed with throttling exception but timeout exception. | ||
| * It also verifies that if limit is set to 0, all tasks are getting timedout. | ||
| */ | ||
|
|
||
| public void testTimeoutWhileThrottling() throws Exception { | ||
| try { | ||
| internalCluster().beforeTest(random()); | ||
| String node = internalCluster().startNode(); | ||
| int throttlingLimit = 0; // throttle all the tasks | ||
| createIndex("test"); | ||
| setPutMappingThrottlingLimit(throttlingLimit); | ||
|
|
||
| AtomicInteger timedoutRequest = new AtomicInteger(); | ||
| int totalRequest = randomIntBetween(1, 5); | ||
| CountDownLatch latch = new CountDownLatch(totalRequest); | ||
|
|
||
| ActionListener listener = new ActionListener() { | ||
| @Override | ||
| public void onResponse(Object o) { | ||
| latch.countDown(); | ||
| throw new AssertionError("Request should not succeed"); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(Exception e) { | ||
| latch.countDown(); | ||
| assertTrue(e instanceof ProcessClusterEventTimeoutException); | ||
| timedoutRequest.incrementAndGet(); | ||
| } | ||
| }; | ||
| executePutMappingRequests(totalRequest, node, listener); | ||
|
|
||
| latch.await(); | ||
| assertEquals(totalRequest, timedoutRequest.get()); // verifying all requests were timed out with 0 throttling limit | ||
| } finally { | ||
| clusterSettingCleanUp(); | ||
| } | ||
| } | ||
|
|
||
| private void executePutMappingRequests(int totalRequest, String node, ActionListener listener) throws Exception { | ||
| Thread[] threads = new Thread[totalRequest]; | ||
| for (int i = 0; i < totalRequest; i++) { | ||
| PutMappingRequest putMappingRequest = new PutMappingRequest("test").source("field" + i, "type=text"); | ||
| threads[i] = new Thread(new Runnable() { | ||
| @Override | ||
| public void run() { | ||
| internalCluster().client(node).admin().indices().putMapping(putMappingRequest, listener); | ||
| } | ||
| }); | ||
| } | ||
| for (int i = 0; i < totalRequest; i++) { | ||
| threads[i].run(); | ||
| } | ||
| for (int i = 0; i < totalRequest; i++) { | ||
| threads[i].join(); | ||
| } | ||
| } | ||
|
|
||
| private void setPutMappingThrottlingLimit(int throttlingLimit) { | ||
| ClusterUpdateSettingsRequest settingsRequest = new ClusterUpdateSettingsRequest(); | ||
| Settings settings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", throttlingLimit).build(); | ||
| settingsRequest.transientSettings(settings); | ||
| assertAcked(client().admin().cluster().updateSettings(settingsRequest).actionGet()); | ||
| } | ||
|
|
||
| private void clusterSettingCleanUp() { | ||
| // We need to remove the throttling limit from setting as part of test cleanup | ||
| ClusterUpdateSettingsRequest settingsRequest = new ClusterUpdateSettingsRequest(); | ||
| Settings settings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", (String) null).build(); | ||
| settingsRequest.transientSettings(settings); | ||
| assertAcked(client().admin().cluster().updateSettings(settingsRequest).actionGet()); | ||
| } | ||
| } |
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
Oops, something went wrong.