From 58a5ce14ba80a0a4eeef0f6b18e58d63113e1159 Mon Sep 17 00:00:00 2001 From: Jon Meredith Date: Mon, 26 Aug 2019 15:07:44 -0600 Subject: [PATCH] In-JVM DTest: Add network topology and tracing support In-JVM DTest: readRepairTest - Set read repair query to CL.ALL The current test relies on the order of nodes returned by the snitch to include node3, which SimpleSnitch does. With support for other snitches coming then the test should be able to handle any order of nodes - so make sure all nodes are present. In-JVM DTest: remove minimum messaging service calculation Match change on trunk to resolves issue with trying to call getMessagingVersion on nodes that are not started. Fixes mixed version upgrades once all branches are updated. Patch by Jon Meredith; Reviewed by Dinesh Joshi and Alex Petrov for CASSANDRA-15319 --- CHANGES.txt | 1 + .../cassandra/service/StorageService.java | 8 +- .../apache/cassandra/distributed/Cluster.java | 7 +- .../distributed/UpgradeableCluster.java | 9 +- .../cassandra/distributed/api/ICluster.java | 2 + .../distributed/api/ICoordinator.java | 5 + .../distributed/api/IInstanceConfig.java | 16 +- .../distributed/impl/AbstractCluster.java | 151 +++++++++++++++++- .../distributed/impl/Coordinator.java | 65 +++++--- .../impl/DistributedTestSnitch.java | 61 +++++++ .../cassandra/distributed/impl/Instance.java | 36 ++++- .../distributed/impl/InstanceClassLoader.java | 3 +- .../distributed/impl/InstanceConfig.java | 32 ++-- .../distributed/impl/IsolatedExecutor.java | 6 +- .../distributed/impl/NetworkTopology.java | 89 +++++++++++ .../distributed/impl/TracingUtil.java | 115 +++++++++++++ .../test/DistributedReadWritePathTest.java | 2 +- .../test/MessageForwardingTest.java | 92 +++++++++++ .../distributed/test/NetworkTopologyTest.java | 99 ++++++++++++ 19 files changed, 738 insertions(+), 61 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java create mode 100644 test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java create mode 100644 test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java diff --git a/CHANGES.txt b/CHANGES.txt index caea0f4ae7..fefd40a0c0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -9,6 +9,7 @@ * Make tools/bin/token-generator py2/3 compatible (CASSANDRA-15012) * Multi-version in-JVM dtests (CASSANDRA-14937) * Allow instance class loaders to be garbage collected for inJVM dtest (CASSANDRA-15170) + * Add support for network topology and query tracing for inJVM dtest (CASSANDRA-15319) 2.2.14 diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 0a9a8da986..b1d8e265eb 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -948,7 +948,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } // if we don't have system_traces keyspace at this point, then create it manually - maybeAddOrUpdateKeyspace(TraceKeyspace.definition()); + ensureTraceKeyspace(); maybeAddOrUpdateKeyspace(SystemDistributedKeyspace.definition()); if (!isSurveyMode) @@ -978,6 +978,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } + @VisibleForTesting + public void ensureTraceKeyspace() + { + maybeAddOrUpdateKeyspace(TraceKeyspace.definition()); + } + public static boolean isReplacingSameAddress() { return DatabaseDescriptor.getReplaceAddress().equals(FBUtilities.getBroadcastAddress()); diff --git a/test/distributed/org/apache/cassandra/distributed/Cluster.java b/test/distributed/org/apache/cassandra/distributed/Cluster.java index 95862b6a62..d3533d3fa9 100644 --- a/test/distributed/org/apache/cassandra/distributed/Cluster.java +++ b/test/distributed/org/apache/cassandra/distributed/Cluster.java @@ -45,9 +45,14 @@ public class Cluster extends AbstractCluster implements IClu return new Wrapper(generation, version, config); } + public static Builder build() + { + return new Builder<>(Cluster::new); + } + public static Builder build(int nodeCount) { - return new Builder<>(nodeCount, Cluster::new); + return build().withNodes(nodeCount); } public static Cluster create(int nodeCount, Consumer configUpdater) throws IOException diff --git a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java index 232ef0b7b3..1fe960a253 100644 --- a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java @@ -19,8 +19,6 @@ package org.apache.cassandra.distributed; import java.io.File; -import java.io.IOException; -import java.nio.file.Files; import java.util.List; import org.apache.cassandra.distributed.api.ICluster; @@ -48,9 +46,14 @@ public class UpgradeableCluster extends AbstractCluster im return new Wrapper(generation, version, config); } + public static Builder build() + { + return new Builder<>(UpgradeableCluster::new); + } + public static Builder build(int nodeCount) { - return new Builder<>(nodeCount, UpgradeableCluster::new); + return build().withNodes(nodeCount); } public static UpgradeableCluster create(int nodeCount) throws Throwable diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICluster.java b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java index 91da61ed34..091e5f048a 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/ICluster.java +++ b/test/distributed/org/apache/cassandra/distributed/api/ICluster.java @@ -29,6 +29,8 @@ public interface ICluster IInstance get(InetAddressAndPort endpoint); int size(); Stream stream(); + Stream stream(String dcName); + Stream stream(String dcName, String rackName); IMessageFilters filters(); } diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java index 59afd8bde6..ef4485318b 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java @@ -19,6 +19,8 @@ package org.apache.cassandra.distributed.api; import java.util.Iterator; +import java.util.UUID; +import java.util.concurrent.Future; // The cross-version API requires that a Coordinator can be constructed without any constructor arguments public interface ICoordinator @@ -28,4 +30,7 @@ public interface ICoordinator Object[][] execute(String query, Enum consistencyLevel, Object... boundValues); Iterator executeWithPaging(String query, Enum consistencyLevel, int pageSize, Object... boundValues); + + Future asyncExecuteWithTracing(UUID sessionId, String query, Enum consistencyLevel, Object... boundValues); + Object[][] executeWithTracing(UUID sessionId, String query, Enum consistencyLevel, Object... boundValues); } diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java index 3e5a18fe7c..dd21b96743 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java @@ -18,15 +18,27 @@ package org.apache.cassandra.distributed.api; -import org.apache.cassandra.locator.InetAddressAndPort; - import java.util.UUID; +import org.apache.cassandra.distributed.impl.NetworkTopology; +import org.apache.cassandra.locator.InetAddressAndPort; + public interface IInstanceConfig { int num(); UUID hostId(); InetAddressAndPort broadcastAddressAndPort(); + NetworkTopology networkTopology(); + + default public String localRack() + { + return networkTopology().localRack(broadcastAddressAndPort()); + } + + default public String localDatacenter() + { + return networkTopology().localDC(broadcastAddressAndPort()); + } /** * write the specified parameters to the Config object; we do not specify Config as the type to support a Config diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 19fb7e57fc..f603497675 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; +import java.util.stream.IntStream; import java.util.stream.Stream; import com.google.common.collect.Sets; @@ -57,6 +58,7 @@ import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.concurrent.SimpleCondition; /** @@ -193,13 +195,15 @@ public abstract class AbstractCluster implements ICluster, } } - protected AbstractCluster(File root, Versions.Version version, List configs, ClassLoader sharedClassLoader) + protected AbstractCluster(File root, Versions.Version version, List configs, + ClassLoader sharedClassLoader) { this.root = root; this.sharedClassLoader = sharedClassLoader; this.instances = new ArrayList<>(); this.instanceMap = new HashMap<>(); int generation = AbstractCluster.generation.incrementAndGet(); + for (InstanceConfig config : configs) { I instance = newInstanceWrapper(generation, version, config); @@ -231,7 +235,20 @@ public abstract class AbstractCluster implements ICluster, { return instances.size(); } + public Stream stream() { return instances.stream(); } + + public Stream stream(String dcName) + { + return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName)); + } + + public Stream stream(String dcName, String rackName) + { + return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName) && + i.config().localRack().equals(rackName)); + } + public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); } public void forEach(Consumer consumer) { instances.forEach(consumer); } public void parallelForEach(IIsolatedExecutor.SerializableConsumer consumer, long timeout, TimeUnit units) @@ -353,15 +370,16 @@ public abstract class AbstractCluster implements ICluster, public static class Builder> { - private final int nodeCount; private final Factory factory; + private int nodeCount; private int subnet; + private Map> nodeIdTopology; private File root; private Versions.Version version; private Consumer configUpdater; - public Builder(int nodeCount, Factory factory) + + public Builder(Factory factory) { - this.nodeCount = nodeCount; this.factory = factory; } @@ -371,6 +389,97 @@ public abstract class AbstractCluster implements ICluster, return this; } + public Builder withNodes(int nodeCount) { + this.nodeCount = nodeCount; + return this; + } + + public Builder withDCs(int dcCount) + { + return withRacks(dcCount, 1); + } + + public Builder withRacks(int dcCount, int racksPerDC) + { + if (nodeCount == 0) + throw new IllegalStateException("Node count will be calculated. Do not supply total node count in the builder"); + + int totalRacks = dcCount * racksPerDC; + int nodesPerRack = (nodeCount + totalRacks - 1) / totalRacks; // round up to next integer + return withRacks(dcCount, racksPerDC, nodesPerRack); + } + + public Builder withRacks(int dcCount, int racksPerDC, int nodesPerRack) + { + if (nodeIdTopology != null) + throw new IllegalStateException("Network topology already created. Call withDCs/withRacks once or before withDC/withRack calls"); + + nodeIdTopology = new HashMap<>(); + int nodeId = 1; + for (int dc = 1; dc <= dcCount; dc++) + { + for (int rack = 1; rack <= racksPerDC; rack++) + { + for (int rackNodeIdx = 0; rackNodeIdx < nodesPerRack; rackNodeIdx++) + nodeIdTopology.put(nodeId++, Pair.create(dcName(dc), rackName(rack))); + } + } + // adjust the node count to match the allocatation + final int adjustedNodeCount = dcCount * racksPerDC * nodesPerRack; + if (adjustedNodeCount != nodeCount) + { + assert adjustedNodeCount > nodeCount : "withRacks should only ever increase the node count"; + logger.info("Network topology of {} DCs with {} racks per DC and {} nodes per rack required increasing total nodes to {}", + dcCount, racksPerDC, nodesPerRack, adjustedNodeCount); + nodeCount = adjustedNodeCount; + } + return this; + } + + public Builder withDC(String dcName, int nodeCount) + { + return withRack(dcName, rackName(1), nodeCount); + } + + public Builder withRack(String dcName, String rackName, int nodesInRack) + { + if (nodeIdTopology == null) + { + if (nodeCount > 0) + throw new IllegalStateException("Node count must not be explicitly set, or allocated using withDCs/withRacks"); + + nodeIdTopology = new HashMap<>(); + } + for (int nodeId = nodeCount + 1; nodeId <= nodeCount + nodesInRack; nodeId++) + nodeIdTopology.put(nodeId, Pair.create(dcName, rackName)); + + nodeCount += nodesInRack; + return this; + } + + // Map of node ids to dc and rack - must be contiguous with an entry nodeId 1 to nodeCount + public Builder withNodeIdTopology(Map> nodeIdTopology) + { + if (nodeIdTopology.isEmpty()) + throw new IllegalStateException("Topology is empty. It must have an entry for every nodeId."); + + IntStream.rangeClosed(1, nodeIdTopology.size()).forEach(nodeId -> { + if (nodeIdTopology.get(nodeId) == null) + throw new IllegalStateException("Topology is missing entry for nodeId " + nodeId); + }); + + if (nodeCount != nodeIdTopology.size()) + { + nodeCount = nodeIdTopology.size(); + logger.info("Adjusting node count to {} for supplied network topology", nodeCount); + + } + + this.nodeIdTopology = new HashMap<>(nodeIdTopology); + + return this; + } + public Builder withRoot(File root) { this.root = root; @@ -396,9 +505,18 @@ public abstract class AbstractCluster implements ICluster, if (root == null) root = Files.createTempDirectory("dtests").toFile(); + if (version == null) version = Versions.CURRENT; + if (nodeCount <= 0) + throw new IllegalStateException("Cluster must have at least one node"); + + if (nodeIdTopology == null) + nodeIdTopology = IntStream.rangeClosed(1, nodeCount).boxed() + .collect(Collectors.toMap(nodeId -> nodeId, + nodeId -> Pair.create(dcName(0), rackName(0)))); + root.mkdirs(); setupLogging(root); @@ -406,17 +524,23 @@ public abstract class AbstractCluster implements ICluster, List configs = new ArrayList<>(); long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount); - for (int i = 0; i < nodeCount; ++i) + + String ipPrefix = "127.0." + subnet + "."; + + NetworkTopology networkTopology = NetworkTopology.build(ipPrefix, 7012, nodeIdTopology); + + for (int i = 0 ; i < nodeCount ; ++i) { - InstanceConfig config = InstanceConfig.generate(i + 1, subnet, root, String.valueOf(token)); + int nodeNum = i + 1; + String ipAddress = ipPrefix + nodeNum; + InstanceConfig config = InstanceConfig.generate(i + 1, ipAddress, networkTopology, root, String.valueOf(token)); if (configUpdater != null) configUpdater.accept(config); configs.add(config); token += increment; } - C cluster = factory.newCluster(root, version, configs, sharedClassLoader); - return cluster; + return factory.newCluster(root, version, configs, sharedClassLoader); } public C start() throws IOException @@ -427,6 +551,15 @@ public abstract class AbstractCluster implements ICluster, } } + static String dcName(int index) + { + return "datacenter" + index; + } + + static String rackName(int index) + { + return "rack" + index; + } private static void setupLogging(File root) { @@ -434,11 +567,13 @@ public abstract class AbstractCluster implements ICluster, { String testConfPath = "test/conf/logback-dtest.xml"; Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml"); + if (!logConfPath.toFile().exists()) { Files.copy(new File(testConfPath).toPath(), logConfPath); } + System.setProperty("logback.configurationFile", "file://" + logConfPath); } catch (IOException e) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index 96b0dd176a..cca364338b 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -22,6 +22,8 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.UUID; +import java.util.concurrent.Future; import org.apache.cassandra.cql3.CQLStatement; import org.apache.cassandra.cql3.QueryOptions; @@ -36,6 +38,7 @@ import org.apache.cassandra.service.pager.Pageable; import org.apache.cassandra.service.pager.QueryPager; import org.apache.cassandra.service.pager.QueryPagers; import org.apache.cassandra.transport.Server; +import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.ByteBufferUtil; @@ -50,36 +53,54 @@ public class Coordinator implements ICoordinator @Override public Object[][] execute(String query, Enum consistencyLevelOrigin, Object... boundValues) { - return instance.sync(() -> { - ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); - CQLStatement prepared = QueryProcessor.getStatement(query, ClientState.forInternalCalls()).statement; - List boundBBValues = new ArrayList<>(); - for (Object boundValue : boundValues) - { - boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); - } + return instance.sync(() -> executeInternal(query, consistencyLevelOrigin, boundValues)).call(); + } - prepared.validate(QueryState.forInternalCalls().getClientState()); - ResultMessage res = prepared.execute(QueryState.forInternalCalls(), - QueryOptions.create(consistencyLevel, - boundBBValues, - false, - Integer.MAX_VALUE, - null, - null, - Server.CURRENT_VERSION)); - - if (res != null && res.kind == ResultMessage.Kind.ROWS) + public Future asyncExecuteWithTracing(UUID sessionId, String query, Enum consistencyLevelOrigin, Object... boundValues) + { + return instance.async(() -> { + try { - return RowUtil.toObjects((ResultMessage.Rows) res); + Tracing.instance.newSession(sessionId); + return executeInternal(query, consistencyLevelOrigin, boundValues); } - else + finally { - return new Object[][]{}; + Tracing.instance.stopSession(); } }).call(); } + private Object[][] executeInternal(String query, Enum consistencyLevelOrigin, Object[] boundValues) + { + ClientState clientState = ClientState.forInternalCalls(); + CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement; + List boundBBValues = new ArrayList<>(); + ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); + for (Object boundValue : boundValues) + boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); + + prepared.validate(QueryState.forInternalCalls().getClientState()); + ResultMessage res = prepared.execute(QueryState.forInternalCalls(), + QueryOptions.create(consistencyLevel, + boundBBValues, + false, + Integer.MAX_VALUE, + null, + null, + Server.CURRENT_VERSION)); + + if (res != null && res.kind == ResultMessage.Kind.ROWS) + return RowUtil.toObjects((ResultMessage.Rows) res); + else + return new Object[][]{}; + } + + public Object[][] executeWithTracing(UUID sessionId, String query, Enum consistencyLevelOrigin, Object... boundValues) + { + return IsolatedExecutor.waitOn(asyncExecuteWithTracing(sessionId, query, consistencyLevelOrigin, boundValues)); + } + @Override public Iterator executeWithPaging(String query, Enum consistencyLevelOrigin, int pageSize, Object... boundValues) { diff --git a/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java b/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java new file mode 100644 index 0000000000..35e2903a17 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.impl; + +import java.net.InetAddress; + +import org.apache.cassandra.config.Config; +import org.apache.cassandra.locator.AbstractNetworkTopologySnitch; +import org.apache.cassandra.locator.InetAddressAndPort; + +public class DistributedTestSnitch extends AbstractNetworkTopologySnitch +{ + private static NetworkTopology mapping = null; + + public String getRack(InetAddress endpoint) + { + assert mapping != null : "network topology must be assigned before using snitch"; + int storage_port = Config.getOverrideLoadConfig().get().storage_port; + return mapping.localRack(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port)); + } + + public String getRack(InetAddressAndPort endpoint) + { + assert mapping != null : "network topology must be assigned before using snitch"; + return mapping.localRack(endpoint); + } + + public String getDatacenter(InetAddress endpoint) + { + assert mapping != null : "network topology must be assigned before using snitch"; + int storage_port = Config.getOverrideLoadConfig().get().storage_port; + return mapping.localDC(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port)); + } + + public String getDatacenter(InetAddressAndPort endpoint) + { + assert mapping != null : "network topology must be assigned before using snitch"; + return mapping.localDC(endpoint); + } + + static void assign(NetworkTopology newMapping) + { + mapping = new NetworkTopology(newMapping); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 29426cbe89..af5ec65970 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -23,6 +23,7 @@ import java.io.DataInputStream; import java.io.File; import java.io.IOException; import java.net.InetAddress; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -69,7 +70,6 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.IMessageSink; -import org.apache.cassandra.net.MessageDeliveryTask; import org.apache.cassandra.net.MessageIn; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; @@ -78,10 +78,13 @@ import org.apache.cassandra.service.PendingRangeCalculatorService; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.StreamCoordinator; +import org.apache.cassandra.tracing.TraceState; +import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis; import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.UUIDGen; import org.apache.cassandra.utils.concurrent.Ref; import static java.util.concurrent.TimeUnit.MINUTES; @@ -240,6 +243,28 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance InetAddressAndPort toFull = lookupAddressAndPort.apply(to); int version = MessagingService.instance().getVersion(to); + // Tracing logic - similar to org.apache.cassandra.net.OutboundTcpConnection.writeConnected + byte[] sessionBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_HEADER); + if (sessionBytes != null) + { + UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes)); + TraceState state = Tracing.instance.get(sessionId); + String message = String.format("Sending %s message to %s", messageOut.verb, toFull.address); + // session may have already finished; see CASSANDRA-5668 + if (state == null) + { + byte[] traceTypeBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_TYPE); + Tracing.TraceType traceType = traceTypeBytes == null ? Tracing.TraceType.QUERY : Tracing.TraceType.deserialize(traceTypeBytes[0]); + TraceState.mutateWithTracing(ByteBuffer.wrap(sessionBytes), message, -1, traceType.getTTL()); + } + else + { + state.trace(message); + if (messageOut.verb == MessagingService.Verb.REQUEST_RESPONSE) + Tracing.instance.doneWithNonLocalSession(state); + } + } + out.writeInt(MessagingService.PROTOCOL_MAGIC); out.writeInt(id); long timestamp = System.currentTimeMillis(); @@ -326,6 +351,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { mkdirs(); + assert config.networkTopology().contains(config.broadcastAddressAndPort()); + DistributedTestSnitch.assign(config.networkTopology()); + DatabaseDescriptor.setDaemonInitialized(); DatabaseDescriptor.createAllDirectories(); @@ -378,6 +406,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance initializeRing(cluster); } + StorageService.instance.ensureTraceKeyspace(); + SystemKeyspace.finishStartup(); if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address)) @@ -443,9 +473,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance ApplicationState.STATUS, new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(tokens.get(i)))); Gossiper.instance.realMarkAlive(ep.address, Gossiper.instance.getEndpointStateForEndpoint(ep.address)); - - int version = Math.min(MessagingService.current_version, cluster.get(ep).getMessagingVersion()); - MessagingService.instance().setVersion(ep.address, version); + MessagingService.instance().setVersion(ep.address, MessagingService.current_version); } // check that all nodes are in token metadata diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java index 363a1df501..ca6d713c84 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java @@ -39,7 +39,8 @@ public class InstanceClassLoader extends URLClassLoader InetAddressAndPort.class, ParameterizedClass.class, SigarLibrary.class, - IInvokableInstance.class + IInvokableInstance.class, + NetworkTopology.class }) .map(Class::getName) .collect(Collectors.toSet()); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index efe9a0f765..88299d6a4c 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -24,7 +24,6 @@ import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.SimpleSeedProvider; -import org.apache.cassandra.locator.SimpleSnitch; import java.io.File; import java.lang.reflect.Field; @@ -43,11 +42,14 @@ public class InstanceConfig implements IInstanceConfig public final int num; public int num() { return num; } + private final NetworkTopology networkTopology; + public NetworkTopology networkTopology() { return networkTopology; } + public final UUID hostId; public UUID hostId() { return hostId; } private final Map params = new TreeMap<>(); - private EnumSet featureFlags; + private final EnumSet featureFlags; private volatile InetAddressAndPort broadcastAddressAndPort; @@ -69,6 +71,7 @@ public class InstanceConfig implements IInstanceConfig } private InstanceConfig(int num, + NetworkTopology networkTopology, String broadcast_address, String listen_address, String broadcast_rpc_address, @@ -81,6 +84,7 @@ public class InstanceConfig implements IInstanceConfig String initial_token) { this.num = num; + this.networkTopology = networkTopology; this.hostId = java.util.UUID.randomUUID(); this .set("broadcast_address", broadcast_address) .set("listen_address", listen_address) @@ -101,8 +105,8 @@ public class InstanceConfig implements IInstanceConfig .set("concurrent_compactors", 1) .set("memtable_heap_space_in_mb", 10) .set("commitlog_sync", "batch") - .set("storage_port", 7010) - .set("endpoint_snitch", SimpleSnitch.class.getName()) + .set("storage_port", 7012) + .set("endpoint_snitch", DistributedTestSnitch.class.getName()) .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(), Collections.singletonMap("seeds", "127.0.0.1"))) // legacy parameters @@ -113,9 +117,11 @@ public class InstanceConfig implements IInstanceConfig private InstanceConfig(InstanceConfig copy) { this.num = copy.num; + this.networkTopology = new NetworkTopology(copy.networkTopology); this.params.putAll(copy.params); this.hostId = copy.hostId; this.featureFlags = copy.featureFlags; + this.broadcastAddressAndPort = copy.broadcastAddressAndPort; } public InstanceConfig with(Feature featureFlag) @@ -192,11 +198,7 @@ public class InstanceConfig implements IInstanceConfig { valueField.set(writeToConfig, value); } - catch (IllegalAccessException e) - { - throw new IllegalStateException(e); - } - catch (IllegalArgumentException e) + catch (IllegalAccessException | IllegalArgumentException e) { throw new IllegalStateException(e); } @@ -217,14 +219,14 @@ public class InstanceConfig implements IInstanceConfig return (String)params.get(name); } - public static InstanceConfig generate(int nodeNum, int subnet, File root, String token) + public static InstanceConfig generate(int nodeNum, String ipAddress, NetworkTopology networkTopology, File root, String token) { - String ipPrefix = "127.0." + subnet + "."; return new InstanceConfig(nodeNum, - ipPrefix + nodeNum, - ipPrefix + nodeNum, - ipPrefix + nodeNum, - ipPrefix + nodeNum, + networkTopology, + ipAddress, + ipAddress, + ipAddress, + ipAddress, String.format("%s/node%d/saved_caches", root, nodeNum), new String[] { String.format("%s/node%d/data", root, nodeNum) }, String.format("%s/node%d/commitlog", root, nodeNum), diff --git a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java index 1d26c5dec3..248155a7e2 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java @@ -77,7 +77,7 @@ public class IsolatedExecutor implements IIsolatedExecutor return t; }; ExecutorService shutdownExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0, TimeUnit.SECONDS, - new LinkedBlockingQueue(), threadFactory); + new LinkedBlockingQueue<>(), threadFactory); return shutdownExecutor.submit(() -> { try { @@ -165,7 +165,7 @@ public class IsolatedExecutor implements IIsolatedExecutor public static Object deserializeOneObject(byte[] bytes) { try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes); - ObjectInputStream ois = new ObjectInputStream(bais);) + ObjectInputStream ois = new ObjectInputStream(bais)) { return ois.readObject(); } @@ -190,7 +190,7 @@ public class IsolatedExecutor implements IIsolatedExecutor } } - private static T waitOn(Future f) + public static T waitOn(Future f) { try { diff --git a/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java b/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java new file mode 100644 index 0000000000..1176b320bc --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.impl; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.Pair; + +public class NetworkTopology +{ + private final Map> map; + + private NetworkTopology() { + map = new HashMap<>(); + } + + @SuppressWarnings("WeakerAccess") + public NetworkTopology(NetworkTopology networkTopology) + { + map = new HashMap<>(networkTopology.map); + } + + public static NetworkTopology build(String ipPrefix, int broadcastPort, Map> nodeIdTopology) + { + final NetworkTopology topology = new NetworkTopology(); + + for (int nodeId = 1; nodeId <= nodeIdTopology.size(); nodeId++) + { + String broadcastAddress = ipPrefix + nodeId; + + try + { + Pair dcAndRack = nodeIdTopology.get(nodeId); + if (dcAndRack == null) + throw new IllegalStateException("nodeId " + nodeId + "not found in instanceMap"); + + InetAddressAndPort broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults( + InetAddress.getByName(broadcastAddress), broadcastPort); + topology.put(broadcastAddressAndPort, dcAndRack); + } + catch (UnknownHostException e) + { + throw new ConfigurationException("Unknown broadcast_address '" + broadcastAddress + '\'', false); + } + } + return topology; + } + + public Pair put(InetAddressAndPort key, Pair value) + { + return map.put(key, value); + } + + public String localRack(InetAddressAndPort key) + { + return map.get(key).right; + } + + public String localDC(InetAddressAndPort key) + { + return map.get(key).left; + } + + public boolean contains(InetAddressAndPort key) + { + return map.containsKey(key); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java new file mode 100644 index 0000000000..d671e553d9 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.impl; + +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.LinkedList; +import java.util.List; +import java.util.UUID; + +import org.apache.cassandra.db.ConsistencyLevel; + + +/** + * Utilities for accessing the system_traces table from in-JVM dtests + */ +public class TracingUtil +{ + /** + * Represents an entry from system_traces + */ + public static class TraceEntry + { + public final UUID sessionId; + public final UUID eventId; + public final String activity; + public final InetAddress source; + public final int sourceElapsed; + public final String thread; + + private TraceEntry(UUID sessionId, UUID eventId, String activity, InetAddress sourceIP, int sourceElapsed, String thread) + { + this.sessionId = sessionId; + this.eventId = eventId; + this.activity = activity; + this.source = sourceIP; + this.sourceElapsed = sourceElapsed; + this.thread = thread; + } + + static TraceEntry fromRowResultObjects(Object[] objects) + { + return new TraceEntry((UUID) objects[0], + (UUID) objects[1], + (String) objects[2], + (InetAddress) objects[3], + (Integer) objects[4], + (String) objects[5]); + } + } + + public static List getTrace(AbstractCluster cluster, UUID sessionId) + { + return getTrace(cluster, sessionId, ConsistencyLevel.ALL); + } + + public static List getTrace(AbstractCluster cluster, UUID sessionId, ConsistencyLevel cl) + { + Object[][] result = cluster.coordinator(1).execute( + "SELECT session_id, event_id, activity, source, source_elapsed, thread " + + "FROM system_traces.events WHERE session_id = ?", cl, sessionId); + + List traces = new LinkedList<>(); + for (Object[] r : result) + { + traces.add(TraceEntry.fromRowResultObjects(r)); + } + return traces; + } + + public static List getTraces(AbstractCluster cluster) + { + return getTraces(cluster, ConsistencyLevel.ALL); + } + + public static List getTraces(AbstractCluster cluster, ConsistencyLevel cl) + { + Object[][] result = cluster.coordinator(1).execute( + "SELECT session_id, event_id, activity, source, source_elapsed, thread " + + "FROM system_traces.events", cl); + + List traces = new ArrayList<>(); + for (Object[] r : result) + { + traces.add(TraceEntry.fromRowResultObjects(r)); + } + return traces; + } + + // Set up the wait for tracing time system property, returning the previous value. + // Handles being called again to reset with the original value, replacing the null + // with the default value. + public static String setWaitForTracingEventTimeoutSecs(String timeoutInSeconds) + { + return System.setProperty("cassandra.wait_for_tracing_events_timeout_secs", + timeoutInSeconds == null ? "0" : timeoutInSeconds); + + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java index 547e41cd08..8cd731de8b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java @@ -80,7 +80,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1")); assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", - ConsistencyLevel.QUORUM), + ConsistencyLevel.ALL), // ensure node3 in preflist row(1, 1, 1)); // Verify that data got repaired to the third node diff --git a/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java b/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java new file mode 100644 index 0000000000..72928e4166 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.Future; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.impl.IsolatedExecutor; +import org.apache.cassandra.distributed.impl.TracingUtil; +import org.apache.cassandra.utils.UUIDGen; + +public class MessageForwardingTest extends DistributedTestBase +{ + @Test + public void mutationsForwardedToAllReplicasTest() + { + String originalTraceTimeout = TracingUtil.setWaitForTracingEventTimeoutSecs("1"); + final int numInserts = 100; + Map commitCounts = new HashMap<>(); + + try (Cluster cluster = init(Cluster.build() + .withDC("dc0", 1) + .withDC("dc1", 3) + .start())) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v text, PRIMARY KEY (pk, ck))"); + + cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0)); + final UUID sessionId = UUIDGen.getTimeUUID(); + Stream> inserts = IntStream.range(0, numInserts).mapToObj((idx) -> + cluster.coordinator(1).asyncExecuteWithTracing(sessionId, + "INSERT INTO " + KEYSPACE + ".tbl(pk,ck,v) VALUES (1, 1, 'x')", + ConsistencyLevel.ALL) + ); + + // Wait for each of the futures to complete before checking the traces, don't care + // about the result so + //noinspection ResultOfMethodCallIgnored + inserts.map(IsolatedExecutor::waitOn).count(); + + cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0)); + List traces = TracingUtil.getTrace(cluster, sessionId, ConsistencyLevel.ALL); + traces.forEach(traceEntry -> { + if (traceEntry.activity.contains("Appending to commitlog")) + { + commitCounts.compute(traceEntry.source, (k, v) -> (v != null ? v : 0) + 1); + } + }); + + // Check that each node received the forwarded messages once (and only once) + commitCounts.forEach((source, count) -> + Assert.assertEquals(source + " appending to commitlog traces", + (long) numInserts, (long) count)); + } + catch (IOException e) + { + Assert.fail("Threw exception: " + e); + } + finally + { + TracingUtil.setWaitForTracingEventTimeoutSecs(originalTraceTimeout); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java new file mode 100644 index 0000000000..2c4f9c8904 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.util.Collections; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.utils.Pair; + +public class NetworkTopologyTest extends DistributedTestBase +{ + @Test + public void namedDcTest() throws Throwable + { + try (Cluster cluster = Cluster.build() + .withNodeIdTopology(Collections.singletonMap(1, Pair.create("somewhere", "rack0"))) + .withRack("elsewhere", "firstrack", 1) + .withRack("elsewhere", "secondrack", 2) + .withDC("nearthere", 4) + .start()) + { + Assert.assertEquals(1, cluster.stream("somewhere").count()); + Assert.assertEquals(1, cluster.stream("elsewhere", "firstrack").count()); + Assert.assertEquals(2, cluster.stream("elsewhere", "secondrack").count()); + Assert.assertEquals(3, cluster.stream("elsewhere").count()); + Assert.assertEquals(4, cluster.stream("nearthere").count()); + + Set expect = cluster.stream().collect(Collectors.toSet()); + Set result = Stream.concat(Stream.concat(cluster.stream("somewhere"), + cluster.stream("elsewhere")), + cluster.stream("nearthere")).collect(Collectors.toSet()); + Assert.assertEquals(expect, result); + } + } + + @Test + public void automaticNamedDcTest() throws Throwable + + { + try (Cluster cluster = Cluster.build() + .withRacks(2, 1, 3) + .start()) + { + Assert.assertEquals(6, cluster.stream().count()); + Assert.assertEquals(3, cluster.stream("datacenter1").count()); + Assert.assertEquals(3, cluster.stream("datacenter2", "rack1").count()); + } + } + + @Test(expected = IllegalStateException.class) + public void noCountsAfterNamingDCsTest() + { + Cluster.build() + .withDC("nameddc", 1) + .withDCs(1); + } + + @Test(expected = IllegalStateException.class) + public void mustProvideNodeCountBeforeWithDCsTest() + { + Cluster.build() + .withDCs(1); + } + + @Test(expected = IllegalStateException.class) + public void noEmptyNodeIdTopologyTest() + { + Cluster.build().withNodeIdTopology(Collections.emptyMap()); + } + + @Test(expected = IllegalStateException.class) + public void noHolesInNodeIdTopologyTest() + { + Cluster.build().withNodeIdTopology(Collections.singletonMap(2, Pair.create("doomed", "rack"))); + } +}