From e75e33fa6dc5e2a3fe061d747cc98679a65ef960 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 21 Jun 2013 10:40:31 -0500 Subject: [PATCH 1/5] refactor reconnecting snitches patch by jasobrown; reviewed by jbellis for CASSANDRA-5681 --- .../locator/Ec2MultiRegionSnitch.java | 71 +-------------- .../locator/GossipingPropertyFileSnitch.java | 63 +------------ .../locator/ReconnectableSnitchHelper.java | 88 +++++++++++++++++++ 3 files changed, 94 insertions(+), 128 deletions(-) create mode 100644 src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java diff --git a/src/java/org/apache/cassandra/locator/Ec2MultiRegionSnitch.java b/src/java/org/apache/cassandra/locator/Ec2MultiRegionSnitch.java index 9317941165..bd5e0917b4 100644 --- a/src/java/org/apache/cassandra/locator/Ec2MultiRegionSnitch.java +++ b/src/java/org/apache/cassandra/locator/Ec2MultiRegionSnitch.java @@ -19,16 +19,11 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; -import java.net.UnknownHostException; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.gms.ApplicationState; -import org.apache.cassandra.gms.EndpointState; import org.apache.cassandra.gms.Gossiper; -import org.apache.cassandra.gms.IEndpointStateChangeSubscriber; -import org.apache.cassandra.gms.VersionedValue; -import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; /** @@ -36,16 +31,13 @@ import org.apache.cassandra.service.StorageService; * * 2) Snitch will set the private IP as a Gossip application state. * - * 3) Snitch implements IESCS and will reset the connection if it is within the + * 3) Uses a helper class that implements IESCS and will reset the public IP connection if it is within the * same region to communicate via private IP. * - * Implements Ec2Snitch to inherit its functionality and extend it for - * Multi-Region. - * * Operational: All the nodes in this cluster needs to be able to (modify the * Security group settings in AWS) communicate via Public IP's. */ -public class Ec2MultiRegionSnitch extends Ec2Snitch implements IEndpointStateChangeSubscriber +public class Ec2MultiRegionSnitch extends Ec2Snitch { private static final String PUBLIC_IP_QUERY_URL = "http://169.254.169.254/latest/meta-data/public-ipv4"; private static final String PRIVATE_IP_QUERY_URL = "http://169.254.169.254/latest/meta-data/local-ipv4"; @@ -62,67 +54,10 @@ public class Ec2MultiRegionSnitch extends Ec2Snitch implements IEndpointStateCha DatabaseDescriptor.setBroadcastAddress(localPublicAddress); } - public void onJoin(InetAddress endpoint, EndpointState epState) - { - if (epState.getApplicationState(ApplicationState.INTERNAL_IP) != null) - reconnect(endpoint, epState.getApplicationState(ApplicationState.INTERNAL_IP)); - } - - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) - { - if (state == ApplicationState.INTERNAL_IP) - reconnect(endpoint, value); - } - - public void onAlive(InetAddress endpoint, EndpointState state) - { - if (state.getApplicationState(ApplicationState.INTERNAL_IP) != null) - reconnect(endpoint, state.getApplicationState(ApplicationState.INTERNAL_IP)); - } - - public void onDead(InetAddress endpoint, EndpointState state) - { - // do nothing - } - - public void onRestart(InetAddress endpoint, EndpointState state) - { - // do nothing - } - - public void onRemove(InetAddress endpoint) - { - // do nothing. - } - - private void reconnect(InetAddress publicAddress, VersionedValue localAddressValue) - { - try - { - reconnect(publicAddress, InetAddress.getByName(localAddressValue.value)); - } - catch (UnknownHostException e) - { - logger.error("Error in getting the IP address resolved: ", e); - } - } - - private void reconnect(InetAddress publicAddress, InetAddress localAddress) - { - if (getDatacenter(publicAddress).equals(getDatacenter(localPublicAddress)) - && MessagingService.instance().getVersion(publicAddress) == MessagingService.current_version - && !MessagingService.instance().getConnectionPool(publicAddress).endPoint().equals(localAddress)) - { - MessagingService.instance().getConnectionPool(publicAddress).reset(localAddress); - logger.debug(String.format("Intiated reconnect to an Internal IP %s for the %s", localAddress, publicAddress)); - } - } - - @Override public void gossiperStarting() { super.gossiperStarting(); Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP, StorageService.instance.valueFactory.internalIP(localPrivateAddress)); - Gossiper.instance.register(this); + Gossiper.instance.register(new ReconnectableSnitchHelper(this, ec2region, true)); } } diff --git a/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java index 071cd099f7..e00239ef6a 100644 --- a/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java @@ -19,7 +19,6 @@ package org.apache.cassandra.locator; import java.net.InetAddress; -import java.net.UnknownHostException; import java.util.Map; import org.apache.cassandra.db.SystemTable; @@ -30,14 +29,11 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.gms.ApplicationState; import org.apache.cassandra.gms.EndpointState; import org.apache.cassandra.gms.Gossiper; -import org.apache.cassandra.gms.IEndpointStateChangeSubscriber; -import org.apache.cassandra.gms.VersionedValue; -import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.service.StorageService; -public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch implements IEndpointStateChangeSubscriber +public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch// implements IEndpointStateChangeSubscriber { private static final Logger logger = LoggerFactory.getLogger(GossipingPropertyFileSnitch.class); @@ -47,7 +43,7 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch i private Map> savedEndpoints; private String DEFAULT_DC = "UNKNOWN_DC"; private String DEFAULT_RACK = "UNKNOWN_RACK"; - private boolean preferLocal; + private final boolean preferLocal; public GossipingPropertyFileSnitch() throws ConfigurationException { @@ -126,64 +122,11 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch i return epState.getApplicationState(ApplicationState.RACK).value; } - // IEndpointStateChangeSubscriber methods - - public void onJoin(InetAddress endpoint, EndpointState epState) - { - if (preferLocal && epState.getApplicationState(ApplicationState.INTERNAL_IP) != null) - reConnect(endpoint, epState.getApplicationState(ApplicationState.INTERNAL_IP)); - } - - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) - { - if (preferLocal && state == ApplicationState.INTERNAL_IP) - reConnect(endpoint, value); - } - - public void onAlive(InetAddress endpoint, EndpointState state) - { - if (preferLocal && state.getApplicationState(ApplicationState.INTERNAL_IP) != null) - reConnect(endpoint, state.getApplicationState(ApplicationState.INTERNAL_IP)); - } - - public void onDead(InetAddress endpoint, EndpointState state) - { - // do nothing - } - - public void onRestart(InetAddress endpoint, EndpointState state) - { - // do nothing - } - - public void onRemove(InetAddress endpoint) - { - // do nothing. - } - - private void reConnect(InetAddress endpoint, VersionedValue versionedValue) - { - if (!getDatacenter(endpoint).equals(myDC)) - return; // do nothing return back... - - try - { - InetAddress remoteIP = InetAddress.getByName(versionedValue.value); - MessagingService.instance().getConnectionPool(endpoint).reset(remoteIP); - logger.debug(String.format("Intiated reconnect to an Internal IP %s for the endpoint %s", remoteIP, endpoint)); - } - catch (UnknownHostException e) - { - logger.error("Error in getting the IP address resolved", e); - } - } - - @Override public void gossiperStarting() { super.gossiperStarting(); Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP, StorageService.instance.valueFactory.internalIP(FBUtilities.getLocalAddress().getHostAddress())); - Gossiper.instance.register(this); + Gossiper.instance.register(new ReconnectableSnitchHelper(this, myDC, preferLocal)); } } diff --git a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java new file mode 100644 index 0000000000..adec95345c --- /dev/null +++ b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java @@ -0,0 +1,88 @@ +package org.apache.cassandra.locator; + +import java.net.InetAddress; +import java.net.UnknownHostException; + +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.IEndpointStateChangeSubscriber; +import org.apache.cassandra.gms.VersionedValue; +import org.apache.cassandra.net.MessagingService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Sidekick helper for snitches that want to reconnect from one IP addr for a node to another. + * Typically, this is for situations like EC2 where a node will have a public address and a private address, + * where we connect on the public, discover the private, and reconnect on the private. + */ +public class ReconnectableSnitchHelper implements IEndpointStateChangeSubscriber +{ + private static final Logger logger = LoggerFactory.getLogger(ReconnectableSnitchHelper.class); + private final IEndpointSnitch snitch; + private final String localDc; + private final boolean preferLocal; + + public ReconnectableSnitchHelper(IEndpointSnitch snitch, String localDc, boolean preferLocal) + { + this.snitch = snitch; + this.localDc = localDc; + this.preferLocal = preferLocal; + } + + private void reconnect(InetAddress publicAddress, VersionedValue localAddressValue) + { + try + { + reconnect(publicAddress, InetAddress.getByName(localAddressValue.value)); + } + catch (UnknownHostException e) + { + logger.error("Error in getting the IP address resolved: ", e); + } + } + + private void reconnect(InetAddress publicAddress, InetAddress localAddress) + { + if (snitch.getDatacenter(publicAddress).equals(localDc) + && MessagingService.instance().getVersion(publicAddress) == MessagingService.current_version + && !MessagingService.instance().getConnectionPool(publicAddress).endPoint().equals(localAddress)) + { + MessagingService.instance().getConnectionPool(publicAddress).reset(localAddress); + logger.debug(String.format("Intiated reconnect to an Internal IP %s for the %s", localAddress, publicAddress)); + } + } + + public void onJoin(InetAddress endpoint, EndpointState epState) + { + if (preferLocal && epState.getApplicationState(ApplicationState.INTERNAL_IP) != null) + reconnect(endpoint, epState.getApplicationState(ApplicationState.INTERNAL_IP)); + } + + public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) + { + if (preferLocal && state == ApplicationState.INTERNAL_IP) + reconnect(endpoint, value); + } + + public void onAlive(InetAddress endpoint, EndpointState state) + { + if (preferLocal && state.getApplicationState(ApplicationState.INTERNAL_IP) != null) + reconnect(endpoint, state.getApplicationState(ApplicationState.INTERNAL_IP)); + } + + public void onDead(InetAddress endpoint, EndpointState state) + { + // do nothing. + } + + public void onRemove(InetAddress endpoint) + { + // do nothing. + } + + public void onRestart(InetAddress endpoint, EndpointState state) + { + // do nothing. + } +} From 110d283afd780774a44368b17177b5e8e781e37f Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 21 Jun 2013 11:13:21 -0500 Subject: [PATCH 2/5] restore fetching global trace state in default .execute method --- .../concurrent/DebuggableThreadPoolExecutor.java | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 26441ec8b1..46a3216e0c 100644 --- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java @@ -133,7 +133,9 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor implements public void execute(Runnable command, TraceState state) { - super.execute(state == null ? command : new TraceSessionWrapper(command, state)); + super.execute(state == null || command instanceof TraceSessionWrapper + ? command + : new TraceSessionWrapper(command, state)); } // execute does not call newTaskFor @@ -141,7 +143,7 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor implements public void execute(Runnable command) { super.execute(isTracing() && !(command instanceof TraceSessionWrapper) - ? new TraceSessionWrapper(command) + ? new TraceSessionWrapper(Executors.callable(command, null)) : command); } @@ -261,11 +263,6 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor implements { private final TraceState state; - public TraceSessionWrapper(Runnable command) - { - this(command, null); - } - public TraceSessionWrapper(Callable callable) { super(callable); From b7e13b89c265c28acfb624a984b97a06a837c3ea Mon Sep 17 00:00:00 2001 From: Jason Brown Date: Fri, 21 Jun 2013 09:23:17 -0700 Subject: [PATCH 3/5] Gossiper.handleMajorStateChange can lose existing node ApplicationState patch by jasobrown; reviewe4d by jbellis for CASSANDRA-5665 --- .../org/apache/cassandra/gms/Gossiper.java | 47 ++++++++++--------- 1 file changed, 26 insertions(+), 21 deletions(-) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index efa9865c25..6b0bbe94c2 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -871,6 +871,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean if (logger.isTraceEnabled()) logger.trace("Updating heartbeat state generation to " + remoteGeneration + " from " + localGeneration + " for " + ep); // major state change will handle the update by inserting the remote state directly + copyNewerApplicationStates(remoteState, localEpStatePtr); handleMajorStateChange(ep, remoteState); } else if ( remoteGeneration == localGeneration ) // generation has not changed, apply new states @@ -880,11 +881,18 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean int remoteMaxVersion = getMaxEndpointStateVersion(remoteState); if ( remoteMaxVersion > localMaxVersion ) { - // apply states, but do not notify since there is no major change - applyNewStates(ep, localEpStatePtr, remoteState); + if (logger.isTraceEnabled()) + { + logger.trace("Updating heartbeat state version to " + remoteState.getHeartBeatState().getHeartBeatVersion() + + " from " + localEpStatePtr.getHeartBeatState().getHeartBeatVersion() + " for " + ep); + } + localEpStatePtr.setHeartBeatState(remoteState.getHeartBeatState()); + Map merged = copyNewerApplicationStates(localEpStatePtr, remoteState); + for (Entry appState : merged.entrySet()) + doNotifications(ep, appState.getKey(), appState.getValue()); } else if (logger.isTraceEnabled()) - logger.trace("Ignoring remote version " + remoteMaxVersion + " <= " + localMaxVersion + " for " + ep); + logger.trace("Ignoring remote version " + remoteMaxVersion + " <= " + localMaxVersion + " for " + ep); if (!localEpStatePtr.isAlive() && !isDeadState(localEpStatePtr)) // unless of course, it was dead markAlive(ep, localEpStatePtr); } @@ -903,28 +911,25 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } } - private void applyNewStates(InetAddress addr, EndpointState localState, EndpointState remoteState) + private Map copyNewerApplicationStates(EndpointState toState, EndpointState fromState) { - // don't assert here, since if the node restarts the version will go back to zero - int oldVersion = localState.getHeartBeatState().getHeartBeatVersion(); - - localState.setHeartBeatState(remoteState.getHeartBeatState()); - if (logger.isTraceEnabled()) - logger.trace("Updating heartbeat state version to " + localState.getHeartBeatState().getHeartBeatVersion() + " from " + oldVersion + " for " + addr + " ..."); - - // we need to make two loops here, one to apply, then another to notify, this way all states in an update are present and current when the notifications are received - for (Entry remoteEntry : remoteState.getApplicationStateMap().entrySet()) + Map merged = new HashMap(); + for (Entry fromEntry : fromState.getApplicationStateMap().entrySet()) { - ApplicationState remoteKey = remoteEntry.getKey(); - VersionedValue remoteValue = remoteEntry.getValue(); + ApplicationState key = fromEntry.getKey(); + VersionedValue value = fromEntry.getValue(); + assert fromState.getHeartBeatState().getGeneration() == toState.getHeartBeatState().getGeneration(); - assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration(); - localState.addApplicationState(remoteKey, remoteValue); - } - for (Entry remoteEntry : remoteState.getApplicationStateMap().entrySet()) - { - doNotifications(addr, remoteEntry.getKey(), remoteEntry.getValue()); + if ( (toState.applicationState.containsKey(key) && toState.applicationState.get(key).compareTo(value) < 0) + || !toState.applicationState.containsKey(key) ) + { + if (logger.isTraceEnabled()) + logger.trace("merging {}:{} into ApplicationState", key, value); + toState.addApplicationState(key, value); + merged.put(key, value); + } } + return merged; } // notify that an application state has changed From 79410e4a887e7adac796339c1a565436deaf254a Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 21 Jun 2013 19:12:31 +0200 Subject: [PATCH 4/5] Versions and license for 1.2.6 release --- NEWS.txt | 1 - build.xml | 2 +- debian/changelog | 6 ++++++ .../locator/ReconnectableSnitchHelper.java | 17 +++++++++++++++++ 4 files changed, 24 insertions(+), 2 deletions(-) diff --git a/NEWS.txt b/NEWS.txt index dbc9aab21d..cb40981c1b 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -21,7 +21,6 @@ Upgrading - hinted_handoff_throttle_in_kb is now reduced by a factor proportional to the number of nodes in the cluster (see https://issues.apache.org/jira/browse/CASSANDRA-5272). - - CQL3 syntax for CREATE CUSTOM INDEX has been updated. See CQL3 documentation for details. diff --git a/build.xml b/build.xml index 46ffbb1cef..4e3235161e 100644 --- a/build.xml +++ b/build.xml @@ -25,7 +25,7 @@ - + diff --git a/debian/changelog b/debian/changelog index 626374fb52..3be9ba0759 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (1.2.6) unstable; urgency=low + + * New release + + -- Sylvain Lebresne Fri, 21 Jun 2013 18:58:24 +0200 + cassandra (1.2.5) unstable; urgency=low * New release diff --git a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java index adec95345c..bef245ec50 100644 --- a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java +++ b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java @@ -1,3 +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. + */ package org.apache.cassandra.locator; import java.net.InetAddress; From 362473fb3772eae38d1241aae45dd0a33804fd5c Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 21 Jun 2013 13:32:25 -0500 Subject: [PATCH 5/5] match dependency to shipped version --- build.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.xml b/build.xml index 4e3235161e..8d54554ef9 100644 --- a/build.xml +++ b/build.xml @@ -340,7 +340,7 @@ - +