From 376fe2a9fe3f13c7555c40cda6d3912d55ef63cc Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Tue, 12 Nov 2024 19:10:15 -0600 Subject: [PATCH] Streamline the serialized format for index status gossip messages patch by Caleb Rackliffe; reviewed by David Capwell for CASSANDRA-20058 --- CHANGES.txt | 1 + .../org/apache/cassandra/index/Index.java | 36 ++- .../cassandra/index/IndexStatusManager.java | 117 +++++++-- .../test/sai/IndexAvailabilityTest.java | 229 ++++++++++++------ .../index/IndexStatusManagerTest.java | 28 ++- 5 files changed, 302 insertions(+), 109 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 003d870a25..bb5f13abce 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.3 + * Streamline the serialized format for index status gossip messages (CASSANDRA-20058) * Batch clusterings into single SAI partition post-filtering reads (CASSANDRA-19497) * Ban the usage of "var" instead of full types in the production code (CASSANDRA-20038) * Suppress CVE-2024-45772 from lucene-core-9.7.0.jar (CASSANDRA-20024) diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java index 8abc800e0f..f9e2265631 100644 --- a/src/java/org/apache/cassandra/index/Index.java +++ b/src/java/org/apache/cassandra/index/Index.java @@ -1031,10 +1031,36 @@ public interface Index */ enum Status { - UNKNOWN, - FULL_REBUILD_STARTED, - BUILD_FAILED, - BUILD_SUCCEEDED, - DROPPED + UNKNOWN(0), + FULL_REBUILD_STARTED(1), + BUILD_FAILED(2), + BUILD_SUCCEEDED(3), + DROPPED(4); + + public final int code; + + Status(int code) + { + this.code = code; + } + + static Status fromCode(int code) + { + switch (code) + { + case 0: + return UNKNOWN; + case 1: + return FULL_REBUILD_STARTED; + case 2: + return BUILD_FAILED; + case 3: + return BUILD_SUCCEEDED; + case 4: + return DROPPED; + } + + throw new IllegalArgumentException("Unrecognized code: " + code); + } } } diff --git a/src/java/org/apache/cassandra/index/IndexStatusManager.java b/src/java/org/apache/cassandra/index/IndexStatusManager.java index 1c0f5887db..6ba91beb7b 100644 --- a/src/java/org/apache/cassandra/index/IndexStatusManager.java +++ b/src/java/org/apache/cassandra/index/IndexStatusManager.java @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -45,6 +46,7 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.serializers.MarshalException; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JsonUtils; @@ -73,8 +75,7 @@ public class IndexStatusManager */ public final Map> peerIndexStatus = new HashMap<>(); - private IndexStatusManager() - {} + private IndexStatusManager() {} /** * Remove endpoints whose indexes are not queryable for the specified {@link Index.QueryPlan}. @@ -148,29 +149,57 @@ public class IndexStatusManager if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort())) return; - Map peerStatus = JsonUtils.fromJsonMap(versionedValue.value); - Map indexStatus = new HashMap<>(); + Map indexStatusMap = statusMapFromString(versionedValue); - for (Map.Entry e : peerStatus.entrySet()) - { - String keyspaceIndex = e.getKey(); - Index.Status status = Index.Status.valueOf(e.getValue()); - indexStatus.put(keyspaceIndex, status); - } - - Map oldStatus = peerIndexStatus.put(endpoint, indexStatus); - Map updated = updatedIndexStatuses(oldStatus, indexStatus); - Set removed = removedIndexStatuses(oldStatus, indexStatus); + Map oldStatus = peerIndexStatus.put(endpoint, indexStatusMap); + Map updated = updatedIndexStatuses(oldStatus, indexStatusMap); + Set removed = removedIndexStatuses(oldStatus, indexStatusMap); if (!updated.isEmpty() || !removed.isEmpty()) logger.debug("Received index status for peer {}:\n Updated: {}\n Removed: {}", endpoint, updated, removed); } - catch (MarshalException | IllegalArgumentException e) + catch (Exception e) { - logger.warn("Unable to parse index status: {}", e.getMessage()); + logger.error("Unable to parse index status: {}", e.getMessage()); } } + private Map statusMapFromString(VersionedValue versionedValue) + { + Map peerStatus = JsonUtils.fromJsonMap(versionedValue.value); + Map indexStatusMap = new HashMap<>(); + + for (Map.Entry endpointStatus : peerStatus.entrySet()) + { + String keyspaceOrIndex = endpointStatus.getKey(); + Object keyspaceOrIndexStatus = endpointStatus.getValue(); + + if (keyspaceOrIndexStatus instanceof String) + { + // This is the legacy format: (fully qualified index name -> enum string) + Index.Status status = Index.Status.valueOf(keyspaceOrIndexStatus.toString()); + indexStatusMap.put(keyspaceOrIndex, status); + } + else if (keyspaceOrIndexStatus instanceof Map) + { + // This is the new format. (keyspace -> (index -> numeric enum code)) + @SuppressWarnings("unchecked") + Map keyspaceIndexStatusMap = (Map) keyspaceOrIndexStatus; + + for (Map.Entry indexStatus : keyspaceIndexStatusMap.entrySet()) + { + Index.Status status = Index.Status.fromCode(indexStatus.getValue()); + indexStatusMap.put(identifier(keyspaceOrIndex, indexStatus.getKey()), status); + } + } + else + { + throw new MarshalException("Invalid index status format: " + endpointStatus); + } + } + return indexStatusMap; + } + /** * Propagate a new index status to the ring. The new index status is added to the current index status map * and the whole map is sent to the ring as a {@link VersionedValue}. @@ -183,35 +212,79 @@ public class IndexStatusManager { try { - Map states = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(), + Map statusMap = peerIndexStatus.computeIfAbsent(FBUtilities.getBroadcastAddressAndPort(), k -> new HashMap<>()); String keyspaceIndex = identifier(keyspace, index); if (status == Index.Status.DROPPED) - states.remove(keyspaceIndex); + statusMap.remove(keyspaceIndex); else - states.put(keyspaceIndex, status); + statusMap.put(keyspaceIndex, status); // Don't try and propagate if the gossiper isn't enabled. This is primarily for tests where the // Gossiper has not been started. If we attempt to propagate when not started an exception is // logged and this causes a number of dtests to fail. if (Gossiper.instance.isEnabled()) { - String newStatus = JsonUtils.JSON_OBJECT_MAPPER.writeValueAsString(states); + // Versions 5.0.0 through 5.0.2 use a much more bloated format that duplicates keyspace names + // and writes full status names instead of their numeric codes. If the minimum cluster version is + // unknown or one of those 3 versions, continue to propagate the old format. + CassandraVersion minVersion = Gossiper.instance.getMinVersion(1, TimeUnit.SECONDS); + String newSerializedStatusMap = shouldWriteLegacyStatusFormat(minVersion) ? JsonUtils.writeAsJsonString(statusMap) + : toSerializedFormat(statusMap); + statusPropagationExecutor.submit(() -> { // schedule gossiper update asynchronously to avoid potential deadlock when another thread is holding // gossiper taskLock. - VersionedValue value = StorageService.instance.valueFactory.indexStatus(newStatus); + VersionedValue value = StorageService.instance.valueFactory.indexStatus(newSerializedStatusMap); Gossiper.instance.addLocalApplicationState(ApplicationState.INDEX_STATUS, value); }); } } - catch (Throwable e) + catch (Exception e) { logger.warn("Unable to propagate index status: {}", e.getMessage()); } } + private static boolean shouldWriteLegacyStatusFormat(CassandraVersion minVersion) + { + return minVersion == null || (minVersion.major == 5 && minVersion.minor == 0 && minVersion.patch < 3); + } + + /** + * Serializes as a JSON string the status of the indexes in the provided map. + *

+ * For example, the map... + * 

+     * {
+     *     ks1.cf1_idx1=FULL_REBUILD_STARTED,
+     *     ks1.cf1_idx2=FULL_REBUILD_STARTED,
+     *     system.PaxosUncommittedIndex=BUILD_SUCCEEDED
+     * }
+     *
+ * ...will be converted to the string... + * 
+     * {
+     *     "system": {"PaxosUncommittedIndex": 3},
+     *     "ks1": {"cf1_idx1": 1, "cf1_idx2": 1}
+     * }
+     *
+ */ + public static String toSerializedFormat(Map indexStatusMap) + { + Map> serialized = new HashMap<>(); + + for (Map.Entry e : indexStatusMap.entrySet()) + { + String[] keyspaceAndIndex = e.getKey().split("\\."); + serialized.computeIfAbsent(keyspaceAndIndex[0], ignore -> new HashMap<>()) + .put(keyspaceAndIndex[1], e.getValue().code); + } + + return JsonUtils.writeAsJsonString(serialized); + } + @VisibleForTesting public synchronized Index.Status getIndexStatus(InetAddressAndPort peer, String keyspace, String index) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java index 3a9e111bad..5731281601 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexAvailabilityTest.java @@ -29,6 +29,9 @@ import java.util.concurrent.TimeUnit; import com.google.common.base.Objects; import org.junit.Test; +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.test.TestBaseImpl; @@ -41,6 +44,7 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.FBUtilities; +import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; import static org.apache.cassandra.distributed.test.sai.SAIUtil.waitForIndexQueryable; @@ -62,88 +66,117 @@ public class IndexAvailabilityTest extends TestBaseImpl public void verifyIndexStatusPropagation() throws Exception { try (Cluster cluster = init(Cluster.build(2) - .withConfig(config -> config.with(GOSSIP) - .with(NETWORK)) + .withConfig(config -> config.with(GOSSIP).with(NETWORK)) .start())) { - String ks1 = "ks1"; - String ks2 = "ks2"; - String ks3 = "ks3"; - String cf1 = "cf1"; - String index1 = "cf1_idx1"; - String index2 = "cf1_idx2"; - - keyspaces = Arrays.asList(ks1, ks2, ks3); - indexesPerKs = Arrays.asList(index1, index2); - - // create 1 tables per keyspace, 2 indexes per table. all indexes are queryable - for (String ks : keyspaces) - { - cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2)); - cluster.schemaChange(String.format(CREATE_TABLE, ks, cf1)); - cluster.schemaChange(String.format(CREATE_INDEX, index1, ks, cf1, "v1")); - cluster.schemaChange(String.format(CREATE_INDEX, index2, ks, cf1, "v2")); - waitForIndexQueryable(cluster, ks); - cluster.forEach(node -> { - expectedNodeIndexQueryability.put(NodeIndex.create(ks, index1, node), Index.Status.BUILD_SUCCEEDED); - expectedNodeIndexQueryability.put(NodeIndex.create(ks, index2, node), Index.Status.BUILD_SUCCEEDED); - }); - } - - // mark ks1 index1 as non-queryable on node1 - markIndexNonQueryable(cluster.get(1), ks1, cf1, index1); - // on node2, it observes that node1 ks1.index1 is not queryable - waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_FAILED); - // other indexes or keyspaces should not be affected - assertIndexingStatus(cluster); - - // mark ks2 index2 as non-queryable on node2 - markIndexNonQueryable(cluster.get(2), ks2, cf1, index2); - // on node1, it observes that node2 ks2.index2 is not queryable - waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.BUILD_FAILED); - // other indexes or keyspaces should not be affected - assertIndexingStatus(cluster); - - // mark ks1 index1 as queryable on node1 - markIndexQueryable(cluster.get(1), ks1, cf1, index1); - // on node2, it observes that node1 ks1.index1 is queryable - waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED); - // other indexes or keyspaces should not be affected - assertIndexingStatus(cluster); - - // mark ks2 index2 as indexing on node1 - markIndexBuilding(cluster.get(1), ks2, cf1, index2); - // on node2, it observes that node1 ks2.index2 is not queryable - waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED); - // other indexes or keyspaces should not be affected - assertIndexingStatus(cluster); - - // drop ks1, ks1 index1/index2 should be non queryable on all nodes - cluster.schemaChange("DROP KEYSPACE " + ks1); - expectedNodeIndexQueryability.keySet().forEach(k -> { - if (k.keyspace.equals(ks1)) - expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); - }); - assertIndexingStatus(cluster); - - // drop ks2 index2, there should be no ks2 index2 status on all node - cluster.schemaChange("DROP INDEX " + ks2 + "." + index2); - expectedNodeIndexQueryability.keySet().forEach(k -> { - if (k.keyspace.equals(ks2) && k.index.equals(index2)) - expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); - }); - assertIndexingStatus(cluster); - - // drop ks3 cf1, there should be no ks3 index1/index2 status - cluster.schemaChange("DROP TABLE " + ks3 + "." + cf1); - expectedNodeIndexQueryability.keySet().forEach(k -> { - if (k.keyspace.equals(ks3)) - expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); - }); - assertIndexingStatus(cluster); + verifyIndexStatusPropagation(cluster); } } + @Test + public void verifyIndexStatusPropagationMixedPatchVersion() throws Exception + { + try (Cluster cluster = init(Cluster.build(2) + .withConfig(config -> config.with(GOSSIP).with(NETWORK)) + .withInstanceInitializer(MixedPatchVersionHelper::setVersions) + .start())) + { + verifyIndexStatusPropagation(cluster); + } + } + + @Test + public void verifyIndexStatusPropagationMixedMajorVersion() throws Exception + { + try (Cluster cluster = init(Cluster.build(2) + .withConfig(config -> config.with(GOSSIP).with(NETWORK)) + .withInstanceInitializer(MixedMajorVersionHelper::setVersions) + .start())) + { + verifyIndexStatusPropagation(cluster); + } + } + + private void verifyIndexStatusPropagation(Cluster cluster) + { + String ks1 = "ks1"; + String ks2 = "ks2"; + String ks3 = "ks3"; + String cf1 = "cf1"; + String index1 = "cf1_idx1"; + String index2 = "cf1_idx2"; + + keyspaces = Arrays.asList(ks1, ks2, ks3); + indexesPerKs = Arrays.asList(index1, index2); + + // create 1 tables per keyspace, 2 indexes per table. all indexes are queryable + for (String ks : keyspaces) + { + cluster.schemaChange(String.format(CREATE_KEYSPACE, ks, 2)); + cluster.schemaChange(String.format(CREATE_TABLE, ks, cf1)); + cluster.schemaChange(String.format(CREATE_INDEX, index1, ks, cf1, "v1")); + cluster.schemaChange(String.format(CREATE_INDEX, index2, ks, cf1, "v2")); + waitForIndexQueryable(cluster, ks); + cluster.forEach(node -> { + expectedNodeIndexQueryability.put(NodeIndex.create(ks, index1, node), Index.Status.BUILD_SUCCEEDED); + expectedNodeIndexQueryability.put(NodeIndex.create(ks, index2, node), Index.Status.BUILD_SUCCEEDED); + }); + } + + // mark ks1 index1 as non-queryable on node1 + markIndexNonQueryable(cluster.get(1), ks1, cf1, index1); + // on node2, it observes that node1 ks1.index1 is not queryable + waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_FAILED); + // other indexes or keyspaces should not be affected + assertIndexingStatus(cluster); + + // mark ks2 index2 as non-queryable on node2 + markIndexNonQueryable(cluster.get(2), ks2, cf1, index2); + // on node1, it observes that node2 ks2.index2 is not queryable + waitForIndexingStatus(cluster.get(1), ks2, index2, cluster.get(2), Index.Status.BUILD_FAILED); + // other indexes or keyspaces should not be affected + assertIndexingStatus(cluster); + + // mark ks1 index1 as queryable on node1 + markIndexQueryable(cluster.get(1), ks1, cf1, index1); + // on node2, it observes that node1 ks1.index1 is queryable + waitForIndexingStatus(cluster.get(2), ks1, index1, cluster.get(1), Index.Status.BUILD_SUCCEEDED); + // other indexes or keyspaces should not be affected + assertIndexingStatus(cluster); + + // mark ks2 index2 as indexing on node1 + markIndexBuilding(cluster.get(1), ks2, cf1, index2); + // on node2, it observes that node1 ks2.index2 is not queryable + waitForIndexingStatus(cluster.get(2), ks2, index2, cluster.get(1), Index.Status.FULL_REBUILD_STARTED); + // other indexes or keyspaces should not be affected + assertIndexingStatus(cluster); + + // drop ks1, ks1 index1/index2 should be non queryable on all nodes + cluster.schemaChange("DROP KEYSPACE " + ks1); + expectedNodeIndexQueryability.keySet().forEach(k -> { + if (k.keyspace.equals(ks1)) + expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); + }); + assertIndexingStatus(cluster); + + // drop ks2 index2, there should be no ks2 index2 status on all node + cluster.schemaChange("DROP INDEX " + ks2 + '.' + index2); + expectedNodeIndexQueryability.keySet().forEach(k -> { + if (k.keyspace.equals(ks2) && k.index.equals(index2)) + expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); + }); + assertIndexingStatus(cluster); + + // drop ks3 cf1, there should be no ks3 index1/index2 status + cluster.schemaChange("DROP TABLE " + ks3 + '.' + cf1); + expectedNodeIndexQueryability.keySet().forEach(k -> { + if (k.keyspace.equals(ks3)) + expectedNodeIndexQueryability.put(k, Index.Status.UNKNOWN); + }); + assertIndexingStatus(cluster); + } + + @SuppressWarnings("DataFlowIssue") private void markIndexNonQueryable(IInvokableInstance node, String keyspace, String table, String indexName) { expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_FAILED); @@ -155,6 +188,7 @@ public class IndexAvailabilityTest extends TestBaseImpl }); } + @SuppressWarnings("DataFlowIssue") private void markIndexQueryable(IInvokableInstance node, String keyspace, String table, String indexName) { expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.BUILD_SUCCEEDED); @@ -166,6 +200,7 @@ public class IndexAvailabilityTest extends TestBaseImpl }); } + @SuppressWarnings("DataFlowIssue") private void markIndexBuilding(IInvokableInstance node, String keyspace, String table, String indexName) { expectedNodeIndexQueryability.put(NodeIndex.create(keyspace, indexName, node), Index.Status.FULL_REBUILD_STARTED); @@ -285,4 +320,44 @@ public class IndexAvailabilityTest extends TestBaseImpl return Objects.hashCode(keyspace, index, node); } } + + public static class MixedMajorVersionHelper + { + @SuppressWarnings({ "unused", "resource" }) + static void setVersions(ClassLoader loader, int node) + { + if (node == 1) + new ByteBuddy().rebase(FBUtilities.class) + .method(named("getReleaseVersionString")) + .intercept(MethodDelegation.to(MixedMajorVersionHelper.class)) + .make() + .load(loader, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings("unused") + public static String getReleaseVersionString() + { + return "4.1.0"; + } + } + + public static class MixedPatchVersionHelper + { + @SuppressWarnings({ "unused", "resource" }) + static void setVersions(ClassLoader loader, int node) + { + if (node == 1) + new ByteBuddy().rebase(FBUtilities.class) + .method(named("getReleaseVersionString")) + .intercept(MethodDelegation.to(MixedPatchVersionHelper.class)) + .make() + .load(loader, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings("unused") + public static String getReleaseVersionString() + { + return "5.0.2"; + } + } } diff --git a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java index d08fec974a..331e079680 100644 --- a/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java +++ b/test/unit/org/apache/cassandra/index/IndexStatusManagerTest.java @@ -19,12 +19,16 @@ package org.apache.cassandra.index; import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.TreeSet; import java.util.stream.Collectors; import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; import org.mockito.Mockito; import org.apache.cassandra.db.ConsistencyLevel; @@ -47,8 +51,21 @@ import static org.apache.cassandra.locator.ReplicaUtils.full; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.Assert.assertArrayEquals; +@RunWith(Parameterized.class) public class IndexStatusManagerTest { + @Parameterized.Parameter + public boolean legacyStatusFormat; + + @Parameterized.Parameters(name = "{0}") + public static List parameters() + { + List parameters = new ArrayList<>(); + parameters.add(new Object[] { true }); + parameters.add(new Object[] { false }); + return parameters; + } + static class Testcase { String keyspace; @@ -389,11 +406,12 @@ public class IndexStatusManagerTest .collect(Collectors.toSet()); // send indexStatus for each endpoint - testcase.indexStatus.forEach((endpoint, indexStatus) -> - IndexStatusManager.instance.receivePeerIndexStatus( - endpoint, - VersionedValue.unsafeMakeVersionedValue(JsonUtils.writeAsJsonString(indexStatus), 1) - )); + testcase.indexStatus.forEach((endpoint, indexStatusMap) -> + { + String serialized = legacyStatusFormat ? JsonUtils.writeAsJsonString(indexStatusMap) + : IndexStatusManager.toSerializedFormat(indexStatusMap); + IndexStatusManager.instance.receivePeerIndexStatus(endpoint, VersionedValue.unsafeMakeVersionedValue(serialized, 1)); + }); // sort the replicas here, so that we can assert the order later EndpointsForRange endpoints = EndpointsForRange.copyOf(new TreeSet<>(replicas));