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

Merge branch 'cassandra-2.1' into trunk

This commit is contained in:
Ariel Weisberg 2015-03-17 19:31:54 +01:00 committed by Robert Stupp
parent ac9ccfdf7c 89cdfd8e07
commit 8d570fa799
6 changed files with 274 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -71,6 +71,7 @@
* Generalize progress reporting (CASSANDRA-8901)
2.1.4
* DatabaseDescriptor throws NPE when rpc_interface is used (CASSANDRA-8839)
* Don't check if an sstable is live for offline compactions (CASSANDRA-8841)
* Don't set clientMode in SSTableLoader (CASSANDRA-8238)
* Fix SSTableRewriter with disabled early open (CASSANDRA-8535)

12
conf/cassandra.yaml
View File

@ -408,8 +408,14 @@ ssl_storage_port: 7001
# address associated with the hostname (it might not be).
#
# Setting listen_address to 0.0.0.0 is always wrong.
#
# If you choose to specify the interface by name and the interface has an ipv4 and an ipv6 address
# you can specify which should be chosen using listen_interface_prefer_ipv6. If false the first ipv4
# address will be used. If true the first ipv6 address will be used. Defaults to false preferring
# ipv4. If there is only one address it will be selected regardless of ipv4/ipv6.
listen_address: localhost
# listen_interface: eth0
# listen_interface_prefer_ipv6: false
# Address to broadcast to other Cassandra nodes
# Leaving this blank will set it to the same value as listen_address
@ -460,8 +466,14 @@ start_rpc: true
# set broadcast_rpc_address to a value other than 0.0.0.0.
#
# For security reasons, you should not expose this port to the internet. Firewall it if needed.
#
# If you choose to specify the interface by name and the interface has an ipv4 and an ipv6 address
# you can specify which should be chosen using rpc_interface_prefer_ipv6. If false the first ipv4
# address will be used. If true the first ipv6 address will be used. Defaults to false preferring
# ipv4. If there is only one address it will be selected regardless of ipv4/ipv6.
rpc_address: localhost
# rpc_interface: eth1
# rpc_interface_prefer_ipv6: false
# port for Thrift to listen for clients on
rpc_port: 9160

6
src/java/org/apache/cassandra/config/Config.java
View File

@ -33,7 +33,7 @@ import org.supercsv.prefs.CsvPreference;
/**
* A class that contains configuration properties for the cassandra node it runs within.
*
*
* Properties declared as volatile can be mutated via JMX.
*/
public class Config
@ -104,12 +104,14 @@ public class Config
public Integer ssl_storage_port = 7001;
public String listen_address;
public String listen_interface;
public Boolean listen_interface_prefer_ipv6 = false;
public String broadcast_address;
public String internode_authenticator;
public Boolean start_rpc = true;
public String rpc_address;
public String rpc_interface;
public Boolean rpc_interface_prefer_ipv6 = false;
public String broadcast_rpc_address;
public Integer rpc_port = 9160;
public Integer rpc_listen_backlog = 50;
@ -159,7 +161,7 @@ public class Config
public Double commitlog_sync_batch_window_in_ms;
public Integer commitlog_sync_period_in_ms;
public int commitlog_segment_size_in_mb = 32;
@Deprecated
public int commitlog_periodic_queue_size = -1;

202
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -125,7 +125,7 @@ public class DatabaseDescriptor
return loader.loadConfig();
}
private static InetAddress getNetworkInterfaceAddress(String intf, String configName) throws ConfigurationException
private static InetAddress getNetworkInterfaceAddress(String intf, String configName, boolean preferIPv6) throws ConfigurationException
{
try
{
@ -135,9 +135,18 @@ public class DatabaseDescriptor
Enumeration<InetAddress> addrs = ni.getInetAddresses();
if (!addrs.hasMoreElements())
throw new ConfigurationException("Configured " + configName + " \"" + intf + "\" was found, but had no addresses", false);
InetAddress retval = listenAddress = addrs.nextElement();
if (addrs.hasMoreElements())
throw new ConfigurationException("Configured " + configName + " \"" + intf + "\" can't have more than one address", false);
/*
* Try to return the first address of the preferred type, otherwise return the first address
*/
InetAddress retval = null;
while (addrs.hasMoreElements())
{
InetAddress temp = addrs.nextElement();
if (preferIPv6 && temp instanceof Inet6Address) return temp;
if (!preferIPv6 && temp instanceof Inet4Address) return temp;
if (retval == null) retval = temp;
}
return retval;
}
catch (SocketException e)
@ -146,6 +155,103 @@ public class DatabaseDescriptor
}
}
@VisibleForTesting
static void applyAddressConfig(Config config) throws ConfigurationException
{
listenAddress = null;
rpcAddress = null;
broadcastAddress = null;
broadcastRpcAddress = null;
/* Local IP, hostname or interface to bind services to */
if (config.listen_address != null && config.listen_interface != null)
{
throw new ConfigurationException("Set listen_address OR listen_interface, not both", false);
}
else if (config.listen_address != null)
{
try
{
listenAddress = InetAddress.getByName(config.listen_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown listen_address '" + config.listen_address + "'", false);
}
if (listenAddress.isAnyLocalAddress())
throw new ConfigurationException("listen_address cannot be a wildcard address (" + config.listen_address + ")!", false);
}
else if (config.listen_interface != null)
{
listenAddress = getNetworkInterfaceAddress(config.listen_interface, "listen_interface", config.listen_interface_prefer_ipv6);
}
/* Gossip Address to broadcast */
if (config.broadcast_address != null)
{
try
{
broadcastAddress = InetAddress.getByName(config.broadcast_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown broadcast_address '" + config.broadcast_address + "'", false);
}
if (broadcastAddress.isAnyLocalAddress())
throw new ConfigurationException("broadcast_address cannot be a wildcard address (" + config.broadcast_address + ")!", false);
}
/* Local IP, hostname or interface to bind RPC server to */
if (config.rpc_address != null && config.rpc_interface != null)
{
throw new ConfigurationException("Set rpc_address OR rpc_interface, not both", false);
}
else if (config.rpc_address != null)
{
try
{
rpcAddress = InetAddress.getByName(config.rpc_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown host in rpc_address " + config.rpc_address, false);
}
}
else if (config.rpc_interface != null)
{
rpcAddress = getNetworkInterfaceAddress(config.rpc_interface, "rpc_interface", config.rpc_interface_prefer_ipv6);
}
else
{
rpcAddress = FBUtilities.getLocalAddress();
}
/* RPC address to broadcast */
if (config.broadcast_rpc_address != null)
{
try
{
broadcastRpcAddress = InetAddress.getByName(config.broadcast_rpc_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown broadcast_rpc_address '" + config.broadcast_rpc_address + "'", false);
}
if (broadcastRpcAddress.isAnyLocalAddress())
throw new ConfigurationException("broadcast_rpc_address cannot be a wildcard address (" + config.broadcast_rpc_address + ")!", false);
}
else
{
if (rpcAddress.isAnyLocalAddress())
throw new ConfigurationException("If rpc_address is set to a wildcard address (" + config.rpc_address + "), then " +
"you must set broadcast_rpc_address to a value other than " + config.rpc_address, false);
broadcastRpcAddress = rpcAddress;
}
}
private static void applyConfig(Config config) throws ConfigurationException
{
conf = config;
@ -300,93 +406,7 @@ public class DatabaseDescriptor
else
logger.info("Global memtable off-heap threshold is enabled at {}MB", conf.memtable_offheap_space_in_mb);
/* Local IP, hostname or interface to bind services to */
if (conf.listen_address != null && conf.listen_interface != null)
{
throw new ConfigurationException("Set listen_address OR listen_interface, not both", false);
}
else if (conf.listen_address != null)
{
try
{
listenAddress = InetAddress.getByName(conf.listen_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown listen_address '" + conf.listen_address + "'", false);
}
if (listenAddress.isAnyLocalAddress())
throw new ConfigurationException("listen_address cannot be a wildcard address (" + conf.listen_address + ")!", false);
}
else if (conf.listen_interface != null)
{
listenAddress = getNetworkInterfaceAddress(conf.listen_interface, "listen_interface");
}
/* Gossip Address to broadcast */
if (conf.broadcast_address != null)
{
try
{
broadcastAddress = InetAddress.getByName(conf.broadcast_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown broadcast_address '" + conf.broadcast_address + "'", false);
}
if (broadcastAddress.isAnyLocalAddress())
throw new ConfigurationException("broadcast_address cannot be a wildcard address (" + conf.broadcast_address + ")!", false);
}
/* Local IP, hostname or interface to bind RPC server to */
if (conf.rpc_address != null && conf.rpc_interface != null)
{
throw new ConfigurationException("Set rpc_address OR rpc_interface, not both", false);
}
else if (conf.rpc_address != null)
{
try
{
rpcAddress = InetAddress.getByName(conf.rpc_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown host in rpc_address " + conf.rpc_address, false);
}
}
else if (conf.rpc_interface != null)
{
rpcAddress = getNetworkInterfaceAddress(conf.rpc_interface, "rpc_interface");
}
else
{
rpcAddress = FBUtilities.getLocalAddress();
}
/* RPC address to broadcast */
if (conf.broadcast_rpc_address != null)
{
try
{
broadcastRpcAddress = InetAddress.getByName(conf.broadcast_rpc_address);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown broadcast_rpc_address '" + conf.broadcast_rpc_address + "'", false);
}
if (broadcastRpcAddress.isAnyLocalAddress())
throw new ConfigurationException("broadcast_rpc_address cannot be a wildcard address (" + conf.broadcast_rpc_address + ")!", false);
}
else
{
if (rpcAddress.isAnyLocalAddress())
throw new ConfigurationException("If rpc_address is set to a wildcard address (" + conf.rpc_address + "), then " +
"you must set broadcast_rpc_address to a value other than " + conf.rpc_address, false);
broadcastRpcAddress = rpcAddress;
}
applyAddressConfig(config);
if (conf.thrift_framed_transport_size_in_mb <= 0)
throw new ConfigurationException("thrift_framed_transport_size_in_mb must be positive", false);

16
src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
View File

@ -50,7 +50,7 @@ public class YamlConfigurationLoader implements ConfigurationLoader
/**
* Inspect the classpath to find storage configuration file
*/
private URL getStorageConfigURL() throws ConfigurationException
static URL getStorageConfigURL() throws ConfigurationException
{
String configUrl = System.getProperty("cassandra.config");
if (configUrl == null)
@ -100,7 +100,7 @@ public class YamlConfigurationLoader implements ConfigurationLoader
// getStorageConfigURL should have ruled this out
throw new AssertionError(e);
}
logConfig(configBytes);
org.yaml.snakeyaml.constructor.Constructor constructor = new org.yaml.snakeyaml.constructor.Constructor(Config.class);
@ -134,16 +134,16 @@ public class YamlConfigurationLoader implements ConfigurationLoader
}
logger.info("Node configuration:[{}]", Joiner.on("; ").join(configMap.entrySet()));
}
private static class MissingPropertiesChecker extends PropertyUtils
private static class MissingPropertiesChecker extends PropertyUtils
{
private final Set<String> missingProperties = new HashSet<>();
public MissingPropertiesChecker()
{
setSkipMissingProperties(true);
}
@Override
public Property getProperty(Class<? extends Object> type, String name) throws IntrospectionException
{
@ -154,10 +154,10 @@ public class YamlConfigurationLoader implements ConfigurationLoader
}
return result;
}
public void check() throws ConfigurationException
{
if (!missingProperties.isEmpty())
if (!missingProperties.isEmpty())
{
throw new ConfigurationException("Invalid yaml. Please remove properties " + missingProperties + " from your cassandra.yaml");
}

139
test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
View File

@ -18,9 +18,17 @@
*/
package org.apache.cassandra.config;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;
import junit.framework.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.Keyspace;
@ -128,4 +136,133 @@ public class DatabaseDescriptorTest
return testConfig;
}
}
static NetworkInterface suitableInterface = null;
static boolean hasIPv4andIPv6 = false;
/*
* Server only accepts interfaces by name if they have a single address
* OS X seems to always have an ipv4 and ipv6 address on all interfaces which means some tests fail
* if not checked for and skipped
*/
@BeforeClass
public static void selectSuitableInterface() throws Exception {
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
while(interfaces.hasMoreElements()) {
NetworkInterface intf = interfaces.nextElement();
System.out.println("Evaluating " + intf.getName());
if (intf.isLoopback()) {
suitableInterface = intf;
boolean hasIPv4 = false;
boolean hasIPv6 = false;
Enumeration<InetAddress> addresses = suitableInterface.getInetAddresses();
while (addresses.hasMoreElements()) {
if (addresses.nextElement() instanceof Inet6Address)
hasIPv6 = true;
else
hasIPv4 = true;
}
hasIPv4andIPv6 = hasIPv4 && hasIPv6;
return;
}
}
}
@Test
public void testRpcInterface() throws Exception
{
Config testConfig = DatabaseDescriptor.loadConfig();
testConfig.rpc_interface = suitableInterface.getName();
testConfig.rpc_address = null;
DatabaseDescriptor.applyAddressConfig(testConfig);
/*
* Confirm ability to select between IPv4 and IPv6
*/
if (hasIPv4andIPv6)
{
testConfig = DatabaseDescriptor.loadConfig();
testConfig.rpc_interface = suitableInterface.getName();
testConfig.rpc_address = null;
testConfig.rpc_interface_prefer_ipv6 = true;
DatabaseDescriptor.applyAddressConfig(testConfig);
assertEquals(DatabaseDescriptor.getRpcAddress().getClass(), Inet6Address.class);
testConfig = DatabaseDescriptor.loadConfig();
testConfig.rpc_interface = suitableInterface.getName();
testConfig.rpc_address = null;
testConfig.rpc_interface_prefer_ipv6 = false;
DatabaseDescriptor.applyAddressConfig(testConfig);
assertEquals(DatabaseDescriptor.getRpcAddress().getClass(), Inet4Address.class);
}
else
{
/*
* Confirm first address of interface is selected
*/
assertEquals(DatabaseDescriptor.getRpcAddress(), suitableInterface.getInetAddresses().nextElement());
}
}
@Test
public void testListenInterface() throws Exception
{
Config testConfig = DatabaseDescriptor.loadConfig();
testConfig.listen_interface = suitableInterface.getName();
testConfig.listen_address = null;
DatabaseDescriptor.applyAddressConfig(testConfig);
/*
* Confirm ability to select between IPv4 and IPv6
*/
if (hasIPv4andIPv6)
{
testConfig = DatabaseDescriptor.loadConfig();
testConfig.listen_interface = suitableInterface.getName();
testConfig.listen_address = null;
testConfig.listen_interface_prefer_ipv6 = true;
DatabaseDescriptor.applyAddressConfig(testConfig);
assertEquals(DatabaseDescriptor.getListenAddress().getClass(), Inet6Address.class);
testConfig = DatabaseDescriptor.loadConfig();
testConfig.listen_interface = suitableInterface.getName();
testConfig.listen_address = null;
testConfig.listen_interface_prefer_ipv6 = false;
DatabaseDescriptor.applyAddressConfig(testConfig);
assertEquals(DatabaseDescriptor.getListenAddress().getClass(), Inet4Address.class);
}
else
{
/*
* Confirm first address of interface is selected
*/
assertEquals(DatabaseDescriptor.getRpcAddress(), suitableInterface.getInetAddresses().nextElement());
}
}
@Test
public void testListenAddress() throws Exception
{
Config testConfig = DatabaseDescriptor.loadConfig();
testConfig.listen_address = suitableInterface.getInterfaceAddresses().get(0).getAddress().getHostAddress();
testConfig.listen_interface = null;
DatabaseDescriptor.applyAddressConfig(testConfig);
}
@Test
public void testRpcAddress() throws Exception
{
Config testConfig = DatabaseDescriptor.loadConfig();
testConfig.rpc_address = suitableInterface.getInterfaceAddresses().get(0).getAddress().getHostAddress();
testConfig.rpc_interface = null;
DatabaseDescriptor.applyAddressConfig(testConfig);
}
}