mirror of https://github.com/apache/cassandra
Add counter support to stress tool
Patch by Sylvain Lebresne; reviewed by Pavel Yaskevich and Jake Luciani for CASSANDRA-2299 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1080314 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
T Jake Luciani
parent
c361d204ae
commit
a7e8d6e2a8
|
|
@ -46,6 +46,7 @@ public class Session
|
|||
{
|
||||
availableOptions.addOption("h", "help", false, "Show this help message and exit");
|
||||
availableOptions.addOption("n", "num-keys", true, "Number of keys, default:1000000");
|
||||
availableOptions.addOption("F", "num-different-keys", true, "Number of different keys (if < NUM-KEYS, the same key will re-used multiple times), default:NUM-KEYS");
|
||||
availableOptions.addOption("N", "skip-keys", true, "Fraction of keys to skip initially, default:0");
|
||||
availableOptions.addOption("t", "threads", true, "Number of threads to use, default:50");
|
||||
availableOptions.addOption("c", "columns", true, "Number of columns per key, default:5");
|
||||
|
|
@ -58,7 +59,7 @@ public class Session
|
|||
availableOptions.addOption("f", "file", true, "Write output to given file");
|
||||
availableOptions.addOption("p", "port", true, "Thrift port, default:9160");
|
||||
availableOptions.addOption("m", "unframed", false, "Use unframed transport, default:false");
|
||||
availableOptions.addOption("o", "operation", true, "Operation to perform (INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET), default:INSERT");
|
||||
availableOptions.addOption("o", "operation", true, "Operation to perform (INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET, COUNTER_ADD, COUNTER_GET), default:INSERT");
|
||||
availableOptions.addOption("u", "supercolumns", true, "Number of super columns per key, default:1");
|
||||
availableOptions.addOption("y", "family-type", true, "Column Family Type (Super, Standard), default:Standard");
|
||||
availableOptions.addOption("K", "keep-trying", true, "Retry on-going operation N times (in case of failure). positive integer, default:10");
|
||||
|
|
@ -70,9 +71,11 @@ public class Session
|
|||
availableOptions.addOption("x", "create-index", true, "Type of index to create on needed column families (KEYS)");
|
||||
availableOptions.addOption("R", "replication-strategy", true, "Replication strategy to use (only on insert if keyspace does not exist), default:org.apache.cassandra.locator.SimpleStrategy");
|
||||
availableOptions.addOption("O", "strategy-properties", true, "Replication strategy properties in the following format <dc_name>:<num>,<dc_name>:<num>,...");
|
||||
availableOptions.addOption("W", "no-replicate-on-write",false, "Set replicate_on_write to false for counters. Only counter add with CL=ONE will work");
|
||||
}
|
||||
|
||||
private int numKeys = 1000 * 1000;
|
||||
private int numDifferentKeys = numKeys;
|
||||
private float skipKeys = 0;
|
||||
private int threads = 50;
|
||||
private int columns = 5;
|
||||
|
|
@ -88,6 +91,7 @@ public class Session
|
|||
private int progressInterval = 10;
|
||||
private int keysPerCall = 1000;
|
||||
private int replicationFactor = 1;
|
||||
private boolean replicateOnWrite = true;
|
||||
private boolean ignoreErrors = false;
|
||||
|
||||
private PrintStream out = System.out;
|
||||
|
|
@ -119,6 +123,11 @@ public class Session
|
|||
if (cmd.hasOption("n"))
|
||||
numKeys = Integer.parseInt(cmd.getOptionValue("n"));
|
||||
|
||||
if (cmd.hasOption("F"))
|
||||
numDifferentKeys = Integer.parseInt(cmd.getOptionValue("F"));
|
||||
else
|
||||
numDifferentKeys = numKeys;
|
||||
|
||||
if (cmd.hasOption("N"))
|
||||
skipKeys = Float.parseFloat(cmd.getOptionValue("N"));
|
||||
|
||||
|
|
@ -207,6 +216,7 @@ public class Session
|
|||
ignoreErrors = true;
|
||||
}
|
||||
|
||||
|
||||
if (cmd.hasOption("i"))
|
||||
progressInterval = Integer.parseInt(cmd.getOptionValue("i"));
|
||||
|
||||
|
|
@ -239,14 +249,17 @@ public class Session
|
|||
replicationStrategyOptions.put(keyAndValue[0], keyAndValue[1]);
|
||||
}
|
||||
}
|
||||
|
||||
if (cmd.hasOption("W"))
|
||||
replicateOnWrite = false;
|
||||
}
|
||||
catch (ParseException e)
|
||||
{
|
||||
throw new IllegalArgumentException(e.getMessage(), e);
|
||||
}
|
||||
|
||||
mean = numKeys / 2;
|
||||
sigma = numKeys * STDev;
|
||||
mean = numDifferentKeys / 2;
|
||||
sigma = numDifferentKeys * STDev;
|
||||
|
||||
operations = new AtomicInteger();
|
||||
keys = new AtomicInteger();
|
||||
|
|
@ -283,6 +296,11 @@ public class Session
|
|||
return numKeys;
|
||||
}
|
||||
|
||||
public int getNumDifferentKeys()
|
||||
{
|
||||
return numDifferentKeys;
|
||||
}
|
||||
|
||||
public int getThreads()
|
||||
{
|
||||
return threads;
|
||||
|
|
@ -305,7 +323,7 @@ public class Session
|
|||
|
||||
public int getTotalKeysLength()
|
||||
{
|
||||
return Integer.toString(numKeys).length();
|
||||
return Integer.toString(numDifferentKeys).length();
|
||||
}
|
||||
|
||||
public ConsistencyLevel getConsistencyLevel()
|
||||
|
|
@ -378,6 +396,12 @@ public class Session
|
|||
// column family with super columns
|
||||
CfDef superCfDef = new CfDef("Keyspace1", "Super1").setColumn_metadata(Arrays.asList(superSubColumn)).setColumn_type("Super");
|
||||
|
||||
// column family for standard counters
|
||||
CfDef counterCfDef = new CfDef("Keyspace1", "Counter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite);
|
||||
|
||||
// column family with counter super columns
|
||||
CfDef counterSuperCfDef = new CfDef("Keyspace1", "SuperCounter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setColumn_type("Super");
|
||||
|
||||
keyspace.setName("Keyspace1");
|
||||
keyspace.setStrategy_class(replicationStrategy);
|
||||
keyspace.setReplication_factor(replicationFactor);
|
||||
|
|
@ -387,7 +411,8 @@ public class Session
|
|||
keyspace.setStrategy_options(replicationStrategyOptions);
|
||||
}
|
||||
|
||||
keyspace.setCf_defs(new ArrayList<CfDef>(Arrays.asList(standardCfDef, superCfDef)));
|
||||
keyspace.setCf_defs(new ArrayList<CfDef>(Arrays.asList(standardCfDef, superCfDef, counterCfDef, counterSuperCfDef)));
|
||||
|
||||
|
||||
Cassandra.Client client = getClient(false);
|
||||
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ public final class Stress
|
|||
{
|
||||
public static enum Operations
|
||||
{
|
||||
INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET
|
||||
INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET, COUNTER_ADD, COUNTER_GET
|
||||
}
|
||||
|
||||
public static Session session;
|
||||
|
|
@ -58,7 +58,7 @@ public final class Stress
|
|||
}
|
||||
|
||||
// creating keyspace and column families
|
||||
if (session.getOperation() == Stress.Operations.INSERT)
|
||||
if (session.getOperation() == Operations.INSERT || session.getOperation() == Operations.COUNTER_ADD)
|
||||
{
|
||||
session.createKeySpaces();
|
||||
}
|
||||
|
|
@ -142,9 +142,15 @@ public final class Stress
|
|||
case READ:
|
||||
return new Reader(index);
|
||||
|
||||
case COUNTER_GET:
|
||||
return new CounterGetter(index);
|
||||
|
||||
case INSERT:
|
||||
return new Inserter(index);
|
||||
|
||||
case COUNTER_ADD:
|
||||
return new CounterAdder(index);
|
||||
|
||||
case RANGE_SLICE:
|
||||
return new RangeSlicer(index);
|
||||
|
||||
|
|
@ -185,7 +191,7 @@ public final class Stress
|
|||
{
|
||||
try
|
||||
{
|
||||
operations.put(createOperation(i));
|
||||
operations.put(createOperation(i % session.getNumDifferentKeys()));
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,138 @@
|
|||
/**
|
||||
* 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.contrib.stress.operations;
|
||||
|
||||
import org.apache.cassandra.contrib.stress.util.Operation;
|
||||
import org.apache.cassandra.db.ColumnFamilyType;
|
||||
import org.apache.cassandra.thrift.*;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class CounterAdder extends Operation
|
||||
{
|
||||
public CounterAdder(int index)
|
||||
{
|
||||
super(index);
|
||||
}
|
||||
|
||||
public void run(Cassandra.Client client) throws IOException
|
||||
{
|
||||
List<CounterColumn> columns = new ArrayList<CounterColumn>();
|
||||
List<CounterSuperColumn> superColumns = new ArrayList<CounterSuperColumn>();
|
||||
|
||||
// format used for keys
|
||||
String format = "%0" + session.getTotalKeysLength() + "d";
|
||||
|
||||
for (int i = 0; i < session.getColumnsPerKey(); i++)
|
||||
{
|
||||
String columnName = ("C" + Integer.toString(i));
|
||||
|
||||
columns.add(new CounterColumn(ByteBufferUtil.bytes(columnName), 1L));
|
||||
}
|
||||
|
||||
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
|
||||
{
|
||||
// supers = [SuperColumn('S' + str(j), columns) for j in xrange(supers_per_key)]
|
||||
for (int i = 0; i < session.getSuperColumns(); i++)
|
||||
{
|
||||
String superColumnName = "S" + Integer.toString(i);
|
||||
superColumns.add(new CounterSuperColumn(ByteBuffer.wrap(superColumnName.getBytes()), columns));
|
||||
}
|
||||
}
|
||||
|
||||
String rawKey = String.format(format, index);
|
||||
Map<ByteBuffer, Map<String, List<CounterMutation>>> record = new HashMap<ByteBuffer, Map<String, List<CounterMutation>>>();
|
||||
|
||||
record.put(ByteBufferUtil.bytes(rawKey), session.getColumnFamilyType() == ColumnFamilyType.Super
|
||||
? getSuperColumnsMutationMap(superColumns)
|
||||
: getColumnsMutationMap(columns));
|
||||
|
||||
long start = System.currentTimeMillis();
|
||||
|
||||
boolean success = false;
|
||||
String exceptionMessage = null;
|
||||
|
||||
for (int t = 0; t < session.getRetryTimes(); t++)
|
||||
{
|
||||
if (success)
|
||||
break;
|
||||
|
||||
try
|
||||
{
|
||||
client.batch_add(record, session.getConsistencyLevel());
|
||||
success = true;
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
exceptionMessage = getExceptionMessage(e);
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (!success)
|
||||
{
|
||||
error(String.format("Operation [%d] retried %d times - error incrementing key %s %s%n",
|
||||
index,
|
||||
session.getRetryTimes(),
|
||||
rawKey,
|
||||
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
|
||||
}
|
||||
|
||||
session.operations.getAndIncrement();
|
||||
session.keys.getAndIncrement();
|
||||
session.latency.getAndAdd(System.currentTimeMillis() - start);
|
||||
}
|
||||
|
||||
private Map<String, List<CounterMutation>> getSuperColumnsMutationMap(List<CounterSuperColumn> superColumns)
|
||||
{
|
||||
List<CounterMutation> mutations = new ArrayList<CounterMutation>();
|
||||
Map<String, List<CounterMutation>> mutationMap = new HashMap<String, List<CounterMutation>>();
|
||||
|
||||
for (CounterSuperColumn s : superColumns)
|
||||
{
|
||||
Counter counter = new Counter().setSuper_column(s);
|
||||
mutations.add(new CounterMutation().setCounter(counter));
|
||||
}
|
||||
|
||||
mutationMap.put("SuperCounter1", mutations);
|
||||
|
||||
return mutationMap;
|
||||
}
|
||||
|
||||
private Map<String, List<CounterMutation>> getColumnsMutationMap(List<CounterColumn> columns)
|
||||
{
|
||||
List<CounterMutation> mutations = new ArrayList<CounterMutation>();
|
||||
Map<String, List<CounterMutation>> mutationMap = new HashMap<String, List<CounterMutation>>();
|
||||
|
||||
for (CounterColumn c : columns)
|
||||
{
|
||||
Counter counter = new Counter().setColumn(c);
|
||||
mutations.add(new CounterMutation().setCounter(counter));
|
||||
}
|
||||
|
||||
mutationMap.put("Counter1", mutations);
|
||||
|
||||
return mutationMap;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,150 @@
|
|||
/**
|
||||
* 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.contrib.stress.operations;
|
||||
|
||||
import org.apache.cassandra.contrib.stress.util.Operation;
|
||||
import org.apache.cassandra.db.ColumnFamilyType;
|
||||
import org.apache.cassandra.thrift.*;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.List;
|
||||
|
||||
public class CounterGetter extends Operation
|
||||
{
|
||||
public CounterGetter(int index)
|
||||
{
|
||||
super(index);
|
||||
}
|
||||
|
||||
public void run(Cassandra.Client client) throws IOException
|
||||
{
|
||||
SliceRange sliceRange = new SliceRange();
|
||||
|
||||
// start/finish
|
||||
sliceRange.setStart(new byte[] {}).setFinish(new byte[] {});
|
||||
|
||||
// reversed/count
|
||||
sliceRange.setReversed(false).setCount(session.getColumnsPerKey());
|
||||
|
||||
// initialize SlicePredicate with existing SliceRange
|
||||
SlicePredicate predicate = new SlicePredicate().setSlice_range(sliceRange);
|
||||
|
||||
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
|
||||
{
|
||||
runSuperCounterGetter(predicate, client);
|
||||
}
|
||||
else
|
||||
{
|
||||
runCounterGetter(predicate, client);
|
||||
}
|
||||
}
|
||||
|
||||
private void runSuperCounterGetter(SlicePredicate predicate, Cassandra.Client client) throws IOException
|
||||
{
|
||||
byte[] rawKey = generateKey();
|
||||
ByteBuffer key = ByteBuffer.wrap(rawKey);
|
||||
|
||||
for (int j = 0; j < session.getSuperColumns(); j++)
|
||||
{
|
||||
String superColumn = 'S' + Integer.toString(j);
|
||||
ColumnParent parent = new ColumnParent("CounterSuper1").setSuper_column(superColumn.getBytes());
|
||||
|
||||
long start = System.currentTimeMillis();
|
||||
|
||||
boolean success = false;
|
||||
String exceptionMessage = null;
|
||||
|
||||
for (int t = 0; t < session.getRetryTimes(); t++)
|
||||
{
|
||||
if (success)
|
||||
break;
|
||||
|
||||
try
|
||||
{
|
||||
List<Counter> counters;
|
||||
counters = client.get_counter_slice(key, parent, predicate, session.getConsistencyLevel());
|
||||
success = (counters.size() != 0);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
exceptionMessage = getExceptionMessage(e);
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (!success)
|
||||
{
|
||||
error(String.format("Operation [%d] retried %d times - error reading counter key %s %s%n",
|
||||
index,
|
||||
session.getRetryTimes(),
|
||||
new String(rawKey),
|
||||
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
|
||||
}
|
||||
|
||||
session.operations.getAndIncrement();
|
||||
session.keys.getAndIncrement();
|
||||
session.latency.getAndAdd(System.currentTimeMillis() - start);
|
||||
}
|
||||
}
|
||||
|
||||
private void runCounterGetter(SlicePredicate predicate, Cassandra.Client client) throws IOException
|
||||
{
|
||||
ColumnParent parent = new ColumnParent("Counter1");
|
||||
|
||||
byte[] key = generateKey();
|
||||
ByteBuffer keyBuffer = ByteBuffer.wrap(key);
|
||||
|
||||
long start = System.currentTimeMillis();
|
||||
|
||||
boolean success = false;
|
||||
String exceptionMessage = null;
|
||||
|
||||
for (int t = 0; t < session.getRetryTimes(); t++)
|
||||
{
|
||||
if (success)
|
||||
break;
|
||||
|
||||
try
|
||||
{
|
||||
List<Counter> counters;
|
||||
counters = client.get_counter_slice(keyBuffer, parent, predicate, session.getConsistencyLevel());
|
||||
success = (counters.size() != 0);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
exceptionMessage = getExceptionMessage(e);
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (!success)
|
||||
{
|
||||
error(String.format("Operation [%d] retried %d times - error reading counter key %s %s%n",
|
||||
index,
|
||||
session.getRetryTimes(),
|
||||
new String(key),
|
||||
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
|
||||
}
|
||||
|
||||
session.operations.getAndIncrement();
|
||||
session.keys.getAndIncrement();
|
||||
session.latency.getAndAdd(System.currentTimeMillis() - start);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -31,7 +31,6 @@ import java.util.Map;
|
|||
|
||||
public class Inserter extends Operation
|
||||
{
|
||||
|
||||
public Inserter(int index)
|
||||
{
|
||||
super(index);
|
||||
|
|
|
|||
|
|
@ -92,7 +92,7 @@ public abstract class Operation
|
|||
private static byte[] generateRandomKey()
|
||||
{
|
||||
String format = "%0" + Stress.session.getTotalKeysLength() + "d";
|
||||
return String.format(format, Stress.randomizer.nextInt(Stress.session.getNumKeys() - 1)).getBytes();
|
||||
return String.format(format, Stress.randomizer.nextInt(Stress.session.getNumDifferentKeys() - 1)).getBytes();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -108,7 +108,7 @@ public abstract class Operation
|
|||
{
|
||||
double token = nextGaussian(session.getMean(), session.getSigma());
|
||||
|
||||
if (0 <= token && token < session.getNumKeys())
|
||||
if (0 <= token && token < session.getNumDifferentKeys())
|
||||
{
|
||||
return String.format(format, (int) token).getBytes();
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue