From 3a1be63f3445bb38bea5898742a2b195c1c26251 Mon Sep 17 00:00:00 2001 From: "opensearch-trigger-bot[bot]" <98922864+opensearch-trigger-bot[bot]@users.noreply.github.com> Date: Thu, 4 Jul 2024 11:36:14 +0530 Subject: [PATCH] Clear templates before Adding; Use NamedWriteableAwareStreamInput for RemoteCustomMetadata; Correct the check for deciding upload of HashesOfConsistentSettings (#14513) (#14639) * Clear templates before Adding; Use NamedWriteableAwareStreamInput for RemoteCustomMetadata * Correct the check for deciding upload of hashes of consistent settings (cherry picked from commit 501a7024d18f7c7bc135b5e686835f0bcdf00d30) Signed-off-by: Sooraj Sinha Signed-off-by: github-actions[bot] Co-authored-by: github-actions[bot] --- .../opensearch/cluster/metadata/Metadata.java | 1 + .../remote/RemoteClusterStateService.java | 5 +- .../remote/model/RemoteCustomMetadata.java | 5 +- .../cluster/metadata/MetadataTests.java | 27 ++++ .../RemoteClusterStateServiceTests.java | 125 +++++++++++++++++- .../model/RemoteCustomMetadataTests.java | 87 +++++++++++- 6 files changed, 239 insertions(+), 11 deletions(-) diff --git a/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java b/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java index e2562ab0902e7..04189a3c0270a 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java @@ -1312,6 +1312,7 @@ public Builder templates(Map templates) { } public Builder templates(TemplatesMetadata templatesMetadata) { + this.templates.clear(); this.templates.putAll(templatesMetadata.getTemplates()); return this; } diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java index ef14adeb42b68..74abe9cd257b4 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -356,7 +356,7 @@ public RemoteClusterStateManifestInfo writeIncrementalMetadata( && clusterState.getNodes().delta(previousClusterState.getNodes()).hasChanges(); final boolean updateClusterBlocks = isPublicationEnabled && !clusterState.blocks().equals(previousClusterState.blocks()); final boolean updateHashesOfConsistentSettings = isPublicationEnabled - || Metadata.isHashesOfConsistentSettingsEqual(previousClusterState.metadata(), clusterState.metadata()) == false; + && Metadata.isHashesOfConsistentSettingsEqual(previousClusterState.metadata(), clusterState.metadata()) == false; uploadedMetadataResults = writeMetadataInParallel( clusterState, @@ -476,7 +476,8 @@ public RemoteClusterStateManifestInfo writeIncrementalMetadata( return manifestDetails; } - private UploadedMetadataResults writeMetadataInParallel( + // package private for testing + UploadedMetadataResults writeMetadataInParallel( ClusterState clusterState, List indexToUpload, Map prevIndexMetadataByName, diff --git a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java index 4c7069ee8be9e..ec5dfbec820d4 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java +++ b/server/src/main/java/org/opensearch/gateway/remote/model/RemoteCustomMetadata.java @@ -12,6 +12,7 @@ import org.opensearch.common.io.Streams; import org.opensearch.common.remote.AbstractRemoteWritableBlobEntity; import org.opensearch.common.remote.BlobPathParameters; +import org.opensearch.core.common.io.stream.NamedWriteableAwareStreamInput; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.compress.Compressor; @@ -122,6 +123,8 @@ public UploadedMetadata getUploadedMetadata() { public static Custom readFrom(StreamInput streamInput, NamedWriteableRegistry namedWriteableRegistry, String customType) throws IOException { - return namedWriteableRegistry.getReader(Custom.class, customType).read(streamInput); + try (StreamInput in = new NamedWriteableAwareStreamInput(streamInput, namedWriteableRegistry)) { + return namedWriteableRegistry.getReader(Custom.class, customType).read(in); + } } } diff --git a/server/src/test/java/org/opensearch/cluster/metadata/MetadataTests.java b/server/src/test/java/org/opensearch/cluster/metadata/MetadataTests.java index 618fcb923bc60..a434a713f330b 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/MetadataTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/MetadataTests.java @@ -1482,6 +1482,33 @@ public void testIsSegmentReplicationDisabled() { assertFalse(metadata.isSegmentReplicationEnabled(indexName)); } + public void testTemplatesMetadata() { + TemplatesMetadata templatesMetadata1 = TemplatesMetadata.builder() + .put( + IndexTemplateMetadata.builder("template_1") + .patterns(Arrays.asList("bar-*", "foo-*")) + .settings(Settings.builder().put("random_index_setting_" + randomAlphaOfLength(3), randomAlphaOfLength(5)).build()) + .build() + ) + .build(); + Metadata metadata1 = Metadata.builder().templates(templatesMetadata1).build(); + assertThat(metadata1.templates(), is(templatesMetadata1.getTemplates())); + + TemplatesMetadata templatesMetadata2 = TemplatesMetadata.builder() + .put( + IndexTemplateMetadata.builder("template_2") + .patterns(Arrays.asList("bar-*", "foo-*")) + .settings(Settings.builder().put("random_index_setting_" + randomAlphaOfLength(3), randomAlphaOfLength(5)).build()) + .build() + ) + .build(); + + Metadata metadata2 = Metadata.builder(metadata1).templates(templatesMetadata2).build(); + + assertThat(metadata2.templates(), is(templatesMetadata2.getTemplates())); + + } + public static Metadata randomMetadata() { Metadata.Builder md = Metadata.builder() .put(buildIndexMetadata("index", "alias", randomBoolean() ? null : randomBoolean()).build(), randomBoolean()) diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java index c8fd982fec1e1..d983a4d8c4027 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java @@ -20,6 +20,7 @@ import org.opensearch.cluster.metadata.IndexTemplateMetadata; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.TemplatesMetadata; +import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.cluster.routing.RoutingTable; import org.opensearch.cluster.routing.remote.InternalRemoteRoutingTableService; @@ -92,6 +93,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.ArgumentMatchers; +import org.mockito.Mockito; import static java.util.stream.Collectors.toList; import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL; @@ -111,6 +113,7 @@ import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY; +import static org.hamcrest.Matchers.anEmptyMap; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.not; @@ -118,6 +121,7 @@ import static org.hamcrest.Matchers.nullValue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -518,11 +522,13 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder().indices(Collections.emptyList()).build(); remoteClusterStateService.start(); - final ClusterMetadataManifest manifest = remoteClusterStateService.writeIncrementalMetadata( + final RemoteClusterStateService rcssSpy = Mockito.spy(remoteClusterStateService); + final RemoteClusterStateManifestInfo manifestInfo = rcssSpy.writeIncrementalMetadata( previousClusterState, clusterState, previousManifest - ).getClusterMetadataManifest(); + ); + final ClusterMetadataManifest manifest = manifestInfo.getClusterMetadataManifest(); final UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "index-uuid", "metadata-filename__2"); final List indices = List.of(uploadedIndexMetadata); @@ -535,6 +541,24 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { .previousClusterUUID("prev-cluster-uuid") .build(); + Mockito.verify(rcssSpy) + .writeMetadataInParallel( + eq(clusterState), + eq(new ArrayList(clusterState.metadata().indices().values())), + eq(Collections.singletonMap(indices.get(0).getIndexName(), null)), + eq(clusterState.metadata().customs()), + eq(true), + eq(true), + eq(true), + eq(false), + eq(false), + eq(false), + eq(Collections.emptyMap()), + eq(false), + eq(Collections.emptyList()) + ); + + assertThat(manifestInfo.getManifestFileName(), notNullValue()); assertThat(manifest.getIndices().size(), is(1)); assertThat(manifest.getIndices().get(0).getIndexName(), is(uploadedIndexMetadata.getIndexName())); assertThat(manifest.getIndices().get(0).getIndexUUID(), is(uploadedIndexMetadata.getIndexUUID())); @@ -543,6 +567,95 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { assertThat(manifest.getStateVersion(), is(expectedManifest.getStateVersion())); assertThat(manifest.getClusterUUID(), is(expectedManifest.getClusterUUID())); assertThat(manifest.getStateUUID(), is(expectedManifest.getStateUUID())); + assertThat(manifest.getHashesOfConsistentSettings(), nullValue()); + assertThat(manifest.getDiscoveryNodesMetadata(), nullValue()); + assertThat(manifest.getClusterBlocksMetadata(), nullValue()); + assertThat(manifest.getClusterStateCustomMap(), anEmptyMap()); + assertThat(manifest.getTransientSettingsMetadata(), nullValue()); + assertThat(manifest.getTemplatesMetadata(), notNullValue()); + assertThat(manifest.getCoordinationMetadata(), notNullValue()); + assertThat(manifest.getCustomMetadataMap().size(), is(2)); + assertThat(manifest.getIndicesRouting().size(), is(0)); + } + + public void testWriteIncrementalMetadataSuccessWhenPublicationEnabled() throws IOException { + publicationEnabled = true; + Settings nodeSettings = Settings.builder().put(REMOTE_PUBLICATION_EXPERIMENTAL, publicationEnabled).build(); + FeatureFlags.initializeFeatureFlags(nodeSettings); + remoteClusterStateService = new RemoteClusterStateService( + "test-node-id", + repositoriesServiceSupplier, + settings, + clusterService, + () -> 0L, + threadPool, + List.of(new RemoteIndexPathUploader(threadPool, settings, repositoriesServiceSupplier, clusterSettings)), + writableRegistry() + ); + final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); + mockBlobStoreObjects(); + final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); + final ClusterState previousClusterState = ClusterState.builder(ClusterName.DEFAULT) + .metadata(Metadata.builder().coordinationMetadata(coordinationMetadata)) + .build(); + + final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder().indices(Collections.emptyList()).build(); + + remoteClusterStateService.start(); + final RemoteClusterStateService rcssSpy = Mockito.spy(remoteClusterStateService); + final RemoteClusterStateManifestInfo manifestInfo = rcssSpy.writeIncrementalMetadata( + previousClusterState, + clusterState, + previousManifest + ); + final ClusterMetadataManifest manifest = manifestInfo.getClusterMetadataManifest(); + final UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "index-uuid", "metadata-filename__2"); + final List indices = List.of(uploadedIndexMetadata); + + final ClusterMetadataManifest expectedManifest = ClusterMetadataManifest.builder() + .indices(indices) + .clusterTerm(1L) + .stateVersion(1L) + .stateUUID("state-uuid") + .clusterUUID("cluster-uuid") + .previousClusterUUID("prev-cluster-uuid") + .build(); + + Mockito.verify(rcssSpy) + .writeMetadataInParallel( + eq(clusterState), + eq(new ArrayList(clusterState.metadata().indices().values())), + eq(Collections.singletonMap(indices.get(0).getIndexName(), null)), + eq(clusterState.metadata().customs()), + eq(true), + eq(true), + eq(true), + eq(true), + eq(false), + eq(false), + eq(Collections.emptyMap()), + eq(true), + Mockito.anyList() + ); + + assertThat(manifestInfo.getManifestFileName(), notNullValue()); + assertThat(manifest.getIndices().size(), is(1)); + assertThat(manifest.getIndices().get(0).getIndexName(), is(uploadedIndexMetadata.getIndexName())); + assertThat(manifest.getIndices().get(0).getIndexUUID(), is(uploadedIndexMetadata.getIndexUUID())); + assertThat(manifest.getIndices().get(0).getUploadedFilename(), notNullValue()); + assertThat(manifest.getClusterTerm(), is(expectedManifest.getClusterTerm())); + assertThat(manifest.getStateVersion(), is(expectedManifest.getStateVersion())); + assertThat(manifest.getClusterUUID(), is(expectedManifest.getClusterUUID())); + assertThat(manifest.getStateUUID(), is(expectedManifest.getStateUUID())); + assertThat(manifest.getHashesOfConsistentSettings(), notNullValue()); + assertThat(manifest.getDiscoveryNodesMetadata(), notNullValue()); + assertThat(manifest.getClusterBlocksMetadata(), nullValue()); + assertThat(manifest.getClusterStateCustomMap(), anEmptyMap()); + assertThat(manifest.getTransientSettingsMetadata(), nullValue()); + assertThat(manifest.getTemplatesMetadata(), notNullValue()); + assertThat(manifest.getCoordinationMetadata(), notNullValue()); + assertThat(manifest.getCustomMetadataMap().size(), is(2)); + assertThat(manifest.getIndicesRouting().size(), is(1)); } /* @@ -2012,7 +2125,9 @@ static ClusterState.Builder generateClusterStateWithOneIndex() { .build(); final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); final Settings settings = Settings.builder().put("mock-settings", true).build(); - final TemplatesMetadata templatesMetadata = TemplatesMetadata.EMPTY_METADATA; + final TemplatesMetadata templatesMetadata = TemplatesMetadata.builder() + .put(IndexTemplateMetadata.builder("template1").settings(idxSettings).patterns(List.of("test*")).build()) + .build(); final CustomMetadata1 customMetadata1 = new CustomMetadata1("custom-metadata-1"); return ClusterState.builder(ClusterName.DEFAULT) .version(1L) @@ -2025,6 +2140,7 @@ static ClusterState.Builder generateClusterStateWithOneIndex() { .coordinationMetadata(coordinationMetadata) .persistentSettings(settings) .templates(templatesMetadata) + .hashesOfConsistentSettings(Map.of("key1", "value1", "key2", "value2")) .putCustom(customMetadata1.getWriteableName(), customMetadata1) .build() ) @@ -2032,7 +2148,8 @@ static ClusterState.Builder generateClusterStateWithOneIndex() { } static DiscoveryNodes nodesWithLocalNodeClusterManager() { - return DiscoveryNodes.builder().clusterManagerNodeId("cluster-manager-id").localNodeId("cluster-manager-id").build(); + final DiscoveryNode localNode = new DiscoveryNode("cluster-manager-id", buildNewFakeTransportAddress(), Version.CURRENT); + return DiscoveryNodes.builder().clusterManagerNodeId("cluster-manager-id").localNodeId("cluster-manager-id").add(localNode).build(); } private static class CustomMetadata1 extends TestCustomMetadata { diff --git a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java index 1e28817be79f2..60cceb205f43d 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/model/RemoteCustomMetadataTests.java @@ -8,6 +8,8 @@ package org.opensearch.gateway.remote.model; +import org.opensearch.Version; +import org.opensearch.cluster.ClusterModule; import org.opensearch.cluster.metadata.IndexGraveyard; import org.opensearch.cluster.metadata.Metadata.Custom; import org.opensearch.common.blobstore.BlobPath; @@ -16,13 +18,20 @@ import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; +import org.opensearch.core.common.io.stream.NamedWriteableRegistry.Entry; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; import org.opensearch.core.compress.Compressor; import org.opensearch.core.compress.NoneCompressor; import org.opensearch.core.index.Index; +import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadata; import org.opensearch.gateway.remote.RemoteClusterStateUtils; import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.translog.transfer.BlobStoreTransferService; +import org.opensearch.persistent.PersistentTaskParams; +import org.opensearch.persistent.PersistentTasksCustomMetadata; +import org.opensearch.persistent.PersistentTasksCustomMetadata.Assignment; import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.test.OpenSearchTestCase; import org.opensearch.threadpool.TestThreadPool; @@ -33,6 +42,7 @@ import java.io.IOException; import java.io.InputStream; import java.util.List; +import java.util.Objects; import static org.opensearch.gateway.remote.RemoteClusterStateUtils.GLOBAL_METADATA_CURRENT_CODEC_VERSION; import static org.opensearch.gateway.remote.model.RemoteCustomMetadata.CUSTOM_DELIMITER; @@ -216,24 +226,93 @@ public void testGetUploadedMetadata() throws IOException { public void testSerDe() throws IOException { Custom customMetadata = getCustomMetadata(); + verifySerDe(customMetadata, IndexGraveyard.TYPE); + } + + public void testSerDeForPersistentTasks() throws IOException { + Custom customMetadata = getPersistentTasksMetadata(); + verifySerDe(customMetadata, PersistentTasksCustomMetadata.TYPE); + } + + private void verifySerDe(Custom objectToUpload, String objectType) throws IOException { RemoteCustomMetadata remoteObjectForUpload = new RemoteCustomMetadata( - customMetadata, - IndexGraveyard.TYPE, + objectToUpload, + objectType, METADATA_VERSION, clusterUUID, compressor, - namedWriteableRegistry + customWritableRegistry() ); try (InputStream inputStream = remoteObjectForUpload.serialize()) { remoteObjectForUpload.setFullBlobName(BlobPath.cleanPath()); assertThat(inputStream.available(), greaterThan(0)); Custom readCustomMetadata = remoteObjectForUpload.deserialize(inputStream); - assertThat(readCustomMetadata, is(customMetadata)); + assertThat(readCustomMetadata, is(objectToUpload)); } } + private NamedWriteableRegistry customWritableRegistry() { + List entries = ClusterModule.getNamedWriteables(); + entries.add(new Entry(PersistentTaskParams.class, TestPersistentTaskParams.PARAM_NAME, TestPersistentTaskParams::new)); + return new NamedWriteableRegistry(entries); + } + public static Custom getCustomMetadata() { return IndexGraveyard.builder().addTombstone(new Index("test-index", "3q2423")).build(); } + private static Custom getPersistentTasksMetadata() { + return PersistentTasksCustomMetadata.builder() + .addTask("_task_1", "testTaskName", new TestPersistentTaskParams("task param data"), new Assignment(null, "_reason")) + .build(); + } + + public static class TestPersistentTaskParams implements PersistentTaskParams { + + private static final String PARAM_NAME = "testTaskName"; + + private final String data; + + public TestPersistentTaskParams(String data) { + this.data = data; + } + + public TestPersistentTaskParams(StreamInput in) throws IOException { + this(in.readString()); + } + + @Override + public String getWriteableName() { + return PARAM_NAME; + } + + @Override + public Version getMinimalSupportedVersion() { + return Version.V_2_13_0; + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeString(data); + } + + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + return builder.startObject().field("data_field", data); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + TestPersistentTaskParams that = (TestPersistentTaskParams) o; + return Objects.equals(data, that.data); + } + + @Override + public int hashCode() { + return Objects.hash(data); + } + } + }