diff --git a/CHANGES.txt b/CHANGES.txt index 467ea1f664..38cadba4a5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -64,6 +64,7 @@ * add asynchronous and half-sync/half-async thrift servers (CASSANDRA-1405) * fix potential use of free'd native memory in SerializingCache (CASSANDRA-2951) + * include files-to-be-streamed in StreamInSession.getSources (CASSANDRA-2972) 0.8.2 diff --git a/src/java/org/apache/cassandra/streaming/StreamInSession.java b/src/java/org/apache/cassandra/streaming/StreamInSession.java index 4c33b72864..d173d644fc 100644 --- a/src/java/org/apache/cassandra/streaming/StreamInSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamInSession.java @@ -201,10 +201,9 @@ public class StreamInSession if (entry.getKey().left.equals(host)) { StreamInSession session = entry.getValue(); - set.addAll(session.files); - if(session.current != null) { + if (session.current != null) set.add(session.current); - } + set.addAll(session.files); } } return set; diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java index 3ffcb3fe17..10fdef5884 100644 --- a/test/distributed/org/apache/cassandra/CassandraServiceController.java +++ b/test/distributed/org/apache/cassandra/CassandraServiceController.java @@ -39,21 +39,21 @@ import org.apache.thrift.transport.TTransport; import org.apache.whirr.service.*; import org.apache.whirr.service.Cluster.Instance; import org.apache.whirr.service.cassandra.CassandraClusterActionHandler; -import org.apache.whirr.service.jclouds.RunUrlStatement; +import org.apache.whirr.service.jclouds.StatementBuilder; import org.jclouds.blobstore.domain.BlobMetadata; import org.jclouds.compute.ComputeService; import org.jclouds.compute.domain.ExecResponse; import org.jclouds.compute.domain.NodeMetadata; import org.jclouds.compute.options.RunScriptOptions; +import org.jclouds.compute.RunScriptOnNodesException; import org.jclouds.domain.Credentials; import org.jclouds.scriptbuilder.domain.OsFamily; +import org.jclouds.scriptbuilder.domain.Statements; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.jclouds.io.Payloads.newStringPayload; - public class CassandraServiceController { private static final Logger LOG = @@ -76,7 +76,6 @@ public class CassandraServiceController private Service service; private Cluster cluster; private ComputeService computeService; - private Credentials credentials; private CompositeConfiguration config; private BlobMetadata tarball; private List hosts; @@ -113,12 +112,12 @@ public class CassandraServiceController try { Cassandra.Client client = createClient(addr); - client.describe_cluster_name(); break; } catch (TException e) { + LOG.debug(e.toString()); try { Thread.sleep(1000); @@ -168,14 +167,13 @@ public class CassandraServiceController for (Instance instance : cluster.getInstances()) { hosts.add(instance.getPublicAddress()); - credentials = instance.getLoginCredentials(); } - waitForClusterInitialization(); - ShutdownHook shutdownHook = new ShutdownHook(this); Runtime.getRuntime().addShutdownHook(shutdownHook); + waitForClusterInitialization(); + running = true; } @@ -231,7 +229,7 @@ public class CassandraServiceController */ public void nodetool(String args, InetAddress... hosts) { - callOnHosts(String.format("apache/cassandra/nodetool %s", args), hosts); + callOnHosts(Arrays.asList(hosts), "nodetool_cassandra", args); } /** @@ -239,25 +237,31 @@ public class CassandraServiceController */ public void wipeHosts(InetAddress... hosts) { - callOnHosts("apache/cassandra/wipe-state", hosts); + callOnHosts(Arrays.asList(hosts), "wipe_cassandra"); } public Failure failHosts(List hosts) { - return new Failure(hosts.toArray(new InetAddress[hosts.size()])).trigger(); + return new Failure(hosts).trigger(); } public Failure failHosts(InetAddress... hosts) { - return new Failure(hosts).trigger(); + return new Failure(Arrays.asList(hosts)).trigger(); } /** TODO: Move to CassandraService? */ - protected void callOnHosts(String payload, InetAddress... hosts) + protected void callOnHosts(List hosts, String functionName, String... functionArgs) { final Set hostset = new HashSet(); + for (InetAddress host : hosts) hostset.add(host.getHostAddress()); + + StatementBuilder statementBuilder = new StatementBuilder(); + statementBuilder.addStatement(Statements.call(functionName, functionArgs)); + Credentials credentials = new Credentials(clusterSpec.getClusterUser(), clusterSpec.getPrivateKey()); + Map results; try { @@ -269,18 +273,27 @@ public class CassandraServiceController intersection.retainAll(node.getPublicAddresses()); return !intersection.isEmpty(); } - }, newStringPayload(new RunUrlStatement(clusterSpec.getRunUrlBase(), payload).render(OsFamily.UNIX)), - RunScriptOptions.Builder.overrideCredentialsWith(credentials)); + }, + statementBuilder, + RunScriptOptions.Builder.overrideCredentialsWith(credentials).wrapInInitScript(false).runAsRoot(false)); } - catch (Exception e) + catch (RunScriptOnNodesException e) { throw new RuntimeException(e); } + if (results.size() != hostset.size()) + { throw new RuntimeException(results.size() + " hosts matched " + hostset + ": " + results); + } + for (ExecResponse response : results.values()) + { if (response.getExitCode() != 0) - throw new RuntimeException("Call " + payload + " failed on at least one of " + hostset + ": " + results.values()); + { + throw new RuntimeException("Call " + functionName + " failed on at least one of " + hostset + ": " + results.values()); + } + } } public List getHosts() @@ -290,24 +303,26 @@ public class CassandraServiceController class Failure { - private InetAddress[] hosts; + private List hosts; - public Failure(InetAddress... hosts) + public Failure(List hosts) { this.hosts = hosts; } public Failure trigger() { - callOnHosts("apache/cassandra/stop", hosts); + callOnHosts(hosts, "stop_cassandra"); return this; } public void resolve() { - callOnHosts("apache/cassandra/start", hosts); + callOnHosts(hosts, "start_cassandra"); for (InetAddress host : hosts) + { waitForNodeInitialization(host); + } } } diff --git a/test/distributed/org/apache/cassandra/MutationTest.java b/test/distributed/org/apache/cassandra/MutationTest.java index ad287efecb..ab93fd3d4c 100644 --- a/test/distributed/org/apache/cassandra/MutationTest.java +++ b/test/distributed/org/apache/cassandra/MutationTest.java @@ -129,6 +129,7 @@ public class MutationTest extends TestBase .expecting(UnavailableException.class).perform(ConsistencyLevel.QUORUM); } finally { failure.resolve(); + Thread.sleep(10000); } // with all nodes up diff --git a/test/distributed/org/apache/cassandra/TestBase.java b/test/distributed/org/apache/cassandra/TestBase.java index c4032044e4..07b68c22b8 100644 --- a/test/distributed/org/apache/cassandra/TestBase.java +++ b/test/distributed/org/apache/cassandra/TestBase.java @@ -61,7 +61,6 @@ public abstract class TestBase cfdef.setRow_cache_size(1000); cfdef.setRow_cache_save_period_in_seconds(0); cfdef.setKey_cache_save_period_in_seconds(3600); - cfdef.setMemtable_flush_after_mins(59); cfdef.setMemtable_throughput_in_mb(255); cfdef.setMemtable_operations_in_millions(0.29); } @@ -87,8 +86,7 @@ public abstract class TestBase client.system_add_keyspace(new KsDef(name, "org.apache.cassandra.locator.SimpleStrategy", Arrays.asList(cfdef)) - .setStrategy_options(stratOptions) - .setCf_defs(Collections.emptyList())); + .setStrategy_options(stratOptions)); // poll, until KS added for (InetAddress host : hosts) @@ -199,7 +197,7 @@ public abstract class TestBase protected long timestamp; private Set> expected = new HashSet>(); - private long timeout = StorageService.RING_DELAY; + private long timeout = StorageService.RING_DELAY * 2; public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key) { diff --git a/test/resources/functions/configure_cassandra.sh b/test/resources/functions/configure_cassandra.sh new file mode 100644 index 0000000000..346465384a --- /dev/null +++ b/test/resources/functions/configure_cassandra.sh @@ -0,0 +1,85 @@ +# +# 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. +# +function configure_cassandra() { + local OPTIND + local OPTARG + + . /etc/profile + + CLOUD_PROVIDER= + while getopts "c:" OPTION; do + case $OPTION in + c) + CLOUD_PROVIDER="$OPTARG" + shift $((OPTIND-1)); OPTIND=1 + ;; + esac + done + + case $CLOUD_PROVIDER in + # We want the gossip protocol to use internal (private) addresses, and the + # client to use public addresses. + # See http://wiki.apache.org/cassandra/FAQ#cant_listen_on_ip_any + ec2 | aws-ec2 ) + PRIVATE_SELF_HOST=`wget -q -O - http://169.254.169.254/latest/meta-data/local-ipv4` + # EC2 is NATed + PUBLIC_SELF_HOST=$PRIVATE_SELF_HOST + ;; + *) + PUBLIC_SELF_HOST=`/sbin/ifconfig eth0 | grep 'inet addr:' | cut -d: -f2 | awk '{ print $1}'` + PRIVATE_SELF_HOST=`/sbin/ifconfig eth1 | grep 'inet addr:' | cut -d: -f2 | awk '{ print $1}'` + ;; + esac + + OH_SIX_CONFIG="/etc/cassandra/conf/storage-conf.xml" + + if [[ -e "$OH_SIX_CONFIG" ]] ; then + config_file=$OH_SIX_CONFIG + seeds="" + for server in "$@"; do + seeds="${seeds}${server}" + done + + #TODO set replication + sed -i -e "s|127.0.0.1|$seeds|" $config_file + sed -i -e "s|localhost|$PRIVATE_SELF_HOST|" $config_file + sed -i -e "s|localhost|$PUBLIC_SELF_HOST|" $config_file + else + config_file="/etc/cassandra/conf/cassandra.yaml" + if [[ "x"`grep -e '^seeds:' $config_file` == "x" ]]; then + seeds="$1" # 08 format seeds + shift + for server in "$@"; do + seeds="${seeds},${server}" + done + sed -i -e "s|- seeds: \"127.0.0.1\"|- seeds: \"${seeds}\"|" $config_file + else + seeds="" # 07 format seeds + for server in "$@"; do + seeds="${seeds}\n - ${server}" + done + sed -i -e "/^seeds:/,/^/d" $config_file ; echo -e "seeds:${seeds}" >> $config_file + fi + + sed -i -e "s|listen_address: localhost|listen_address: $PRIVATE_SELF_HOST|" $config_file + sed -i -e "s|rpc_address: localhost|rpc_address: $PUBLIC_SELF_HOST|" $config_file + fi + + # Now that it's configured, start Cassandra + nohup /etc/rc.local & + +} diff --git a/test/resources/functions/install_cassandra.sh b/test/resources/functions/install_cassandra.sh new file mode 100644 index 0000000000..59f3e98c81 --- /dev/null +++ b/test/resources/functions/install_cassandra.sh @@ -0,0 +1,55 @@ +# +# 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. +# +function install_cassandra() { + + C_MAJOR_VERSION=$1 + C_TAR_URL=$2 + + c_tar_file=`basename $C_TAR_URL` + c_tar_dir=`echo $c_tar_file | awk -F '-bin' '{print $1}'` + + CASSANDRA_HOME=/usr/local/$c_tar_dir + C_CONF_DIR=/etc/cassandra/conf + C_LOG_DIR=/var/log/cassandra + + install_tarball $C_TAR_URL + + echo "export CASSANDRA_HOME=$CASSANDRA_HOME" >> /etc/profile + echo "export CASSANDRA_CONF=$C_CONF_DIR" >> /etc/profile + echo 'export PATH=$CASSANDRA_HOME/bin:$PATH' >> /etc/profile + + mkdir -p /mnt/cassandra/logs + ln -s /mnt/cassandra/logs $C_LOG_DIR + mkdir -p $C_CONF_DIR + cp $CASSANDRA_HOME/conf/log4j*.properties $C_CONF_DIR + if [[ "0.6" == "$C_MAJOR_VERSION" ]] ; then + cp $CASSANDRA_HOME/conf/storage-conf.xml $C_CONF_DIR + sed -i -e "s|CASSANDRA_CONF=\$cassandra_home/conf|CASSANDRA_CONF=$C_CONF_DIR|" $CASSANDRA_HOME/bin/cassandra.in.sh + else + cp $CASSANDRA_HOME/conf/cassandra.yaml $C_CONF_DIR + cp $CASSANDRA_HOME/conf/cassandra-env.sh $C_CONF_DIR + # FIXME: this is only necessary because CASSANDRA_CONF/HOME are not in root's environment as they should be + sed -i -e "s|CASSANDRA_CONF=\$CASSANDRA_HOME/conf|CASSANDRA_CONF=$C_CONF_DIR|" $CASSANDRA_HOME/bin/cassandra.in.sh + fi + + # Ensure Cassandra starts on boot + sed -i -e "s/exit 0//" /etc/rc.local + cat >> /etc/rc.local < /dev/null 2>&1 & +EOF + +} diff --git a/test/resources/functions/nodetool_cassandra.sh b/test/resources/functions/nodetool_cassandra.sh new file mode 100644 index 0000000000..d7c6bd23b4 --- /dev/null +++ b/test/resources/functions/nodetool_cassandra.sh @@ -0,0 +1,26 @@ +# +# 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. +# + +function nodetool_cassandra() { + set -x + set -e + + . /etc/profile + + sudo nodetool -h localhost $* + exit $? +} diff --git a/test/resources/functions/start_cassandra.sh b/test/resources/functions/start_cassandra.sh new file mode 100644 index 0000000000..8347e16c63 --- /dev/null +++ b/test/resources/functions/start_cassandra.sh @@ -0,0 +1,27 @@ +# +# 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. +# + +function start_cassandra() { + set -x + set -e + + . /etc/profile + + # launch using the script created by install + sudo nohup /etc/rc.local & + sleep 5 +} diff --git a/test/resources/functions/stop_cassandra.sh b/test/resources/functions/stop_cassandra.sh new file mode 100644 index 0000000000..8565c10caa --- /dev/null +++ b/test/resources/functions/stop_cassandra.sh @@ -0,0 +1,28 @@ +# +# 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. +# + +function stop_cassandra() { + set -x + set -e + + . /etc/profile + + # assume we are the only java process + # FIXME: unsafe assumption, but easier than true init + sudo killall java + exit $? +} diff --git a/test/resources/functions/wipe_cassandra.sh b/test/resources/functions/wipe_cassandra.sh new file mode 100644 index 0000000000..b3e8754597 --- /dev/null +++ b/test/resources/functions/wipe_cassandra.sh @@ -0,0 +1,27 @@ +# +# 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. +# + +function wipe_cassandra() { + set -x + set -e + + . /etc/profile + + DD=/var/lib/cassandra + sudo rm -Rf $DD/data $DD/commitlog $DD/saved_caches + exit $? +}