feat(datastream): add Modify Data Streams API to add/remove backing indices

client/rest-high-level/src/test/java/org/opensearch/client/RestHighLevelClientTests.java

@@ -867,6 +867,7 @@ public void testApiNamingConventions() throws Exception {
             "indices.recovery",
             "indices.segments",
             "indices.stats",
+            "indices.modify_data_stream",
             "ingest.processor_grok",
             "nodes.info",
             "nodes.stats",

rest-api-spec/src/main/resources/rest-api-spec/api/indices.modify_data_stream.json

@@ -0,0 +1,33 @@
+{
+  "indices.modify_data_stream":{
+    "documentation":{
+      "url":"https://opensearch.org/docs/latest/im-plugin/data-streams/",
+      "description":"Modifies a data stream by adding or removing backing indices"
+    },
+    "stability":"experimental",
+    "url":{
+      "paths":[
+        {
+          "path":"/_data_stream/_modify",
+          "methods":[
+            "POST"
+          ]
+        }
+      ]
+    },
+    "params":{
+      "cluster_manager_timeout":{
+        "type":"time",
+        "description":"Specify timeout for connection to cluster-manager"
+      },
+      "timeout":{
+        "type":"time",
+        "description":"Explicit operation timeout"
+      }
+    },
+    "body":{
+      "description":"The data stream modifications",
+      "required":true
+    }
+  }
+}

rest-api-spec/src/main/resources/rest-api-spec/test/indices.modify_data_stream/10_basic.yml

@@ -0,0 +1,99 @@
+---
+setup:
+  - skip:
+      version: " - 3.7.99"
+      reason: "the modify data stream API was added in 3.8.0"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [my-template] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template
+        body:
+          index_patterns: [ logs-* ]
+          data_stream: {}
+          template:
+            settings:
+              number_of_shards: 1
+              number_of_replicas: 0
+
+  - do:
+      indices.create_data_stream:
+        name: logs-foo
+
+---
+teardown:
+  - do:
+      indices.delete_data_stream:
+        name: logs-foo
+        ignore: 404
+
+  - do:
+      indices.delete_index_template:
+        name: my-template
+        ignore: 404
+
+---
+"Remove and add a backing index":
+  # roll over so there are two backing indices and the first one is no longer the write index
+  - do:
+      indices.rollover:
+        alias: logs-foo
+
+  - do:
+      indices.get_data_stream:
+        name: logs-foo
+  - length: { data_streams.0.indices: 2 }
+  - match: { data_streams.0.generation: 2 }
+
+  # remove the (non-write) first backing index
+  - do:
+      indices.modify_data_stream:
+        body:
+          actions:
+            - remove_backing_index:
+                data_stream: logs-foo
+                index: .ds-logs-foo-000001
+
+  - do:
+      indices.get_data_stream:
+        name: logs-foo
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: .ds-logs-foo-000002 }
+  # generation is unchanged by modifying backing indices
+  - match: { data_streams.0.generation: 2 }
+
+  # add it back
+  - do:
+      indices.modify_data_stream:
+        body:
+          actions:
+            - add_backing_index:
+                data_stream: logs-foo
+                index: .ds-logs-foo-000001
+
+  - do:
+      indices.get_data_stream:
+        name: logs-foo
+  - length: { data_streams.0.indices: 2 }
+  - match: { data_streams.0.generation: 2 }
+
+---
+"Removing the write index is rejected":
+  - do:
+      catch: bad_request
+      indices.modify_data_stream:
+        body:
+          actions:
+            - remove_backing_index:
+                data_stream: logs-foo
+                index: .ds-logs-foo-000001
+
+---
+"Empty actions is rejected":
+  - do:
+      catch: bad_request
+      indices.modify_data_stream:
+        body:
+          actions: []

server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/DataStreamTestCase.java

@@ -63,6 +63,14 @@ public DataStreamsStatsAction.Response getDataStreamsStats(String... names) thro
         return client().execute(DataStreamsStatsAction.INSTANCE, request).get();
     }
 
+    public AcknowledgedResponse modifyDataStream(DataStreamAction... actions) throws Exception {
+        ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(List.of(actions));
+        AcknowledgedResponse response = client().admin().indices().modifyDataStream(request).get();
+        assertThat(response.isAcknowledged(), is(true));
+        performRemoteStoreTestAction();
+        return response;
+    }
+
     public RolloverResponse rolloverDataStream(String name) throws Exception {
         RolloverRequest request = new RolloverRequest(name, null);
         RolloverResponse response = client().admin().indices().rolloverIndex(request).get();

server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsIT.java

@@ -0,0 +1,101 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.datastream;
+
+import org.opensearch.ExceptionsHelper;
+import org.opensearch.cluster.metadata.DataStream;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.core.index.Index;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
+
+public class ModifyDataStreamsIT extends DataStreamTestCase {
+
+    private List<String> backingIndexNames(String dataStreamName) throws Exception {
+        DataStream dataStream = getDataStreams(dataStreamName).getDataStreams().get(0).getDataStream();
+        return dataStream.getIndices().stream().map(Index::getName).collect(Collectors.toList());
+    }
+
+    public void testRemoveThenAddBackingIndex() throws Exception {
+        createDataStreamIndexTemplate("demo-template", Collections.singletonList("logs-*"));
+        createDataStream("logs-demo");
+        rolloverDataStream("logs-demo");
+
+        // two backing indices, write index is generation 2
+        assertThat(backingIndexNames("logs-demo"), containsInAnyOrder(".ds-logs-demo-000001", ".ds-logs-demo-000002"));
+
+        // remove the (non-write) first backing index; it becomes a standalone index again
+        modifyDataStream(DataStreamAction.removeBackingIndex("logs-demo", ".ds-logs-demo-000001"));
+        assertThat(backingIndexNames("logs-demo"), containsInAnyOrder(".ds-logs-demo-000002"));
+        IndexMetadata removed = clusterAdmin().prepareState().get().getState().metadata().index(".ds-logs-demo-000001");
+        assertNotNull(removed);
+        assertFalse(IndexMetadata.INDEX_HIDDEN_SETTING.get(removed.getSettings()));
+
+        // add it back; the data stream references it again and it is hidden once more
+        modifyDataStream(DataStreamAction.addBackingIndex("logs-demo", ".ds-logs-demo-000001"));
+        assertThat(backingIndexNames("logs-demo"), containsInAnyOrder(".ds-logs-demo-000001", ".ds-logs-demo-000002"));
+        IndexMetadata added = clusterAdmin().prepareState().get().getState().metadata().index(".ds-logs-demo-000001");
+        assertTrue(IndexMetadata.INDEX_HIDDEN_SETTING.get(added.getSettings()));
+
+        // generation is unchanged by add/remove of backing indices
+        assertThat(getDataStreams("logs-demo").getDataStreams().get(0).getDataStream().getGeneration(), equalTo(2L));
+    }
+
+    public void testAddStandaloneIndexAsBackingIndex() throws Exception {
+        createDataStreamIndexTemplate("demo-template", Collections.singletonList("logs-*"));
+        createDataStream("logs-demo");
+
+        // an externally created index (arbitrary name) with a @timestamp field
+        assertAcked(client().admin().indices().prepareCreate("restored-001").setMapping("@timestamp", "type=date").get());
+
+        modifyDataStream(DataStreamAction.addBackingIndex("logs-demo", "restored-001"));
+
+        // the arbitrarily named index is now a backing index (allowed for non-write backing indices)
+        assertThat(backingIndexNames("logs-demo"), hasItem("restored-001"));
+        // and it was auto-adapted to be hidden
+        IndexMetadata added = clusterAdmin().prepareState().get().getState().metadata().index("restored-001");
+        assertTrue(IndexMetadata.INDEX_HIDDEN_SETTING.get(added.getSettings()));
+    }
+
+    public void testRemoveWriteIndexFails() throws Exception {
+        createDataStreamIndexTemplate("demo-template", Collections.singletonList("logs-*"));
+        createDataStream("logs-demo");
+
+        // .ds-logs-demo-000001 is the only (and therefore write) index
+        ExecutionException e = expectThrows(
+            ExecutionException.class,
+            () -> client().admin()
+                .indices()
+                .modifyDataStream(
+                    new ModifyDataStreamsAction.Request(List.of(DataStreamAction.removeBackingIndex("logs-demo", ".ds-logs-demo-000001")))
+                )
+                .get()
+        );
+        // the request may be handled on a remote cluster-manager node, in which case the cause is wrapped in a
+        // RemoteTransportException; unwrap it before asserting on the underlying IllegalArgumentException
+        Throwable cause = ExceptionsHelper.unwrapCause(e.getCause());
+        assertThat(cause, instanceOf(IllegalArgumentException.class));
+        assertThat(cause.getMessage(), containsString("is the write index"));
+
+        // the data stream is unchanged
+        assertThat(backingIndexNames("logs-demo"), containsInAnyOrder(".ds-logs-demo-000001"));
+        assertThat(backingIndexNames("logs-demo"), not(hasItem("nonexistent")));
+    }
+}

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -157,6 +157,7 @@
 import org.opensearch.action.admin.indices.datastream.DataStreamsStatsAction;
 import org.opensearch.action.admin.indices.datastream.DeleteDataStreamAction;
 import org.opensearch.action.admin.indices.datastream.GetDataStreamAction;
+import org.opensearch.action.admin.indices.datastream.ModifyDataStreamsAction;
 import org.opensearch.action.admin.indices.delete.DeleteIndexAction;
 import org.opensearch.action.admin.indices.delete.TransportDeleteIndexAction;
 import org.opensearch.action.admin.indices.exists.indices.IndicesExistsAction;
@@ -429,6 +430,7 @@
 import org.opensearch.rest.action.admin.indices.RestIndicesSegmentsAction;
 import org.opensearch.rest.action.admin.indices.RestIndicesShardStoresAction;
 import org.opensearch.rest.action.admin.indices.RestIndicesStatsAction;
+import org.opensearch.rest.action.admin.indices.RestModifyDataStreamsAction;
 import org.opensearch.rest.action.admin.indices.RestOpenIndexAction;
 import org.opensearch.rest.action.admin.indices.RestPauseIngestionAction;
 import org.opensearch.rest.action.admin.indices.RestPutComponentTemplateAction;
@@ -803,6 +805,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         // Data streams:
         actions.register(CreateDataStreamAction.INSTANCE, CreateDataStreamAction.TransportAction.class);
         actions.register(DeleteDataStreamAction.INSTANCE, DeleteDataStreamAction.TransportAction.class);
+        actions.register(ModifyDataStreamsAction.INSTANCE, ModifyDataStreamsAction.TransportAction.class);
         actions.register(GetDataStreamAction.INSTANCE, GetDataStreamAction.TransportAction.class);
         actions.register(ResolveIndexAction.INSTANCE, ResolveIndexAction.TransportAction.class);
         actions.register(DataStreamsStatsAction.INSTANCE, DataStreamsStatsAction.TransportAction.class);
@@ -1032,6 +1035,7 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestCreateDataStreamAction());
         registerHandler.accept(new RestDeleteDataStreamAction());
         registerHandler.accept(new RestGetDataStreamsAction());
+        registerHandler.accept(new RestModifyDataStreamsAction());
         registerHandler.accept(new RestResolveIndexAction());
         registerHandler.accept(new RestDataStreamsStatsAction());