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 <himshikh@amazon.com>
Co-authored-by: Bukhtawar Khan <bukhtawa@amazon.com>

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;
  }
+
 }

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;
  }