diff --git a/CHANGELOG.md b/CHANGELOG.md index 333b1cd19b404..b69c33df3b6a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Add success and failure metrics for async shard fetch ([#15976](https://github.com/opensearch-project/OpenSearch/pull/15976)) - Add new metric REMOTE_STORE to NodeStats API response ([#15611](https://github.com/opensearch-project/OpenSearch/pull/15611)) - [S3 Repository] Change default retry mechanism of s3 clients to Standard Mode ([#15978](https://github.com/opensearch-project/OpenSearch/pull/15978)) +- Add changes to block calls in cat shards, indices and segments based on dynamic limit settings ([#15986](https://github.com/opensearch-project/OpenSearch/pull/15986)) ### Dependencies - Bump `com.azure:azure-identity` from 1.13.0 to 1.13.2 ([#15578](https://github.com/opensearch-project/OpenSearch/pull/15578)) diff --git a/server/src/main/java/org/opensearch/OpenSearchServerException.java b/server/src/main/java/org/opensearch/OpenSearchServerException.java index b0ab62259d5c0..0ee3debb4819a 100644 --- a/server/src/main/java/org/opensearch/OpenSearchServerException.java +++ b/server/src/main/java/org/opensearch/OpenSearchServerException.java @@ -14,6 +14,7 @@ import static org.opensearch.Version.V_2_10_0; import static org.opensearch.Version.V_2_13_0; import static org.opensearch.Version.V_2_17_0; +import static org.opensearch.Version.V_2_18_0; import static org.opensearch.Version.V_2_1_0; import static org.opensearch.Version.V_2_4_0; import static org.opensearch.Version.V_2_5_0; @@ -1210,6 +1211,14 @@ public static void registerExceptions() { V_2_17_0 ) ); + registerExceptionHandle( + new OpenSearchExceptionHandle( + org.opensearch.common.breaker.ResponseLimitBreachedException.class, + org.opensearch.common.breaker.ResponseLimitBreachedException::new, + 175, + V_2_18_0 + ) + ); registerExceptionHandle( new OpenSearchExceptionHandle( org.opensearch.cluster.block.IndexCreateBlockException.class, diff --git a/server/src/main/java/org/opensearch/action/ActionModule.java b/server/src/main/java/org/opensearch/action/ActionModule.java index 937d7509fe68c..98bcd6ba9c3f8 100644 --- a/server/src/main/java/org/opensearch/action/ActionModule.java +++ b/server/src/main/java/org/opensearch/action/ActionModule.java @@ -300,6 +300,7 @@ import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.common.NamedRegistry; import org.opensearch.common.annotation.PublicApi; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.inject.AbstractModule; import org.opensearch.common.inject.TypeLiteral; import org.opensearch.common.inject.multibindings.MapBinder; @@ -531,6 +532,7 @@ public class ActionModule extends AbstractModule { private final RequestValidators indicesAliasesRequestRequestValidators; private final ThreadPool threadPool; private final ExtensionsManager extensionsManager; + private final ResponseLimitSettings responseLimitSettings; public ActionModule( Settings settings, @@ -583,6 +585,7 @@ public ActionModule( ); restController = new RestController(headers, restWrapper, nodeClient, circuitBreakerService, usageService); + responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); } public Map> getActions() { @@ -968,8 +971,8 @@ public void initRestHandlers(Supplier nodesInCluster) { registerHandler.accept(new RestClusterManagerAction()); registerHandler.accept(new RestNodesAction()); registerHandler.accept(new RestTasksAction(nodesInCluster)); - registerHandler.accept(new RestIndicesAction()); - registerHandler.accept(new RestSegmentsAction()); + registerHandler.accept(new RestIndicesAction(responseLimitSettings)); + registerHandler.accept(new RestSegmentsAction(responseLimitSettings)); // Fully qualified to prevent interference with rest.action.count.RestCountAction registerHandler.accept(new org.opensearch.rest.action.cat.RestCountAction()); // Fully qualified to prevent interference with rest.action.indices.RestRecoveryAction @@ -989,7 +992,7 @@ public void initRestHandlers(Supplier nodesInCluster) { registerHandler.accept(new RestTemplatesAction()); // LIST API - registerHandler.accept(new RestIndicesListAction()); + registerHandler.accept(new RestIndicesListAction(responseLimitSettings)); // Point in time API registerHandler.accept(new RestCreatePitAction()); @@ -1060,6 +1063,8 @@ protected void configure() { // register dynamic ActionType -> transportAction Map used by NodeClient bind(DynamicActionRegistry.class).toInstance(dynamicActionRegistry); + + bind(ResponseLimitSettings.class).toInstance(responseLimitSettings); } public ActionFilters getActionFilters() { diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java index 49299777db8ae..76aa25b9a96b5 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/shards/CatShardsRequest.java @@ -27,11 +27,13 @@ public class CatShardsRequest extends ClusterManagerNodeReadRequest headers) { return new ClusterAdminTask(id, type, action, parentTaskId, headers, this.cancelAfterTimeInterval); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java index 224d3cbc5f10a..6b073a16f7d28 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/shards/TransportCatShardsAction.java @@ -16,6 +16,8 @@ import org.opensearch.action.support.HandledTransportAction; import org.opensearch.action.support.TimeoutTaskCancellationUtility; import org.opensearch.client.node.NodeClient; +import org.opensearch.common.breaker.ResponseLimitBreachedException; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.inject.Inject; import org.opensearch.core.action.ActionListener; import org.opensearch.core.action.NotifyOnceListener; @@ -23,6 +25,10 @@ import org.opensearch.tasks.Task; import org.opensearch.transport.TransportService; +import java.util.Objects; + +import static org.opensearch.common.breaker.ResponseLimitSettings.LimitEntity.SHARDS; + /** * Perform cat shards action * @@ -31,11 +37,18 @@ public class TransportCatShardsAction extends HandledTransportAction { private final NodeClient client; + private final ResponseLimitSettings responseLimitSettings; @Inject - public TransportCatShardsAction(NodeClient client, TransportService transportService, ActionFilters actionFilters) { + public TransportCatShardsAction( + NodeClient client, + TransportService transportService, + ActionFilters actionFilters, + ResponseLimitSettings responseLimitSettings + ) { super(CatShardsAction.NAME, transportService, actionFilters, CatShardsRequest::new); this.client = client; + this.responseLimitSettings = responseLimitSettings; } @Override @@ -73,6 +86,7 @@ protected void innerOnFailure(Exception e) { client.admin().cluster().state(clusterStateRequest, new ActionListener() { @Override public void onResponse(ClusterStateResponse clusterStateResponse) { + validateRequestLimit(shardsRequest, clusterStateResponse, cancellableListener); catShardsResponse.setClusterStateResponse(clusterStateResponse); IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest(); indicesStatsRequest.setShouldCancelOnTimeout(true); @@ -107,4 +121,19 @@ public void onFailure(Exception e) { } } + + private void validateRequestLimit( + final CatShardsRequest shardsRequest, + final ClusterStateResponse clusterStateResponse, + final ActionListener listener + ) { + if (shardsRequest.isRequestLimitCheckSupported() + && Objects.nonNull(clusterStateResponse) + && Objects.nonNull(clusterStateResponse.getState())) { + int limit = responseLimitSettings.getCatShardsResponseLimit(); + if (ResponseLimitSettings.isResponseLimitBreached(clusterStateResponse.getState().getRoutingTable(), SHARDS, limit)) { + listener.onFailure(new ResponseLimitBreachedException("Too many shards requested.", limit, SHARDS)); + } + } + } } diff --git a/server/src/main/java/org/opensearch/common/breaker/ResponseLimitBreachedException.java b/server/src/main/java/org/opensearch/common/breaker/ResponseLimitBreachedException.java new file mode 100644 index 0000000000000..db6785067edf6 --- /dev/null +++ b/server/src/main/java/org/opensearch/common/breaker/ResponseLimitBreachedException.java @@ -0,0 +1,66 @@ +/* + * 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.breaker; + +import org.opensearch.OpenSearchException; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.core.xcontent.XContentBuilder; + +import java.io.IOException; + +/** + * Thrown when api response breaches threshold limit. + * + * @opensearch.internal + */ +public class ResponseLimitBreachedException extends OpenSearchException { + + private final int responseLimit; + private final ResponseLimitSettings.LimitEntity limitEntity; + + public ResponseLimitBreachedException(StreamInput in) throws IOException { + super(in); + responseLimit = in.readVInt(); + limitEntity = in.readEnum(ResponseLimitSettings.LimitEntity.class); + } + + public ResponseLimitBreachedException(String msg, int responseLimit, ResponseLimitSettings.LimitEntity limitEntity) { + super(msg); + this.responseLimit = responseLimit; + this.limitEntity = limitEntity; + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + super.writeTo(out); + out.writeVInt(responseLimit); + out.writeEnum(limitEntity); + } + + public int getResponseLimit() { + return responseLimit; + } + + public ResponseLimitSettings.LimitEntity getLimitEntity() { + return limitEntity; + } + + @Override + public RestStatus status() { + return RestStatus.TOO_MANY_REQUESTS; + } + + @Override + protected void metadataToXContent(XContentBuilder builder, Params params) throws IOException { + builder.field("response_limit", responseLimit); + builder.field("limit_entity", limitEntity); + } +} diff --git a/server/src/main/java/org/opensearch/common/breaker/ResponseLimitSettings.java b/server/src/main/java/org/opensearch/common/breaker/ResponseLimitSettings.java new file mode 100644 index 0000000000000..8eb47c9aaf147 --- /dev/null +++ b/server/src/main/java/org/opensearch/common/breaker/ResponseLimitSettings.java @@ -0,0 +1,180 @@ +/* + * 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.breaker; + +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.routing.IndexRoutingTable; +import org.opensearch.cluster.routing.IndexShardRoutingTable; +import org.opensearch.cluster.routing.RoutingTable; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.settings.Settings; +import org.opensearch.rest.action.cat.RestIndicesAction; +import org.opensearch.rest.action.cat.RestSegmentsAction; +import org.opensearch.rest.action.cat.RestShardsAction; + +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; + +/** + * Class to define dynamic settings for putting response limits on the actions and methods to evaluate if block is required. + */ +public class ResponseLimitSettings { + + /** + * Enum to represent entity against which we need to perform limit checks. + */ + public enum LimitEntity { + INDICES, + SHARDS + } + + private volatile int catIndicesResponseLimit; + private volatile int catShardsResponseLimit; + private volatile int catSegmentsResponseLimit; + + /** + * Setting to enable response limit on {@link RestIndicesAction}. The limit will be applied on number of indices. + */ + public static final Setting CAT_INDICES_RESPONSE_LIMIT_SETTING = Setting.intSetting( + "cat.indices.response.limit.number_of_indices", + -1, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + /** + * Setting to enable response limit on {@link RestShardsAction}. The limit will be applied on number of shards. + */ + public static final Setting CAT_SHARDS_RESPONSE_LIMIT_SETTING = Setting.intSetting( + "cat.shards.response.limit.number_of_shards", + -1, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + /** + * Setting to enable response limit on {@link RestSegmentsAction}. The limit will be applied on number of indices. + */ + public static final Setting CAT_SEGMENTS_RESPONSE_LIMIT_SETTING = Setting.intSetting( + "cat.segments.response.limit.number_of_indices", + -1, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + public ResponseLimitSettings(ClusterSettings clusterSettings, Settings settings) { + setCatShardsResponseLimit(CAT_SHARDS_RESPONSE_LIMIT_SETTING.get(settings)); + setCatIndicesResponseLimit(CAT_INDICES_RESPONSE_LIMIT_SETTING.get(settings)); + setCatSegmentsResponseLimit(CAT_SEGMENTS_RESPONSE_LIMIT_SETTING.get(settings)); + + clusterSettings.addSettingsUpdateConsumer(CAT_SHARDS_RESPONSE_LIMIT_SETTING, this::setCatShardsResponseLimit); + clusterSettings.addSettingsUpdateConsumer(CAT_INDICES_RESPONSE_LIMIT_SETTING, this::setCatIndicesResponseLimit); + clusterSettings.addSettingsUpdateConsumer(CAT_SEGMENTS_RESPONSE_LIMIT_SETTING, this::setCatSegmentsResponseLimit); + } + + /** + * Method to check if the response limit has reached for an action. + * The limits are controlled via dynamic settings. + * + * @param metadata {@link Metadata} + * @param limitEntity {@link LimitEntity} + * @param limit Integer limit on block entity + * @return True/False + */ + public static boolean isResponseLimitBreached(final Metadata metadata, final LimitEntity limitEntity, final int limit) { + if (Objects.isNull(metadata) || limit <= 0) return false; + if (limitEntity == LimitEntity.INDICES) { + int indicesCount = getTotalIndicesFromMetadata.apply(metadata); + return indicesCount > limit; + } else { + throw new IllegalArgumentException("Unsupported limit entity [" + limitEntity + "]"); + } + } + + /** + * Method to check if the response limit has reached for an action. + * The limits are controlled via dynamic settings. + * + * @param routingTable {@link RoutingTable} + * @param limitEntity {@link LimitEntity} + * @param limit Integer limit on block entity + * @return True/False + */ + public static boolean isResponseLimitBreached(final RoutingTable routingTable, final LimitEntity limitEntity, final int limit) { + if (Objects.isNull(routingTable) || limit <= 0) return false; + if (Objects.isNull(limitEntity)) { + throw new IllegalArgumentException("Limit entity cannot be null"); + } + switch (limitEntity) { + case INDICES: + int indicesCount = getTotalIndicesFromRoutingTable.apply(routingTable); + if (indicesCount > limit) return true; + break; + case SHARDS: + if (isShardsLimitBreached(routingTable, limit)) return true; + break; + default: + throw new IllegalArgumentException("Unsupported limit entity [" + limitEntity + "]"); + } + return false; + } + + private static boolean isShardsLimitBreached(final RoutingTable routingTable, final int limit) { + final Map indexRoutingTableMap = routingTable.getIndicesRouting(); + int totalShards = 0; + for (final Map.Entry entry : indexRoutingTableMap.entrySet()) { + for (final Map.Entry indexShardRoutingTableEntry : entry.getValue().getShards().entrySet()) { + totalShards += indexShardRoutingTableEntry.getValue().getShards().size(); + // Fail fast if limit value is breached and avoid unnecessary computation. + if (totalShards > limit) return true; + } + } + return false; + } + + private void setCatShardsResponseLimit(final int catShardsResponseLimit) { + this.catShardsResponseLimit = catShardsResponseLimit; + } + + private void setCatIndicesResponseLimit(final int catIndicesResponseLimit) { + this.catIndicesResponseLimit = catIndicesResponseLimit; + } + + private void setCatSegmentsResponseLimit(final int catSegmentsResponseLimit) { + this.catSegmentsResponseLimit = catSegmentsResponseLimit; + } + + public int getCatShardsResponseLimit() { + return this.catShardsResponseLimit; + } + + public int getCatIndicesResponseLimit() { + return this.catIndicesResponseLimit; + } + + public int getCatSegmentsResponseLimit() { + return this.catSegmentsResponseLimit; + } + + static Function getTotalIndicesFromMetadata = (metadata) -> { + if (Objects.nonNull(metadata) && Objects.nonNull(metadata.getIndices())) { + return metadata.getIndices().size(); + } + return 0; + }; + + static Function getTotalIndicesFromRoutingTable = (routingTable) -> { + if (Objects.nonNull(routingTable) && Objects.nonNull(routingTable.getIndicesRouting())) { + return routingTable.getIndicesRouting().size(); + } + return 0; + }; +} diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index ecdd23530c648..e18c4dabb29eb 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -84,6 +84,7 @@ import org.opensearch.cluster.service.ClusterManagerTaskThrottler; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.annotation.PublicApi; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.cache.CacheType; import org.opensearch.common.cache.settings.CacheSettings; import org.opensearch.common.cache.store.settings.OpenSearchOnHeapCacheSettings; @@ -796,7 +797,12 @@ public void apply(Settings value, Settings current, Settings previous) { WorkloadManagementSettings.NODE_LEVEL_CPU_REJECTION_THRESHOLD, WorkloadManagementSettings.NODE_LEVEL_CPU_CANCELLATION_THRESHOLD, WorkloadManagementSettings.NODE_LEVEL_MEMORY_REJECTION_THRESHOLD, - WorkloadManagementSettings.NODE_LEVEL_MEMORY_CANCELLATION_THRESHOLD + WorkloadManagementSettings.NODE_LEVEL_MEMORY_CANCELLATION_THRESHOLD, + + // Settings to be used for limiting rest requests + ResponseLimitSettings.CAT_INDICES_RESPONSE_LIMIT_SETTING, + ResponseLimitSettings.CAT_SHARDS_RESPONSE_LIMIT_SETTING, + ResponseLimitSettings.CAT_SEGMENTS_RESPONSE_LIMIT_SETTING ) ) ); diff --git a/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java b/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java index 6f4e060363bfb..506f5f1529776 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/AbstractCatAction.java @@ -33,6 +33,7 @@ import org.opensearch.client.node.NodeClient; import org.opensearch.common.Table; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.io.Streams; import org.opensearch.common.io.UTF8StreamWriter; import org.opensearch.core.common.io.stream.BytesStream; @@ -98,4 +99,12 @@ protected Set responseParams() { return RESPONSE_PARAMS; } + /** + * Method to check if limits defined in {@link ResponseLimitSettings} are applicable to an action. + * + * @return True / False status + */ + protected boolean isRequestLimitCheckSupported() { + return false; + } } diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java index 1e76008ff8c64..7562e09ddd704 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java @@ -50,6 +50,8 @@ import org.opensearch.cluster.health.ClusterIndexHealth; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.common.Table; +import org.opensearch.common.breaker.ResponseLimitBreachedException; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.collect.Tuple; import org.opensearch.common.logging.DeprecationLogger; import org.opensearch.common.settings.Settings; @@ -86,6 +88,7 @@ import static java.util.Arrays.asList; import static java.util.Collections.unmodifiableList; import static org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest.DEFAULT_CLUSTER_MANAGER_NODE_TIMEOUT; +import static org.opensearch.common.breaker.ResponseLimitSettings.LimitEntity.INDICES; import static org.opensearch.rest.RestRequest.Method.GET; /** @@ -102,6 +105,12 @@ public class RestIndicesAction extends AbstractListAction { private static final String DUPLICATE_PARAMETER_ERROR_MESSAGE = "Please only use one of the request parameters [master_timeout, cluster_manager_timeout]."; + private final ResponseLimitSettings responseLimitSettings; + + public RestIndicesAction(ResponseLimitSettings responseLimitSettings) { + this.responseLimitSettings = responseLimitSettings; + } + @Override public List routes() { return unmodifiableList(asList(new Route(GET, "/_cat/indices"), new Route(GET, "/_cat/indices/{index}"))); @@ -123,6 +132,11 @@ protected void documentation(StringBuilder sb) { sb.append("/_cat/indices/{index}\n"); } + @Override + public boolean isRequestLimitCheckSupported() { + return true; + } + @Override public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) { final String[] indices = Strings.splitStringByCommaToArray(request.param("index")); @@ -162,7 +176,6 @@ public void onResponse(final GetSettingsResponse getSettingsResponse) { // type of request in the presence of security plugins (looking at you, ClusterHealthRequest), so // force the IndicesOptions for all the sub-requests to be as inclusive as possible. final IndicesOptions subRequestIndicesOptions = IndicesOptions.lenientExpandHidden(); - // Indices that were successfully resolved during the get settings request might be deleted when the // subsequent cluster state, cluster health and indices stats requests execute. We have to distinguish two cases: // 1) the deleted index was explicitly passed as parameter to the /_cat/indices request. In this case we @@ -181,6 +194,7 @@ public void onResponse(final GetSettingsResponse getSettingsResponse) { new ActionListener() { @Override public void onResponse(ClusterStateResponse clusterStateResponse) { + validateRequestLimit(clusterStateResponse, listener); IndexPaginationStrategy paginationStrategy = getPaginationStrategy(clusterStateResponse); // For non-paginated queries, indicesToBeQueried would be same as indices retrieved from // rest request and unresolved, while for paginated queries, it would be a list of indices @@ -234,6 +248,15 @@ public void onFailure(final Exception e) { } + private void validateRequestLimit(final ClusterStateResponse clusterStateResponse, final ActionListener listener) { + if (isRequestLimitCheckSupported() && Objects.nonNull(clusterStateResponse) && Objects.nonNull(clusterStateResponse.getState())) { + int limit = responseLimitSettings.getCatIndicesResponseLimit(); + if (ResponseLimitSettings.isResponseLimitBreached(clusterStateResponse.getState().getMetadata(), INDICES, limit)) { + listener.onFailure(new ResponseLimitBreachedException("Too many indices requested.", limit, INDICES)); + } + } + } + /** * We're using the Get Settings API here to resolve the authorized indices for the user. * This is because the Cluster State and Cluster Health APIs do not filter output based diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestSegmentsAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestSegmentsAction.java index b88af4ac3eeed..69216cb46492d 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestSegmentsAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestSegmentsAction.java @@ -42,6 +42,8 @@ import org.opensearch.client.node.NodeClient; import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.common.Table; +import org.opensearch.common.breaker.ResponseLimitBreachedException; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.logging.DeprecationLogger; import org.opensearch.core.common.Strings; import org.opensearch.index.engine.Segment; @@ -52,9 +54,11 @@ import java.util.List; import java.util.Map; +import java.util.Objects; import static java.util.Arrays.asList; import static java.util.Collections.unmodifiableList; +import static org.opensearch.common.breaker.ResponseLimitSettings.LimitEntity.INDICES; import static org.opensearch.rest.RestRequest.Method.GET; /** @@ -66,6 +70,12 @@ public class RestSegmentsAction extends AbstractCatAction { private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(RestSegmentsAction.class); + private final ResponseLimitSettings responseLimitSettings; + + public RestSegmentsAction(ResponseLimitSettings responseLimitSettings) { + this.responseLimitSettings = responseLimitSettings; + } + @Override public List routes() { return unmodifiableList(asList(new Route(GET, "/_cat/segments"), new Route(GET, "/_cat/segments/{index}"))); @@ -81,6 +91,11 @@ public boolean allowSystemIndexAccessByDefault() { return true; } + @Override + public boolean isRequestLimitCheckSupported() { + return true; + } + @Override public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) { final String[] indices = Strings.splitStringByCommaToArray(request.param("index")); @@ -96,6 +111,7 @@ public RestChannelConsumer doCatRequest(final RestRequest request, final NodeCli return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener(channel) { @Override public void processResponse(final ClusterStateResponse clusterStateResponse) { + validateRequestLimit(clusterStateResponse); final IndicesSegmentsRequest indicesSegmentsRequest = new IndicesSegmentsRequest(); indicesSegmentsRequest.indices(indices); client.admin().indices().segments(indicesSegmentsRequest, new RestResponseListener(channel) { @@ -110,6 +126,15 @@ public RestResponse buildResponse(final IndicesSegmentResponse indicesSegmentRes }); } + private void validateRequestLimit(final ClusterStateResponse clusterStateResponse) { + if (isRequestLimitCheckSupported() && Objects.nonNull(clusterStateResponse) && Objects.nonNull(clusterStateResponse.getState())) { + int limit = responseLimitSettings.getCatSegmentsResponseLimit(); + if (ResponseLimitSettings.isResponseLimitBreached(clusterStateResponse.getState().getRoutingTable(), INDICES, limit)) { + throw new ResponseLimitBreachedException("Segments from too many indices requested.", limit, INDICES); + } + } + } + @Override protected void documentation(StringBuilder sb) { sb.append("/_cat/segments\n"); diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java index a7ad5fe6c14a3..c8b4e7472927e 100644 --- a/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java +++ b/server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java @@ -105,6 +105,11 @@ protected void documentation(StringBuilder sb) { sb.append("/_cat/shards/{index}\n"); } + @Override + public boolean isRequestLimitCheckSupported() { + return true; + } + @Override public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) { final String[] indices = Strings.splitStringByCommaToArray(request.param("index")); @@ -113,6 +118,7 @@ public RestChannelConsumer doCatRequest(final RestRequest request, final NodeCli shardsRequest.clusterManagerNodeTimeout(request.paramAsTime("cluster_manager_timeout", shardsRequest.clusterManagerNodeTimeout())); shardsRequest.setCancelAfterTimeInterval(request.paramAsTime("cancel_after_time_interval", NO_TIMEOUT)); shardsRequest.setIndices(indices); + shardsRequest.setRequestLimitCheckSupported(isRequestLimitCheckSupported()); parseDeprecatedMasterTimeoutParameter(shardsRequest, request, deprecationLogger, getName()); return channel -> client.execute(CatShardsAction.INSTANCE, shardsRequest, new RestResponseListener(channel) { @Override diff --git a/server/src/main/java/org/opensearch/rest/action/list/RestIndicesListAction.java b/server/src/main/java/org/opensearch/rest/action/list/RestIndicesListAction.java index ad5c58c86ce90..971cab0535be9 100644 --- a/server/src/main/java/org/opensearch/rest/action/list/RestIndicesListAction.java +++ b/server/src/main/java/org/opensearch/rest/action/list/RestIndicesListAction.java @@ -9,6 +9,7 @@ package org.opensearch.rest.action.list; import org.opensearch.action.admin.cluster.state.ClusterStateResponse; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.Settings; import org.opensearch.rest.RestRequest; @@ -35,6 +36,10 @@ public class RestIndicesListAction extends RestIndicesAction { private static final int MAX_SUPPORTED_LIST_INDICES_PAGE_SIZE = 5000; private static final int DEFAULT_LIST_INDICES_PAGE_SIZE = 500; + public RestIndicesListAction(final ResponseLimitSettings responseLimitSettings) { + super(responseLimitSettings); + } + @Override public List routes() { return unmodifiableList(asList(new Route(GET, "/_list/indices"), new Route(GET, "/_list/indices/{index}"))); @@ -70,6 +75,11 @@ protected PageParams validateAndGetPageParams(RestRequest restRequest) { return pageParams; } + @Override + public boolean isRequestLimitCheckSupported() { + return false; + } + protected int defaultPageSize() { return DEFAULT_LIST_INDICES_PAGE_SIZE; } diff --git a/server/src/test/java/org/opensearch/ExceptionSerializationTests.java b/server/src/test/java/org/opensearch/ExceptionSerializationTests.java index 2e4a2d7bdd59c..c1972daeab6d3 100644 --- a/server/src/test/java/org/opensearch/ExceptionSerializationTests.java +++ b/server/src/test/java/org/opensearch/ExceptionSerializationTests.java @@ -65,6 +65,7 @@ import org.opensearch.cluster.routing.UnsupportedWeightedRoutingStateException; import org.opensearch.cluster.service.ClusterManagerThrottlingException; import org.opensearch.common.UUIDs; +import org.opensearch.common.breaker.ResponseLimitBreachedException; import org.opensearch.common.collect.Tuple; import org.opensearch.common.io.PathUtils; import org.opensearch.common.io.stream.BytesStreamOutput; @@ -898,6 +899,7 @@ public void testIds() { ids.put(172, ViewNotFoundException.class); ids.put(173, ViewAlreadyExistsException.class); ids.put(174, InvalidIndexContextException.class); + ids.put(175, ResponseLimitBreachedException.class); ids.put(10001, IndexCreateBlockException.class); Map, Integer> reverse = new HashMap<>(); diff --git a/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java b/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java index c7e1039686cc9..26c4670b4288c 100644 --- a/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java +++ b/server/src/test/java/org/opensearch/action/RenamedTimeoutRequestParameterTests.java @@ -11,7 +11,9 @@ import org.opensearch.OpenSearchParseException; import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest; import org.opensearch.client.node.NodeClient; +import org.opensearch.common.breaker.ResponseLimitSettings; import org.opensearch.common.logging.DeprecationLogger; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.settings.SettingsFilter; import org.opensearch.core.common.bytes.BytesArray; @@ -155,7 +157,10 @@ public void testCatAllocation() { } public void testCatIndices() { - RestIndicesAction action = new RestIndicesAction(); + ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + Settings settings = Settings.builder().build(); + ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + RestIndicesAction action = new RestIndicesAction(responseLimitSettings); Exception e = assertThrows(OpenSearchParseException.class, () -> action.doCatRequest(getRestRequestWithBothParams(), client)); assertThat(e.getMessage(), containsString(DUPLICATE_PARAMETER_ERROR_MESSAGE)); assertWarnings(MASTER_TIMEOUT_DEPRECATED_MESSAGE); @@ -239,7 +244,10 @@ public void testCatThreadPool() { } public void testCatSegments() { - RestSegmentsAction action = new RestSegmentsAction(); + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final Settings settings = Settings.builder().build(); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + RestSegmentsAction action = new RestSegmentsAction(responseLimitSettings); Exception e = assertThrows(OpenSearchParseException.class, () -> action.doCatRequest(getRestRequestWithBothParams(), client)); assertThat(e.getMessage(), containsString(DUPLICATE_PARAMETER_ERROR_MESSAGE)); assertWarnings(MASTER_TIMEOUT_DEPRECATED_MESSAGE); diff --git a/server/src/test/java/org/opensearch/common/breaker/ResponseLimitSettingsTests.java b/server/src/test/java/org/opensearch/common/breaker/ResponseLimitSettingsTests.java new file mode 100644 index 0000000000000..4633d055b3505 --- /dev/null +++ b/server/src/test/java/org/opensearch/common/breaker/ResponseLimitSettingsTests.java @@ -0,0 +1,228 @@ +/* + * 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.breaker; + +import org.opensearch.Version; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.routing.IndexRoutingTable; +import org.opensearch.cluster.routing.IndexShardRoutingTable; +import org.opensearch.cluster.routing.RoutingTable; +import org.opensearch.cluster.routing.ShardRouting; +import org.opensearch.cluster.routing.ShardRoutingState; +import org.opensearch.cluster.routing.TestShardRouting; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.Index; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.test.OpenSearchTestCase; + +import java.util.HashMap; +import java.util.Map; + +import static org.opensearch.common.breaker.ResponseLimitSettings.CAT_INDICES_RESPONSE_LIMIT_SETTING; +import static org.opensearch.common.breaker.ResponseLimitSettings.CAT_SEGMENTS_RESPONSE_LIMIT_SETTING; +import static org.opensearch.common.breaker.ResponseLimitSettings.CAT_SHARDS_RESPONSE_LIMIT_SETTING; + +public class ResponseLimitSettingsTests extends OpenSearchTestCase { + + public void testIsResponseLimitBreachedForNullMetadataExpectNotBreached() { + final Settings settings = Settings.builder().build(); + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + final boolean breached = ResponseLimitSettings.isResponseLimitBreached( + (Metadata) null, + ResponseLimitSettings.LimitEntity.INDICES, + 0 + ); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForNullRoutingTableExpectNotBreached() { + final Settings settings = Settings.builder().build(); + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + final boolean breached = ResponseLimitSettings.isResponseLimitBreached( + (RoutingTable) null, + ResponseLimitSettings.LimitEntity.INDICES, + 0 + ); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForNullLimitEntityWithRoutingTableExpectException() { + final ClusterState clusterState = buildClusterState("test-index-1", "test-index-2", "test-index-3"); + expectThrows( + IllegalArgumentException.class, + () -> ResponseLimitSettings.isResponseLimitBreached(clusterState.getRoutingTable(), null, 4) + ); + } + + public void testIsResponseLimitBreachedForNullLimitEntityWithMetadataExpectException() { + final ClusterState clusterState = buildClusterState("test-index-1", "test-index-2", "test-index-3"); + expectThrows( + IllegalArgumentException.class, + () -> ResponseLimitSettings.isResponseLimitBreached(clusterState.getMetadata(), null, 4) + ); + } + + public void testIsResponseLimitBreachedForCatIndicesWithNoSettingPassedExpectNotBreached() { + // Don't set setting + final Settings settings = Settings.builder().build(); + final boolean breached = isBreachedForIndices(settings); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForCatIndicesWithSettingDisabledExpectNotBreached() { + // Don't enable limit + final Settings settings = Settings.builder().put(CAT_INDICES_RESPONSE_LIMIT_SETTING.getKey(), -1).build(); + final boolean breached = isBreachedForIndices(settings); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForCatIndicesWithSettingEnabledExpectBreached() { + // Set limit of 1 index + final Settings settings = Settings.builder().put(CAT_INDICES_RESPONSE_LIMIT_SETTING.getKey(), 1).build(); + final boolean breached = isBreachedForIndices(settings); + assertTrue(breached); + } + + public void testIsResponseLimitBreachedForCatIndicesWithSettingEnabledExpectNotBreached() { + // Set limit of 5 indices + final Settings settings = Settings.builder().put(CAT_INDICES_RESPONSE_LIMIT_SETTING.getKey(), 5).build(); + final boolean breached = isBreachedForIndices(settings); + assertFalse(breached); + } + + private static boolean isBreachedForIndices(final Settings settings) { + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + // Pass cluster state with 3 indices + final ClusterState clusterState = buildClusterState("test-index-1", "test-index-2", "test-index-3"); + return ResponseLimitSettings.isResponseLimitBreached( + clusterState.getMetadata(), + ResponseLimitSettings.LimitEntity.INDICES, + responseLimitSettings.getCatIndicesResponseLimit() + ); + } + + public void testIsResponseLimitBreachedForCatShardsWithSettingDisabledExpectNotBreached() { + // Don't enable limit + final Settings settings = Settings.builder().put(CAT_SHARDS_RESPONSE_LIMIT_SETTING.getKey(), -1).build(); + final boolean breached = isBreachedForShards(settings); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForCatShardsWithNoSettingPassedExpectNotBreached() { + // Don't set setting + final Settings settings = Settings.builder().build(); + final boolean breached = isBreachedForShards(settings); + assertFalse(breached); + } + + private static boolean isBreachedForShards(Settings settings) { + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + // Build cluster state with 3 shards + final ClusterState clusterState = buildClusterState("test-index-1", "test-index-2", "test-index-3"); + return ResponseLimitSettings.isResponseLimitBreached( + clusterState.getRoutingTable(), + ResponseLimitSettings.LimitEntity.SHARDS, + responseLimitSettings.getCatShardsResponseLimit() + ); + } + + public void testIsResponseLimitBreachedForCatShardsWithSettingEnabledExpectBreached() { + // Set limit of 2 shards + final Settings settings = Settings.builder().put(CAT_SHARDS_RESPONSE_LIMIT_SETTING.getKey(), 2).build(); + final boolean breached = isBreachedForShards(settings); + assertTrue(breached); + } + + public void testIsResponseLimitBreachedForCatShardsWithSettingEnabledExpectNotBreached() { + // Set limit of 9 shards + final Settings settings = Settings.builder().put(CAT_SHARDS_RESPONSE_LIMIT_SETTING.getKey(), 9).build(); + final boolean breached = isBreachedForShards(settings); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForCatSegmentsWithSettingDisabledExpectNotBreached() { + // Don't enable limit + final Settings settings = Settings.builder().put(CAT_SEGMENTS_RESPONSE_LIMIT_SETTING.getKey(), -1).build(); + final boolean breached = isBreachedForSegments(settings); + assertFalse(breached); + } + + public void testIsResponseLimitBreachedForCatSegmentsWithNoSettingPassedExpectNotBreached() { + // Don't set setting + final Settings settings = Settings.builder().build(); + final boolean breached = isBreachedForSegments(settings); + assertFalse(breached); + } + + private static boolean isBreachedForSegments(Settings settings) { + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + // Build cluster state with 3 indices + final ClusterState clusterState = buildClusterState("test-index-1", "test-index-2", "test-index-3"); + final boolean breached = ResponseLimitSettings.isResponseLimitBreached( + clusterState.getRoutingTable(), + ResponseLimitSettings.LimitEntity.INDICES, + responseLimitSettings.getCatSegmentsResponseLimit() + ); + return breached; + } + + public void testIsResponseLimitBreachedForCatSegmentsWithSettingEnabledExpectBreached() { + // Set limit of 1 index + final Settings settings = Settings.builder().put(CAT_SEGMENTS_RESPONSE_LIMIT_SETTING.getKey(), 1).build(); + final boolean breached = isBreachedForSegments(settings); + assertTrue(breached); + } + + public void testIsResponseLimitBreachedForCatSegmentsWithSettingEnabledExpectNotBreached() { + // Set limit of 5 indices + final Settings settings = Settings.builder().put(CAT_SEGMENTS_RESPONSE_LIMIT_SETTING.getKey(), 5).build(); + final boolean breached = isBreachedForSegments(settings); + assertFalse(breached); + } + + private static ClusterState buildClusterState(String... indices) { + final Metadata.Builder metadata = Metadata.builder(); + for (String index : indices) { + metadata.put(IndexMetadata.builder(index).settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(0)); + } + final Map indexRoutingTableMap = new HashMap<>(); + for (String s : indices) { + final Index index = new Index(s, "uuid"); + final ShardId primaryShardId = new ShardId(index, 0); + final ShardRouting primaryShardRouting = createShardRouting(primaryShardId, true); + final ShardId replicaShardId = new ShardId(index, 1); + final ShardRouting replicaShardRouting = createShardRouting(replicaShardId, false); + final IndexShardRoutingTable.Builder indexShardRoutingTableBuilder = new IndexShardRoutingTable.Builder(primaryShardId); + indexShardRoutingTableBuilder.addShard(primaryShardRouting); + indexShardRoutingTableBuilder.addShard(replicaShardRouting); + final IndexRoutingTable.Builder indexRoutingTable = IndexRoutingTable.builder(index) + .addShard(primaryShardRouting) + .addShard(replicaShardRouting) + .addIndexShard(indexShardRoutingTableBuilder.build()); + indexRoutingTableMap.put(index.getName(), indexRoutingTable.build()); + } + final RoutingTable routingTable = new RoutingTable(1, indexRoutingTableMap); + return ClusterState.builder(org.opensearch.cluster.ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)) + .metadata(metadata) + .routingTable(routingTable) + .build(); + } + + private static ShardRouting createShardRouting(ShardId shardId, boolean isPrimary) { + return TestShardRouting.newShardRouting(shardId, randomAlphaOfLength(4), isPrimary, ShardRoutingState.STARTED); + } +} diff --git a/server/src/test/java/org/opensearch/rest/action/cat/RestIndicesActionTests.java b/server/src/test/java/org/opensearch/rest/action/cat/RestIndicesActionTests.java index 1d1b509ae94e5..2debb00a0d14f 100644 --- a/server/src/test/java/org/opensearch/rest/action/cat/RestIndicesActionTests.java +++ b/server/src/test/java/org/opensearch/rest/action/cat/RestIndicesActionTests.java @@ -43,6 +43,8 @@ import org.opensearch.cluster.routing.TestShardRouting; import org.opensearch.common.Table; import org.opensearch.common.UUIDs; +import org.opensearch.common.breaker.ResponseLimitSettings; +import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; @@ -144,7 +146,10 @@ public void setup() { } public void testBuildTable() { - final RestIndicesAction action = new RestIndicesAction(); + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final Settings settings = Settings.builder().build(); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + final RestIndicesAction action = new RestIndicesAction(responseLimitSettings); final Table table = action.buildTable( new FakeRestRequest(), indicesSettings, @@ -165,8 +170,11 @@ public void testBuildTable() { } public void testBuildPaginatedTable() { - final RestIndicesAction action = new RestIndicesAction(); - final RestIndicesListAction indicesListAction = new RestIndicesListAction(); + final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + final Settings settings = Settings.builder().build(); + final ResponseLimitSettings responseLimitSettings = new ResponseLimitSettings(clusterSettings, settings); + final RestIndicesAction action = new RestIndicesAction(responseLimitSettings); + final RestIndicesListAction indicesListAction = new RestIndicesListAction(responseLimitSettings); List indicesList = new ArrayList<>(indicesMetadatas.keySet()); // Using half of the indices from metadata list for a page String[] indicesToBeQueried = indicesList.subList(0, indicesMetadatas.size() / 2).toArray(new String[0]);