From 9694494e7f3dc184edb2d3d583ed48c6f0d7a711 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 5 Jan 2011 20:16:14 +0000 Subject: [PATCH] Distributed test harness. Patch by Kelvin Kakugawa, Stu Hood, and Ryan King, reviewed by brandonwilliams for CASSANDRA-1859. git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1055618 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + build.xml | 47 ++- ivysettings.xml | 17 +- test/distributed/README.txt | 56 +++ test/distributed/ivy.xml | 25 ++ .../cassandra/CassandraServiceController.java | 333 ++++++++++++++++++ .../org/apache/cassandra/MovementTest.java | 128 +++++++ .../org/apache/cassandra/MutationTest.java | 266 ++++++++++++++ .../org/apache/cassandra/TestBase.java | 114 ++++++ .../org/apache/cassandra/utils/BlobUtils.java | 122 +++++++ .../org/apache/cassandra/utils/KeyPair.java | 49 +++ test/resources/whirr-default.properties | 21 ++ 12 files changed, 1168 insertions(+), 11 deletions(-) create mode 100644 test/distributed/README.txt create mode 100644 test/distributed/ivy.xml create mode 100644 test/distributed/org/apache/cassandra/CassandraServiceController.java create mode 100644 test/distributed/org/apache/cassandra/MovementTest.java create mode 100644 test/distributed/org/apache/cassandra/MutationTest.java create mode 100644 test/distributed/org/apache/cassandra/TestBase.java create mode 100644 test/distributed/org/apache/cassandra/utils/BlobUtils.java create mode 100644 test/distributed/org/apache/cassandra/utils/KeyPair.java create mode 100644 test/resources/whirr-default.properties diff --git a/CHANGES.txt b/CHANGES.txt index 73861527d9..6a7a7e59e3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -12,6 +12,7 @@ dev * 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 a6758a3127..555d93f898 100644 --- a/build.xml +++ b/build.xml @@ -40,12 +40,14 @@ + + @@ -105,6 +107,7 @@ + @@ -165,10 +168,17 @@ + + + + + + @@ -453,28 +463,49 @@ + + + + + + + + + + + + + + + - + - + + + + @@ -485,13 +516,23 @@ - + + + + + + + + + + + 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/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