diff --git a/CHANGELOG.md b/CHANGELOG.md index 1da2f241a4c03..cce5d921450d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [Unreleased 2.x] ### Added +- Add API to modify a data stream's backing indices, and an `attach_to_data_stream` option on snapshot restore ([#22487](https://github.com/opensearch-project/OpenSearch/pull/22487), [#22539](https://github.com/opensearch-project/OpenSearch/pull/22539)) ### Dependencies - Updated Netty to 4.1.137.Final ([#21491](https://github.com/opensearch-project/OpenSearch/pull/21491), [#21968](https://github.com/opensearch-project/OpenSearch/pull/21968), [#22692](https://github.com/opensearch-project/OpenSearch/pull/22692)) diff --git a/buildSrc/version.properties b/buildSrc/version.properties index af8e41f5d4925..47ba92ed367d0 100644 --- a/buildSrc/version.properties +++ b/buildSrc/version.properties @@ -1 +1 @@ -opensearch = 2.19.6 +opensearch = 2.19.7 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 b0476a65e6d8c..8d6ca27b2697e 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 @@ -884,7 +884,8 @@ public void testApiNamingConventions() throws Exception { "cluster.delete_weighted_routing", "cluster.put_decommission_awareness", "cluster.get_decommission_awareness", - "cluster.delete_decommission_awareness", }; + "cluster.delete_decommission_awareness", + "indices.modify_data_stream" }; List booleanReturnMethods = Arrays.asList("security.enable_user", "security.disable_user", "security.change_password"); Set deprecatedMethods = new HashSet<>(); deprecatedMethods.add("indices.force_merge"); diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 3ab94743acc78..e60d87537ffaf 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -1,5 +1,5 @@ [versions] -opensearch = "2.19.6" +opensearch = "2.19.7" lucene = "9.12.3" bundled_jdk_vendor = "adoptium" diff --git a/libs/core/src/main/java/org/opensearch/Version.java b/libs/core/src/main/java/org/opensearch/Version.java index 8a62de05af63a..aed3280070008 100644 --- a/libs/core/src/main/java/org/opensearch/Version.java +++ b/libs/core/src/main/java/org/opensearch/Version.java @@ -151,7 +151,8 @@ public class Version implements Comparable, ToXContentFragment { public static final Version V_2_19_4 = new Version(2190499, org.apache.lucene.util.Version.LUCENE_9_12_3); public static final Version V_2_19_5 = new Version(2190599, org.apache.lucene.util.Version.LUCENE_9_12_3); public static final Version V_2_19_6 = new Version(2190699, org.apache.lucene.util.Version.LUCENE_9_12_3); - public static final Version CURRENT = V_2_19_6; + public static final Version V_2_19_7 = new Version(2190799, org.apache.lucene.util.Version.LUCENE_9_12_3); + public static final Version CURRENT = V_2_19_7; public static Version fromId(int id) { final Version known = LegacyESVersion.idToVersion.get(id); 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..9b780186b017e --- /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":"Adds or removes backing indices of a data stream with metadata-only actions" + }, + "stability":"experimental", + "url":{ + "paths":[ + { + "path":"/_data_stream/_modify", + "methods":[ + "POST" + ] + } + ] + }, + "params":{ + "cluster_manager_timeout":{ + "type":"time", + "description":"Explicit operation timeout for connection to cluster-manager node" + }, + "timeout":{ + "type":"time", + "description":"Explicit operation timeout" + } + }, + "body":{ + "description":"The data stream modify actions", + "required":true + } + } +} 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..cbe3f1a1c66bc --- /dev/null +++ b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsIT.java @@ -0,0 +1,204 @@ +/* + * 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.support.master.AcknowledgedResponse; +import org.opensearch.cluster.metadata.DataStream; + +import java.util.List; +import java.util.stream.Collectors; + +import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_HIDDEN_SETTING; +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; + +/** + * End-to-end coverage for the {@code POST /_data_stream/_modify} API: metadata-only add/remove of backing indices. + * A data stream's generation is derived from its backing indices, so it is never set directly. The API is used to + * re-attach backing indices that exist but are not (or are no longer) part of the stream — for example, indices + * restored from cold storage that were not re-associated with the stream (opensearch-project/OpenSearch#8271). + */ +public class ModifyDataStreamsIT extends DataStreamTestCase { + + private static final String DS = "logs-modify"; + + private List backingIndices() throws Exception { + return getDataStreams(DS).getDataStreams() + .get(0) + .getDataStream() + .getIndices() + .stream() + .map(i -> i.getName()) + .collect(Collectors.toList()); + } + + private long generation() throws Exception { + return getDataStreams(DS).getDataStreams().get(0).getDataStream().getGeneration(); + } + + private AcknowledgedResponse modify(List actions) throws Exception { + return client().execute(ModifyDataStreamsAction.INSTANCE, new ModifyDataStreamsAction.Request(actions)).get(); + } + + public void testRemoveBackingIndex() throws Exception { + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); // gen 2 + rolloverDataStream(DS); // gen 3 + assertThat(backingIndices(), hasSize(3)); + + String firstBackingIndex = DataStream.getDefaultBackingIndexName(DS, 1); + assertAcked(modify(List.of(DataStreamAction.removeBackingIndex(DS, firstBackingIndex)))); + + List after = backingIndices(); + assertThat(after, hasSize(2)); + assertThat(after.contains(firstBackingIndex), equalTo(false)); + // The removed index still exists as a standalone index; it is only detached from the stream. + assertTrue(client().admin().indices().prepareExists(firstBackingIndex).get().isExists()); + // Generation is unchanged: removing a non-write index does not affect it. + assertThat(generation(), equalTo(3L)); + } + + public void testCannotRemoveWriteIndex() throws Exception { + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); + String writeIndex = DataStream.getDefaultBackingIndexName(DS, generation()); + + Exception e = expectThrows(Exception.class, () -> modify(List.of(DataStreamAction.removeBackingIndex(DS, writeIndex)))); + assertThat(org.opensearch.ExceptionsHelper.unwrapCause(e).getMessage(), containsString("because it is the write index")); + } + + public void testReattachDetachedBackingIndex() throws Exception { + // Models the #8271 flow: a backing index is detached (e.g. tiered to cold storage) and later re-attached. + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); // gen 2 + rolloverDataStream(DS); // gen 3, backing [1, 2, 3] + + String detached = DataStream.getDefaultBackingIndexName(DS, 1); + assertAcked(modify(List.of(DataStreamAction.removeBackingIndex(DS, detached)))); + assertThat(backingIndices(), hasSize(2)); + + // Re-attach it; the API inserts it back in generation order and leaves the derived generation untouched. + assertAcked(modify(List.of(DataStreamAction.addBackingIndex(DS, detached)))); + assertThat( + backingIndices(), + contains( + DataStream.getDefaultBackingIndexName(DS, 1), + DataStream.getDefaultBackingIndexName(DS, 2), + DataStream.getDefaultBackingIndexName(DS, 3) + ) + ); + assertThat(generation(), equalTo(3L)); + } + + public void testDetachUnhidesAndReattachHidesBackingIndex() throws Exception { + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); // gen 2, backing [1, 2] + + String detached = DataStream.getDefaultBackingIndexName(DS, 1); + // Backing indices are hidden. + assertThat(hidden(detached), equalTo(true)); + + // Detaching makes it visible again (mirrors hide-on-attach). + assertAcked(modify(List.of(DataStreamAction.removeBackingIndex(DS, detached)))); + assertThat(hidden(detached), equalTo(false)); + + // Re-attaching hides it again, like every backing index. + assertAcked(modify(List.of(DataStreamAction.addBackingIndex(DS, detached)))); + assertThat(hidden(detached), equalTo(true)); + } + + private boolean hidden(String index) { + return INDEX_HIDDEN_SETTING.get(client().admin().cluster().prepareState().get().getState().metadata().index(index).getSettings()); + } + + public void testMigrateArbitraryNamedIndexIntoDataStream() throws Exception { + // A user with a pre-existing regular index wants to migrate to data streams: the legacy index is attached as an + // older (non-write) backing index even though its name does not follow the .ds--NNNNNN convention. + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); // gen 2, backing [1, 2] + long genBefore = generation(); + + String legacy = "legacy-logs-2023"; + // The index must map the data stream's @timestamp field as a date for data stream search to work. + assertAcked( + client().admin().indices().prepareCreate(legacy).setMapping("{\"properties\":{\"@timestamp\":{\"type\":\"date\"}}}").get() + ); + + assertAcked(modify(List.of(DataStreamAction.addBackingIndex(DS, legacy)))); + + List after = backingIndices(); + // Arbitrary-named index is the oldest backing index; the convention-following write index remains last. + assertThat(after.get(0), equalTo(legacy)); + assertThat(after.get(after.size() - 1), equalTo(DataStream.getDefaultBackingIndexName(DS, genBefore))); + assertThat(after, hasSize(3)); + // Generation is unchanged (the migrated index is not the write index). + assertThat(generation(), equalTo(genBefore)); + // The migrated index is now hidden like every backing index. + assertThat(hidden(legacy), equalTo(true)); + + // Rollover still advances correctly off the convention-following write index. + rolloverDataStream(DS); + assertThat(generation(), equalTo(genBefore + 1)); + } + + public void testCannotAddIndexWithoutTimestampMapping() throws Exception { + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); + + // Index has no @timestamp date field, so data stream search would not work; the add must be rejected. + String noTimestamp = "no-timestamp-idx"; + assertAcked( + client().admin().indices().prepareCreate(noTimestamp).setMapping("{\"properties\":{\"msg\":{\"type\":\"text\"}}}").get() + ); + + Exception e = expectThrows(Exception.class, () -> modify(List.of(DataStreamAction.addBackingIndex(DS, noTimestamp)))); + assertThat( + org.opensearch.ExceptionsHelper.unwrapCause(e).getMessage(), + containsString("does not have a [@timestamp] field mapped as a date type") + ); + } + + public void testRemoveAndReattachMultipleIndicesInSingleRequest() throws Exception { + createDataStreamIndexTemplate("template", List.of("logs-*")); + createDataStream(DS); + rolloverDataStream(DS); // gen 2 + rolloverDataStream(DS); // gen 3 + rolloverDataStream(DS); // gen 4, backing [1, 2, 3, 4] + + String gen1 = DataStream.getDefaultBackingIndexName(DS, 1); + String gen2 = DataStream.getDefaultBackingIndexName(DS, 2); + + // Remove two non-write indices in one request. + assertAcked(modify(List.of(DataStreamAction.removeBackingIndex(DS, gen1), DataStreamAction.removeBackingIndex(DS, gen2)))); + assertThat(backingIndices(), hasSize(2)); + + // Re-attach both in one request. + assertAcked(modify(List.of(DataStreamAction.addBackingIndex(DS, gen1), DataStreamAction.addBackingIndex(DS, gen2)))); + assertThat( + backingIndices(), + containsInAnyOrder(gen1, gen2, DataStream.getDefaultBackingIndexName(DS, 3), DataStream.getDefaultBackingIndexName(DS, 4)) + ); + assertThat(generation(), equalTo(4L)); + + // Rollover still computes the correct next generation, proving generation stayed in sync. + rolloverDataStream(DS); + assertThat(generation(), equalTo(5L)); + assertTrue(backingIndices().contains(DataStream.getDefaultBackingIndexName(DS, 5))); + } +} diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/DataStreamRestoreAutoAttachIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/DataStreamRestoreAutoAttachIT.java new file mode 100644 index 0000000000000..de89a3bdb8fb0 --- /dev/null +++ b/server/src/internalClusterTest/java/org/opensearch/snapshots/DataStreamRestoreAutoAttachIT.java @@ -0,0 +1,170 @@ +/* + * 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.snapshots; + +import org.opensearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse; +import org.opensearch.action.admin.indices.datastream.CreateDataStreamAction; +import org.opensearch.action.admin.indices.datastream.DataStreamAction; +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.rollover.RolloverRequest; +import org.opensearch.action.admin.indices.template.put.PutComposableIndexTemplateAction; +import org.opensearch.cluster.metadata.ComposableIndexTemplate; +import org.opensearch.cluster.metadata.DataStream; +import org.opensearch.cluster.metadata.Template; +import org.opensearch.common.settings.Settings; + +import java.util.List; +import java.util.stream.Collectors; + +import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.equalTo; + +/** + * Verifies that restoring a data stream backing index attaches it to a pre-existing data stream of the same name in the + * same cluster-state update when {@code attach_to_data_stream} is set, advancing the stream generation when the + * attached index has a higher counter than the current write index. + */ +public class DataStreamRestoreAutoAttachIT extends AbstractSnapshotIntegTestCase { + + private static final String REPO = "test-repo"; + private static final String DS = "logs-attach"; + + private void createTemplate() throws Exception { + ComposableIndexTemplate template = new ComposableIndexTemplate( + List.of("logs-*"), + new Template(Settings.builder().put("number_of_shards", 1).put("number_of_replicas", 0).build(), null, null), + null, + null, + null, + null, + new ComposableIndexTemplate.DataStreamTemplate(new DataStream.TimestampField("@timestamp")) + ); + PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request("ds-template"); + request.indexTemplate(template); + assertAcked(client().execute(PutComposableIndexTemplateAction.INSTANCE, request).get()); + } + + private List backingIndices() throws Exception { + return client().admin() + .indices() + .getDataStreams(new GetDataStreamAction.Request(new String[] { DS })) + .get() + .getDataStreams() + .get(0) + .getDataStream() + .getIndices() + .stream() + .map(i -> i.getName()) + .collect(Collectors.toList()); + } + + private long generation() throws Exception { + return client().admin() + .indices() + .getDataStreams(new GetDataStreamAction.Request(new String[] { DS })) + .get() + .getDataStreams() + .get(0) + .getDataStream() + .getGeneration(); + } + + public void testRestoreReattachesDetachedBackingIndex() throws Exception { + createRepository(REPO, "fs"); + createTemplate(); + assertAcked(client().admin().indices().createDataStream(new CreateDataStreamAction.Request(DS)).get()); + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(DS, null)).get().isRolledOver(), equalTo(true)); // gen 2 + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(DS, null)).get().isRolledOver(), equalTo(true)); // gen 3 + assertThat(generation(), equalTo(3L)); + + // Snapshot the whole data stream (includes the gen-3 backing index). + createSnapshot(REPO, "snap", List.of(DS)); + + // Roll once more so the gen-3 index is no longer the write index, then detach and delete it to model a backing + // index that is missing locally but present in the snapshot. + String gen3 = DataStream.getDefaultBackingIndexName(DS, 3); + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(DS, null)).get().isRolledOver(), equalTo(true)); // gen 4 + assertAcked( + client().execute( + ModifyDataStreamsAction.INSTANCE, + new ModifyDataStreamsAction.Request(List.of(DataStreamAction.removeBackingIndex(DS, gen3))) + ).get() + ); + assertAcked(client().admin().indices().delete(new DeleteIndexRequest(gen3)).get()); + assertThat(backingIndices().contains(gen3), equalTo(false)); + + // Restore just the detached backing index; with attach_to_data_stream it re-attaches to the stream. + RestoreSnapshotResponse restore = client().admin() + .cluster() + .prepareRestoreSnapshot(REPO, "snap") + .setIndices(gen3) + .setAttachToDataStream(true) + .setWaitForCompletion(true) + .get(); + assertThat(restore.getRestoreInfo().successfulShards(), equalTo(restore.getRestoreInfo().totalShards())); + + // The restored index is back in the stream, in generation order. + assertThat(backingIndices().contains(gen3), equalTo(true)); + assertThat( + backingIndices(), + contains( + DataStream.getDefaultBackingIndexName(DS, 1), + DataStream.getDefaultBackingIndexName(DS, 2), + DataStream.getDefaultBackingIndexName(DS, 3), + DataStream.getDefaultBackingIndexName(DS, 4) + ) + ); + // Generation stays at 4 (gen-3 is not the write index); reattaching a lower-counter index does not change it. + assertThat(generation(), equalTo(4L)); + } + + public void testRestoreWithRenameAdvancesGenerationOfRenamedStream() throws Exception { + // Exercises the two things a unit test of attachRestoredBackingIndices cannot: that the restore pipeline feeds + // the post-rename index name into attach, and that attaching a higher-generation index passes cluster-state + // validation atomically and leaves rollover working. The source is a generation ahead of the target. + createRepository(REPO, "fs"); + createTemplate(); + + // Source stream rolled to generation 3, snapshotted; its gen-3 index is the write index. + String source = "logs-source"; + assertAcked(client().admin().indices().createDataStream(new CreateDataStreamAction.Request(source)).get()); + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(source, null)).get().isRolledOver(), equalTo(true)); // 2 + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(source, null)).get().isRolledOver(), equalTo(true)); // 3 + createSnapshot(REPO, "snap", List.of(source)); + + // Target stream logs-attach is a generation behind at 2 (backing [1, 2]). + assertAcked(client().admin().indices().createDataStream(new CreateDataStreamAction.Request(DS)).get()); + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(DS, null)).get().isRolledOver(), equalTo(true)); // 2 + assertThat(generation(), equalTo(2L)); + + // Restore the source's gen-3 index, renamed into the target stream. Auto-attach keys off the post-rename name, + // and the higher counter advances the target generation to 3, making it the new write index. + String targetGen3 = DataStream.getDefaultBackingIndexName(DS, 3); + RestoreSnapshotResponse restore = client().admin() + .cluster() + .prepareRestoreSnapshot(REPO, "snap") + .setIndices(DataStream.getDefaultBackingIndexName(source, 3)) + .setRenamePattern(source) + .setRenameReplacement(DS) + .setAttachToDataStream(true) + .setWaitForCompletion(true) + .get(); + assertThat(restore.getRestoreInfo().successfulShards(), equalTo(restore.getRestoreInfo().totalShards())); + + assertTrue(backingIndices().contains(targetGen3)); + assertThat(generation(), equalTo(3L)); + assertThat(backingIndices().get(backingIndices().size() - 1), equalTo(targetGen3)); + // A subsequent rollover advances off the new write index, confirming generation stayed in sync. + assertThat(client().admin().indices().rolloverIndex(new RolloverRequest(DS, null)).get().isRolledOver(), equalTo(true)); + assertThat(generation(), equalTo(4L)); + } +} diff --git a/server/src/main/java/org/opensearch/action/ActionModule.java b/server/src/main/java/org/opensearch/action/ActionModule.java index 1072247357c7a..05a6e78cb807c 100644 --- a/server/src/main/java/org/opensearch/action/ActionModule.java +++ b/server/src/main/java/org/opensearch/action/ActionModule.java @@ -151,6 +151,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; @@ -404,6 +405,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.RestPutComponentTemplateAction; import org.opensearch.rest.action.admin.indices.RestPutComposableIndexTemplateAction; @@ -746,6 +748,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); @@ -947,6 +950,7 @@ public void initRestHandlers(Supplier nodesInCluster) { // Data Stream API registerHandler.accept(new RestCreateDataStreamAction()); registerHandler.accept(new RestDeleteDataStreamAction()); + registerHandler.accept(new RestModifyDataStreamsAction()); registerHandler.accept(new RestGetDataStreamsAction()); registerHandler.accept(new RestResolveIndexAction()); registerHandler.accept(new RestDataStreamsStatsAction()); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java index 4d678893a8851..101e3358a4639 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java @@ -130,6 +130,8 @@ private static StorageType fromString(String string) { @Nullable // if any snapshot UUID will do private String snapshotUuid; + private boolean attachToDataStream = false; + public RestoreSnapshotRequest() {} /** @@ -178,6 +180,9 @@ public RestoreSnapshotRequest(StreamInput in) throws IOException { if (in.getVersion().onOrAfter(Version.V_2_18_0)) { renameAliasReplacement = in.readOptionalString(); } + if (in.getVersion().onOrAfter(Version.V_2_19_7)) { + attachToDataStream = in.readBoolean(); + } } @Override @@ -220,6 +225,9 @@ public void writeTo(StreamOutput out) throws IOException { if (out.getVersion().onOrAfter(Version.V_2_18_0)) { out.writeOptionalString(renameAliasReplacement); } + if (out.getVersion().onOrAfter(Version.V_2_19_7)) { + out.writeBoolean(attachToDataStream); + } } @Override @@ -655,6 +663,26 @@ public String getSourceRemoteTranslogRepository() { return sourceRemoteTranslogRepository; } + /** + * When {@code true}, a restored index whose name matches the data stream backing-index convention + * ({@code .ds--NNNNNN}) is attached to a pre-existing data stream of the same name as part of the + * restore. Defaults to {@code false}, in which case such an index is restored as a standalone index. + * + * @param attachToDataStream whether to attach matching restored indices to their data stream + * @return this request + */ + public RestoreSnapshotRequest attachToDataStream(boolean attachToDataStream) { + this.attachToDataStream = attachToDataStream; + return this; + } + + /** + * Returns whether matching restored indices are attached to their data stream. + */ + public boolean attachToDataStream() { + return attachToDataStream; + } + /** * Parses restore definition * @@ -744,6 +772,8 @@ public RestoreSnapshotRequest source(Map source) { } else { throw new IllegalArgumentException("malformed source_remote_translog_repository"); } + } else if (name.equals("attach_to_data_stream")) { + attachToDataStream(nodeBooleanValue(entry.getValue(), "attach_to_data_stream")); } else { if (IndicesOptions.isIndicesOptions(name) == false) { throw new IllegalArgumentException("Unknown parameter " + name); @@ -801,6 +831,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws if (sourceRemoteTranslogRepository != null) { builder.field("source_remote_translog_repository", sourceRemoteTranslogRepository); } + builder.field("attach_to_data_stream", attachToDataStream); builder.endObject(); return builder; } @@ -832,7 +863,8 @@ public boolean equals(Object o) { && Objects.equals(snapshotUuid, that.snapshotUuid) && Objects.equals(storageType, that.storageType) && Objects.equals(sourceRemoteStoreRepository, that.sourceRemoteStoreRepository) - && Objects.equals(sourceRemoteTranslogRepository, that.sourceRemoteTranslogRepository); + && Objects.equals(sourceRemoteTranslogRepository, that.sourceRemoteTranslogRepository) + && attachToDataStream == that.attachToDataStream; return equals; } @@ -855,7 +887,8 @@ public int hashCode() { snapshotUuid, storageType, sourceRemoteStoreRepository, - sourceRemoteTranslogRepository + sourceRemoteTranslogRepository, + attachToDataStream ); result = 31 * result + Arrays.hashCode(indices); result = 31 * result + Arrays.hashCode(ignoreIndexSettings); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java index 038d62ad7f4cb..2177d7f391713 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java @@ -302,4 +302,13 @@ public RestoreSnapshotRequestBuilder setSourceRemoteTranslogRepository(String re request.setSourceRemoteTranslogRepository(repositoryName); return this; } + + /** + * If set to true, a restored index whose name matches the data stream backing-index convention is attached to a + * pre-existing data stream of the same name instead of being restored as a standalone index. + */ + public RestoreSnapshotRequestBuilder setAttachToDataStream(boolean attachToDataStream) { + request.attachToDataStream(attachToDataStream); + return this; + } } 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..3be2747a64997 --- /dev/null +++ b/server/src/main/java/org/opensearch/action/admin/indices/datastream/DataStreamAction.java @@ -0,0 +1,184 @@ +/* + * 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.annotation.ExperimentalApi; +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; + +/** + * A single add- or remove-backing-index operation within a {@link ModifyDataStreamsAction} request. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class DataStreamAction implements Writeable, ToXContentObject { + + private final Type type; + private final String dataStream; + private final String index; + + /** + * The type of modification applied to a data stream. + * + * @opensearch.experimental + */ + @ExperimentalApi + 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 + "]"); + } + } + } + + public DataStreamAction(Type type, String dataStream, String index) { + Objects.requireNonNull(type, "[type] is required"); + Objects.requireNonNull(dataStream, "[data_stream] is required"); + Objects.requireNonNull(index, "[index] is required"); + this.type = type; + this.dataStream = dataStream; + this.index = index; + } + + public Type type() { + return type; + } + + public String dataStream() { + return dataStream; + } + + public String index() { + return 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); + } + + public DataStreamAction(StreamInput in) throws IOException { + this(Type.fromValue(in.readByte()), in.readString(), 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"); + + // Parser per action type; the enclosing action is a single-key object whose key is the action type. + private static final ConstructingObjectParser ADD_PARSER = actionParser(Type.ADD_BACKING_INDEX); + private static final ConstructingObjectParser REMOVE_PARSER = actionParser(Type.REMOVE_BACKING_INDEX); + + private static final ObjectParser PARSER = new ObjectParser<>("data_stream_action", ActionHolder::new); + + static { + PARSER.declareObject((holder, action) -> holder.action = action, ADD_PARSER, new ParseField(Type.ADD_BACKING_INDEX.fieldName())); + PARSER.declareObject( + (holder, action) -> holder.action = action, + REMOVE_PARSER, + new ParseField(Type.REMOVE_BACKING_INDEX.fieldName()) + ); + } + + private static ConstructingObjectParser actionParser(Type type) { + ConstructingObjectParser parser = new ConstructingObjectParser<>( + type.fieldName(), + false, + args -> new DataStreamAction(type, (String) args[0], (String) args[1]) + ); + parser.declareString(ConstructingObjectParser.constructorArg(), DATA_STREAM_FIELD); + parser.declareString(ConstructingObjectParser.constructorArg(), INDEX_FIELD); + return parser; + } + + public static DataStreamAction fromXContent(XContentParser parser) throws IOException { + return PARSER.parse(parser, null).action; + } + + @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; + } + + private static class ActionHolder { + private DataStreamAction action; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DataStreamAction other = (DataStreamAction) o; + return type == other.type && dataStream.equals(other.dataStream) && index.equals(other.index); + } + + @Override + public int hashCode() { + return Objects.hash(type, dataStream, index); + } + + @Override + public String toString() { + return "DataStreamAction[type=" + type + ", dataStream=" + dataStream + ", index=" + 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..6e4aa16566b31 --- /dev/null +++ b/server/src/main/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsAction.java @@ -0,0 +1,153 @@ +/* + * 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.ActionType; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction; +import org.opensearch.action.support.master.AcknowledgedRequest; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.block.ClusterBlockException; +import org.opensearch.cluster.block.ClusterBlockLevel; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; +import org.opensearch.cluster.metadata.MetadataDataStreamsService; +import org.opensearch.cluster.metadata.MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.common.inject.Inject; +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.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +import static org.opensearch.action.ValidateActions.addValidationError; + +/** + * Applies a batch of {@link DataStreamAction} operations (add/remove backing index) to data-stream metadata atomically. + * + * @opensearch.experimental + */ +@ExperimentalApi +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 carrying the data-stream actions to apply. + * + * @opensearch.experimental + */ + @ExperimentalApi + public static class Request extends AcknowledgedRequest { + + private final List actions; + + public Request(List actions) { + this.actions = new ArrayList<>(Objects.requireNonNull(actions, "actions must not be null")); + } + + public Request(StreamInput in) throws IOException { + super(in); + this.actions = 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 ActionRequestValidationException validate() { + ActionRequestValidationException validationException = null; + if (actions == null || actions.isEmpty()) { + validationException = addValidationError("at least one data stream action must be specified", validationException); + } + return validationException; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Request request = (Request) o; + return actions.equals(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 MetadataDataStreamsService metadataDataStreamsService; + + @Inject + public TransportAction( + TransportService transportService, + ClusterService clusterService, + ThreadPool threadPool, + ActionFilters actionFilters, + IndexNameExpressionResolver indexNameExpressionResolver, + MetadataDataStreamsService metadataDataStreamsService + ) { + super(NAME, transportService, clusterService, threadPool, actionFilters, Request::new, indexNameExpressionResolver); + this.metadataDataStreamsService = metadataDataStreamsService; + } + + @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) { + ModifyDataStreamsClusterStateUpdateRequest updateRequest = new ModifyDataStreamsClusterStateUpdateRequest( + request.getActions(), + request.clusterManagerNodeTimeout(), + request.timeout() + ); + metadataDataStreamsService.modifyDataStream(updateRequest, listener); + } + + @Override + protected ClusterBlockException checkBlock(Request request, ClusterState state) { + return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE); + } + } +} 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..ec019c7e62fbb 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/DataStream.java @@ -148,6 +148,61 @@ public DataStream removeBackingIndex(Index index) { return new DataStream(name, timeStampField, backingIndices, generation); } + /** + * Adds the given index to the data stream's backing indices and returns a new {@code DataStream} instance. Backing + * indices are ordered oldest-to-newest so the write index remains last: convention-named indices + * ({@code .ds--NNNNNN}) order by their counter, and non-convention names sort first. The generation is derived + * as the highest backing-index counter, so adding an index whose counter exceeds the current generation advances + * the generation and makes it the new write index; adding a lower-counter or non-convention index leaves the + * generation unchanged. + * + * @param index the backing index to add + * @return new {@code DataStream} instance with the index added + */ + public DataStream addBackingIndex(Index index) { + List backingIndices = new ArrayList<>(indices); + backingIndices.add(index); + backingIndices.sort(Comparator.comparingLong(i -> backingIndexCounterOrMin(name, i.getName()))); + long newGeneration = Math.max(generation, backingIndexCounterOrMin(name, index.getName())); + return new DataStream(name, timeStampField, backingIndices, newGeneration); + } + + /** + * Returns the generation counter of a backing index named {@code .ds--NNNNNN}, or + * {@link Long#MIN_VALUE} for any other name so that non-convention indices sort before convention-named ones and + * never become the write index. + */ + static long backingIndexCounterOrMin(String dataStreamName, String indexName) { + if (dataStreamName.equals(parseDataStreamName(indexName)) == false) { + return Long.MIN_VALUE; + } + return Long.parseLong(indexName.substring(indexName.lastIndexOf('-') + 1)); + } + + /** + * If {@code indexName} follows the data stream backing-index naming convention {@code .ds--NNNNNN} + * (a numeric counter suffix), returns the {@code } portion; otherwise returns {@code null}. + */ + public static String parseDataStreamName(String indexName) { + if (indexName.startsWith(BACKING_INDEX_PREFIX) == false) { + return null; + } + int counterDash = indexName.lastIndexOf('-'); + if (counterDash <= BACKING_INDEX_PREFIX.length()) { + return null; + } + String counter = indexName.substring(counterDash + 1); + if (counter.isEmpty()) { + return null; + } + for (int i = 0; i < counter.length(); i++) { + if (Character.isDigit(counter.charAt(i)) == false) { + return null; + } + } + return indexName.substring(BACKING_INDEX_PREFIX.length(), counterDash); + } + /** * Replaces the specified backing index with a new index and returns a new {@code DataStream} instance with * the modified backing indices. An {@code IllegalArgumentException} is thrown if the index to be replaced diff --git a/server/src/main/java/org/opensearch/cluster/metadata/MetadataDataStreamsService.java b/server/src/main/java/org/opensearch/cluster/metadata/MetadataDataStreamsService.java new file mode 100644 index 0000000000000..5694425a8dedb --- /dev/null +++ b/server/src/main/java/org/opensearch/cluster/metadata/MetadataDataStreamsService.java @@ -0,0 +1,303 @@ +/* + * 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.cluster.metadata; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.action.admin.indices.datastream.DataStreamAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.cluster.AckedClusterStateUpdateTask; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.ack.ClusterStateUpdateRequest; +import org.opensearch.cluster.ack.ClusterStateUpdateResponse; +import org.opensearch.cluster.service.ClusterManagerTaskKeys; +import org.opensearch.cluster.service.ClusterManagerTaskThrottler; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.Priority; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.index.Index; +import org.opensearch.index.mapper.DateFieldMapper; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Adds or removes backing indices of data streams in a single atomic cluster-state update. These are metadata-only + * mutations; they never create, delete, restore, open, close, or relocate indices. The write index (and hence the + * generation) is preserved: added indices become non-write backing indices, and the write index cannot be removed. + * Added indices need not follow the {@code .ds--NNNNNN} naming convention, which allows migrating + * pre-existing regular indices into a data stream. + *

+ * An added index is marked hidden (as all backing indices are) and a removed index is made visible again. If an index + * was hidden before it was ever a backing index, detaching it therefore leaves it visible; the caller can hide it + * again. + *

+ * This service is injectable, so components such as cross-cluster replication can call {@link #modifyDataStream} + * directly rather than through {@code ModifyDataStreamsAction}'s transport layer. + * + * @opensearch.internal + */ +public class MetadataDataStreamsService { + + private static final Logger logger = LogManager.getLogger(MetadataDataStreamsService.class); + + private final ClusterService clusterService; + private final ClusterManagerTaskThrottler.ThrottlingKey modifyDataStreamTaskKey; + + public MetadataDataStreamsService(ClusterService clusterService) { + this.clusterService = clusterService; + // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction. + this.modifyDataStreamTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.MODIFY_DATA_STREAM_KEY, true); + } + + /** + * Convenience entry point for in-process callers (e.g. cross-cluster replication) that want to apply a batch of + * data-stream actions without constructing a {@link ModifyDataStreamsClusterStateUpdateRequest}. + * + * @param actions the metadata-only actions to apply atomically + * @param clusterManagerNodeTimeout timeout for reaching the cluster-manager node + * @param ackTimeout timeout for the cluster-state update to be acknowledged + * @param listener notified when the update completes + */ + public void modifyDataStream( + final List actions, + final TimeValue clusterManagerNodeTimeout, + final TimeValue ackTimeout, + final ActionListener listener + ) { + modifyDataStream(new ModifyDataStreamsClusterStateUpdateRequest(actions, clusterManagerNodeTimeout, ackTimeout), listener); + } + + public void modifyDataStream( + final ModifyDataStreamsClusterStateUpdateRequest request, + final ActionListener listener + ) { + if (request.actions.isEmpty()) { + listener.onResponse(new AcknowledgedResponse(true)); + return; + } + ActionListener wrappedListener = ActionListener.map( + listener, + response -> new AcknowledgedResponse(response.isAcknowledged()) + ); + clusterService.submitStateUpdateTask( + "update-data-streams", + new AckedClusterStateUpdateTask(Priority.HIGH, request, wrappedListener) { + @Override + public ClusterManagerTaskThrottler.ThrottlingKey getClusterManagerThrottlingKey() { + return modifyDataStreamTaskKey; + } + + @Override + public ClusterState execute(ClusterState currentState) { + return modifyDataStream(currentState, request.actions); + } + + @Override + protected ClusterStateUpdateResponse newResponse(boolean acknowledged) { + return new ClusterStateUpdateResponse(acknowledged); + } + } + ); + } + + /** + * Applies the given actions to the current cluster state and returns the updated state. Visible for testing. + */ + static ClusterState modifyDataStream(ClusterState currentState, Iterable actions) { + Metadata.Builder metadataBuilder = Metadata.builder(currentState.metadata()); + // Track streams we've mutated so multiple actions against the same stream compose within this single update. + Map updated = new LinkedHashMap<>(); + for (DataStreamAction action : actions) { + String dataStreamName = action.dataStream(); + DataStream dataStream = updated.getOrDefault(dataStreamName, currentState.metadata().dataStreams().get(dataStreamName)); + if (dataStream == null) { + throw new IllegalArgumentException("data stream [" + dataStreamName + "] not found"); + } + switch (action.type()) { + case ADD_BACKING_INDEX: + dataStream = applyAddBackingIndex(metadataBuilder, dataStream, action.index()); + break; + case REMOVE_BACKING_INDEX: + dataStream = applyRemoveBackingIndex(metadataBuilder, dataStream, action.index()); + break; + default: + throw new IllegalArgumentException("unsupported data stream action type [" + action.type() + "]"); + } + updated.put(dataStreamName, dataStream); + } + + for (DataStream dataStream : updated.values()) { + logger.info("updating data stream [{}]", dataStream.getName()); + metadataBuilder.put(dataStream); + } + validateNoSharedBackingIndices(currentState.metadata().dataStreams(), updated); + // Metadata.build() additionally rejects a stream whose prefix matches an index with a counter above its + // generation. + return ClusterState.builder(currentState).metadata(metadataBuilder).build(); + } + + /** + * Rejects a resulting state in which any index is a backing index of more than one data stream, checking the full + * final membership so the outcome does not depend on action order. + */ + private static void validateNoSharedBackingIndices(Map currentStreams, Map updatedStreams) { + Map finalStreams = new HashMap<>(currentStreams); + finalStreams.putAll(updatedStreams); + Map indexToStream = new HashMap<>(); + for (DataStream dataStream : finalStreams.values()) { + for (Index index : dataStream.getIndices()) { + String previousOwner = indexToStream.putIfAbsent(index.getName(), dataStream.getName()); + if (previousOwner != null && previousOwner.equals(dataStream.getName()) == false) { + throw new IllegalArgumentException( + "index [" + + index.getName() + + "] cannot be a backing index of more than one data stream, but is claimed by [" + + previousOwner + + "] and [" + + dataStream.getName() + + "]" + ); + } + } + } + } + + private static DataStream applyAddBackingIndex(Metadata.Builder metadataBuilder, DataStream dataStream, String indexName) { + IndexMetadata indexMetadata = metadataBuilder.get(indexName); + if (indexMetadata == null) { + throw new IllegalArgumentException("index [" + indexName + "] not found"); + } + Index index = indexMetadata.getIndex(); + if (dataStream.getIndices().contains(index)) { + // Idempotent: already a member. + return dataStream; + } + + // Data stream search relies on the timestamp field, so every attached index must map it as a date, regardless + // of the index name (a .ds--NNNNNN name is not proof the index came from create/rollover; any index can + // be created with that name and an arbitrary mapping). + validateTimestampFieldMapping(indexMetadata, dataStream.getTimeStampField().getName()); + + // Backing indices are hidden; mark the index hidden if it is not already, matching data stream creation and + // rollover. put(IndexMetadata.Builder) bumps the index metadata version; the top-level Metadata version is + // bumped by the cluster-manager service on publish. + if (IndexMetadata.INDEX_HIDDEN_SETTING.get(indexMetadata.getSettings()) == false) { + IndexMetadata.Builder hiddenIndexMetadata = IndexMetadata.builder(indexMetadata) + .settings(Settings.builder().put(indexMetadata.getSettings()).put(IndexMetadata.SETTING_INDEX_HIDDEN, true)) + .settingsVersion(indexMetadata.getSettingsVersion() + 1); + metadataBuilder.put(hiddenIndexMetadata); + index = hiddenIndexMetadata.build().getIndex(); + } + + // The added index is always a non-write index, so the generation is unchanged: an index whose counter exceeds + // the current generation cannot exist alongside a lower-generation stream (Metadata#validateDataStreams rejects + // it), and this API never creates indices. New write indices come from rollover. + return dataStream.addBackingIndex(index); + } + + /** + * Verifies the index maps the data stream's timestamp field as a date type, which data stream search requires. + */ + @SuppressWarnings("unchecked") + public static void validateTimestampFieldMapping(IndexMetadata indexMetadata, String timestampFieldName) { + MappingMetadata mapping = indexMetadata.mapping(); + Object type = null; + if (mapping != null) { + Object properties = mapping.sourceAsMap().get("properties"); + if (properties instanceof Map) { + Object field = ((Map) properties).get(timestampFieldName); + if (field instanceof Map) { + type = ((Map) field).get("type"); + } + } + } + if (DateFieldMapper.CONTENT_TYPE.equals(type) == false && DateFieldMapper.DATE_NANOS_CONTENT_TYPE.equals(type) == false) { + throw new IllegalArgumentException( + "index [" + + indexMetadata.getIndex().getName() + + "] cannot be added as a backing index because it does not have a [" + + timestampFieldName + + "] field mapped as a date type" + ); + } + } + + private static DataStream applyRemoveBackingIndex(Metadata.Builder metadataBuilder, DataStream dataStream, String indexName) { + Index toRemove = null; + for (Index index : dataStream.getIndices()) { + if (index.getName().equals(indexName)) { + toRemove = index; + break; + } + } + if (toRemove == null) { + throw new IllegalArgumentException("index [" + indexName + "] is not part of data stream [" + dataStream.getName() + "]"); + } + if (dataStream.getIndices().size() == 1) { + throw new IllegalArgumentException( + "cannot remove backing index [" + + indexName + + "] of data stream [" + + dataStream.getName() + + "] because it is the last backing index; delete the data stream instead" + ); + } + // The write index (last, highest-generation) cannot be removed: it would orphan the stream's generation. + Index writeIndex = dataStream.getIndices().get(dataStream.getIndices().size() - 1); + if (writeIndex.equals(toRemove)) { + throw new IllegalArgumentException( + "cannot remove backing index [" + + indexName + + "] of data stream [" + + dataStream.getName() + + "] because it is the write index" + ); + } + // Mirror the hidden setting applied on attach: a detached index is no longer a backing index, so make it + // visible again. If it was hidden before it was ever a backing index, the caller can hide it again. + IndexMetadata detached = metadataBuilder.get(indexName); + if (detached != null && IndexMetadata.INDEX_HIDDEN_SETTING.get(detached.getSettings())) { + metadataBuilder.put( + IndexMetadata.builder(detached) + .settings(Settings.builder().put(detached.getSettings()).put(IndexMetadata.SETTING_INDEX_HIDDEN, false)) + .settingsVersion(detached.getSettingsVersion() + 1) + ); + } + return dataStream.removeBackingIndex(toRemove); + } + + /** + * A cluster-state update request carrying the batch of data-stream actions to apply. + * + * @opensearch.internal + */ + public static final class ModifyDataStreamsClusterStateUpdateRequest extends ClusterStateUpdateRequest { + + private final List actions; + + public ModifyDataStreamsClusterStateUpdateRequest( + List actions, + TimeValue clusterManagerNodeTimeout, + TimeValue ackTimeout + ) { + this.actions = actions; + masterNodeTimeout(clusterManagerNodeTimeout); + ackTimeout(ackTimeout); + } + + public List getActions() { + return actions; + } + } +} diff --git a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTaskKeys.java b/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTaskKeys.java index c88bea56cb9bd..ce65327582df5 100644 --- a/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTaskKeys.java +++ b/server/src/main/java/org/opensearch/cluster/service/ClusterManagerTaskKeys.java @@ -21,6 +21,7 @@ public final class ClusterManagerTaskKeys { public static final String DELETE_DANGLING_INDEX_KEY = "delete-dangling-index"; public static final String CREATE_DATA_STREAM_KEY = "create-data-stream"; public static final String REMOVE_DATA_STREAM_KEY = "remove-data-stream"; + public static final String MODIFY_DATA_STREAM_KEY = "modify-data-stream"; public static final String ROLLOVER_INDEX_KEY = "rollover-index"; public static final String INDEX_ALIASES_KEY = "index-aliases"; public static final String PUT_MAPPING_KEY = "put-mapping"; diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java index b84be7742394c..d4e09edbc6496 100644 --- a/server/src/main/java/org/opensearch/node/Node.java +++ b/server/src/main/java/org/opensearch/node/Node.java @@ -76,6 +76,7 @@ import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.MetadataCreateDataStreamService; import org.opensearch.cluster.metadata.MetadataCreateIndexService; +import org.opensearch.cluster.metadata.MetadataDataStreamsService; import org.opensearch.cluster.metadata.MetadataIndexUpgradeService; import org.opensearch.cluster.metadata.SystemIndexMetadataUpgradeService; import org.opensearch.cluster.metadata.TemplateUpgradeService; @@ -996,6 +997,7 @@ protected Node( clusterService, metadataCreateIndexService ); + final MetadataDataStreamsService metadataDataStreamsService = new MetadataDataStreamsService(clusterService); Collection pluginComponents = pluginsService.filterPlugins(Plugin.class) .stream() @@ -1456,6 +1458,7 @@ protected Node( b.bind(MetadataCreateIndexService.class).toInstance(metadataCreateIndexService); b.bind(AwarenessReplicaBalance.class).toInstance(awarenessReplicaBalance); b.bind(MetadataCreateDataStreamService.class).toInstance(metadataCreateDataStreamService); + b.bind(MetadataDataStreamsService.class).toInstance(metadataDataStreamsService); b.bind(SearchService.class).toInstance(searchService); b.bind(SearchTransportService.class).toInstance(searchTransportService); b.bind(SearchPhaseController.class) 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..18190bd7c1dc7 --- /dev/null +++ b/server/src/main/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsAction.java @@ -0,0 +1,73 @@ +/* + * 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.rest.action.admin.indices; + +import org.opensearch.action.admin.indices.datastream.DataStreamAction; +import org.opensearch.action.admin.indices.datastream.ModifyDataStreamsAction; +import org.opensearch.client.node.NodeClient; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * REST action to add or remove 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 Collections.singletonList(new Route(RestRequest.Method.POST, "/_data_stream/_modify")); + } + + @Override + protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException { + List actions; + try (XContentParser parser = request.contentParser()) { + actions = parseActions(parser); + } + ModifyDataStreamsAction.Request modifyRequest = new ModifyDataStreamsAction.Request(actions); + modifyRequest.clusterManagerNodeTimeout(request.paramAsTime("cluster_manager_timeout", modifyRequest.clusterManagerNodeTimeout())); + modifyRequest.timeout(request.paramAsTime("timeout", modifyRequest.timeout())); + return channel -> client.execute(ModifyDataStreamsAction.INSTANCE, modifyRequest, new RestToXContentListener<>(channel)); + } + + private static List parseActions(XContentParser parser) throws IOException { + List actions = new ArrayList<>(); + XContentParser.Token token = parser.nextToken(); + if (token != XContentParser.Token.START_OBJECT) { + throw new IllegalArgumentException("expected an object with an [actions] array"); + } + String currentFieldName = null; + while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { + if (token == XContentParser.Token.FIELD_NAME) { + currentFieldName = parser.currentName(); + } else if ("actions".equals(currentFieldName) && token == XContentParser.Token.START_ARRAY) { + while (parser.nextToken() != XContentParser.Token.END_ARRAY) { + actions.add(DataStreamAction.fromXContent(parser)); + } + } else { + throw new IllegalArgumentException("unexpected field [" + currentFieldName + "], only [actions] is supported"); + } + } + return actions; + } +} diff --git a/server/src/main/java/org/opensearch/snapshots/RestoreService.java b/server/src/main/java/org/opensearch/snapshots/RestoreService.java index 0fbd5fcbfdee4..640b00d6ce0f2 100644 --- a/server/src/main/java/org/opensearch/snapshots/RestoreService.java +++ b/server/src/main/java/org/opensearch/snapshots/RestoreService.java @@ -58,6 +58,7 @@ import org.opensearch.cluster.metadata.IndexTemplateMetadata; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.MetadataCreateIndexService; +import org.opensearch.cluster.metadata.MetadataDataStreamsService; import org.opensearch.cluster.metadata.MetadataIndexStateService; import org.opensearch.cluster.metadata.MetadataIndexUpgradeService; import org.opensearch.cluster.metadata.RepositoriesMetadata; @@ -630,6 +631,9 @@ public ClusterState execute(ClusterState currentState) { .map(ds -> updateDataStream(ds, mdBuilder, request)) .collect(Collectors.toMap(DataStream::getName, Function.identity())) ); + if (request.attachToDataStream()) { + attachRestoredBackingIndices(indices.keySet(), mdBuilder, updatedDataStreams); + } mdBuilder.dataStreams(updatedDataStreams); // Restore global state if needed @@ -977,6 +981,36 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS } } + /** + * Attaches each restored {@code .ds--NNNNNN} index to a pre-existing data stream of the same name, in + * the same cluster-state update as the restore. Adding the index and advancing the generation together avoids the + * transient state {@link Metadata.Builder} validation rejects (a convention-named index above the generation). + * Updates {@code updatedDataStreams} in place. Visible for testing. + */ + static void attachRestoredBackingIndices( + Set restoredIndexNames, + Metadata.Builder metadata, + Map updatedDataStreams + ) { + for (String restoredIndexName : restoredIndexNames) { + String streamName = DataStream.parseDataStreamName(restoredIndexName); + if (streamName == null) { + continue; + } + DataStream currentDs = updatedDataStreams.get(streamName); + IndexMetadata restoredIndexMetadata = metadata.get(restoredIndexName); + if (currentDs == null || restoredIndexMetadata == null) { + continue; + } + if (currentDs.getIndices().contains(restoredIndexMetadata.getIndex())) { + continue; + } + // A backing index must map the timestamp field as a date, or data stream search breaks. + MetadataDataStreamsService.validateTimestampFieldMapping(restoredIndexMetadata, currentDs.getTimeStampField().getName()); + updatedDataStreams.put(streamName, currentDs.addBackingIndex(restoredIndexMetadata.getIndex())); + } + } + // visible for testing static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metadata, RestoreSnapshotRequest request) { String dataStreamName = dataStream.getName(); diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilderTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilderTests.java new file mode 100644 index 0000000000000..c63ae79f7a4a8 --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilderTests.java @@ -0,0 +1,51 @@ +/* + * 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.cluster.snapshots.restore; + +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.client.NoOpClient; +import org.junit.After; +import org.junit.Before; + +public class RestoreSnapshotRequestBuilderTests extends OpenSearchTestCase { + + private NoOpClient testClient; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + this.testClient = new NoOpClient(getTestName()); + } + + @Override + @After + public void tearDown() throws Exception { + this.testClient.close(); + super.tearDown(); + } + + public void testSetAttachToDataStream() { + RestoreSnapshotRequestBuilder builder = new RestoreSnapshotRequestBuilder( + this.testClient, + RestoreSnapshotAction.INSTANCE, + "repo", + "snap" + ); + assertFalse(builder.request().attachToDataStream()); + + RestoreSnapshotRequestBuilder returned = builder.setAttachToDataStream(true); + // The setter is fluent and mutates the wrapped request. + assertSame(builder, returned); + assertTrue(builder.request().attachToDataStream()); + + builder.setAttachToDataStream(false); + assertFalse(builder.request().attachToDataStream()); + } +} diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestTests.java index 04cc45f3477c6..ef9bee1c871c9 100644 --- a/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestTests.java +++ b/server/src/test/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestTests.java @@ -32,9 +32,12 @@ package org.opensearch.action.admin.cluster.snapshots.restore; +import org.opensearch.Version; import org.opensearch.action.support.IndicesOptions; +import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.xcontent.XContentFactory; import org.opensearch.core.common.bytes.BytesReference; +import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.common.io.stream.Writeable; import org.opensearch.core.xcontent.MediaTypeRegistry; import org.opensearch.core.xcontent.NamedXContentRegistry; @@ -126,6 +129,8 @@ private RestoreSnapshotRequest randomState(RestoreSnapshotRequest instance) { instance.setSourceRemoteStoreRepository(randomAlphaOfLengthBetween(5, 10)); } + instance.attachToDataStream(randomBoolean()); + return instance; } @@ -166,4 +171,72 @@ public void testSource() throws IOException { assertEquals(original, processed); } + + public void testSourceParsesAttachToDataStream() throws IOException { + // Explicitly drive the "attach_to_data_stream" branch of source(Map) in both directions. + RestoreSnapshotRequest request = new RestoreSnapshotRequest("repo", "snap"); + assertFalse(request.attachToDataStream()); + + Map source = new HashMap<>(); + source.put("attach_to_data_stream", true); + request.source(source); + assertTrue(request.attachToDataStream()); + + source.put("attach_to_data_stream", "false"); + request.source(source); + assertFalse(request.attachToDataStream()); + } + + public void testToXContentAlwaysEmitsAttachToDataStream() throws IOException { + for (boolean attach : new boolean[] { true, false }) { + RestoreSnapshotRequest request = new RestoreSnapshotRequest("repo", "snap").attachToDataStream(attach); + XContentBuilder builder = request.toXContent(XContentFactory.jsonBuilder(), new ToXContent.MapParams(Collections.emptyMap())); + XContentParser parser = MediaTypeRegistry.JSON.xContent() + .createParser(NamedXContentRegistry.EMPTY, null, BytesReference.bytes(builder).streamInput()); + Map map = parser.mapOrdered(); + assertEquals(attach, map.get("attach_to_data_stream")); + } + } + + public void testEqualsAndHashCodeDifferOnlyOnAttachToDataStream() { + RestoreSnapshotRequest off = new RestoreSnapshotRequest("repo", "snap").attachToDataStream(false); + RestoreSnapshotRequest on = new RestoreSnapshotRequest("repo", "snap").attachToDataStream(true); + + // Everything else is identical, so the equals() chain reaches the attachToDataStream comparison and it decides. + assertNotEquals(off, on); + assertNotEquals(off.hashCode(), on.hashCode()); + + on.attachToDataStream(false); + assertEquals(off, on); + assertEquals(off.hashCode(), on.hashCode()); + } + + public void testAttachToDataStreamNotSentToOlderNodes() throws IOException { + // The wire guard is Version.V_2_19_7; AbstractWireSerializingTestCase pins its round trip at Version.CURRENT + // (== V_2_19_7), so the "older node" side of the guard is only reachable through an explicit stream version. + RestoreSnapshotRequest request = new RestoreSnapshotRequest("repo", "snap").attachToDataStream(true); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + out.setVersion(Version.V_2_19_6); + request.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + in.setVersion(Version.V_2_19_6); + RestoreSnapshotRequest read = new RestoreSnapshotRequest(in); + // The flag is dropped on the wire, so an older node falls back to the default. + assertFalse(read.attachToDataStream()); + assertEquals("repo", read.repository()); + assertEquals("snap", read.snapshot()); + } + } + + // Same request at the current version keeps the flag. + try (BytesStreamOutput out = new BytesStreamOutput()) { + out.setVersion(Version.V_2_19_7); + request.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + in.setVersion(Version.V_2_19_7); + assertTrue(new RestoreSnapshotRequest(in).attachToDataStream()); + } + } + } } 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..cb263a979134b --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/DataStreamActionTests.java @@ -0,0 +1,326 @@ +/* + * 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.io.stream.BytesStreamOutput; +import org.opensearch.common.xcontent.XContentFactory; +import org.opensearch.common.xcontent.XContentType; +import org.opensearch.core.common.bytes.BytesReference; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.core.xcontent.XContentParseException; +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; +import static org.hamcrest.Matchers.not; + +public class DataStreamActionTests extends OpenSearchTestCase { + + private static final String DS = "logs-foo"; + private static final String INDEX = ".ds-logs-foo-000001"; + + // ---------------------------------------------------------------- Type enum + + public void testTypeValue() { + assertThat(DataStreamAction.Type.ADD_BACKING_INDEX.value(), equalTo((byte) 0)); + assertThat(DataStreamAction.Type.REMOVE_BACKING_INDEX.value(), equalTo((byte) 1)); + } + + public void testTypeFieldName() { + assertThat(DataStreamAction.Type.ADD_BACKING_INDEX.fieldName(), equalTo("add_backing_index")); + assertThat(DataStreamAction.Type.REMOVE_BACKING_INDEX.fieldName(), equalTo("remove_backing_index")); + } + + public void testTypeFromValueValid() { + assertThat(DataStreamAction.Type.fromValue((byte) 0), equalTo(DataStreamAction.Type.ADD_BACKING_INDEX)); + assertThat(DataStreamAction.Type.fromValue((byte) 1), equalTo(DataStreamAction.Type.REMOVE_BACKING_INDEX)); + } + + public void testTypeFromValueInvalidThrows() { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.Type.fromValue((byte) 2)); + assertThat(e.getMessage(), equalTo("no data stream action type for [2]")); + + IllegalArgumentException negative = expectThrows(IllegalArgumentException.class, () -> DataStreamAction.Type.fromValue((byte) -1)); + assertThat(negative.getMessage(), equalTo("no data stream action type for [-1]")); + } + + public void testTypeValuesAreRoundTrippableThroughFromValue() { + for (DataStreamAction.Type type : DataStreamAction.Type.values()) { + assertThat(DataStreamAction.Type.fromValue(type.value()), equalTo(type)); + } + } + + // ---------------------------------------------------------------- constructor null checks + + public void testConstructorRejectsNullType() { + NullPointerException e = expectThrows(NullPointerException.class, () -> new DataStreamAction(null, DS, INDEX)); + assertThat(e.getMessage(), equalTo("[type] is required")); + } + + public void testConstructorRejectsNullDataStream() { + NullPointerException e = expectThrows( + NullPointerException.class, + () -> new DataStreamAction(DataStreamAction.Type.ADD_BACKING_INDEX, null, INDEX) + ); + assertThat(e.getMessage(), equalTo("[data_stream] is required")); + } + + public void testConstructorRejectsNullIndex() { + NullPointerException e = expectThrows( + NullPointerException.class, + () -> new DataStreamAction(DataStreamAction.Type.REMOVE_BACKING_INDEX, DS, null) + ); + assertThat(e.getMessage(), equalTo("[index] is required")); + } + + // ---------------------------------------------------------------- accessors + factories + + public void testAccessors() { + DataStreamAction action = new DataStreamAction(DataStreamAction.Type.REMOVE_BACKING_INDEX, DS, INDEX); + assertThat(action.type(), equalTo(DataStreamAction.Type.REMOVE_BACKING_INDEX)); + assertThat(action.dataStream(), equalTo(DS)); + assertThat(action.index(), equalTo(INDEX)); + } + + public void testAddBackingIndexFactory() { + DataStreamAction action = DataStreamAction.addBackingIndex(DS, INDEX); + assertThat(action.type(), equalTo(DataStreamAction.Type.ADD_BACKING_INDEX)); + assertThat(action.dataStream(), equalTo(DS)); + assertThat(action.index(), equalTo(INDEX)); + assertThat(action, equalTo(new DataStreamAction(DataStreamAction.Type.ADD_BACKING_INDEX, DS, INDEX))); + } + + public void testRemoveBackingIndexFactory() { + DataStreamAction action = DataStreamAction.removeBackingIndex(DS, INDEX); + assertThat(action.type(), equalTo(DataStreamAction.Type.REMOVE_BACKING_INDEX)); + assertThat(action.dataStream(), equalTo(DS)); + assertThat(action.index(), equalTo(INDEX)); + assertThat(action, equalTo(new DataStreamAction(DataStreamAction.Type.REMOVE_BACKING_INDEX, DS, INDEX))); + } + + // ---------------------------------------------------------------- wire serialization + + public void testSerializationRoundTripAdd() throws IOException { + assertSerializationRoundTrip(DataStreamAction.addBackingIndex(DS, INDEX)); + } + + public void testSerializationRoundTripRemove() throws IOException { + assertSerializationRoundTrip(DataStreamAction.removeBackingIndex(DS, INDEX)); + } + + public void testSerializationRoundTripRandom() throws IOException { + for (DataStreamAction.Type type : DataStreamAction.Type.values()) { + String ds = randomAlphaOfLength(8); + String index = randomAlphaOfLength(12); + assertSerializationRoundTrip(new DataStreamAction(type, ds, index)); + } + } + + public void testStreamInputRejectsUnknownTypeByte() throws IOException { + try (BytesStreamOutput out = new BytesStreamOutput()) { + out.writeByte((byte) 7); + out.writeString(DS); + out.writeString(INDEX); + try (StreamInput in = out.bytes().streamInput()) { + IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new DataStreamAction(in)); + assertThat(e.getMessage(), equalTo("no data stream action type for [7]")); + } + } + } + + private void assertSerializationRoundTrip(DataStreamAction original) throws IOException { + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + DataStreamAction deserialized = new DataStreamAction(in); + assertThat(deserialized.type(), equalTo(original.type())); + assertThat(deserialized.dataStream(), equalTo(original.dataStream())); + assertThat(deserialized.index(), equalTo(original.index())); + assertThat(deserialized, equalTo(original)); + assertThat(deserialized.hashCode(), equalTo(original.hashCode())); + } + } + } + + // ---------------------------------------------------------------- toXContent + + public void testToXContentAdd() throws IOException { + assertThat( + toJson(DataStreamAction.addBackingIndex(DS, INDEX)), + equalTo("{\"add_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\"}}") + ); + } + + public void testToXContentRemove() throws IOException { + assertThat( + toJson(DataStreamAction.removeBackingIndex(DS, INDEX)), + equalTo("{\"remove_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\"}}") + ); + } + + private String toJson(DataStreamAction action) throws IOException { + try (XContentBuilder builder = XContentFactory.jsonBuilder()) { + action.toXContent(builder, ToXContent.EMPTY_PARAMS); + return BytesReference.bytes(builder).utf8ToString(); + } + } + + // ---------------------------------------------------------------- fromXContent + + public void testFromXContentAdd() throws IOException { + String json = "{\"add_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + DataStreamAction action = DataStreamAction.fromXContent(parser); + assertThat(action, equalTo(DataStreamAction.addBackingIndex(DS, INDEX))); + } + } + + public void testFromXContentRemove() throws IOException { + String json = "{\"remove_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + DataStreamAction action = DataStreamAction.fromXContent(parser); + assertThat(action, equalTo(DataStreamAction.removeBackingIndex(DS, INDEX))); + } + } + + public void testXContentRoundTrip() throws IOException { + for (DataStreamAction.Type type : DataStreamAction.Type.values()) { + DataStreamAction original = new DataStreamAction(type, randomAlphaOfLength(6), randomAlphaOfLength(10)); + try (XContentParser parser = createParser(XContentType.JSON.xContent(), toJson(original))) { + assertThat(DataStreamAction.fromXContent(parser), equalTo(original)); + } + } + } + + public void testFromXContentUnknownActionNameThrows() throws IOException { + String json = "{\"rotate_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + XContentParseException e = expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(e.getMessage(), containsString("rotate_backing_index")); + } + } + + public void testFromXContentMissingIndexThrows() throws IOException { + String json = "{\"add_backing_index\":{\"data_stream\":\"" + DS + "\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + XContentParseException e = expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(e.getMessage(), containsString("index")); + } + } + + public void testFromXContentMissingDataStreamThrows() throws IOException { + String json = "{\"remove_backing_index\":{\"index\":\"" + INDEX + "\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + XContentParseException e = expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(e.getMessage(), containsString("data_stream")); + } + } + + public void testFromXContentUnknownInnerFieldThrows() throws IOException { + String json = "{\"add_backing_index\":{\"data_stream\":\"" + DS + "\",\"index\":\"" + INDEX + "\",\"bogus\":\"value\"}}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + XContentParseException e = expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(e.getMessage(), containsString("add_backing_index")); + assertNotNull(e.getCause()); + assertThat(e.getCause().getMessage(), containsString("bogus")); + } + } + + public void testFromXContentNotAnObjectThrows() throws IOException { + try (XContentParser parser = createParser(XContentType.JSON.xContent(), "[]")) { + XContentParseException e = expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + assertThat(e.getMessage(), containsString("Expected START_OBJECT")); + } + } + + public void testFromXContentEmptyObjectYieldsNoAction() throws IOException { + try (XContentParser parser = createParser(XContentType.JSON.xContent(), "{}")) { + assertNull(DataStreamAction.fromXContent(parser)); + } + } + + public void testFromXContentWrongInnerTypeThrows() throws IOException { + String json = "{\"add_backing_index\":\"not-an-object\"}"; + try (XContentParser parser = createParser(XContentType.JSON.xContent(), json)) { + expectThrows(XContentParseException.class, () -> DataStreamAction.fromXContent(parser)); + } + } + + // ---------------------------------------------------------------- equals / hashCode / toString + + public void testEqualsSameInstance() { + DataStreamAction action = DataStreamAction.addBackingIndex(DS, INDEX); + assertTrue(action.equals(action)); + } + + public void testEqualsNull() { + DataStreamAction action = DataStreamAction.addBackingIndex(DS, INDEX); + assertFalse(action.equals(null)); + } + + public void testEqualsDifferentClass() { + DataStreamAction action = DataStreamAction.addBackingIndex(DS, INDEX); + assertFalse(action.equals("add_backing_index")); + assertFalse(action.equals(new Object())); + } + + public void testEqualsEquivalentInstances() { + DataStreamAction one = DataStreamAction.addBackingIndex(DS, INDEX); + DataStreamAction two = new DataStreamAction(DataStreamAction.Type.ADD_BACKING_INDEX, new String(DS), new String(INDEX)); + assertNotSame(one, two); + assertTrue(one.equals(two)); + assertTrue(two.equals(one)); + assertThat(one.hashCode(), equalTo(two.hashCode())); + } + + public void testNotEqualsOnType() { + DataStreamAction add = DataStreamAction.addBackingIndex(DS, INDEX); + DataStreamAction remove = DataStreamAction.removeBackingIndex(DS, INDEX); + assertFalse(add.equals(remove)); + assertThat(add.hashCode(), not(equalTo(remove.hashCode()))); + } + + public void testNotEqualsOnDataStream() { + DataStreamAction one = DataStreamAction.addBackingIndex(DS, INDEX); + DataStreamAction two = DataStreamAction.addBackingIndex("logs-other", INDEX); + assertFalse(one.equals(two)); + assertThat(one.hashCode(), not(equalTo(two.hashCode()))); + } + + public void testNotEqualsOnIndex() { + DataStreamAction one = DataStreamAction.addBackingIndex(DS, INDEX); + DataStreamAction two = DataStreamAction.addBackingIndex(DS, ".ds-logs-foo-000002"); + assertFalse(one.equals(two)); + assertThat(one.hashCode(), not(equalTo(two.hashCode()))); + } + + public void testHashCodeIsStable() { + DataStreamAction action = DataStreamAction.removeBackingIndex(DS, INDEX); + assertThat(action.hashCode(), equalTo(action.hashCode())); + } + + public void testToStringAdd() { + assertThat( + DataStreamAction.addBackingIndex(DS, INDEX).toString(), + equalTo("DataStreamAction[type=ADD_BACKING_INDEX, dataStream=" + DS + ", index=" + INDEX + "]") + ); + } + + public void testToStringRemove() { + assertThat( + DataStreamAction.removeBackingIndex(DS, INDEX).toString(), + equalTo("DataStreamAction[type=REMOVE_BACKING_INDEX, dataStream=" + DS + ", index=" + INDEX + "]") + ); + } +} 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..3e6e3e963f38d --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsRequestTests.java @@ -0,0 +1,198 @@ +/* + * 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.support.master.AcknowledgedRequest; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.common.io.stream.BytesStreamOutput; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.test.OpenSearchTestCase; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.not; + +public class ModifyDataStreamsRequestTests extends OpenSearchTestCase { + + private DataStreamAction randomAction() { + String ds = randomAlphaOfLength(6).toLowerCase(java.util.Locale.ROOT); + String index = ".ds-" + ds + "-" + String.format(java.util.Locale.ROOT, "%06d", randomIntBetween(1, 1000)); + return randomBoolean() ? DataStreamAction.addBackingIndex(ds, index) : DataStreamAction.removeBackingIndex(ds, index); + } + + public void testActionSerializationRoundTrip() throws IOException { + DataStreamAction original = randomAction(); + DataStreamAction deserialized = copy(original); + assertThat(deserialized, equalTo(original)); + assertThat(deserialized.hashCode(), equalTo(original.hashCode())); + } + + public void testRequestSerializationRoundTrip() throws IOException { + List actions = new ArrayList<>(); + int count = randomIntBetween(1, 5); + for (int i = 0; i < count; i++) { + actions.add(randomAction()); + } + ModifyDataStreamsAction.Request original = new ModifyDataStreamsAction.Request(actions); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ModifyDataStreamsAction.Request deserialized = new ModifyDataStreamsAction.Request(in); + assertThat(deserialized, equalTo(original)); + assertThat(deserialized.getActions(), equalTo(original.getActions())); + } + } + } + + public void testEmptyRequestFailsValidation() { + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(new ArrayList<>()); + assertNotNull(request.validate()); + } + + /** + * The empty-request validation failure carries the documented message. Covers the {@code actions.isEmpty()} true + * branch of Request#validate(). + */ + public void testEmptyRequestValidationMessage() { + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(Collections.emptyList()); + ActionRequestValidationException e = request.validate(); + assertNotNull(e); + assertThat(e.validationErrors(), hasSize(1)); + assertThat(e.validationErrors().get(0), containsString("at least one data stream action must be specified")); + } + + /** + * A request with at least one action validates cleanly. Covers the {@code actions.isEmpty()} false branch, i.e. + * the {@code return null} path of Request#validate(). + */ + public void testNonEmptyRequestPassesValidation() { + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request( + Collections.singletonList(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001")) + ); + assertNull(request.validate()); + } + + /** + * The constructor rejects a null action list, so {@code actions == null} can never be observed by validate(). + */ + public void testNullActionListRejectedByConstructor() { + final List nullActions = null; + NullPointerException e = expectThrows(NullPointerException.class, () -> new ModifyDataStreamsAction.Request(nullActions)); + assertThat(e.getMessage(), containsString("actions must not be null")); + } + + /** + * The constructor defensively copies the caller's list: later mutation of the source list is not visible. + */ + public void testConstructorCopiesActionList() { + List source = new ArrayList<>(); + source.add(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001")); + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(source); + + source.add(DataStreamAction.removeBackingIndex("logs-foo", ".ds-logs-foo-000002")); + + assertThat(request.getActions(), hasSize(1)); + assertThat(request.getActions(), contains(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001"))); + } + + /** + * Drives every branch of Request#equals(Object): identity, null, foreign class, unequal actions and equal actions. + */ + public void testEqualsAndHashCode() { + DataStreamAction add = DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001"); + DataStreamAction remove = DataStreamAction.removeBackingIndex("logs-foo", ".ds-logs-foo-000002"); + + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(Collections.singletonList(add)); + ModifyDataStreamsAction.Request same = new ModifyDataStreamsAction.Request(Collections.singletonList(add)); + ModifyDataStreamsAction.Request different = new ModifyDataStreamsAction.Request(Collections.singletonList(remove)); + ModifyDataStreamsAction.Request longer = new ModifyDataStreamsAction.Request(Arrays.asList(add, remove)); + + // this == o + assertTrue(request.equals(request)); + // o == null + assertFalse(request.equals(null)); + // getClass() != o.getClass() + assertFalse(request.equals("not a request")); + // actions.equals(...) == false, same size + assertThat(request, not(equalTo(different))); + // actions.equals(...) == false, different size + assertThat(request, not(equalTo(longer))); + // actions.equals(...) == true + assertThat(request, equalTo(same)); + assertThat(request.hashCode(), equalTo(same.hashCode())); + } + + /** + * Non-default acknowledgement and cluster-manager timeouts survive the wire round trip, exercising the + * {@code super.writeTo(out)} / {@code super(in)} halves of the request serialization. + */ + public void testTimeoutsSurviveSerialization() throws IOException { + ModifyDataStreamsAction.Request original = new ModifyDataStreamsAction.Request( + Collections.singletonList(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001")) + ); + original.timeout(TimeValue.timeValueSeconds(7)); + original.clusterManagerNodeTimeout(TimeValue.timeValueSeconds(11)); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ModifyDataStreamsAction.Request deserialized = new ModifyDataStreamsAction.Request(in); + assertThat(deserialized.timeout(), equalTo(TimeValue.timeValueSeconds(7))); + assertThat(deserialized.clusterManagerNodeTimeout(), equalTo(TimeValue.timeValueSeconds(11))); + assertThat(deserialized, equalTo(original)); + } + } + } + + /** + * A freshly built request carries the standard acknowledged-request default ack timeout. + */ + public void testDefaultTimeouts() { + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request( + Collections.singletonList(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001")) + ); + assertThat(request.timeout(), equalTo(AcknowledgedRequest.DEFAULT_ACK_TIMEOUT)); + } + + /** + * The action singleton exposes the registered transport name and a reader that yields an AcknowledgedResponse. + */ + public void testActionTypeSingleton() throws IOException { + assertThat(ModifyDataStreamsAction.NAME, equalTo("indices:admin/data_stream/modify")); + assertThat(ModifyDataStreamsAction.INSTANCE.name(), equalTo(ModifyDataStreamsAction.NAME)); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + new AcknowledgedResponse(true).writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + AcknowledgedResponse response = ModifyDataStreamsAction.INSTANCE.getResponseReader().read(in); + assertTrue(response.isAcknowledged()); + } + } + } + + private DataStreamAction copy(DataStreamAction action) throws IOException { + try (BytesStreamOutput out = new BytesStreamOutput()) { + action.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + return new DataStreamAction(in); + } + } + } +} diff --git a/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsTransportActionTests.java b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsTransportActionTests.java new file mode 100644 index 0000000000000..701400c5a63d5 --- /dev/null +++ b/server/src/test/java/org/opensearch/action/admin/indices/datastream/ModifyDataStreamsTransportActionTests.java @@ -0,0 +1,247 @@ +/* + * 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.support.ActionFilters; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.block.ClusterBlockException; +import org.opensearch.cluster.block.ClusterBlocks; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.metadata.MetadataDataStreamsService; +import org.opensearch.cluster.metadata.MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.io.stream.BytesStreamOutput; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.concurrent.ThreadContext; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.telemetry.tracing.noop.NoopTracer; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.transport.CapturingTransport; +import org.opensearch.threadpool.TestThreadPool; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; +import org.junit.After; +import org.junit.Before; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.opensearch.test.ClusterServiceUtils.createClusterService; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.notNullValue; + +/** + * Unit tests for the transport half of {@link ModifyDataStreamsAction}: executor choice, response reading, + * cluster-block handling and delegation to {@link MetadataDataStreamsService}. + */ +public class ModifyDataStreamsTransportActionTests extends OpenSearchTestCase { + + private ThreadPool threadPool; + private ClusterService clusterService; + private TransportService transportService; + private CapturingMetadataDataStreamsService dataStreamsService; + private ModifyDataStreamsAction.TransportAction action; + + /** + * Records the update request handed to the service so the test can assert on delegation without running a real + * cluster-state update, and lets each test choose whether the service succeeds or fails. + */ + private static class CapturingMetadataDataStreamsService extends MetadataDataStreamsService { + + private final AtomicReference captured = new AtomicReference<>(); + private volatile boolean acknowledged = true; + private volatile RuntimeException failure = null; + + CapturingMetadataDataStreamsService(ClusterService clusterService) { + super(clusterService); + } + + @Override + public void modifyDataStream(ModifyDataStreamsClusterStateUpdateRequest request, ActionListener listener) { + captured.set(request); + if (failure != null) { + listener.onFailure(failure); + } else { + listener.onResponse(new AcknowledgedResponse(acknowledged)); + } + } + } + + @Before + public void setUp() throws Exception { + super.setUp(); + threadPool = new TestThreadPool("ModifyDataStreamsTransportActionTests"); + clusterService = createClusterService(threadPool); + CapturingTransport capturingTransport = new CapturingTransport(); + transportService = capturingTransport.createTransportService( + clusterService.getSettings(), + threadPool, + TransportService.NOOP_TRANSPORT_INTERCEPTOR, + boundAddress -> clusterService.localNode(), + null, + Collections.emptySet(), + NoopTracer.INSTANCE + ); + transportService.start(); + transportService.acceptIncomingRequests(); + + dataStreamsService = new CapturingMetadataDataStreamsService(clusterService); + action = new ModifyDataStreamsAction.TransportAction( + transportService, + clusterService, + threadPool, + new ActionFilters(Collections.emptySet()), + new IndexNameExpressionResolver(new ThreadContext(clusterService.getSettings())), + dataStreamsService + ); + } + + @After + public void tearDown() throws Exception { + super.tearDown(); + clusterService.close(); + transportService.close(); + ThreadPool.terminate(threadPool, 30, TimeUnit.SECONDS); + } + + /** + * The action is metadata-only and cheap, so it runs on the calling thread rather than a dedicated pool. + */ + public void testExecutorIsSame() { + assertThat(action.executor(), equalTo(ThreadPool.Names.SAME)); + } + + /** + * read() reconstructs an acknowledged response from the wire, for both acknowledged values. + */ + public void testReadResponse() throws IOException { + for (boolean acknowledged : new boolean[] { true, false }) { + try (BytesStreamOutput out = new BytesStreamOutput()) { + new AcknowledgedResponse(acknowledged).writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + AcknowledgedResponse response = action.read(in); + assertThat(response.isAcknowledged(), equalTo(acknowledged)); + } + } + } + } + + /** + * With no global blocks in place checkBlock returns null, so the operation proceeds. + */ + public void testCheckBlockPassesWithoutBlocks() { + ClusterState state = ClusterState.builder(ClusterName.DEFAULT).build(); + assertNull(action.checkBlock(newRequest(), state)); + } + + /** + * A global METADATA_WRITE block (cluster read-only) is surfaced as a ClusterBlockException. + */ + public void testCheckBlockRejectsMetadataWriteBlock() { + ClusterState state = ClusterState.builder(ClusterName.DEFAULT) + .blocks(ClusterBlocks.builder().addGlobalBlock(Metadata.CLUSTER_READ_ONLY_BLOCK).build()) + .build(); + ClusterBlockException e = action.checkBlock(newRequest(), state); + assertThat(e, notNullValue()); + assertThat(e.getMessage(), containsString("cluster read-only (api)")); + } + + /** + * A global block that does not include METADATA_WRITE (create-index only) does not block this action. + */ + public void testCheckBlockIgnoresUnrelatedBlock() { + ClusterState state = ClusterState.builder(ClusterName.DEFAULT) + .blocks(ClusterBlocks.builder().addGlobalBlock(Metadata.CLUSTER_CREATE_INDEX_BLOCK).build()) + .build(); + assertNull(action.checkBlock(newRequest(), state)); + } + + /** + * clusterManagerOperation forwards the request's actions and both timeouts to the metadata service, and passes the + * service's acknowledgement back to the listener. + */ + public void testClusterManagerOperationDelegatesToService() { + List actions = Arrays.asList( + DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001"), + DataStreamAction.removeBackingIndex("logs-bar", ".ds-logs-bar-000002") + ); + ModifyDataStreamsAction.Request request = new ModifyDataStreamsAction.Request(actions); + request.timeout(TimeValue.timeValueSeconds(13)); + request.clusterManagerNodeTimeout(TimeValue.timeValueSeconds(17)); + + AtomicReference responseRef = new AtomicReference<>(); + action.clusterManagerOperation( + request, + ClusterState.builder(ClusterName.DEFAULT).build(), + ActionListener.wrap(responseRef::set, e -> { + throw new AssertionError("unexpected failure", e); + }) + ); + + ModifyDataStreamsClusterStateUpdateRequest captured = dataStreamsService.captured.get(); + assertThat(captured, notNullValue()); + assertThat(captured.getActions(), contains(actions.toArray())); + assertThat(captured.ackTimeout(), equalTo(TimeValue.timeValueSeconds(13))); + assertThat(captured.masterNodeTimeout(), equalTo(TimeValue.timeValueSeconds(17))); + + assertThat(responseRef.get(), notNullValue()); + assertTrue(responseRef.get().isAcknowledged()); + } + + /** + * A non-acknowledged service result is relayed unchanged rather than being coerced to true. + */ + public void testClusterManagerOperationRelaysNotAcknowledged() { + dataStreamsService.acknowledged = false; + + AtomicReference responseRef = new AtomicReference<>(); + action.clusterManagerOperation( + newRequest(), + ClusterState.builder(ClusterName.DEFAULT).build(), + ActionListener.wrap(responseRef::set, e -> { + throw new AssertionError("unexpected failure", e); + }) + ); + + assertThat(responseRef.get(), notNullValue()); + assertFalse(responseRef.get().isAcknowledged()); + } + + /** + * A failure raised by the metadata service reaches the caller's listener untouched. + */ + public void testClusterManagerOperationPropagatesFailure() { + dataStreamsService.failure = new IllegalArgumentException("data stream [logs-foo] not found"); + + AtomicReference failureRef = new AtomicReference<>(); + action.clusterManagerOperation(newRequest(), ClusterState.builder(ClusterName.DEFAULT).build(), ActionListener.wrap(r -> { + throw new AssertionError("expected failure but got " + r); + }, failureRef::set)); + + assertThat(failureRef.get(), instanceOf(IllegalArgumentException.class)); + assertThat(failureRef.get().getMessage(), containsString("data stream [logs-foo] not found")); + } + + private static ModifyDataStreamsAction.Request newRequest() { + return new ModifyDataStreamsAction.Request( + Collections.singletonList(DataStreamAction.addBackingIndex("logs-foo", ".ds-logs-foo-000001")) + ); + } +} 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..dbb771538a650 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,86 @@ public void testRemoveBackingIndex() { } } + public void testAddBackingIndexAdvancesGenerationForHigherCounter() { + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + List indices = new ArrayList<>(); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 1), UUIDs.randomBase64UUID(random()))); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 2), UUIDs.randomBase64UUID(random()))); + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, 2); + + Index next = new Index(getDefaultBackingIndexName(dataStreamName, 3), UUIDs.randomBase64UUID(random())); + DataStream updated = original.addBackingIndex(next); + + assertThat(updated.getGeneration(), equalTo(3L)); + // The higher-counter index becomes the write (last) index. + assertThat(updated.getIndices().get(updated.getIndices().size() - 1), equalTo(next)); + } + + public void testAddBackingIndexReordersAndKeepsGenerationForLowerCounter() { + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + List indices = new ArrayList<>(); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 1), UUIDs.randomBase64UUID(random()))); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 3), UUIDs.randomBase64UUID(random()))); + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, 3); + + Index gen2 = new Index(getDefaultBackingIndexName(dataStreamName, 2), UUIDs.randomBase64UUID(random())); + DataStream updated = original.addBackingIndex(gen2); + + // Generation unchanged; the re-added index lands in counter order, not at the end. + assertThat(updated.getGeneration(), equalTo(3L)); + assertThat(updated.getIndices().get(1), equalTo(gen2)); + assertThat(updated.getIndices().get(2).getName(), equalTo(getDefaultBackingIndexName(dataStreamName, 3))); + } + + public void testAddArbitraryNamedBackingIndexSortsFirstAndKeepsGeneration() { + String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT); + List indices = new ArrayList<>(); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 1), UUIDs.randomBase64UUID(random()))); + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, 1); + + Index legacy = new Index("legacy-index", UUIDs.randomBase64UUID(random())); + DataStream updated = original.addBackingIndex(legacy); + + assertThat(updated.getGeneration(), equalTo(1L)); + // Arbitrary-named index sorts first (oldest); the convention write index stays last. + assertThat(updated.getIndices().get(0), equalTo(legacy)); + assertThat(updated.getIndices().get(1).getName(), equalTo(getDefaultBackingIndexName(dataStreamName, 1))); + } + + public void testParseDataStreamName() { + assertThat(DataStream.parseDataStreamName(".ds-logs-foo-000003"), equalTo("logs-foo")); + assertThat(DataStream.parseDataStreamName(".ds-logs-foo-1234567"), equalTo("logs-foo")); + assertNull(DataStream.parseDataStreamName("logs-foo-000003")); + assertNull(DataStream.parseDataStreamName(".ds-logs-foo-notanumber")); + assertNull(DataStream.parseDataStreamName(".ds-000003")); + assertNull(DataStream.parseDataStreamName(".ds-logs-foo")); + } + + public void testParseDataStreamNameWithEmptyCounter() { + // Trailing dash: the counter substring is empty. Without the isEmpty() guard the empty suffix would fall + // through the digit loop and the name would be (wrongly) accepted as a backing index. + assertNull(DataStream.parseDataStreamName(".ds-logs-foo-")); + assertNull(DataStream.parseDataStreamName(".ds-a-")); + // And an arbitrary index whose name ends in a dash must not be treated as belonging to any stream. + assertThat(DataStream.backingIndexCounterOrMin("logs-foo", ".ds-logs-foo-"), equalTo(Long.MIN_VALUE)); + } + + public void testAddBackingIndexWithCounterAboveIntMax() { + // A counter beyond Integer.MAX_VALUE must still parse: generation is a long, and parseDataStreamName accepts a + // numeric suffix of any length, so backingIndexCounterOrMin must not overflow an int. + String dataStreamName = "logs-foo"; + long bigCounter = ((long) Integer.MAX_VALUE) + 5; + List indices = new ArrayList<>(); + indices.add(new Index(getDefaultBackingIndexName(dataStreamName, 1), UUIDs.randomBase64UUID(random()))); + DataStream original = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, 1); + + Index big = new Index(getDefaultBackingIndexName(dataStreamName, bigCounter), UUIDs.randomBase64UUID(random())); + DataStream updated = original.addBackingIndex(big); + + assertThat(updated.getGeneration(), equalTo(bigCounter)); + assertThat(updated.getIndices().get(updated.getIndices().size() - 1), equalTo(big)); + } + 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/cluster/metadata/MetadataDataStreamsServiceTests.java b/server/src/test/java/org/opensearch/cluster/metadata/MetadataDataStreamsServiceTests.java new file mode 100644 index 0000000000000..962bc5246f16f --- /dev/null +++ b/server/src/test/java/org/opensearch/cluster/metadata/MetadataDataStreamsServiceTests.java @@ -0,0 +1,851 @@ +/* + * 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.cluster.metadata; + +import org.opensearch.OpenSearchException; +import org.opensearch.Version; +import org.opensearch.action.admin.indices.datastream.DataStreamAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.cluster.AckedClusterStateUpdateTask; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.ClusterStateUpdateTask; +import org.opensearch.cluster.ack.ClusterStateUpdateResponse; +import org.opensearch.cluster.service.ClusterManagerTaskThrottler; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.Priority; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.index.Index; +import org.opensearch.test.OpenSearchTestCase; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import org.mockito.ArgumentCaptor; + +import static org.opensearch.cluster.DataStreamTestHelper.createTimestampField; +import static org.opensearch.cluster.DataStreamTestHelper.generateMapping; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; +import static org.hamcrest.Matchers.sameInstance; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +public class MetadataDataStreamsServiceTests extends OpenSearchTestCase { + + private static final String DS = "logs-foo"; + + /** A backing index with the @timestamp date mapping every data stream index requires. */ + private static IndexMetadata.Builder createBackingIndex(String dataStreamName, int generation) { + try { + return org.opensearch.cluster.DataStreamTestHelper.createBackingIndex(dataStreamName, generation) + .putMapping(generateMapping("@timestamp")); + } catch (IOException e) { + throw new AssertionError(e); + } + } + + /** An arbitrary-named index carrying the given @timestamp mapping type ("date", "date_nanos", or "text"). */ + private static IndexMetadata.Builder arbitraryIndex(String name, String timestampType) { + try { + return IndexMetadata.builder(name) + .settings(org.opensearch.common.settings.Settings.builder().put("index.version.created", org.opensearch.Version.CURRENT)) + .numberOfShards(1) + .numberOfReplicas(0) + .putMapping(generateMapping("@timestamp", timestampType)); + } catch (IOException e) { + throw new AssertionError(e); + } + } + + /** + * Builds a cluster state whose data stream {@code DS} has backing indices for exactly the given generations (the + * highest of which must equal {@code generation}, so the write index follows the naming convention), plus any + * {@code standaloneIndices} that exist as plain indices but are NOT part of the stream. Every standalone backing + * index counter must be {@code <= generation}, otherwise {@link Metadata.Builder#build()} validation rejects it as a + * would-be rollover conflict. + */ + private ClusterState state(long generation, List backingGenerations, int... standaloneIndices) { + Metadata.Builder metadata = Metadata.builder(); + List streamIndices = new ArrayList<>(); + for (int gen : backingGenerations) { + IndexMetadata im = createBackingIndex(DS, gen).build(); + metadata.put(im, false); + streamIndices.add(im.getIndex()); + } + for (int gen : standaloneIndices) { + metadata.put(createBackingIndex(DS, gen).build(), false); + } + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), streamIndices, generation)); + return ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + } + + private static List backingIndexNames(ClusterState state) { + return state.metadata().dataStreams().get(DS).getIndices().stream().map(Index::getName).collect(Collectors.toList()); + } + + public void testAddBackingIndexReinsertsInGenerationOrder() { + // Stream foo is at gen 3 with backing [1, 3] (gen 2 was previously detached and still exists standalone). + ClusterState state = state(3, List.of(1, 3), 2); + String reAdded = DataStream.getDefaultBackingIndexName(DS, 2); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.addBackingIndex(DS, reAdded))); + + // The re-added index lands in generation order, not at the end. + assertThat( + backingIndexNames(updated), + contains( + DataStream.getDefaultBackingIndexName(DS, 1), + DataStream.getDefaultBackingIndexName(DS, 2), + DataStream.getDefaultBackingIndexName(DS, 3) + ) + ); + // Generation is derived and unchanged: the write index is still gen 3. + assertThat(updated.metadata().dataStreams().get(DS).getGeneration(), equalTo(3L)); + } + + public void testAddBackingIndexIsIdempotent() { + ClusterState state = state(2, List.of(1, 2)); + String existing = DataStream.getDefaultBackingIndexName(DS, 2); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.addBackingIndex(DS, existing))); + + assertThat(backingIndexNames(updated).size(), equalTo(2)); + assertThat(updated.metadata().dataStreams().get(DS).getGeneration(), equalTo(2L)); + } + + public void testAddUnknownIndexFails() { + ClusterState state = state(2, List.of(1, 2)); + String missing = DataStream.getDefaultBackingIndexName(DS, 5); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.addBackingIndex(DS, missing))) + ); + assertThat(e.getMessage(), containsString("not found")); + } + + public void testAddArbitraryNamedIndexAsOldestBackingIndex() { + // Migrating a pre-existing regular index into a data stream: the arbitrary-named index is attached as the + // oldest backing index, ahead of the convention-following write index, and the generation is unchanged. + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 1).build(); + metadata.put(writeIndex, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(writeIndex.getIndex()), 1)); + metadata.put(arbitraryIndex("legacy-logs-2023", "date").build(), false); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex(DS, "legacy-logs-2023")) + ); + + // Arbitrary-named index sorts first (oldest); the convention-following write index stays last. + assertThat(backingIndexNames(updated), contains("legacy-logs-2023", DataStream.getDefaultBackingIndexName(DS, 1))); + assertThat(updated.metadata().dataStreams().get(DS).getGeneration(), equalTo(1L)); + // The migrated index is marked hidden like every backing index. + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(updated.metadata().index("legacy-logs-2023").getSettings()), equalTo(true)); + } + + public void testAddIndexWithoutTimestampMappingFails() { + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 1).build(); + metadata.put(writeIndex, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(writeIndex.getIndex()), 1)); + // Candidate index maps @timestamp as text, not a date type. + metadata.put(arbitraryIndex("legacy-logs-2023", "text").build(), false); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.addBackingIndex(DS, "legacy-logs-2023"))) + ); + assertThat(e.getMessage(), containsString("does not have a [@timestamp] field mapped as a date type")); + } + + public void testAddConventionNamedIndexWithoutTimestampMappingFails() { + // A .ds--NNNNNN name is not proof the index came from create/rollover; a manually created index with the + // convention name but no timestamp mapping must still be rejected. + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 2).build(); + metadata.put(writeIndex, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(writeIndex.getIndex()), 2)); + // Convention-named gen-1 index created manually with a non-date @timestamp mapping. + metadata.put(arbitraryIndex(DataStream.getDefaultBackingIndexName(DS, 1), "text").build(), false); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex(DS, DataStream.getDefaultBackingIndexName(DS, 1))) + ) + ); + assertThat(e.getMessage(), containsString("does not have a [@timestamp] field mapped as a date type")); + } + + public void testAddDateNanosTimestampMappingSucceeds() { + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 1).build(); + metadata.put(writeIndex, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(writeIndex.getIndex()), 1)); + metadata.put(arbitraryIndex("legacy-logs-2023", "date_nanos").build(), false); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex(DS, "legacy-logs-2023")) + ); + assertThat(backingIndexNames(updated), contains("legacy-logs-2023", DataStream.getDefaultBackingIndexName(DS, 1))); + } + + public void testRemoveBackingIndex() { + ClusterState state = state(3, List.of(1, 2, 3)); + String toRemove = DataStream.getDefaultBackingIndexName(DS, 1); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex(DS, toRemove)) + ); + + assertThat( + backingIndexNames(updated), + contains(DataStream.getDefaultBackingIndexName(DS, 2), DataStream.getDefaultBackingIndexName(DS, 3)) + ); + // Generation is unaffected by removing a non-write index. + assertThat(updated.metadata().dataStreams().get(DS).getGeneration(), equalTo(3L)); + } + + public void testRemoveBackingIndexUnhidesIt() { + // Backing indices are hidden; detaching one makes it visible again, mirroring the hide-on-attach behavior. + ClusterState state = state(3, List.of(1, 2, 3)); + String toRemove = DataStream.getDefaultBackingIndexName(DS, 1); + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(state.metadata().index(toRemove).getSettings()), equalTo(true)); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex(DS, toRemove)) + ); + + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(updated.metadata().index(toRemove).getSettings()), equalTo(false)); + } + + public void testRemoveWriteIndexFails() { + ClusterState state = state(3, List.of(1, 2, 3)); + String writeIndex = DataStream.getDefaultBackingIndexName(DS, 3); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.removeBackingIndex(DS, writeIndex))) + ); + assertThat(e.getMessage(), containsString("because it is the write index")); + } + + public void testRemoveIndexNotPartOfStreamFails() { + // Removing an index that is not a backing index of the stream is rejected (remove only checks stream + // membership, so the index need not exist elsewhere). + ClusterState state = state(2, List.of(1, 2)); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.removeBackingIndex(DS, "not-a-member"))) + ); + assertThat(e.getMessage(), containsString("is not part of data stream [" + DS + "]")); + } + + public void testRemoveThenReAddComposeInSingleUpdate() { + ClusterState state = state(3, List.of(1, 2, 3)); + String middle = DataStream.getDefaultBackingIndexName(DS, 2); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex(DS, middle), DataStreamAction.addBackingIndex(DS, middle)) + ); + + DataStream ds = updated.metadata().dataStreams().get(DS); + assertThat(ds.getIndices().size(), equalTo(3)); + assertThat(ds.getGeneration(), equalTo(3L)); + } + + public void testAddMarksBackingIndexHidden() throws IOException { + // A standalone, non-hidden backing-index-named index at a counter <= generation. + String DS_NAME = DS; + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata b1 = createBackingIndex(DS_NAME, 1).build(); + IndexMetadata b2 = createBackingIndex(DS_NAME, 2).build(); + metadata.put(b1, false); + metadata.put(b2, false); + // gen 1 exists standalone and is explicitly NOT hidden + IndexMetadata visible = IndexMetadata.builder(DataStream.getDefaultBackingIndexName(DS_NAME, 1)) + .settings( + org.opensearch.common.settings.Settings.builder() + .put("index.version.created", org.opensearch.Version.CURRENT) + .put(IndexMetadata.SETTING_INDEX_HIDDEN, false) + ) + .numberOfShards(1) + .numberOfReplicas(0) + .putMapping(generateMapping("@timestamp")) + .build(); + // Stream at gen 2 with backing [2] only; gen-1 index is standalone and visible. + metadata.put(visible, true); + metadata.put(new DataStream(DS_NAME, createTimestampField("@timestamp"), List.of(b2.getIndex()), 2)); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + String toAdd = DataStream.getDefaultBackingIndexName(DS_NAME, 1); + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(state.metadata().index(toAdd).getSettings()), equalTo(false)); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex(DS_NAME, toAdd)) + ); + + // The attached index is now hidden, matching every other backing index. + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(updated.metadata().index(toAdd).getSettings()), equalTo(true)); + assertThat(backingIndexNames(updated), contains(toAdd, DataStream.getDefaultBackingIndexName(DS_NAME, 2))); + } + + public void testUnknownDataStreamFails() { + ClusterState state = state(2, List.of(1, 2)); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex("no-such-stream", ".ds-no-such-stream-000001")) + ) + ); + assertThat(e.getMessage(), containsString("data stream [no-such-stream] not found")); + } + + /** + * Builds a cluster state with two data streams. {@code logs-a} has backing gen 1 (arbitrary index {@code shared} + * belongs to it), and {@code logs-b} has backing gen 1. {@code shared} is a standalone arbitrary-named index that + * belongs to logs-a in the initial state. + */ + private ClusterState twoStreamsWithSharedCandidate() { + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata a1 = createBackingIndex("logs-a", 1).build(); + IndexMetadata b1 = createBackingIndex("logs-b", 1).build(); + IndexMetadata shared = arbitraryIndex("shared-idx", "date").build(); + metadata.put(a1, false); + metadata.put(b1, false); + metadata.put(shared, false); + // logs-a owns both its convention write index and the arbitrary "shared-idx". + metadata.put(new DataStream("logs-a", createTimestampField("@timestamp"), List.of(shared.getIndex(), a1.getIndex()), 1)); + metadata.put(new DataStream("logs-b", createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + return ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + } + + private static List backingIndexNames(ClusterState state, String dataStream) { + return state.metadata().dataStreams().get(dataStream).getIndices().stream().map(Index::getName).collect(Collectors.toList()); + } + + public void testCannotAddSameIndexToTwoStreamsInOneRequest() { + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata a1 = createBackingIndex("logs-a", 1).build(); + IndexMetadata b1 = createBackingIndex("logs-b", 1).build(); + metadata.put(a1, false); + metadata.put(b1, false); + metadata.put(new DataStream("logs-a", createTimestampField("@timestamp"), List.of(a1.getIndex()), 1)); + metadata.put(new DataStream("logs-b", createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + metadata.put(arbitraryIndex("legacy-idx", "date").build(), false); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex("logs-a", "legacy-idx"), DataStreamAction.addBackingIndex("logs-b", "legacy-idx")) + ) + ); + assertThat(e.getMessage(), containsString("more than one data stream")); + } + + public void testMoveBackingIndexBetweenStreamsRemoveThenAdd() { + ClusterState state = twoStreamsWithSharedCandidate(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex("logs-a", "shared-idx"), DataStreamAction.addBackingIndex("logs-b", "shared-idx")) + ); + + assertThat(backingIndexNames(updated, "logs-a"), contains(DataStream.getDefaultBackingIndexName("logs-a", 1))); + assertThat(backingIndexNames(updated, "logs-b"), contains("shared-idx", DataStream.getDefaultBackingIndexName("logs-b", 1))); + } + + public void testMoveBackingIndexBetweenStreamsAddThenRemove() { + // Reverse action order from the previous test; the outcome must be identical. + ClusterState state = twoStreamsWithSharedCandidate(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex("logs-b", "shared-idx"), DataStreamAction.removeBackingIndex("logs-a", "shared-idx")) + ); + + assertThat(backingIndexNames(updated, "logs-a"), contains(DataStream.getDefaultBackingIndexName("logs-a", 1))); + assertThat(backingIndexNames(updated, "logs-b"), contains("shared-idx", DataStream.getDefaultBackingIndexName("logs-b", 1))); + } + + public void testAddWithoutRemoveStillRejectedRegardlessOfOrder() { + // Adding shared-idx to logs-b without removing it from logs-a must fail, in either action order. + ClusterState state = twoStreamsWithSharedCandidate(); + + IllegalArgumentException forward = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.addBackingIndex("logs-b", "shared-idx"))) + ); + assertThat(forward.getMessage(), containsString("more than one data stream")); + } + + // ------------------------------------------------------------------------------------------------------------ + // Timestamp-field mapping validation: every shape that leaves `type` null must be rejected. + // ------------------------------------------------------------------------------------------------------------ + + /** An index named {@code name} carrying the given raw mapping source, or no mapping at all when {@code null}. */ + private static IndexMetadata.Builder indexWithRawMapping(String name, String mappingSource) { + IndexMetadata.Builder builder = IndexMetadata.builder(name) + .settings(Settings.builder().put("index.version.created", Version.CURRENT)) + .numberOfShards(1) + .numberOfReplicas(0); + if (mappingSource != null) { + try { + builder.putMapping(mappingSource); + } catch (IOException e) { + throw new AssertionError(e); + } + } + return builder; + } + + /** A stream {@code DS} at generation 1 (single convention-named write index) plus one standalone candidate index. */ + private ClusterState stateWithCandidate(IndexMetadata candidate) { + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 1).build(); + metadata.put(writeIndex, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(writeIndex.getIndex()), 1)); + metadata.put(candidate, false); + return ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + } + + private void expectTimestampMappingRejection(IndexMetadata candidate) { + ClusterState state = stateWithCandidate(candidate); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex(DS, candidate.getIndex().getName())) + ) + ); + assertThat( + e.getMessage(), + equalTo( + "index [" + + candidate.getIndex().getName() + + "] cannot be added as a backing index because it does not have a [@timestamp] field mapped as a date type" + ) + ); + } + + public void testAddIndexWithNoMappingAtAllFails() { + // mapping() is null: the whole property lookup is skipped and `type` stays null. + IndexMetadata candidate = indexWithRawMapping("no-mapping-idx", null).build(); + assertThat(candidate.mapping(), nullValue()); + expectTimestampMappingRejection(candidate); + } + + public void testAddIndexWhoseMappingHasNoPropertiesFails() { + // A mapping exists but carries no "properties" block at all. + IndexMetadata candidate = indexWithRawMapping("no-properties-idx", "{\"_meta\":{\"origin\":\"manual\"}}").build(); + assertThat(candidate.mapping(), notNullValue()); + expectTimestampMappingRejection(candidate); + } + + public void testAddIndexWhosePropertiesIsNotAnObjectFails() { + // "properties" is present but is a scalar, so the `properties instanceof Map` guard rejects it. + IndexMetadata candidate = indexWithRawMapping("scalar-properties-idx", "{\"properties\":\"oops\"}").build(); + expectTimestampMappingRejection(candidate); + } + + public void testAddIndexWhoseMappingLacksTimestampFieldFails() { + // "properties" exists but has no @timestamp entry at all. + IndexMetadata candidate = indexWithRawMapping("other-field-idx", "{\"properties\":{\"other\":{\"type\":\"date\"}}}").build(); + expectTimestampMappingRejection(candidate); + } + + public void testAddIndexWhoseTimestampFieldIsNotAnObjectFails() { + // @timestamp is present but is a scalar, so the `field instanceof Map` guard rejects it. + IndexMetadata candidate = indexWithRawMapping("scalar-timestamp-idx", "{\"properties\":{\"@timestamp\":\"date\"}}").build(); + expectTimestampMappingRejection(candidate); + } + + public void testAddIndexWhoseTimestampFieldHasNoTypeFails() { + // @timestamp is an object but declares no "type", so `type` is null. + IndexMetadata candidate = indexWithRawMapping("untyped-timestamp-idx", "{\"properties\":{\"@timestamp\":{\"index\":true}}}") + .build(); + expectTimestampMappingRejection(candidate); + } + + public void testValidateTimestampFieldMappingAcceptsBothDateTypes() throws IOException { + // The accepting side of both halves of the `date`/`date_nanos` condition, exercised directly. + MetadataDataStreamsService.validateTimestampFieldMapping(arbitraryIndex("d", "date").build(), "@timestamp"); + MetadataDataStreamsService.validateTimestampFieldMapping(arbitraryIndex("dn", "date_nanos").build(), "@timestamp"); + // A stream whose timestamp field is not the default name is validated against that name. + IndexMetadata custom = IndexMetadata.builder("custom-ts") + .settings(Settings.builder().put("index.version.created", Version.CURRENT)) + .numberOfShards(1) + .numberOfReplicas(0) + .putMapping(generateMapping("event.ingested", "date")) + .build(); + MetadataDataStreamsService.validateTimestampFieldMapping(custom, "event.ingested"); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.validateTimestampFieldMapping(custom, "@timestamp") + ); + assertThat(e.getMessage(), containsString("does not have a [@timestamp] field mapped as a date type")); + } + + // ------------------------------------------------------------------------------------------------------------ + // Remove-backing-index guards. + // ------------------------------------------------------------------------------------------------------------ + + public void testRemoveLastBackingIndexFails() { + // Size-1 check runs before the write-index check, so the message must be the "last backing index" one even + // though the single index is also the write index. + ClusterState state = state(1, List.of(1)); + String only = DataStream.getDefaultBackingIndexName(DS, 1); + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> MetadataDataStreamsService.modifyDataStream(state, List.of(DataStreamAction.removeBackingIndex(DS, only))) + ); + assertThat( + e.getMessage(), + equalTo( + "cannot remove backing index [" + + only + + "] of data stream [" + + DS + + "] because it is the last backing index; delete the data stream instead" + ) + ); + } + + public void testRemoveAlreadyVisibleBackingIndexLeavesSettingsUntouched() { + // A backing index that is not hidden takes the false side of the unhide guard: no settings write at all. + Metadata.Builder metadata = Metadata.builder(); + IndexMetadata writeIndex = createBackingIndex(DS, 1).build(); + IndexMetadata visible = IndexMetadata.builder("legacy-visible") + .settings(Settings.builder().put("index.version.created", Version.CURRENT).put(IndexMetadata.SETTING_INDEX_HIDDEN, false)) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + metadata.put(writeIndex, false); + metadata.put(visible, false); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(visible.getIndex(), writeIndex.getIndex()), 1)); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + long settingsVersionBefore = state.metadata().index("legacy-visible").getSettingsVersion(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex(DS, "legacy-visible")) + ); + + assertThat(backingIndexNames(updated), contains(DataStream.getDefaultBackingIndexName(DS, 1))); + assertThat(IndexMetadata.INDEX_HIDDEN_SETTING.get(updated.metadata().index("legacy-visible").getSettings()), equalTo(false)); + // The settings version is untouched, proving the unhide branch was skipped rather than re-applied. + assertThat(updated.metadata().index("legacy-visible").getSettingsVersion(), equalTo(settingsVersionBefore)); + } + + public void testRemoveBackingIndexWithNoIndexMetadataSkipsUnhide() { + // Defensive path: the stream references indices for which no IndexMetadata exists (Metadata skips data stream + // lookup building entirely when there are no indices), so metadataBuilder.get(...) returns null. + Index first = new Index(DataStream.getDefaultBackingIndexName(DS, 1), "uuid-1"); + Index writeIndex = new Index(DataStream.getDefaultBackingIndexName(DS, 2), "uuid-2"); + Metadata.Builder metadata = Metadata.builder(); + metadata.put(new DataStream(DS, createTimestampField("@timestamp"), List.of(first, writeIndex), 2)); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + assertThat(state.metadata().index(first.getName()), nullValue()); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.removeBackingIndex(DS, first.getName())) + ); + + assertThat(backingIndexNames(updated), contains(writeIndex.getName())); + assertThat(updated.metadata().indices().size(), equalTo(0)); + } + + // ------------------------------------------------------------------------------------------------------------ + // Shared-backing-index validation. + // ------------------------------------------------------------------------------------------------------------ + + public void testIndexNameListedTwiceWithinOneStreamIsNotReportedAsShared() { + // The shared-index check keys on index name, so a stream that lists the same name twice would collide with + // itself; the owner-equality half of the guard must let that through instead of failing the whole update. + IndexMetadata a1 = createBackingIndex("logs-a", 1).build(); + IndexMetadata b1 = createBackingIndex("logs-b", 1).build(); + Metadata.Builder metadata = Metadata.builder(); + metadata.put(a1, false); + metadata.put(b1, false); + // logs-a lists .ds-logs-a-000001 twice (same name, different UUIDs). + metadata.put( + new DataStream( + "logs-a", + createTimestampField("@timestamp"), + List.of(new Index(a1.getIndex().getName(), "other-uuid"), a1.getIndex()), + 1 + ) + ); + metadata.put(new DataStream("logs-b", createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + // Any action re-runs the full shared-index validation across every stream, including logs-a. + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of(DataStreamAction.addBackingIndex("logs-b", b1.getIndex().getName())) + ); + + assertThat(updated.metadata().dataStreams().get("logs-a").getIndices().size(), equalTo(2)); + assertThat(backingIndexNames(updated, "logs-b"), contains(b1.getIndex().getName())); + } + + public void testSeveralActionsAcrossTwoStreamsComposeInOneUpdate() { + // One request, four actions, two streams: each stream's intermediate result feeds the next action on it. + IndexMetadata a1 = createBackingIndex("logs-a", 1).build(); + IndexMetadata a2 = createBackingIndex("logs-a", 2).build(); + IndexMetadata b1 = createBackingIndex("logs-b", 1).build(); + IndexMetadata shared = arbitraryIndex("roaming-idx", "date").build(); + Metadata.Builder metadata = Metadata.builder(); + metadata.put(a1, false); + metadata.put(a2, false); + metadata.put(b1, false); + metadata.put(shared, false); + metadata.put( + new DataStream("logs-a", createTimestampField("@timestamp"), List.of(shared.getIndex(), a1.getIndex(), a2.getIndex()), 2) + ); + metadata.put(new DataStream("logs-b", createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(metadata).build(); + + ClusterState updated = MetadataDataStreamsService.modifyDataStream( + state, + List.of( + DataStreamAction.removeBackingIndex("logs-a", "roaming-idx"), + DataStreamAction.removeBackingIndex("logs-a", a1.getIndex().getName()), + DataStreamAction.addBackingIndex("logs-b", "roaming-idx"), + DataStreamAction.addBackingIndex("logs-b", a1.getIndex().getName()) + ) + ); + + assertThat(backingIndexNames(updated, "logs-a"), contains(a2.getIndex().getName())); + // The arbitrary name sorts first; .ds-logs-a-000001 is not a logs-b convention name so it also sorts first. + assertThat(backingIndexNames(updated, "logs-b").size(), equalTo(3)); + assertThat(backingIndexNames(updated, "logs-b").get(2), equalTo(b1.getIndex().getName())); + assertThat(updated.metadata().dataStreams().get("logs-b").getGeneration(), equalTo(1L)); + } + + // ------------------------------------------------------------------------------------------------------------ + // Service entry points: request handling and the submitted cluster-manager task. + // ------------------------------------------------------------------------------------------------------------ + + /** Captures whichever ActionListener callback fires, so both success and failure can be asserted. */ + private static final class CapturingListener implements ActionListener { + private AcknowledgedResponse response; + private Exception failure; + + @Override + public void onResponse(AcknowledgedResponse acknowledgedResponse) { + this.response = acknowledgedResponse; + } + + @Override + public void onFailure(Exception e) { + this.failure = e; + } + } + + private ClusterManagerTaskThrottler.ThrottlingKey throttlingKey; + + private ClusterService mockClusterService() { + ClusterService clusterService = mock(ClusterService.class); + throttlingKey = mock(ClusterManagerTaskThrottler.ThrottlingKey.class); + when(clusterService.registerClusterManagerTask(anyString(), anyBoolean())).thenReturn(throttlingKey); + return clusterService; + } + + @SuppressWarnings("unchecked") + private AckedClusterStateUpdateTask captureSubmittedTask(ClusterService clusterService) { + ArgumentCaptor captor = ArgumentCaptor.forClass(ClusterStateUpdateTask.class); + verify(clusterService).submitStateUpdateTask(eq("update-data-streams"), captor.capture()); + return (AckedClusterStateUpdateTask) captor.getValue(); + } + + public void testConstructorRegistersThrottledClusterManagerTask() { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + assertThat(service, notNullValue()); + verify(clusterService).registerClusterManagerTask("modify-data-stream", true); + } + + public void testEmptyActionListIsAcknowledgedWithoutSubmittingAnyTask() { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + CapturingListener listener = new CapturingListener(); + + service.modifyDataStream( + new MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest( + Collections.emptyList(), + TimeValue.timeValueSeconds(30), + TimeValue.timeValueSeconds(30) + ), + listener + ); + + assertThat(listener.failure, nullValue()); + assertThat(listener.response.isAcknowledged(), equalTo(true)); + verify(clusterService, never()).submitStateUpdateTask(anyString(), any(ClusterStateUpdateTask.class)); + } + + public void testSubmittedTaskCarriesRequestConfigurationAndAppliesActions() throws Exception { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + CapturingListener listener = new CapturingListener(); + String toRemove = DataStream.getDefaultBackingIndexName(DS, 1); + + service.modifyDataStream( + new MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest( + List.of(DataStreamAction.removeBackingIndex(DS, toRemove)), + TimeValue.timeValueSeconds(11), + TimeValue.timeValueSeconds(22) + ), + listener + ); + + AckedClusterStateUpdateTask task = captureSubmittedTask(clusterService); + assertThat(task.priority(), equalTo(Priority.HIGH)); + assertThat(task.timeout(), equalTo(TimeValue.timeValueSeconds(11))); + assertThat(task.ackTimeout(), equalTo(TimeValue.timeValueSeconds(22))); + assertThat(task.getClusterManagerThrottlingKey(), sameInstance(throttlingKey)); + + // The task's execute() is the same metadata mutation the static helper performs. + ClusterState updated = task.execute(state(2, List.of(1, 2))); + assertThat(backingIndexNames(updated), contains(DataStream.getDefaultBackingIndexName(DS, 2))); + + // Nothing is reported to the caller until the state update is acked. + assertThat(listener.response, nullValue()); + task.onAllNodesAcked(null); + assertThat(listener.response.isAcknowledged(), equalTo(true)); + } + + public void testAckedWithExceptionAndAckTimeoutBothYieldUnacknowledged() { + // Both non-happy acknowledgement paths map to an unacknowledged AcknowledgedResponse rather than a failure. + ClusterService ackTimeoutClusterService = mockClusterService(); + CapturingListener ackTimeoutListener = new CapturingListener(); + new MetadataDataStreamsService(ackTimeoutClusterService).modifyDataStream( + List.of(DataStreamAction.removeBackingIndex(DS, DataStream.getDefaultBackingIndexName(DS, 1))), + TimeValue.timeValueSeconds(5), + TimeValue.timeValueSeconds(5), + ackTimeoutListener + ); + captureSubmittedTask(ackTimeoutClusterService).onAckTimeout(); + assertThat(ackTimeoutListener.failure, nullValue()); + assertThat(ackTimeoutListener.response.isAcknowledged(), equalTo(false)); + + ClusterService ackFailureClusterService = mockClusterService(); + CapturingListener ackFailureListener = new CapturingListener(); + new MetadataDataStreamsService(ackFailureClusterService).modifyDataStream( + List.of(DataStreamAction.removeBackingIndex(DS, DataStream.getDefaultBackingIndexName(DS, 1))), + TimeValue.timeValueSeconds(5), + TimeValue.timeValueSeconds(5), + ackFailureListener + ); + captureSubmittedTask(ackFailureClusterService).onAllNodesAcked(new OpenSearchException("ack failed")); + assertThat(ackFailureListener.failure, nullValue()); + assertThat(ackFailureListener.response.isAcknowledged(), equalTo(false)); + } + + public void testConvenienceOverloadPassesTimeoutsThrough() throws Exception { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + CapturingListener listener = new CapturingListener(); + + service.modifyDataStream( + List.of(DataStreamAction.addBackingIndex(DS, DataStream.getDefaultBackingIndexName(DS, 2))), + TimeValue.timeValueSeconds(7), + TimeValue.timeValueSeconds(13), + listener + ); + + AckedClusterStateUpdateTask task = captureSubmittedTask(clusterService); + assertThat(task.timeout(), equalTo(TimeValue.timeValueSeconds(7))); + assertThat(task.ackTimeout(), equalTo(TimeValue.timeValueSeconds(13))); + // Idempotent add: the stream is unchanged but the task still runs. + assertThat(backingIndexNames(task.execute(state(2, List.of(1, 2)))).size(), equalTo(2)); + } + + public void testTaskFailurePropagatesToTheCallersListener() { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + CapturingListener listener = new CapturingListener(); + + service.modifyDataStream( + List.of(DataStreamAction.removeBackingIndex(DS, "nope")), + TimeValue.timeValueSeconds(5), + TimeValue.timeValueSeconds(5), + listener + ); + + AckedClusterStateUpdateTask task = captureSubmittedTask(clusterService); + // A rejected action surfaces from execute() and is reported through onFailure. + IllegalArgumentException thrown = expectThrows(IllegalArgumentException.class, () -> task.execute(state(2, List.of(1, 2)))); + assertThat(thrown.getMessage(), containsString("is not part of data stream")); + + task.onFailure("update-data-streams", thrown); + assertThat(listener.response, nullValue()); + assertThat(listener.failure, sameInstance(thrown)); + } + + public void testEmptyActionListPathDoesNotTouchTheClusterServiceBeyondRegistration() { + ClusterService clusterService = mockClusterService(); + MetadataDataStreamsService service = new MetadataDataStreamsService(clusterService); + verify(clusterService).registerClusterManagerTask(anyString(), anyBoolean()); + + service.modifyDataStream( + Collections.emptyList(), + TimeValue.timeValueSeconds(1), + TimeValue.timeValueSeconds(1), + new CapturingListener() + ); + + verifyNoMoreInteractions(clusterService); + } + + public void testUpdateRequestExposesItsActionsAndTimeouts() { + List actions = List.of(DataStreamAction.addBackingIndex(DS, "a"), DataStreamAction.removeBackingIndex(DS, "b")); + MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest request = + new MetadataDataStreamsService.ModifyDataStreamsClusterStateUpdateRequest( + actions, + TimeValue.timeValueSeconds(3), + TimeValue.timeValueSeconds(9) + ); + + assertThat(request.getActions(), equalTo(actions)); + assertThat(request.masterNodeTimeout(), equalTo(TimeValue.timeValueSeconds(3))); + assertThat(request.ackTimeout(), equalTo(TimeValue.timeValueSeconds(9))); + } +} diff --git a/server/src/test/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsActionTests.java b/server/src/test/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsActionTests.java new file mode 100644 index 0000000000000..ed73c0d70f222 --- /dev/null +++ b/server/src/test/java/org/opensearch/rest/action/admin/indices/RestModifyDataStreamsActionTests.java @@ -0,0 +1,227 @@ +/* + * 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.rest.action.admin.indices; + +import org.opensearch.OpenSearchParseException; +import org.opensearch.action.admin.indices.datastream.DataStreamAction; +import org.opensearch.action.admin.indices.datastream.ModifyDataStreamsAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.common.CheckedConsumer; +import org.opensearch.core.common.bytes.BytesArray; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.core.xcontent.MediaTypeRegistry; +import org.opensearch.rest.RestChannel; +import org.opensearch.rest.RestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.test.rest.FakeRestChannel; +import org.opensearch.test.rest.FakeRestRequest; +import org.opensearch.test.rest.RestActionTestCase; +import org.junit.Before; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; + +public class RestModifyDataStreamsActionTests extends RestActionTestCase { + + private RestModifyDataStreamsAction action; + + @Before + public void setupAction() { + action = new RestModifyDataStreamsAction(); + controller().registerHandler(action); + } + + public void testGetName() { + assertThat(action.getName(), equalTo("modify_data_stream_action")); + } + + public void testRoutes() { + List routes = action.routes(); + assertThat(routes, hasSize(1)); + RestHandler.Route route = routes.get(0); + assertThat(route.getMethod(), equalTo(RestRequest.Method.POST)); + assertThat(route.getPath(), equalTo("/_data_stream/_modify")); + } + + public void testAddBackingIndex() throws Exception { + String body = "{\"actions\":[{\"add_backing_index\":{\"data_stream\":\"logs-foo\",\"index\":\".ds-logs-foo-000001\"}}]}"; + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest(body, emptyParams())); + + assertThat(request.getActions(), hasSize(1)); + DataStreamAction dsAction = request.getActions().get(0); + assertThat(dsAction.type(), equalTo(DataStreamAction.Type.ADD_BACKING_INDEX)); + assertThat(dsAction.dataStream(), equalTo("logs-foo")); + assertThat(dsAction.index(), equalTo(".ds-logs-foo-000001")); + assertNull(request.validate()); + } + + public void testRemoveBackingIndex() throws Exception { + String body = "{\"actions\":[{\"remove_backing_index\":{\"data_stream\":\"logs-bar\",\"index\":\".ds-logs-bar-000003\"}}]}"; + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest(body, emptyParams())); + + assertThat(request.getActions(), hasSize(1)); + DataStreamAction dsAction = request.getActions().get(0); + assertThat(dsAction.type(), equalTo(DataStreamAction.Type.REMOVE_BACKING_INDEX)); + assertThat(dsAction.dataStream(), equalTo("logs-bar")); + assertThat(dsAction.index(), equalTo(".ds-logs-bar-000003")); + } + + public void testMultipleActionsInOneBody() throws Exception { + String body = "{\"actions\":[" + + "{\"remove_backing_index\":{\"data_stream\":\"logs-foo\",\"index\":\".ds-logs-foo-000001\"}}," + + "{\"add_backing_index\":{\"data_stream\":\"logs-bar\",\"index\":\".ds-logs-foo-000001\"}}," + + "{\"add_backing_index\":{\"data_stream\":\"logs-bar\",\"index\":\"standalone-index\"}}" + + "]}"; + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest(body, emptyParams())); + + assertThat(request.getActions(), hasSize(3)); + assertThat(request.getActions().get(0), equalTo(DataStreamAction.removeBackingIndex("logs-foo", ".ds-logs-foo-000001"))); + assertThat(request.getActions().get(1), equalTo(DataStreamAction.addBackingIndex("logs-bar", ".ds-logs-foo-000001"))); + assertThat(request.getActions().get(2), equalTo(DataStreamAction.addBackingIndex("logs-bar", "standalone-index"))); + } + + public void testEmptyActionsArrayProducesEmptyRequestThatFailsValidation() throws Exception { + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest("{\"actions\":[]}", emptyParams())); + + assertThat(request.getActions(), hasSize(0)); + assertNotNull(request.validate()); + assertThat(request.validate().getMessage(), containsString("at least one data stream action must be specified")); + } + + public void testEmptyObjectBodyProducesEmptyRequest() throws Exception { + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest("{}", emptyParams())); + + assertThat(request.getActions(), hasSize(0)); + assertNotNull(request.validate()); + } + + public void testDefaultTimeoutsWhenNoParams() throws Exception { + String body = "{\"actions\":[{\"add_backing_index\":{\"data_stream\":\"logs-foo\",\"index\":\"i1\"}}]}"; + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest(body, emptyParams())); + + ModifyDataStreamsAction.Request untouched = new ModifyDataStreamsAction.Request( + List.of(DataStreamAction.addBackingIndex("logs-foo", "i1")) + ); + assertThat(request.clusterManagerNodeTimeout(), equalTo(untouched.clusterManagerNodeTimeout())); + assertThat(request.timeout(), equalTo(untouched.timeout())); + } + + public void testTimeoutsReadFromParams() throws Exception { + String body = "{\"actions\":[{\"add_backing_index\":{\"data_stream\":\"logs-foo\",\"index\":\"i1\"}}]}"; + Map params = new HashMap<>(); + params.put("cluster_manager_timeout", "45s"); + params.put("timeout", "17s"); + ModifyDataStreamsAction.Request request = executeAndCapture(buildRestRequest(body, params)); + + assertThat(request.clusterManagerNodeTimeout().millis(), equalTo(45_000L)); + assertThat(request.timeout().millis(), equalTo(17_000L)); + } + + public void testMissingBodyIsRejected() { + RestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry()).withMethod(RestRequest.Method.POST) + .withPath("/_data_stream/_modify") + .build(); + + OpenSearchParseException e = expectThrows( + OpenSearchParseException.class, + () -> action.prepareRequest(restRequest, verifyingClient) + ); + assertThat(e.getMessage(), containsString("request body is required")); + } + + public void testBodyThatIsNotAnObjectIsRejected() { + RestRequest restRequest = buildRestRequest("[]", emptyParams()); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(restRequest, verifyingClient) + ); + assertThat(e.getMessage(), equalTo("expected an object with an [actions] array")); + } + + public void testUnexpectedTopLevelFieldIsRejected() { + RestRequest restRequest = buildRestRequest("{\"not_actions\":\"whatever\"}", emptyParams()); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(restRequest, verifyingClient) + ); + assertThat(e.getMessage(), equalTo("unexpected field [not_actions], only [actions] is supported")); + } + + public void testActionsFieldThatIsNotAnArrayIsRejected() { + RestRequest restRequest = buildRestRequest("{\"actions\":\"not-an-array\"}", emptyParams()); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(restRequest, verifyingClient) + ); + assertThat(e.getMessage(), equalTo("unexpected field [actions], only [actions] is supported")); + } + + public void testUnknownActionTypeIsRejected() { + RestRequest restRequest = buildRestRequest( + "{\"actions\":[{\"rename_backing_index\":{\"data_stream\":\"logs-foo\",\"index\":\"i1\"}}]}", + emptyParams() + ); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> action.prepareRequest(restRequest, verifyingClient) + ); + assertThat(e.getMessage(), containsString("rename_backing_index")); + } + + public void testMalformedJsonIsRejected() { + RestRequest restRequest = buildRestRequest("{\"actions\":[", emptyParams()); + + expectThrows(Exception.class, () -> action.prepareRequest(restRequest, verifyingClient)); + } + + /** + * Runs {@link RestModifyDataStreamsAction#prepareRequest} and the returned consumer against the verifying client, + * returning the {@link ModifyDataStreamsAction.Request} that was handed to the client. + */ + private ModifyDataStreamsAction.Request executeAndCapture(RestRequest restRequest) throws Exception { + AtomicReference captured = new AtomicReference<>(); + verifyingClient.setExecuteVerifier((actionType, request) -> { + assertThat(actionType, equalTo(ModifyDataStreamsAction.INSTANCE)); + captured.set((ModifyDataStreamsAction.Request) request); + return new AcknowledgedResponse(true); + }); + + CheckedConsumer consumer = action.prepareRequest(restRequest, verifyingClient); + FakeRestChannel channel = new FakeRestChannel(restRequest, true, 1); + consumer.accept(channel); + + assertThat(channel.responses().get(), equalTo(1)); + assertThat(channel.errors().get(), equalTo(0)); + assertThat(channel.capturedResponse().status(), equalTo(RestStatus.OK)); + assertNotNull(captured.get()); + return captured.get(); + } + + private Map emptyParams() { + return new HashMap<>(); + } + + private RestRequest buildRestRequest(String content, Map params) { + return new FakeRestRequest.Builder(xContentRegistry()).withMethod(RestRequest.Method.POST) + .withPath("/_data_stream/_modify") + .withParams(params) + .withContent(new BytesArray(content), MediaTypeRegistry.JSON) + .build(); + } +} diff --git a/server/src/test/java/org/opensearch/snapshots/RestoreServiceTests.java b/server/src/test/java/org/opensearch/snapshots/RestoreServiceTests.java index 3629f324a62e4..c1d7f6b40cf7e 100644 --- a/server/src/test/java/org/opensearch/snapshots/RestoreServiceTests.java +++ b/server/src/test/java/org/opensearch/snapshots/RestoreServiceTests.java @@ -41,10 +41,16 @@ import org.opensearch.indices.replication.common.ReplicationType; import org.opensearch.test.OpenSearchTestCase; +import java.io.IOException; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Set; +import static org.opensearch.cluster.DataStreamTestHelper.createBackingIndex; import static org.opensearch.cluster.DataStreamTestHelper.createTimestampField; +import static org.opensearch.cluster.DataStreamTestHelper.generateMapping; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SEARCH_REPLICAS; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_REPLICATION_TYPE; import static org.mockito.Mockito.eq; @@ -127,6 +133,110 @@ public void testPrefixNotChanged() { assertEquals(Collections.singletonList(renamedIndex), renamedDataStream.getIndices()); } + private static IndexMetadata backingIndex(String dataStreamName, int generation) { + try { + return createBackingIndex(dataStreamName, generation).putMapping(generateMapping("@timestamp")).build(); + } catch (IOException e) { + throw new AssertionError(e); + } + } + + public void testAttachRestoredBackingIndexAdvancesGeneration() { + String ds = "logs-attach"; + IndexMetadata b1 = backingIndex(ds, 1); + IndexMetadata b2 = backingIndex(ds, 2); + // Stream is at generation 1; the restored gen-2 index is present in the builder but not yet a member. + Metadata.Builder metadata = Metadata.builder().put(b1, false).put(b2, false); + Map updatedDataStreams = new HashMap<>(); + updatedDataStreams.put(ds, new DataStream(ds, createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + + RestoreService.attachRestoredBackingIndices(Set.of(b2.getIndex().getName()), metadata, updatedDataStreams); + + DataStream result = updatedDataStreams.get(ds); + assertEquals(List.of(b1.getIndex(), b2.getIndex()), result.getIndices()); + assertEquals(2L, result.getGeneration()); + } + + public void testAttachSkipsNonConventionAndNonMemberCases() { + String ds = "logs-attach"; + IndexMetadata b1 = backingIndex(ds, 1); + Metadata.Builder metadata = Metadata.builder().put(b1, false); + DataStream original = new DataStream(ds, createTimestampField("@timestamp"), List.of(b1.getIndex()), 1); + Map updatedDataStreams = new HashMap<>(); + updatedDataStreams.put(ds, original); + + // A non-convention name (no matching stream) and an already-member index are both skipped, leaving the + // stream unchanged. + RestoreService.attachRestoredBackingIndices(Set.of("some-regular-index", b1.getIndex().getName()), metadata, updatedDataStreams); + + assertEquals(original, updatedDataStreams.get(ds)); + } + + public void testAttachSkipsWhenStreamDoesNotExist() { + // Restored index parses to a stream name, but no such stream exists on the target: skipped, no exception. + String ds = "logs-missing"; + IndexMetadata b1 = backingIndex(ds, 1); + Metadata.Builder metadata = Metadata.builder().put(b1, false); + Map updatedDataStreams = new HashMap<>(); + + RestoreService.attachRestoredBackingIndices(Set.of(b1.getIndex().getName()), metadata, updatedDataStreams); + + assertTrue(updatedDataStreams.isEmpty()); + } + + public void testAttachRejectsIndexWithoutTimestampMapping() { + String ds = "logs-attach"; + IndexMetadata b1 = backingIndex(ds, 1); + // A convention-named gen-2 index whose @timestamp is not a date type. + IndexMetadata bad; + try { + bad = createBackingIndex(ds, 2).putMapping(generateMapping("@timestamp", "text")).build(); + } catch (IOException e) { + throw new AssertionError(e); + } + Metadata.Builder metadata = Metadata.builder().put(b1, false).put(bad, false); + Map updatedDataStreams = new HashMap<>(); + updatedDataStreams.put(ds, new DataStream(ds, createTimestampField("@timestamp"), List.of(b1.getIndex()), 1)); + + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> RestoreService.attachRestoredBackingIndices(Set.of(bad.getIndex().getName()), metadata, updatedDataStreams) + ); + assertTrue(e.getMessage().contains("does not have a [@timestamp] field mapped as a date type")); + } + + public void testAttachSkipsWhenRestoredIndexIsNotInMetadata() { + // The restored name parses to an existing stream, but no IndexMetadata was built for it: the + // restoredIndexMetadata == null side of the guard must be taken, leaving the stream untouched. + String ds = "logs-attach"; + IndexMetadata b1 = backingIndex(ds, 1); + Metadata.Builder metadata = Metadata.builder().put(b1, false); + DataStream original = new DataStream(ds, createTimestampField("@timestamp"), List.of(b1.getIndex()), 1); + Map updatedDataStreams = new HashMap<>(); + updatedDataStreams.put(ds, original); + + String absentButConventionNamed = DataStream.getDefaultBackingIndexName(ds, 2); + assertNull(metadata.get(absentButConventionNamed)); + + RestoreService.attachRestoredBackingIndices(Set.of(absentButConventionNamed), metadata, updatedDataStreams); + + assertEquals(original, updatedDataStreams.get(ds)); + assertEquals(1L, updatedDataStreams.get(ds).getGeneration()); + } + + public void testAttachWithNoRestoredIndicesIsANoOp() { + String ds = "logs-attach"; + IndexMetadata b1 = backingIndex(ds, 1); + Metadata.Builder metadata = Metadata.builder().put(b1, false); + DataStream original = new DataStream(ds, createTimestampField("@timestamp"), List.of(b1.getIndex()), 1); + Map updatedDataStreams = new HashMap<>(); + updatedDataStreams.put(ds, original); + + RestoreService.attachRestoredBackingIndices(Collections.emptySet(), metadata, updatedDataStreams); + + assertEquals(original, updatedDataStreams.get(ds)); + } + public void testValidateReplicationTypeRestoreSettings_WhenSnapshotIsDocument_RestoreToDocument() { SnapshotId snapshotId = new SnapshotId("snapshotId", "123"); Snapshot snapshot = new Snapshot("testRepo", snapshotId);