From cf9a581bf7957d9404a6a5f33f77cf32468c6336 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 25 May 2012 15:21:27 -0500 Subject: [PATCH] Add GossipingPropertyFileSnitch. Patch by brandonwilliams reviewed by Vijay for CASSANDRA-1974 --- conf/cassandra.yaml | 5 + .../locator/GossipingPropertyFileSnitch.java | 140 ++++++++++++++++++ .../cassandra/locator/PropertyFileSnitch.java | 4 +- 3 files changed, 147 insertions(+), 2 deletions(-) create mode 100644 src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 14cc525a75..d64e2c9be0 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -437,6 +437,11 @@ rpc_timeout_in_ms: 10000 # - PropertyFileSnitch: # Proximity is determined by rack and data center, which are # explicitly configured in cassandra-topology.properties. +# - GossipingPropertyFileSnitch +# The rack and datacenter for the local node are defined in +# cassandra-rackdc.properties and propagated to other nodes via gossip. If +# cassandra-topology.properties exists, it is used as a fallback, allowing +# migration from the PropertyFileSnitch. # - RackInferringSnitch: # Proximity is determined by rack and data center, which are # assumed to correspond to the 3rd and 2nd octet of each node's diff --git a/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java new file mode 100644 index 0000000000..9a22544eb8 --- /dev/null +++ b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java @@ -0,0 +1,140 @@ +/** + * 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.locator; + +import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.service.StorageService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.InputStream; +import java.net.InetAddress; +import java.util.Map; +import java.util.Properties; + +public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch +{ + private static final Logger logger = LoggerFactory.getLogger(GossipingPropertyFileSnitch.class); + + public static final String RACKDC_PROPERTY_FILENAME = "cassandra-rackdc.properties"; + private PropertyFileSnitch psnitch; + private String myDC; + private String myRack; + + public GossipingPropertyFileSnitch() throws ConfigurationException + { + try + { + loadConfiguration(); + } + catch (ConfigurationException e) + { + throw new RuntimeException("Unable to load " + RACKDC_PROPERTY_FILENAME + " : ", e); + } + try + { + psnitch = new PropertyFileSnitch(); + logger.info("Loaded " + PropertyFileSnitch.RACK_PROPERTY_FILENAME + " for compatibility"); + } + catch (ConfigurationException e) + { + logger.info("Unable to load " + PropertyFileSnitch.RACK_PROPERTY_FILENAME + "; compatibility mode disabled"); + } + } + + private void loadConfiguration() throws ConfigurationException + { + InputStream stream = GossipingPropertyFileSnitch.class.getClassLoader().getResourceAsStream(RACKDC_PROPERTY_FILENAME); + Properties properties = new Properties(); + try + { + properties.load(stream); + } + catch (Exception e) + { + throw new ConfigurationException("Unable to read " + RACKDC_PROPERTY_FILENAME, e); + } + finally + { + FileUtils.closeQuietly(stream); + } + for (Map.Entry entry : properties.entrySet()) + { + String key = (String) entry.getKey(); + String value = (String) entry.getValue(); + if (key.equals("dc")) + myDC = value; + else if (key.equals("rack")) + myRack = value; + } + if (myDC == null || myRack == null) + throw new ConfigurationException("DC or rack not found in " + RACKDC_PROPERTY_FILENAME); + } + + /** + * Return the data center for which an endpoint resides in + * + * @param endpoint the endpoint to process + * @return string of data center + */ + public String getDatacenter(InetAddress endpoint) + { + EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + if (epState == null || epState.getApplicationState(ApplicationState.DC) == null) + { + if (psnitch == null) + throw new RuntimeException("Could not retrieve DC for " + endpoint + " from gossip and PFS compatibility is disabled"); + else + return psnitch.getDatacenter(endpoint); + } + return epState.getApplicationState(ApplicationState.DC).value; + } + + /** + * Return the rack for which an endpoint resides in + * + * @param endpoint the endpoint to process + * @return string of rack + */ + public String getRack(InetAddress endpoint) + { + EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + if (epState == null || epState.getApplicationState(ApplicationState.RACK) == null) + { + if (psnitch == null) + throw new RuntimeException("Could not retrieve rack for " + endpoint + " from gossip and PFS compatibility is disabled"); + else + return psnitch.getRack(endpoint); + } + return epState.getApplicationState(ApplicationState.RACK).value; + } + + @Override + public void gossiperStarting() + { + // Share info via gossip. + logger.info("Adding ApplicationState DC=" + myDC + " rack=" + myRack); + Gossiper.instance.addLocalApplicationState(ApplicationState.DC, StorageService.instance.valueFactory.datacenter(myDC)); + Gossiper.instance.addLocalApplicationState(ApplicationState.RACK, StorageService.instance.valueFactory.rack(myRack)); + } +} diff --git a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java index 7c9084da1d..00adc7e20c 100644 --- a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java @@ -50,7 +50,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch { private static final Logger logger = LoggerFactory.getLogger(PropertyFileSnitch.class); - private static final String RACK_PROPERTY_FILENAME = "cassandra-topology.properties"; + public static final String RACK_PROPERTY_FILENAME = "cassandra-topology.properties"; private static volatile Map endpointMap; private static volatile String[] defaultDCRack; @@ -126,7 +126,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch stream = getClass().getClassLoader().getResourceAsStream(RACK_PROPERTY_FILENAME); properties.load(stream); } - catch (IOException e) + catch (Exception e) { throw new ConfigurationException("Unable to read " + RACK_PROPERTY_FILENAME, e); }