[Paginating ClusterManager Read APIs] Blocking non-paginated calls

[gargharsh3134]

Is your feature request related to a problem? Please describe

To prevent unfiltered or non-paginated Admin API queries from stressing out large clusters (clusters having large number of indices or shards), a mechanism to invalidate/fail those requests is required. Each API can define it's own dynamic cluster setting with a value denoting the limit it can support, say _cat/indices can define a setting named cat.indices.limit = 1000 denoting it can only serve requests which query less than 1k indices at a time.

This would help to prevent performance degradations in the cluster by the read (_cat) APIs which are merely used for monitoring the cluster and also encourage users to move to paginated counterparts of such APIs (#15014).

Describe the solution you'd like

The approach should be able to incorporate all the _cat APIs and provide abstraction such that, the concrete RestActions can simply define the actual validations specific to them and offload rest of the work.
On an initial thought, all (most) of the _cat APIs would require hold of clusterState to validate and resolve the RestRequest, to identify the number of shards, indices etc. that are being queried. A pre-validation can thus make use of local cluster state to validate the request.

A new RestCatLimitSettings class can be introduced, that will can be used by each of the API to define it's own cluster setting at a common place.

public class RestCatLimitSettings {

    public static final Setting<Integer> CAT_INDICES_LIMIT_SETTING = Setting.intSetting(
        "cat.indices.limit",
        -1,
        Setting.Property.NodeScope,
        Setting.Property.Dynamic
    );

    public static final Setting<Integer> CAT_SHARDS_LIMIT_SETTING = Setting.intSetting(
        "cat.shards.limit",
        -1,
        Setting.Property.NodeScope,
        Setting.Property.Dynamic
    );

    public static final Setting<Integer> CAT_SEGMENTS_LIMIT_SETTING = Setting.intSetting(
        "cat.segments.limit",
        -1,
        Setting.Property.NodeScope,
        Setting.Property.Dynamic
    );

    private static final List<Setting<Integer>> restCatLimitSettingsList = unmodifiableList(
        Arrays.asList(CAT_INDICES_LIMIT_SETTING, CAT_SHARDS_LIMIT_SETTING, CAT_SEGMENTS_LIMIT_SETTING));

    private Map<String, Integer> restCatLimitsMap;

    public RestCatLimitSettings(ClusterSettings clusterSettings, Settings settings) {
        restCatLimitsMap = new HashMap<>();
        restCatLimitSettingsList.forEach(setting -> restCatLimitsMap.put(setting.getKey(), setting.get(settings)));
        
        // define consumer to update the restCatLimitsMap if any of the settings change
    }
    
    public Map<String, Integer> getRestCatLimitsMap() {
        return restCatLimitsMap;
    }
}

Introducing requiresPreValidation and PreValidationPredicate in AbstractCatAction which the onboarding APIs can override and provide their own implementations.

public abstract class AbstractCatAction extends BaseRestHandler {

   // To check if the RestAction is onboarded to fail fast mechanism
    protected Boolean requiresPreValidation(RestRequest request) {
        return false;
    }

   // To be overriden by the onbaording actions if they require specific validations
   // else, by default every call would be blocked
    protected Predicate<ClusterStateResponse> preValidateRestRequest(RestRequest request) {
        return ClusterStateResponse -> false;
    }

    @Override
    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
        
        if (request.getLargeClusterValidationRequired()) {
            if (requiresPreValidation()) {
                final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
                clusterStateRequest.local(true);
                return restChannel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(restChannel) {
                    @Override
                    public void processResponse(final ClusterStateResponse clusterStateResponse) throws Exception {
                        if (preValidateRestRequest(request).test(clusterStateResponse)) {
                            throw new Exception();
                        }
                        RestChannelConsumer action = doCatRequest(request, client);
                        action.accept(restChannel);
                    }
                });
            }
        }
        return doCatRequest(request, client);
        
    }

Related component

Cluster Manager

Describe alternatives you've considered

No response

Additional context

No response

[sumitasr]

Raised PR to resolve this issue. @rajiv-kv can you assign this to me.

[sumitasr]

Follow up task - Evaluate if we need to configure a lower and upper bound value for response limit settings.

[sumitasr]

Follow up task - Evaluate if we can introduce stable integ tests for #15986