From 5d6a0ad25b77c2ba69e8480559b428893df671b6 Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Fri, 9 Feb 2024 01:43:30 +0530 Subject: [PATCH] Upload all the metadata attributes in parallel Signed-off-by: Shivansh Arora --- .../cluster/metadata/IndexMetadata.java | 4 + .../opensearch/cluster/metadata/Metadata.java | 33 +- .../cluster/metadata/TemplatesMetadata.java | 11 +- .../remote/ClusterMetadataManifest.java | 140 +++- .../remote/RemoteClusterStateService.java | 688 ++++++++---------- .../remote/ClusterMetadataManifestTests.java | 45 +- .../RemoteClusterStateServiceTests.java | 600 +++++++++++---- 7 files changed, 964 insertions(+), 557 deletions(-) diff --git a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java index 03784df509ed6..5adb07bd91572 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java @@ -752,6 +752,10 @@ public String getIndexUUID() { return index.getUUID(); } + public String getIndexName() { + return index.getName(); + } + /** * Test whether the current index UUID is the same as the given one. Returns true if either are _na_ */ 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 464a6e89effbd..0141a74d01d0c 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/Metadata.java @@ -948,7 +948,7 @@ public Iterator iterator() { } public static boolean isGlobalStateEquals(Metadata metadata1, Metadata metadata2) { - if (!metadata1.coordinationMetadata.equals(metadata2.coordinationMetadata)) { + if (!isCoordinationMetadataEqual(metadata1, metadata2)) { return false; } if (!metadata1.hashesOfConsistentSettings.equals(metadata2.hashesOfConsistentSettings)) { @@ -967,13 +967,29 @@ public static boolean isGlobalStateEquals(Metadata metadata1, Metadata metadata2 * Compares Metadata entities persisted in Remote Store. */ public static boolean isGlobalResourcesMetadataEquals(Metadata metadata1, Metadata metadata2) { - if (!metadata1.persistentSettings.equals(metadata2.persistentSettings)) { + if (!isSettingsMetadataEqual(metadata1, metadata2)) { return false; } - if (!metadata1.templates.equals(metadata2.templates)) { + if (!isTemplatesMetadataEqual(metadata1, metadata2)) { return false; } // Check if any persistent metadata needs to be saved + return isCustomMetadataEqual(metadata1, metadata2); + } + + public static boolean isCoordinationMetadataEqual(Metadata metadata1, Metadata metadata2) { + return metadata1.coordinationMetadata.equals(metadata2.coordinationMetadata); + } + + public static boolean isSettingsMetadataEqual(Metadata metadata1, Metadata metadata2) { + return metadata1.persistentSettings.equals(metadata2.persistentSettings); + } + + public static boolean isTemplatesMetadataEqual(Metadata metadata1, Metadata metadata2) { + return metadata1.templates.equals(metadata2.templates); + } + + public static boolean isCustomMetadataEqual(Metadata metadata1, Metadata metadata2) { int customCount1 = 0; for (Map.Entry cursor : metadata1.customs.entrySet()) { if (cursor.getValue().context().contains(XContentContext.GATEWAY)) { @@ -987,8 +1003,7 @@ public static boolean isGlobalResourcesMetadataEquals(Metadata metadata1, Metada customCount2++; } } - if (customCount1 != customCount2) return false; - return true; + return customCount1 == customCount2; } @Override @@ -1037,7 +1052,11 @@ private static class MetadataDiff implements Diff { persistentSettings = after.persistentSettings; hashesOfConsistentSettings = after.hashesOfConsistentSettings.diff(before.hashesOfConsistentSettings); indices = DiffableUtils.diff(before.indices, after.indices, DiffableUtils.getStringKeySerializer()); - templates = DiffableUtils.diff(before.templates.getTemplates(), after.templates.getTemplates(), DiffableUtils.getStringKeySerializer()); + templates = DiffableUtils.diff( + before.templates.getTemplates(), + after.templates.getTemplates(), + DiffableUtils.getStringKeySerializer() + ); customs = DiffableUtils.diff(before.customs, after.customs, DiffableUtils.getStringKeySerializer(), CUSTOM_VALUE_SERIALIZER); } @@ -1765,7 +1784,9 @@ public static void toXContent(Metadata metadata, XContentBuilder builder, ToXCon builder.endObject(); } + builder.startObject("templates"); metadata.templatesMetadata().toXContent(builder, params); + builder.endObject(); if (context == XContentContext.API) { builder.startObject("indices"); diff --git a/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java b/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java index ff4c7f62fd9c5..179d5aa6fbc52 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java @@ -9,9 +9,6 @@ package org.opensearch.cluster.metadata; import org.opensearch.cluster.AbstractDiffable; -import org.opensearch.cluster.coordination.CoordinationMetadata; -import org.opensearch.common.annotation.PublicApi; -import org.opensearch.common.settings.Settings; import org.opensearch.core.common.io.stream.StreamOutput; import org.opensearch.core.xcontent.ToXContentFragment; import org.opensearch.core.xcontent.XContentBuilder; @@ -66,7 +63,7 @@ public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - TemplatesMetadata that = (TemplatesMetadata) o; + TemplatesMetadata that = (TemplatesMetadata) o; return Objects.equals(templates, that.templates); } @@ -111,11 +108,11 @@ public TemplatesMetadata build() { } public static void toXContent(TemplatesMetadata templates, XContentBuilder builder, Params params) throws IOException { -// builder.startObject("templates-metadata"); - for(IndexTemplateMetadata cursor : templates.getTemplates().values()) { + // builder.startObject("templates-metadata"); + for (IndexTemplateMetadata cursor : templates.getTemplates().values()) { IndexTemplateMetadata.Builder.toXContentWithTypes(cursor, builder, params); } -// builder.endObject(); + // builder.endObject(); } public static TemplatesMetadata fromXContent(XContentParser parser) throws IOException { diff --git a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java index a9f59fc8b1b8c..a3a84924e4563 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java +++ b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -210,10 +211,26 @@ private static void declareParser(ConstructingObjectParser getCustomMetadataMap() { } public boolean hasMetadataAttributesFiles() { - return uploadedCoordinationMetadata != null || uploadedSettingsMetadata != null || uploadedTemplatesMetadata != null || uploadedCustomMetadataMap != null; + return uploadedCoordinationMetadata != null + || uploadedSettingsMetadata != null + || uploadedTemplatesMetadata != null + || !uploadedCustomMetadataMap.isEmpty(); } public ClusterMetadataManifest( @@ -369,7 +389,9 @@ public ClusterMetadataManifest( this.uploadedCoordinationMetadata = uploadedCoordinationMetadata; this.uploadedSettingsMetadata = uploadedSettingsMetadata; this.uploadedTemplatesMetadata = uploadedTemplatesMetadata; - this.uploadedCustomMetadataMap = Collections.unmodifiableMap(uploadedCustomMetadataMap); + this.uploadedCustomMetadataMap = Collections.unmodifiableMap( + uploadedCustomMetadataMap != null ? uploadedCustomMetadataMap : new HashMap<>() + ); } public ClusterMetadataManifest(StreamInput in) throws IOException { @@ -388,7 +410,9 @@ public ClusterMetadataManifest(StreamInput in) throws IOException { this.uploadedCoordinationMetadata = new UploadedMetadataAttribute(in); this.uploadedSettingsMetadata = new UploadedMetadataAttribute(in); this.uploadedTemplatesMetadata = new UploadedMetadataAttribute(in); - this.uploadedCustomMetadataMap = Collections.unmodifiableMap(in.readMap(StreamInput::readString, UploadedMetadataAttribute::new)); + this.uploadedCustomMetadataMap = Collections.unmodifiableMap( + in.readMap(StreamInput::readString, UploadedMetadataAttribute::new) + ); this.globalMetadataFileName = null; } else if (in.getVersion().onOrAfter(Version.V_2_12_0)) { this.codecVersion = in.readInt(); @@ -435,17 +459,23 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws builder.field(CLUSTER_UUID_COMMITTED.getPreferredName(), isClusterUUIDCommitted()); if (onOrAfterCodecVersion(CODEC_V2)) { builder.field(CODEC_VERSION_FIELD.getPreferredName(), getCodecVersion()); - builder.startObject(UPLOADED_COORDINATOR_METADATA.getPreferredName()); - getCoordinationMetadata().toXContent(builder, params); - builder.endObject(); - builder.startObject(UPLOADED_SETTINGS_METADATA.getPreferredName()); - getSettingsMetadata().toXContent(builder, params); - builder.endObject(); - builder.startObject(UPLOADED_TEMPLATES_METADATA.getPreferredName()); - getTemplatesMetadata().toXContent(builder, params); - builder.endObject(); + if (getCoordinationMetadata() != null) { + builder.startObject(UPLOADED_COORDINATOR_METADATA.getPreferredName()); + getCoordinationMetadata().toXContent(builder, params); + builder.endObject(); + } + if (getSettingsMetadata() != null) { + builder.startObject(UPLOADED_SETTINGS_METADATA.getPreferredName()); + getSettingsMetadata().toXContent(builder, params); + builder.endObject(); + } + if (getTemplatesMetadata() != null) { + builder.startObject(UPLOADED_TEMPLATES_METADATA.getPreferredName()); + getTemplatesMetadata().toXContent(builder, params); + builder.endObject(); + } builder.startObject(UPLOADED_CUSTOM_METADATA.getPreferredName()); - for (UploadedMetadataAttribute attribute: getCustomMetadataMap().values()) { + for (UploadedMetadataAttribute attribute : getCustomMetadataMap().values()) { attribute.toXContent(builder, params); } builder.endObject(); @@ -534,6 +564,10 @@ public static ClusterMetadataManifest fromXContentV0(XContentParser parser) thro return PARSER_V0.parse(parser, null); } + public static ClusterMetadataManifest fromXContentV1(XContentParser parser) throws IOException { + return PARSER_V1.parse(parser, null); + } + public static ClusterMetadataManifest fromXContent(XContentParser parser) throws IOException { return CURRENT_PARSER.parse(parser, null); } @@ -546,6 +580,10 @@ public static ClusterMetadataManifest fromXContent(XContentParser parser) throws public static class Builder { private String globalMetadataFileName; + private UploadedMetadataAttribute coordinationMetadata; + private UploadedMetadataAttribute settingsMetadata; + private UploadedMetadataAttribute templatesMetadata; + private Map customMetadataMap; private int codecVersion; private List indices; private long clusterTerm; @@ -573,6 +611,31 @@ public Builder globalMetadataFileName(String globalMetadataFileName) { return this; } + public Builder coordinationMetadata(UploadedMetadataAttribute coordinationMetadata) { + this.coordinationMetadata = coordinationMetadata; + return this; + } + + public Builder settingMetadata(UploadedMetadataAttribute settingsMetadata) { + this.settingsMetadata = settingsMetadata; + return this; + } + + public Builder templatesMetadata(UploadedMetadataAttribute templatesMetadata) { + this.templatesMetadata = templatesMetadata; + return this; + } + + public Builder customMetadataMap(Map customMetadataMap) { + this.customMetadataMap = customMetadataMap; + return this; + } + + public Builder put(String custom, UploadedMetadataAttribute customMetadata) { + this.customMetadataMap.put(custom, customMetadata); + return this; + } + public Builder clusterTerm(long clusterTerm) { this.clusterTerm = clusterTerm; return this; @@ -624,6 +687,7 @@ public Builder clusterUUIDCommitted(boolean clusterUUIDCommitted) { public Builder() { indices = new ArrayList<>(); + customMetadataMap = new HashMap<>(); } public Builder(ClusterMetadataManifest manifest) { @@ -635,6 +699,10 @@ public Builder(ClusterMetadataManifest manifest) { this.nodeId = manifest.nodeId; this.committed = manifest.committed; this.globalMetadataFileName = manifest.globalMetadataFileName; + this.coordinationMetadata = manifest.uploadedCoordinationMetadata; + this.settingsMetadata = manifest.uploadedSettingsMetadata; + this.templatesMetadata = manifest.uploadedTemplatesMetadata; + this.customMetadataMap = manifest.uploadedCustomMetadataMap; this.codecVersion = manifest.codecVersion; this.indices = new ArrayList<>(manifest.indices); this.previousClusterUUID = manifest.previousClusterUUID; @@ -654,18 +722,28 @@ public ClusterMetadataManifest build() { globalMetadataFileName, indices, previousClusterUUID, - clusterUUIDCommitted + clusterUUIDCommitted, + coordinationMetadata, + settingsMetadata, + templatesMetadata, + customMetadataMap ); } } + public static interface UploadedMetadata { + String getComponent(); + + String getUploadedFilename(); + } + /** * Metadata for uploaded index metadata * * @opensearch.internal */ - public static class UploadedIndexMetadata implements Writeable, ToXContentFragment { + public static class UploadedIndexMetadata implements UploadedMetadata, Writeable, ToXContentFragment { private static final ParseField INDEX_NAME_FIELD = new ParseField("index_name"); private static final ParseField INDEX_UUID_FIELD = new ParseField("index_uuid"); @@ -714,6 +792,11 @@ public String getUploadedFilePath() { return uploadedFilename; } + @Override + public String getComponent() { + return getIndexName(); + } + public String getUploadedFilename() { String[] splitPath = uploadedFilename.split("/"); return splitPath[splitPath.length - 1]; @@ -772,7 +855,7 @@ public static UploadedIndexMetadata fromXContent(XContentParser parser) throws I } } - public static class UploadedMetadataAttribute implements Writeable, ToXContentFragment { + public static class UploadedMetadataAttribute implements UploadedMetadata, Writeable, ToXContentFragment { private static final ParseField UPLOADED_FILENAME_FIELD = new ParseField("uploaded_filename"); private static final ObjectParser.NamedObjectParser PARSER; @@ -807,6 +890,11 @@ public String getAttributeName() { return attributeName; } + @Override + public String getComponent() { + return getAttributeName(); + } + public String getUploadedFilename() { return uploadedFilename; } @@ -830,10 +918,14 @@ public static UploadedMetadataAttribute fromXContent(XContentParser parser) thro @Override public String toString() { - return "UploadedMetadataAttribute{" + - "attributeName='" + attributeName + '\'' + - ", uploadedFilename='" + uploadedFilename + '\'' + - '}'; + return "UploadedMetadataAttribute{" + + "attributeName='" + + attributeName + + '\'' + + ", uploadedFilename='" + + uploadedFilename + + '\'' + + '}'; } } } 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 13a1a565886fa..718197618ccf1 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -18,6 +18,7 @@ import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.Metadata; import org.opensearch.cluster.metadata.TemplatesMetadata; +import org.opensearch.common.CheckedRunnable; import org.opensearch.common.Nullable; import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobMetadata; @@ -29,7 +30,6 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.io.IOUtils; import org.opensearch.core.action.ActionListener; -import org.opensearch.core.index.Index; import org.opensearch.core.xcontent.ToXContent; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; @@ -42,7 +42,6 @@ import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat; import org.opensearch.threadpool.ThreadPool; -import reactor.util.annotation.NonNull; import java.io.Closeable; import java.io.IOException; @@ -67,6 +66,8 @@ import java.util.function.Supplier; import java.util.stream.Collectors; +import reactor.util.annotation.NonNull; + import static org.opensearch.gateway.PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled; @@ -138,8 +139,8 @@ public class RemoteClusterStateService implements Closeable { Settings::fromXContent ); - public static final ChecksumBlobStoreFormat TEMPLATE_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "template", + public static final ChecksumBlobStoreFormat TEMPLATES_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "templates", METADATA_NAME_FORMAT, TemplatesMetadata::fromXContent ); @@ -150,8 +151,6 @@ public class RemoteClusterStateService implements Closeable { Metadata.Custom::fromXContent ); - private static final Map metadataComponentBlobStoreMap = new HashMap<>(); - /** * Manifest format compatible with older codec v0, where codec version was missing. */ @@ -159,7 +158,13 @@ public class RemoteClusterStateService implements Closeable { new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV0); /** - * Manifest format compatible with codec v1, where we introduced codec versions/global metadata. + * Manifest format compatible with older codec v1, where global metadata was missing. + */ + public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT_V1 = + new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV1); + + /** + * Manifest format compatible with codec v2, where we introduced codec versions/global metadata. */ public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT = new ChecksumBlobStoreFormat<>( "cluster-metadata-manifest", @@ -185,7 +190,7 @@ public class RemoteClusterStateService implements Closeable { public static final String METADATA_FILE_PREFIX = "metadata"; public static final String COORDINATION_METADATA = "coordination"; public static final String SETTING_METADATA = "settings"; - public static final String TEMPLATE_METADATA = "template"; + public static final String TEMPLATES_METADATA = "templates"; public static final String CUSTOM_METADATA = "custom"; public static final int SPLITED_MANIFEST_FILE_LENGTH = 6; // file name manifest__term__version__C/P__timestamp__codecversion @@ -240,11 +245,6 @@ public RemoteClusterStateService( clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setGlobalMetadataUploadTimeout); clusterSettings.addSettingsUpdateConsumer(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, this::setMetadataManifestUploadTimeout); this.remoteStateStats = new RemotePersistenceStats(); - - metadataComponentBlobStoreMap.put("coordination", COORDINATION_METADATA_FORMAT); - metadataComponentBlobStoreMap.put("settings", SETTINGS_METADATA_FORMAT); - metadataComponentBlobStoreMap.put("templates", TEMPLATE_METADATA_FORMAT); - metadataComponentBlobStoreMap.put("customs", CUSTOM_METADATA_FORMAT); } private BlobStoreTransferService getBlobStoreTransferService() { @@ -268,26 +268,22 @@ public ClusterMetadataManifest writeFullMetadata(ClusterState clusterState, Stri return null; } - // TODO: we can upload global metadata and index metadata in parallel. [issue: #10645] - // Write globalMetadata fragments - UploadedMetadataAttribute uploadedCoordinationMetadata = writeCoordinationMetadata(clusterState); - UploadedMetadataAttribute uploadedSettingsMetadata = writeSettingsMetadata(clusterState); - UploadedMetadataAttribute uploadedTemplateMetadata = writeTemplateMetadata(clusterState); - Map uploadedCustomMetadataMap = writeCustomMetadataInParallel(clusterState); - - // any validations before/after upload ? - final List allUploadedIndexMetadata = writeIndexMetadataParallel( + UploadedMetadataResults uploadedMetadataResults = writeMetadataInParallel( clusterState, - new ArrayList<>(clusterState.metadata().indices().values()) + new ArrayList<>(clusterState.metadata().indices().values()), + clusterState.metadata().customs(), + true, + true, + true ); final ClusterMetadataManifest manifest = uploadManifest( clusterState, - allUploadedIndexMetadata, + uploadedMetadataResults.uploadedIndexMetadata, previousClusterUUID, - uploadedCoordinationMetadata, - uploadedSettingsMetadata, - uploadedTemplateMetadata, - uploadedCustomMetadataMap, + uploadedMetadataResults.uploadedCoordinationMetadata, + uploadedMetadataResults.uploadedSettingsMetadata, + uploadedMetadataResults.uploadedTemplatesMetadata, + uploadedMetadataResults.uploadedCustomMetadataMap, false ); final long durationMillis = TimeValue.nsecToMSec(relativeTimeNanosSupplier.getAsLong() - startTimeNanos); @@ -298,13 +294,13 @@ public ClusterMetadataManifest writeFullMetadata(ClusterState clusterState, Stri "writing cluster state took [{}ms] which is above the warn threshold of [{}]; " + "wrote full state with [{}] indices", durationMillis, slowWriteLoggingThreshold, - allUploadedIndexMetadata.size() + uploadedMetadataResults.uploadedIndexMetadata.size() ); } else { logger.info( "writing cluster state took [{}ms]; " + "wrote full state with [{}] indices and global metadata", durationMillis, - allUploadedIndexMetadata.size() + uploadedMetadataResults.uploadedIndexMetadata.size() ); } return manifest; @@ -335,23 +331,24 @@ public ClusterMetadataManifest writeIncrementalMetadata( previousClusterState.metadata(), clusterState.metadata() ) == false; - UploadedMetadataAttribute uploadedCoordinationMetadata; - UploadedMetadataAttribute uploadedSettingsMetadata; - UploadedMetadataAttribute uploadedTemplateMetadata; - Map uploadedCustomMetadataMap; - // For migration case from codec V0 or V1 to V2, we have added null check on metadata attribute files, - // If file is empty and codec is 1 then write global metadata. - if (updateGlobalMetadata || !previousManifest.hasMetadataAttributesFiles()) { - uploadedCoordinationMetadata = writeCoordinationMetadata(clusterState); - uploadedSettingsMetadata = writeSettingsMetadata(clusterState); - uploadedTemplateMetadata = writeTemplateMetadata(clusterState); - uploadedCustomMetadataMap = writeCustomMetadataInParallel(clusterState); - } else { - logger.debug("Global metadata has not updated in cluster state, skipping upload of it"); - uploadedCoordinationMetadata = previousManifest.getCoordinationMetadata(); - uploadedSettingsMetadata = previousManifest.getSettingsMetadata(); - uploadedTemplateMetadata = previousManifest.getTemplatesMetadata(); - uploadedCustomMetadataMap = previousManifest.getCustomMetadataMap(); + + final boolean updateCoordinationMetadata = Metadata.isCoordinationMetadataEqual( + previousClusterState.metadata(), + clusterState.metadata() + ) == false; + final boolean updateSettingsMetadata = Metadata.isSettingsMetadataEqual( + previousClusterState.metadata(), + clusterState.metadata() + ) == false; + final boolean updateTemplatesMetadata = Metadata.isTemplatesMetadataEqual( + previousClusterState.metadata(), + clusterState.metadata() + ) == false; + final Map previousStateCustomMap = new HashMap<>(previousManifest.getCustomMetadataMap()); + final Map customsToUpload = getUpdatedCustoms(clusterState, previousClusterState); + final Map allUploadedCustomMap = new HashMap<>(previousManifest.getCustomMetadataMap()); + for (final String custom : clusterState.metadata().customs().keySet()) { + previousStateCustomMap.remove(custom); } // Write Index Metadata @@ -384,12 +381,28 @@ public ClusterMetadataManifest writeIncrementalMetadata( } previousStateIndexMetadataVersionByName.remove(indexMetadata.getIndex().getName()); } + UploadedMetadataResults uploadedMetadataResults; + boolean firstUpload = !previousManifest.hasMetadataAttributesFiles(); + // For migration case from codec V0 or V1 to V2, we have added null check on metadata attribute files, + // If file is empty and codec is 1 then write global metadata. + if (firstUpload) { + uploadedMetadataResults = writeMetadataInParallel(clusterState, toUpload, clusterState.metadata().customs(), true, true, true); + } else { + uploadedMetadataResults = writeMetadataInParallel( + clusterState, + toUpload, + customsToUpload, + updateCoordinationMetadata, + updateSettingsMetadata, + updateTemplatesMetadata + ); + } - List uploadedIndexMetadataList = writeIndexMetadataParallel(clusterState, toUpload); - uploadedIndexMetadataList.forEach( + uploadedMetadataResults.uploadedIndexMetadata.forEach( uploadedIndexMetadata -> allUploadedIndexMetadata.put(uploadedIndexMetadata.getIndexName(), uploadedIndexMetadata) ); - + allUploadedCustomMap.putAll(uploadedMetadataResults.uploadedCustomMetadataMap); + previousStateCustomMap.keySet().forEach(allUploadedCustomMap::remove); for (String removedIndexName : previousStateIndexMetadataVersionByName.keySet()) { allUploadedIndexMetadata.remove(removedIndexName); } @@ -397,10 +410,16 @@ public ClusterMetadataManifest writeIncrementalMetadata( clusterState, new ArrayList<>(allUploadedIndexMetadata.values()), previousManifest.getPreviousClusterUUID(), - uploadedCoordinationMetadata, - uploadedSettingsMetadata, - uploadedTemplateMetadata, - uploadedCustomMetadataMap, + firstUpload || updateCoordinationMetadata + ? uploadedMetadataResults.uploadedCoordinationMetadata + : previousManifest.getCoordinationMetadata(), + firstUpload || updateSettingsMetadata + ? uploadedMetadataResults.uploadedSettingsMetadata + : previousManifest.getSettingsMetadata(), + firstUpload || updateTemplatesMetadata + ? uploadedMetadataResults.uploadedTemplatesMetadata + : previousManifest.getTemplatesMetadata(), + firstUpload || !customsToUpload.isEmpty() ? allUploadedCustomMap : previousManifest.getCustomMetadataMap(), false ); deleteStaleClusterMetadata(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), RETAINED_MANIFESTS); @@ -411,344 +430,170 @@ public ClusterMetadataManifest writeIncrementalMetadata( if (durationMillis >= slowWriteLoggingThreshold.getMillis()) { logger.warn( "writing cluster state took [{}ms] which is above the warn threshold of [{}]; " - + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, global metadata updated : [{}]", + + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, coordination metadata updated : [{}], " + + "settings metadata updated : [{}], templates metadata updated : [{}], custom metadata updated : [{}]", durationMillis, slowWriteLoggingThreshold, numIndicesUpdated, numIndicesUnchanged, - updateGlobalMetadata + updateCoordinationMetadata, + updateSettingsMetadata, + updateTemplatesMetadata, + customsToUpload.size() ); } else { logger.info( "writing cluster state for version [{}] took [{}ms]; " - + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, global metadata updated : [{}]", + + "wrote metadata for [{}] indices and skipped [{}] unchanged indices, coordination metadata updated : [{}], " + + "settings metadata updated : [{}], templates metadata updated : [{}], custom metadata updated : [{}]", manifest.getStateVersion(), durationMillis, numIndicesUpdated, numIndicesUnchanged, - updateGlobalMetadata + updateCoordinationMetadata, + updateSettingsMetadata, + updateTemplatesMetadata, + customsToUpload.size() ); } return manifest; } - /** - * Uploads provided ClusterState's global Metadata to remote store in parallel. - * The call is blocking so the method waits for upload to finish and then return. - * - * @param clusterState current ClusterState - * @return String file name where globalMetadata file is stored. - */ - private String writeGlobalMetadata(ClusterState clusterState) throws IOException { - - AtomicReference result = new AtomicReference(); - AtomicReference exceptionReference = new AtomicReference(); - - final BlobContainer globalMetadataContainer = globalMetadataContainer( - clusterState.getClusterName().value(), - clusterState.metadata().clusterUUID() - ); - final String globalMetadataFilename = globalMetadataFileName(clusterState.metadata()); - - // latch to wait until upload is not finished - CountDownLatch latch = new CountDownLatch(1); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap(resp -> { - logger.trace(String.format(Locale.ROOT, "GlobalMetadata uploaded successfully.")); - result.set(globalMetadataContainer.path().buildAsString() + globalMetadataFilename); - }, ex -> { exceptionReference.set(ex); }), latch); - - GLOBAL_METADATA_FORMAT.writeAsyncWithUrgentPriority( - clusterState.metadata(), - globalMetadataContainer, - globalMetadataFilename, - blobStoreRepository.getCompressor(), - completionListener, - FORMAT_PARAMS - ); - - try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - // TODO: We should add metrics where transfer is timing out. [Issue: #10687] - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of global metadata to complete") - ); - throw ex; - } - } catch (InterruptedException ex) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of global metadata to complete - %s"), - ex - ); - Thread.currentThread().interrupt(); - throw exception; - } - if (exceptionReference.get() != null) { - throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); - } - return result.get(); - } - - private UploadedMetadataAttribute writeCoordinationMetadata(ClusterState clusterState) throws IOException { - AtomicReference result = new AtomicReference<>(); - AtomicReference exceptionReference = new AtomicReference<>(); - CountDownLatch latch = new CountDownLatch(1); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap( - (UploadedMetadataAttribute uploadedMetadataAttribute) -> { - logger.trace(String.format(Locale.ROOT, "CoordinationMetadata uploaded successfully.")); - result.set(uploadedMetadataAttribute); - }, ex -> { - logger.error( - () -> new ParameterizedMessage("Exception during transfer of Metadata Fragment to Remote {}", ex.getMessage()), - ex - ); - exceptionReference.set(ex); - }), - latch); - - writeMetadataComponentAsync(clusterState, COORDINATION_METADATA, COORDINATION_METADATA_FORMAT, clusterState.metadata().coordinationMetadata(), completionListener); - - try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of coordination metadata to complete") - ); - throw ex; - } - } catch (InterruptedException ex) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of coordination metadata to complete - %s"), - ex - ); - Thread.currentThread().interrupt(); - throw exception; - } - if (exceptionReference.get() != null) { - throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); - } - return result.get(); - } - - private UploadedMetadataAttribute writeSettingsMetadata(ClusterState clusterState) throws IOException { - AtomicReference result = new AtomicReference<>(); - AtomicReference exceptionReference = new AtomicReference<>(); - CountDownLatch latch = new CountDownLatch(1); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap( - (UploadedMetadataAttribute uploadedMetadataAttribute) -> { - logger.trace(String.format(Locale.ROOT, "Settings Metadata uploaded successfully.")); - result.set(uploadedMetadataAttribute); - }, ex -> { - logger.error( - () -> new ParameterizedMessage("Exception during transfer of Metadata Fragment to Remote {}", ex.getMessage()), - ex - ); - exceptionReference.set(ex); - }), - latch); - - writeMetadataComponentAsync(clusterState, SETTING_METADATA, SETTINGS_METADATA_FORMAT, clusterState.metadata().persistentSettings(), completionListener); - - try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of settings metadata to complete") - ); - throw ex; - } - } catch (InterruptedException ex) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of settings metadata to complete - %s"), - ex - ); - Thread.currentThread().interrupt(); - throw exception; - } - if (exceptionReference.get() != null) { - throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); - } - return result.get(); - } - - private UploadedMetadataAttribute writeTemplateMetadata(ClusterState clusterState) throws IOException { - AtomicReference result = new AtomicReference<>(); - AtomicReference exceptionReference = new AtomicReference<>(); - CountDownLatch latch = new CountDownLatch(1); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap( - (UploadedMetadataAttribute uploadedMetadataAttribute) -> { - logger.trace(String.format(Locale.ROOT, "Templates Metadata uploaded successfully.")); - result.set(uploadedMetadataAttribute); - }, ex -> { - logger.error( - () -> new ParameterizedMessage("Exception during transfer of Metadata Fragment to Remote {}", ex.getMessage()), - ex - ); - exceptionReference.set(ex); - }), - latch); - - writeMetadataComponentAsync(clusterState, TEMPLATE_METADATA, TEMPLATE_METADATA_FORMAT, clusterState.metadata().templatesMetadata(), completionListener); - - try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of templates metadata to complete") - ); - throw ex; - } - } catch (InterruptedException ex) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of templates metadata to complete - %s"), - ex - ); - Thread.currentThread().interrupt(); - throw exception; - } - if (exceptionReference.get() != null) { - throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); - } - return result.get(); - } - - private Map writeCustomMetadataInParallel(ClusterState clusterState) throws IOException { - Map result = new HashMap<>(); - List exceptionReference = new ArrayList<>(); - - Map customToUpload = clusterState.metadata().customs(); - - final CountDownLatch latch = new CountDownLatch(customToUpload.size()); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap( - (UploadedMetadataAttribute uploadedMetadataAttribute) -> { - logger.trace(String.format(Locale.ROOT, "Custom Metadata uploaded successfully.")); - result.put(uploadedMetadataAttribute.getAttributeName(), uploadedMetadataAttribute); + private UploadedMetadataResults writeMetadataInParallel( + ClusterState clusterState, + List indexToUpload, + Map customToUpload, + boolean uploadCoordinationMetadata, + boolean uploadSettingsMetadata, + boolean uploadTemplateMetadata + ) throws IOException { + int totalUploadTasks = indexToUpload.size() + customToUpload.size() + (uploadCoordinationMetadata ? 1 : 0) + (uploadSettingsMetadata + ? 1 + : 0) + (uploadTemplateMetadata ? 1 : 0); + CountDownLatch latch = new CountDownLatch(totalUploadTasks); + List> uploadTasks = new ArrayList<>(totalUploadTasks); + Map results = new HashMap<>(totalUploadTasks); + List exceptionList = Collections.synchronizedList(new ArrayList<>(totalUploadTasks)); + + LatchedActionListener listener = new LatchedActionListener<>( + ActionListener.wrap((ClusterMetadataManifest.UploadedMetadata uploadedMetadata) -> { + logger.trace(String.format(Locale.ROOT, "Metadata component %s uploaded successfully.", uploadedMetadata.getComponent())); + results.put(uploadedMetadata.getComponent(), uploadedMetadata); }, ex -> { logger.error( () -> new ParameterizedMessage("Exception during transfer of Metadata Fragment to Remote {}", ex.getMessage()), ex ); - exceptionReference.add(ex); + exceptionList.add(ex); }), latch ); - for (Map.Entry custom : customToUpload.entrySet()) { - writeMetadataComponentAsync(clusterState, custom.getKey(), CUSTOM_METADATA_FORMAT, custom.getValue(), completionListener); + if (uploadSettingsMetadata) { + uploadTasks.add( + getAsyncMetadataWriteAction( + clusterState, + SETTING_METADATA, + SETTINGS_METADATA_FORMAT, + clusterState.metadata().persistentSettings(), + listener + ) + ); } - - try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format( - Locale.ROOT, - "Timed out waiting for transfer of custom metadata to complete - %s", - String.join(", ", customToUpload.keySet()) - ) - ); - exceptionReference.forEach(ex::addSuppressed); - throw ex; - } - } catch (InterruptedException ex) { - exceptionReference.forEach(ex::addSuppressed); - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format( - Locale.ROOT, - "Timed out waiting for transfer of custom metadata to complete - %s", - String.join(", ", customToUpload.keySet()) - ), - ex + if (uploadCoordinationMetadata) { + uploadTasks.add( + getAsyncMetadataWriteAction( + clusterState, + COORDINATION_METADATA, + COORDINATION_METADATA_FORMAT, + clusterState.metadata().coordinationMetadata(), + listener + ) ); - Thread.currentThread().interrupt(); - throw exception; } - if (!exceptionReference.isEmpty()) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format( - Locale.ROOT, - "Exception during transfer of custom metadata to complete - %s", - String.join(", ", customToUpload.keySet()) + if (uploadTemplateMetadata) { + uploadTasks.add( + getAsyncMetadataWriteAction( + clusterState, + TEMPLATES_METADATA, + TEMPLATES_METADATA_FORMAT, + clusterState.metadata().templatesMetadata(), + listener ) ); - exceptionReference.forEach(exception::addSuppressed); - throw exception; } - return result; - } - - /** - * Uploads provided IndexMetadata's to remote store in parallel. The call is blocking so the method waits for upload to finish and then return. - * - * @param clusterState current ClusterState - * @param toUpload list of IndexMetadata to upload - * @return {@code List} list of IndexMetadata uploaded to remote - */ - private List writeIndexMetadataParallel(ClusterState clusterState, List toUpload) - throws IOException { - List exceptionList = Collections.synchronizedList(new ArrayList<>(toUpload.size())); - final CountDownLatch latch = new CountDownLatch(toUpload.size()); - List result = new ArrayList<>(toUpload.size()); - - LatchedActionListener latchedActionListener = new LatchedActionListener<>( - ActionListener.wrap((UploadedIndexMetadata uploadedIndexMetadata) -> { - logger.trace( - String.format(Locale.ROOT, "IndexMetadata uploaded successfully for %s", uploadedIndexMetadata.getIndexName()) - ); - result.add(uploadedIndexMetadata); - }, ex -> { - assert ex instanceof RemoteStateTransferException; - logger.error( - () -> new ParameterizedMessage("Exception during transfer of IndexMetadata to Remote {}", ex.getMessage()), - ex - ); - exceptionList.add(ex); - }), - latch + customToUpload.forEach( + (key, value) -> uploadTasks.add( + getAsyncMetadataWriteAction( + clusterState, + String.join(DELIMITER, CUSTOM_METADATA, key), + CUSTOM_METADATA_FORMAT, + value, + listener + ) + ) ); + indexToUpload.forEach(indexMetadata -> { uploadTasks.add(getIndexMetadataAsyncAction(clusterState, indexMetadata, listener)); }); - for (IndexMetadata indexMetadata : toUpload) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index/ftqsCnn9TgOX/metadata_4_1690947200 - writeIndexMetadataAsync(clusterState, indexMetadata, latchedActionListener); + // start async upload of all required metadata files + for (CheckedRunnable uploadTask : uploadTasks) { + uploadTask.run(); } try { - if (latch.await(getIndexMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { + if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { + // TODO: We should add metrics where transfer is timing out. [Issue: #10687] RemoteStateTransferException ex = new RemoteStateTransferException( - String.format( - Locale.ROOT, - "Timed out waiting for transfer of index metadata to complete - %s", - toUpload.stream().map(IndexMetadata::getIndex).map(Index::toString).collect(Collectors.joining("")) - ) + String.format(Locale.ROOT, "Timed out waiting for transfer of metadata to complete") ); - exceptionList.forEach(ex::addSuppressed); throw ex; } } catch (InterruptedException ex) { - exceptionList.forEach(ex::addSuppressed); RemoteStateTransferException exception = new RemoteStateTransferException( - String.format( - Locale.ROOT, - "Timed out waiting for transfer of index metadata to complete - %s", - toUpload.stream().map(IndexMetadata::getIndex).map(Index::toString).collect(Collectors.joining("")) - ), + String.format(Locale.ROOT, "Timed out waiting for transfer of metadata to complete - %s"), ex ); Thread.currentThread().interrupt(); throw exception; } - if (exceptionList.size() > 0) { + if (!exceptionList.isEmpty()) { RemoteStateTransferException exception = new RemoteStateTransferException( String.format( Locale.ROOT, - "Exception during transfer of IndexMetadata to Remote %s", - toUpload.stream().map(IndexMetadata::getIndex).map(Index::toString).collect(Collectors.joining("")) + "Exception during transfer of following metadata to Remote - %s, %s, %s", + indexToUpload.stream().map(IndexMetadata::getIndexName).collect(Collectors.joining(",")), + customToUpload.keySet().stream().collect(Collectors.joining(", ")), + String.join( + ", ", + (uploadSettingsMetadata ? "settings" : ""), + (uploadCoordinationMetadata ? "coordination" : ""), + (uploadTemplateMetadata ? "templates" : "") + ) ) ); exceptionList.forEach(exception::addSuppressed); throw exception; } - return result; + UploadedMetadataResults response = new UploadedMetadataResults(); + for (Map.Entry entry : results.entrySet()) { + final String name = entry.getKey(); + final ClusterMetadataManifest.UploadedMetadata uploadedMetadata = entry.getValue(); + if (uploadedMetadata.getClass().equals(UploadedIndexMetadata.class)) { + response.uploadedIndexMetadata.add((UploadedIndexMetadata) uploadedMetadata); + } else if (uploadedMetadata.getComponent().contains(CUSTOM_METADATA)) { + // component name for custom metadata will look like custom__ + response.uploadedCustomMetadataMap.put(name.split(DELIMITER)[1], (UploadedMetadataAttribute) uploadedMetadata); + } else if (COORDINATION_METADATA.equals(uploadedMetadata.getComponent())) { + response.uploadedCoordinationMetadata = (UploadedMetadataAttribute) uploadedMetadata; + } else if (SETTING_METADATA.equals(uploadedMetadata.getComponent())) { + response.uploadedSettingsMetadata = (UploadedMetadataAttribute) uploadedMetadata; + } else if (TEMPLATES_METADATA.equals(uploadedMetadata.getComponent())) { + response.uploadedTemplatesMetadata = (UploadedMetadataAttribute) uploadedMetadata; + } else { + throw new IllegalStateException("Unexpected metadata component " + uploadedMetadata.getComponent()); + } + } + return response; } /** @@ -758,11 +603,11 @@ private List writeIndexMetadataParallel(ClusterState clus * @param indexMetadata {@link IndexMetadata} to upload * @param latchedActionListener listener to respond back on after upload finishes */ - private void writeIndexMetadataAsync( + private CheckedRunnable getIndexMetadataAsyncAction( ClusterState clusterState, IndexMetadata indexMetadata, - LatchedActionListener latchedActionListener - ) throws IOException { + LatchedActionListener latchedActionListener + ) { final BlobContainer indexMetadataContainer = indexMetadataContainer( clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), @@ -780,7 +625,7 @@ private void writeIndexMetadataAsync( ex -> latchedActionListener.onFailure(new RemoteStateTransferException(indexMetadata.getIndex().toString(), ex)) ); - INDEX_METADATA_FORMAT.writeAsyncWithUrgentPriority( + return () -> INDEX_METADATA_FORMAT.writeAsyncWithUrgentPriority( indexMetadata, indexMetadataContainer, indexMetadataFilename, @@ -794,29 +639,23 @@ private void writeIndexMetadataAsync( * Allows async upload of Metadata components to remote */ - private void writeMetadataComponentAsync( + private CheckedRunnable getAsyncMetadataWriteAction( ClusterState clusterState, String component, ChecksumBlobStoreFormat componentMetadataBlobStore, ToXContent componentMetadata, - LatchedActionListener latchedActionListener - ) throws IOException { + LatchedActionListener latchedActionListener + ) { final BlobContainer globalMetadataContainer = globalMetadataContainer( clusterState.getClusterName().value(), clusterState.metadata().clusterUUID() ); final String componentMetadataFilename = metadataAttributeFileName(component, clusterState.metadata().version()); ActionListener completionListener = ActionListener.wrap( - resp -> latchedActionListener.onResponse( - new UploadedMetadataAttribute( - component, - componentMetadataFilename - ) - ), + resp -> latchedActionListener.onResponse(new UploadedMetadataAttribute(component, componentMetadataFilename)), ex -> latchedActionListener.onFailure(new RemoteStateTransferException(component, ex)) ); - - componentMetadataBlobStore.writeAsyncWithUrgentPriority( + return () -> componentMetadataBlobStore.writeAsyncWithUrgentPriority( componentMetadata, globalMetadataContainer, componentMetadataFilename, @@ -867,18 +706,56 @@ public void start() { blobStoreRepository = (BlobStoreRepository) repository; } + private ClusterMetadataManifest uploadV1Manifest( + ClusterState clusterState, + List uploadedIndexMetadata, + String previousClusterUUID, + String globalMetadataFileName, + boolean committed + ) throws IOException { + synchronized (this) { + final String manifestFileName = getManifestFileName( + clusterState.term(), + clusterState.version(), + committed, + ClusterMetadataManifest.CODEC_V1 + ); + final ClusterMetadataManifest manifest = new ClusterMetadataManifest( + clusterState.term(), + clusterState.getVersion(), + clusterState.metadata().clusterUUID(), + clusterState.stateUUID(), + Version.CURRENT, + nodeId, + committed, + ClusterMetadataManifest.CODEC_V1, + globalMetadataFileName, + uploadedIndexMetadata, + previousClusterUUID, + clusterState.metadata().clusterUUIDCommitted() + ); + writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); + return manifest; + } + } + private ClusterMetadataManifest uploadManifest( ClusterState clusterState, List uploadedIndexMetadata, String previousClusterUUID, UploadedMetadataAttribute uploadedCoordinationMetadata, UploadedMetadataAttribute uploadedSettingsMetadata, - UploadedMetadataAttribute uploadedTemplateMetadata, + UploadedMetadataAttribute uploadedTemplatesMetadata, Map uploadedCustomMetadataMap, boolean committed ) throws IOException { synchronized (this) { - final String manifestFileName = getManifestFileName(clusterState.term(), clusterState.version(), committed); + final String manifestFileName = getManifestFileName( + clusterState.term(), + clusterState.version(), + committed, + MANIFEST_CURRENT_CODEC_VERSION + ); final ClusterMetadataManifest manifest = new ClusterMetadataManifest( clusterState.term(), clusterState.getVersion(), @@ -894,7 +771,7 @@ private ClusterMetadataManifest uploadManifest( clusterState.metadata().clusterUUIDCommitted(), uploadedCoordinationMetadata, uploadedSettingsMetadata, - uploadedTemplateMetadata, + uploadedTemplatesMetadata, uploadedCustomMetadataMap ); writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); @@ -916,7 +793,7 @@ private void writeMetadataManifest(String clusterName, String clusterUUID, Clust logger.trace(String.format(Locale.ROOT, "Manifest file uploaded successfully.")); }, ex -> { exceptionReference.set(ex); }), latch); - CLUSTER_METADATA_MANIFEST_FORMAT.writeAsyncWithUrgentPriority( + getClusterMetadataManifestBlobStoreFormat(fileName).writeAsyncWithUrgentPriority( uploadManifest, metadataManifestContainer, fileName, @@ -1006,6 +883,29 @@ private void setMetadataManifestUploadTimeout(TimeValue newMetadataManifestUploa this.metadataManifestUploadTimeout = newMetadataManifestUploadTimeout; } + private Map getUpdatedCustoms(ClusterState currentState, ClusterState previousState) { + if (Metadata.isCustomMetadataEqual(previousState.metadata(), currentState.metadata())) { + return new HashMap<>(); + } + Map updatedCustom = new HashMap<>(); + Set currentCustoms = new HashSet<>(currentState.metadata().customs().keySet()); + for (Map.Entry cursor : previousState.metadata().customs().entrySet()) { + if (cursor.getValue().context().contains(Metadata.XContentContext.GATEWAY)) { + if (!cursor.getValue().equals(currentState.metadata().custom(cursor.getKey()))) { + updatedCustom.put(cursor.getKey(), currentState.metadata().custom(cursor.getKey())); + } + currentCustoms.remove(cursor.getKey()); + } + } + for (String custom : currentCustoms) { + Metadata.Custom cursor = currentState.metadata().custom(custom); + if (cursor.context().contains(Metadata.XContentContext.GATEWAY)) { + updatedCustom.put(custom, cursor); + } + } + return updatedCustom; + } + public TimeValue getIndexMetadataUploadTimeout() { return this.indexMetadataUploadTimeout; } @@ -1018,7 +918,7 @@ public TimeValue getMetadataManifestUploadTimeout() { return this.metadataManifestUploadTimeout; } - static String getManifestFileName(long term, long version, boolean committed) { + static String getManifestFileName(long term, long version, boolean committed, int codecVersion) { // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest______C/P____ return String.join( DELIMITER, @@ -1027,8 +927,8 @@ static String getManifestFileName(long term, long version, boolean committed) { RemoteStoreUtils.invertLong(version), (committed ? "C" : "P"), // C for committed and P for published RemoteStoreUtils.invertLong(System.currentTimeMillis()), - String.valueOf(MANIFEST_CURRENT_CODEC_VERSION) // Keep the codec version at last place only, during read we reads last place to - // determine codec version. + String.valueOf(codecVersion) // Keep the codec version at last place only, during read we reads last place to + // determine codec version. ); } @@ -1162,20 +1062,29 @@ private Metadata getGlobalMetadata(String clusterName, String clusterUUID, Clust blobStoreRepository.getNamedXContentRegistry() ); } else if (clusterMetadataManifest.hasMetadataAttributesFiles()) { - CoordinationMetadata coordinationMetadata = getCoordinationMetadata(clusterName, clusterUUID, - clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename()); - Settings settingsMetadata = getSettingsMetadata(clusterName, clusterUUID, - clusterMetadataManifest.getSettingsMetadata().getUploadedFilename()); - TemplatesMetadata templatesMetadata = getTemplatesMetadata(clusterName, clusterUUID, - clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename()); + CoordinationMetadata coordinationMetadata = getCoordinationMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename() + ); + Settings settingsMetadata = getSettingsMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getSettingsMetadata().getUploadedFilename() + ); + TemplatesMetadata templatesMetadata = getTemplatesMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename() + ); Metadata.Builder builder = new Metadata.Builder(); builder.coordinationMetadata(coordinationMetadata); builder.persistentSettings(settingsMetadata); builder.templates(templatesMetadata); - clusterMetadataManifest.getCustomMetadataMap().forEach( - (key, value) -> - builder.putCustom(key, getCustomsMetadata(clusterName, clusterUUID, value.getUploadedFilename(), key)) - ); + clusterMetadataManifest.getCustomMetadataMap().forEach((key, value) -> { + String custom = key.split(DELIMITER)[1]; + builder.putCustom(custom, getCustomsMetadata(clusterName, clusterUUID, value.getUploadedFilename(), custom)); + }); return builder.build(); } else { return Metadata.EMPTY_METADATA; @@ -1235,7 +1144,7 @@ private TemplatesMetadata getTemplatesMetadata(String clusterName, String cluste // Fetch Templates metadata if (templatesMetadataFileName != null) { String[] splitPath = templatesMetadataFileName.split("/"); - return TEMPLATE_METADATA_FORMAT.read( + return TEMPLATES_METADATA_FORMAT.read( globalMetadataContainer(clusterName, clusterUUID), splitPath[splitPath.length - 1], blobStoreRepository.getNamedXContentRegistry() @@ -1251,11 +1160,16 @@ private TemplatesMetadata getTemplatesMetadata(String clusterName, String cluste } } - private Metadata.Custom getCustomsMetadata(String clusterName, String clusterUUID, @NonNull String customMetadataFileName, String custom) { + private Metadata.Custom getCustomsMetadata( + String clusterName, + String clusterUUID, + @NonNull String customMetadataFileName, + String custom + ) { try { // Fetch Custom metadata String[] splitPath = customMetadataFileName.split("/"); - ChecksumBlobStoreFormat customChecksumBlobStoreFormat = new ChecksumBlobStoreFormat<>( + ChecksumBlobStoreFormat customChecksumBlobStoreFormat = new ChecksumBlobStoreFormat<>( "custom", METADATA_NAME_FORMAT, (parser -> Metadata.Custom.fromXContent(parser, custom)) @@ -1515,6 +1429,8 @@ private ChecksumBlobStoreFormat getClusterMetadataManif long codecVersion = getManifestCodecVersion(fileName); if (codecVersion == MANIFEST_CURRENT_CODEC_VERSION) { return CLUSTER_METADATA_MANIFEST_FORMAT; + } else if (codecVersion == ClusterMetadataManifest.CODEC_V1) { + return CLUSTER_METADATA_MANIFEST_FORMAT_V1; } else if (codecVersion == ClusterMetadataManifest.CODEC_V0) { return CLUSTER_METADATA_MANIFEST_FORMAT_V0; } @@ -1736,4 +1652,34 @@ public void deleteStaleClusterUUIDs(ClusterState clusterState, ClusterMetadataMa public RemotePersistenceStats getStats() { return remoteStateStats; } + + private static class UploadedMetadataResults { + List uploadedIndexMetadata; + Map uploadedCustomMetadataMap; + UploadedMetadataAttribute uploadedCoordinationMetadata; + UploadedMetadataAttribute uploadedSettingsMetadata; + UploadedMetadataAttribute uploadedTemplatesMetadata; + + public UploadedMetadataResults( + List uploadedIndexMetadata, + Map uploadedCustomMetadataMap, + UploadedMetadataAttribute uploadedCoordinationMetadata, + UploadedMetadataAttribute uploadedSettingsMetadata, + UploadedMetadataAttribute uploadedTemplatesMetadata + ) { + this.uploadedIndexMetadata = uploadedIndexMetadata; + this.uploadedCustomMetadataMap = uploadedCustomMetadataMap; + this.uploadedCoordinationMetadata = uploadedCoordinationMetadata; + this.uploadedSettingsMetadata = uploadedSettingsMetadata; + this.uploadedTemplatesMetadata = uploadedTemplatesMetadata; + } + + public UploadedMetadataResults() { + this.uploadedIndexMetadata = new ArrayList<>(); + this.uploadedCustomMetadataMap = new HashMap<>(); + this.uploadedCoordinationMetadata = null; + this.uploadedSettingsMetadata = null; + this.uploadedTemplatesMetadata = null; + } + } } diff --git a/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java b/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java index 6c9a3201656d7..5235dd456af7b 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java @@ -16,12 +16,14 @@ import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.core.xcontent.XContentParser; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; +import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; import org.opensearch.test.EqualsHashCodeTestUtils; import org.opensearch.test.OpenSearchTestCase; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; public class ClusterMetadataManifestTests extends OpenSearchTestCase { @@ -53,7 +55,7 @@ public void testClusterMetadataManifestXContentV0() throws IOException { } } - public void testClusterMetadataManifestXContent() throws IOException { + public void testClusterMetadataManifestXContentV1() throws IOException { UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "test-uuid", "/test/upload/path"); ClusterMetadataManifest originalManifest = new ClusterMetadataManifest( 1L, @@ -74,6 +76,37 @@ public void testClusterMetadataManifestXContent() throws IOException { originalManifest.toXContent(builder, ToXContent.EMPTY_PARAMS); builder.endObject(); + try (XContentParser parser = createParser(JsonXContent.jsonXContent, BytesReference.bytes(builder))) { + final ClusterMetadataManifest fromXContentManifest = ClusterMetadataManifest.fromXContentV1(parser); + assertEquals(originalManifest, fromXContentManifest); + } + } + + public void testClusterMetadataManifestXContent() throws IOException { + UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "test-uuid", "/test/upload/path"); + ClusterMetadataManifest originalManifest = new ClusterMetadataManifest( + 1L, + 1L, + "test-cluster-uuid", + "test-state-uuid", + Version.CURRENT, + "test-node-id", + false, + ClusterMetadataManifest.CODEC_V2, + null, + Collections.singletonList(uploadedIndexMetadata), + "prev-cluster-uuid", + true, + new UploadedMetadataAttribute(RemoteClusterStateService.COORDINATION_METADATA, "coordination-file"), + new UploadedMetadataAttribute(RemoteClusterStateService.SETTING_METADATA, "setting-file"), + new UploadedMetadataAttribute(RemoteClusterStateService.TEMPLATES_METADATA, "templates-file"), + new HashMap<>() + ); + final XContentBuilder builder = JsonXContent.contentBuilder(); + builder.startObject(); + originalManifest.toXContent(builder, ToXContent.EMPTY_PARAMS); + builder.endObject(); + try (XContentParser parser = createParser(JsonXContent.jsonXContent, BytesReference.bytes(builder))) { final ClusterMetadataManifest fromXContentManifest = ClusterMetadataManifest.fromXContent(parser); assertEquals(originalManifest, fromXContentManifest); @@ -89,11 +122,15 @@ public void testClusterMetadataManifestSerializationEqualsHashCode() { Version.CURRENT, "B10RX1f5RJenMQvYccCgSQ", true, - 1, - "test-global-metadata-file", + 2, + null, randomUploadedIndexMetadataList(), "yfObdx8KSMKKrXf8UyHhM", - true + true, + new UploadedMetadataAttribute(RemoteClusterStateService.COORDINATION_METADATA, "coordination-file"), + new UploadedMetadataAttribute(RemoteClusterStateService.SETTING_METADATA, "setting-file"), + new UploadedMetadataAttribute(RemoteClusterStateService.TEMPLATES_METADATA, "templates-file"), + new HashMap<>() ); { // Mutate Cluster Term EqualsHashCodeTestUtils.checkEqualsAndHashCode( 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 65477051cdb30..510f33210e5c9 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java @@ -15,7 +15,9 @@ import org.opensearch.cluster.coordination.CoordinationMetadata; import org.opensearch.cluster.metadata.IndexGraveyard; import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.IndexTemplateMetadata; import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.metadata.TemplatesMetadata; import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; import org.opensearch.common.blobstore.BlobContainer; @@ -38,6 +40,7 @@ import org.opensearch.core.index.Index; import org.opensearch.core.xcontent.NamedXContentRegistry; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; +import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.indices.IndicesModule; import org.opensearch.repositories.FilterRepository; @@ -46,6 +49,7 @@ import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.repositories.fs.FsRepository; import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.test.TestCustomMetadata; import org.opensearch.test.VersionUtils; import org.opensearch.threadpool.TestThreadPool; import org.opensearch.threadpool.ThreadPool; @@ -59,6 +63,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.EnumSet; import java.util.HashMap; import java.util.List; import java.util.Locale; @@ -69,14 +74,19 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import java.util.stream.Stream; import org.mockito.ArgumentCaptor; import org.mockito.ArgumentMatchers; import static java.util.stream.Collectors.toList; +import static org.opensearch.gateway.remote.RemoteClusterStateService.COORDINATION_METADATA; +import static org.opensearch.gateway.remote.RemoteClusterStateService.CUSTOM_METADATA; import static org.opensearch.gateway.remote.RemoteClusterStateService.DELIMITER; import static org.opensearch.gateway.remote.RemoteClusterStateService.FORMAT_PARAMS; import static org.opensearch.gateway.remote.RemoteClusterStateService.INDEX_METADATA_CURRENT_CODEC_VERSION; @@ -84,11 +94,14 @@ import static org.opensearch.gateway.remote.RemoteClusterStateService.MANIFEST_FILE_PREFIX; import static org.opensearch.gateway.remote.RemoteClusterStateService.METADATA_FILE_PREFIX; import static org.opensearch.gateway.remote.RemoteClusterStateService.RETAINED_MANIFESTS; +import static org.opensearch.gateway.remote.RemoteClusterStateService.SETTING_METADATA; +import static org.opensearch.gateway.remote.RemoteClusterStateService.TEMPLATES_METADATA; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY; 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.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; import static org.mockito.ArgumentMatchers.any; @@ -223,6 +236,11 @@ public void testWriteFullMetadataSuccess() throws IOException { assertThat(manifest.getClusterUUID(), is(expectedManifest.getClusterUUID())); assertThat(manifest.getStateUUID(), is(expectedManifest.getStateUUID())); assertThat(manifest.getPreviousClusterUUID(), is(expectedManifest.getPreviousClusterUUID())); + assertThat(manifest.getGlobalMetadataFileName(), nullValue()); + assertThat(manifest.getCoordinationMetadata(), notNullValue()); + assertThat(manifest.getSettingsMetadata(), notNullValue()); + assertThat(manifest.getTemplatesMetadata(), notNullValue()); + assertFalse(manifest.getCustomMetadataMap().isEmpty()); } public void testWriteFullMetadataInParallelSuccess() throws IOException { @@ -235,6 +253,18 @@ public void testWriteFullMetadataInParallelSuccess() throws IOException { doAnswer((i) -> { actionListenerArgumentCaptor.getValue().onResponse(null); return null; + }).doAnswer((i) -> { + actionListenerArgumentCaptor.getValue().onResponse(null); + return null; + }).doAnswer((i) -> { + actionListenerArgumentCaptor.getValue().onResponse(null); + return null; + }).doAnswer((i) -> { + actionListenerArgumentCaptor.getValue().onResponse(null); + return null; + }).doAnswer((i) -> { + actionListenerArgumentCaptor.getValue().onResponse(null); + return null; }).doAnswer((i) -> { actionListenerArgumentCaptor.getValue().onResponse(null); capturedWriteContext.set(writeContextArgumentCaptor.getValue()); @@ -263,15 +293,19 @@ public void testWriteFullMetadataInParallelSuccess() throws IOException { 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.getGlobalMetadataFileName(), notNullValue()); + assertThat(manifest.getGlobalMetadataFileName(), nullValue()); + assertThat(manifest.getCoordinationMetadata(), notNullValue()); + assertThat(manifest.getSettingsMetadata(), notNullValue()); + assertThat(manifest.getTemplatesMetadata(), notNullValue()); + assertThat(manifest.getCustomMetadataMap().size(), not(0)); 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.getPreviousClusterUUID(), is(expectedManifest.getPreviousClusterUUID())); - assertEquals(actionListenerArgumentCaptor.getAllValues().size(), 3); - assertEquals(writeContextArgumentCaptor.getAllValues().size(), 3); + assertEquals(7, actionListenerArgumentCaptor.getAllValues().size()); + assertEquals(7, writeContextArgumentCaptor.getAllValues().size()); byte[] writtenBytes = capturedWriteContext.get() .getStreamProvider(Integer.MAX_VALUE) @@ -351,7 +385,7 @@ public void testTimeoutWhileWritingManifestFile() throws IOException { remoteClusterStateService.writeFullMetadata(clusterState, randomAlphaOfLength(10)); } catch (Exception e) { assertTrue(e instanceof RemoteClusterStateService.RemoteStateTransferException); - assertTrue(e.getMessage().contains("Timed out waiting for transfer of manifest file to complete")); + assertTrue(e.getMessage().contains("Timed out waiting for transfer of metadata to complete")); } } @@ -436,13 +470,28 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { } /* - * Here we will verify the migration of manifest file from codec V0 and V1. + * Here we will verify the migration of manifest file from codec V0. * * Initially codec version is 0 and global metadata is also null, we will perform index metadata update. - * In final manifest codec version should be 1 and - * global metadata should be updated, even if it was not changed in this cluster state update + * In final manifest codec version should be 2 and have metadata files updated, + * even if it was not changed in this cluster state update + */ + public void testMigrationFromCodecV0ManifestToCodecV2Manifest() throws IOException { + verifyCodecMigrationManifest(ClusterMetadataManifest.CODEC_V0); + } + + /* + * Here we will verify the migration of manifest file from codec V1. + * + * Initially codec version is 1 and a global metadata file is there, we will perform index metadata update. + * In final manifest codec version should be 2 and have metadata files updated, + * even if it was not changed in this cluster state update */ - public void testMigrationFromCodecV0ManifestToCodecV1Manifest() throws IOException { + public void testMigrationFromCodecV1ManifestToCodecV2Manifest() throws IOException { + verifyCodecMigrationManifest(ClusterMetadataManifest.CODEC_V1); + } + + private void verifyCodecMigrationManifest(int previousCodec) throws IOException { mockBlobStoreObjects(); final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); final ClusterState previousClusterState = ClusterState.builder(ClusterName.DEFAULT) @@ -462,7 +511,7 @@ public void testMigrationFromCodecV0ManifestToCodecV1Manifest() throws IOExcepti // previous manifest with codec 0 and null global metadata final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder() - .codecVersion(ClusterMetadataManifest.CODEC_V0) + .codecVersion(previousCodec) .globalMetadataFileName(null) .indices(Collections.emptyList()) .build(); @@ -475,12 +524,28 @@ public void testMigrationFromCodecV0ManifestToCodecV1Manifest() throws IOExcepti ); // global metadata is updated - assertThat(manifestAfterUpdate.getGlobalMetadataFileName(), notNullValue()); + assertThat(manifestAfterUpdate.hasMetadataAttributesFiles(), is(true)); // Manifest file with codec version with 1 is updated. - assertThat(manifestAfterUpdate.getCodecVersion(), is(ClusterMetadataManifest.CODEC_V1)); + assertThat(manifestAfterUpdate.getCodecVersion(), is(MANIFEST_CURRENT_CODEC_VERSION)); } - public void testWriteIncrementalGlobalMetadataSuccess() throws IOException { + public void testWriteIncrementalGlobalMetadataFromCodecV0Success() throws IOException { + final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder().indices(Collections.emptyList()).build(); + + verifyWriteIncrementalGlobalMetadataFromOlderCodecSuccess(previousManifest); + } + + public void testWriteIncrementalGlobalMetadataFromCodecV1Success() throws IOException { + final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder() + .codecVersion(1) + .globalMetadataFileName("global-metadata-file") + .indices(Collections.emptyList()) + .build(); + + verifyWriteIncrementalGlobalMetadataFromOlderCodecSuccess(previousManifest); + } + + private void verifyWriteIncrementalGlobalMetadataFromOlderCodecSuccess(ClusterMetadataManifest previousManifest) throws IOException { final ClusterState clusterState = generateClusterStateWithGlobalMetadata().nodes(nodesWithLocalNodeClusterManager()).build(); mockBlobStoreObjects(); final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); @@ -488,12 +553,6 @@ public void testWriteIncrementalGlobalMetadataSuccess() throws IOException { .metadata(Metadata.builder().coordinationMetadata(coordinationMetadata)) .build(); - final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder() - .codecVersion(2) - .globalMetadataFileName("global-metadata-file") - .indices(Collections.emptyList()) - .build(); - remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.writeIncrementalMetadata( previousClusterState, @@ -502,8 +561,8 @@ public void testWriteIncrementalGlobalMetadataSuccess() throws IOException { ); final ClusterMetadataManifest expectedManifest = ClusterMetadataManifest.builder() + .codecVersion(2) .indices(Collections.emptyList()) - .globalMetadataFileName("mock-filename") .clusterTerm(1L) .stateVersion(1L) .stateUUID("state-uuid") @@ -511,130 +570,239 @@ public void testWriteIncrementalGlobalMetadataSuccess() throws IOException { .previousClusterUUID("prev-cluster-uuid") .build(); - assertThat(manifest.getGlobalMetadataFileName(), 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())); + assertNull(manifest.getGlobalMetadataFileName()); + assertNotNull(manifest.getCoordinationMetadata()); + assertNotNull(manifest.getSettingsMetadata()); + assertNotNull(manifest.getTemplatesMetadata()); + assertNotEquals(0, manifest.getCustomMetadataMap().size()); + + assertEquals(expectedManifest.getClusterTerm(), manifest.getClusterTerm()); + assertEquals(expectedManifest.getStateVersion(), manifest.getStateVersion()); + assertEquals(expectedManifest.getClusterUUID(), manifest.getClusterUUID()); + assertEquals(expectedManifest.getStateUUID(), manifest.getStateUUID()); + assertEquals(expectedManifest.getCodecVersion(), manifest.getCodecVersion()); } - /* - * Here we will verify index metadata is not uploaded again if change is only in global metadata - */ - public void testGlobalMetadataOnlyUpdated() throws IOException { - // setup - mockBlobStoreObjects(); - final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); - final ClusterState initialClusterState = ClusterState.builder(ClusterName.DEFAULT) - .metadata(Metadata.builder().coordinationMetadata(coordinationMetadata).version(randomNonNegativeLong())) + public void testCoordinationMetadataOnlyUpdated() throws IOException { + // Updating the voting config, as updating the term will upload the full cluster state and other files will also get updated + Function updater = (initialClusterState) -> ClusterState.builder(initialClusterState) + .metadata( + Metadata.builder(initialClusterState.metadata()) + .coordinationMetadata( + CoordinationMetadata.builder(initialClusterState.coordinationMetadata()) + .addVotingConfigExclusion(new CoordinationMetadata.VotingConfigExclusion("excludedNodeId", "excludedNodeName")) + .build() + ) + .build() + ) .build(); - final ClusterMetadataManifest initialManifest = ClusterMetadataManifest.builder() - .codecVersion(2) - .globalMetadataFileName("global-metadata-file") - .indices(Collections.emptyList()) + verifyMetadataAttributeOnlyUpdated(updater, (initialMetadata, metadataAfterUpdate) -> { + // Verify that index metadata information is same in manifest files + assertEquals(metadataAfterUpdate.getIndices().size(), initialMetadata.getIndices().size()); + IntStream.range(0, initialMetadata.getIndices().size()).forEach(i -> { + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexName(), initialMetadata.getIndices().get(i).getIndexName()); + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexUUID(), initialMetadata.getIndices().get(i).getIndexUUID()); + // since timestamp is part of file name, if file name is same we can confirm that file is not update in global metadata + // update + assertEquals( + metadataAfterUpdate.getIndices().get(i).getUploadedFilename(), + initialMetadata.getIndices().get(i).getUploadedFilename() + ); + }); + + // coordination metadata file would have changed + assertFalse( + metadataAfterUpdate.getCoordinationMetadata() + .getUploadedFilename() + .equalsIgnoreCase(initialMetadata.getCoordinationMetadata().getUploadedFilename()) + ); + // Other files will be equal + assertEquals( + metadataAfterUpdate.getSettingsMetadata().getUploadedFilename(), + initialMetadata.getSettingsMetadata().getUploadedFilename() + ); + assertEquals(metadataAfterUpdate.getTemplatesMetadata(), initialMetadata.getTemplatesMetadata()); + assertEquals(metadataAfterUpdate.getCustomMetadataMap(), initialMetadata.getCustomMetadataMap()); + }); + } + + public void testSettingsMetadataOnlyUpdated() throws IOException { + Function updater = (initialClusterState) -> ClusterState.builder(initialClusterState) + .metadata( + Metadata.builder(initialClusterState.metadata()).persistentSettings(Settings.builder().put("foo", "bar").build()).build() + ) .build(); - remoteClusterStateService.start(); - // Initial cluster state with index. - final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); - // Updating remote cluster state with changing index metadata - final ClusterMetadataManifest manifestAfterIndexMetadataUpdate = remoteClusterStateService.writeIncrementalMetadata( - initialClusterState, - clusterState, - initialManifest - ); + verifyMetadataAttributeOnlyUpdated(updater, (initialMetadata, metadataAfterUpdate) -> { + // Verify that index metadata information is same in manifest files + assertEquals(metadataAfterUpdate.getIndices().size(), initialMetadata.getIndices().size()); + IntStream.range(0, initialMetadata.getIndices().size()).forEach(i -> { + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexName(), initialMetadata.getIndices().get(i).getIndexName()); + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexUUID(), initialMetadata.getIndices().get(i).getIndexUUID()); + // since timestamp is part of file name, if file name is same we can confirm that file is not update in global metadata + // update + assertEquals( + metadataAfterUpdate.getIndices().get(i).getUploadedFilename(), + initialMetadata.getIndices().get(i).getUploadedFilename() + ); + }); + + // setting metadata file would have changed + assertFalse( + metadataAfterUpdate.getSettingsMetadata() + .getUploadedFilename() + .equalsIgnoreCase(initialMetadata.getSettingsMetadata().getUploadedFilename()) + ); + assertEquals(metadataAfterUpdate.getCoordinationMetadata(), initialMetadata.getCoordinationMetadata()); + assertEquals(metadataAfterUpdate.getTemplatesMetadata(), initialMetadata.getTemplatesMetadata()); + assertEquals(metadataAfterUpdate.getCustomMetadataMap(), initialMetadata.getCustomMetadataMap()); + }); + } - // new cluster state where only global metadata is different - Metadata newMetadata = Metadata.builder(clusterState.metadata()) - .persistentSettings(Settings.builder().put("cluster.blocks.read_only", true).build()) - .version(randomNonNegativeLong()) + public void testTemplatesMetadataOnlyUpdated() throws IOException { + Function updater = (initialClusterState) -> ClusterState.builder(initialClusterState) + .metadata( + Metadata.builder(initialClusterState.metadata()) + .templates( + TemplatesMetadata.builder() + .put( + IndexTemplateMetadata.builder("template" + randomAlphaOfLength(3)) + .patterns(Arrays.asList("bar-*", "foo-*")) + .settings( + Settings.builder() + .put("random_index_setting_" + randomAlphaOfLength(3), randomAlphaOfLength(5)) + .build() + ) + .build() + ) + .build() + ) + .build() + ) .build(); - ClusterState newClusterState = ClusterState.builder(clusterState).metadata(newMetadata).build(); - // updating remote cluster state with global metadata - final ClusterMetadataManifest manifestAfterGlobalMetadataUpdate = remoteClusterStateService.writeIncrementalMetadata( - clusterState, - newClusterState, - manifestAfterIndexMetadataUpdate - ); + verifyMetadataAttributeOnlyUpdated(updater, (initialMetadata, metadataAfterUpdate) -> { + // Verify that index metadata information is same in manifest files + assertEquals(metadataAfterUpdate.getIndices().size(), initialMetadata.getIndices().size()); + IntStream.range(0, initialMetadata.getIndices().size()).forEach(i -> { + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexName(), initialMetadata.getIndices().get(i).getIndexName()); + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexUUID(), initialMetadata.getIndices().get(i).getIndexUUID()); + // since timestamp is part of file name, if file name is same we can confirm that file is not update in global metadata + // update + assertEquals( + metadataAfterUpdate.getIndices().get(i).getUploadedFilename(), + initialMetadata.getIndices().get(i).getUploadedFilename() + ); + }); - // Verify that index metadata information is same in manifest files - assertThat(manifestAfterIndexMetadataUpdate.getIndices().size(), is(manifestAfterGlobalMetadataUpdate.getIndices().size())); - assertThat( - manifestAfterIndexMetadataUpdate.getIndices().get(0).getIndexName(), - is(manifestAfterGlobalMetadataUpdate.getIndices().get(0).getIndexName()) - ); - assertThat( - manifestAfterIndexMetadataUpdate.getIndices().get(0).getIndexUUID(), - is(manifestAfterGlobalMetadataUpdate.getIndices().get(0).getIndexUUID()) - ); + // template metadata file would have changed + assertFalse( + metadataAfterUpdate.getTemplatesMetadata() + .getUploadedFilename() + .equalsIgnoreCase(initialMetadata.getTemplatesMetadata().getUploadedFilename()) + ); + assertEquals(metadataAfterUpdate.getCoordinationMetadata(), initialMetadata.getCoordinationMetadata()); + assertEquals(metadataAfterUpdate.getSettingsMetadata(), initialMetadata.getSettingsMetadata()); + assertEquals(metadataAfterUpdate.getCustomMetadataMap(), initialMetadata.getCustomMetadataMap()); + }); + } - // since timestamp is part of file name, if file name is same we can confirm that file is not update in global metadata update - assertThat( - manifestAfterIndexMetadataUpdate.getIndices().get(0).getUploadedFilename(), - is(manifestAfterGlobalMetadataUpdate.getIndices().get(0).getUploadedFilename()) - ); + public void testCustomMetadataOnlyUpdated() throws IOException { + Function updater = (initialClusterState) -> ClusterState.builder(initialClusterState) + .metadata( + Metadata.builder(initialClusterState.metadata()) + .putCustom("custom_metadata_type", new CustomMetadata1("mock_custom_metadata")) + .build() + ) + .build(); - // global metadata file would have changed - assertFalse( - manifestAfterIndexMetadataUpdate.getGlobalMetadataFileName() - .equalsIgnoreCase(manifestAfterGlobalMetadataUpdate.getGlobalMetadataFileName()) - ); + verifyMetadataAttributeOnlyUpdated(updater, (initialMetadata, metadataAfterUpdate) -> { + // Verify that index metadata information is same in manifest files + assertEquals(metadataAfterUpdate.getIndices().size(), initialMetadata.getIndices().size()); + IntStream.range(0, initialMetadata.getIndices().size()).forEach(i -> { + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexName(), initialMetadata.getIndices().get(i).getIndexName()); + assertEquals(metadataAfterUpdate.getIndices().get(i).getIndexUUID(), initialMetadata.getIndices().get(i).getIndexUUID()); + // since timestamp is part of file name, if file name is same we can confirm that file is not update in global metadata + // update + assertEquals( + metadataAfterUpdate.getIndices().get(i).getUploadedFilename(), + initialMetadata.getIndices().get(i).getUploadedFilename() + ); + // custom metadata map would have changed + assertNotEquals(metadataAfterUpdate.getCustomMetadataMap(), initialMetadata.getCustomMetadataMap()); + assertEquals(initialMetadata.getCustomMetadataMap().size() + 1, metadataAfterUpdate.getCustomMetadataMap().size()); + initialMetadata.getCustomMetadataMap().forEach((k, v) -> { + assertTrue(metadataAfterUpdate.getCustomMetadataMap().containsKey(k)); + assertEquals(v, metadataAfterUpdate.getCustomMetadataMap().get(k)); + }); + assertEquals(metadataAfterUpdate.getCoordinationMetadata(), initialMetadata.getCoordinationMetadata()); + assertEquals(metadataAfterUpdate.getSettingsMetadata(), initialMetadata.getSettingsMetadata()); + assertEquals(metadataAfterUpdate.getTemplatesMetadata(), initialMetadata.getTemplatesMetadata()); + }); + }); } /* * Here we will verify global metadata is not uploaded again if change is only in index metadata */ public void testIndexMetadataOnlyUpdated() throws IOException { - // setup - mockBlobStoreObjects(); - final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); - final ClusterState initialClusterState = ClusterState.builder(ClusterName.DEFAULT) - .metadata(Metadata.builder().coordinationMetadata(coordinationMetadata)) - .build(); - final ClusterMetadataManifest initialManifest = ClusterMetadataManifest.builder() - .codecVersion(2) - .indices(Collections.emptyList()) + Function updater = (initialState) -> ClusterState.builder(initialState) + .metadata( + Metadata.builder(initialState.metadata()) + .put( + IndexMetadata.builder("test" + randomAlphaOfLength(3)) + .settings( + Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexMetadata.SETTING_INDEX_UUID, "uuid") + .build() + ) + .numberOfShards(1) + .numberOfReplicas(0) + ) + .build() + ) .build(); - remoteClusterStateService.start(); - // Initial cluster state with global metadata. - final ClusterState clusterState = generateClusterStateWithGlobalMetadata().nodes(nodesWithLocalNodeClusterManager()).build(); + verifyMetadataAttributeOnlyUpdated(updater, (initialMetadata, metadataAfterUpdate) -> { + assertEquals(metadataAfterUpdate.getCoordinationMetadata(), initialMetadata.getCoordinationMetadata()); + assertEquals(metadataAfterUpdate.getSettingsMetadata(), initialMetadata.getSettingsMetadata()); + assertEquals(metadataAfterUpdate.getTemplatesMetadata(), initialMetadata.getTemplatesMetadata()); + assertEquals(metadataAfterUpdate.getCustomMetadataMap(), initialMetadata.getCustomMetadataMap()); + assertEquals(initialMetadata.getIndices().size() + 1, metadataAfterUpdate.getIndices().size()); + }); + } - // Updating remote cluster state with changing global metadata - final ClusterMetadataManifest manifestAfterGlobalMetadataUpdate = remoteClusterStateService.writeIncrementalMetadata( - initialClusterState, - clusterState, - initialManifest - ); + /* + * Here we will verify index metadata is not uploaded again if change is only in global metadata + */ + private void verifyMetadataAttributeOnlyUpdated( + Function clusterStateUpdater, + BiConsumer assertions + ) throws IOException { + // setup + mockBlobStoreObjects(); - // new cluster state where only Index metadata is different - final IndexMetadata indexMetadata = new IndexMetadata.Builder("test").settings( - Settings.builder() - .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) - .put(IndexMetadata.SETTING_INDEX_UUID, "uuid") - .build() - ).numberOfShards(1).numberOfReplicas(0).build(); - Metadata newMetadata = Metadata.builder(clusterState.metadata()).put(indexMetadata, true).build(); - ClusterState newClusterState = ClusterState.builder(clusterState).metadata(newMetadata).build(); + // Initial cluster state with index. + final ClusterState initialClusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); + remoteClusterStateService.start(); + final ClusterMetadataManifest initialManifest = remoteClusterStateService.writeFullMetadata(initialClusterState, "_na_"); - // updating remote cluster state with index metadata - final ClusterMetadataManifest manifestAfterIndexMetadataUpdate = remoteClusterStateService.writeIncrementalMetadata( - clusterState, - newClusterState, - manifestAfterGlobalMetadataUpdate - ); + ClusterState newClusterState = clusterStateUpdater.apply(initialClusterState); - // Verify that global metadata information is same in manifest files after updating index Metadata - // since timestamp is part of file name, if file name is same we can confirm that file is not update in index metadata update - assertThat( - manifestAfterIndexMetadataUpdate.getGlobalMetadataFileName(), - is(manifestAfterGlobalMetadataUpdate.getGlobalMetadataFileName()) - ); + // updating remote cluster state with global metadata + final ClusterMetadataManifest manifestAfterMetadataUpdate; + if (initialClusterState.term() == newClusterState.term()) { + manifestAfterMetadataUpdate = remoteClusterStateService.writeIncrementalMetadata( + initialClusterState, + newClusterState, + initialManifest + ); + } else { + manifestAfterMetadataUpdate = remoteClusterStateService.writeFullMetadata(newClusterState, initialClusterState.stateUUID()); + } - // Index metadata would have changed - assertThat(manifestAfterGlobalMetadataUpdate.getIndices().size(), is(0)); - assertThat(manifestAfterIndexMetadataUpdate.getIndices().size(), is(1)); + assertions.accept(initialManifest, manifestAfterMetadataUpdate); } public void testReadLatestMetadataManifestFailedIOException() throws IOException { @@ -795,7 +963,10 @@ public void testReadGlobalMetadata() throws IOException { .stateUUID("state-uuid") .clusterUUID("cluster-uuid") .codecVersion(MANIFEST_CURRENT_CODEC_VERSION) - .globalMetadataFileName("global-metadata-file") + .coordinationMetadata(new UploadedMetadataAttribute(COORDINATION_METADATA, "mock-coordination-file")) + .settingMetadata(new UploadedMetadataAttribute(SETTING_METADATA, "mock-setting-file")) + .templatesMetadata(new UploadedMetadataAttribute(TEMPLATES_METADATA, "mock-templates-file")) + .put(IndexGraveyard.TYPE, new UploadedMetadataAttribute(CUSTOM_METADATA + DELIMITER + IndexGraveyard.TYPE, "mock-custom-" +IndexGraveyard.TYPE+ "-file")) .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) .previousClusterUUID("prev-cluster-uuid") @@ -830,7 +1001,7 @@ public void testReadGlobalMetadataIOException() throws IOException { .stateVersion(1L) .stateUUID("state-uuid") .clusterUUID("cluster-uuid") - .codecVersion(MANIFEST_CURRENT_CODEC_VERSION) + .codecVersion(ClusterMetadataManifest.CODEC_V1) .globalMetadataFileName(globalIndexMetadataName) .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) @@ -1124,18 +1295,24 @@ public void testFileNames() { assertThat(splittedIndexMetadataFileName[1], is(RemoteStoreUtils.invertLong(indexMetadata.getVersion()))); assertThat(splittedIndexMetadataFileName[3], is(String.valueOf(INDEX_METADATA_CURRENT_CODEC_VERSION))); + verifyManifestFileNameWithCodec(MANIFEST_CURRENT_CODEC_VERSION); + verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V1); + verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V0); + } + + private void verifyManifestFileNameWithCodec(int codecVersion) { int term = randomIntBetween(5, 10); int version = randomIntBetween(5, 10); - String manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, true); + String manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, true, codecVersion); assertThat(manifestFileName.split(DELIMITER).length, is(6)); String[] splittedName = manifestFileName.split(DELIMITER); assertThat(splittedName[0], is(MANIFEST_FILE_PREFIX)); assertThat(splittedName[1], is(RemoteStoreUtils.invertLong(term))); assertThat(splittedName[2], is(RemoteStoreUtils.invertLong(version))); assertThat(splittedName[3], is("C")); - assertThat(splittedName[5], is(String.valueOf(MANIFEST_CURRENT_CODEC_VERSION))); + assertThat(splittedName[5], is(String.valueOf(codecVersion))); - manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, false); + manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, false, codecVersion); splittedName = manifestFileName.split(DELIMITER); assertThat(splittedName[3], is("P")); } @@ -1251,12 +1428,16 @@ private void mockObjectsForGettingPreviousClusterUUID( new UploadedIndexMetadata("index1", "index-uuid1", "key1"), new UploadedIndexMetadata("index2", "index-uuid2", "key2") ); + Map customMetadataMap = new HashMap<>(); final ClusterMetadataManifest clusterManifest1 = generateClusterMetadataManifest( "cluster-uuid1", clusterUUIDsPointers.get("cluster-uuid1"), randomAlphaOfLength(10), uploadedIndexMetadataList1, - "test-metadata1", + customMetadataMap, + new UploadedMetadataAttribute(COORDINATION_METADATA, "key3"), + new UploadedMetadataAttribute(SETTING_METADATA, "key4"), + new UploadedMetadataAttribute(TEMPLATES_METADATA, "key5"), clusterUUIDCommitted.getOrDefault("cluster-uuid1", true) ); Settings indexSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT).build(); @@ -1275,7 +1456,7 @@ private void mockObjectsForGettingPreviousClusterUUID( .build(); Map indexMetadataMap1 = Map.of("index-uuid1", indexMetadata1, "index-uuid2", indexMetadata2); mockBlobContainerForGlobalMetadata(blobContainer1, clusterManifest1, metadata1); - mockBlobContainer(blobContainer1, clusterManifest1, indexMetadataMap1, ClusterMetadataManifest.CODEC_V1); + mockBlobContainer(blobContainer1, clusterManifest1, indexMetadataMap1, ClusterMetadataManifest.CODEC_V2); List uploadedIndexMetadataList2 = List.of( new UploadedIndexMetadata("index1", "index-uuid1", "key1"), @@ -1286,7 +1467,10 @@ private void mockObjectsForGettingPreviousClusterUUID( clusterUUIDsPointers.get("cluster-uuid2"), randomAlphaOfLength(10), uploadedIndexMetadataList2, - "test-metadata2", + customMetadataMap, + new UploadedMetadataAttribute(COORDINATION_METADATA, "key3"), + new UploadedMetadataAttribute(SETTING_METADATA, "key4"), + new UploadedMetadataAttribute(TEMPLATES_METADATA, "key5"), clusterUUIDCommitted.getOrDefault("cluster-uuid2", true) ); IndexMetadata indexMetadata3 = IndexMetadata.builder("index1") @@ -1304,7 +1488,7 @@ private void mockObjectsForGettingPreviousClusterUUID( .build(); Map indexMetadataMap2 = Map.of("index-uuid1", indexMetadata3, "index-uuid2", indexMetadata4); mockBlobContainerForGlobalMetadata(blobContainer2, clusterManifest2, metadata2); - mockBlobContainer(blobContainer2, clusterManifest2, indexMetadataMap2, ClusterMetadataManifest.CODEC_V1); + mockBlobContainer(blobContainer2, clusterManifest2, indexMetadataMap2, ClusterMetadataManifest.CODEC_V2); // differGlobalMetadata controls which one of IndexMetadata or Metadata object would be different // when comparing cluster-uuid3 and cluster-uuid1 state. @@ -1326,17 +1510,19 @@ private void mockObjectsForGettingPreviousClusterUUID( Metadata metadata3 = Metadata.builder() .persistentSettings(Settings.builder().put(Metadata.SETTING_READ_ONLY_SETTING.getKey(), !differGlobalMetadata).build()) .build(); - final ClusterMetadataManifest clusterManifest3 = generateClusterMetadataManifest( "cluster-uuid3", clusterUUIDsPointers.get("cluster-uuid3"), randomAlphaOfLength(10), uploadedIndexMetadataList3, - "test-metadata3", + customMetadataMap, + new UploadedMetadataAttribute(COORDINATION_METADATA, "key3"), + new UploadedMetadataAttribute(SETTING_METADATA, "key4"), + new UploadedMetadataAttribute(TEMPLATES_METADATA, "key5"), clusterUUIDCommitted.getOrDefault("cluster-uuid3", true) ); mockBlobContainerForGlobalMetadata(blobContainer3, clusterManifest3, metadata3); - mockBlobContainer(blobContainer3, clusterManifest3, indexMetadataMap3, ClusterMetadataManifest.CODEC_V1); + mockBlobContainer(blobContainer3, clusterManifest3, indexMetadataMap3, ClusterMetadataManifest.CODEC_V2); ArrayList mockBlobContainerOrderedList = new ArrayList<>( List.of(blobContainer1, blobContainer1, blobContainer3, blobContainer3, blobContainer2, blobContainer2) @@ -1356,7 +1542,7 @@ private void mockObjectsForGettingPreviousClusterUUID( when(blobStoreRepository.getCompressor()).thenReturn(new DeflateCompressor()); } - private ClusterMetadataManifest generateClusterMetadataManifest( + private ClusterMetadataManifest generateV1ClusterMetadataManifest( String clusterUUID, String previousClusterUUID, String stateUUID, @@ -1380,6 +1566,36 @@ private ClusterMetadataManifest generateClusterMetadataManifest( .build(); } + private ClusterMetadataManifest generateClusterMetadataManifest( + String clusterUUID, + String previousClusterUUID, + String stateUUID, + List uploadedIndexMetadata, + Map customMetadataMap, + UploadedMetadataAttribute coordinationMetadata, + UploadedMetadataAttribute settingsMetadata, + UploadedMetadataAttribute templatesMetadata, + Boolean isUUIDCommitted + ) { + return ClusterMetadataManifest.builder() + .indices(uploadedIndexMetadata) + .clusterTerm(1L) + .stateVersion(1L) + .stateUUID(stateUUID) + .clusterUUID(clusterUUID) + .nodeId("nodeA") + .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID(previousClusterUUID) + .committed(true) + .clusterUUIDCommitted(isUUIDCommitted) + .coordinationMetadata(coordinationMetadata) + .settingMetadata(settingsMetadata) + .templatesMetadata(templatesMetadata) + .customMetadataMap(customMetadataMap) + .codecVersion(MANIFEST_CURRENT_CODEC_VERSION) + .build(); + } + private BlobContainer mockBlobStoreObjects() { return mockBlobStoreObjects(BlobContainer.class); } @@ -1419,7 +1635,7 @@ private void mockBlobContainer( int codecVersion ) throws IOException { String manifestFileName = codecVersion >= ClusterMetadataManifest.CODEC_V1 - ? "manifest__manifestFileName__abcd__abcd__abcd__1" + ? "manifest__manifestFileName__abcd__abcd__abcd__" + codecVersion : "manifestFileName"; BlobMetadata blobMetadata = new PlainBlobMetadata(manifestFileName, 1); when(blobContainer.listBlobsByPrefixInSortedOrder("manifest" + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC)) @@ -1460,7 +1676,8 @@ private void mockBlobContainerForGlobalMetadata( ClusterMetadataManifest clusterMetadataManifest, Metadata metadata ) throws IOException { - String mockManifestFileName = "manifest__1__2__C__456__1"; + int codecVersion = clusterMetadataManifest.getCodecVersion(); + String mockManifestFileName = "manifest__1__2__C__456__" + codecVersion; BlobMetadata blobMetadata = new PlainBlobMetadata(mockManifestFileName, 1); when( blobContainer.listBlobsByPrefixInSortedOrder( @@ -1477,19 +1694,84 @@ private void mockBlobContainerForGlobalMetadata( FORMAT_PARAMS ); when(blobContainer.readBlob(mockManifestFileName)).thenReturn(new ByteArrayInputStream(bytes.streamInput().readAllBytes())); + if (codecVersion >= ClusterMetadataManifest.CODEC_V2) { + String coordinationFileName = getFileNameFromPath(clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename()); + when(blobContainer.readBlob(RemoteClusterStateService.COORDINATION_METADATA_FORMAT.blobName(coordinationFileName))).thenAnswer( + (invocationOnMock) -> { + BytesReference bytesReference = RemoteClusterStateService.COORDINATION_METADATA_FORMAT.serialize( + metadata.coordinationMetadata(), + coordinationFileName, + blobStoreRepository.getCompressor(), + FORMAT_PARAMS + ); + return new ByteArrayInputStream(bytesReference.streamInput().readAllBytes()); + } + ); + + String settingsFileName = getFileNameFromPath(clusterMetadataManifest.getSettingsMetadata().getUploadedFilename()); + when(blobContainer.readBlob(RemoteClusterStateService.SETTINGS_METADATA_FORMAT.blobName(settingsFileName))).thenAnswer( + (invocationOnMock) -> { + BytesReference bytesReference = RemoteClusterStateService.SETTINGS_METADATA_FORMAT.serialize( + metadata.persistentSettings(), + settingsFileName, + blobStoreRepository.getCompressor(), + FORMAT_PARAMS + ); + return new ByteArrayInputStream(bytesReference.streamInput().readAllBytes()); + } + ); - String[] splitPath = clusterMetadataManifest.getGlobalMetadataFileName().split("/"); - when(blobContainer.readBlob(RemoteClusterStateService.GLOBAL_METADATA_FORMAT.blobName(splitPath[splitPath.length - 1]))).thenAnswer( - (invocationOnMock) -> { - BytesReference bytesGlobalMetadata = RemoteClusterStateService.GLOBAL_METADATA_FORMAT.serialize( - metadata, - "global-metadata-file", - blobStoreRepository.getCompressor(), - FORMAT_PARAMS + String templatesFileName = getFileNameFromPath(clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename()); + when(blobContainer.readBlob(RemoteClusterStateService.TEMPLATES_METADATA_FORMAT.blobName(templatesFileName))).thenAnswer( + (invocationOnMock) -> { + BytesReference bytesReference = RemoteClusterStateService.TEMPLATES_METADATA_FORMAT.serialize( + metadata.templatesMetadata(), + templatesFileName, + blobStoreRepository.getCompressor(), + FORMAT_PARAMS + ); + return new ByteArrayInputStream(bytesReference.streamInput().readAllBytes()); + } + ); + + Map customFileMap = clusterMetadataManifest.getCustomMetadataMap() + .entrySet() + .stream() + .collect(Collectors.toMap(Map.Entry::getKey, entry -> getFileNameFromPath(entry.getValue().getUploadedFilename()))); + + for (Map.Entry entry : customFileMap.entrySet()) { + String custom = entry.getKey(); + String fileName = entry.getValue(); + when(blobContainer.readBlob(RemoteClusterStateService.CUSTOM_METADATA_FORMAT.blobName(fileName))).thenAnswer( + (invocation) -> { + BytesReference bytesReference = RemoteClusterStateService.CUSTOM_METADATA_FORMAT.serialize( + metadata.custom(custom), + fileName, + blobStoreRepository.getCompressor(), + FORMAT_PARAMS + ); + return new ByteArrayInputStream(bytesReference.streamInput().readAllBytes()); + } ); - return new ByteArrayInputStream(bytesGlobalMetadata.streamInput().readAllBytes()); } - ); + } else if (codecVersion == ClusterMetadataManifest.CODEC_V1) { + String[] splitPath = clusterMetadataManifest.getGlobalMetadataFileName().split("/"); + when(blobContainer.readBlob(RemoteClusterStateService.GLOBAL_METADATA_FORMAT.blobName(splitPath[splitPath.length - 1]))) + .thenAnswer((invocationOnMock) -> { + BytesReference bytesGlobalMetadata = RemoteClusterStateService.GLOBAL_METADATA_FORMAT.serialize( + metadata, + "global-metadata-file", + blobStoreRepository.getCompressor(), + FORMAT_PARAMS + ); + return new ByteArrayInputStream(bytesGlobalMetadata.streamInput().readAllBytes()); + }); + } + } + + private String getFileNameFromPath(String filePath) { + String[] splitPath = filePath.split("/"); + return splitPath[splitPath.length - 1]; } private static ClusterState.Builder generateClusterStateWithGlobalMetadata() { @@ -1519,7 +1801,9 @@ private static ClusterState.Builder generateClusterStateWithOneIndex() { .numberOfReplicas(0) .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 CustomMetadata1 customMetadata1 = new CustomMetadata1("custom-metadata-1"); return ClusterState.builder(ClusterName.DEFAULT) .version(1L) .stateUUID("state-uuid") @@ -1529,6 +1813,9 @@ private static ClusterState.Builder generateClusterStateWithOneIndex() { .put(indexMetadata, true) .clusterUUID("cluster-uuid") .coordinationMetadata(coordinationMetadata) + .persistentSettings(settings) + .templates(templatesMetadata) + .putCustom(customMetadata1.getWriteableName(), customMetadata1) .build() ); } @@ -1537,4 +1824,27 @@ private static DiscoveryNodes nodesWithLocalNodeClusterManager() { return DiscoveryNodes.builder().clusterManagerNodeId("cluster-manager-id").localNodeId("cluster-manager-id").build(); } + private static class CustomMetadata1 extends TestCustomMetadata { + public static final String TYPE = "custom_md_1"; + + CustomMetadata1(String data) { + super(data); + } + + @Override + public String getWriteableName() { + return TYPE; + } + + @Override + public Version getMinimalSupportedVersion() { + return Version.CURRENT; + } + + @Override + public EnumSet context() { + return EnumSet.of(Metadata.XContentContext.GATEWAY); + } + } + }