natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support Dynamic Port Allocation for in-jvm dtest framework 

patch by Francisco Guerrero; reviewed by Dinesh Joshi, Jon Meredith, Yifan Cai for CASSANDRA-18722
This commit is contained in:
Francisco Guerrero 2023-08-25 11:10:48 -06:00 committed by Jon Meredith
parent 4034fbb6dd
commit 6ffa43f68b
5 changed files with 142 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -140,6 +140,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private final Map<Integer, NetworkTopology.DcAndRack> nodeIdTopology;
private final Consumer<IInstanceConfig> configUpdater;
private final int broadcastPort;
private final Map<String, Integer> portMap;
// mutated by starting/stopping a node
private final List<I> instances;
@ -163,6 +164,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
extends org.apache.cassandra.distributed.shared.AbstractBuilder<I, C, B>
{
private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces;
private boolean dynamicPortAllocation = false;
{
// Indicate that we are running in the in-jvm dtest environment
@ -177,10 +179,32 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
super(factory);
}
@SuppressWarnings("unchecked")
private B self()
{
return (B) this;
}
public B withNodeProvisionStrategy(INodeProvisionStrategy.Strategy nodeProvisionStrategy)
{
this.nodeProvisionStrategy = nodeProvisionStrategy;
return (B) this;
return self();
}
/**
* When {@code dynamicPortAllocation} is {@code true}, it will ask {@link INodeProvisionStrategy} to provision
* available storage, native and JMX ports in the given interface. When {@code dynamicPortAllocation} is
* {@code false} (the default behavior), it will use statically allocated ports based on the number of
* interfaces available and the node number.
*
* @param dynamicPortAllocation {@code true} for dynamic port allocation, {@code false} for static port
* allocation
* @return a reference to this Builder
*/
public B withDynamicPortAllocation(boolean dynamicPortAllocation)
{
this.dynamicPortAllocation = dynamicPortAllocation;
return self();
}
}
@ -408,6 +432,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
this.filters = new MessageFilters();
this.instanceInitializer = builder.getInstanceInitializer();
this.datadirCount = builder.getDatadirCount();
this.portMap = builder.dynamicPortAllocation ? new ConcurrentHashMap<>() : null;
int generation = GENERATION.incrementAndGet();
for (int i = 0; i < builder.getNodeCount(); ++i)
@ -431,7 +456,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private InstanceConfig createInstanceConfig(int nodeNum)
{
INodeProvisionStrategy provisionStrategy = nodeProvisionStrategy.create(subnet);
INodeProvisionStrategy provisionStrategy = nodeProvisionStrategy.create(subnet, portMap);
long token = tokenSupplier.token(nodeNum);
NetworkTopology topology = buildNetworkTopology(provisionStrategy, nodeIdTopology);
InstanceConfig config = InstanceConfig.generate(nodeNum, provisionStrategy, topology, root, Long.toString(token), datadirCount);

91
test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java
View File

@ -18,43 +18,68 @@
package org.apache.cassandra.distributed.impl;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.cassandra.net.SocketUtils;
public interface INodeProvisionStrategy
{
public enum Strategy
enum Strategy
{
OneNetworkInterface
{
INodeProvisionStrategy create(int subnet) {
@Override
INodeProvisionStrategy create(int subnet, @Nullable Map<String, Integer> portMap)
{
String ipAdress = "127.0." + subnet + ".1";
return new INodeProvisionStrategy()
{
@Override
public String seedIp()
{
return "127.0." + subnet + ".1";
return ipAdress;
}
@Override
public int seedPort()
{
return 7012;
return storagePort(1);
}
@Override
public String ipAddress(int nodeNum)
{
return "127.0." + subnet + ".1";
return ipAdress;
}
@Override
public int storagePort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("storagePort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 7011 + nodeNum));
}
return 7011 + nodeNum;
}
@Override
public int nativeTransportPort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("nativeTransportPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 9041 + nodeNum));
}
return 9041 + nodeNum;
}
@Override
public int jmxPort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("jmxPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 7199 + nodeNum));
}
return 7199 + nodeNum;
}
};
@ -62,49 +87,81 @@ public interface INodeProvisionStrategy
},
MultipleNetworkInterfaces
{
INodeProvisionStrategy create(int subnet) {
String ipPrefix = "127.0." + subnet + ".";
@Override
INodeProvisionStrategy create(int subnet, @Nullable Map<String, Integer> portMap)
{
String ipPrefix = "127.0." + subnet + '.';
return new INodeProvisionStrategy()
{
@Override
public String seedIp()
{
return ipPrefix + "1";
return ipPrefix + '1';
}
@Override
public int seedPort()
{
return 7012;
return storagePort(1);
}
@Override
public String ipAddress(int nodeNum)
{
return ipPrefix + nodeNum;
}
@Override
public int storagePort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("storagePort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 7012));
}
return 7012;
}
@Override
public int nativeTransportPort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("nativeTransportPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 9042));
}
return 9042;
}
@Override
public int jmxPort(int nodeNum)
{
if (portMap != null)
{
return portMap.computeIfAbsent("jmxPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 7199));
}
return 7199;
}
};
}
};
abstract INodeProvisionStrategy create(int subnet);
INodeProvisionStrategy create(int subnet)
{
return create(subnet, null);
}
abstract INodeProvisionStrategy create(int subnet, @Nullable Map<String, Integer> portMap);
}
abstract String seedIp();
abstract int seedPort();
abstract String ipAddress(int nodeNum);
abstract int storagePort(int nodeNum);
abstract int nativeTransportPort(int nodeNum);
abstract int jmxPort(int nodeNum);
String seedIp();
int seedPort();
String ipAddress(int nodeNum);
int storagePort(int nodeNum);
int nativeTransportPort(int nodeNum);
int jmxPort(int nodeNum);
}

2
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
View File

@ -110,7 +110,7 @@ public class InstanceConfig implements IInstanceConfig
.set("native_transport_port", native_transport_port)
.set("endpoint_snitch", DistributedTestSnitch.class.getName())
.set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(),
Collections.singletonMap("seeds", seedIp + ":" + seedPort)))
Collections.singletonMap("seeds", seedIp + ':' + seedPort)))
// required settings for dtest functionality
.set("diagnostic_events_enabled", true)
.set("auto_bootstrap", false)

3
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java
View File

@ -72,6 +72,7 @@ public class JMXFeatureTest extends TestBaseImpl
for (int i = 0; i < iterations; i++)
{
try (Cluster cluster = Cluster.build(2)
.withDynamicPortAllocation(true)
.withNodeProvisionStrategy(provisionStrategy)
.withConfig(c -> c.with(Feature.values())).start())
{
@ -136,7 +137,7 @@ public class JMXFeatureTest extends TestBaseImpl
// to check that we are actually connecting to the correct instance
String defaultDomain = mbsc.getDefaultDomain();
instancesContacted.add(defaultDomain);
Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort()));
Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ':' + config.jmxPort()));
}
catch (Throwable t)
{

61
test/unit/org/apache/cassandra/net/SocketUtils.java
View File

@ -19,39 +19,54 @@
package org.apache.cassandra.net;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.UnknownHostException;
import com.google.common.base.Throwables;
public class SocketUtils
{
public static synchronized int findAvailablePort() throws RuntimeException
/**
* Returns an available port for the given {@code bindAddress}. When an {@link IOException} occurs when opening a
* socket or if a {@link SecurityException} is raised because a manager exists and its checkListen method does
* not allow the operation, the {@code fallbackPort} is returned.
*
* @param bindAddress the ip address for the interface where we need an available port number
* @param fallbackPort a port to return in case {@link SecurityException} or {@link IOException} is encountered
* @return an available port the given {@code bindAddress} when succeeds, otherwise the {@code fallbackPort}
* @throws RuntimeException if no IP address for the {@code bindAddress} could be found
*/
public static synchronized int findAvailablePort(String bindAddress, int fallbackPort) throws RuntimeException
{
ServerSocket ss = null;
try
{
// let the system pick an ephemeral port
ss = new ServerSocket(0);
ss.setReuseAddress(true);
return ss.getLocalPort();
return findAvailablePort(InetAddress.getByName(bindAddress), fallbackPort);
}
catch (IOException e)
catch (UnknownHostException e)
{
throw Throwables.propagate(e);
}
finally
{
if (ss != null)
{
try
{
ss.close();
}
catch (IOException e)
{
Throwables.propagate(e);
}
}
throw new RuntimeException(e);
}
}
}
/**
* Returns an available port for the given {@code bindAddress}. When an {@link IOException} occurs when opening a
* socket or if a {@link SecurityException} is raised because a manager exists and its checkListen method does
* not allow the operation, the {@code fallbackPort} is returned.
*
* @param bindAddress the ip address for the interface where we need an available port number
* @param fallbackPort a port to return in case {@link SecurityException} or {@link IOException} is encountered
* @return an available port the given {@code bindAddress} when succeeds, otherwise the {@code fallbackPort}
*/
public static synchronized int findAvailablePort(InetAddress bindAddress, int fallbackPort)
{
try (ServerSocket socket = new ServerSocket(0, 50, bindAddress))
{
return socket.getLocalPort();
}
catch (SecurityException | IOException exception)
{
return fallbackPort;
}
}
}