diff --git a/CHANGES.txt b/CHANGES.txt
index 0cd72d398e..7a5f4d21c3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -17,6 +17,7 @@
* implement describeOwnership for BOP, COPP (CASSANDRA-1928)
* make read repair behave as expected for ConsistencyLevel > ONE
(CASSANDRA-982)
+ * distributed test harness (CASSANDRA-1859)
0.7.0-rc4
diff --git a/build.xml b/build.xml
index 5e42c358cf..b4b01c6ca2 100644
--- a/build.xml
+++ b/build.xml
@@ -41,12 +41,14 @@
+
+
@@ -108,6 +110,7 @@
+
@@ -189,10 +192,17 @@
+
+
+
+
+
+
@@ -479,28 +489,49 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
-
+
+
+
+
@@ -511,8 +542,9 @@
-
+
+
+
@@ -520,6 +552,14 @@
timeout="${test.long.timeout}" />
+
+
+
+
+
+
+
+
diff --git a/ivysettings.xml b/ivysettings.xml
index 12dddd4d35..6f53f1c775 100644
--- a/ivysettings.xml
+++ b/ivysettings.xml
@@ -17,17 +17,18 @@
~ under the License.
-->
-
+
-
-
-
+
+
+
+
+
+
+
+
-
-
-
-
diff --git a/lib/high-scale-lib.jar b/lib/high-scale-lib.jar
index 421a436eed..fbeec1aa3c 100644
Binary files a/lib/high-scale-lib.jar and b/lib/high-scale-lib.jar differ
diff --git a/test/distributed/README.txt b/test/distributed/README.txt
new file mode 100644
index 0000000000..44e1ab9107
--- /dev/null
+++ b/test/distributed/README.txt
@@ -0,0 +1,56 @@
+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.provider=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=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
+
+
diff --git a/test/distributed/ivy.xml b/test/distributed/ivy.xml
new file mode 100644
index 0000000000..b0baecc69a
--- /dev/null
+++ b/test/distributed/ivy.xml
@@ -0,0 +1,25 @@
+
+
+
+
+
+
+
+
diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java
new file mode 100644
index 0000000000..f597b3ee3a
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/CassandraServiceController.java
@@ -0,0 +1,333 @@
+/**
+ * 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.net.URI;
+import java.util.*;
+
+import org.apache.cassandra.thrift.Cassandra;
+import org.apache.cassandra.thrift.TokenRange;
+import org.apache.cassandra.utils.KeyPair;
+import org.apache.cassandra.utils.BlobUtils;
+import org.apache.cassandra.utils.Pair;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.CompositeConfiguration;
+import org.apache.commons.configuration.PropertiesConfiguration;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.*;
+import org.apache.thrift.transport.*;
+
+import org.apache.whirr.service.Cluster;
+import org.apache.whirr.service.Cluster.Instance;
+import org.apache.whirr.service.ClusterSpec;
+import org.apache.whirr.service.ComputeServiceContextBuilder;
+import org.apache.whirr.service.Service;
+import org.apache.whirr.service.ServiceFactory;
+import org.apache.whirr.service.cassandra.CassandraService;
+import org.apache.whirr.service.cassandra.CassandraClusterActionHandler;
+import org.apache.whirr.service.jclouds.RunUrlStatement;
+
+import org.jclouds.blobstore.domain.BlobMetadata;
+
+import org.jclouds.compute.ComputeService;
+import org.jclouds.compute.domain.NodeMetadata;
+import org.jclouds.compute.options.RunScriptOptions;
+import org.jclouds.domain.Credentials;
+import org.jclouds.io.Payload;
+import org.jclouds.scriptbuilder.domain.OsFamily;
+import org.jclouds.ssh.ExecResponse;
+import static org.jclouds.io.Payloads.newStringPayload;
+
+import com.google.common.base.Predicate;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.Assert.assertThat;
+
+public class CassandraServiceController
+{
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CassandraServiceController.class);
+
+ protected static int CLIENT_PORT = 9160;
+ protected static int JMX_PORT = 8080;
+
+ private static final CassandraServiceController INSTANCE =
+ new CassandraServiceController();
+
+ public static CassandraServiceController getInstance()
+ {
+ return INSTANCE;
+ }
+
+ private boolean running;
+
+ private ClusterSpec clusterSpec;
+ private CassandraService service;
+ private Cluster cluster;
+ private ComputeService computeService;
+ private Credentials credentials;
+ private CompositeConfiguration config;
+ private BlobMetadata tarball;
+ private List hosts;
+
+ private CassandraServiceController()
+ {
+ }
+
+ public Cassandra.Client createClient(InetAddress addr)
+ throws TTransportException, 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)
+ {
+ 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 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 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.7");
+ }
+
+ service = (CassandraService)new ServiceFactory().create(clusterSpec.getServiceName());
+ cluster = service.launchCluster(clusterSpec);
+ computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService();
+ hosts = new ArrayList();
+ for (Instance instance : cluster.getInstances())
+ {
+ hosts.add(instance.getPublicAddress());
+ credentials = instance.getLoginCredentials();
+ }
+
+ waitForClusterInitialization();
+
+ ShutdownHook shutdownHook = new ShutdownHook(this);
+ Runtime.getRuntime().addShutdownHook(shutdownHook);
+
+ running = true;
+ }
+
+ public synchronized void shutdown()
+ {
+ // catch and log errors, we're in a runtime shutdown hook
+ try
+ {
+ LOG.info("Shutting down cluster...");
+ if (service != null)
+ service.destroyCluster(clusterSpec);
+ if (tarball != null)
+ BlobUtils.deleteBlob(config, clusterSpec, tarball);
+ running = false;
+ }
+ catch (Exception e)
+ {
+ LOG.error(String.format("Error shutting down cluster: %s", 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(String.format("apache/cassandra/nodetool %s", args), hosts);
+ }
+
+ /**
+ * Wipes all persisted state for the given node, leaving it as if it had just started.
+ */
+ public void wipeHosts(InetAddress... hosts)
+ {
+ callOnHosts("apache/cassandra/wipe-state", hosts);
+ }
+
+ public Failure failHosts(List hosts)
+ {
+ return new Failure(hosts.toArray(new InetAddress[hosts.size()])).trigger();
+ }
+
+ public Failure failHosts(InetAddress... hosts)
+ {
+ return new Failure(hosts).trigger();
+ }
+
+ /** TODO: Move to CassandraService? */
+ protected void callOnHosts(String payload, InetAddress... hosts)
+ {
+ final Set hostset = new HashSet();
+ for (InetAddress host : hosts)
+ hostset.add(host.getHostAddress());
+ Map results;
+ try
+ {
+ results = computeService.runScriptOnNodesMatching(new Predicate()
+ {
+ public boolean apply(NodeMetadata node)
+ {
+ Set intersection = new HashSet(hostset);
+ intersection.retainAll(node.getPublicAddresses());
+ return !intersection.isEmpty();
+ }
+ }, newStringPayload(new RunUrlStatement(clusterSpec.getRunUrlBase(), payload).render(OsFamily.UNIX)),
+ RunScriptOptions.Builder.overrideCredentialsWith(credentials));
+ }
+ catch (Exception 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());
+ }
+
+ public List getHosts()
+ {
+ return hosts;
+ }
+
+ class Failure
+ {
+ private InetAddress[] hosts;
+
+ public Failure(InetAddress... hosts)
+ {
+ this.hosts = hosts;
+ }
+
+ public Failure trigger()
+ {
+ callOnHosts("apache/cassandra/stop", hosts);
+ return this;
+ }
+
+ public void resolve()
+ {
+ callOnHosts("apache/cassandra/start", hosts);
+ 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);
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/MovementTest.java b/test/distributed/org/apache/cassandra/MovementTest.java
new file mode 100644
index 0000000000..5577e2be9b
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/MovementTest.java
@@ -0,0 +1,128 @@
+/**
+ * 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.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> insertBatch(Cassandra.Client client) throws Exception
+ {
+ final int N = 1000;
+ Column col1 = new Column(
+ ByteBuffer.wrap("c1".getBytes()),
+ ByteBuffer.wrap("v1".getBytes()),
+ 0
+ );
+ Column col2 = new Column(
+ ByteBuffer.wrap("c2".getBytes()),
+ ByteBuffer.wrap("v2".getBytes()),
+ 0
+ );
+
+ // build N rows
+ Map> rows = new HashMap>();
+ Map>> batch = new HashMap>>();
+ for (int i = 0; i < N; i++)
+ {
+ String rawKey = String.format("test.key.%d", i);
+ ByteBuffer key = ByteBuffer.wrap(rawKey.getBytes());
+ 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> rowmap = new HashMap>();
+ 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> batch) throws Exception
+ {
+ for (Map.Entry> 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 hosts = controller.getHosts();
+ Cassandra.Client client = controller.createClient(hosts.get(0));
+ client.set_keyspace(keyspace);
+
+ // add keys to each node
+ Map> 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);
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/MutationTest.java b/test/distributed/org/apache/cassandra/MutationTest.java
new file mode 100644
index 0000000000..42bc4bd7a4
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/MutationTest.java
@@ -0,0 +1,266 @@
+/**
+ * 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.IOException;
+import java.io.Writer;
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.ArrayList;
+
+import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.thrift.TException;
+import org.apache.cassandra.client.*;
+import org.apache.cassandra.dht.RandomPartitioner;
+
+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
+{
+ @Test
+ public void testInsert() throws Exception
+ {
+ List 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);
+
+
+ assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE));
+ assertColumnEqual("c2", "v2", 0, getColumn(client, key, "Standard1", "c2", ConsistencyLevel.ONE));
+
+ List 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 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);
+ assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE));
+
+ List endpoints = endpointsForKey(hosts.get(0), key, keyspace);
+ InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
+ Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size()));
+
+ Thread.sleep(10000); // let gossip catch up
+
+ try {
+ client = controller.createClient(coordinator);
+ client.set_keyspace(keyspace);
+
+ assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE));
+
+ insert(client, key, "Standard1", "c3", "v3", 0, ConsistencyLevel.ALL);
+ assert false;
+ } catch (UnavailableException e) {
+ // [this is good]
+ } finally {
+ failure.resolve();
+ Thread.sleep(10000);
+ }
+ }
+
+ @Test
+ public void testWriteQuorumReadQuorum() throws Exception
+ {
+ List 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 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
+
+ Thread.sleep(10000);
+ client = controller.createClient(coordinator);
+ client.set_keyspace(keyspace);
+ try {
+ insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.QUORUM);
+ assert false;
+ } catch (UnavailableException e) {
+ // [this is good]
+ } finally {
+ failure.resolve();
+ Thread.sleep(10000);
+ }
+
+ // with all nodes up
+ insert(client, key, "Standard1", "c2", "v2", 0, ConsistencyLevel.QUORUM);
+
+ failure = controller.failHosts(endpoints.get(0));
+ Thread.sleep(10000);
+ try {
+ getColumn(client, key, "Standard1", "c2", ConsistencyLevel.QUORUM);
+ } finally {
+ failure.resolve();
+ Thread.sleep(10000);
+ }
+ }
+
+ @Test
+ public void testWriteOneReadAll() throws Exception
+ {
+ List 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 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);
+ Thread.sleep(10000);
+ try {
+ insert(client, key, "Standard1", "c2", "v2", 0, ConsistencyLevel.ONE);
+ assert false;
+ } catch (UnavailableException e) {
+ // this is good
+ } finally {
+ failure.resolve();
+ }
+ }
+
+ 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()),
+ ByteBuffer.wrap(value.getBytes()),
+ 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 List 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 endpointsForKey(InetAddress seed, ByteBuffer key, String keyspace)
+ throws IOException
+ {
+ RingCache ring = new RingCache(keyspace, new RandomPartitioner(), seed.getHostAddress(), 9160);
+ List privateendpoints = ring.getEndpoint(key);
+ List endpoints = new ArrayList();
+ for (InetAddress endpoint : privateendpoints)
+ {
+ endpoints.add(controller.getPublicHost(endpoint));
+ }
+ return endpoints;
+ }
+
+ protected InetAddress nonEndpointForKey(InetAddress seed, ByteBuffer key, String keyspace)
+ throws IOException
+ {
+ List endpoints = endpointsForKey(seed, key, keyspace);
+ for (InetAddress host : controller.getHosts())
+ {
+ if (!endpoints.contains(host))
+ {
+ return host;
+ }
+ }
+ return null;
+ }
+
+ protected ByteBuffer newKey()
+ {
+ return ByteBuffer.wrap(String.format("test.key.%d", System.currentTimeMillis()).getBytes());
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/TestBase.java b/test/distributed/org/apache/cassandra/TestBase.java
new file mode 100644
index 0000000000..53c242cb0a
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/TestBase.java
@@ -0,0 +1,114 @@
+/**
+ * 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.util.LinkedList;
+import java.util.List;
+
+import org.apache.thrift.TException;
+
+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
+{
+ protected static CassandraServiceController controller =
+ CassandraServiceController.getInstance();
+
+ protected static void addKeyspace(String name, int rf) throws Exception
+ {
+ List cfDefList = new LinkedList();
+
+ CfDef standard1 = new CfDef(name, "Standard1");
+ standard1.setComparator_type("BytesType");
+ standard1.setKey_cache_size(10000);
+ standard1.setRow_cache_size(1000);
+ standard1.setRow_cache_save_period_in_seconds(0);
+ standard1.setKey_cache_save_period_in_seconds(3600);
+ standard1.setMemtable_flush_after_mins(59);
+ standard1.setMemtable_throughput_in_mb(255);
+ standard1.setMemtable_operations_in_millions(0.29);
+ cfDefList.add(standard1);
+
+ List hosts = controller.getHosts();
+ Cassandra.Client client = controller.createClient(hosts.get(0));
+
+ client.system_add_keyspace(
+ new KsDef(
+ name,
+ "org.apache.cassandra.locator.SimpleStrategy",
+ rf,
+ cfDefList));
+
+ // poll, until KS added
+ for (InetAddress host : hosts)
+ {
+ try
+ {
+ client = controller.createClient(host);
+ poll:
+ while (true)
+ {
+ List 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;
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void setUp() throws Exception
+ {
+ controller.ensureClusterRunning();
+ }
+
+ protected static String createTemporaryKey()
+ {
+ return String.format("test.key.%d", System.currentTimeMillis());
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/utils/BlobUtils.java b/test/distributed/org/apache/cassandra/utils/BlobUtils.java
new file mode 100644
index 0000000000..94474cf086
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/utils/BlobUtils.java
@@ -0,0 +1,122 @@
+/**
+ * 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.net.URI;
+
+import org.apache.commons.configuration.Configuration;
+
+import org.apache.whirr.service.ClusterSpec;
+
+import org.jclouds.blobstore.BlobStore;
+import org.jclouds.blobstore.BlobStoreContext;
+import org.jclouds.blobstore.BlobStoreContextFactory;
+import org.jclouds.blobstore.domain.BlobMetadata;
+import org.jclouds.blobstore.InputStreamMap;
+
+import org.jclouds.aws.s3.S3Client;
+import org.jclouds.aws.s3.S3AsyncClient;
+import org.jclouds.aws.s3.domain.AccessControlList;
+import org.jclouds.aws.s3.domain.CannedAccessPolicy;
+
+import org.jclouds.rest.RestContext;
+
+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 storeBlob(Configuration config, ClusterSpec spec, String filename)
+ {
+ File file = new File(filename);
+ String container = getContainer(config);
+ String provider = getProvider(config);
+ String blobName = System.nanoTime() + "/" + file.getName();
+ BlobStoreContext context = getContext(config, spec);
+ try
+ {
+ InputStreamMap map = context.createInputStreamMap(container);
+ map.putFile(blobName, file);
+ // 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.equals("s3"))
+ {
+ S3Client sss = context.getProviderSpecificContext().getApi();
+ String ownerId = sss.getObjectACL(container, blobName).getOwner().getId();
+ sss.putObjectACL(container,
+ blobName,
+ 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(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();
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/utils/KeyPair.java b/test/distributed/org/apache/cassandra/utils/KeyPair.java
new file mode 100644
index 0000000000..a8d9049027
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/utils/KeyPair.java
@@ -0,0 +1,49 @@
+/**
+ * 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 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. of("public", publicKey,
+ "private", privateKey);
+ }
+}
diff --git a/test/resources/whirr-default.properties b/test/resources/whirr-default.properties
new file mode 100644
index 0000000000..fc33c48122
--- /dev/null
+++ b/test/resources/whirr-default.properties
@@ -0,0 +1,21 @@
+#
+# 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.
+#
+whirr.service-name=cassandra
+whirr.cluster-name=cassandra_test
+whirr.instance-templates=4 cassandra
+whirr.version=0.3.0-incubating-SNAPSHOT