Simplify Feature State Snapshot Code (#69433) (#69523)

We can do a lot of the checks on the request and its version
compatibility off the cluster state update thread.
Also, the logic for working out the indices to add to the snapshot
due to requested features was quit complicated and did a lot of redundant
checks and loops so I simplified it where possible.

Author: original-brownbear

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -385,6 +385,44 @@ public void createSnapshot(final CreateSnapshotRequest request, final ActionList
             return;
         }
         final Snapshot snapshot = new Snapshot(repositoryName, snapshotId);
+
+        // We should only use the feature states logic if we're sure we'll be able to finish the snapshot without a lower-version
+        // node taking over and causing problems. Therefore, if we're in a mixed cluster with versions that don't know how to handle
+        // feature states, skip all feature states logic, and if `feature_states` is explicitly configured, throw an exception.
+        final List<String> requestedStates = Arrays.asList(request.featureStates());
+        final Version initialMinNodeVersion = clusterService.state().nodes().getMinNodeVersion();
+        final Set<String> featureStatesSet;
+        if (initialMinNodeVersion.onOrAfter(FEATURE_STATES_VERSION)) {
+            if (request.includeGlobalState() || requestedStates.isEmpty() == false) {
+                if (request.includeGlobalState() && requestedStates.isEmpty()) {
+                    // If we're including global state and feature states aren't specified, include all of them
+                    featureStatesSet = systemIndexDescriptorMap.keySet();
+                } else if (requestedStates.size() == 1 && NO_FEATURE_STATES_VALUE.equalsIgnoreCase(requestedStates.get(0))) {
+                    // If there's exactly one value and it's "none", include no states
+                    featureStatesSet = Collections.emptySet();
+                } else {
+                    // Otherwise, check for "none" then use the list of requested states
+                    if (requestedStates.contains(NO_FEATURE_STATES_VALUE)) {
+                        listener.onFailure(new IllegalArgumentException("the feature_states value [" +
+                                SnapshotsService.NO_FEATURE_STATES_VALUE + "] indicates that no feature states should be snapshotted, " +
+                                "but other feature states were requested: " + requestedStates));
+                        return;
+                    }
+                    featureStatesSet = new HashSet<>(requestedStates);
+                    featureStatesSet.retainAll(systemIndexDescriptorMap.keySet());
+                }
+            } else {
+                featureStatesSet = Collections.emptySet();
+            }
+        } else if (requestedStates.isEmpty() == false) {
+            listener.onFailure(new SnapshotException(snapshot, "feature_states can only be used when all nodes in cluster are version ["
+                    + FEATURE_STATES_VERSION + "] or higher, but at least one node in this cluster is on version ["
+                    + initialMinNodeVersion + "]"));
+            return;
+        } else {
+            featureStatesSet = Collections.emptySet();
+        }
+
         final Map<String, Object> userMeta = repository.adaptUserMetadata(request.userMetadata());
         repository.executeConsistentStateUpdate(repositoryData -> new ClusterStateUpdateTask(request.masterNodeTimeout()) {
 
@@ -421,57 +459,32 @@ public ClusterState execute(ClusterState currentState) {
                 // Store newSnapshot here to be processed in clusterStateProcessed
                 List<String> indices = Arrays.asList(indexNameExpressionResolver.concreteIndexNames(currentState, request));
 
-                List<SnapshotFeatureInfo> featureStates = Collections.emptyList();
-                final List<String> requestedStates = Arrays.asList(request.featureStates());
-
-                // We should only use the feature states logic if we're sure we'll be able to finish the snapshot without a lower-version
-                // node taking over and causing problems. Therefore, if we're in a mixed cluster with versions that don't know how to handle
-                // feature states, skip all feature states logic, and if `feature_states` is explicitly configured, throw an exception.
-                if (currentState.nodes().getMinNodeVersion().onOrAfter(FEATURE_STATES_VERSION)) {
-                    if (request.includeGlobalState() || requestedStates.isEmpty() == false) {
-                        final Set<String> featureStatesSet;
-                        if (request.includeGlobalState() && requestedStates.isEmpty()) {
-                            // If we're including global state and feature states aren't specified, include all of them
-                            featureStatesSet = new HashSet<>(systemIndexDescriptorMap.keySet());
-                        } else if (requestedStates.size() == 1 && NO_FEATURE_STATES_VALUE.equalsIgnoreCase(requestedStates.get(0))) {
-                            // If there's exactly one value and it's "none", include no states
-                            featureStatesSet = Collections.emptySet();
-                        } else {
-                            // Otherwise, check for "none" then use the list of requested states
-                            if (requestedStates.contains(NO_FEATURE_STATES_VALUE)) {
-                                throw new IllegalArgumentException("the feature_states value [" + SnapshotsService.NO_FEATURE_STATES_VALUE +
-                                    "] indicates that no feature states should be snapshotted, but other feature states were requested: " +
-                                    requestedStates);
-                            }
-                            featureStatesSet = new HashSet<>(requestedStates);
-                        }
-
-                        featureStates = systemIndexDescriptorMap.keySet().stream()
-                            .filter(feature -> featureStatesSet.contains(feature))
-                            .map(feature -> new SnapshotFeatureInfo(feature, resolveFeatureIndexNames(currentState, feature)))
+                final List<SnapshotFeatureInfo> featureStates;
+                // if we have any feature states in the snapshot, we add their required indices to the snapshot indices if they haven't
+                // been requested by the request directly
+                if (featureStatesSet.isEmpty()) {
+                    featureStates = Collections.emptyList();
+                } else {
+                    final Set<String> indexNames = new HashSet<>(indices);
+                    featureStates = featureStatesSet.stream()
+                            .map(feature -> new SnapshotFeatureInfo(feature,
+                                    systemIndexDescriptorMap.get(feature).getIndexDescriptors().stream()
+                                            .flatMap(descriptor -> descriptor.getMatchingIndices(currentState.metadata()).stream())
+                                            .collect(Collectors.toList())))
                             .filter(featureInfo -> featureInfo.getIndices().isEmpty() == false) // Omit any empty featureStates
                             .collect(Collectors.toList());
-                        final Stream<String> featureStateIndices = featureStates.stream().flatMap(feature -> feature.getIndices().stream());
-
-                        final Stream<String> associatedIndices = systemIndexDescriptorMap.keySet().stream()
-                            .filter(feature -> featureStatesSet.contains(feature))
-                            .flatMap(feature -> resolveAssociatedIndices(currentState, feature).stream());
+                    for (SnapshotFeatureInfo featureState : featureStates) {
+                        indexNames.addAll(featureState.getIndices());
+                    }
 
-                        // Add all resolved indices from the feature states to the list of indices
-                        indices = Stream.of(indices.stream(), featureStateIndices, associatedIndices)
-                            .flatMap(s -> s)
-                            .distinct()
-                            .collect(Collectors.toList());
+                    // Add all resolved indices from the feature states to the list of indices
+                    for (String feature : featureStatesSet) {
+                        for (String pattern : systemIndexDescriptorMap.get(feature).getAssociatedIndexPatterns()) {
+                            Collections.addAll(indexNames, indexNameExpressionResolver.concreteIndexNamesWithSystemIndexAccess(
+                                    currentState, LENIENT_EXPAND_OPEN_CLOSED_HIDDEN, pattern));
+                        }
                     }
-                } else if (requestedStates.isEmpty() == false) {
-                    throw new SnapshotException(
-                        new Snapshot(repositoryName, snapshotId),
-                        "feature_states can only be used when all nodes in cluster are version ["
-                            + FEATURE_STATES_VERSION
-                            + "] or higher, but at least one node in this cluster is on version ["
-                            + currentState.nodes().getMinNodeVersion()
-                            + "]"
-                    );
+                    indices = Collections.unmodifiableList(new ArrayList<>(indexNames));
                 }
 
                 final List<String> dataStreams =
@@ -524,29 +537,6 @@ public void clusterStateProcessed(String source, ClusterState oldState, final Cl
         }, "create_snapshot [" + snapshotName + ']', listener::onFailure);
     }
 
-    private List<String> resolveFeatureIndexNames(ClusterState currentState, String featureName) {
-        if (systemIndexDescriptorMap.containsKey(featureName) == false) {
-            throw new IllegalArgumentException("requested snapshot of feature state for unknown feature [" + featureName + "]");
-        }
-
-        final SystemIndices.Feature feature = systemIndexDescriptorMap.get(featureName);
-        return feature.getIndexDescriptors().stream()
-            .flatMap(descriptor -> descriptor.getMatchingIndices(currentState.metadata()).stream())
-            .collect(Collectors.toList());
-    }
-
-    private List<String> resolveAssociatedIndices(ClusterState currentState, String featureName) {
-        if (systemIndexDescriptorMap.containsKey(featureName) == false) {
-            throw new IllegalArgumentException("requested associated indices for feature state for unknown feature [" + featureName + "]");
-        }
-
-        final SystemIndices.Feature feature = systemIndexDescriptorMap.get(featureName);
-        return feature.getAssociatedIndexPatterns().stream()
-            .flatMap(pattern -> Arrays.stream(indexNameExpressionResolver.concreteIndexNamesWithSystemIndexAccess(currentState,
-                LENIENT_EXPAND_OPEN_CLOSED_HIDDEN, pattern)))
-            .collect(Collectors.toList());
-    }
-
     private static void ensureSnapshotNameNotRunning(List<SnapshotsInProgress.Entry> runningSnapshots, String repositoryName,
                                                      String snapshotName) {
         if (runningSnapshots.stream().anyMatch(s -> {