diff --git a/conf/datacenters.properties b/conf/datacenters.properties new file mode 100644 index 0000000000..b186cb1174 --- /dev/null +++ b/conf/datacenters.properties @@ -0,0 +1,20 @@ +# 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. + +# datacenter=replication factor +dc1=3 +dc2=5 +dc3=1 diff --git a/src/java/org/apache/cassandra/locator/DatacenterEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DatacenterEndpointSnitch.java deleted file mode 100644 index 6e507cc3a5..0000000000 --- a/src/java/org/apache/cassandra/locator/DatacenterEndpointSnitch.java +++ /dev/null @@ -1,173 +0,0 @@ -package org.apache.cassandra.locator; -/* - * - * 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. - * - */ - - -import java.io.IOException; -import java.net.InetAddress; -import java.net.UnknownHostException; -import java.util.*; -import javax.xml.parsers.ParserConfigurationException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.utils.XMLUtils; -import org.xml.sax.SAXException; - -/** - * DataCenterEndpointSnitch - *

- * This class basically reads the configuration and sets the IP Ranges to a - * hashMap which can be read later. this class also provides a way to compare 2 - * Endpoints and also get details from the same. - */ - -public class DatacenterEndpointSnitch extends AbstractRackAwareSnitch -{ - /** - * This Map will contain the information of the Endpoints and its Location - * (Datacenter and RAC) - */ - private Map> ipDC = new HashMap>(); - private Map> ipRAC = new HashMap>(); - private Map dcRepFactor = new HashMap(); - - private XMLUtils xmlUtils; - private Map quorumDCMap = new HashMap(); - /** - * The default rack property file to be read. - */ - private static String DEFAULT_RACK_CONFIG_FILE = "/etc/cassandra/DC-Config.xml"; - - /** - * Reference to the logger. - */ - private static Logger logger_ = LoggerFactory.getLogger(DatacenterEndpointSnitch.class); - - /** - * Constructor, intialize XML config and read the config in... - */ - public DatacenterEndpointSnitch() throws IOException, ParserConfigurationException, SAXException - { - xmlUtils = new XMLUtils(DEFAULT_RACK_CONFIG_FILE); - reloadConfiguration(); - } - - /** - * Return the rack for which an endpoint resides in - */ - public String getRack(InetAddress endpoint) throws UnknownHostException - { - byte[] ip = getIPAddress(endpoint.getHostAddress()); - return ipRAC.get(ip[1]).get(ip[2]); - } - - /** - * Return the datacenter for which an endpoint resides in - */ - public String getDatacenter(InetAddress endpoint) throws UnknownHostException - { - byte[] ip = getIPAddress(endpoint.getHostAddress()); - return ipDC.get(ip[1]).get(ip[2]); - } - - /** - * This method will load the configuration from the xml file. Mandatory - * fields are Atleast 1 DC and 1RAC configurations. Name of the DC/RAC, IP - * Quadrents for RAC and DC. - *

- * This method will not be called everytime - */ - public void reloadConfiguration() throws IOException - { - try - { - String[] dcNames = xmlUtils.getNodeValues("/Endpoints/DataCenter/name"); - for (String dcName : dcNames) - { - // Parse the Datacenter Quaderant. - String dcXPath = "/Endpoints/DataCenter[name='" + dcName + "']"; - String dcIPQuad = xmlUtils.getNodeValue(dcXPath + "/ip2ndQuad"); - String replicationFactor = xmlUtils.getNodeValue(dcXPath + "/replicationFactor"); - byte dcByte = intToByte(Integer.parseInt(dcIPQuad)); - // Parse the replication factor for a DC - int dcReF = Integer.parseInt(replicationFactor); - dcRepFactor.put(dcName, dcReF); - quorumDCMap.put(dcName, (dcReF / 2 + 1)); - String[] racNames = xmlUtils.getNodeValues(dcXPath + "/rack/name"); - Map dcRackMap = ipDC.get(dcByte); - if (null == dcRackMap) - { - dcRackMap = new HashMap(); - } - Map rackDcMap = ipRAC.get(dcByte); - if (null == rackDcMap) - { - rackDcMap = new HashMap(); - } - for (String racName : racNames) - { - // Parse the RAC ip Quaderant. - String racIPQuad = xmlUtils.getNodeValue(dcXPath + "/rack[name = '" + racName + "']/ip3rdQuad"); - byte racByte = intToByte(Integer.parseInt(racIPQuad)); - dcRackMap.put(racByte, dcName); - rackDcMap.put(racByte, racName); - } - ipDC.put(dcByte, dcRackMap); - ipRAC.put(dcByte, rackDcMap); - } - } - catch (Exception ioe) - { - throw new IOException("Could not process " + DEFAULT_RACK_CONFIG_FILE, ioe); - } - } - - /** - * Returns a DC replication map, the key will be the dc name and the value - * will be the replication factor of that Datacenter. - */ - public HashMap getMapReplicationFactor() - { - return new HashMap(dcRepFactor); - } - - /** - * Returns a DC replication map, the key will be the dc name and the value - * will be the replication factor of that Datacenter. - */ - public HashMap getMapQuorumFactor() - { - return new HashMap(quorumDCMap); - } - - private byte[] getIPAddress(String host) throws UnknownHostException - { - InetAddress ia = InetAddress.getByName(host); - return ia.getAddress(); - } - - public static byte intToByte(int n) - { - return (byte) (n & 0x000000ff); - } -} diff --git a/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java b/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java index 8f1009ad67..321f45b7b4 100644 --- a/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java +++ b/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java @@ -21,13 +21,16 @@ package org.apache.cassandra.locator; */ +import java.io.FileReader; import java.io.IOException; import java.io.IOError; import java.net.InetAddress; +import java.net.URL; import java.net.UnknownHostException; import java.util.*; import java.util.Map.Entry; +import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.*; import org.apache.cassandra.thrift.ConsistencyLevel; @@ -50,6 +53,7 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy ArrayList tokens; private List localEndpoints = new ArrayList(); + private static final String DATACENTER_PROPERTIES_FILENAME = "datacenters.properties"; private List getLocalEndpoints() { @@ -65,15 +69,15 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy * This Method will get the required information of the Endpoint from the * DataCenterEndpointSnitch and poopulates this singleton class. */ - private synchronized void loadEndpoints(TokenMetadata metadata) throws IOException + private synchronized void loadEndpoints(TokenMetadata metadata) throws UnknownHostException { this.tokens = new ArrayList(metadata.sortedTokens()); - String localDC = ((DatacenterEndpointSnitch)snitch_).getDatacenter(InetAddress.getLocalHost()); + String localDC = ((AbstractRackAwareSnitch)snitch_).getDatacenter(InetAddress.getLocalHost()); dcMap = new HashMap>(); for (Token token : this.tokens) { InetAddress endpoint = metadata.getEndpoint(token); - String dataCenter = ((DatacenterEndpointSnitch)snitch_).getDatacenter(endpoint); + String dataCenter = ((AbstractRackAwareSnitch)snitch_).getDatacenter(endpoint); if (dataCenter.equals(localDC)) { localEndpoints.add(endpoint); @@ -92,7 +96,6 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy Collections.sort(valueList); dcMap.put(entry.getKey(), valueList); } - dcReplicationFactor = ((DatacenterEndpointSnitch)snitch_).getMapReplicationFactor(); for (Entry entry : dcReplicationFactor.entrySet()) { String datacenter = entry.getKey(); @@ -105,13 +108,33 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy } } - public DatacenterShardStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) - throws UnknownHostException + public DatacenterShardStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) throws ConfigurationException { super(tokenMetadata, snitch); - if ((!(snitch instanceof DatacenterEndpointSnitch))) + if (!(snitch instanceof AbstractRackAwareSnitch)) { - throw new IllegalArgumentException("DatacenterShardStrategy requires DatacenterEndpointSnitch"); + throw new IllegalArgumentException("DatacenterShardStrategy requires a rack-aware endpointsnitch"); + } + + // load replication factors for each DC + ClassLoader loader = PropertyFileSnitch.class.getClassLoader(); + URL scpurl = loader.getResource(DATACENTER_PROPERTIES_FILENAME); + if (scpurl == null) + throw new ConfigurationException("unable to locate " + DATACENTER_PROPERTIES_FILENAME); + + String rackPropertyFilename = scpurl.getFile(); + try + { + Properties p = new Properties(); + p.load(new FileReader(rackPropertyFilename)); + for (Entry entry : p.entrySet()) + { + dcReplicationFactor.put((String)entry.getKey(), Integer.valueOf((String)entry.getValue())); + } + } + catch (IOException ioe) + { + throw new ConfigurationException("Could not process " + rackPropertyFilename, ioe); } } @@ -224,4 +247,9 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy } return super.getWriteResponseHandler(blockFor, consistency_level, table); } + + int getReplicationFactor(String datacenter) + { + return dcReplicationFactor.get(datacenter); + } } diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java index cc11877fbc..44c5a5a30c 100644 --- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java @@ -29,7 +29,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.locator.DatacenterEndpointSnitch; +import org.apache.cassandra.locator.AbstractRackAwareSnitch; import org.apache.cassandra.net.Message; /** @@ -41,14 +41,14 @@ public class DatacenterSyncWriteResponseHandler extends WriteResponseHandler { private final Map dcResponses = new HashMap(); private final Map responseCounts; - private final DatacenterEndpointSnitch endpointSnitch; + private final AbstractRackAwareSnitch endpointSnitch; public DatacenterSyncWriteResponseHandler(Map responseCounts, String table) { // Response is been managed by the map so make it 1 for the superclass. super(1, table); this.responseCounts = responseCounts; - endpointSnitch = (DatacenterEndpointSnitch) DatabaseDescriptor.getEndpointSnitch(); + endpointSnitch = (AbstractRackAwareSnitch) DatabaseDescriptor.getEndpointSnitch(); } @Override diff --git a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java index 8fb7749876..3256483e17 100644 --- a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java @@ -29,7 +29,7 @@ import java.net.UnknownHostException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.locator.DatacenterEndpointSnitch; +import org.apache.cassandra.locator.AbstractRackAwareSnitch; import org.apache.cassandra.net.Message; import org.apache.cassandra.utils.FBUtilities; @@ -41,7 +41,7 @@ import org.apache.cassandra.utils.FBUtilities; public class DatacenterWriteResponseHandler extends WriteResponseHandler { private final AtomicInteger blockFor; - private final DatacenterEndpointSnitch endpointsnitch; + private final AbstractRackAwareSnitch endpointsnitch; private final InetAddress localEndpoint; public DatacenterWriteResponseHandler(int blockFor, String table) @@ -49,7 +49,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler // Response is been managed by the map so the waitlist size really doesnt matter. super(blockFor, table); this.blockFor = new AtomicInteger(blockFor); - endpointsnitch = (DatacenterEndpointSnitch) DatabaseDescriptor.getEndpointSnitch(); + endpointsnitch = (AbstractRackAwareSnitch) DatabaseDescriptor.getEndpointSnitch(); localEndpoint = FBUtilities.getLocalAddress(); } diff --git a/test/conf/datacenters.properties b/test/conf/datacenters.properties new file mode 100644 index 0000000000..b186cb1174 --- /dev/null +++ b/test/conf/datacenters.properties @@ -0,0 +1,20 @@ +# 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. + +# datacenter=replication factor +dc1=3 +dc2=5 +dc3=1 diff --git a/test/unit/org/apache/cassandra/locator/DatacenterStrategyTest.java b/test/unit/org/apache/cassandra/locator/DatacenterStrategyTest.java new file mode 100644 index 0000000000..c53c2f622b --- /dev/null +++ b/test/unit/org/apache/cassandra/locator/DatacenterStrategyTest.java @@ -0,0 +1,18 @@ +package org.apache.cassandra.locator; + +import org.junit.Test; + +import org.apache.cassandra.config.ConfigurationException; + +public class DatacenterStrategyTest +{ + @Test + public void testProperties() throws ConfigurationException + { + DatacenterShardStrategy strategy = new DatacenterShardStrategy(new TokenMetadata(), new RackInferringSnitch()); + assert strategy.getReplicationFactor("dc1") == 3; + assert strategy.getReplicationFactor("dc2") == 5; + assert strategy.getReplicationFactor("dc3") == 1; + } + +}