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

merge from 1.0 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1214927 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-12-15 19:34:40 +00:00
parent 16da34e1ba b681f4bc28
commit e99cd6253b
24 changed files with 154 additions and 1388 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.rat-excludes
View File

@ -29,3 +29,4 @@ drivers/txpy/txcql/cassandra/*
drivers/py/cql/cassandra/*
doc/cql/CQL*
build.properties.default
test/data/legacy-sstables/**

12
CHANGES.txt
View File

@ -27,7 +27,12 @@
* more efficient allocation of small bloom filters (CASSANDRA-3618)
1.0.7
* fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614)
1.0.6
* (CQL) fix cqlsh support for replicate_on_write (CASSANDRA-3596)
* fix adding to leveled manifest after streaming (CASSANDRA-3536)
* filter out unavailable cipher suites when using encryption (CASSANDRA-3178)
* (HADOOP) add old-style api support for CFIF and CFRR (CASSANDRA-2799)
@ -49,13 +54,20 @@
* add back partitioner to sstable metadata (CASSANDRA-3540)
* fix NPE in get_count for counters (CASSANDRA-3601)
Merged from 0.8:
* remove invalid assertion that table was opened before dropping it
(CASSANDRA-3580)
* range and index scans now only send requests to enough replicas to
satisfy requested CL + RR (CASSANDRA-3598)
* use cannonical host for local node in nodetool info (CASSANDRA-3556)
* remove nonlocal DC write optimization since it only worked with
CL.ONE or CL.LOCAL_QUORUM (CASSANDRA-3577, 3585)
* detect misuses of CounterColumnType (CASSANDRA-3422)
* turn off string interning in json2sstable, take 2 (CASSANDRA-2189)
* validate compression parameters on add/update of the ColumnFamily
(CASSANDRA-3573)
* Check for 0.0.0.0 is incorrect in CFIF (CASSANDRA-3584)
* Increase vm.max_map_count in debian packaging (CASSANDRA-3563)
* gossiper will never add itself to saved endpoints (CASSANDRA-3485)
1.0.5

25
NEWS.txt
View File

@ -33,15 +33,28 @@ Upgrading
want to use such large batches.
1.0.6
=====
Upgrading
---------
- This release fixes an issue related to the chunk_length_kb option for
compressed sstables. If you use compression on some column families, it
is recommended after the upgrade to check the value for this option on
these column families (the default value is 64). In case the option would
not be set correctly, you should update the column family definition,
setting the right value and then run scrub on the column family.
- Please report to instruction for 1.0.5 if coming from an older version.
1.0.5
=====
JMX
---
- A command has been added to stop running compaction. It is available
through JMX and through nodetool stop <type> (see the nodetool help for
details). Please note that stopped compaction are terminated and cannot
be restarted afterwards.
Upgrading
---------
- 1.0.5 comes to fix two important regression of 1.0.4. So all information
concerning 1.0.4 are valid for this release, but please avoids upgrading
to 1.0.4.
1.0.4

31
build.xml
View File

@ -59,7 +59,6 @@
<property name="test.name" value="*Test"/>
<property name="test.unit.src" value="${test.dir}/unit"/>
<property name="test.long.src" value="${test.dir}/long"/>
<property name="test.distributed.src" value="${test.dir}/distributed"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<condition property="version" value="${base.version}">
<isset property="release"/>
@ -285,9 +284,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<artifact:remoteRepository id="central" url="${artifact.remoteRepository.central}"/>
<artifact:remoteRepository id="java.net2" url="${artifact.remoteRepository.java.net2}"/>
<artifact:remoteRepository id="apache" url="${artifact.remoteRepository.apache}"/>
<!-- for distributed tests -->
<artifact:remoteRepository id="jclouds" url="${artifact.remoteRepository.jclouds}"/>
<artifact:remoteRepository id="oauth" url="${artifact.remoteRepository.oauth}"/>
<macrodef name="install">
<attribute name="pomFile"/>
@ -969,22 +965,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
</copy>
</target>
<target name="build-distributed-test" depends="build-test,maven-ant-tasks-retrieve-test" description="Compile distributed test classes (which have additional deps)">
<javac
debug="true"
debuglevel="${debuglevel}"
destdir="${test.classes}">
<classpath>
<path refid="cassandra.classpath"/>
<pathelement location="${test.classes}"/>
<fileset dir="${test.lib}">
<include name="**/*.jar" />
</fileset>
</classpath>
<src path="${test.distributed.src}"/>
</javac>
</target>
<macrodef name="testmacro">
<attribute name="suitename" />
<attribute name="inputdir" />
@ -1036,6 +1016,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<testmacro suitename="unit" inputdir="${test.unit.src}" timeout="60000">
<jvmarg value="-Dlegacy-sstable-root=${test.data}/legacy-sstables"/>
<jvmarg value="-Dcorrupt-sstable-root=${test.data}/corrupt-sstables"/>
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
</testmacro>
</target>
@ -1044,6 +1025,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<jvmarg value="-Dlegacy-sstable-root=${test.data}/legacy-sstables"/>
<jvmarg value="-Dcorrupt-sstable-root=${test.data}/corrupt-sstables"/>
<jvmarg value="-Dcassandra.test.compression=true"/>
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
</testmacro>
</target>
@ -1070,14 +1052,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<target name="long-test" depends="build-test" description="Execute functional tests">
<testmacro suitename="long" inputdir="${test.long.src}"
timeout="${test.long.timeout}" />
</target>
<!-- Depends on artifacts so that we can push a tarball to remote nodes, and has its own build target for cloudy deps. -->
<target name="distributed-test" depends="build-distributed-test,artifacts" description="Execute distributed tests: see ${test.distributed.src}/README.txt">
<testmacro suitename="distributed" inputdir="${test.distributed.src}" timeout="1200000" forkmode="once">
<jvmarg value="-Dwhirr.config=${whirr.config}"/>
<jvmarg value="-Dwhirr.cassandra_tarball=${build.dir}/${final.name}-bin.tar.gz"/>
</testmacro>
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
</target>
<!-- instruments the classes to later create code coverage reports -->

1
debian/cassandra-sysctl.conf vendored Normal file
View File

@ -0,0 +1 @@
vm.max_map_count = 1048575

12
debian/changelog vendored
View File

@ -1,3 +1,15 @@
cassandra (1.0.6) unstable; urgency=low
* New release
-- Sylvain Lebresne <slebresne@apache.org> Sat, 10 Dec 2011 18:21:50 -0600
cassandra (1.0.5) unstable; urgency=low
* New release
-- Sylvain Lebresne <slebresne@apache.org> Tue, 29 Nov 2011 19:36:09 +0100
cassandra (1.0.4) unstable; urgency=low
* New release

2
pylib/cqlshlib/cqlhandling.py
View File

@ -37,7 +37,7 @@ columnfamily_options = (
('max_compaction_threshold', None),
('row_cache_save_period_in_seconds', None),
('key_cache_save_period_in_seconds', None),
('replication_on_write', 'replicate_on_write')
('replicate_on_write', None)
)
cql_type_to_apache_class = {

5
src/java/org/apache/cassandra/db/DataTracker.java
View File

@ -292,6 +292,11 @@ public class DataTracker
}
while (!view.compareAndSet(currentView, newView));
if (notCompacting.isEmpty())
{
// notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
return;
}
notifySSTablesChanged(notCompacting, Collections.<SSTableReader>emptySet());
postReplace(notCompacting, Collections.<SSTableReader>emptySet());
}

2
src/java/org/apache/cassandra/db/SystemTable.java
View File

@ -138,6 +138,8 @@ public class SystemTable
*/
public static synchronized void updateToken(InetAddress ep, Token token)
{
if (ep == FBUtilities.getLocalAddress())
return;
IPartitioner p = StorageService.getPartitioner();
ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, STATUS_CF);
cf.addColumn(new Column(p.getTokenFactory().toByteArray(token), ByteBuffer.wrap(ep.getAddress()), System.currentTimeMillis()));

3
src/java/org/apache/cassandra/db/migration/DropKeyspace.java
View File

@ -63,8 +63,7 @@ public class DropKeyspace extends Migration
}
// remove the table from the static instances.
Table table = Table.clear(ksm.name, schema);
assert table != null;
Table.clear(ksm.name, schema);
// reset defs.
schema.clearTableDefinition(ksm, newVersion);
}

70
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -377,9 +377,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
* @param endpoint - the endpoint being removed
* @param token - the token being removed
* @param mytoken - my own token for replication coordination
* @param delay
*/
public void advertiseRemoving(InetAddress endpoint, Token token, Token mytoken, int delay)
public void advertiseRemoving(InetAddress endpoint, Token token, Token mytoken)
{
EndpointState epState = endpointStateMap.get(endpoint);
// remember this node's generation
@ -388,7 +387,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
logger.info("Sleeping for " + StorageService.RING_DELAY + "ms to ensure " + endpoint + " does not change");
try
{
Thread.sleep(delay);
Thread.sleep(StorageService.RING_DELAY);
}
catch (InterruptedException e)
{
@ -432,6 +431,66 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
}
/**
* Do not call this method unless you know what you are doing.
* It will try extremely hard to obliterate any endpoint from the ring,
* even if it does not know about it.
* This should only ever be called by human via JMX.
* @param address
* @throws UnknownHostException
*/
public void unsafeAssassinateEndpoint(String address) throws UnknownHostException
{
InetAddress endpoint = InetAddress.getByName(address);
EndpointState epState = endpointStateMap.get(endpoint);
Token token = null;
logger.warn("Assassinating {} via gossip", endpoint);
if (epState == null)
{
epState = new EndpointState(new HeartBeatState((int)((System.currentTimeMillis() + 60000) / 1000), 9999));
}
else
{
try
{
token = StorageService.instance.getTokenMetadata().getToken(endpoint);
}
catch (AssertionError e)
{
}
int generation = epState.getHeartBeatState().getGeneration();
logger.info("Sleeping for " + StorageService.RING_DELAY + "ms to ensure " + endpoint + " does not change");
try
{
Thread.sleep(StorageService.RING_DELAY);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
// make sure it did not change
epState = endpointStateMap.get(endpoint);
if (epState.getHeartBeatState().getGeneration() != generation)
throw new RuntimeException("Endpoint " + endpoint + " generation changed while trying to remove it");
epState.updateTimestamp(); // make sure we don't evict it too soon
epState.getHeartBeatState().forceNewerGenerationUnsafe();
}
if (token == null)
token = StorageService.instance.getBootstrapToken();
// do not pass go, do not collect 200 dollars, just gtfo
epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.left(token, computeExpireTime()));
handleMajorStateChange(endpoint, epState);
try
{
Thread.sleep(intervalInMillis * 4);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
logger.warn("Finished killing {}", endpoint);
}
public boolean isKnownEndpoint(InetAddress endpoint)
{
return endpointStateMap.containsKey(endpoint);
@ -1035,6 +1094,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
*/
public void addSavedEndpoint(InetAddress ep)
{
if (ep == FBUtilities.getLocalAddress())
{
logger.debug("Attempt to add self as saved endpoint");
return;
}
EndpointState epState = new EndpointState(new HeartBeatState(0));
epState.markDead();
epState.setHasToken(true);

2
src/java/org/apache/cassandra/gms/GossiperMBean.java
View File

@ -28,4 +28,6 @@ public interface GossiperMBean
public int getCurrentGenerationNumber(String address) throws UnknownHostException;
public void unsafeAssassinateEndpoint(String address) throws UnknownHostException;
}

4
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -862,7 +862,7 @@ public class StorageProxy implements StorageProxyMBean
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
ReadCallback<Iterable<Row>> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints);
handler.assureSufficientLiveNodes();
for (InetAddress endpoint : liveEndpoints)
for (InetAddress endpoint : handler.endpoints)
{
MessagingService.instance().sendRR(c2, endpoint, handler);
if (logger.isDebugEnabled())
@ -1140,7 +1140,7 @@ public class StorageProxy implements StorageProxyMBean
IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range);
MessageProducer producer = new CachingMessageProducer(command);
for (InetAddress endpoint : liveEndpoints)
for (InetAddress endpoint : handler.endpoints)
{
MessagingService.instance().sendRR(producer, endpoint, handler);
if (logger.isDebugEnabled())

32
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -79,7 +79,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
{
private static Logger logger_ = LoggerFactory.getLogger(StorageService.class);
public static final int RING_DELAY = 30 * 1000; // delay after which we assume ring has stablized
public static final int RING_DELAY = getRingDelay(); // delay after which we assume ring has stablized
/* All verb handler identifiers */
public enum Verb
@ -150,6 +150,17 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
put(Verb.UNUSED_3, Stage.INTERNAL_RESPONSE);
}};
private static int getRingDelay()
{
String newdelay = System.getProperty("cassandra.ring_delay_ms");
if (newdelay != null)
{
logger_.warn("Overriding RING_DELAY to {}ms", newdelay);
return Integer.parseInt(newdelay);
}
else
return 30 * 1000;
}
/**
* This pool is used for periodic short (sub-second) tasks.
@ -404,8 +415,16 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
logger_.info("Loading persisted ring state");
for (Map.Entry<Token, InetAddress> entry : SystemTable.loadTokens().entrySet())
{
tokenMetadata_.updateNormalToken(entry.getKey(), entry.getValue());
Gossiper.instance.addSavedEndpoint(entry.getValue());
if (entry.getValue() == FBUtilities.getLocalAddress())
{
// entry has been mistakenly added, delete it
SystemTable.removeToken(entry.getKey());
}
else
{
tokenMetadata_.updateNormalToken(entry.getKey(), entry.getValue());
Gossiper.instance.addSavedEndpoint(entry.getValue());
}
}
}
@ -2330,11 +2349,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
* @param tokenString token for the node
*/
public void removeToken(String tokenString)
{
removeToken(tokenString, RING_DELAY);
}
public void removeToken(String tokenString, int delay)
{
InetAddress myAddress = FBUtilities.getBroadcastAddress();
Token localToken = tokenMetadata_.getToken(myAddress);
@ -2382,7 +2396,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
calculatePendingRanges();
// the gossiper will handle spoofing this node's state to REMOVING_TOKEN for us
// we add our own token so other nodes to let us know when they're done
Gossiper.instance.advertiseRemoving(endpoint, token, localToken, delay);
Gossiper.instance.advertiseRemoving(endpoint, token, localToken);
// kick off streaming commands
restoreReplicaCount(endpoint, myAddress);

3
test/cassandra.in.sh
View File

@ -50,4 +50,5 @@ JVM_OPTS=" \
-XX:+HeapDumpOnOutOfMemoryError \
-Dcom.sun.management.jmxremote.port=8090 \
-Dcom.sun.management.jmxremote.ssl=false \
-Dcom.sun.management.jmxremote.authenticate=false"
-Dcom.sun.management.jmxremote.authenticate=false \
-Dcassandra.ring_delay_ms=1000"

57
test/distributed/README.txt
View File

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

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

@ -1,336 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra;
import java.net.InetAddress;
import java.net.URI;
import java.util.*;
import com.google.common.base.Predicate;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.utils.BlobUtils;
import org.apache.cassandra.utils.KeyPair;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.whirr.service.*;
import org.apache.whirr.service.Cluster.Instance;
import org.apache.whirr.service.cassandra.CassandraClusterActionHandler;
import org.apache.whirr.service.jclouds.StatementBuilder;
import org.jclouds.blobstore.domain.BlobMetadata;
import org.jclouds.compute.ComputeService;
import org.jclouds.compute.domain.ExecResponse;
import org.jclouds.compute.domain.NodeMetadata;
import org.jclouds.compute.options.RunScriptOptions;
import org.jclouds.compute.RunScriptOnNodesException;
import org.jclouds.domain.Credentials;
import org.jclouds.scriptbuilder.domain.OsFamily;
import org.jclouds.scriptbuilder.domain.Statements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CassandraServiceController
{
private static final Logger LOG =
LoggerFactory.getLogger(CassandraServiceController.class);
protected static int CLIENT_PORT = 9160;
protected static int JMX_PORT = 7199;
private static final CassandraServiceController INSTANCE =
new CassandraServiceController();
public static CassandraServiceController getInstance()
{
return INSTANCE;
}
private boolean running;
private ClusterSpec clusterSpec;
private Service service;
private Cluster cluster;
private ComputeService computeService;
private CompositeConfiguration config;
private BlobMetadata tarball;
private List<InetAddress> hosts;
private CassandraServiceController()
{
}
public Cassandra.Client createClient(InetAddress addr) throws TException
{
TTransport transport = new TSocket(
addr.getHostAddress(),
CLIENT_PORT,
200000);
transport = new TFramedTransport(transport);
TProtocol protocol = new TBinaryProtocol(transport);
Cassandra.Client client = new Cassandra.Client(protocol);
transport.open();
return client;
}
private void waitForClusterInitialization()
{
for (InetAddress host : hosts)
waitForNodeInitialization(host);
}
private void waitForNodeInitialization(InetAddress addr)
{
while (true)
{
try
{
Cassandra.Client client = createClient(addr);
client.describe_cluster_name();
break;
}
catch (TException e)
{
LOG.debug(e.toString());
try
{
Thread.sleep(1000);
}
catch (InterruptedException ie)
{
break;
}
}
}
}
public synchronized void startup() throws Exception
{
LOG.info("Starting up cluster...");
config = new CompositeConfiguration();
if (System.getProperty("whirr.config") != null)
{
config.addConfiguration(
new PropertiesConfiguration(System.getProperty("whirr.config")));
}
config.addConfiguration(new PropertiesConfiguration("whirr-default.properties"));
clusterSpec = new ClusterSpec(config);
if (clusterSpec.getPrivateKey() == null)
{
Map<String, String> pair = KeyPair.generate();
clusterSpec.setPublicKey(pair.get("public"));
clusterSpec.setPrivateKey(pair.get("private"));
}
// if a local tarball is available deploy it to the blobstore where it will be available to cassandra
if (System.getProperty("whirr.cassandra_tarball") != null)
{
Pair<BlobMetadata,URI> blob = BlobUtils.storeBlob(config, clusterSpec, System.getProperty("whirr.cassandra_tarball"));
tarball = blob.left;
config.setProperty(CassandraClusterActionHandler.BIN_TARBALL, blob.right.toURL().toString());
// TODO: parse the CassandraVersion property file instead
config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.8");
}
service = new ServiceFactory().create(clusterSpec.getServiceName());
cluster = service.launchCluster(clusterSpec);
computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService();
hosts = new ArrayList<InetAddress>();
for (Instance instance : cluster.getInstances())
{
hosts.add(instance.getPublicAddress());
}
ShutdownHook shutdownHook = new ShutdownHook(this);
Runtime.getRuntime().addShutdownHook(shutdownHook);
waitForClusterInitialization();
running = true;
}
public synchronized void shutdown()
{
// catch and log errors, we're in a runtime shutdown hook
try
{
LOG.info("Shutting down cluster...");
if (tarball != null)
BlobUtils.deleteBlob(config, clusterSpec, tarball);
if (service != null)
service.destroyCluster(clusterSpec);
running = false;
}
catch (Exception e)
{
LOG.error("Error shutting down cluster.", e);
}
}
public class ShutdownHook extends Thread
{
private CassandraServiceController controller;
public ShutdownHook(CassandraServiceController controller)
{
this.controller = controller;
}
public void run()
{
controller.shutdown();
}
}
public synchronized boolean ensureClusterRunning() throws Exception
{
if (running)
{
LOG.info("Cluster already running.");
return false;
}
else
{
startup();
return true;
}
}
/**
* Execute nodetool with args against localhost from the given host.
*/
public void nodetool(String args, InetAddress... hosts)
{
callOnHosts(Arrays.asList(hosts), "nodetool_cassandra", args);
}
/**
* Wipes all persisted state for the given node, leaving it as if it had just started.
*/
public void wipeHosts(InetAddress... hosts)
{
callOnHosts(Arrays.asList(hosts), "wipe_cassandra");
}
public Failure failHosts(List<InetAddress> hosts)
{
return new Failure(hosts).trigger();
}
public Failure failHosts(InetAddress... hosts)
{
return new Failure(Arrays.asList(hosts)).trigger();
}
/** TODO: Move to CassandraService? */
protected void callOnHosts(List<InetAddress> hosts, String functionName, String... functionArgs)
{
final Set<String> hostset = new HashSet<String>();
for (InetAddress host : hosts)
hostset.add(host.getHostAddress());
StatementBuilder statementBuilder = new StatementBuilder();
statementBuilder.addStatement(Statements.call(functionName, functionArgs));
Credentials credentials = new Credentials(clusterSpec.getClusterUser(), clusterSpec.getPrivateKey());
Map<? extends NodeMetadata,ExecResponse> results;
try
{
results = computeService.runScriptOnNodesMatching(new Predicate<NodeMetadata>()
{
public boolean apply(NodeMetadata node)
{
Set<String> intersection = new HashSet<String>(hostset);
intersection.retainAll(node.getPublicAddresses());
return !intersection.isEmpty();
}
},
statementBuilder,
RunScriptOptions.Builder.overrideCredentialsWith(credentials).wrapInInitScript(false).runAsRoot(false));
}
catch (RunScriptOnNodesException e)
{
throw new RuntimeException(e);
}
if (results.size() != hostset.size())
{
throw new RuntimeException(results.size() + " hosts matched " + hostset + ": " + results);
}
for (ExecResponse response : results.values())
{
if (response.getExitCode() != 0)
{
throw new RuntimeException("Call " + functionName + " failed on at least one of " + hostset + ": " + results.values());
}
}
}
public List<InetAddress> getHosts()
{
return hosts;
}
class Failure
{
private List<InetAddress> hosts;
public Failure(List<InetAddress> hosts)
{
this.hosts = hosts;
}
public Failure trigger()
{
callOnHosts(hosts, "stop_cassandra");
return this;
}
public void resolve()
{
callOnHosts(hosts, "start_cassandra");
for (InetAddress host : hosts)
{
waitForNodeInitialization(host);
}
}
}
public InetAddress getPublicHost(InetAddress privateHost)
{
for (Instance instance : cluster.getInstances())
if (privateHost.equals(instance.getPrivateAddress()))
return instance.getPublicAddress();
throw new RuntimeException("No public host for private host " + privateHost);
}
}

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

@ -1,88 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra;
import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.thrift.TException;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class CountersTest extends TestBase
{
@Test
public void testWriteOneReadAll() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
// create a keyspace that performs counter validation
final String keyspace = "TestOneNodeWrite";
keyspace(keyspace).rf(3).validator("CounterColumnType").create();
for (InetAddress host : hosts)
{
Cassandra.Client client = controller.createClient(host);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
add(client, key, "Standard1", "c1", 1, ConsistencyLevel.ONE);
add(client, key, "Standard1", "c2", 2, ConsistencyLevel.ONE);
new CounterGet(client, "Standard1", key).name("c1").value(1L).perform(ConsistencyLevel.ALL);
new CounterGet(client, "Standard1", key).name("c2").value(2L).perform(ConsistencyLevel.ALL);
}
}
protected class CounterGet extends RetryingAction<Long>
{
public CounterGet(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
ByteBuffer bname = ByteBuffer.wrap(name.getBytes());
ColumnPath cpath = new ColumnPath(cf).setColumn(bname);
CounterColumn col = client.get(key, cpath, cl).counter_column;
assertEquals(bname, col.name);
assertEquals(value.longValue(), col.value);
}
}
/** NB: Counter increments are unfortunately not idempotent, so we don't provide a RetyingAction to perform them. */
protected void add(Cassandra.Client client, ByteBuffer key, String cf, String name, long value, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
CounterColumn col = new CounterColumn(ByteBuffer.wrap(name.getBytes()), value);
client.add(key, new ColumnParent(cf), col, cl);
}
}

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

@ -1,125 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class MovementTest extends TestBase
{
private static final String STANDARD_CF = "Standard1";
private static final ColumnParent STANDARD = new ColumnParent(STANDARD_CF);
/** Inserts 1000 keys with names such that at least 1 key ends up on each host. */
private static Map<ByteBuffer,List<ColumnOrSuperColumn>> insertBatch(Cassandra.Client client) throws Exception
{
final int N = 1000;
Column col1 = new Column(ByteBufferUtil.bytes("c1"))
.setValue(ByteBufferUtil.bytes("v1"))
.setTimestamp(0);
Column col2 = new Column(ByteBufferUtil.bytes("c2"))
.setValue(ByteBufferUtil.bytes("v2"))
.setTimestamp(0);
// build N rows
Map<ByteBuffer,List<ColumnOrSuperColumn>> rows = new HashMap<ByteBuffer, List<ColumnOrSuperColumn>>();
Map<ByteBuffer,Map<String,List<Mutation>>> batch = new HashMap<ByteBuffer,Map<String,List<Mutation>>>();
for (int i = 0; i < N; i++)
{
String rawKey = String.format("test.key.%d", i);
ByteBuffer key = ByteBufferUtil.bytes(rawKey);
Mutation m1 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col1));
Mutation m2 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col2));
rows.put(key, Arrays.asList(m1.getColumn_or_supercolumn(),
m2.getColumn_or_supercolumn()));
// add row to batch
Map<String,List<Mutation>> rowmap = new HashMap<String,List<Mutation>>();
rowmap.put(STANDARD_CF, Arrays.asList(m1, m2));
batch.put(key, rowmap);
}
// insert the batch
client.batch_mutate(batch, ConsistencyLevel.ONE);
return rows;
}
private static void verifyBatch(Cassandra.Client client, Map<ByteBuffer,List<ColumnOrSuperColumn>> batch) throws Exception
{
for (Map.Entry<ByteBuffer,List<ColumnOrSuperColumn>> entry : batch.entrySet())
{
// verify slice
SlicePredicate sp = new SlicePredicate();
sp.setSlice_range(
new SliceRange(
ByteBuffer.wrap(new byte[0]),
ByteBuffer.wrap(new byte[0]),
false,
1000
)
);
assertEquals(client.get_slice(entry.getKey(), STANDARD, sp, ConsistencyLevel.ONE),
entry.getValue());
}
}
@Test
public void testLoadbalance() throws Exception
{
final String keyspace = "TestLoadbalance";
addKeyspace(keyspace, 1);
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
client.set_keyspace(keyspace);
// add keys to each node
Map<ByteBuffer,List<ColumnOrSuperColumn>> rows = insertBatch(client);
Thread.sleep(100);
// ask a node to move to a new location
controller.nodetool("loadbalance", hosts.get(0));
// trigger cleanup on all nodes
for (InetAddress host : hosts)
controller.nodetool("cleanup", host);
// check that all keys still exist
verifyBatch(client, rows);
}
}

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

@ -1,188 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.client.RingCache;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.thrift.TException;
import org.apache.cassandra.CassandraServiceController.Failure;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public class MutationTest extends TestBase
{
private static final Logger logger = LoggerFactory.getLogger(MutationTest.class);
@Test
public void testInsert() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
final String keyspace = "TestInsert";
addKeyspace(keyspace, 3);
Cassandra.Client client = controller.createClient(hosts.get(0));
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE);
insert(client, key, "Standard1", "c2", "v2", 0, ConsistencyLevel.ONE);
// block until the column is available
new Get(client, "Standard1", key).name("c1").value("v1").perform(ConsistencyLevel.ONE);
new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.ONE);
List<ColumnOrSuperColumn> coscs = get_slice(client, key, "Standard1", ConsistencyLevel.ONE);
assertColumnEqual("c1", "v1", 0, coscs.get(0).column);
assertColumnEqual("c2", "v2", 0, coscs.get(1).column);
}
@Test
public void testWriteAllReadOne() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteAllReadOne";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ALL);
// should be instantly available
assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE));
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size()));
try {
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
new Get(client, "Standard1", key).name("c1").value("v1")
.perform(ConsistencyLevel.ONE);
new Insert(client, "Standard1", key).name("c3").value("v3")
.expecting(UnavailableException.class).perform(ConsistencyLevel.ALL);
} finally {
failure.resolve();
Thread.sleep(10000);
}
}
@Test
public void testWriteQuorumReadQuorum() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteQuorumReadQuorum";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
// with quorum-1 nodes up
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size())); //kill all but one nodes
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
try {
new Insert(client, "Standard1", key).name("c1").value("v1")
.expecting(UnavailableException.class).perform(ConsistencyLevel.QUORUM);
} finally {
failure.resolve();
Thread.sleep(10000);
}
// with all nodes up
new Insert(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM);
failure = controller.failHosts(endpoints.get(0));
try {
new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM);
} finally {
failure.resolve();
Thread.sleep(10000);
}
}
@Test
public void testWriteOneReadAll() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
final String keyspace = "TestWriteOneReadAll";
addKeyspace(keyspace, 3);
client.set_keyspace(keyspace);
ByteBuffer key = newKey();
List<InetAddress> endpoints = endpointsForKey(hosts.get(0), key, keyspace);
InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace);
client = controller.createClient(coordinator);
client.set_keyspace(keyspace);
insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE);
assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ALL));
// with each of HH, read repair and proactive repair:
// with one node up
// write with one (success)
// read with all (failure)
// bring nodes up
// repair
// read with all (success)
Failure failure = controller.failHosts(endpoints);
try {
new Insert(client, "Standard1", key).name("c2").value("v2")
.expecting(UnavailableException.class).perform(ConsistencyLevel.ONE);
} finally {
failure.resolve();
}
}
protected ByteBuffer newKey()
{
return ByteBufferUtil.bytes(String.format("test.key.%d", System.currentTimeMillis()));
}
}

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

@ -1,329 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra;
import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.thrift.TException;
import org.apache.cassandra.client.*;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.*;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
public abstract class TestBase
{
private static final Logger logger = LoggerFactory.getLogger(TestBase.class);
protected static CassandraServiceController controller =
CassandraServiceController.getInstance();
static class KeyspaceCreation
{
private String name;
private int rf;
private CfDef cfdef;
public KeyspaceCreation(String name)
{
this.name = name;
cfdef = new CfDef(name, "Standard1");
cfdef.setComparator_type("BytesType");
cfdef.setKey_cache_size(10000);
cfdef.setRow_cache_size(1000);
cfdef.setRow_cache_save_period_in_seconds(0);
cfdef.setKey_cache_save_period_in_seconds(3600);
cfdef.setMemtable_throughput_in_mb(255);
cfdef.setMemtable_operations_in_millions(0.29);
}
public KeyspaceCreation validator(String validator)
{
cfdef.setDefault_validation_class(validator);
return this;
}
public KeyspaceCreation rf(int rf)
{
this.rf = rf;
return this;
}
public void create() throws Exception
{
List<InetAddress> hosts = controller.getHosts();
Cassandra.Client client = controller.createClient(hosts.get(0));
Map<String,String> stratOptions = new HashMap<String,String>();
stratOptions.put("replication_factor", "" + rf);
client.system_add_keyspace(new KsDef(name,
"org.apache.cassandra.locator.SimpleStrategy",
Arrays.asList(cfdef))
.setStrategy_options(stratOptions));
// poll, until KS added
for (InetAddress host : hosts)
{
try
{
client = controller.createClient(host);
poll:
while (true)
{
List<KsDef> ksDefList = client.describe_keyspaces();
for (KsDef ks : ksDefList)
{
if (ks.name.equals(name))
break poll;
}
try
{
Thread.sleep(1000);
}
catch (InterruptedException e)
{
break poll;
}
}
}
catch (TException te)
{
continue;
}
}
}
}
protected static KeyspaceCreation keyspace(String name)
{
return new KeyspaceCreation(name);
}
protected static void addKeyspace(String name, int rf) throws Exception
{
keyspace(name).rf(rf).create();
}
@BeforeClass
public static void setUp() throws Exception
{
controller.ensureClusterRunning();
}
protected ByteBuffer newKey()
{
return ByteBuffer.wrap(String.format("test.key.%d", System.currentTimeMillis()).getBytes());
}
protected void insert(Cassandra.Client client, ByteBuffer key, String cf, String name, String value, long timestamp, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
Column col = new Column(ByteBuffer.wrap(name.getBytes()))
.setValue(ByteBuffer.wrap(value.getBytes()))
.setTimestamp(timestamp);
client.insert(key, new ColumnParent(cf), col, cl);
}
protected Column getColumn(Cassandra.Client client, ByteBuffer key, String cf, String col, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException
{
ColumnPath cpath = new ColumnPath(cf);
cpath.setColumn(col.getBytes());
return client.get(key, cpath, cl).column;
}
protected class Get extends RetryingAction<String>
{
public Get(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
assertColumnEqual(name, value, timestamp, getColumn(client, key, cf, name, cl));
}
}
protected class Insert extends RetryingAction<String>
{
public Insert(Cassandra.Client client, String cf, ByteBuffer key)
{
super(client, cf, key);
}
public void tryPerformAction(ConsistencyLevel cl) throws Exception
{
insert(client, key, cf, name, value, timestamp, cl);
}
}
/** Performs an action repeatedly until timeout, success or failure. */
protected abstract class RetryingAction<T>
{
protected Cassandra.Client client;
protected String cf;
protected ByteBuffer key;
protected String name;
protected T value;
protected long timestamp;
private Set<Class<Exception>> expected = new HashSet<Class<Exception>>();
private long timeout = StorageService.RING_DELAY * 2;
public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key)
{
this.client = client;
this.cf = cf;
this.key = key;
this.timestamp = 0;
}
public RetryingAction name(String name)
{
this.name = name; return this;
}
/** A parameterized value for the action. */
public RetryingAction value(T value)
{
this.value = value; return this;
}
/** The total time to allow before failing. */
public RetryingAction timeout(long timeout)
{
this.timeout = timeout; return this;
}
/** The expected timestamp of the returned column. */
public RetryingAction timestamp(long timestamp)
{
this.timestamp = timestamp; return this;
}
/** The exception classes that indicate success. */
public RetryingAction expecting(Class... tempExceptions)
{
this.expected.clear();
for (Class exclass : tempExceptions)
expected.add((Class<Exception>)exclass);
return this;
}
public void perform(ConsistencyLevel cl) throws AssertionError
{
long deadline = System.currentTimeMillis() + timeout;
int attempts = 0;
String template = "%s for " + this + " after %d attempt(s) with %d ms to spare.";
Exception e = null;
while(deadline > System.currentTimeMillis())
{
try
{
attempts++;
tryPerformAction(cl);
logger.info(String.format(template, "Succeeded", attempts, deadline - System.currentTimeMillis()));
return;
}
catch (Exception ex)
{
e = ex;
if (!expected.contains(ex.getClass()))
continue;
logger.info(String.format(template, "Caught expected exception: " + e, attempts, deadline - System.currentTimeMillis()));
return;
}
}
String err = String.format(template, "Caught unexpected: " + e, attempts, deadline - System.currentTimeMillis());
logger.error(err, e);
throw new AssertionError(err);
}
public String toString()
{
return this.getClass().getSimpleName() + "(" + key + "," + name + ")";
}
protected abstract void tryPerformAction(ConsistencyLevel cl) throws Exception;
}
protected List<ColumnOrSuperColumn> get_slice(Cassandra.Client client, ByteBuffer key, String cf, ConsistencyLevel cl)
throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
SlicePredicate sp = new SlicePredicate();
sp.setSlice_range(
new SliceRange(
ByteBuffer.wrap(new byte[0]),
ByteBuffer.wrap(new byte[0]),
false,
1000
)
);
return client.get_slice(key, new ColumnParent(cf), sp, cl);
}
protected void assertColumnEqual(String name, String value, long timestamp, Column col)
{
assertEquals(ByteBuffer.wrap(name.getBytes()), col.name);
assertEquals(ByteBuffer.wrap(value.getBytes()), col.value);
assertEquals(timestamp, col.timestamp);
}
protected List<InetAddress> endpointsForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
Configuration conf = new Configuration();
RingCache ring = new RingCache(keyspace, new RandomPartitioner(), seed.getHostAddress(), 9160);
List<InetAddress> privateendpoints = ring.getEndpoint(key);
List<InetAddress> endpoints = new ArrayList<InetAddress>();
for (InetAddress endpoint : privateendpoints)
{
endpoints.add(controller.getPublicHost(endpoint));
}
return endpoints;
}
protected InetAddress nonEndpointForKey(InetAddress seed, ByteBuffer key, String keyspace)
throws IOException
{
List<InetAddress> endpoints = endpointsForKey(seed, key, keyspace);
for (InetAddress host : controller.getHosts())
{
if (!endpoints.contains(host))
{
return host;
}
}
return null;
}
}

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

@ -1,163 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.utils;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URI;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import com.google.common.io.Files;
import org.apache.commons.configuration.Configuration;
import org.apache.whirr.service.ClusterSpec;
import org.jclouds.blobstore.BlobStoreContext;
import org.jclouds.blobstore.BlobStoreContextFactory;
import org.jclouds.blobstore.InputStreamMap;
import org.jclouds.blobstore.domain.BlobMetadata;
import org.jclouds.s3.S3AsyncClient;
import org.jclouds.s3.S3Client;
import org.jclouds.s3.domain.AccessControlList;
import org.jclouds.s3.domain.CannedAccessPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public final class BlobUtils
{
private static final Logger LOG = LoggerFactory.getLogger(BlobUtils.class);
public static final String BLOB_PROVIDER = "whirr.blobstore.provider";
public static final String BLOB_CONTAINER = "whirr.blobstore.container";
private static BlobStoreContext getContext(Configuration config, ClusterSpec spec)
{
return new BlobStoreContextFactory().createContext(getProvider(config), spec.getIdentity(), spec.getCredential());
}
private static String getProvider(Configuration config)
{
String provider = config.getString(BLOB_PROVIDER, null);
if (provider == null)
throw new RuntimeException("Please set " + BLOB_PROVIDER + " to a jclouds supported provider.");
return provider;
}
private static String getContainer(Configuration config)
{
String container = config.getString(BLOB_CONTAINER, null);
if (container == null)
throw new RuntimeException("Please set " + BLOB_CONTAINER + " to an existing container for your chosen provider.");
return container;
}
/**
* Stores the given local file as a public blob, and returns metadata for the blob.
*/
public static Pair<BlobMetadata,URI> storeBlob(Configuration config, ClusterSpec spec, String filename)
{
File file = new File(filename);
String container = getContainer(config);
String provider = getProvider(config);
// blob name and checksum of the file
String blobName = System.nanoTime() + "/" + file.getName();
String blobNameChecksum = blobName + ".md5";
BlobStoreContext context = getContext(config, spec);
File checksumFile;
try
{
checksumFile = File.createTempFile("dtchecksum", "md5");
checksumFile.deleteOnExit();
FileWriter checksumWriter = new FileWriter(checksumFile);
String checksum = FBUtilities.bytesToHex(Files.getDigest(file, MessageDigest.getInstance("MD5")));
checksumWriter.write(String.format("%s %s", checksum, file.getName()));
checksumWriter.close();
}
catch (IOException e)
{
throw new RuntimeException("Can't create a checksum of the file: " + filename);
}
catch (NoSuchAlgorithmException e)
{
throw new RuntimeException(e.getMessage());
}
try
{
InputStreamMap map = context.createInputStreamMap(container);
map.putFile(blobName, file);
map.putFile(blobNameChecksum, checksumFile);
// TODO: magic! in order to expose the blob as public, we need to dive into provider specific APIs
// the hope is that permissions are encapsulated in jclouds in the future
if (provider.contains("s3"))
{
S3Client sss = context.<S3Client,S3AsyncClient>getProviderSpecificContext().getApi();
String ownerId = sss.getObjectACL(container, blobName).getOwner().getId();
sss.putObjectACL(container,
blobName,
AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId));
sss.putObjectACL(container,
blobNameChecksum,
AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId));
}
else
{
LOG.warn(provider + " may not be properly supported for tarball transfer.");
}
// resolve the full URI of the blob (see http://code.google.com/p/jclouds/issues/detail?id=431)
BlobMetadata blob = context.getBlobStore().blobMetadata(container, blobName);
URI uri = context.getProviderSpecificContext().getEndpoint().resolve("/" + container + "/" + blob.getName());
return new Pair<BlobMetadata, URI>(blob, uri);
}
finally
{
context.close();
}
}
public static void deleteBlob(Configuration config, ClusterSpec spec, BlobMetadata blob)
{
String container = getContainer(config);
BlobStoreContext context = getContext(config, spec);
try
{
context.getBlobStore().removeBlob(container, blob.getName());
}
finally
{
context.close();
}
}
}

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

@ -1,49 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.utils;
import java.io.ByteArrayOutputStream;
import java.util.Map;
import com.google.common.collect.ImmutableMap;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
/**
* A convenience class for generating an RSA key pair.
*/
public class KeyPair {
/**
* return a "public" -> rsa public key, "private" -> its corresponding
* private key
*/
public static Map<String,String> generate() throws JSchException {
com.jcraft.jsch.KeyPair pair = com.jcraft.jsch.KeyPair.genKeyPair(
new JSch(), com.jcraft.jsch.KeyPair.RSA);
ByteArrayOutputStream publicKeyOut = new ByteArrayOutputStream();
ByteArrayOutputStream privateKeyOut = new ByteArrayOutputStream();
pair.writePublicKey(publicKeyOut, "whirr");
pair.writePrivateKey(privateKeyOut);
String publicKey = new String(publicKeyOut.toByteArray());
String privateKey = new String(privateKeyOut.toByteArray());
return ImmutableMap.<String, String> of("public", publicKey,
"private", privateKey);
}
}

2
test/unit/org/apache/cassandra/service/RemoveTest.java
View File

@ -125,7 +125,7 @@ public class RemoveTest extends CleanupHelper
{
try
{
ss.removeToken(token, 0);
ss.removeToken(token);
}
catch (Exception e)
{