[main] System data streams are not being upgraded in the feature migration API (elastic#126409)

This commit adds support for system data streams reindexing. The system data stream migration extends the existing system indices migration task and uses the data stream reindex API.
The system index migration task starts a reindex data stream task and tracks its status every second. Only one system index or system data stream is migrated at a time. If a data stream migration fails, the entire system index migration task will also fail.

Port of elastic#123926

Author: alexey-ivanov-es

docs/changelog/126409.yaml

@@ -0,0 +1,6 @@
+pr: 126409
+summary: System data streams are not being upgraded in the feature migration API
+area: Infra/Core
+type: bug
+issues:
+  - 122949

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

@@ -326,6 +326,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );

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

@@ -275,6 +275,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                         .build(),
                     Map.of(),
                     Collections.singletonList("test"),
+                    "test",
                     new ExecutorNames(ThreadPool.Names.SYSTEM_CRITICAL_READ, ThreadPool.Names.SYSTEM_READ, ThreadPool.Names.SYSTEM_WRITE)
                 )
             );

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

@@ -213,6 +213,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );

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

@@ -1094,6 +1094,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                         .build(),
                     Map.of(),
                     List.of(),
+                    "test",
                     ExecutorNames.DEFAULT_SYSTEM_INDEX_THREAD_POOLS
                 )
             );

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SystemResourceSnapshotIT.java

@@ -1153,6 +1153,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );
@@ -1192,6 +1193,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );
@@ -1231,6 +1233,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );
@@ -1299,6 +1302,7 @@ public Collection<SystemDataStreamDescriptor> getSystemDataStreamDescriptors() {
                             .build(),
                         Map.of(),
                         List.of("product"),
+                        "product",
                         ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
                     )
                 );

server/src/main/java/module-info.java

@@ -282,6 +282,7 @@
     exports org.elasticsearch.indices.recovery;
     exports org.elasticsearch.indices.recovery.plan;
     exports org.elasticsearch.indices.store;
+    exports org.elasticsearch.indices.system;
     exports org.elasticsearch.inference;
     exports org.elasticsearch.ingest;
     exports org.elasticsearch.internal

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDataStreamsService.java

@@ -382,6 +382,7 @@ private static void addBackingIndex(
                 mapperSupplier,
                 false,
                 failureStore,
+                dataStream.isSystem(),
                 nodeSettings
             );
         } catch (IOException e) {

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMigrateToDataStreamService.java

@@ -156,7 +156,7 @@ static ClusterState migrateToDataStream(
         ProjectMetadata.Builder mb = ProjectMetadata.builder(project);
         for (Index index : alias.getIndices()) {
             IndexMetadata im = project.index(index);
-            prepareBackingIndex(mb, im, request.aliasName, mapperSupplier, true, false, Settings.EMPTY);
+            prepareBackingIndex(mb, im, request.aliasName, mapperSupplier, true, false, false, Settings.EMPTY);
         }
         ClusterState updatedState = ClusterState.builder(projectState.cluster()).putProjectMetadata(mb).build();
 
@@ -212,6 +212,8 @@ static void validateRequest(ProjectMetadata project, MigrateToDataStreamClusterS
      *                    exception should be thrown in that case instead
      * @param failureStore <code>true</code> if the index is being migrated into the data stream's failure store, <code>false</code> if it
      *                     is being migrated into the data stream's backing indices
+     * @param makeSystem <code>true</code> if the index is being migrated into the system data stream, <code>false</code> if it
+     *                     is being migrated into non-system data stream
      * @param nodeSettings The settings for the current node
      */
     static void prepareBackingIndex(
@@ -221,6 +223,7 @@ static void prepareBackingIndex(
         Function<IndexMetadata, MapperService> mapperSupplier,
         boolean removeAlias,
         boolean failureStore,
+        boolean makeSystem,
         Settings nodeSettings
     ) throws IOException {
         MappingMetadata mm = im.mapping();
@@ -251,6 +254,7 @@ static void prepareBackingIndex(
         imb.mappingVersion(im.getMappingVersion() + 1)
             .mappingsUpdatedVersion(IndexVersion.current())
             .putMapping(new MappingMetadata(mapper));
+        imb.system(makeSystem);
         b.put(imb);
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/SystemIndexMetadataUpgradeService.java

@@ -241,6 +241,10 @@ private List<IndexMetadata> updateIndices(ClusterState currentState, List<Index>
             final List<IndexMetadata> updatedMetadata = new ArrayList<>();
             for (Index index : indices) {
                 IndexMetadata indexMetadata = metadata.indexMetadata(index);
+                // this might happen because update is async and the index might have been deleted between task creation and execution
+                if (indexMetadata == null) {
+                    continue;
+                }
                 final boolean shouldBeSystem = shouldBeSystem(indexMetadata);
                 IndexMetadata updatedIndexMetadata = updateIndexIfNecessary(indexMetadata, shouldBeSystem);
                 if (updatedIndexMetadata != null) {

server/src/main/java/org/elasticsearch/cluster/routing/allocation/shards/ShardsAvailabilityHealthIndicatorService.java

@@ -78,7 +78,6 @@
 import java.util.stream.Stream;
 
 import static java.util.stream.Collectors.joining;
-import static java.util.stream.Collectors.toMap;
 import static java.util.stream.Collectors.toSet;
 import static org.elasticsearch.cluster.health.ClusterShardHealth.getInactivePrimaryHealth;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_ROUTING_INCLUDE_GROUP_PREFIX;
@@ -1152,7 +1151,8 @@ public List<HealthIndicatorImpact> getImpacts() {
 
         /**
          * Returns the diagnosis for unassigned primary and replica shards.
-         * @param verbose true if the diagnosis should be generated, false if they should be omitted.
+         *
+         * @param verbose                   true if the diagnosis should be generated, false if they should be omitted.
          * @param maxAffectedResourcesCount the max number of affected resources to be returned as part of the diagnosis
          * @return The diagnoses list the indicator identified. Alternatively, an empty list if none were found or verbose is false.
          */
@@ -1243,23 +1243,6 @@ static List<Diagnosis.Resource> getRestoreFromSnapshotAffectedResources(
                 }
             }
 
-            Map<String, Set<ProjectIndexName>> featureToDsBackingIndices = getSystemDsBackingIndicesForProjects(
-                systemIndices,
-                affectedProjects,
-                metadata
-            );
-
-            // the shards_availability indicator works with indices so let's remove the feature states data streams backing indices from
-            // the list of affected indices (the feature state will cover the restore of these indices too)
-            for (Map.Entry<String, Set<ProjectIndexName>> featureToBackingIndices : featureToDsBackingIndices.entrySet()) {
-                for (ProjectIndexName featureIndex : featureToBackingIndices.getValue()) {
-                    if (restoreFromSnapshotIndices.contains(featureIndex)) {
-                        affectedFeatureStates.add(featureToBackingIndices.getKey());
-                        affectedIndices.remove(featureIndex);
-                    }
-                }
-            }
-
             if (affectedIndices.isEmpty() == false) {
                 affectedResources.add(
                     new Diagnosis.Resource(
@@ -1281,7 +1264,7 @@ static List<Diagnosis.Resource> getRestoreFromSnapshotAffectedResources(
         }
 
         /**
-         * Retrieve the system indices for the projects and group them by Feature
+         * Retrieve the system indices and indices backing system data streams for the projects and group them by Feature
          */
         private static Map<String, Set<ProjectIndexName>> getSystemIndicesForProjects(
             SystemIndices systemIndices,
@@ -1293,7 +1276,7 @@ private static Map<String, Set<ProjectIndexName>> getSystemIndicesForProjects(
                 .collect(
                     Collectors.toMap(
                         SystemIndices.Feature::getName,
-                        feature -> feature.getIndexDescriptors()
+                        feature -> feature.getSystemResourceDescriptors()
                             .stream()
                             .flatMap(
                                 descriptor -> projects.stream()
@@ -1307,34 +1290,6 @@ private static Map<String, Set<ProjectIndexName>> getSystemIndicesForProjects(
                     )
                 );
         }
-
-        /**
-         * Retrieve the backing indices for system data stream for the projects and group them by Feature
-         */
-        private static Map<String, Set<ProjectIndexName>> getSystemDsBackingIndicesForProjects(
-            SystemIndices systemIndices,
-            Set<ProjectId> projects,
-            Metadata metadata
-        ) {
-            return systemIndices.getFeatures()
-                .stream()
-                .collect(
-                    toMap(
-                        SystemIndices.Feature::getName,
-                        feature -> feature.getDataStreamDescriptors()
-                            .stream()
-                            .flatMap(
-                                descriptor -> projects.stream()
-                                    .flatMap(
-                                        projectId -> descriptor.getBackingIndexNames(metadata.getProject(projectId))
-                                            .stream()
-                                            .map(index -> new ProjectIndexName(projectId, index))
-                                    )
-                            )
-                            .collect(Collectors.toSet())
-                    )
-                );
-        }
     }
 
     public static class SearchableSnapshotsState {

server/src/main/java/org/elasticsearch/indices/AssociatedIndexDescriptor.java

@@ -13,6 +13,7 @@
 import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.apache.lucene.util.automaton.RegExp;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.indices.system.IndexPatternMatcher;
 
 import java.util.List;
 import java.util.Objects;

server/src/main/java/org/elasticsearch/indices/IndexPatternMatcher.java renamed to server/src/main/java/org/elasticsearch/indices/IndexMatcher.java

@@ -15,23 +15,15 @@
 import java.util.List;
 
 /**
- * An IndexPatternMatcher holds an index pattern in a string and, given a
- * {@link Metadata} object, can return a list of index names matching that pattern.
+ * An IndexMatcher given a {@link Metadata} object, can return a list of index names matching that pattern.
  */
-public interface IndexPatternMatcher {
-    /**
-     * @return A pattern, either with a wildcard or simple regex, describing indices that are
-     * related to a system feature. Such indices may be system indices or associated
-     * indices.
-     */
-    String getIndexPattern();
-
+public interface IndexMatcher {
     /**
      * Retrieves a list of all indices which match this descriptor's pattern. Implementations
      * may include other special information when matching indices, such as aliases.
-     *
+     * <p>
      * This cannot be done via {@link org.elasticsearch.cluster.metadata.IndexNameExpressionResolver} because that class can only handle
-     * simple wildcard expressions, but system index name patterns may use full Lucene regular expression syntax,
+     * simple wildcard expressions, but system index name patterns may use full Lucene regular expression syntax.
      *
      * @param project The current metadata to get the list of matching indices from
      * @return A list of index names that match this descriptor

server/src/main/java/org/elasticsearch/indices/SystemDataStreamDescriptor.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.indices.system.SystemResourceDescriptor;
 
 import java.util.Collections;
 import java.util.List;
@@ -45,14 +46,15 @@
  * <p>The descriptor also provides names for the thread pools that Elasticsearch should use to read, search, or modify the descriptor’s
  * indices.
  */
-public class SystemDataStreamDescriptor {
+public class SystemDataStreamDescriptor implements SystemResourceDescriptor {
 
     private final String dataStreamName;
     private final String description;
     private final Type type;
     private final ComposableIndexTemplate composableIndexTemplate;
     private final Map<String, ComponentTemplate> componentTemplates;
     private final List<String> allowedElasticProductOrigins;
+    private final String origin;
     private final ExecutorNames executorNames;
 
     /**
@@ -66,6 +68,7 @@ public class SystemDataStreamDescriptor {
      *                           {@link ComposableIndexTemplate}
      * @param allowedElasticProductOrigins a list of product origin values that are allowed to access this data stream if the
      *                                     type is {@link Type#EXTERNAL}. Must not be {@code null}
+     * @param origin specifies the origin to use when creating or updating the data stream
      * @param executorNames thread pools that should be used for operations on the system data stream
      */
     public SystemDataStreamDescriptor(
@@ -75,6 +78,7 @@ public SystemDataStreamDescriptor(
         ComposableIndexTemplate composableIndexTemplate,
         Map<String, ComponentTemplate> componentTemplates,
         List<String> allowedElasticProductOrigins,
+        String origin,
         ExecutorNames executorNames
     ) {
         this.dataStreamName = Objects.requireNonNull(dataStreamName, "dataStreamName must be specified");
@@ -96,6 +100,7 @@ public SystemDataStreamDescriptor(
             throw new IllegalArgumentException("External system data stream without allowed products is not a valid combination");
         }
         this.executorNames = Objects.nonNull(executorNames) ? executorNames : ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS;
+        this.origin = origin;
     }
 
     public String getDataStreamName() {
@@ -125,6 +130,11 @@ public List<String> getBackingIndexNames(ProjectMetadata projectMetadata) {
         return Stream.concat(dataStream.getIndices().stream(), dataStream.getFailureIndices().stream()).map(Index::getName).toList();
     }
 
+    @Override
+    public List<String> getMatchingIndices(ProjectMetadata metadata) {
+        return getBackingIndexNames(metadata);
+    }
+
     public String getDescription() {
         return description;
     }
@@ -133,6 +143,17 @@ public ComposableIndexTemplate getComposableIndexTemplate() {
         return composableIndexTemplate;
     }
 
+    @Override
+    public String getOrigin() {
+        return origin;
+    }
+
+    @Override
+    public boolean isAutomaticallyManaged() {
+        return true;
+    }
+
+    @Override
     public boolean isExternal() {
         return type == Type.EXTERNAL;
     }
@@ -142,9 +163,10 @@ public String getBackingIndexPattern() {
     }
 
     private static String backingIndexPatternForDataStream(String dataStream) {
-        return DataStream.BACKING_INDEX_PREFIX + dataStream + "-*";
+        return ".(migrated-){0,}[fd]s-" + dataStream + "-*";
     }
 
+    @Override
     public List<String> getAllowedElasticProductOrigins() {
         return allowedElasticProductOrigins;
     }
@@ -157,6 +179,7 @@ public Map<String, ComponentTemplate> getComponentTemplates() {
      * Get the names of the thread pools that should be used for operations on this data stream.
      * @return Names for get, search, and write executors.
      */
+    @Override
     public ExecutorNames getThreadPoolNames() {
         return this.executorNames;
     }