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

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1153727 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-08-04 02:04:41 +00:00
parent 032b476c52
commit 3120b65d34
11 changed files with 290 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

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

5
src/java/org/apache/cassandra/streaming/StreamInSession.java
View File

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

57
test/distributed/org/apache/cassandra/CassandraServiceController.java
View File

@ -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<InetAddress> 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<InetAddress> 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<InetAddress> hosts, String functionName, String... functionArgs)
{
final Set<String> hostset = new HashSet<String>();
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<? extends NodeMetadata,ExecResponse> 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<InetAddress> getHosts()
@ -290,24 +303,26 @@ public class CassandraServiceController
class Failure
{
private InetAddress[] hosts;
private List<InetAddress> hosts;
public Failure(InetAddress... hosts)
public Failure(List<InetAddress> 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);
}
}
}

1
test/distributed/org/apache/cassandra/MutationTest.java
View File

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

6
test/distributed/org/apache/cassandra/TestBase.java
View File

@ -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.<CfDef>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<Class<Exception>> expected = new HashSet<Class<Exception>>();
private long timeout = StorageService.RING_DELAY;
private long timeout = StorageService.RING_DELAY * 2;
public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key)
{

85
test/resources/functions/configure_cassandra.sh Normal file
View File

@ -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}<Seed>${server}</Seed>"
done
#TODO set replication
sed -i -e "s|<Seed>127.0.0.1</Seed>|$seeds|" $config_file
sed -i -e "s|<ListenAddress>localhost</ListenAddress>|<ListenAddress>$PRIVATE_SELF_HOST</ListenAddress>|" $config_file
sed -i -e "s|<ThriftAddress>localhost</ThriftAddress>|<ThriftAddress>$PUBLIC_SELF_HOST</ThriftAddress>|" $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 &
}

55
test/resources/functions/install_cassandra.sh Normal file
View File

@ -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 <<EOF
$CASSANDRA_HOME/bin/cassandra > /dev/null 2>&1 &
EOF
}

26
test/resources/functions/nodetool_cassandra.sh Normal file
View File

@ -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 $?
}

27
test/resources/functions/start_cassandra.sh Normal file
View File

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

28
test/resources/functions/stop_cassandra.sh Normal file
View File

@ -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 $?
}

27
test/resources/functions/wipe_cassandra.sh Normal file
View File

@ -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 $?
}