diff --git a/build.xml b/build.xml index 8bc9dc7392..a25e5db446 100644 --- a/build.xml +++ b/build.xml @@ -60,7 +60,6 @@ - @@ -287,9 +286,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - @@ -972,22 +968,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - - - - - - - - - - - @@ -1076,14 +1056,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} timeout="${test.long.timeout}" /> - - - - - - - - diff --git a/test/distributed/README.txt b/test/distributed/README.txt deleted file mode 100644 index 651038b77c..0000000000 --- a/test/distributed/README.txt +++ /dev/null @@ -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 - - diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java deleted file mode 100644 index 10fdef5884..0000000000 --- a/test/distributed/org/apache/cassandra/CassandraServiceController.java +++ /dev/null @@ -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 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 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.8"); - } - - service = 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()); - } - - 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 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 hosts, String functionName, String... functionArgs) - { - final Set hostset = new HashSet(); - - for (InetAddress host : hosts) - hostset.add(host.getHostAddress()); - - StatementBuilder statementBuilder = new StatementBuilder(); - statementBuilder.addStatement(Statements.call(functionName, functionArgs)); - Credentials credentials = new Credentials(clusterSpec.getClusterUser(), clusterSpec.getPrivateKey()); - - Map results; - try - { - results = computeService.runScriptOnNodesMatching(new Predicate() - { - public boolean apply(NodeMetadata node) - { - Set intersection = new HashSet(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 getHosts() - { - return hosts; - } - - class Failure - { - private List hosts; - - public Failure(List 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); - } -} diff --git a/test/distributed/org/apache/cassandra/CountersTest.java b/test/distributed/org/apache/cassandra/CountersTest.java deleted file mode 100644 index 71e635d367..0000000000 --- a/test/distributed/org/apache/cassandra/CountersTest.java +++ /dev/null @@ -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 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 - { - 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); - } -} diff --git a/test/distributed/org/apache/cassandra/MovementTest.java b/test/distributed/org/apache/cassandra/MovementTest.java deleted file mode 100644 index 68d9086318..0000000000 --- a/test/distributed/org/apache/cassandra/MovementTest.java +++ /dev/null @@ -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> 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> rows = new HashMap>(); - Map>> batch = new HashMap>>(); - 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> 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 deleted file mode 100644 index ab93fd3d4c..0000000000 --- a/test/distributed/org/apache/cassandra/MutationTest.java +++ /dev/null @@ -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 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 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); - // should be instantly available - 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())); - - 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 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 - - 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 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); - 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())); - } -} diff --git a/test/distributed/org/apache/cassandra/TestBase.java b/test/distributed/org/apache/cassandra/TestBase.java deleted file mode 100644 index e182270391..0000000000 --- a/test/distributed/org/apache/cassandra/TestBase.java +++ /dev/null @@ -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 hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - Map stratOptions = new HashMap(); - 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 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 - { - 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 - { - 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 - { - protected Cassandra.Client client; - protected String cf; - protected ByteBuffer key; - protected String name; - protected T value; - protected long timestamp; - - private Set> expected = new HashSet>(); - 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)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 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 - { - Configuration conf = new Configuration(); - - 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; - } -} diff --git a/test/distributed/org/apache/cassandra/utils/BlobUtils.java b/test/distributed/org/apache/cassandra/utils/BlobUtils.java deleted file mode 100644 index ddcbfa5fc1..0000000000 --- a/test/distributed/org/apache/cassandra/utils/BlobUtils.java +++ /dev/null @@ -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 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.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(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 deleted file mode 100644 index a8d9049027..0000000000 --- a/test/distributed/org/apache/cassandra/utils/KeyPair.java +++ /dev/null @@ -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 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); - } -}