[8.2] Allow the modify data stream api to remove broken reference to a backing index (#87030) (#87084)

* Allow the modify data stream api to remove broken reference to a backing index (#87030)

This adds a `force_remove_backing_index` action to the modify data
stream api to allow removing broken reference to backing indices from a
data stream.

```
POST _data_stream/_modify
{
  "actions": [
    {
      "force_remove_backing_index": {
        "data_stream": "my-logs",
        "index": ".ds-my-logs-2099.01.01-000001"
      }
    }
  ]
}
```

* Fix compilation for backport

* Fix test

Co-authored-by: Lee Hinman <[email protected]>

Author: martijnvg

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
 import org.elasticsearch.action.admin.indices.rollover.RolloverResponse;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
 import org.elasticsearch.action.admin.indices.template.delete.DeleteComposableIndexTemplateAction;
 import org.elasticsearch.action.admin.indices.template.get.GetComposableIndexTemplateAction;
 import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
@@ -40,6 +41,7 @@
 import org.elasticsearch.action.datastreams.DeleteDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.DataStreamInfo;
+import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
@@ -50,17 +52,23 @@
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateTaskExecutor;
+import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.DataStreamAction;
 import org.elasticsearch.cluster.metadata.DataStreamAlias;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.Template;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
 import org.elasticsearch.index.mapper.DateFieldMapper;
@@ -88,9 +96,11 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -1711,6 +1721,74 @@ public void testCreateIndexAliasWithSameNameAsDataStreamAlias() throws Exception
         }
     }
 
+    public void testRemoveGhostReference() throws Exception {
+        String dataStreamName = "logs-es";
+        DataStreamIT.putComposableIndexTemplate("my-template", List.of("logs-*"));
+        var request = new CreateDataStreamAction.Request(dataStreamName);
+        assertAcked(client().execute(CreateDataStreamAction.INSTANCE, request).actionGet());
+        assertAcked(client().admin().indices().rolloverIndex(new RolloverRequest(dataStreamName, null)).actionGet());
+        var indicesStatsResponse = client().admin().indices().stats(new IndicesStatsRequest()).actionGet();
+        assertThat(indicesStatsResponse.getIndices().size(), equalTo(2));
+        ClusterState before = internalCluster().getCurrentMasterNodeInstance(ClusterService.class).state();
+        assertThat(before.getMetadata().dataStreams().get(dataStreamName).getIndices(), hasSize(2));
+
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<DataStream> brokenDataStreamHolder = new AtomicReference<>();
+        internalCluster().getCurrentMasterNodeInstance(ClusterService.class)
+            .submitStateUpdateTask(getTestName(), new ClusterStateUpdateTask() {
+                @Override
+                public ClusterState execute(ClusterState currentState) throws Exception {
+                    DataStream original = currentState.getMetadata().dataStreams().get(dataStreamName);
+                    DataStream broken = new DataStream(
+                        original.getName(),
+                        List.of(new Index(original.getIndices().get(0).getName(), "broken"), original.getIndices().get(1)),
+                        original.getGeneration(),
+                        original.getMetadata(),
+                        original.isHidden(),
+                        original.isReplicated(),
+                        original.isSystem(),
+                        original.isAllowCustomRouting(),
+                        original.getIndexMode()
+                    );
+                    brokenDataStreamHolder.set(broken);
+                    return ClusterState.builder(currentState)
+                        .metadata(Metadata.builder(currentState.getMetadata()).put(broken).build())
+                        .build();
+                }
+
+                @Override
+                public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
+                    latch.countDown();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    logger.error("error while adding a broken data stream", e);
+                    latch.countDown();
+                }
+            }, ClusterStateTaskExecutor.unbatched());
+        latch.await();
+        var ghostReference = brokenDataStreamHolder.get().getIndices().get(0);
+
+        // Many APIs fail with NPE, because of broken data stream:
+        expectThrows(NullPointerException.class, () -> client().admin().indices().stats(new IndicesStatsRequest()).actionGet());
+        expectThrows(NullPointerException.class, () -> client().search(new SearchRequest()).actionGet());
+
+        assertAcked(
+            client().execute(
+                ModifyDataStreamsAction.INSTANCE,
+                new ModifyDataStreamsAction.Request(List.of(DataStreamAction.removeBackingIndex(dataStreamName, ghostReference.getName())))
+            ).actionGet()
+        );
+        ClusterState after = internalCluster().getCurrentMasterNodeInstance(ClusterService.class).state();
+        assertThat(after.getMetadata().dataStreams().get(dataStreamName).getIndices(), hasSize(1));
+        // Data stream resolves now to one backing index.
+        // Note, that old backing index still exists and has been unhidden.
+        // The modify data stream api only fixed the data stream by removing a broken reference to a backing index.
+        indicesStatsResponse = client().admin().indices().stats(new IndicesStatsRequest()).actionGet();
+        assertThat(indicesStatsResponse.getIndices().size(), equalTo(2));
+    }
+
     private static void verifyResolvability(String dataStream, ActionRequestBuilder<?, ?> requestBuilder, boolean fail) {
         verifyResolvability(dataStream, requestBuilder, fail, 0);
     }

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

@@ -41,16 +41,16 @@ private ModifyDataStreamsAction() {
 
     public static final class Request extends AcknowledgedRequest<Request> implements IndicesRequest, ToXContentObject {
 
-        // relevant only for authorizing the request, so require every specified
-        // index to exist, expand wildcards only to open indices, prohibit
-        // wildcard expressions that resolve to zero indices, and do not attempt
-        // to resolve expressions as aliases
+        // The actual DataStreamAction don't support wildcards, so supporting it doesn't make sense.
+        // Also supporting wildcards it would prohibit this api from removing broken references to backing indices. (in case of bugs).
+        // For this case, when removing broken backing indices references that don't exist, we need to allow ignore_unavailable and
+        // allow_no_indices. Otherwise, the data stream can't be repaired.
         private static final IndicesOptions INDICES_OPTIONS = IndicesOptions.fromOptions(
+            true,
+            true,
             false,
             false,
-            true,
             false,
-            true,
             false,
             true,
             false
@@ -108,7 +108,9 @@ public ActionRequestValidationException validate() {
 
         @Override
         public String[] indices() {
-            return actions.stream().map(DataStreamAction::getDataStream).toArray(String[]::new);
+            // Return the indices instead of data streams, this api can be used to repair a broken data stream definition and
+            // in that case, exceptions can occur while resolving data streams for doing authorization or looking up index blocks.
+            return actions.stream().map(DataStreamAction::getIndex).toArray(String[]::new);
         }
 
         @Override

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

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.indices.IndicesService;
 
@@ -122,17 +123,29 @@ private static void addBackingIndex(
     }
 
     private static void removeBackingIndex(Metadata metadata, Metadata.Builder builder, String dataStreamName, String indexName) {
-        var dataStream = validateDataStream(metadata, dataStreamName);
-        var index = validateIndex(metadata, indexName);
-        var writeIndex = metadata.index(index.getWriteIndex());
-        builder.put(dataStream.getDataStream().removeBackingIndex(writeIndex.getIndex()));
+        boolean indexNotRemoved = true;
+        var dataStream = validateDataStream(metadata, dataStreamName).getDataStream();
+        for (Index backingIndex : dataStream.getIndices()) {
+            if (backingIndex.getName().equals(indexName)) {
+                builder.put(dataStream.removeBackingIndex(backingIndex));
+                indexNotRemoved = false;
+                break;
+            }
+        }
+
+        if (indexNotRemoved) {
+            throw new IllegalArgumentException("index [" + indexName + "] not found");
+        }
 
         // un-hide index
-        builder.put(
-            IndexMetadata.builder(writeIndex)
-                .settings(Settings.builder().put(writeIndex.getSettings()).put("index.hidden", "false").build())
-                .settingsVersion(writeIndex.getSettingsVersion() + 1)
-        );
+        var indexMetadata = builder.get(indexName);
+        if (indexMetadata != null) {
+            builder.put(
+                IndexMetadata.builder(indexMetadata)
+                    .settings(Settings.builder().put(indexMetadata.getSettings()).put("index.hidden", "false").build())
+                    .settingsVersion(indexMetadata.getSettingsVersion() + 1)
+            );
+        }
     }
 
     private static IndexAbstraction.DataStream validateDataStream(Metadata metadata, String dataStreamName) {

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
@@ -26,6 +27,7 @@
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.notNullValue;
 
 public class MetadataDataStreamsServiceTests extends MapperServiceTestCase {
@@ -337,6 +339,48 @@ public void testMissingIndex() {
         assertThat(e.getMessage(), equalTo("index [" + missingIndex + "] not found"));
     }
 
+    public void testRemoveBrokenBackingIndexReference() {
+        var dataStreamName = "my-logs";
+        var state = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>(dataStreamName, 2)), List.of());
+        var original = state.getMetadata().dataStreams().get(dataStreamName);
+        var broken = new DataStream(
+            original.getName(),
+            List.of(new Index(original.getIndices().get(0).getName(), "broken"), original.getIndices().get(1)),
+            original.getGeneration(),
+            original.getMetadata(),
+            original.isHidden(),
+            original.isReplicated(),
+            original.isSystem(),
+            original.isAllowCustomRouting(),
+            original.getIndexMode()
+        );
+        var brokenState = ClusterState.builder(state).metadata(Metadata.builder(state.getMetadata()).put(broken).build()).build();
+
+        var result = MetadataDataStreamsService.modifyDataStream(
+            brokenState,
+            List.of(DataStreamAction.removeBackingIndex(dataStreamName, broken.getIndices().get(0).getName())),
+            this::getMapperService
+        );
+        assertThat(result.getMetadata().dataStreams().get(dataStreamName).getIndices(), hasSize(1));
+        assertThat(result.getMetadata().dataStreams().get(dataStreamName).getIndices().get(0), equalTo(original.getIndices().get(1)));
+    }
+
+    public void testRemoveBackingIndexThatDoesntExist() {
+        var dataStreamName = "my-logs";
+        var state = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>(dataStreamName, 2)), List.of());
+
+        String indexToRemove = DataStream.getDefaultBackingIndexName(dataStreamName, 3);
+        var e = expectThrows(
+            IllegalArgumentException.class,
+            () -> MetadataDataStreamsService.modifyDataStream(
+                state,
+                List.of(DataStreamAction.removeBackingIndex(dataStreamName, indexToRemove)),
+                this::getMapperService
+            )
+        );
+        assertThat(e.getMessage(), equalTo("index [" + indexToRemove + "] not found"));
+    }
+
     private MapperService getMapperService(IndexMetadata im) {
         try {
             String mapping = im.mapping().source().toString();