[Backport] Add cluster state checksum in manifest #15218

CHANGELOG.md

@@ -39,6 +39,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Reader Writer Separation] Add searchOnly replica routing configuration ([#15410](https://github.com/opensearch-project/OpenSearch/pull/15410))
 - Add index creation using the context field ([#15290](https://github.com/opensearch-project/OpenSearch/pull/15290))
 - [Remote Publication] Add remote download stats ([#15291](https://github.com/opensearch-project/OpenSearch/pull/15291)))
+- [Remote Publication] Added checksum validation for cluster state behind a cluster setting ([#15218](https://github.com/opensearch-project/OpenSearch/pull/15218))
 
 ### Dependencies
 - Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))

libs/core/src/main/java/org/opensearch/core/common/io/stream/StreamOutput.java

@@ -633,7 +633,7 @@ public final <K, V> void writeMapOfLists(final Map<K, List<V>> map, final Writer
      * @param keyWriter The key writer
      * @param valueWriter The value writer
      */
-    public final <K, V> void writeMap(final Map<K, V> map, final Writer<K> keyWriter, final Writer<V> valueWriter) throws IOException {
+    public <K, V> void writeMap(final Map<K, V> map, final Writer<K> keyWriter, final Writer<V> valueWriter) throws IOException {
         writeVInt(map.size());
         for (final Map.Entry<K, V> entry : map.entrySet()) {
             keyWriter.write(this, entry.getKey());

libs/core/src/main/java/org/opensearch/core/common/io/stream/VerifiableWriteable.java

@@ -0,0 +1,20 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.core.common.io.stream;
+
+import java.io.IOException;
+
+/**
+ * Provides a method for serialization which will give ordered stream, creating same byte array on every invocation.
+ * This should be invoked with a stream that provides ordered serialization.
+ */
+public interface VerifiableWriteable extends Writeable {
+
+    void writeVerifiableTo(StreamOutput out) throws IOException;
+}

server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteRoutingTableServiceIT.java

@@ -67,6 +67,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
             )
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, REMOTE_ROUTING_TABLE_REPO)
             .put(REMOTE_PUBLICATION_EXPERIMENTAL, true)
+            .put(RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_ENABLED_SETTING.getKey(), true)
             .build();
     }
 

server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteStatePublicationIT.java

@@ -90,6 +90,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, routingTableRepoName)
             .put(routingTableRepoTypeAttributeKey, ReloadableFsRepository.TYPE)
             .put(routingTableRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath)
+            .put(RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_ENABLED_SETTING.getKey(), true)
             .build();
     }
 

server/src/main/java/org/opensearch/cluster/AbstractDiffable.java

@@ -83,6 +83,11 @@
             this.part = part;
         }
 
+        @Override
+        public String toString() {
+            return "CompleteDiff{" + "part=" + part + '}';
+        }
+
         /**
          * Creates simple diff without changes
          */

server/src/main/java/org/opensearch/cluster/ClusterState.java

@@ -156,7 +156,7 @@
 
     }
 
-    private static final NamedDiffableValueSerializer<Custom> CUSTOM_VALUE_SERIALIZER = new NamedDiffableValueSerializer<>(Custom.class);
+    public static final NamedDiffableValueSerializer<Custom> CUSTOM_VALUE_SERIALIZER = new NamedDiffableValueSerializer<>(Custom.class);
 
     public static final String UNKNOWN_UUID = "_na_";
 
@@ -839,6 +839,34 @@
             minimumClusterManagerNodesOnPublishingClusterManager = after.minimumClusterManagerNodesOnPublishingClusterManager;
         }
 
+        @Override
+        public String toString() {
+            return new StringBuilder().append("ClusterStateDiff{toVersion=")
+                .append(toVersion)
+                .append(", fromUuid='")
+                .append(fromUuid)
+                .append('\'')
+                .append(", toUuid='")
+                .append(toUuid)
+                .append('\'')
+                .append(", clusterName=")
+                .append(clusterName)
+                .append(", routingTable=")
+                .append(routingTable)
+                .append(", nodes=")
+                .append(nodes)
+                .append(", metadata=")
+                .append(metadata)
+                .append(", blocks=")
+                .append(blocks)
+                .append(", customs=")
+                .append(customs)
+                .append(", minimumClusterManagerNodesOnPublishingClusterManager=")
+                .append(minimumClusterManagerNodesOnPublishingClusterManager)
+                .append("}")
+                .toString();
+        }
+
         ClusterStateDiff(StreamInput in, DiscoveryNode localNode) throws IOException {
             clusterName = new ClusterName(in);
             fromUuid = in.readString();

server/src/main/java/org/opensearch/cluster/DiffableUtils.java

@@ -271,6 +271,18 @@ public Map<K, T> getUpserts() {
             return upserts;
         }
 
+        @Override
+        public String toString() {
+            return new StringBuilder().append("MapDiff{deletes=")
+                .append(deletes)
+                .append(", diffs=")
+                .append(diffs)
+                .append(", upserts=")
+                .append(upserts)
+                .append("}")
+                .toString();
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeCollection(deletes, (o, v) -> keySerializer.writeKey(v, o));

server/src/main/java/org/opensearch/cluster/block/ClusterBlock.java

@@ -52,7 +52,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public class ClusterBlock implements Writeable, ToXContentFragment {
+public class ClusterBlock implements Writeable, ToXContentFragment, Comparable<ClusterBlock> {
 
     private final int id;
     @Nullable
@@ -217,7 +217,13 @@ public int hashCode() {
         return Objects.hash(id, uuid);
     }
 
+    @Override
+    public int compareTo(ClusterBlock block) {
+        return Integer.compare(block.id(), this.id());
+    }
+
     public boolean isAllowReleaseResources() {
         return allowReleaseResources;
     }
+
 }

server/src/main/java/org/opensearch/cluster/block/ClusterBlocks.java

@@ -303,6 +303,10 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeMap(indicesBlocks, StreamOutput::writeString, (o, s) -> writeBlockSet(s, o));
     }
 
+    public void writeVerifiableTo(StreamOutput out) throws IOException {
+        writeTo(out);
+    }
+
     private static void writeBlockSet(Set<ClusterBlock> blocks, StreamOutput out) throws IOException {
         out.writeCollection(blocks);
     }

server/src/main/java/org/opensearch/cluster/coordination/CoordinationMetadata.java

@@ -37,6 +37,7 @@
 import org.opensearch.core.ParseField;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.VerifiableWriteable;
 import org.opensearch.core.common.io.stream.Writeable;
 import org.opensearch.core.xcontent.ConstructingObjectParser;
 import org.opensearch.core.xcontent.ToXContentFragment;
@@ -59,7 +60,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public class CoordinationMetadata implements Writeable, ToXContentFragment {
+public class CoordinationMetadata implements VerifiableWriteable, ToXContentFragment {
 
     public static final CoordinationMetadata EMPTY_METADATA = builder().build();
 
@@ -149,6 +150,11 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeCollection(votingConfigExclusions);
     }
 
+    @Override
+    public void writeVerifiableTo(StreamOutput out) throws IOException {
+        writeTo(out);
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         return builder.field(TERM_PARSE_FIELD.getPreferredName(), term)
@@ -272,7 +278,7 @@ public CoordinationMetadata build() {
      * @opensearch.api
      */
     @PublicApi(since = "1.0.0")
-    public static class VotingConfigExclusion implements Writeable, ToXContentFragment {
+    public static class VotingConfigExclusion implements Writeable, ToXContentFragment, Comparable<VotingConfigExclusion> {
         public static final String MISSING_VALUE_MARKER = "_absent_";
         private final String nodeId;
         private final String nodeName;
@@ -361,6 +367,10 @@ public String toString() {
             return sb.toString();
         }
 
+        @Override
+        public int compareTo(VotingConfigExclusion votingConfigExclusion) {
+            return votingConfigExclusion.getNodeId().compareTo(this.getNodeId());
+        }
     }
 
     /**

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -56,6 +56,7 @@
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.VerifiableWriteable;
 import org.opensearch.core.common.io.stream.Writeable;
 import org.opensearch.core.index.Index;
 import org.opensearch.core.index.shard.ShardId;
@@ -69,6 +70,7 @@
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.seqno.SequenceNumbers;
+import org.opensearch.index.translog.BufferedChecksumStreamOutput;
 import org.opensearch.indices.replication.SegmentReplicationSource;
 import org.opensearch.indices.replication.common.ReplicationType;
 
@@ -88,6 +90,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.function.Function;
 
 import static org.opensearch.cluster.metadata.Metadata.CONTEXT_MODE_PARAM;
@@ -103,7 +106,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public class IndexMetadata implements Diffable<IndexMetadata>, ToXContentFragment {
+public class IndexMetadata implements Diffable<IndexMetadata>, ToXContentFragment, VerifiableWriteable {
 
     public static final ClusterBlock INDEX_READ_ONLY_BLOCK = new ClusterBlock(
         5,
@@ -1287,6 +1290,32 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    @Override
+    public void writeVerifiableTo(StreamOutput out) throws IOException {
+        out.writeString(index.getName()); // uuid will come as part of settings
+        out.writeLong(version);
+        out.writeVLong(mappingVersion);
+        out.writeVLong(settingsVersion);
+        out.writeVLong(aliasesVersion);
+        out.writeInt(routingNumShards);
+        out.writeByte(state.id());
+        writeSettingsToStream(settings, out);
+        out.writeVLongArray(primaryTerms);
+        ((BufferedChecksumStreamOutput) out).writeMapValues(mappings, (stream, val) -> val.writeTo(stream));
+        ((BufferedChecksumStreamOutput) out).writeMapValues(aliases, (stream, val) -> val.writeTo(stream));
+        out.writeMap(customData, StreamOutput::writeString, (stream, val) -> val.writeTo(stream));
+        out.writeMap(
+            inSyncAllocationIds,
+            StreamOutput::writeVInt,
+            (stream, val) -> DiffableUtils.StringSetValueSerializer.getInstance().write(new TreeSet<>(val), stream)
+        );
+        ((BufferedChecksumStreamOutput) out).writeMapValues(rolloverInfos, (stream, val) -> val.writeTo(stream));
+        out.writeBoolean(isSystem);
+        if (out.getVersion().onOrAfter(Version.V_2_17_0)) {
+            out.writeOptionalWriteable(context);
+        }
+    }
+
     public boolean isSystem() {
         return isSystem;
     }

server/src/main/java/org/opensearch/cluster/metadata/IndexTemplateMetadata.java

@@ -50,6 +50,7 @@
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.index.mapper.MapperService;
+import org.opensearch.index.translog.BufferedChecksumStreamOutput;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -257,6 +258,16 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalVInt(version);
     }
 
+    public void writeVerifiableTo(StreamOutput out) throws IOException {
+        out.writeString(name);
+        out.writeInt(order);
+        out.writeStringCollection(patterns);
+        Settings.writeSettingsToStream(settings, out);
+        out.writeMap(mappings, StreamOutput::writeString, (stream, val) -> val.writeTo(stream));
+        ((BufferedChecksumStreamOutput) out).writeMapValues(aliases, (stream, val) -> val.writeTo(stream));
+        out.writeOptionalVInt(version);
+    }
+
     @Override
     public String toString() {
         try {

server/src/main/java/org/opensearch/cluster/metadata/Metadata.java

@@ -254,7 +254,7 @@ static Custom fromXContent(XContentParser parser, String name) throws IOExceptio
 
     public static final String GLOBAL_STATE_FILE_PREFIX = "global-";
 
-    private static final NamedDiffableValueSerializer<Custom> CUSTOM_VALUE_SERIALIZER = new NamedDiffableValueSerializer<>(Custom.class);
+    public static final NamedDiffableValueSerializer<Custom> CUSTOM_VALUE_SERIALIZER = new NamedDiffableValueSerializer<>(Custom.class);
 
     private final String clusterUUID;
     private final boolean clusterUUIDCommitted;

server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java

@@ -11,9 +11,11 @@
 import org.opensearch.cluster.AbstractDiffable;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.VerifiableWriteable;
 import org.opensearch.core.xcontent.ToXContentFragment;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.index.translog.BufferedChecksumStreamOutput;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -27,7 +29,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "2.15.0")
-public class TemplatesMetadata extends AbstractDiffable<TemplatesMetadata> implements ToXContentFragment {
+public class TemplatesMetadata extends AbstractDiffable<TemplatesMetadata> implements ToXContentFragment, VerifiableWriteable {
     public static TemplatesMetadata EMPTY_METADATA = builder().build();
     private final Map<String, IndexTemplateMetadata> templates;
 
@@ -65,6 +67,11 @@
         }
     }
 
+    @Override
+    public void writeVerifiableTo(StreamOutput out) throws IOException {
+        ((BufferedChecksumStreamOutput) out).writeMapValues(templates, (stream, value) -> value.writeVerifiableTo(stream));
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -80,6 +87,11 @@
         return templates != null ? templates.hashCode() : 0;
     }
 
+    @Override
+    public String toString() {
+        return "TemplatesMetadata{" + "templates=" + templates + '}';
+    }
+
     /**
      * Builder for the templates metadata
      *