[Failure store - selector syntax] Replace failureOptions with selector options internally. (#114812) (#114882)

**Introduction**

> In order to make adoption of failure stores simpler for all users, we
are introducing a new syntactical feature to index expression
resolution: The selector. > > Selectors, denoted with a :: followed by a
recognized suffix will allow users to specify which component of an
index abstraction they would like to operate on within an API call. In
this case, an index abstraction is a concrete index, data stream, or
alias; Any abstraction that can be resolved to a set of indices/shards.
We define a component of an index abstraction to be some searchable unit
of the index abstraction. > > To start, we will support two components:
data and failures. Concrete indices are their own data components, while
the data component for index aliases are all of the indices contained
therein. For data streams, the data component corresponds to their
backing indices. Data stream aliases mirror this, treating all backing
indices of the data streams they correspond to as their data component.
>  > The failure component is only supported by data streams and data
stream aliases. The failure component of these abstractions refer to the
data streams' failure stores. Indices and index aliases do not have a
failure component.

For more details and examples see
https://github.com/elastic/elasticsearch/pull/113144. All this work has
been cherry picked from there.

**Purpose of this PR**

This PR is replacing the `FailureStoreOptions` with the
`SelectorOptions`, there shouldn't be any perceivable change to the user
since we kept the query parameter "failure_store" for now. It will be
removed in the next PR which will introduce the parsing of the
expressions. 

_The current PR is just a refactoring and does not and should not change
any existing behaviour._

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamsSnapshotsIT.java

@@ -132,9 +132,7 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         // Initialize the failure store.
         RolloverRequest rolloverRequest = new RolloverRequest("with-fs", null);
         rolloverRequest.setIndicesOptions(
-            IndicesOptions.builder(rolloverRequest.indicesOptions())
-                .failureStoreOptions(b -> b.includeRegularIndices(false).includeFailureIndices(true))
-                .build()
+            IndicesOptions.builder(rolloverRequest.indicesOptions()).selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES).build()
         );
         response = client.execute(RolloverAction.INSTANCE, rolloverRequest).get();
         assertTrue(response.isAcknowledged());

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/IngestFailureStoreMetricsIT.java

@@ -198,9 +198,7 @@ public class IngestFailureStoreMetricsIT extends ESIntegTestCase {
         // Initialize failure store.
         var rolloverRequest = new RolloverRequest(dataStream, null);
         rolloverRequest.setIndicesOptions(
-            IndicesOptions.builder(rolloverRequest.indicesOptions())
-                .failureStoreOptions(opts -> opts.includeFailureIndices(true).includeRegularIndices(false))
-                .build()
+            IndicesOptions.builder(rolloverRequest.indicesOptions()).selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES).build()
         );
         var rolloverResponse = client().execute(RolloverAction.INSTANCE, rolloverRequest).actionGet();
         var failureStoreIndex = rolloverResponse.getNewIndex();

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java

@@ -946,7 +946,7 @@ public class DataStreamLifecycleService implements ClusterStateListener, Closeab
                 UpdateSettingsRequest updateMergePolicySettingsRequest = new UpdateSettingsRequest();
                 updateMergePolicySettingsRequest.indicesOptions(
                     IndicesOptions.builder(updateMergePolicySettingsRequest.indicesOptions())
-                        .failureStoreOptions(new IndicesOptions.FailureStoreOptions(true, true))
+                        .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                         .build()
                 );
                 updateMergePolicySettingsRequest.indices(indexName);
@@ -1409,7 +1409,7 @@ public class DataStreamLifecycleService implements ClusterStateListener, Closeab
         if (rolloverFailureStore) {
             rolloverRequest.setIndicesOptions(
                 IndicesOptions.builder(rolloverRequest.indicesOptions())
-                    .failureStoreOptions(opts -> opts.includeFailureIndices(true).includeRegularIndices(false))
+                    .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                     .build()
             );
         }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/rest/RestGetDataStreamsAction.java

@@ -43,7 +43,7 @@ public class RestGetDataStreamsAction extends BaseRestHandler {
                 IndicesOptions.GatekeeperOptions.IGNORE_THROTTLED,
                 "verbose"
             ),
-            DataStream.isFailureStoreFeatureFlagEnabled() ? Set.of(IndicesOptions.FailureStoreOptions.FAILURE_STORE) : Set.of()
+            DataStream.isFailureStoreFeatureFlagEnabled() ? Set.of(IndicesOptions.FAILURE_STORE_QUERY_PARAM) : Set.of()
         )
     );
 

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java

@@ -225,17 +225,11 @@ public class DataStreamLifecycleServiceTests extends ESTestCase {
         assertThat(clientSeenRequests.get(0), instanceOf(RolloverRequest.class));
         RolloverRequest rolloverBackingIndexRequest = (RolloverRequest) clientSeenRequests.get(0);
         assertThat(rolloverBackingIndexRequest.getRolloverTarget(), is(dataStreamName));
-        assertThat(
-            rolloverBackingIndexRequest.indicesOptions().failureStoreOptions(),
-            equalTo(new IndicesOptions.FailureStoreOptions(true, false))
-        );
+        assertThat(rolloverBackingIndexRequest.indicesOptions().selectorOptions(), equalTo(IndicesOptions.SelectorOptions.ONLY_DATA));
         assertThat(clientSeenRequests.get(1), instanceOf(RolloverRequest.class));
         RolloverRequest rolloverFailureIndexRequest = (RolloverRequest) clientSeenRequests.get(1);
         assertThat(rolloverFailureIndexRequest.getRolloverTarget(), is(dataStreamName));
-        assertThat(
-            rolloverFailureIndexRequest.indicesOptions().failureStoreOptions(),
-            equalTo(new IndicesOptions.FailureStoreOptions(false, true))
-        );
+        assertThat(rolloverFailureIndexRequest.indicesOptions().selectorOptions(), equalTo(IndicesOptions.SelectorOptions.ONLY_FAILURES));
         List<DeleteIndexRequest> deleteRequests = clientSeenRequests.subList(2, 5)
             .stream()
             .map(transportRequest -> (DeleteIndexRequest) transportRequest)
@@ -1549,17 +1543,11 @@ public class DataStreamLifecycleServiceTests extends ESTestCase {
         assertThat(clientSeenRequests.get(0), instanceOf(RolloverRequest.class));
         RolloverRequest rolloverBackingIndexRequest = (RolloverRequest) clientSeenRequests.get(0);
         assertThat(rolloverBackingIndexRequest.getRolloverTarget(), is(dataStreamName));
-        assertThat(
-            rolloverBackingIndexRequest.indicesOptions().failureStoreOptions(),
-            equalTo(new IndicesOptions.FailureStoreOptions(true, false))
-        );
+        assertThat(rolloverBackingIndexRequest.indicesOptions().selectorOptions(), equalTo(IndicesOptions.SelectorOptions.ONLY_DATA));
         assertThat(clientSeenRequests.get(1), instanceOf(RolloverRequest.class));
         RolloverRequest rolloverFailureIndexRequest = (RolloverRequest) clientSeenRequests.get(1);
         assertThat(rolloverFailureIndexRequest.getRolloverTarget(), is(dataStreamName));
-        assertThat(
-            rolloverFailureIndexRequest.indicesOptions().failureStoreOptions(),
-            equalTo(new IndicesOptions.FailureStoreOptions(false, true))
-        );
+        assertThat(rolloverFailureIndexRequest.indicesOptions().selectorOptions(), equalTo(IndicesOptions.SelectorOptions.ONLY_FAILURES));
         assertThat(
             ((DeleteIndexRequest) clientSeenRequests.get(2)).indices()[0],
             is(dataStream.getFailureIndices().getIndices().get(0).getName())

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -245,6 +245,7 @@ public class TransportVersions {
     public static final TransportVersion QUERY_RULE_TEST_API = def(8_769_00_0);
     public static final TransportVersion ESQL_PER_AGGREGATE_FILTER = def(8_770_00_0);
     public static final TransportVersion ML_INFERENCE_ATTACH_TO_EXISTSING_DEPLOYMENT = def(8_771_00_0);
+    public static final TransportVersion CONVERT_FAILURE_STORE_OPTIONS_TO_SELECTOR_OPTIONS_INTERNALLY = def(8_772_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/indices/get/GetIndexRequest.java

@@ -98,9 +98,7 @@ public class GetIndexRequest extends ClusterInfoRequest<GetIndexRequest> {
         super(
             DataStream.isFailureStoreFeatureFlagEnabled()
                 ? IndicesOptions.builder(IndicesOptions.strictExpandOpen())
-                    .failureStoreOptions(
-                        IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true)
-                    )
+                    .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                     .build()
                 : IndicesOptions.strictExpandOpen()
         );

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverRequest.java

@@ -147,8 +147,8 @@ public class RolloverRequest extends AcknowledgedRequest<RolloverRequest> implem
             );
         }
 
-        var failureStoreOptions = indicesOptions.failureStoreOptions();
-        if (failureStoreOptions.includeRegularIndices() && failureStoreOptions.includeFailureIndices()) {
+        var selectors = indicesOptions.selectorOptions().defaultSelectors();
+        if (selectors.size() > 1) {
             validationException = addValidationError(
                 "rollover cannot be applied to both regular and failure indices at the same time",
                 validationException
@@ -188,7 +188,7 @@ public class RolloverRequest extends AcknowledgedRequest<RolloverRequest> implem
      * @return true of the rollover request targets the failure store, false otherwise.
      */
     public boolean targetsFailureStore() {
-        return DataStream.isFailureStoreFeatureFlagEnabled() && indicesOptions.failureStoreOptions().includeFailureIndices();
+        return DataStream.isFailureStoreFeatureFlagEnabled() && indicesOptions.includeFailureIndices();
     }
 
     public void setIndicesOptions(IndicesOptions indicesOptions) {

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -150,7 +150,7 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
                 .matchClosed(request.indicesOptions().expandWildcardsClosed())
                 .build(),
             IndicesOptions.GatekeeperOptions.DEFAULT,
-            request.indicesOptions().failureStoreOptions()
+            request.indicesOptions().selectorOptions()
         );
 
         return state.blocks()
@@ -247,7 +247,7 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
             IndicesOptions.ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS,
             IndicesOptions.WildcardOptions.builder().matchClosed(true).allowEmptyExpressions(false).build(),
             IndicesOptions.GatekeeperOptions.DEFAULT,
-            rolloverRequest.indicesOptions().failureStoreOptions()
+            rolloverRequest.indicesOptions().selectorOptions()
         );
         IndicesStatsRequest statsRequest = new IndicesStatsRequest().indices(rolloverRequest.getRolloverTarget())
             .clear()

server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java

@@ -212,7 +212,7 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
                 RolloverRequest rolloverRequest = new RolloverRequest(dataStream, null);
                 rolloverRequest.setIndicesOptions(
                     IndicesOptions.builder(rolloverRequest.indicesOptions())
-                        .failureStoreOptions(new IndicesOptions.FailureStoreOptions(false, true))
+                        .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                         .build()
                 );
                 // We are executing a lazy rollover because it is an action specialised for this situation, when we want an

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -418,7 +418,7 @@ public class TransportBulkAction extends TransportAbstractBulkAction {
             if (targetFailureStore) {
                 rolloverRequest.setIndicesOptions(
                     IndicesOptions.builder(rolloverRequest.indicesOptions())
-                        .failureStoreOptions(new IndicesOptions.FailureStoreOptions(false, true))
+                        .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                         .build()
                 );
             }

server/src/main/java/org/elasticsearch/action/datastreams/DataStreamsStatsAction.java

@@ -61,9 +61,7 @@ public class DataStreamsStatsAction extends ActionType<DataStreamsStatsAction.Re
                             .allowFailureIndices(true)
                             .build()
                     )
-                    .failureStoreOptions(
-                        IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true).build()
-                    )
+                    .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                     .build()
             );
         }

server/src/main/java/org/elasticsearch/action/support/IndexComponentSelector.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.support;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * We define as index components the two different sets of indices a data stream could consist of:
+ * - DATA: represents the backing indices
+ * - FAILURES: represent the failing indices
+ * Note: An index is its own DATA component, but it cannot have a FAILURE component.
+ */
+public enum IndexComponentSelector {
+    DATA("data"),
+    FAILURES("failures");
+
+    private final String key;
+
+    IndexComponentSelector(String key) {
+        this.key = key;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    private static final Map<String, IndexComponentSelector> REGISTRY;
+
+    static {
+        Map<String, IndexComponentSelector> registry = new HashMap<>(IndexComponentSelector.values().length);
+        for (IndexComponentSelector value : IndexComponentSelector.values()) {
+            registry.put(value.getKey(), value);
+        }
+        REGISTRY = Collections.unmodifiableMap(registry);
+    }
+
+    public static IndexComponentSelector getByKey(String key) {
+        return REGISTRY.get(key);
+    }
+}

server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java

@@ -31,7 +31,6 @@ import java.util.EnumSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.xcontent.support.XContentMapValues.nodeBooleanValue;
@@ -48,24 +47,36 @@ import static org.elasticsearch.common.xcontent.support.XContentMapValues.nodeSt
  * @param gatekeeperOptions, applies to all the resolved indices and defines if throttled will be included and if certain type of
  *                        aliases or indices are allowed, or they will throw an error. It acts as a gatekeeper when an action
  *                        does not support certain options.
- * @param failureStoreOptions, applies to all indices already matched and controls the type of indices that will be returned. Currently,
- *                             there are two types, data stream failure indices (only certain data streams have them) and data stream
- *                             backing indices or stand-alone indices.
+ * @param selectorOptions, applies to all resolved expressions, and it specifies the index component that should be included, if there
+ *                         is no index component defined on the expression level.
  */
 public record IndicesOptions(
     ConcreteTargetOptions concreteTargetOptions,
     WildcardOptions wildcardOptions,
     GatekeeperOptions gatekeeperOptions,
-    FailureStoreOptions failureStoreOptions
+    SelectorOptions selectorOptions
 ) implements ToXContentFragment {
 
-    public IndicesOptions(
-        ConcreteTargetOptions concreteTargetOptions,
-        WildcardOptions wildcardOptions,
-        GatekeeperOptions gatekeeperOptions
-    ) {
-        this(concreteTargetOptions, wildcardOptions, gatekeeperOptions, FailureStoreOptions.DEFAULT);
-    }
+    /**
+     * @deprecated this query param will be replaced by the selector `::` on the expression level
+     */
+    @Deprecated
+    public static final String FAILURE_STORE_QUERY_PARAM = "failure_store";
+    /**
+     * @deprecated this value will be replaced by the selector `::*` on the expression level
+     */
+    @Deprecated
+    public static final String INCLUDE_ALL = "include";
+    /**
+     * @deprecated this value will be replaced by the selector `::data` on the expression level
+     */
+    @Deprecated
+    public static final String INCLUDE_ONLY_REGULAR_INDICES = "exclude";
+    /**
+     * @deprecated this value will be replaced by the selector `::failures` on the expression level
+     */
+    @Deprecated
+    public static final String INCLUDE_ONLY_FAILURE_INDICES = "only";
 
     public static IndicesOptions.Builder builder() {
         return new Builder();
@@ -310,7 +321,7 @@ public record IndicesOptions(
      * - The "allow*" flags, which purpose is to enable actions to define certain conditions that need to apply on the concrete indices
      * they accept. For example, single-index actions will set allowAliasToMultipleIndices to false, while search will not accept a
      * closed index etc. These options are not configurable by the end-user.
-     * - The ignoreThrottled flag, which is a depricared flag that will filter out frozen indices.
+     * - The ignoreThrottled flag, which is a deprecated flag that will filter out frozen indices.
      * @param allowAliasToMultipleIndices, allow aliases to multiple indices, true by default.
      * @param allowClosedIndices, allow closed indices, true by default.
      * @param allowFailureIndices, allow failure indices in the response, true by default
@@ -408,87 +419,53 @@ public record IndicesOptions(
     }
 
     /**
-     * Applies to all indices already matched and controls the type of indices that will be returned. There are two types, data stream
-     * failure indices (only certain data streams have them) and data stream backing indices or stand-alone indices.
-     * @param includeRegularIndices, when true regular or data stream backing indices will be retrieved.
-     * @param includeFailureIndices, when true data stream failure indices will be included.
+     * Defines which selectors should be used by default for an index operation in the event that no selectors are provided.
      */
-    public record FailureStoreOptions(boolean includeRegularIndices, boolean includeFailureIndices)
-        implements
-            Writeable,
-            ToXContentFragment {
+    public record SelectorOptions(EnumSet<IndexComponentSelector> defaultSelectors) implements Writeable {
 
-        public static final String FAILURE_STORE = "failure_store";
-        public static final String INCLUDE_ALL = "include";
-        public static final String INCLUDE_ONLY_REGULAR_INDICES = "exclude";
-        public static final String INCLUDE_ONLY_FAILURE_INDICES = "only";
-
-        public static final FailureStoreOptions DEFAULT = new FailureStoreOptions(true, false);
-
-        public static FailureStoreOptions read(StreamInput in) throws IOException {
-            return new FailureStoreOptions(in.readBoolean(), in.readBoolean());
-        }
-
-        public static FailureStoreOptions parseParameters(Object failureStoreValue, FailureStoreOptions defaultOptions) {
-            if (failureStoreValue == null) {
-                return defaultOptions;
-            }
-            FailureStoreOptions.Builder builder = defaultOptions == null
-                ? new FailureStoreOptions.Builder()
-                : new FailureStoreOptions.Builder(defaultOptions);
-            return switch (failureStoreValue.toString()) {
-                case INCLUDE_ALL -> builder.includeRegularIndices(true).includeFailureIndices(true).build();
-                case INCLUDE_ONLY_REGULAR_INDICES -> builder.includeRegularIndices(true).includeFailureIndices(false).build();
-                case INCLUDE_ONLY_FAILURE_INDICES -> builder.includeRegularIndices(false).includeFailureIndices(true).build();
-                default -> throw new IllegalArgumentException("No valid " + FAILURE_STORE + " value [" + failureStoreValue + "]");
-            };
-        }
-
-        @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            return builder.field(FAILURE_STORE, displayValue());
-        }
+        public static final SelectorOptions DATA_AND_FAILURE = new SelectorOptions(
+            EnumSet.of(IndexComponentSelector.DATA, IndexComponentSelector.FAILURES)
+        );
+        public static final SelectorOptions ONLY_DATA = new SelectorOptions(EnumSet.of(IndexComponentSelector.DATA));
+        public static final SelectorOptions ONLY_FAILURES = new SelectorOptions(EnumSet.of(IndexComponentSelector.FAILURES));
+        /**
+         * Default instance. Uses <pre>::data</pre> as the default selector if none are present in an index expression.
+         */
+        public static final SelectorOptions DEFAULT = ONLY_DATA;
 
-        public String displayValue() {
-            if (includeRegularIndices && includeFailureIndices) {
-                return INCLUDE_ALL;
-            } else if (includeRegularIndices) {
-                return INCLUDE_ONLY_REGULAR_INDICES;
-            }
-            return INCLUDE_ONLY_FAILURE_INDICES;
+        public static SelectorOptions read(StreamInput in) throws IOException {
+            return new SelectorOptions(in.readEnumSet(IndexComponentSelector.class));
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeBoolean(includeRegularIndices);
-            out.writeBoolean(includeFailureIndices);
+            out.writeEnumSet(defaultSelectors);
         }
 
         public static class Builder {
-            private boolean includeRegularIndices;
-            private boolean includeFailureIndices;
+            private EnumSet<IndexComponentSelector> defaultSelectors;
 
             public Builder() {
                 this(DEFAULT);
             }
 
-            Builder(FailureStoreOptions options) {
-                includeRegularIndices = options.includeRegularIndices;
-                includeFailureIndices = options.includeFailureIndices;
+            Builder(SelectorOptions options) {
+                defaultSelectors = EnumSet.copyOf(options.defaultSelectors);
             }
 
-            public Builder includeRegularIndices(boolean includeRegularIndices) {
-                this.includeRegularIndices = includeRegularIndices;
+            public Builder setDefaultSelectors(IndexComponentSelector first, IndexComponentSelector... remaining) {
+                defaultSelectors = EnumSet.of(first, remaining);
                 return this;
             }
 
-            public Builder includeFailureIndices(boolean includeFailureIndices) {
-                this.includeFailureIndices = includeFailureIndices;
+            public Builder setDefaultSelectors(EnumSet<IndexComponentSelector> defaultSelectors) {
+                this.defaultSelectors = EnumSet.copyOf(defaultSelectors);
                 return this;
             }
 
-            public FailureStoreOptions build() {
-                return new FailureStoreOptions(includeRegularIndices, includeFailureIndices);
+            public SelectorOptions build() {
+                assert defaultSelectors.isEmpty() != true : "Default selectors cannot be an empty set";
+                return new SelectorOptions(EnumSet.copyOf(defaultSelectors));
             }
         }
 
@@ -496,8 +473,8 @@ public record IndicesOptions(
             return new Builder();
         }
 
-        public static Builder builder(FailureStoreOptions failureStoreOptions) {
-            return new Builder(failureStoreOptions);
+        public static Builder builder(SelectorOptions selectorOptions) {
+            return new Builder(selectorOptions);
         }
     }
 
@@ -550,7 +527,7 @@ public record IndicesOptions(
         ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
         WildcardOptions.DEFAULT,
         GatekeeperOptions.DEFAULT,
-        FailureStoreOptions.DEFAULT
+        SelectorOptions.DEFAULT
     );
 
     public static final IndicesOptions STRICT_EXPAND_OPEN = IndicesOptions.builder()
@@ -570,7 +547,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_FAILURE_STORE = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -589,7 +566,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+        .selectorOptions(SelectorOptions.DATA_AND_FAILURE)
         .build();
     public static final IndicesOptions LENIENT_EXPAND_OPEN = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
@@ -608,7 +585,25 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
+        .build();
+    public static final IndicesOptions LENIENT_EXPAND_OPEN_NO_SELECTORS = IndicesOptions.builder()
+        .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
+        .wildcardOptions(
+            WildcardOptions.builder()
+                .matchOpen(true)
+                .matchClosed(false)
+                .includeHidden(false)
+                .allowEmptyExpressions(true)
+                .resolveAliases(true)
+        )
+        .gatekeeperOptions(
+            GatekeeperOptions.builder()
+                .allowAliasToMultipleIndices(true)
+                .allowClosedIndices(true)
+                .allowFailureIndices(false)
+                .ignoreThrottled(false)
+        )
         .build();
     public static final IndicesOptions LENIENT_EXPAND_OPEN_HIDDEN = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
@@ -627,7 +622,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions LENIENT_EXPAND_OPEN_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
@@ -646,7 +641,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions LENIENT_EXPAND_OPEN_CLOSED_HIDDEN = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
@@ -660,7 +655,20 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
+        .build();
+    public static final IndicesOptions LENIENT_EXPAND_OPEN_CLOSED_HIDDEN_NO_SELECTOR = IndicesOptions.builder()
+        .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
+        .wildcardOptions(
+            WildcardOptions.builder().matchOpen(true).matchClosed(true).includeHidden(true).allowEmptyExpressions(true).resolveAliases(true)
+        )
+        .gatekeeperOptions(
+            GatekeeperOptions.builder()
+                .allowAliasToMultipleIndices(true)
+                .allowClosedIndices(true)
+                .allowFailureIndices(false)
+                .ignoreThrottled(false)
+        )
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -679,7 +687,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_CLOSED_HIDDEN = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -693,7 +701,20 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
+        .build();
+    public static final IndicesOptions STRICT_EXPAND_OPEN_CLOSED_HIDDEN_NO_SELECTORS = IndicesOptions.builder()
+        .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
+        .wildcardOptions(
+            WildcardOptions.builder().matchOpen(true).matchClosed(true).includeHidden(true).allowEmptyExpressions(true).resolveAliases(true)
+        )
+        .gatekeeperOptions(
+            GatekeeperOptions.builder()
+                .allowAliasToMultipleIndices(true)
+                .allowClosedIndices(true)
+                .allowFailureIndices(false)
+                .ignoreThrottled(false)
+        )
         .build();
     public static final IndicesOptions LENIENT_EXPAND_OPEN_CLOSED_FAILURE_STORE = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
@@ -712,7 +733,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+        .selectorOptions(SelectorOptions.DATA_AND_FAILURE)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_CLOSED_HIDDEN_FAILURE_STORE = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -726,7 +747,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+        .selectorOptions(SelectorOptions.DATA_AND_FAILURE)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_CLOSED_FAILURE_STORE = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -745,7 +766,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+        .selectorOptions(SelectorOptions.DATA_AND_FAILURE)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_FORBID_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -764,7 +785,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_HIDDEN_FORBID_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -783,7 +804,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_EXPAND_OPEN_FORBID_CLOSED_IGNORE_THROTTLED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -802,7 +823,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .allowAliasToMultipleIndices(true)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_SINGLE_INDEX_NO_EXPAND_FORBID_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -821,7 +842,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
     public static final IndicesOptions STRICT_NO_EXPAND_FORBID_CLOSED = IndicesOptions.builder()
         .concreteTargetOptions(ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS)
@@ -840,7 +861,7 @@ public record IndicesOptions(
                 .allowFailureIndices(true)
                 .ignoreThrottled(false)
         )
-        .failureStoreOptions(FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(false))
+        .selectorOptions(SelectorOptions.ONLY_DATA)
         .build();
 
     /**
@@ -929,14 +950,14 @@ public record IndicesOptions(
      * @return whether regular indices (stand-alone or backing indices) will be included in the response
      */
     public boolean includeRegularIndices() {
-        return failureStoreOptions().includeRegularIndices();
+        return selectorOptions().defaultSelectors().contains(IndexComponentSelector.DATA);
     }
 
     /**
      * @return whether failure indices (only supported by certain data streams) will be included in the response
      */
     public boolean includeFailureIndices() {
-        return failureStoreOptions().includeFailureIndices();
+        return selectorOptions().defaultSelectors().contains(IndexComponentSelector.FAILURES);
     }
 
     public void writeIndicesOptions(StreamOutput out) throws IOException {
@@ -977,8 +998,13 @@ public record IndicesOptions(
             states.add(WildcardStates.HIDDEN);
         }
         out.writeEnumSet(states);
-        if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)) {
-            failureStoreOptions.writeTo(out);
+        if (out.getTransportVersion()
+            .between(TransportVersions.V_8_14_0, TransportVersions.CONVERT_FAILURE_STORE_OPTIONS_TO_SELECTOR_OPTIONS_INTERNALLY)) {
+            out.writeBoolean(includeRegularIndices());
+            out.writeBoolean(includeFailureIndices());
+        }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.CONVERT_FAILURE_STORE_OPTIONS_TO_SELECTOR_OPTIONS_INTERNALLY)) {
+            out.writeEnumSet(selectorOptions.defaultSelectors);
         }
     }
 
@@ -999,16 +1025,30 @@ public record IndicesOptions(
             .allowFailureIndices(allowFailureIndices)
             .ignoreThrottled(options.contains(Option.IGNORE_THROTTLED))
             .build();
-        FailureStoreOptions failureStoreOptions = in.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)
-            ? FailureStoreOptions.read(in)
-            : FailureStoreOptions.DEFAULT;
+        SelectorOptions selectorOptions = SelectorOptions.DEFAULT;
+        if (in.getTransportVersion()
+            .between(TransportVersions.V_8_14_0, TransportVersions.CONVERT_FAILURE_STORE_OPTIONS_TO_SELECTOR_OPTIONS_INTERNALLY)) {
+            // Reading from an older node, which will be sending two booleans that we must read out and ignore.
+            var includeData = in.readBoolean();
+            var includeFailures = in.readBoolean();
+            if (includeData && includeFailures) {
+                selectorOptions = SelectorOptions.DATA_AND_FAILURE;
+            } else if (includeData) {
+                selectorOptions = SelectorOptions.ONLY_DATA;
+            } else {
+                selectorOptions = SelectorOptions.ONLY_FAILURES;
+            }
+        }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.CONVERT_FAILURE_STORE_OPTIONS_TO_SELECTOR_OPTIONS_INTERNALLY)) {
+            selectorOptions = new SelectorOptions(in.readEnumSet(IndexComponentSelector.class));
+        }
         return new IndicesOptions(
             options.contains(Option.ALLOW_UNAVAILABLE_CONCRETE_TARGETS)
                 ? ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS
                 : ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
             wildcardOptions,
             gatekeeperOptions,
-            failureStoreOptions
+            selectorOptions
         );
     }
 
@@ -1016,7 +1056,7 @@ public record IndicesOptions(
         private ConcreteTargetOptions concreteTargetOptions;
         private WildcardOptions wildcardOptions;
         private GatekeeperOptions gatekeeperOptions;
-        private FailureStoreOptions failureStoreOptions;
+        private SelectorOptions selectorOptions;
 
         Builder() {
             this(DEFAULT);
@@ -1026,7 +1066,7 @@ public record IndicesOptions(
             concreteTargetOptions = indicesOptions.concreteTargetOptions;
             wildcardOptions = indicesOptions.wildcardOptions;
             gatekeeperOptions = indicesOptions.gatekeeperOptions;
-            failureStoreOptions = indicesOptions.failureStoreOptions;
+            selectorOptions = indicesOptions.selectorOptions;
         }
 
         public Builder concreteTargetOptions(ConcreteTargetOptions concreteTargetOptions) {
@@ -1054,25 +1094,18 @@ public record IndicesOptions(
             return this;
         }
 
-        public Builder failureStoreOptions(FailureStoreOptions failureStoreOptions) {
-            this.failureStoreOptions = failureStoreOptions;
+        public Builder selectorOptions(SelectorOptions selectorOptions) {
+            this.selectorOptions = selectorOptions;
             return this;
         }
 
-        public Builder failureStoreOptions(FailureStoreOptions.Builder failureStoreOptions) {
-            this.failureStoreOptions = failureStoreOptions.build();
-            return this;
-        }
-
-        public Builder failureStoreOptions(Consumer<FailureStoreOptions.Builder> failureStoreOptionsConfig) {
-            FailureStoreOptions.Builder failureStoreOptionsBuilder = FailureStoreOptions.builder(failureStoreOptions);
-            failureStoreOptionsConfig.accept(failureStoreOptionsBuilder);
-            this.failureStoreOptions = failureStoreOptionsBuilder.build();
+        public Builder selectorOptions(SelectorOptions.Builder selectorOptions) {
+            this.selectorOptions = selectorOptions.build();
             return this;
         }
 
         public IndicesOptions build() {
-            return new IndicesOptions(concreteTargetOptions, wildcardOptions, gatekeeperOptions, failureStoreOptions);
+            return new IndicesOptions(concreteTargetOptions, wildcardOptions, gatekeeperOptions, selectorOptions);
         }
     }
 
@@ -1171,11 +1204,12 @@ public record IndicesOptions(
             .allowClosedIndices(forbidClosedIndices == false)
             .ignoreThrottled(ignoreThrottled)
             .build();
+        final SelectorOptions selectorOptions = SelectorOptions.DEFAULT;
         return new IndicesOptions(
             ignoreUnavailable ? ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS : ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
             wildcards,
             gatekeeperOptions,
-            FailureStoreOptions.DEFAULT
+            selectorOptions
         );
     }
 
@@ -1189,9 +1223,7 @@ public record IndicesOptions(
             request.param(ConcreteTargetOptions.IGNORE_UNAVAILABLE),
             request.param(WildcardOptions.ALLOW_NO_INDICES),
             request.param(GatekeeperOptions.IGNORE_THROTTLED),
-            DataStream.isFailureStoreFeatureFlagEnabled()
-                ? request.param(FailureStoreOptions.FAILURE_STORE)
-                : FailureStoreOptions.INCLUDE_ONLY_REGULAR_INDICES,
+            DataStream.isFailureStoreFeatureFlagEnabled() ? request.param(FAILURE_STORE_QUERY_PARAM) : INCLUDE_ONLY_REGULAR_INDICES,
             defaultSettings
         );
     }
@@ -1207,7 +1239,7 @@ public record IndicesOptions(
                 map.containsKey(GatekeeperOptions.IGNORE_THROTTLED)
                     ? map.get(GatekeeperOptions.IGNORE_THROTTLED)
                     : map.get("ignoreThrottled"),
-                map.containsKey(FailureStoreOptions.FAILURE_STORE) ? map.get(FailureStoreOptions.FAILURE_STORE) : map.get("failureStore"),
+                map.containsKey(FAILURE_STORE_QUERY_PARAM) ? map.get(FAILURE_STORE_QUERY_PARAM) : map.get("failureStore"),
                 defaultSettings
             );
         }
@@ -1235,7 +1267,7 @@ public record IndicesOptions(
             || "ignoreThrottled".equals(name)
             || WildcardOptions.ALLOW_NO_INDICES.equals(name)
             || "allowNoIndices".equals(name)
-            || (DataStream.isFailureStoreFeatureFlagEnabled() && FailureStoreOptions.FAILURE_STORE.equals(name))
+            || (DataStream.isFailureStoreFeatureFlagEnabled() && FAILURE_STORE_QUERY_PARAM.equals(name))
             || (DataStream.isFailureStoreFeatureFlagEnabled() && "failureStore".equals(name));
     }
 
@@ -1267,26 +1299,51 @@ public record IndicesOptions(
 
         WildcardOptions wildcards = WildcardOptions.parseParameters(wildcardsString, allowNoIndicesString, defaultSettings.wildcardOptions);
         GatekeeperOptions gatekeeperOptions = GatekeeperOptions.parseParameter(ignoreThrottled, defaultSettings.gatekeeperOptions);
-        FailureStoreOptions failureStoreOptions = DataStream.isFailureStoreFeatureFlagEnabled()
-            ? FailureStoreOptions.parseParameters(failureStoreString, defaultSettings.failureStoreOptions)
-            : FailureStoreOptions.DEFAULT;
+        SelectorOptions selectorOptions = DataStream.isFailureStoreFeatureFlagEnabled()
+            ? parseFailureStoreParameters(failureStoreString, defaultSettings.selectorOptions)
+            : SelectorOptions.DEFAULT;
 
         // note that allowAliasesToMultipleIndices is not exposed, always true (only for internal use)
         return IndicesOptions.builder()
             .concreteTargetOptions(ConcreteTargetOptions.fromParameter(ignoreUnavailableString, defaultSettings.concreteTargetOptions))
             .wildcardOptions(wildcards)
             .gatekeeperOptions(gatekeeperOptions)
-            .failureStoreOptions(failureStoreOptions)
+            .selectorOptions(selectorOptions)
             .build();
     }
 
+    /**
+     * @deprecated This method parses the query parameter failure_store. This is a deprecated param, and it will be replaced
+     * the selector suffix, for example `my-data-stream::data` or `my-data-stream::failures`
+     */
+    @Deprecated
+    private static SelectorOptions parseFailureStoreParameters(Object failureStoreValue, SelectorOptions defaultOptions) {
+        if (failureStoreValue == null) {
+            return defaultOptions;
+        }
+        return switch (failureStoreValue.toString()) {
+            case INCLUDE_ALL -> SelectorOptions.DATA_AND_FAILURE;
+            case INCLUDE_ONLY_REGULAR_INDICES -> SelectorOptions.ONLY_DATA;
+            case INCLUDE_ONLY_FAILURE_INDICES -> SelectorOptions.ONLY_FAILURES;
+            default -> throw new IllegalArgumentException("No valid " + FAILURE_STORE_QUERY_PARAM + " value [" + failureStoreValue + "]");
+        };
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
         concreteTargetOptions.toXContent(builder, params);
         wildcardOptions.toXContent(builder, params);
         gatekeeperOptions.toXContent(builder, params);
         if (DataStream.isFailureStoreFeatureFlagEnabled()) {
-            failureStoreOptions.toXContent(builder, params);
+            String displayValue;
+            if (SelectorOptions.DATA_AND_FAILURE.equals(selectorOptions())) {
+                displayValue = INCLUDE_ALL;
+            } else if (SelectorOptions.ONLY_DATA.equals(selectorOptions())) {
+                displayValue = INCLUDE_ONLY_REGULAR_INDICES;
+            } else {
+                displayValue = INCLUDE_ONLY_FAILURE_INDICES;
+            }
+            builder.field(FAILURE_STORE_QUERY_PARAM, displayValue);
         }
         return builder;
     }
@@ -1295,7 +1352,7 @@ public record IndicesOptions(
     private static final ParseField IGNORE_UNAVAILABLE_FIELD = new ParseField(ConcreteTargetOptions.IGNORE_UNAVAILABLE);
     private static final ParseField IGNORE_THROTTLED_FIELD = new ParseField(GatekeeperOptions.IGNORE_THROTTLED).withAllDeprecated();
     private static final ParseField ALLOW_NO_INDICES_FIELD = new ParseField(WildcardOptions.ALLOW_NO_INDICES);
-    private static final ParseField FAILURE_STORE_FIELD = new ParseField(FailureStoreOptions.FAILURE_STORE);
+    private static final ParseField FAILURE_STORE_FIELD = new ParseField(FAILURE_STORE_QUERY_PARAM);
 
     public static IndicesOptions fromXContent(XContentParser parser) throws IOException {
         return fromXContent(parser, null);
@@ -1306,7 +1363,7 @@ public record IndicesOptions(
         WildcardOptions.Builder wildcards = defaults == null ? null : WildcardOptions.builder(defaults.wildcardOptions());
         GatekeeperOptions.Builder generalOptions = GatekeeperOptions.builder()
             .ignoreThrottled(defaults != null && defaults.gatekeeperOptions().ignoreThrottled());
-        FailureStoreOptions failureStoreOptions = defaults == null ? FailureStoreOptions.DEFAULT : defaults.failureStoreOptions();
+        SelectorOptions selectorOptions = defaults == null ? SelectorOptions.DEFAULT : defaults.selectorOptions();
         Boolean allowNoIndices = defaults == null ? null : defaults.allowNoIndices();
         Boolean ignoreUnavailable = defaults == null ? null : defaults.ignoreUnavailable();
         Token token = parser.currentToken() == Token.START_OBJECT ? parser.currentToken() : parser.nextToken();
@@ -1358,7 +1415,7 @@ public record IndicesOptions(
                     generalOptions.ignoreThrottled(parser.booleanValue());
                 } else if (DataStream.isFailureStoreFeatureFlagEnabled()
                     && FAILURE_STORE_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
-                        failureStoreOptions = FailureStoreOptions.parseParameters(parser.text(), failureStoreOptions);
+                        selectorOptions = parseFailureStoreParameters(parser.text(), selectorOptions);
                     } else {
                         throw new ElasticsearchParseException(
                             "could not read indices options. Unexpected index option [" + currentFieldName + "]"
@@ -1389,7 +1446,7 @@ public record IndicesOptions(
             .concreteTargetOptions(new ConcreteTargetOptions(ignoreUnavailable))
             .wildcardOptions(wildcards)
             .gatekeeperOptions(generalOptions)
-            .failureStoreOptions(failureStoreOptions)
+            .selectorOptions(selectorOptions)
             .build();
     }
 

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRolloverIndexAction.java

@@ -71,7 +71,7 @@ public class RestRolloverIndexAction extends BaseRestHandler {
             if (failureStore) {
                 rolloverIndexRequest.setIndicesOptions(
                     IndicesOptions.builder(rolloverIndexRequest.indicesOptions())
-                        .failureStoreOptions(new IndicesOptions.FailureStoreOptions(false, true))
+                        .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                         .build()
                 );
             }

server/src/test/java/org/elasticsearch/action/admin/indices/get/GetIndexRequestTests.java

@@ -82,9 +82,6 @@ public class GetIndexRequestTests extends ESTestCase {
         );
         assertThat(getIndexRequest.indicesOptions().wildcardOptions(), equalTo(IndicesOptions.strictExpandOpen().wildcardOptions()));
         assertThat(getIndexRequest.indicesOptions().gatekeeperOptions(), equalTo(IndicesOptions.strictExpandOpen().gatekeeperOptions()));
-        assertThat(
-            getIndexRequest.indicesOptions().failureStoreOptions(),
-            equalTo(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true).build())
-        );
+        assertThat(getIndexRequest.indicesOptions().selectorOptions(), equalTo(IndicesOptions.SelectorOptions.DATA_AND_FAILURE));
     }
 }

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java

@@ -13,7 +13,7 @@ import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.support.ActiveShardCount;
-import org.elasticsearch.action.support.IndicesOptions.FailureStoreOptions;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.AliasAction;
@@ -747,14 +747,14 @@ public class MetadataRolloverServiceTests extends ESTestCase {
         final String defaultRolloverIndexName;
         final boolean useDataStream = randomBoolean();
         final Metadata.Builder builder = Metadata.builder();
-        var failureStoreOptions = FailureStoreOptions.DEFAULT;
+        var defaultSelectorOptions = IndicesOptions.SelectorOptions.DEFAULT;
         if (useDataStream) {
             DataStream dataStream = DataStreamTestHelper.randomInstance()
                 // ensure no replicate data stream
                 .promoteDataStream();
             rolloverTarget = dataStream.getName();
             if (dataStream.isFailureStoreEnabled() && randomBoolean()) {
-                failureStoreOptions = new FailureStoreOptions(false, true);
+                defaultSelectorOptions = IndicesOptions.SelectorOptions.ONLY_FAILURES;
                 sourceIndexName = dataStream.getFailureStoreWriteIndex().getName();
                 defaultRolloverIndexName = DataStream.getDefaultFailureStoreName(
                     dataStream.getName(),
@@ -815,7 +815,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
             true,
             null,
             null,
-            failureStoreOptions.includeFailureIndices()
+            IndicesOptions.SelectorOptions.ONLY_FAILURES.equals(defaultSelectorOptions)
         );
 
         newIndexName = newIndexName == null ? defaultRolloverIndexName : newIndexName;

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/RolloverRequestTests.java

@@ -11,6 +11,7 @@ package org.elasticsearch.action.admin.indices.rollover;
 
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -36,7 +37,9 @@ import org.elasticsearch.xcontent.json.JsonXContent;
 import org.junit.Before;
 
 import java.io.IOException;
+import java.util.EnumSet;
 import java.util.Map;
+import java.util.Set;
 
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -179,7 +182,12 @@ public class RolloverRequestTests extends ESTestCase {
         originalRequest.lazy(randomBoolean());
         originalRequest.setIndicesOptions(
             IndicesOptions.builder(originalRequest.indicesOptions())
-                .failureStoreOptions(new IndicesOptions.FailureStoreOptions(randomBoolean(), randomBoolean()))
+                .selectorOptions(
+                    IndicesOptions.SelectorOptions.builder()
+                        .setDefaultSelectors(
+                            EnumSet.copyOf(randomNonEmptySubsetOf(Set.of(IndexComponentSelector.DATA, IndexComponentSelector.FAILURES)))
+                        )
+                )
                 .build()
         );
 
@@ -191,10 +199,7 @@ public class RolloverRequestTests extends ESTestCase {
                 assertThat(cloneRequest.getNewIndexName(), equalTo(originalRequest.getNewIndexName()));
                 assertThat(cloneRequest.getRolloverTarget(), equalTo(originalRequest.getRolloverTarget()));
                 assertThat(cloneRequest.isLazy(), equalTo(originalRequest.isLazy()));
-                assertThat(
-                    cloneRequest.indicesOptions().failureStoreOptions(),
-                    equalTo(originalRequest.indicesOptions().failureStoreOptions())
-                );
+                assertThat(cloneRequest.indicesOptions().selectorOptions(), equalTo(originalRequest.indicesOptions().selectorOptions()));
                 for (Map.Entry<String, Condition<?>> entry : cloneRequest.getConditions().getConditions().entrySet()) {
                     Condition<?> condition = originalRequest.getConditions().getConditions().get(entry.getKey());
                     // here we compare the string representation as there is some information loss when serializing
@@ -264,7 +269,7 @@ public class RolloverRequestTests extends ESTestCase {
             RolloverRequest rolloverRequest = new RolloverRequest("alias-index", "new-index-name");
             rolloverRequest.setIndicesOptions(
                 IndicesOptions.builder(rolloverRequest.indicesOptions())
-                    .failureStoreOptions(new IndicesOptions.FailureStoreOptions(true, true))
+                    .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                     .build()
             );
             ActionRequestValidationException validationException = rolloverRequest.validate();

server/src/test/java/org/elasticsearch/action/support/IndicesOptionsTests.java

@@ -10,7 +10,6 @@
 package org.elasticsearch.action.support;
 
 import org.elasticsearch.action.support.IndicesOptions.ConcreteTargetOptions;
-import org.elasticsearch.action.support.IndicesOptions.FailureStoreOptions;
 import org.elasticsearch.action.support.IndicesOptions.GatekeeperOptions;
 import org.elasticsearch.action.support.IndicesOptions.WildcardOptions;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -31,9 +30,11 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 
@@ -57,8 +58,12 @@ public class IndicesOptionsTests extends ESTestCase {
                         .allowAliasToMultipleIndices(randomBoolean())
                         .allowClosedIndices(randomBoolean())
                 )
-                .failureStoreOptions(
-                    FailureStoreOptions.builder().includeRegularIndices(randomBoolean()).includeFailureIndices(randomBoolean())
+                .selectorOptions(
+                    IndicesOptions.SelectorOptions.builder()
+                        .setDefaultSelectors(
+                            EnumSet.copyOf(randomNonEmptySubsetOf(Set.of(IndexComponentSelector.DATA, IndexComponentSelector.FAILURES)))
+                        )
+                        .build()
                 )
                 .build();
 
@@ -345,9 +350,11 @@ public class IndicesOptionsTests extends ESTestCase {
             randomBoolean()
         );
         GatekeeperOptions gatekeeperOptions = new GatekeeperOptions(randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean());
-        FailureStoreOptions failureStoreOptions = new IndicesOptions.FailureStoreOptions(randomBoolean(), randomBoolean());
+        IndicesOptions.SelectorOptions selectorOptions = new IndicesOptions.SelectorOptions(
+            EnumSet.copyOf(randomNonEmptySubsetOf(Set.of(IndexComponentSelector.DATA, IndexComponentSelector.FAILURES)))
+        );
 
-        IndicesOptions indicesOptions = new IndicesOptions(concreteTargetOptions, wildcardOptions, gatekeeperOptions, failureStoreOptions);
+        IndicesOptions indicesOptions = new IndicesOptions(concreteTargetOptions, wildcardOptions, gatekeeperOptions, selectorOptions);
 
         XContentType type = randomFrom(XContentType.values());
         BytesReference xContentBytes = toXContentBytes(indicesOptions, type);
@@ -362,7 +369,15 @@ public class IndicesOptionsTests extends ESTestCase {
         assertThat(map.get("ignore_unavailable"), equalTo(concreteTargetOptions.allowUnavailableTargets()));
         assertThat(map.get("allow_no_indices"), equalTo(wildcardOptions.allowEmptyExpressions()));
         assertThat(map.get("ignore_throttled"), equalTo(gatekeeperOptions.ignoreThrottled()));
-        assertThat(map.get("failure_store"), equalTo(failureStoreOptions.displayValue()));
+        String displayValue;
+        if (IndicesOptions.SelectorOptions.DATA_AND_FAILURE.equals(selectorOptions)) {
+            displayValue = "include";
+        } else if (IndicesOptions.SelectorOptions.ONLY_DATA.equals(selectorOptions)) {
+            displayValue = "exclude";
+        } else {
+            displayValue = "only";
+        }
+        assertThat(map.get("failure_store"), equalTo(displayValue));
     }
 
     public void testFromXContent() throws IOException {

server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java

@@ -2319,7 +2319,8 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
                 new IndicesOptions(
                     IndicesOptions.ConcreteTargetOptions.ERROR_WHEN_UNAVAILABLE_TARGETS,
                     IndicesOptions.WildcardOptions.DEFAULT,
-                    IndicesOptions.GatekeeperOptions.builder().ignoreThrottled(true).build()
+                    IndicesOptions.GatekeeperOptions.builder().ignoreThrottled(true).build(),
+                    IndicesOptions.SelectorOptions.DEFAULT
                 ),
                 "ind*",
                 "test-index"
@@ -2757,7 +2758,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test include failure store with an exact data stream name
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-data-stream");
             assertThat(result.length, equalTo(4));
@@ -2771,7 +2772,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // We expect that they will be skipped
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                 .gatekeeperOptions(IndicesOptions.GatekeeperOptions.builder().allowFailureIndices(false).build())
                 .concreteTargetOptions(IndicesOptions.ConcreteTargetOptions.ALLOW_UNAVAILABLE_TARGETS)
                 .build();
@@ -2785,7 +2786,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // We expect an error
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                 .gatekeeperOptions(IndicesOptions.GatekeeperOptions.builder().allowFailureIndices(false).build())
                 .build();
             FailureIndexNotSupportedException failureIndexNotSupportedException = expectThrows(
@@ -2801,7 +2802,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test only failure store with an exact data stream name
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(false).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-data-stream");
             assertThat(result.length, equalTo(2));
@@ -2828,7 +2829,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test include failure store without any expressions
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true);
             assertThat(result.length, equalTo(5));
@@ -2848,7 +2849,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test only failure store without any expressions
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(false).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true);
             assertThat(result.length, equalTo(2));
@@ -2881,7 +2882,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test include failure store with wildcard expression
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(true).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.DATA_AND_FAILURE)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-*");
             assertThat(result.length, equalTo(5));
@@ -2901,7 +2902,7 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
         // Test only failure store with wildcard expression
         {
             IndicesOptions indicesOptions = IndicesOptions.builder(IndicesOptions.STRICT_EXPAND_OPEN)
-                .failureStoreOptions(IndicesOptions.FailureStoreOptions.builder().includeRegularIndices(false).includeFailureIndices(true))
+                .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                 .build();
             Index[] result = indexNameExpressionResolver.concreteIndices(state, indicesOptions, true, "my-*");
             assertThat(result.length, equalTo(2));

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/RolloverStep.java

@@ -127,7 +127,7 @@ public class RolloverStep extends AsyncActionStep {
         if (targetFailureStore) {
             rolloverRequest.setIndicesOptions(
                 IndicesOptions.builder(rolloverRequest.indicesOptions())
-                    .failureStoreOptions(opts -> opts.includeFailureIndices(true).includeRegularIndices(false))
+                    .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                     .build()
             );
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStep.java

@@ -248,7 +248,7 @@ public class WaitForRolloverReadyStep extends AsyncWaitStep {
         if (targetFailureStore) {
             rolloverRequest.setIndicesOptions(
                 IndicesOptions.builder(rolloverRequest.indicesOptions())
-                    .failureStoreOptions(opts -> opts.includeFailureIndices(true).includeRegularIndices(false))
+                    .selectorOptions(IndicesOptions.SelectorOptions.ONLY_FAILURES)
                     .build()
             );
         }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStepTests.java

@@ -308,8 +308,8 @@ public class WaitForRolloverReadyStepTests extends AbstractStepTestCase<WaitForR
         verify(indicesClient, Mockito.only()).rolloverIndex(requestCaptor.capture(), Mockito.any());
 
         RolloverRequest request = requestCaptor.getValue();
-        assertThat(request.indicesOptions().failureStoreOptions().includeFailureIndices(), equalTo(failureStoreIndex));
-        assertThat(request.indicesOptions().failureStoreOptions().includeRegularIndices(), not(equalTo(failureStoreIndex)));
+        assertThat(request.indicesOptions().includeFailureIndices(), equalTo(failureStoreIndex));
+        assertThat(request.indicesOptions().includeRegularIndices(), not(equalTo(failureStoreIndex)));
     }
 
     public void testSkipRolloverIfDataStreamIsAlreadyRolledOver() {