forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Cluster manager task throttling feature [Final PR] (opensearch-projec…
…t#4986) Basic Throttler Framework / Exponential Basic back off policy. Add basic thorttler/exponential backoff policy for retry/Defination o… opensearch-project#3527 Changes required in Master node to perform throttling. Master node changes for master task throttling opensearch-project#3882 Changes required in Data node to perform retry on throttling. Data node changes for master task throttling opensearch-project#4204 Provide support for all task type in throttling framework. Onboarding of few task types to throttling opensearch-project#4542 Integration Tests (Fix timeout exception and Add Integ test for Master task throttling opensearch-project#4588 Signed-off-by: Dhwanil Patel <dhwanip@amazon.com>
- Loading branch information
1 parent
d3e95d6
commit 0fb0835
Showing
49 changed files
with
2,059 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
Oops, something went wrong.