Converting an Existing Data Stream to a System DataStream is Broken

docs/changelog/121392.yaml

@@ -0,0 +1,5 @@
+pr: 121392
+summary: Converting an Existing Data Stream to a System `DataStream` is Broken
+area: Infra/Core
+type: bug
+issues: []

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

@@ -301,6 +301,13 @@ public boolean isFailureStoreIndex(String indexName) {
         return failureIndices.containsIndex(indexName);
     }
 
+    /**
+     * Returns true if the index name provided belongs to this data stream.
+     */
+    public boolean containsIndex(String indexName) {
+        return backingIndices.containsIndex(indexName) || failureIndices.containsIndex(indexName);
+    }
+
     public DataStreamOptions getDataStreamOptions() {
         return dataStreamOptions;
     }
@@ -1039,7 +1046,7 @@ private boolean isIndexOlderThan(
      * we return false.
      */
     public boolean isIndexManagedByDataStreamLifecycle(Index index, Function<String, IndexMetadata> indexMetadataSupplier) {
-        if (backingIndices.containsIndex(index.getName()) == false && failureIndices.containsIndex(index.getName()) == false) {
+        if (containsIndex(index.getName()) == false) {
             return false;
         }
         IndexMetadata indexMetadata = indexMetadataSupplier.apply(index.getName());

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

@@ -129,14 +129,28 @@ private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String
     }
 
     // visible for testing
-    SystemIndexMetadataUpdateTask getTask() {
-        return new SystemIndexMetadataUpdateTask();
+    ClusterState executeMetadataUpdateTask(ClusterState clusterState) {
+        return new SystemIndexMetadataUpdateTask().execute(clusterState);
     }
 
-    public class SystemIndexMetadataUpdateTask extends ClusterStateUpdateTask {
+    private class SystemIndexMetadataUpdateTask extends ClusterStateUpdateTask {
 
         @Override
-        public ClusterState execute(ClusterState currentState) throws Exception {
+        public ClusterState execute(ClusterState currentState) {
+            List<IndexMetadata> updatedMetadata = updateIndices(currentState);
+            List<DataStream> updatedDataStreams = updateDataStreams(currentState);
+
+            if (updatedMetadata.isEmpty() == false || updatedDataStreams.isEmpty() == false) {
+                Metadata.Builder builder = Metadata.builder(currentState.metadata());
+                updatedMetadata.forEach(idxMeta -> builder.put(idxMeta, true));
+                updatedDataStreams.forEach(builder::put);
+
+                return ClusterState.builder(currentState).metadata(builder).build();
+            }
+            return currentState;
+        }
+
+        private List<IndexMetadata> updateIndices(ClusterState currentState) {
             final Map<String, IndexMetadata> indexMetadataMap = currentState.metadata().indices();
             final List<IndexMetadata> updatedMetadata = new ArrayList<>();
             for (Map.Entry<String, IndexMetadata> entry : indexMetadataMap.entrySet()) {
@@ -172,13 +186,23 @@ public ClusterState execute(ClusterState currentState) throws Exception {
                     updatedMetadata.add(builder.build());
                 }
             }
+            return updatedMetadata;
+        }
 
-            if (updatedMetadata.isEmpty() == false) {
-                final Metadata.Builder builder = Metadata.builder(currentState.metadata());
-                updatedMetadata.forEach(idxMeta -> builder.put(idxMeta, true));
-                return ClusterState.builder(currentState).metadata(builder).build();
+        private List<DataStream> updateDataStreams(ClusterState currentState) {
+            List<DataStream> updatedDataStreams = new ArrayList<>();
+            for (DataStream dataStream : currentState.getMetadata().dataStreams().values()) {
+                boolean shouldBeSystem = systemIndices.isSystemDataStream(dataStream.getName());
+                if (dataStream.isSystem() != shouldBeSystem) {
+                    DataStream.Builder dataStreamBuilder = dataStream.copy().setSystem(shouldBeSystem);
+                    if (shouldBeSystem) {
+                        dataStreamBuilder.setHidden(true);
+                    }
+
+                    updatedDataStreams.add(dataStreamBuilder.build());
+                }
             }
-            return currentState;
+            return updatedDataStreams;
         }
 
         @Override

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

@@ -14,11 +14,14 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.indices.ExecutorNames;
+import org.elasticsearch.indices.SystemDataStreamDescriptor;
 import org.elasticsearch.indices.SystemIndexDescriptor;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -49,13 +52,35 @@ public class SystemIndexMetadataUpgradeServiceTests extends ESTestCase {
         .setOrigin("FAKE_ORIGIN")
         .build();
 
+    private static final String SYSTEM_DATA_STREAM_NAME = ".my-ds";
+    private static final String SYSTEM_DATA_STREAM_INDEX_NAME = DataStream.BACKING_INDEX_PREFIX + SYSTEM_DATA_STREAM_NAME + "-1";
+    private static final SystemDataStreamDescriptor SYSTEM_DATA_STREAM_DESCRIPTOR = new SystemDataStreamDescriptor(
+        SYSTEM_DATA_STREAM_NAME,
+        "System datastream for test",
+        SystemDataStreamDescriptor.Type.INTERNAL,
+        ComposableIndexTemplate.builder().build(),
+        Collections.emptyMap(),
+        Collections.singletonList("FAKE_ORIGIN"),
+        ExecutorNames.DEFAULT_SYSTEM_DATA_STREAM_THREAD_POOLS
+    );
+
     private SystemIndexMetadataUpgradeService service;
 
     @Before
     public void setUpTest() {
         // set up a system index upgrade service
         this.service = new SystemIndexMetadataUpgradeService(
-            new SystemIndices(List.of(new SystemIndices.Feature("foo", "a test feature", List.of(DESCRIPTOR)))),
+            new SystemIndices(
+                List.of(
+                    new SystemIndices.Feature("foo", "a test feature", List.of(DESCRIPTOR)),
+                    new SystemIndices.Feature(
+                        "sds",
+                        "system data stream feature",
+                        Collections.emptyList(),
+                        Collections.singletonList(SYSTEM_DATA_STREAM_DESCRIPTOR)
+                    )
+                )
+            ),
             mock(ClusterService.class)
         );
     }
@@ -75,6 +100,40 @@ public void testUpgradeVisibleIndexToSystemIndex() throws Exception {
         assertSystemUpgradeAppliesHiddenSetting(hiddenIndexMetadata);
     }
 
+    public void testUpgradeDataStreamToSystemDataStream() {
+        IndexMetadata dsIndexMetadata = IndexMetadata.builder(SYSTEM_DATA_STREAM_INDEX_NAME)
+            .system(false)
+            .settings(getSettingsBuilder().put(IndexMetadata.SETTING_INDEX_HIDDEN, true))
+            .build();
+        DataStream dataStream = DataStream.builder(SYSTEM_DATA_STREAM_NAME, Collections.singletonList(dsIndexMetadata.getIndex()))
+            .setHidden(false)
+            .setSystem(false)
+            .build();
+
+        assertTrue(dataStream.containsIndex(dsIndexMetadata.getIndex().getName()));
+        assertTrue("Metadata should require update but does not", service.requiresUpdate(dsIndexMetadata));
+
+        Metadata.Builder clusterMetadata = new Metadata.Builder();
+        clusterMetadata.put(dataStream);
+        clusterMetadata.put(dsIndexMetadata, true);
+
+        ClusterState clusterState = ClusterState.builder(new ClusterName("system-index-metadata-upgrade-service-tests"))
+            .metadata(clusterMetadata.build())
+            .customs(Map.of())
+            .build();
+
+        // Execute a metadata upgrade task on the initial cluster state
+        ClusterState newState = service.executeMetadataUpdateTask(clusterState);
+
+        DataStream updatedDataStream = newState.metadata().dataStreams().get(dataStream.getName());
+        assertThat(updatedDataStream.isSystem(), equalTo(true));
+        assertThat(updatedDataStream.isHidden(), equalTo(true));
+
+        IndexMetadata updatedIndexMetadata = newState.metadata().index(dsIndexMetadata.getIndex().getName());
+        assertThat(updatedIndexMetadata.isSystem(), equalTo(true));
+        assertThat(updatedIndexMetadata.isHidden(), equalTo(true));
+    }
+
     /**
      * If a system index erroneously is set to visible, we should remedy that situation.
      */
@@ -209,7 +268,7 @@ public void testIsVisible() {
         assertThat(service.requiresUpdate(systemVisibleIndex), equalTo(true));
     }
 
-    private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMetadata) throws Exception {
+    private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMetadata) {
         assertTrue("Metadata should require update but does not", service.requiresUpdate(hiddenIndexMetadata));
         Metadata.Builder clusterMetadata = new Metadata.Builder();
         clusterMetadata.put(IndexMetadata.builder(hiddenIndexMetadata));
@@ -220,7 +279,7 @@ private void assertSystemUpgradeAppliesHiddenSetting(IndexMetadata hiddenIndexMe
             .build();
 
         // Get a metadata upgrade task and execute it on the initial cluster state
-        ClusterState newState = service.getTask().execute(clusterState);
+        ClusterState newState = service.executeMetadataUpdateTask(clusterState);
 
         IndexMetadata result = newState.metadata().index(SYSTEM_INDEX_NAME);
         assertThat(result.isSystem(), equalTo(true));
@@ -238,7 +297,7 @@ private void assertSystemUpgradeHidesAlias(IndexMetadata visibleAliasMetadata) t
             .build();
 
         // Get a metadata upgrade task and execute it on the initial cluster state
-        ClusterState newState = service.getTask().execute(clusterState);
+        ClusterState newState = service.executeMetadataUpdateTask(clusterState);
 
         IndexMetadata result = newState.metadata().index(SYSTEM_INDEX_NAME);
         assertThat(result.isSystem(), equalTo(true));