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

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/124884.yaml

@@ -0,0 +1,6 @@
+pr: 124884
+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

@@ -323,6 +323,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

@@ -210,6 +210,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

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

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

@@ -283,6 +283,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

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

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

@@ -205,7 +205,7 @@ static void prepareBackingIndex(
         Function<IndexMetadata, MapperService> mapperSupplier,
         boolean removeAlias
     ) throws IOException {
-        prepareBackingIndex(b, im, dataStreamName, mapperSupplier, removeAlias, false, Settings.EMPTY);
+        prepareBackingIndex(b, im, dataStreamName, mapperSupplier, removeAlias, false, false, Settings.EMPTY);
     }
 
     /**
@@ -219,6 +219,8 @@ static void prepareBackingIndex(
      *                    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(
@@ -228,6 +230,7 @@ static void prepareBackingIndex(
         Function<IndexMetadata, MapperService> mapperSupplier,
         boolean removeAlias,
         boolean failureStore,
+        boolean makeSystem,
         Settings nodeSettings
     ) throws IOException {
         MappingMetadata mm = im.mapping();
@@ -258,6 +261,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

@@ -239,6 +239,10 @@ private List<IndexMetadata> updateIndices(ClusterState currentState, List<Index>
             final List<IndexMetadata> updatedMetadata = new ArrayList<>();
             for (Index index : indices) {
                 IndexMetadata indexMetadata = metadata.index(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

@@ -1195,7 +1195,7 @@ static List<Diagnosis.Resource> getRestoreFromSnapshotAffectedResources(
                 .collect(
                     toMap(
                         SystemIndices.Feature::getName,
-                        feature -> feature.getIndexDescriptors()
+                        feature -> feature.getSystemResourceDescriptors()
                             .stream()
                             .flatMap(descriptor -> descriptor.getMatchingIndices(metadata).stream())
                             .collect(toSet())
@@ -1211,29 +1211,6 @@ static List<Diagnosis.Resource> getRestoreFromSnapshotAffectedResources(
                 }
             }
 
-            Map<String, Set<String>> featureToDsBackingIndices = systemIndices.getFeatures()
-                .stream()
-                .collect(
-                    toMap(
-                        SystemIndices.Feature::getName,
-                        feature -> feature.getDataStreamDescriptors()
-                            .stream()
-                            .flatMap(descriptor -> descriptor.getBackingIndexNames(metadata).stream())
-                            .collect(toSet())
-                    )
-                );
-
-            // 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<String>> featureToBackingIndices : featureToDsBackingIndices.entrySet()) {
-                for (String featureIndex : featureToBackingIndices.getValue()) {
-                    if (restoreFromSnapshotIndices.contains(featureIndex)) {
-                        affectedFeatureStates.add(featureToBackingIndices.getKey());
-                        affectedIndices.remove(featureIndex);
-                    }
-                }
-            }
-
             if (affectedIndices.isEmpty() == false) {
                 affectedResources.add(new Diagnosis.Resource(INDEX, affectedIndices.stream().limit(maxAffectedResourcesCount).toList()));
             }