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

Remove old distributed tests. 

Patch by brandonwilliams reviewed by jbellis for CASSANDRA-3602

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1212572 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Brandon Williams 2011-12-09 18:45:39 +00:00
parent bf0f80cbb6
commit 250e63bc74
9 changed files with 0 additions and 1363 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
build.xml
View File

@ -60,7 +60,6 @@
<property name="test.name" value="*Test"/>
<property name="test.unit.src" value="${test.dir}/unit"/>
<property name="test.long.src" value="${test.dir}/long"/>
<property name="test.distributed.src" value="${test.dir}/distributed"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<condition property="version" value="${base.version}">
<isset property="release"/>
@ -287,9 +286,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<artifact:remoteRepository id="central" url="${artifact.remoteRepository.central}"/>
<artifact:remoteRepository id="java.net2" url="${artifact.remoteRepository.java.net2}"/>
<artifact:remoteRepository id="apache" url="${artifact.remoteRepository.apache}"/>
<!-- for distributed tests -->
<artifact:remoteRepository id="jclouds" url="${artifact.remoteRepository.jclouds}"/>
<artifact:remoteRepository id="oauth" url="${artifact.remoteRepository.oauth}"/>
<macrodef name="install">
<attribute name="pomFile"/>
@ -972,22 +968,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
</copy>
</target>
<target name="build-distributed-test" depends="build-test,maven-ant-tasks-retrieve-test" description="Compile distributed test classes (which have additional deps)">
<javac
debug="true"
debuglevel="${debuglevel}"
destdir="${test.classes}">
<classpath>
<path refid="cassandra.classpath"/>
<pathelement location="${test.classes}"/>
<fileset dir="${test.lib}">
<include name="**/*.jar" />
</fileset>
</classpath>
<src path="${test.distributed.src}"/>
</javac>
</target>
<macrodef name="testmacro">
<attribute name="suitename" />
<attribute name="inputdir" />
@ -1076,14 +1056,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
timeout="${test.long.timeout}" />
</target>
<!-- Depends on artifacts so that we can push a tarball to remote nodes, and has its own build target for cloudy deps. -->
<target name="distributed-test" depends="build-distributed-test,artifacts" description="Execute distributed tests: see ${test.distributed.src}/README.txt">
<testmacro suitename="distributed" inputdir="${test.distributed.src}" timeout="1200000" forkmode="once">
<jvmarg value="-Dwhirr.config=${whirr.config}"/>
<jvmarg value="-Dwhirr.cassandra_tarball=${build.dir}/${final.name}-bin.tar.gz"/>
</testmacro>
</target>
<!-- instruments the classes to later create code coverage reports -->
<target name="cobertura-instrument" depends="build,build-test">
<taskdef resource="tasks.properties">

57
test/distributed/README.txt
View File

@ -1,57 +0,0 @@
Distributed Test Harness
Sub-project description
-----------------------
A distributed test harness that deploys a cluster to a cloud provider,
via Apache Whirr, runs tests against that cluster, then tears down
the deployed cluster.
Requirements
------------
* A cloud provider account. [see: http://incubator.apache.org/whirr/]
Getting started
---------------
First, setup an account w/ a supported cloud provider. Then, refer to
the Whirr documentation for configuration instructions. Refer to:
* http://incubator.apache.org/whirr/quick-start-guide.html
Setup your personal whirr configuration properties. The shared whirr
configuration is located at:
* test/resources/whirr-default.properties
An example EC2/S3 whirr configuration would be:
###############################################
whirr.cluster-user=[username]
whirr.provider=aws-ec2
whirr.location-id=us-west-1
whirr.image-id=us-west-1/ami-16f3a253
whirr.hardware-id=m1.large
whirr.identity=[EC2 Access Key ID]
whirr.credential=[EC2 Secret Access Key]
whirr.private-key-file=${sys:user.home}/.ssh/id_rsa
whirr.public-key-file=${sys:user.home}/.ssh/id_rsa.pub
whirr.run-url-base=http://hoodidge.net/scripts/
whirr.blobstore.provider=aws-s3
whirr.blobstore.container=cassandratests
###############################################
The distributed tests are located in:
* test/distributed
Run the tests via ant:
* ant distributed-test -Dwhirr.config=my-whirr.properties
The ant target will:
* download extra dependencies via Apache Ivy
* compile the distributed tests
* push the local working copy to a blobstore to fetch from the test nodes
* deploy a cluster via Apache Whirr
* run the distributed tests against the cluster
* tear down the deployed cluster

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

@ -1,336 +0,0 @@
/**
* 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;
import java.net.InetAddress;
import java.net.URI;
import java.util.*;
import com.google.common.base.Predicate;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.utils.BlobUtils;
import org.apache.cassandra.utils.KeyPair;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
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.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;
public class CassandraServiceController
{
private static final Logger LOG =
LoggerFactory.getLogger(CassandraServiceController.class);
protected static int CLIENT_PORT = 9160;
protected static int JMX_PORT = 7199;
private static final CassandraServiceController INSTANCE =
new CassandraServiceController();
public static CassandraServiceController getInstance()
{
return INSTANCE;
}
private boolean running;
private ClusterSpec clusterSpec;
private Service service;
private Cluster cluster;
private ComputeService computeService;
private CompositeConfiguration config;
private BlobMetadata tarball;
private List<InetAddress> hosts;
private CassandraServiceController()
{
}
public Cassandra.Client createClient(InetAddress addr) throws TException
{
TTransport transport = new TSocket(
addr.getHostAddress(),
CLIENT_PORT,
200000);
transport = new TFramedTransport(transport);
TProtocol protocol = new TBinaryProtocol(transport);
Cassandra.Client client = new Cassandra.Client(protocol);
transport.open();
return client;
}
private void waitForClusterInitialization()
{
for (InetAddress host : hosts)
waitForNodeInitialization(host);
}
private void waitForNodeInitialization(InetAddress addr)
{
while (true)
{
try
{
Cassandra.Client client = createClient(addr);
client.describe_cluster_name();
break;
}
catch (TException e)
{
LOG.debug(e.toString());
try
{
Thread.sleep(1000);
}
catch (InterruptedException ie)
{
break;
}
}
}
}
public synchronized void startup() throws Exception
{
LOG.info("Starting up cluster...");
config = new CompositeConfiguration();
if (System.getProperty("whirr.config") != null)
{
config.addConfiguration(
new PropertiesConfiguration(System.getProperty("whirr.config")));
}
config.addConfiguration(new PropertiesConfiguration("whirr-default.properties"));
clusterSpec = new ClusterSpec(config);
if (clusterSpec.getPrivateKey() == null)
{
Map<String, String> pair = KeyPair.generate();
clusterSpec.setPublicKey(pair.get("public"));
clusterSpec.setPrivateKey(pair.get("private"));
}
// if a local tarball is available deploy it to the blobstore where it will be available to cassandra
if (System.getProperty("whirr.cassandra_tarball") != null)
{
Pair<BlobMetadata,URI> blob = BlobUtils.storeBlob(config, clusterSpec, System.getProperty("whirr.cassandra_tarball"));
tarball = blob.left;
config.setProperty(CassandraClusterActionHandler.BIN_TARBALL, blob.right.toURL().toString());
// TODO: parse the CassandraVersion property file instead
config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.8");
}
service = new ServiceFactory().create(clusterSpec.getServiceName());
cluster = service.launchCluster(clusterSpec);
computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService();
hosts = new ArrayList<InetAddress>();
for (Instance instance : cluster.getInstances())
{
hosts.add(instance.getPublicAddress());
}
ShutdownHook shutdownHook = new ShutdownHook(this);
Runtime.getRuntime().addShutdownHook(shutdownHook);
waitForClusterInitialization();
running = true;
}
public synchronized void shutdown()
{
// catch and log errors, we're in a runtime shutdown hook
try
{
LOG.info("Shutting down cluster...");
if (tarball != null)
BlobUtils.deleteBlob(config, clusterSpec, tarball);
if (service != null)
service.destroyCluster(clusterSpec);
running = false;
}
catch (Exception e)
{
LOG.error("Error shutting down cluster.", e);
}
}
public class ShutdownHook extends Thread
{
private CassandraServiceController controller;
public ShutdownHook(CassandraServiceController controller)
{
this.controller = controller;
}
public void run()
{
controller.shutdown();
}
}
public synchronized boolean ensureClusterRunning() throws Exception
{
if (running)
{
LOG.info("Cluster already running.");
return false;
}
else
{
startup();
return true;
}
}
/**
* Execute nodetool with args against localhost from the given host.
*/
public void nodetool(String args, InetAddress... hosts)
{
callOnHosts(Arrays.asList(hosts), "nodetool_cassandra", args);
}
/**
* Wipes all persisted state for the given node, leaving it as if it had just started.
*/
public void wipeHosts(InetAddress... hosts)
{
callOnHosts(Arrays.asList(hosts), "wipe_cassandra");
}
public Failure failHosts(List<InetAddress> hosts)
{
return new Failure(hosts).trigger();
}
public Failure failHosts(InetAddress... hosts)
{
return new Failure(Arrays.asList(hosts)).trigger();
}
/** TODO: Move to CassandraService? */
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
{
results = computeService.runScriptOnNodesMatching(new Predicate<NodeMetadata>()
{
public boolean apply(NodeMetadata node)
{
Set<String> intersection = new HashSet<String>(hostset);
intersection.retainAll(node.getPublicAddresses());
return !intersection.isEmpty();
}
},
statementBuilder,
RunScriptOptions.Builder.overrideCredentialsWith(credentials).wrapInInitScript(false).runAsRoot(false));
}
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 " + functionName + " failed on at least one of " + hostset + ": " + results.values());
}
}
}
public List<InetAddress> getHosts()
{
return hosts;
}
class Failure
{
private List<InetAddress> hosts;
public Failure(List<InetAddress> hosts)
{
this.hosts = hosts;
}
public Failure trigger()
{
callOnHosts(hosts, "stop_cassandra");
return this;
}
public void resolve()
{
callOnHosts(hosts, "start_cassandra");
for (InetAddress host : hosts)
{
waitForNodeInitialization(host);
}
}
}
public InetAddress getPublicHost(InetAddress privateHost)
{
for (Instance instance : cluster.getInstances())
if (privateHost.equals(instance.getPrivateAddress()))
return instance.getPublicAddress();
throw new RuntimeException("No public host for private host " + privateHost);
}
}

88
test/distributed/org/apache/cassandra/CountersTest.java
View File

@ -1,88 +0,0 @@
/**
* 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;
import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.thrift.TException;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class CountersTest extends TestBase
{
@Test
public void testWriteOneReadAll() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
// create a keyspace that performs counter validation
final String keyspace = "TestOneNodeWrite";
keyspace(keyspace).rf(3).validator("CounterColumnType").create();
for (InetAddress host : hosts)
{
Cassandra.Client client = controller.createClient(host);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
add(client, key, "Standard1", "c1", 1, ConsistencyLevel.ONE);
add(client, key, "Standard1", "c2", 2, ConsistencyLevel.ONE);
new CounterGet(client, "Standard1", key).name("c1").value(1L).perform(ConsistencyLevel.ALL);
new CounterGet(client, "Standard1", key).name("c2").value(2L).perform(ConsistencyLevel.ALL);
}
}
protected class CounterGet extends RetryingAction<Long>
{
public CounterGet(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
ByteBuffer bname = ByteBuffer.wrap(name.getBytes());
ColumnPath cpath = new ColumnPath(cf).setColumn(bname);
CounterColumn col = client.get(key, cpath, cl).counter_column;
assertEquals(bname, col.name);
assertEquals(value.longValue(), col.value);
}
}
/** NB: Counter increments are unfortunately not idempotent, so we don't provide a RetyingAction to perform them. */
protected void add(Cassandra.Client client, ByteBuffer key, String cf, String name, long value, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
CounterColumn col = new CounterColumn(ByteBuffer.wrap(name.getBytes()), value);
client.add(key, new ColumnParent(cf), col, cl);
}
}

125
test/distributed/org/apache/cassandra/MovementTest.java
View File

@ -1,125 +0,0 @@
/**
* 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;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class MovementTest extends TestBase
{
private static final String STANDARD_CF = "Standard1";
private static final ColumnParent STANDARD = new ColumnParent(STANDARD_CF);
/** Inserts 1000 keys with names such that at least 1 key ends up on each host. */
private static Map<ByteBuffer,List<ColumnOrSuperColumn>> insertBatch(Cassandra.Client client) throws Exception
{
final int N = 1000;
Column col1 = new Column(ByteBufferUtil.bytes("c1"))
.setValue(ByteBufferUtil.bytes("v1"))
.setTimestamp(0);
Column col2 = new Column(ByteBufferUtil.bytes("c2"))
.setValue(ByteBufferUtil.bytes("v2"))
.setTimestamp(0);
// build N rows
Map<ByteBuffer,List<ColumnOrSuperColumn>> rows = new HashMap<ByteBuffer, List<ColumnOrSuperColumn>>();
Map<ByteBuffer,Map<String,List<Mutation>>> batch = new HashMap<ByteBuffer,Map<String,List<Mutation>>>();
for (int i = 0; i < N; i++)
{
String rawKey = String.format("test.key.%d", i);
ByteBuffer key = ByteBufferUtil.bytes(rawKey);
Mutation m1 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col1));
Mutation m2 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col2));
rows.put(key, Arrays.asList(m1.getColumn_or_supercolumn(),
m2.getColumn_or_supercolumn()));
// add row to batch
Map<String,List<Mutation>> rowmap = new HashMap<String,List<Mutation>>();
rowmap.put(STANDARD_CF, Arrays.asList(m1, m2));
batch.put(key, rowmap);
}
// insert the batch
client.batch_mutate(batch, ConsistencyLevel.ONE);
return rows;
}
private static void verifyBatch(Cassandra.Client client, Map<ByteBuffer,List<ColumnOrSuperColumn>> batch) throws Exception
{
for (Map.Entry<ByteBuffer,List<ColumnOrSuperColumn>> entry : batch.entrySet())
{
// verify slice
SlicePredicate sp = new SlicePredicate();
sp.setSlice_range(
new SliceRange(
ByteBuffer.wrap(new byte[0]),
ByteBuffer.wrap(new byte[0]),
false,
1000
)
);
assertEquals(client.get_slice(entry.getKey(), STANDARD, sp, ConsistencyLevel.ONE),
entry.getValue());
}
}
@Test
public void testLoadbalance() throws Exception
{
final String keyspace = "TestLoadbalance";
addKeyspace(keyspace, 1);
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
client.set_keyspace(keyspace);
// add keys to each node
Map<ByteBuffer,List<ColumnOrSuperColumn>> rows = insertBatch(client);
Thread.sleep(100);
// ask a node to move to a new location
controller.nodetool("loadbalance", hosts.get(0));
// trigger cleanup on all nodes
for (InetAddress host : hosts)
controller.nodetool("cleanup", host);
// check that all keys still exist
verifyBatch(client, rows);
}
}

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

@ -1,188 +0,0 @@
/**
* 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;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.client.RingCache;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.thrift.TException;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class MutationTest extends TestBase
{
private static final Logger logger = LoggerFactory.getLogger(MutationTest.class);
@Test
public void testInsert() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
final String keyspace = "TestInsert";
addKeyspace(keyspace, 3);
Cassandra.Client client = controller.createClient(hosts.get(0));
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE);
insert(client, key, "Standard1", "c2", "v2", 0, ConsistencyLevel.ONE);
// block until the column is available
new Get(client, "Standard1", key).name("c1").value("v1").perform(ConsistencyLevel.ONE);
new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.ONE);
List<ColumnOrSuperColumn> coscs = get_slice(client, key, "Standard1", ConsistencyLevel.ONE);
assertColumnEqual("c1", "v1", 0, coscs.get(0).column);
assertColumnEqual("c2", "v2", 0, coscs.get(1).column);
}
@Test
public void testWriteAllReadOne() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteAllReadOne";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ALL);
// should be instantly available
assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE));
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size()));
try {
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
new Get(client, "Standard1", key).name("c1").value("v1")
.perform(ConsistencyLevel.ONE);
new Insert(client, "Standard1", key).name("c3").value("v3")
.expecting(UnavailableException.class).perform(ConsistencyLevel.ALL);
} finally {
failure.resolve();
Thread.sleep(10000);
}
}
@Test
public void testWriteQuorumReadQuorum() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteQuorumReadQuorum";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
// with quorum-1 nodes up
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size())); //kill all but one nodes
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
try {
new Insert(client, "Standard1", key).name("c1").value("v1")
.expecting(UnavailableException.class).perform(ConsistencyLevel.QUORUM);
} finally {
failure.resolve();
Thread.sleep(10000);
}
// with all nodes up
new Insert(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM);
failure = controller.failHosts(endpoints.get(0));
try {
new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM);
} finally {
failure.resolve();
Thread.sleep(10000);
}
}
@Test
public void testWriteOneReadAll() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteOneReadAll";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE);
assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ALL));
// with each of HH, read repair and proactive repair:
// with one node up
// write with one (success)
// read with all (failure)
// bring nodes up
// repair
// read with all (success)
Failure failure = controller.failHosts(endpoints);
try {
new Insert(client, "Standard1", key).name("c2").value("v2")
.expecting(UnavailableException.class).perform(ConsistencyLevel.ONE);
} finally {
failure.resolve();
}
}
protected ByteBuffer newKey()
{
return ByteBufferUtil.bytes(String.format("test.key.%d", System.currentTimeMillis()));
}
}

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

@ -1,329 +0,0 @@
/**
* 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;
import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.thrift.TException;
import org.apache.cassandra.client.*;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.*;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public abstract class TestBase
{
private static final Logger logger = LoggerFactory.getLogger(TestBase.class);
protected static CassandraServiceController controller =
CassandraServiceController.getInstance();
static class KeyspaceCreation
{
private String name;
private int rf;
private CfDef cfdef;
public KeyspaceCreation(String name)
{
this.name = name;
cfdef = new CfDef(name, "Standard1");
cfdef.setComparator_type("BytesType");
cfdef.setKey_cache_size(10000);
cfdef.setRow_cache_size(1000);
cfdef.setRow_cache_save_period_in_seconds(0);
cfdef.setKey_cache_save_period_in_seconds(3600);
cfdef.setMemtable_throughput_in_mb(255);
cfdef.setMemtable_operations_in_millions(0.29);
}
public KeyspaceCreation validator(String validator)
{
cfdef.setDefault_validation_class(validator);
return this;
}
public KeyspaceCreation rf(int rf)
{
this.rf = rf;
return this;
}
public void create() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
Map<String,String> stratOptions = new HashMap<String,String>();
stratOptions.put("replication_factor", "" + rf);
client.system_add_keyspace(new KsDef(name,
"org.apache.cassandra.locator.SimpleStrategy",
Arrays.asList(cfdef))
.setStrategy_options(stratOptions));
// poll, until KS added
for (InetAddress host : hosts)
{
try
{
client = controller.createClient(host);
poll:
while (true)
{
List<KsDef> ksDefList = client.describe_keyspaces();
for (KsDef ks : ksDefList)
{
if (ks.name.equals(name))
break poll;
}
try
{
Thread.sleep(1000);
}
catch (InterruptedException e)
{
break poll;
}
}
}
catch (TException te)
{
continue;
}
}
}
}
protected static KeyspaceCreation keyspace(String name)
{
return new KeyspaceCreation(name);
}
protected static void addKeyspace(String name, int rf) throws Exception
{
keyspace(name).rf(rf).create();
}
@BeforeClass
public static void setUp() throws Exception
{
controller.ensureClusterRunning();
}
protected ByteBuffer newKey()
{
return ByteBuffer.wrap(String.format("test.key.%d", System.currentTimeMillis()).getBytes());
}
protected void insert(Cassandra.Client client, ByteBuffer key, String cf, String name, String value, long timestamp, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
Column col = new Column(ByteBuffer.wrap(name.getBytes()))
.setValue(ByteBuffer.wrap(value.getBytes()))
.setTimestamp(timestamp);
client.insert(key, new ColumnParent(cf), col, cl);
}
protected Column getColumn(Cassandra.Client client, ByteBuffer key, String cf, String col, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException
{
ColumnPath cpath = new ColumnPath(cf);
cpath.setColumn(col.getBytes());
return client.get(key, cpath, cl).column;
}
protected class Get extends RetryingAction<String>
{
public Get(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
assertColumnEqual(name, value, timestamp, getColumn(client, key, cf, name, cl));
}
}
protected class Insert extends RetryingAction<String>
{
public Insert(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
insert(client, key, cf, name, value, timestamp, cl);
}
}
/** Performs an action repeatedly until timeout, success or failure. */
protected abstract class RetryingAction<T>
{
protected Cassandra.Client client;
protected String cf;
protected ByteBuffer key;
protected String name;
protected T value;
protected long timestamp;
private Set<Class<Exception>> expected = new HashSet<Class<Exception>>();
private long timeout = StorageService.RING_DELAY * 2;
public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key)
{
this.client = client;
this.cf = cf;
this.key = key;
this.timestamp = 0;
}
public RetryingAction name(String name)
{
this.name = name; return this;
}
/** A parameterized value for the action. */
public RetryingAction value(T value)
{
this.value = value; return this;
}
/** The total time to allow before failing. */
public RetryingAction timeout(long timeout)
{
this.timeout = timeout; return this;
}
/** The expected timestamp of the returned column. */
public RetryingAction timestamp(long timestamp)
{
this.timestamp = timestamp; return this;
}
/** The exception classes that indicate success. */
public RetryingAction expecting(Class... tempExceptions)
{
this.expected.clear();
for (Class exclass : tempExceptions)
expected.add((Class<Exception>)exclass);
return this;
}
public void perform(ConsistencyLevel cl) throws AssertionError
{
long deadline = System.currentTimeMillis() + timeout;
int attempts = 0;
String template = "%s for " + this + " after %d attempt(s) with %d ms to spare.";
Exception e = null;
while(deadline > System.currentTimeMillis())
{
try
{
attempts++;
tryPerformAction(cl);
logger.info(String.format(template, "Succeeded", attempts, deadline - System.currentTimeMillis()));
return;
}
catch (Exception ex)
{
e = ex;
if (!expected.contains(ex.getClass()))
continue;
logger.info(String.format(template, "Caught expected exception: " + e, attempts, deadline - System.currentTimeMillis()));
return;
}
}
String err = String.format(template, "Caught unexpected: " + e, attempts, deadline - System.currentTimeMillis());
logger.error(err, e);
throw new AssertionError(err);
}
public String toString()
{
return this.getClass().getSimpleName() + "(" + key + "," + name + ")";
}
protected abstract void tryPerformAction(ConsistencyLevel cl) throws Exception;
}
protected List<ColumnOrSuperColumn> get_slice(Cassandra.Client client, ByteBuffer key, String cf, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
SlicePredicate sp = new SlicePredicate();
sp.setSlice_range(
new SliceRange(
ByteBuffer.wrap(new byte[0]),
ByteBuffer.wrap(new byte[0]),
false,
1000
)
);
return client.get_slice(key, new ColumnParent(cf), sp, cl);
}
protected void assertColumnEqual(String name, String value, long timestamp, Column col)
{
assertEquals(ByteBuffer.wrap(name.getBytes()), col.name);
assertEquals(ByteBuffer.wrap(value.getBytes()), col.value);
assertEquals(timestamp, col.timestamp);
}
protected List<InetAddress> endpointsForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
Configuration conf = new Configuration();
RingCache ring = new RingCache(keyspace, new RandomPartitioner(), seed.getHostAddress(), 9160);
List<InetAddress> privateendpoints = ring.getEndpoint(key);
List<InetAddress> endpoints = new ArrayList<InetAddress>();
for (InetAddress endpoint : privateendpoints)
{
endpoints.add(controller.getPublicHost(endpoint));
}
return endpoints;
}
protected InetAddress nonEndpointForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
List<InetAddress> endpoints = endpointsForKey(seed, key, keyspace);
for (InetAddress host : controller.getHosts())
{
if (!endpoints.contains(host))
{
return host;
}
}
return null;
}
}

163
test/distributed/org/apache/cassandra/utils/BlobUtils.java
View File

@ -1,163 +0,0 @@
/**
* 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.utils;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URI;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import com.google.common.io.Files;
import org.apache.commons.configuration.Configuration;
import org.apache.whirr.service.ClusterSpec;
import org.jclouds.blobstore.BlobStoreContext;
import org.jclouds.blobstore.BlobStoreContextFactory;
import org.jclouds.blobstore.InputStreamMap;
import org.jclouds.blobstore.domain.BlobMetadata;
import org.jclouds.s3.S3AsyncClient;
import org.jclouds.s3.S3Client;
import org.jclouds.s3.domain.AccessControlList;
import org.jclouds.s3.domain.CannedAccessPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public final class BlobUtils
{
private static final Logger LOG = LoggerFactory.getLogger(BlobUtils.class);
public static final String BLOB_PROVIDER = "whirr.blobstore.provider";
public static final String BLOB_CONTAINER = "whirr.blobstore.container";
private static BlobStoreContext getContext(Configuration config, ClusterSpec spec)
{
return new BlobStoreContextFactory().createContext(getProvider(config), spec.getIdentity(), spec.getCredential());
}
private static String getProvider(Configuration config)
{
String provider = config.getString(BLOB_PROVIDER, null);
if (provider == null)
throw new RuntimeException("Please set " + BLOB_PROVIDER + " to a jclouds supported provider.");
return provider;
}
private static String getContainer(Configuration config)
{
String container = config.getString(BLOB_CONTAINER, null);
if (container == null)
throw new RuntimeException("Please set " + BLOB_CONTAINER + " to an existing container for your chosen provider.");
return container;
}
/**
* Stores the given local file as a public blob, and returns metadata for the blob.
*/
public static Pair<BlobMetadata,URI> storeBlob(Configuration config, ClusterSpec spec, String filename)
{
File file = new File(filename);
String container = getContainer(config);
String provider = getProvider(config);
// blob name and checksum of the file
String blobName = System.nanoTime() + "/" + file.getName();
String blobNameChecksum = blobName + ".md5";
BlobStoreContext context = getContext(config, spec);
File checksumFile;
try
{
checksumFile = File.createTempFile("dtchecksum", "md5");
checksumFile.deleteOnExit();
FileWriter checksumWriter = new FileWriter(checksumFile);
String checksum = FBUtilities.bytesToHex(Files.getDigest(file, MessageDigest.getInstance("MD5")));
checksumWriter.write(String.format("%s %s", checksum, file.getName()));
checksumWriter.close();
}
catch (IOException e)
{
throw new RuntimeException("Can't create a checksum of the file: " + filename);
}
catch (NoSuchAlgorithmException e)
{
throw new RuntimeException(e.getMessage());
}
try
{
InputStreamMap map = context.createInputStreamMap(container);
map.putFile(blobName, file);
map.putFile(blobNameChecksum, checksumFile);
// TODO: magic! in order to expose the blob as public, we need to dive into provider specific APIs
// the hope is that permissions are encapsulated in jclouds in the future
if (provider.contains("s3"))
{
S3Client sss = context.<S3Client,S3AsyncClient>getProviderSpecificContext().getApi();
String ownerId = sss.getObjectACL(container, blobName).getOwner().getId();
sss.putObjectACL(container,
blobName,
AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId));
sss.putObjectACL(container,
blobNameChecksum,
AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId));
}
else
{
LOG.warn(provider + " may not be properly supported for tarball transfer.");
}
// resolve the full URI of the blob (see http://code.google.com/p/jclouds/issues/detail?id=431)
BlobMetadata blob = context.getBlobStore().blobMetadata(container, blobName);
URI uri = context.getProviderSpecificContext().getEndpoint().resolve("/" + container + "/" + blob.getName());
return new Pair<BlobMetadata, URI>(blob, uri);
}
finally
{
context.close();
}
}
public static void deleteBlob(Configuration config, ClusterSpec spec, BlobMetadata blob)
{
String container = getContainer(config);
BlobStoreContext context = getContext(config, spec);
try
{
context.getBlobStore().removeBlob(container, blob.getName());
}
finally
{
context.close();
}
}
}

49
test/distributed/org/apache/cassandra/utils/KeyPair.java
View File

@ -1,49 +0,0 @@
/**
* 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.utils;
import java.io.ByteArrayOutputStream;
import java.util.Map;
import com.google.common.collect.ImmutableMap;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
/**
* A convenience class for generating an RSA key pair.
*/
public class KeyPair {
/**
* return a "public" -> rsa public key, "private" -> its corresponding
* private key
*/
public static Map<String,String> generate() throws JSchException {
com.jcraft.jsch.KeyPair pair = com.jcraft.jsch.KeyPair.genKeyPair(
new JSch(), com.jcraft.jsch.KeyPair.RSA);
ByteArrayOutputStream publicKeyOut = new ByteArrayOutputStream();
ByteArrayOutputStream privateKeyOut = new ByteArrayOutputStream();
pair.writePublicKey(publicKeyOut, "whirr");
pair.writePrivateKey(privateKeyOut);
String publicKey = new String(publicKeyOut.toByteArray());
String privateKey = new String(privateKeyOut.toByteArray());
return ImmutableMap.<String, String> of("public", publicKey,
"private", privateKey);
}
}