diff --git a/client/rest-high-level/src/test/java/org/opensearch/client/RestHighLevelClientTests.java b/client/rest-high-level/src/test/java/org/opensearch/client/RestHighLevelClientTests.java index 00d1a261b344a..2f7f842dab5e8 100644 --- a/client/rest-high-level/src/test/java/org/opensearch/client/RestHighLevelClientTests.java +++ b/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", diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.modify_data_stream.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.modify_data_stream.json new file mode 100644 index 0000000000000..09c36cd66e775 --- /dev/null +++ b/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 + } + } +} diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.modify_data_stream/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.modify_data_stream/10_basic.yml new file mode 100644 index 0000000000000..cb3acbe1537fe --- /dev/null +++ b/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: [] diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/DataStreamTestCase.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/DataStreamTestCase.java index c36f8e38d2cdd..93deaa980e6fa 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/DataStreamTestCase.java +++ b/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(); diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsIT.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsIT.java new file mode 100644 index 0000000000000..578335771ee5f --- /dev/null +++ b/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 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"))); + } +} diff --git a/server/src/main/java/org/opensearch/action/ActionModule.java b/server/src/main/java/org/opensearch/action/ActionModule.java index b70be5bd416a6..53749ff7d09ff 100644 --- a/server/src/main/java/org/opensearch/action/ActionModule.java +++ b/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 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 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()); diff --git a/server/src/main/java/org/opensearch/action/admin/indices/datastream/DataStreamAction.java b/server/src/main/java/org/opensearch/action/admin/indices/datastream/DataStreamAction.java new file mode 100644 index 0000000000000..02d99c6575e79 --- /dev/null +++ b/server/src/main/java/org/opensearch/action/admin/indices/datastream/DataStreamAction.java @@ -0,0 +1,221 @@ +/* + * 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. + */ + +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.action.admin.indices.datastream; + +import org.opensearch.common.annotation.PublicApi; +import org.opensearch.core.ParseField; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.common.io.stream.Writeable; +import org.opensearch.core.xcontent.ConstructingObjectParser; +import org.opensearch.core.xcontent.ObjectParser; +import org.opensearch.core.xcontent.ToXContentObject; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.core.xcontent.XContentParser; + +import java.io.IOException; +import java.util.Objects; +import java.util.function.Supplier; + +import static org.opensearch.core.xcontent.ConstructingObjectParser.optionalConstructorArg; + +/** + * Operations on a data stream's backing indices that are submitted as part of a + * {@link ModifyDataStreamsAction} request. A single {@code DataStreamAction} either adds or removes a single + * backing index from a single data stream. + * + * @opensearch.api + */ +@PublicApi(since = "3.8.0") +public class DataStreamAction implements Writeable, ToXContentObject { + + /** + * The type of action to perform on a data stream's backing indices. + * + * @opensearch.api + */ + @PublicApi(since = "3.8.0") + public enum Type { + ADD_BACKING_INDEX((byte) 0, "add_backing_index"), + REMOVE_BACKING_INDEX((byte) 1, "remove_backing_index"); + + private final byte value; + private final String fieldName; + + Type(byte value, String fieldName) { + this.value = value; + this.fieldName = fieldName; + } + + public byte value() { + return value; + } + + public String fieldName() { + return fieldName; + } + + public static Type fromValue(byte value) { + switch (value) { + case 0: + return ADD_BACKING_INDEX; + case 1: + return REMOVE_BACKING_INDEX; + default: + throw new IllegalArgumentException("no data stream action type for [" + value + "]"); + } + } + } + + private final Type type; + private String dataStream; + private String index; + + public static DataStreamAction addBackingIndex(String dataStream, String index) { + return new DataStreamAction(Type.ADD_BACKING_INDEX, dataStream, index); + } + + public static DataStreamAction removeBackingIndex(String dataStream, String index) { + return new DataStreamAction(Type.REMOVE_BACKING_INDEX, dataStream, index); + } + + private DataStreamAction(Type type) { + this.type = type; + } + + private DataStreamAction(Type type, String dataStream, String index) { + this.type = type; + this.dataStream = dataStream; + this.index = index; + } + + public Type getType() { + return type; + } + + public String getDataStream() { + return dataStream; + } + + public String getIndex() { + return index; + } + + public DataStreamAction(StreamInput in) throws IOException { + this.type = Type.fromValue(in.readByte()); + this.dataStream = in.readString(); + this.index = in.readString(); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeByte(type.value()); + out.writeString(dataStream); + out.writeString(index); + } + + private static final ParseField DATA_STREAM_FIELD = new ParseField("data_stream"); + private static final ParseField INDEX_FIELD = new ParseField("index"); + + private static final ObjectParser ADD_BACKING_INDEX_PARSER = parser( + Type.ADD_BACKING_INDEX.fieldName(), + () -> new DataStreamAction(Type.ADD_BACKING_INDEX) + ); + private static final ObjectParser REMOVE_BACKING_INDEX_PARSER = parser( + Type.REMOVE_BACKING_INDEX.fieldName(), + () -> new DataStreamAction(Type.REMOVE_BACKING_INDEX) + ); + + private static ObjectParser parser(String name, Supplier supplier) { + ObjectParser parser = new ObjectParser<>(name, supplier); + parser.declareString((action, dataStream) -> action.dataStream = dataStream, DATA_STREAM_FIELD); + parser.declareString((action, index) -> action.index = index, INDEX_FIELD); + parser.declareRequiredFieldSet(DATA_STREAM_FIELD.getPreferredName()); + parser.declareRequiredFieldSet(INDEX_FIELD.getPreferredName()); + return parser; + } + + public static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( + "data_stream_action", + a -> { + // only a single action is allowed per entry + DataStreamAction action = null; + int actionCount = 0; + for (Object o : a) { + if (o != null) { + action = (DataStreamAction) o; + actionCount++; + } + } + if (actionCount == 0) { + throw new IllegalArgumentException("no data stream operation declared on operation entry"); + } + if (actionCount > 1) { + throw new IllegalArgumentException("too many data stream operations declared on operation entry"); + } + return action; + } + ); + + static { + PARSER.declareObject(optionalConstructorArg(), ADD_BACKING_INDEX_PARSER, new ParseField(Type.ADD_BACKING_INDEX.fieldName())); + PARSER.declareObject(optionalConstructorArg(), REMOVE_BACKING_INDEX_PARSER, new ParseField(Type.REMOVE_BACKING_INDEX.fieldName())); + } + + public static DataStreamAction fromXContent(XContentParser parser) throws IOException { + return PARSER.parse(parser, null); + } + + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + builder.startObject(); + builder.startObject(type.fieldName()); + builder.field(DATA_STREAM_FIELD.getPreferredName(), dataStream); + builder.field(INDEX_FIELD.getPreferredName(), index); + builder.endObject(); + builder.endObject(); + return builder; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + DataStreamAction that = (DataStreamAction) o; + return type == that.type && Objects.equals(dataStream, that.dataStream) && Objects.equals(index, that.index); + } + + @Override + public int hashCode() { + return Objects.hash(type, dataStream, index); + } +} diff --git a/server/src/main/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsAction.java b/server/src/main/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsAction.java new file mode 100644 index 0000000000000..5dd36bcfb4bc2 --- /dev/null +++ b/server/src/main/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsAction.java @@ -0,0 +1,459 @@ +/* + * 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. + */ + +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.action.admin.indices.datastream; + +import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.action.ActionType; +import org.opensearch.action.IndicesRequest; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.IndicesOptions; +import org.opensearch.action.support.clustermanager.AcknowledgedRequest; +import org.opensearch.action.support.clustermanager.AcknowledgedResponse; +import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.ClusterStateUpdateTask; +import org.opensearch.cluster.block.ClusterBlockException; +import org.opensearch.cluster.block.ClusterBlockLevel; +import org.opensearch.cluster.metadata.ComposableIndexTemplate; +import org.opensearch.cluster.metadata.DataStream; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; +import org.opensearch.cluster.metadata.MappingMetadata; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.metadata.MetadataCreateDataStreamService; +import org.opensearch.cluster.service.ClusterManagerTaskThrottler; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.CheckedFunction; +import org.opensearch.common.Priority; +import org.opensearch.common.annotation.PublicApi; +import org.opensearch.common.inject.Inject; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.io.IOUtils; +import org.opensearch.core.ParseField; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.xcontent.ConstructingObjectParser; +import org.opensearch.core.xcontent.ToXContentObject; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.index.mapper.MapperService; +import org.opensearch.indices.IndicesService; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.opensearch.action.ValidateActions.addValidationError; +import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_HIDDEN_SETTING; +import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_INDEX_HIDDEN; +import static org.opensearch.cluster.service.ClusterManagerTask.MODIFY_DATA_STREAM; +import static org.opensearch.core.xcontent.ConstructingObjectParser.constructorArg; + +/** + * Transport action for modifying the backing indices of one or more data streams. The supported actions are adding and + * removing backing indices. Multiple actions are applied atomically in a single cluster state update. + * + * @opensearch.api + */ +@PublicApi(since = "3.8.0") +public class ModifyDataStreamsAction extends ActionType { + + public static final ModifyDataStreamsAction INSTANCE = new ModifyDataStreamsAction(); + public static final String NAME = "indices:admin/data_stream/modify"; + + private ModifyDataStreamsAction() { + super(NAME, AcknowledgedResponse::new); + } + + /** + * Request for modifying one or more data streams. + * + * @opensearch.api + */ + @PublicApi(since = "3.8.0") + public static class Request extends AcknowledgedRequest implements IndicesRequest, ToXContentObject { + + private final List actions; + + private static final IndicesOptions INDICES_OPTIONS = IndicesOptions.fromOptions( + false, + true, + false, + true, + false, + false, + true, + false + ); + + @SuppressWarnings("unchecked") + private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( + "data_stream_actions", + args -> new Request((List) args[0]) + ); + static { + PARSER.declareObjectArray(constructorArg(), (p, c) -> DataStreamAction.fromXContent(p), new ParseField("actions")); + } + + public static Request fromXContent(XContentParser parser) throws IOException { + return PARSER.parse(parser, null); + } + + public Request(List actions) { + this.actions = Collections.unmodifiableList(actions); + } + + @Override + public ActionRequestValidationException validate() { + if (actions.isEmpty()) { + return addValidationError("must specify at least one data stream modification action", null); + } + return null; + } + + public Request(StreamInput in) throws IOException { + super(in); + this.actions = Collections.unmodifiableList(in.readList(DataStreamAction::new)); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + super.writeTo(out); + out.writeList(actions); + } + + public List getActions() { + return actions; + } + + @Override + public String[] indices() { + // resolve to the backing indices being added/removed (not the data streams). This allows the API to repair a + // broken data stream definition by referencing indices that may no longer be members of any data stream, and + // a DataStreamAction does not support wildcards. + return actions.stream().map(DataStreamAction::getIndex).toArray(String[]::new); + } + + @Override + public IndicesOptions indicesOptions() { + return INDICES_OPTIONS; + } + + @Override + public boolean includeDataStreams() { + return true; + } + + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + builder.startObject(); + builder.startArray("actions"); + for (DataStreamAction action : actions) { + action.toXContent(builder, params); + } + builder.endArray(); + builder.endObject(); + return builder; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Request request = (Request) o; + return Objects.equals(actions, request.actions); + } + + @Override + public int hashCode() { + return Objects.hash(actions); + } + } + + /** + * Transport action for modifying data streams. + * + * @opensearch.internal + */ + public static class TransportAction extends TransportClusterManagerNodeAction { + + private final IndicesService indicesService; + private final ClusterManagerTaskThrottler.ThrottlingKey modifyDataStreamTaskKey; + + @Inject + public TransportAction( + TransportService transportService, + ClusterService clusterService, + ThreadPool threadPool, + ActionFilters actionFilters, + IndexNameExpressionResolver indexNameExpressionResolver, + IndicesService indicesService + ) { + super(NAME, transportService, clusterService, threadPool, actionFilters, Request::new, indexNameExpressionResolver); + this.indicesService = indicesService; + // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction. + modifyDataStreamTaskKey = clusterService.registerClusterManagerTask(MODIFY_DATA_STREAM, true); + } + + @Override + protected String executor() { + return ThreadPool.Names.SAME; + } + + @Override + protected AcknowledgedResponse read(StreamInput in) throws IOException { + return new AcknowledgedResponse(in); + } + + @Override + protected void clusterManagerOperation(Request request, ClusterState state, ActionListener listener) { + clusterService.submitStateUpdateTask("update-backing-indices", new ClusterStateUpdateTask(Priority.URGENT) { + + @Override + public TimeValue timeout() { + return request.clusterManagerNodeTimeout(); + } + + @Override + public void onFailure(String source, Exception e) { + listener.onFailure(e); + } + + @Override + public ClusterManagerTaskThrottler.ThrottlingKey getClusterManagerThrottlingKey() { + return modifyDataStreamTaskKey; + } + + @Override + public ClusterState execute(ClusterState currentState) throws IOException { + return modifyDataStream(currentState, request.getActions(), indicesService::createIndexMapperService); + } + + @Override + public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) { + listener.onResponse(new AcknowledgedResponse(true)); + } + }); + } + + @Override + protected ClusterBlockException checkBlock(Request request, ClusterState state) { + return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE); + } + } + + /** + * Applies the given data stream modification actions to the cluster state. All actions are applied to a single + * {@link Metadata.Builder} so that the request is processed atomically: if any action fails, the whole request + * fails and no partial changes are published. + * + * @param mapperServiceFactory builds a throwaway {@link MapperService} for a candidate index so that the + * {@code _data_stream_timestamp} meta field can be merged into its mapping when the + * index is added as a backing index. + */ + static ClusterState modifyDataStream( + ClusterState currentState, + Iterable actions, + CheckedFunction mapperServiceFactory + ) throws IOException { + Metadata.Builder updatedMetadata = Metadata.builder(currentState.metadata()); + for (DataStreamAction action : actions) { + switch (action.getType()) { + case ADD_BACKING_INDEX: + addBackingIndex(updatedMetadata, action.getDataStream(), action.getIndex(), mapperServiceFactory); + break; + case REMOVE_BACKING_INDEX: + removeBackingIndex(updatedMetadata, action.getDataStream(), action.getIndex()); + break; + default: + throw new IllegalStateException("unsupported data stream action type [" + action.getType() + "]"); + } + } + return ClusterState.builder(currentState).metadata(updatedMetadata).build(); + } + + private static void addBackingIndex( + Metadata.Builder builder, + String dataStreamName, + String indexName, + CheckedFunction mapperServiceFactory + ) throws IOException { + final DataStream dataStream = validateDataStream(builder, dataStreamName); + + // an index whose name matches this data stream's backing-index naming pattern with a generation counter greater than + // the stream's current generation cannot be added: it would collide with a future backing index produced by a rollover + // and make the metadata invalid (Metadata.Builder.validateDataStreams would later throw IllegalStateException -> HTTP 500) + final String backingIndexPrefix = DataStream.BACKING_INDEX_PREFIX + dataStreamName + "-"; + if (indexName.startsWith(backingIndexPrefix) + && Metadata.NUMBER_PATTERN.matcher(indexName.substring(backingIndexPrefix.length())).matches() + && IndexMetadata.parseIndexNameCounter(indexName) > dataStream.getGeneration()) { + throw new IllegalArgumentException( + "cannot add index [" + + indexName + + "] to data stream [" + + dataStreamName + + "] because its name conflicts with a future backing index of the data stream " + + "(current generation [" + + dataStream.getGeneration() + + "])" + ); + } + + final IndexMetadata index = validateIndex(builder, indexName); + + // an index that is already a backing index of this data stream cannot be added again + if (dataStream.getIndices().contains(index.getIndex())) { + throw new IllegalArgumentException( + "index [" + indexName + "] is already a backing index of data stream [" + dataStreamName + "] and cannot be added" + ); + } + // an index that already belongs to a different data stream cannot be added, as that would leave the index + // referenced by two data streams and corrupt the cluster state's indices lookup + for (DataStream other : builder.dataStreams().values()) { + if (other.getName().equals(dataStreamName) == false && other.getIndices().contains(index.getIndex())) { + throw new IllegalArgumentException( + "index [" + + indexName + + "] is already a backing index of data stream [" + + other.getName() + + "] and cannot be added to data stream [" + + dataStreamName + + "]" + ); + } + } + // an index with aliases cannot become a backing index, as aliases and data streams cannot point at the same index + if (index.getAliases().isEmpty() == false) { + throw new IllegalArgumentException( + "cannot add index [" + + indexName + + "] to data stream [" + + dataStreamName + + "] because it has aliases " + + index.getAliases().keySet() + ); + } + + // auto-adapt the index so it is a valid backing index: hide it and merge in the _data_stream_timestamp meta field, + // then validate the timestamp mapping (mirrors how a backing index is created from a data stream template) + final IndexMetadata updatedIndex = prepareBackingIndex(index, dataStream.getTimeStampField().getName(), mapperServiceFactory); + builder.put(updatedIndex, true); + builder.put(dataStream.addBackingIndex(updatedIndex.getIndex())); + } + + /** + * Returns an updated copy of {@code index} that is a valid backing index for a data stream: it is hidden and its + * mapping contains an enabled {@code _data_stream_timestamp} meta field for the given timestamp field. + */ + private static IndexMetadata prepareBackingIndex( + IndexMetadata index, + String timestampFieldName, + CheckedFunction mapperServiceFactory + ) throws IOException { + final IndexMetadata hiddenIndex = IndexMetadata.builder(index) + .settings(Settings.builder().put(index.getSettings()).put(SETTING_INDEX_HIDDEN, true)) + .build(); + + final MapperService mapperService = mapperServiceFactory.apply(hiddenIndex); + try { + if (hiddenIndex.mapping() != null) { + mapperService.merge(hiddenIndex, MapperService.MergeReason.MAPPING_RECOVERY); + } + mapperService.merge( + MapperService.SINGLE_MAPPING_NAME, + new ComposableIndexTemplate.DataStreamTemplate(new DataStream.TimestampField(timestampFieldName)) + .getDataStreamMappingSnippet(), + MapperService.MergeReason.MAPPING_UPDATE + ); + MetadataCreateDataStreamService.validateTimestampFieldMapping(mapperService); + + final IndexMetadata.Builder updated = IndexMetadata.builder(hiddenIndex); + + // Only bump the settings version if hiding the index actually changed its settings; re-adding an index that was + // already hidden leaves the settings unchanged and the version must not move (IndexService.updateMetadata asserts this). + if (index.getSettings().equals(hiddenIndex.getSettings()) == false) { + updated.settingsVersion(1 + hiddenIndex.getSettingsVersion()); + } + + // Likewise, only update the mapping (and bump the mapping version) if merging the _data_stream_timestamp meta field + // actually changed the mapping; re-adding an index that was already a backing index leaves the mapping unchanged + // (MapperService.updateMapping asserts that a mapping version bump implies a mapping change). + final MappingMetadata mergedMapping = new MappingMetadata(mapperService.documentMapper().mappingSource()); + final MappingMetadata currentMapping = hiddenIndex.mapping(); + if (currentMapping == null || currentMapping.source().equals(mergedMapping.source()) == false) { + updated.putMapping(mergedMapping).mappingVersion(1 + hiddenIndex.getMappingVersion()); + } + + return updated.build(); + } finally { + IOUtils.close(mapperService); + } + } + + private static void removeBackingIndex(Metadata.Builder builder, String dataStreamName, String indexName) { + final DataStream dataStream = validateDataStream(builder, dataStreamName); + final IndexMetadata index = validateIndex(builder, indexName); + + builder.put(dataStream.removeBackingIndex(index.getIndex())); + // un-hide the index that was removed from the data stream so that it becomes a regular standalone index again + if (INDEX_HIDDEN_SETTING.get(index.getSettings())) { + builder.put( + IndexMetadata.builder(index) + .settings(Settings.builder().put(index.getSettings()).put(SETTING_INDEX_HIDDEN, false)) + .settingsVersion(1 + index.getSettingsVersion()) + .build(), + true + ); + } + } + + private static DataStream validateDataStream(Metadata.Builder builder, String dataStreamName) { + DataStream dataStream = builder.dataStream(dataStreamName); + if (dataStream == null) { + throw new IllegalArgumentException("data stream [" + dataStreamName + "] not found"); + } + return dataStream; + } + + private static IndexMetadata validateIndex(Metadata.Builder builder, String indexName) { + IndexMetadata index = builder.get(indexName); + if (index == null) { + throw new IllegalArgumentException("index [" + indexName + "] not found"); + } + return index; + } +} diff --git a/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java b/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java index 54df245b1b835..070f34cbeb42d 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java @@ -134,14 +134,47 @@ public DataStream rollover(Index newWriteIndex) { return new DataStream(name, timeStampField, backingIndices, generation + 1); } + /** + * Adds the specified index as a backing index and returns a new {@code DataStream} instance with the new combination + * of backing indices. The index is added at the start of the backing indices list so that the existing write index + * (which is always the last backing index) remains the write index. + * + * @param index the index to add to the data stream as a backing index + * @return new {@code DataStream} instance with the added backing index + */ + public DataStream addBackingIndex(Index index) { + // validate that index is not part of another data stream is the responsibility of the caller + List backingIndices = new ArrayList<>(indices); + backingIndices.add(0, index); + assert backingIndices.size() == indices.size() + 1; + return new DataStream(name, timeStampField, backingIndices, generation); + } + /** * Removes the specified backing index and returns a new {@code DataStream} instance with - * the remaining backing indices. + * the remaining backing indices. An {@code IllegalArgumentException} is thrown if the index to be removed + * is not a backing index for this data stream or if it is the {@code DataStream}'s write index. * * @param index the backing index to remove * @return new {@code DataStream} instance with the remaining backing indices */ public DataStream removeBackingIndex(Index index) { + int backingIndexPosition = indices.indexOf(index); + if (backingIndexPosition == -1) { + throw new IllegalArgumentException( + String.format(Locale.ROOT, "index [%s] is not part of data stream [%s]", index.getName(), name) + ); + } + if (indices.size() == (backingIndexPosition + 1)) { + throw new IllegalArgumentException( + String.format( + Locale.ROOT, + "cannot remove backing index [%s] of data stream [%s] because it is the write index", + index.getName(), + name + ) + ); + } List backingIndices = new ArrayList<>(indices); backingIndices.remove(index); assert backingIndices.size() == indices.size() - 1; diff --git a/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java b/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java index 03b343e94b5ab..62209834ec427 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java @@ -1375,7 +1375,13 @@ public Builder removeIndexTemplate(String name) { } public DataStream dataStream(String dataStreamName) { - return ((DataStreamMetadata) customs.get(DataStreamMetadata.TYPE)).dataStreams().get(dataStreamName); + return dataStreams().get(dataStreamName); + } + + public Map dataStreams() { + return Optional.ofNullable((DataStreamMetadata) this.customs.get(DataStreamMetadata.TYPE)) + .map(DataStreamMetadata::dataStreams) + .orElse(Collections.emptyMap()); } public Builder dataStreams(Map dataStreams) { diff --git a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTask.java b/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTask.java index eb5b6d0812ff2..c4de9705ce03b 100644 --- a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTask.java +++ b/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTask.java @@ -24,6 +24,7 @@ public enum ClusterManagerTask { DELETE_DANGLING_INDEX("delete-dangling-index", 50), CREATE_DATA_STREAM("create-data-stream", 50), REMOVE_DATA_STREAM("remove-data-stream", 50), + MODIFY_DATA_STREAM("modify-data-stream", 50), CREATE_INDEX_TEMPLATE("create-index-template", 50), REMOVE_INDEX_TEMPLATE("remove-index-template", 50), CREATE_COMPONENT_TEMPLATE("create-component-template", 50), diff --git a/server/src/main/java/org/opensearch/index/mapper/DataStreamFieldMapper.java b/server/src/main/java/org/opensearch/index/mapper/DataStreamFieldMapper.java index d1da942c2a7fd..f2eb5aeecb439 100644 --- a/server/src/main/java/org/opensearch/index/mapper/DataStreamFieldMapper.java +++ b/server/src/main/java/org/opensearch/index/mapper/DataStreamFieldMapper.java @@ -21,6 +21,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; /** * Field mapper for a datastream field @@ -48,7 +49,10 @@ public static final class Defaults { * @opensearch.internal */ public static final class Builder extends MetadataFieldMapper.Builder { - final Parameter enabledParam = Parameter.boolParam("enabled", false, mapper -> toType(mapper).enabled, Defaults.ENABLED); + // The _data_stream_timestamp meta field can be enabled on a mapping update (e.g. when an existing index is + // adapted into a data stream backing index) but, once enabled, it cannot be disabled again. + final Parameter enabledParam = Parameter.boolParam("enabled", false, mapper -> toType(mapper).enabled, Defaults.ENABLED) + .setMergeValidator((previous, current) -> Objects.equals(previous, current) || (previous == false && current)); final Parameter timestampFieldParam = new Parameter<>( "timestamp_field", diff --git a/server/src/main/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsAction.java b/server/src/main/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsAction.java new file mode 100644 index 0000000000000..8cfeb528aafbb --- /dev/null +++ b/server/src/main/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsAction.java @@ -0,0 +1,76 @@ +/* + * 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. + */ + +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +/* + * Modifications Copyright OpenSearch Contributors. See + * GitHub history for details. + */ + +package org.opensearch.rest.action.admin.indices; + +import org.opensearch.action.admin.indices.datastream.ModifyDataStreamsAction; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; +import org.opensearch.transport.client.node.NodeClient; + +import java.io.IOException; +import java.util.List; + +import static java.util.Collections.singletonList; +import static org.opensearch.rest.RestRequest.Method.POST; + +/** + * Transport action to modify the backing indices of a data stream + * + * @opensearch.api + */ +public class RestModifyDataStreamsAction extends BaseRestHandler { + + @Override + public String getName() { + return "modify_data_stream_action"; + } + + @Override + public List routes() { + return singletonList(new Route(POST, "/_data_stream/_modify")); + } + + @Override + protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException { + final ModifyDataStreamsAction.Request modifyDataStreamRequest; + try (XContentParser parser = request.contentParser()) { + modifyDataStreamRequest = ModifyDataStreamsAction.Request.fromXContent(parser); + } + modifyDataStreamRequest.clusterManagerNodeTimeout( + request.paramAsTime("cluster_manager_timeout", modifyDataStreamRequest.clusterManagerNodeTimeout()) + ); + modifyDataStreamRequest.timeout(request.paramAsTime("timeout", modifyDataStreamRequest.timeout())); + return channel -> client.admin().indices().modifyDataStream(modifyDataStreamRequest, new RestToXContentListener<>(channel)); + } +} diff --git a/server/src/main/java/org/opensearch/transport/client/IndicesAdminClient.java b/server/src/main/java/org/opensearch/transport/client/IndicesAdminClient.java index 46b270e7eec7c..96b489bf78fa6 100644 --- a/server/src/main/java/org/opensearch/transport/client/IndicesAdminClient.java +++ b/server/src/main/java/org/opensearch/transport/client/IndicesAdminClient.java @@ -51,6 +51,7 @@ import org.opensearch.action.admin.indices.datastream.CreateDataStreamAction; 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.DeleteIndexRequest; import org.opensearch.action.admin.indices.delete.DeleteIndexRequestBuilder; import org.opensearch.action.admin.indices.exists.indices.IndicesExistsRequest; @@ -832,6 +833,16 @@ public interface IndicesAdminClient extends OpenSearchClient { */ ActionFuture deleteDataStream(DeleteDataStreamAction.Request request); + /** + * Modify the backing indices of one or more data streams + */ + void modifyDataStream(ModifyDataStreamsAction.Request request, ActionListener listener); + + /** + * Modify the backing indices of one or more data streams + */ + ActionFuture modifyDataStream(ModifyDataStreamsAction.Request request); + /** * Get data streams */ @@ -1141,6 +1152,13 @@ default CompletionStage deleteDataStreamAsync(DeleteDataSt return future; } + /** Modify data stream - CompletionStage version */ + default CompletionStage modifyDataStreamAsync(ModifyDataStreamsAction.Request request) { + CompletableFuture future = new CompletableFuture<>(); + modifyDataStream(request, ActionListener.wrap(future::complete, future::completeExceptionally)); + return future; + } + /** Get data streams - CompletionStage version */ default CompletionStage getDataStreamsAsync(GetDataStreamAction.Request request) { CompletableFuture future = new CompletableFuture<>(); diff --git a/server/src/main/java/org/opensearch/transport/client/support/AbstractClient.java b/server/src/main/java/org/opensearch/transport/client/support/AbstractClient.java index 498d6483c2e3d..09a1f8c0114f3 100644 --- a/server/src/main/java/org/opensearch/transport/client/support/AbstractClient.java +++ b/server/src/main/java/org/opensearch/transport/client/support/AbstractClient.java @@ -217,6 +217,7 @@ import org.opensearch.action.admin.indices.datastream.CreateDataStreamAction; 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.DeleteIndexRequest; import org.opensearch.action.admin.indices.delete.DeleteIndexRequestBuilder; @@ -2124,6 +2125,16 @@ public ActionFuture deleteDataStream(DeleteDataStreamActio return execute(DeleteDataStreamAction.INSTANCE, request); } + @Override + public void modifyDataStream(ModifyDataStreamsAction.Request request, ActionListener listener) { + execute(ModifyDataStreamsAction.INSTANCE, request, listener); + } + + @Override + public ActionFuture modifyDataStream(ModifyDataStreamsAction.Request request) { + return execute(ModifyDataStreamsAction.INSTANCE, request); + } + @Override public void getDataStreams(GetDataStreamAction.Request request, ActionListener listener) { execute(GetDataStreamAction.INSTANCE, request, listener); diff --git a/server/src/test/java/org/opensearch/action/admin/indices/datastream/DataStreamActionTests.java b/server/src/test/java/org/opensearch/action/admin/indices/datastream/DataStreamActionTests.java new file mode 100644 index 0000000000000..a9f8ebd9ed364 --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/DataStreamActionTests.java @@ -0,0 +1,84 @@ +/* + * 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.common.xcontent.XContentFactory; +import org.opensearch.common.xcontent.json.JsonXContent; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.test.OpenSearchTestCase; + +import java.io.IOException; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; + +public class DataStreamActionTests extends OpenSearchTestCase { + + public void testAddBackingIndexXContentRoundTrip() throws IOException { + assertXContentRoundTrip(DataStreamAction.addBackingIndex("my-data-stream", "my-index")); + } + + public void testRemoveBackingIndexXContentRoundTrip() throws IOException { + assertXContentRoundTrip(DataStreamAction.removeBackingIndex("my-data-stream", "my-index")); + } + + private void assertXContentRoundTrip(DataStreamAction action) throws IOException { + XContentBuilder builder = XContentFactory.jsonBuilder(); + action.toXContent(builder, ToXContent.EMPTY_PARAMS); + try (XContentParser parser = createParser(JsonXContent.jsonXContent, builder.toString())) { + DataStreamAction parsed = DataStreamAction.fromXContent(parser); + assertThat(parsed, equalTo(action)); + } + } + + public void testRejectsEntryWithoutAction() throws IOException { + String json = "{}"; + try (XContentParser parser = createParser(JsonXContent.jsonXContent, json)) { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(causeChainMessages(e), containsString("no data stream operation declared")); + } + } + + public void testRejectsEntryWithMultipleActions() throws IOException { + String json = "{ \"add_backing_index\": { \"data_stream\": \"ds\", \"index\": \"i1\" }, " + + "\"remove_backing_index\": { \"data_stream\": \"ds\", \"index\": \"i2\" } }"; + try (XContentParser parser = createParser(JsonXContent.jsonXContent, json)) { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(causeChainMessages(e), containsString("too many data stream operations declared")); + } + } + + public void testRejectsActionWithoutIndex() throws IOException { + String json = "{ \"add_backing_index\": { \"data_stream\": \"ds\" } }"; + try (XContentParser parser = createParser(JsonXContent.jsonXContent, json)) { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(causeChainMessages(e), containsString("index")); + } + } + + public void testRejectsActionWithoutDataStream() throws IOException { + String json = "{ \"remove_backing_index\": { \"index\": \"i1\" } }"; + try (XContentParser parser = createParser(JsonXContent.jsonXContent, json)) { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(causeChainMessages(e), containsString("data_stream")); + } + } + + // ConstructingObjectParser wraps validation errors thrown from its builder lambda, so the custom message + // surfaces somewhere in the cause chain rather than on the top-level exception. + private static String causeChainMessages(Throwable t) { + StringBuilder sb = new StringBuilder(); + for (Throwable c = t; c != null; c = c.getCause()) { + sb.append(c.getMessage()).append('\n'); + } + return sb.toString(); + } +} diff --git a/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsActionTests.java b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsActionTests.java new file mode 100644 index 0000000000000..4af1a198e38fb --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsActionTests.java @@ -0,0 +1,304 @@ +/* + * 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.Version; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.AliasMetadata; +import org.opensearch.cluster.metadata.DataStream; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.common.CheckedFunction; +import org.opensearch.common.collect.Tuple; +import org.opensearch.common.settings.Settings; +import org.opensearch.index.MapperTestUtils; +import org.opensearch.index.mapper.MapperService; + +import java.io.IOException; +import java.util.List; + +import static org.opensearch.cluster.DataStreamTestHelper.getClusterStateWithDataStreams; +import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_VERSION_CREATED; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItem; +import static org.hamcrest.Matchers.not; + +public class ModifyDataStreamsActionTests extends org.opensearch.test.OpenSearchTestCase { + + private static final String DATA_STREAM = "my-data-stream"; + + /** + * Builds a {@link MapperService} for an arbitrary index so the add-backing-index path can merge the + * {@code _data_stream_timestamp} meta field into it without a running node. + */ + private CheckedFunction mapperServiceFactory() { + return indexMetadata -> MapperTestUtils.newMapperService( + xContentRegistry(), + createTempDir(), + indexMetadata.getSettings(), + indexMetadata.getIndex().getName() + ); + } + + private ClusterState modify(ClusterState cs, List actions) { + try { + return ModifyDataStreamsAction.modifyDataStream(cs, actions, mapperServiceFactory()); + } catch (IOException e) { + throw new AssertionError(e); + } + } + + public void testAddBackingIndex() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + + ClusterState newState = modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"))); + + DataStream ds = newState.metadata().dataStreams().get(DATA_STREAM); + assertThat(ds.getIndices().size(), equalTo(3)); + // new backing index is added to the front so the existing write index stays last + assertThat(ds.getIndices().get(0).getName(), equalTo("standalone-index")); + assertThat( + ds.getIndices().get(ds.getIndices().size() - 1).getName(), + equalTo(DataStream.getDefaultBackingIndexName(DATA_STREAM, 2)) + ); + // generation is unchanged when adding a backing index + assertThat(ds.getGeneration(), equalTo(2L)); + // the added index is now hidden + assertTrue(IndexMetadata.INDEX_HIDDEN_SETTING.get(newState.metadata().index("standalone-index").getSettings())); + } + + public void testAddBackingIndexMergesDataStreamTimestampMetaField() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + + ClusterState newState = modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"))); + + // the added index has had the _data_stream_timestamp meta field merged into its mapping (auto-adapt) + IndexMetadata added = newState.metadata().index("standalone-index"); + assertNotNull(added.mapping()); + assertThat(added.mapping().source().string(), containsString("_data_stream_timestamp")); + } + + public void testAddBackingIndexWithExistingMappingMergesTimestampMetaField() throws IOException { + ClusterState base = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of()); + // a standalone index that already has a mapping with a plain @timestamp field (but no _data_stream_timestamp + // meta field), mirroring an externally created index being adopted as a backing index + IndexMetadata mapped = IndexMetadata.builder("mapped-index") + .settings(Settings.builder().put(SETTING_VERSION_CREATED, Version.CURRENT)) + .numberOfShards(1) + .numberOfReplicas(0) + .putMapping("{\"properties\":{\"@timestamp\":{\"type\":\"date\"}}}") + .build(); + ClusterState cs = ClusterState.builder(new ClusterName("_name")) + .metadata(Metadata.builder(base.metadata()).put(mapped, false)) + .build(); + + ClusterState newState = modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "mapped-index"))); + + DataStream ds = newState.metadata().dataStreams().get(DATA_STREAM); + assertThat(ds.getIndices().stream().map(i -> i.getName()).toList(), hasItem("mapped-index")); + IndexMetadata added = newState.metadata().index("mapped-index"); + assertThat(added.mapping().source().string(), containsString("_data_stream_timestamp")); + assertTrue(IndexMetadata.INDEX_HIDDEN_SETTING.get(added.getSettings())); + } + + public void testReAddBackingIndexOnlyBumpsSettingsVersion() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + + // add the standalone index: it is adapted (hidden + _data_stream_timestamp mapping merged in) + ClusterState afterAdd = modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"))); + // remove it again: it stays in the cluster state with its (now canonical) mapping but is un-hidden + ClusterState afterRemove = modify(afterAdd, List.of(DataStreamAction.removeBackingIndex(DATA_STREAM, "standalone-index"))); + IndexMetadata adapted = afterRemove.metadata().index("standalone-index"); + + // re-add it: its mapping already carries the _data_stream_timestamp meta field, so re-adapting must not change + // the mapping, but hiding it again does change the settings (this is the testRemoveThenAddBackingIndex IT case) + ClusterState afterReAdd = modify(afterRemove, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"))); + + IndexMetadata reAdded = afterReAdd.metadata().index("standalone-index"); + assertThat( + afterReAdd.metadata().dataStreams().get(DATA_STREAM).getIndices().stream().map(i -> i.getName()).toList(), + hasItem("standalone-index") + ); + assertTrue(IndexMetadata.INDEX_HIDDEN_SETTING.get(reAdded.getSettings())); + // mapping is unchanged on re-add, so the mapping version must NOT move (node asserts bump implies content change) + assertThat(reAdded.getMappingVersion(), equalTo(adapted.getMappingVersion())); + // but hiding the index changed its settings, so the settings version MUST move + assertThat(reAdded.getSettingsVersion(), equalTo(adapted.getSettingsVersion() + 1)); + } + + public void testAddBackingIndexBumpsSettingsVersion() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + long beforeVersion = cs.metadata().index("standalone-index").getSettingsVersion(); + + ClusterState newState = modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"))); + + // hidden flips false -> true, so the settings version must be bumped (IndexService.updateMetadata asserts this) + assertThat(newState.metadata().index("standalone-index").getSettingsVersion(), equalTo(beforeVersion + 1)); + } + + public void testRemoveBackingIndex() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 3)), List.of()); + String indexToRemove = DataStream.getDefaultBackingIndexName(DATA_STREAM, 1); + long beforeVersion = cs.metadata().index(indexToRemove).getSettingsVersion(); + + ClusterState newState = modify(cs, List.of(DataStreamAction.removeBackingIndex(DATA_STREAM, indexToRemove))); + + DataStream ds = newState.metadata().dataStreams().get(DATA_STREAM); + assertThat(ds.getIndices().size(), equalTo(2)); + assertThat(ds.getIndices().stream().map(i -> i.getName()).toList(), not(hasItem(indexToRemove))); + // the removed index still exists as a standalone index and is no longer hidden + assertNotNull(newState.metadata().index(indexToRemove)); + assertFalse(IndexMetadata.INDEX_HIDDEN_SETTING.get(newState.metadata().index(indexToRemove).getSettings())); + // un-hiding changes settings, so the settings version must be bumped (IndexService.updateMetadata asserts this) + assertThat(newState.metadata().index(indexToRemove).getSettingsVersion(), equalTo(beforeVersion + 1)); + } + + public void testRemoveWriteIndexFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of()); + String writeIndex = DataStream.getDefaultBackingIndexName(DATA_STREAM, 2); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.removeBackingIndex(DATA_STREAM, writeIndex))) + ); + assertThat(e.getMessage(), containsString("is the write index")); + } + + public void testRemoveIndexNotPartOfDataStreamFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.removeBackingIndex(DATA_STREAM, "standalone-index"))) + ); + assertThat(e.getMessage(), containsString("is not part of data stream")); + } + + public void testAddToNonexistentDataStreamFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex("no-such-stream", "standalone-index"))) + ); + assertThat(e.getMessage(), containsString("data stream [no-such-stream] not found")); + } + + public void testAddNonexistentIndexFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of()); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "no-such-index"))) + ); + assertThat(e.getMessage(), containsString("index [no-such-index] not found")); + } + + public void testAddIndexWithAliasFails() { + ClusterState base = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of()); + IndexMetadata aliased = IndexMetadata.builder("aliased-index") + .settings(Settings.builder().put(SETTING_VERSION_CREATED, Version.CURRENT)) + .numberOfShards(1) + .numberOfReplicas(0) + .putAlias(AliasMetadata.builder("my-alias")) + .build(); + ClusterState cs = ClusterState.builder(new ClusterName("_name")) + .metadata(Metadata.builder(base.metadata()).put(aliased, false)) + .build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, "aliased-index"))) + ); + assertThat(e.getMessage(), containsString("aliases")); + } + + public void testAddIndexAlreadyBackingIndexFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 3)), List.of()); + // this index is already a backing index of the data stream + String existingBackingIndex = DataStream.getDefaultBackingIndexName(DATA_STREAM, 1); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, existingBackingIndex))) + ); + assertThat(e.getMessage(), containsString("already")); + } + + public void testAddIndexBelongingToAnotherDataStreamFails() { + ClusterState cs = getClusterStateWithDataStreams( + List.of(new Tuple<>(DATA_STREAM, 2), new Tuple<>("other-data-stream", 2)), + List.of() + ); + // this index is already a backing index of a different data stream + String otherBackingIndex = DataStream.getDefaultBackingIndexName("other-data-stream", 1); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, otherBackingIndex))) + ); + assertThat(e.getMessage(), containsString("is already a backing index of data stream [other-data-stream]")); + } + + public void testAddIndexWithGenerationConflictNameFails() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of()); + // an index whose name matches the backing-index naming pattern but with a counter greater than the stream's + // current generation would collide with a future rollover-created backing index; the request must be rejected with a + // 400 (IllegalArgumentException) up front rather than failing the cluster state build with a 500 (IllegalStateException) + String conflictingName = DataStream.getDefaultBackingIndexName(DATA_STREAM, 99); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> modify(cs, List.of(DataStreamAction.addBackingIndex(DATA_STREAM, conflictingName))) + ); + assertThat(e.getMessage(), containsString("conflicts with a future backing index")); + } + + public void testMultipleActionsAppliedAtomically() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 3)), List.of("standalone-index")); + String indexToRemove = DataStream.getDefaultBackingIndexName(DATA_STREAM, 1); + + ClusterState newState = modify( + cs, + List.of( + DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"), + DataStreamAction.removeBackingIndex(DATA_STREAM, indexToRemove) + ) + ); + + DataStream ds = newState.metadata().dataStreams().get(DATA_STREAM); + // started with 3 backing indices, +1 added, -1 removed = 3 + assertThat(ds.getIndices().size(), equalTo(3)); + assertThat(ds.getIndices().stream().map(i -> i.getName()).toList(), hasItem("standalone-index")); + assertThat(ds.getIndices().stream().map(i -> i.getName()).toList(), not(hasItem(indexToRemove))); + } + + public void testFailedActionLeavesClusterStateUnchanged() { + ClusterState cs = getClusterStateWithDataStreams(List.of(new Tuple<>(DATA_STREAM, 2)), List.of("standalone-index")); + String writeIndex = DataStream.getDefaultBackingIndexName(DATA_STREAM, 2); + Metadata before = cs.metadata(); + + // the second action targets the write index and must fail; the whole request fails with no partial state + expectThrows( + IllegalArgumentException.class, + () -> modify( + cs, + List.of( + DataStreamAction.addBackingIndex(DATA_STREAM, "standalone-index"), + DataStreamAction.removeBackingIndex(DATA_STREAM, writeIndex) + ) + ) + ); + // the input cluster state object is never mutated + assertThat(cs.metadata().dataStreams().get(DATA_STREAM).getIndices().size(), equalTo(2)); + assertFalse(IndexMetadata.INDEX_HIDDEN_SETTING.get(before.index("standalone-index").getSettings())); + } +} diff --git a/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsClientTests.java b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsClientTests.java new file mode 100644 index 0000000000000..6cc58b24792f6 --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsClientTests.java @@ -0,0 +1,96 @@ +/* + * 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.action.ActionRequest; +import org.opensearch.action.ActionType; +import org.opensearch.action.support.clustermanager.AcknowledgedResponse; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.action.ActionResponse; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.client.NoOpClient; + +import java.util.List; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.atomic.AtomicReference; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.sameInstance; + +/** + * Verifies that the {@code IndicesAdminClient} wiring for the modify-data-stream API routes the request through the + * client to {@link ModifyDataStreamsAction#INSTANCE}. These code paths are otherwise only exercised by the internal + * cluster IT, which does not feed the unit-test coverage report. + */ +public class ModifyDataStreamsClientTests extends OpenSearchTestCase { + + private ModifyDataStreamsAction.Request newRequest() { + return new ModifyDataStreamsAction.Request(List.of(DataStreamAction.addBackingIndex("my-data-stream", "my-index"))); + } + + public void testListenerVariantRoutesToAction() { + ModifyDataStreamsAction.Request request = newRequest(); + AtomicReference> capturedAction = new AtomicReference<>(); + AtomicReference capturedRequest = new AtomicReference<>(); + + try (NoOpClient client = capturingClient(capturedAction, capturedRequest)) { + AtomicReference response = new AtomicReference<>(); + client.admin().indices().modifyDataStream(request, ActionListener.wrap(response::set, e -> fail("unexpected failure: " + e))); + + assertThat(capturedAction.get(), sameInstance(ModifyDataStreamsAction.INSTANCE)); + assertThat(capturedRequest.get(), sameInstance(request)); + assertThat(response.get().isAcknowledged(), equalTo(true)); + } + } + + public void testFutureVariantRoutesToAction() throws Exception { + ModifyDataStreamsAction.Request request = newRequest(); + AtomicReference> capturedAction = new AtomicReference<>(); + AtomicReference capturedRequest = new AtomicReference<>(); + + try (NoOpClient client = capturingClient(capturedAction, capturedRequest)) { + AcknowledgedResponse response = client.admin().indices().modifyDataStream(request).get(); + + assertThat(capturedAction.get(), sameInstance(ModifyDataStreamsAction.INSTANCE)); + assertThat(capturedRequest.get(), sameInstance(request)); + assertThat(response.isAcknowledged(), equalTo(true)); + } + } + + public void testCompletionStageVariantRoutesToAction() throws Exception { + ModifyDataStreamsAction.Request request = newRequest(); + AtomicReference> capturedAction = new AtomicReference<>(); + AtomicReference capturedRequest = new AtomicReference<>(); + + try (NoOpClient client = capturingClient(capturedAction, capturedRequest)) { + CompletionStage stage = client.admin().indices().modifyDataStreamAsync(request); + AcknowledgedResponse response = stage.toCompletableFuture().get(); + + assertThat(capturedAction.get(), sameInstance(ModifyDataStreamsAction.INSTANCE)); + assertThat(capturedRequest.get(), sameInstance(request)); + assertThat(response.isAcknowledged(), equalTo(true)); + } + } + + private NoOpClient capturingClient(AtomicReference> capturedAction, AtomicReference capturedRequest) { + return new NoOpClient(getTestName()) { + @Override + @SuppressWarnings("unchecked") + protected void doExecute( + ActionType action, + Request request, + ActionListener listener + ) { + capturedAction.set(action); + capturedRequest.set(request); + listener.onResponse((Response) new AcknowledgedResponse(true)); + } + }; + } +} diff --git a/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsRequestTests.java b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsRequestTests.java new file mode 100644 index 0000000000000..33c4820fce0da --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsRequestTests.java @@ -0,0 +1,69 @@ +/* + * 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.action.ActionRequestValidationException; +import org.opensearch.action.admin.indices.datastream.ModifyDataStreamsAction.Request; +import org.opensearch.core.common.io.stream.Writeable; +import org.opensearch.test.AbstractWireSerializingTestCase; + +import java.util.ArrayList; +import java.util.List; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; + +public class ModifyDataStreamsRequestTests extends AbstractWireSerializingTestCase { + + @Override + protected Writeable.Reader instanceReader() { + return Request::new; + } + + @Override + protected Request createTestInstance() { + int numActions = randomIntBetween(1, 4); + List actions = new ArrayList<>(numActions); + for (int i = 0; i < numActions; i++) { + String dataStream = randomAlphaOfLength(6); + String index = randomAlphaOfLength(6); + if (randomBoolean()) { + actions.add(DataStreamAction.addBackingIndex(dataStream, index)); + } else { + actions.add(DataStreamAction.removeBackingIndex(dataStream, index)); + } + } + return new Request(actions); + } + + public void testValidateRequest() { + Request req = new Request(List.of(DataStreamAction.addBackingIndex("my-data-stream", "my-index"))); + ActionRequestValidationException e = req.validate(); + assertNull(e); + } + + public void testValidateRequestWithoutActions() { + Request req = new Request(List.of()); + ActionRequestValidationException e = req.validate(); + assertNotNull(e); + assertThat(e.validationErrors().size(), equalTo(1)); + assertThat(e.validationErrors().get(0), containsString("must specify at least one data stream modification action")); + } + + public void testIndicesResolveToBackingIndices() { + Request req = new Request( + List.of( + DataStreamAction.addBackingIndex("my-data-stream", "index-1"), + DataStreamAction.removeBackingIndex("my-data-stream", "index-2") + ) + ); + assertThat(req.indices(), equalTo(new String[] { "index-1", "index-2" })); + assertTrue(req.includeDataStreams()); + } +} diff --git a/server/src/test/java/org/opensearch/cluster/metadata/DataStreamTests.java b/server/src/test/java/org/opensearch/cluster/metadata/DataStreamTests.java index a70438307f94a..57b0dc33592e4 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/DataStreamTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/DataStreamTests.java @@ -113,6 +113,71 @@ public void testRemoveBackingIndex() { } } + public void testAddBackingIndex() { + int numBackingIndices = randomIntBetween(2, 32); + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + + List indices = new ArrayList<>(numBackingIndices); + for (int k = 1; k <= numBackingIndices; k++) { + indices.add(new Index(DataStream.getDefaultBackingIndexName(dataStreamName, k), UUIDs.randomBase64UUID(random()))); + } + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices); + + Index indexToAdd = new Index("index-to-add", UUIDs.randomBase64UUID(random())); + DataStream updated = original.addBackingIndex(indexToAdd); + assertThat(updated.getName(), equalTo(original.getName())); + // adding a backing index does not change the generation (only rollover does) + assertThat(updated.getGeneration(), equalTo(original.getGeneration())); + assertThat(updated.getTimeStampField(), equalTo(original.getTimeStampField())); + assertThat(updated.getIndices().size(), equalTo(numBackingIndices + 1)); + // the new index is added to the front so the existing write index stays last + assertThat(updated.getIndices().get(0), equalTo(indexToAdd)); + assertThat(updated.getIndices().get(updated.getIndices().size() - 1), equalTo(original.getIndices().get(numBackingIndices - 1))); + for (int k = 0; k < numBackingIndices; k++) { + assertThat(updated.getIndices().get(k + 1), equalTo(original.getIndices().get(k))); + } + } + + public void testRemoveBackingIndexThrowsExceptionIfIndexNotPartOfDataStream() { + int numBackingIndices = randomIntBetween(2, 32); + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + + List indices = new ArrayList<>(numBackingIndices); + for (int k = 1; k <= numBackingIndices; k++) { + indices.add(new Index(DataStream.getDefaultBackingIndexName(dataStreamName, k), UUIDs.randomBase64UUID(random()))); + } + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices); + + Index standaloneIndex = new Index("index-foo", UUIDs.randomBase64UUID(random())); + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> original.removeBackingIndex(standaloneIndex)); + assertThat(e.getMessage(), equalTo("index [index-foo] is not part of data stream [" + dataStreamName + "]")); + } + + public void testRemoveBackingIndexThrowsExceptionIfRemovingWriteIndex() { + int numBackingIndices = randomIntBetween(2, 32); + int writeIndexPosition = numBackingIndices - 1; + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + + List indices = new ArrayList<>(numBackingIndices); + for (int k = 1; k <= numBackingIndices; k++) { + indices.add(new Index(DataStream.getDefaultBackingIndexName(dataStreamName, k), UUIDs.randomBase64UUID(random()))); + } + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices); + + Index writeIndex = indices.get(writeIndexPosition); + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> original.removeBackingIndex(writeIndex)); + assertThat( + e.getMessage(), + equalTo( + "cannot remove backing index [" + + writeIndex.getName() + + "] of data stream [" + + dataStreamName + + "] because it is the write index" + ) + ); + } + public void testDefaultBackingIndexName() { // this test does little more than flag that changing the default naming convention for backing indices // will also require changing a lot of hard-coded values in REST tests and docs diff --git a/server/src/test/java/org/opensearch/index/mapper/DataStreamFieldMapperTests.java b/server/src/test/java/org/opensearch/index/mapper/DataStreamFieldMapperTests.java index 8fdc1b8a62be6..877babd9f9cbf 100644 --- a/server/src/test/java/org/opensearch/index/mapper/DataStreamFieldMapperTests.java +++ b/server/src/test/java/org/opensearch/index/mapper/DataStreamFieldMapperTests.java @@ -113,6 +113,79 @@ public void testDeeplyNestedCustomTimestampField() throws Exception { ); } + public void testEnableOnMappingMergeIsAllowed() throws Exception { + MapperService mapperService = createIndex("test").mapperService(); + + // an index that starts out without the _data_stream_timestamp meta field (the default disabled state) + mapperService.merge( + "_doc", + new CompressedXContent("{\"_doc\":{\"properties\":{\"@timestamp\":{\"type\":\"date\"}}}}"), + MapperService.MergeReason.MAPPING_UPDATE + ); + + // enabling the meta field on a subsequent merge is allowed (this is how an existing index is adapted into a backing index) + String enableMapping = XContentFactory.jsonBuilder() + .startObject() + .startObject("_doc") + .startObject("_data_stream_timestamp") + .field("enabled", true) + .endObject() + .endObject() + .endObject() + .toString(); + DocumentMapper merged = mapperService.merge( + "_doc", + new CompressedXContent(enableMapping), + MapperService.MergeReason.MAPPING_UPDATE + ); + + // the meta field is now enabled, so the timestamp validation in postParse is active + MapperException exception = expectThrows( + MapperException.class, + () -> merged.parse( + new SourceToParse( + "test", + "1", + BytesReference.bytes(XContentFactory.jsonBuilder().startObject().endObject()), + MediaTypeRegistry.JSON + ) + ) + ); + assertThat(exception.getCause().getMessage(), containsString("documents must contain a single-valued timestamp field")); + } + + public void testDisableOnceEnabledIsRejected() throws Exception { + MapperService mapperService = createIndex("test").mapperService(); + + // start with the meta field enabled + String enableMapping = XContentFactory.jsonBuilder() + .startObject() + .startObject("_doc") + .startObject("_data_stream_timestamp") + .field("enabled", true) + .endObject() + .endObject() + .endObject() + .toString(); + mapperService.merge("_doc", new CompressedXContent(enableMapping), MapperService.MergeReason.MAPPING_UPDATE); + + // attempting to disable it again on a subsequent merge is rejected by the merge validator + String disableMapping = XContentFactory.jsonBuilder() + .startObject() + .startObject("_doc") + .startObject("_data_stream_timestamp") + .field("enabled", false) + .endObject() + .endObject() + .endObject() + .toString(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> mapperService.merge("_doc", new CompressedXContent(disableMapping), MapperService.MergeReason.MAPPING_UPDATE) + ); + assertThat(exception.getMessage(), containsString("Cannot update parameter [enabled] from [true] to [false]")); + } + private void assertDataStreamFieldMapper(String mapping, String timestampFieldName) throws Exception { DocumentMapper mapper = createIndex("test").mapperService() .merge("_doc", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE);