Skip to content

Commit

Permalink
Add cluster state checksum in manifest (opensearch-project#15218)
Browse files Browse the repository at this point in the history
* 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]>
  • Loading branch information
himshikha and Bukhtawar authored Sep 3, 2024
1 parent 0f53bf9 commit 82762d4
Show file tree
Hide file tree
Showing 38 changed files with 2,085 additions and 61 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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);
}
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
*/
Expand Down
30 changes: 29 additions & 1 deletion server/src/main/java/org/opensearch/cluster/ClusterState.java
Original file line number Diff line number Diff line change
Expand Up @@ -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_";

Expand Down Expand Up @@ -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();
Expand Down
12 changes: 12 additions & 0 deletions server/src/main/java/org/opensearch/cluster/DiffableUtils.java
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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();

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -361,6 +368,10 @@ public String toString() {
return sb.toString();
}

@Override
public int compareTo(VotingConfigExclusion votingConfigExclusion) {
return votingConfigExclusion.getNodeId().compareTo(this.getNodeId());
}
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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,
Expand Down Expand Up @@ -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;
}
Expand Down
Loading

0 comments on commit 82762d4

Please sign in to comment.