diff --git a/CHANGES.txt b/CHANGES.txt index 925a90a720..5a2460f589 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -39,6 +39,7 @@ * Fixing invalid CQL in security documentation (CASSANDRA-15020) * 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) 3.0.18 diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 8c29601149..5ebdf897b1 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -987,7 +987,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.metadata()); + ensureTraceKeyspace(); maybeAddOrUpdateKeyspace(SystemDistributedKeyspace.metadata()); if (!isSurveyMode) @@ -1019,6 +1019,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } + @VisibleForTesting + public void ensureTraceKeyspace() + { + maybeAddOrUpdateKeyspace(TraceKeyspace.metadata()); + } + 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 c77d72514b..1fe960a253 100644 --- a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java @@ -46,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..effa4296e9 100644 --- a/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java @@ -18,8 +18,11 @@ package org.apache.cassandra.distributed.api; +import org.apache.cassandra.distributed.impl.NetworkTopology; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.utils.Pair; +import java.util.Map; import java.util.UUID; public interface IInstanceConfig @@ -27,6 +30,17 @@ 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 d09ed9d819..32006610d6 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) @@ -309,7 +326,7 @@ public abstract class AbstractCluster implements ICluster, private void signal() { - if (schemaHasChanged && 1 == instances.stream().map(IInstance::schemaVersion).distinct().count()) + if (schemaHasChanged && 1 == instances.stream().filter(i -> !i.isShutdown()).map(IInstance::schemaVersion).distinct().count()) agreement.signalAll(); } @@ -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 c27523f239..69c2e44a69 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -23,6 +23,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; @@ -35,6 +37,7 @@ import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.pager.QueryPager; 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; import org.apache.cassandra.utils.FBUtilities; @@ -50,36 +53,55 @@ public class Coordinator implements ICoordinator @Override public Object[][] execute(String query, Enum consistencyLevelOrigin, Object... boundValues) { - return instance.sync(() -> { - ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); - ClientState clientState = makeFakeClientState(); - CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement; - List boundBBValues = new ArrayList<>(); - for (Object boundValue : boundValues) - { - boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); - } + return instance.sync(() -> executeInternal(query, consistencyLevelOrigin, boundValues)).call(); + } - 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) + { + ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name()); + ClientState clientState = makeFakeClientState(); + CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement; + List boundBBValues = new ArrayList<>(); + for (Object boundValue : boundValues) + { + boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); + } + + 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 aea21e2790..1530f9eb34 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.impl; 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; @@ -70,7 +71,6 @@ import org.apache.cassandra.io.util.DataInputBuffer; 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; @@ -80,10 +80,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 org.apache.cassandra.utils.memory.BufferPool; @@ -243,6 +246,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(); @@ -327,6 +352,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { mkdirs(); + assert config.networkTopology().contains(config.broadcastAddressAndPort()); + DistributedTestSnitch.assign(config.networkTopology()); + DatabaseDescriptor.setDaemonInitialized(); DatabaseDescriptor.createAllDirectories(); @@ -379,6 +407,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance initializeRing(cluster); } + StorageService.instance.ensureTraceKeyspace(); + SystemKeyspace.finishStartup(); if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address)) @@ -449,8 +479,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token))); 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 0ef5a69113..ff3aced3d8 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java @@ -37,7 +37,8 @@ public class InstanceClassLoader extends URLClassLoader Pair.class, InetAddressAndPort.class, ParameterizedClass.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 992d6b8e6c..67166565b3 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 238e9e7efa..ac5028611c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java @@ -85,7 +85,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"))); + } +}