Add cluster state checksum in manifest

CHANGELOG.md

@@ -46,6 +46,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Remote Publication] Add remote download stats ([#15291](https://github.com/opensearch-project/OpenSearch/pull/15291)))
 - Add support for comma-separated list of index names to be used with Snapshot Status API ([#15409](https://github.com/opensearch-project/OpenSearch/pull/15409))
 - Add prefix support to hashed prefix & infix path types on remote store ([#15557](https://github.com/opensearch-project/OpenSearch/pull/15557))
+- [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/BufferedChecksumStreamOutput.java

@@ -33,9 +33,18 @@
 package org.opensearch.core.common.io.stream;
 
 import org.apache.lucene.store.BufferedChecksum;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
 import java.util.zip.CRC32;
 import java.util.zip.Checksum;
 
@@ -90,4 +99,75 @@
     public void resetDigest() {
         digest.reset();
     }
+
+    @Override
+    public void writeMap(@Nullable Map<String, Object> map) throws IOException {
+        Map<String, Object> newMap = new TreeMap<>(map);
+        writeGenericValue(newMap);
+    }
+
+    @Override
+    public <K, V> void writeMap(Map<K, V> map, final Writeable.Writer<K> keyWriter, final Writeable.Writer<V> valueWriter)
+        throws IOException {
+        writeVInt(map.size());
+        map.keySet().stream().sorted().forEachOrdered(key -> {
+            try {
+                keyWriter.write(this, key);
+                valueWriter.write(this, map.get(key));
+            } catch (IOException e) {
+                throw new RuntimeException("Failed to write map values.", e);
+            }
+        });
+    }
+
+    public <K, V> void writeMapValues(Map<K, V> map, final Writeable.Writer<V> valueWriter) throws IOException {
+        writeVInt(map.size());
+        map.keySet().stream().sorted().forEachOrdered(key -> {
+            try {
+                valueWriter.write(this, map.get(key));
+            } catch (IOException e) {
+                throw new RuntimeException("Failed to write map values.", e);
+            }
+        });
+    }
+
+    @Override
+    public void writeStringArray(String[] array) throws IOException {
+        String[] copyArray = Arrays.copyOf(array, array.length);
+        Arrays.sort(copyArray);
+        super.writeStringArray(copyArray);
+    }
+
+    @Override
+    public void writeVLongArray(long[] values) throws IOException {
+        long[] copyValues = Arrays.copyOf(values, values.length);
+        Arrays.sort(copyValues);
+        super.writeVLongArray(copyValues);
+    }
+
+    @Override
+    public void writeCollection(final Collection<? extends Writeable> collection) throws IOException {
+        List<? extends Writeable> sortedList = collection.stream().sorted().collect(Collectors.toList());
+        super.writeCollection(sortedList, (o, v) -> v.writeTo(o));
+    }
+
+    @Override
+    public void writeStringCollection(final Collection<String> collection) throws IOException {
+        List<String> listCollection = new ArrayList<>(collection);
+        Collections.sort(listCollection);
+        writeCollection(listCollection, StreamOutput::writeString);
+    }
+
+    @Override
+    public void writeOptionalStringCollection(final Collection<String> collection) throws IOException {
+        if (collection != null) {
+            List<String> listCollection = new ArrayList<>(collection);
+            Collections.sort(listCollection);
+            writeBoolean(true);
+            writeCollection(listCollection, StreamOutput::writeString);
+        } else {
+            writeBoolean(false);
+        }
+    }
+
 }

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(BufferedChecksumStreamOutput 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

@@ -39,8 +39,10 @@
 import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.util.set.Sets;
+import org.opensearch.core.common.io.stream.BufferedChecksumStreamOutput;
 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.rest.RestStatus;
 
 import java.io.IOException;
@@ -62,7 +64,7 @@
  * @opensearch.api
  */
 @PublicApi(since = "1.0.0")
-public class ClusterBlocks extends AbstractDiffable<ClusterBlocks> {
+public class ClusterBlocks extends AbstractDiffable<ClusterBlocks> implements VerifiableWriteable {
     public static final ClusterBlocks EMPTY_CLUSTER_BLOCK = new ClusterBlocks(emptySet(), Map.of());
 
     private final Set<ClusterBlock> global;
@@ -303,6 +305,11 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeMap(indicesBlocks, StreamOutput::writeString, (o, s) -> writeBlockSet(s, o));
     }
 
+    @Override
+    public void writeVerifiableTo(BufferedChecksumStreamOutput 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

@@ -35,8 +35,10 @@
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.util.set.Sets;
 import org.opensearch.core.ParseField;
+import org.opensearch.core.common.io.stream.BufferedChecksumStreamOutput;
 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 +61,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 +151,11 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeCollection(votingConfigExclusions);
     }
 
+    @Override
+    public void writeVerifiableTo(BufferedChecksumStreamOutput 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 +279,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 +368,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

@@ -54,8 +54,10 @@
 import org.opensearch.common.xcontent.XContentHelper;
 import org.opensearch.core.Assertions;
 import org.opensearch.core.common.Strings;
+import org.opensearch.core.common.io.stream.BufferedChecksumStreamOutput;
 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;
@@ -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,
@@ -1264,6 +1267,32 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    @Override
+    public void writeVerifiableTo(BufferedChecksumStreamOutput 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);
+        out.writeMapValues(mappings, (stream, val) -> val.writeTo(stream));
+        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)
+        );
+        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;
     }