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

move per-datacenter replication factor into datacenters.properties 

patch by jbellis; reviewed by eevans for CASSANDRA-994


git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@938568 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-04-27 17:06:04 +00:00
parent 40df0ce87e
commit 2acfb43599
7 changed files with 100 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
conf/datacenters.properties Normal file
View File

@ -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

173
src/java/org/apache/cassandra/locator/DatacenterEndpointSnitch.java
View File

@ -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
* <p/>
* 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<Byte, Map<Byte, String>> ipDC = new HashMap<Byte, Map<Byte, String>>();
private Map<Byte, Map<Byte, String>> ipRAC = new HashMap<Byte, Map<Byte, String>>();
private Map<String, Integer> dcRepFactor = new HashMap<String, Integer>();
private XMLUtils xmlUtils;
private Map<String, Integer> quorumDCMap = new HashMap<String, Integer>();
/**
* 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.
* <p/>
* 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<Byte, String> dcRackMap = ipDC.get(dcByte);
if (null == dcRackMap)
{
dcRackMap = new HashMap<Byte, String>();
}
Map<Byte, String> rackDcMap = ipRAC.get(dcByte);
if (null == rackDcMap)
{
rackDcMap = new HashMap<Byte, String>();
}
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<String, Integer> getMapReplicationFactor()
{
return new HashMap<String, Integer>(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<String, Integer> getMapQuorumFactor()
{
return new HashMap<String, Integer>(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);
}
}

44
src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java
View File

@ -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<Token> tokens;
private List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
private static final String DATACENTER_PROPERTIES_FILENAME = "datacenters.properties";
private List<InetAddress> 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<Token>(metadata.sortedTokens());
String localDC = ((DatacenterEndpointSnitch)snitch_).getDatacenter(InetAddress.getLocalHost());
String localDC = ((AbstractRackAwareSnitch)snitch_).getDatacenter(InetAddress.getLocalHost());
dcMap = new HashMap<String, List<Token>>();
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<String, Integer> 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<Object, Object> 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);
}
}

6
src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
View File

@ -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<String, Integer> dcResponses = new HashMap<String, Integer>();
private final Map<String, Integer> responseCounts;
private final DatacenterEndpointSnitch endpointSnitch;
private final AbstractRackAwareSnitch endpointSnitch;
public DatacenterSyncWriteResponseHandler(Map<String, Integer> 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

6
src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java
View File

@ -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();
}

20
test/conf/datacenters.properties Normal file
View File

@ -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

18
test/unit/org/apache/cassandra/locator/DatacenterStrategyTest.java Normal file
View File

@ -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;
}
}