Add cluster state checksum in manifest (opensearch-project#15218)

* Add cluster state checksum in manifest for remote state and routing table publication

Signed-off-by: Himshikha Gupta <[email protected]>
Co-authored-by: Bukhtawar Khan <[email protected]>

Author: 2 people

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 reset() throws IOException {
     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 @@ private static class CompleteDiff<T extends Diffable<T>> implements Diff<T> {
             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 @@ default boolean isPrivate() {
 
     }
 
-    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 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
             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;
     }
+
 }