diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java index 29ff0161c6..94c33f4a0c 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java @@ -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 :,:,..."); + 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(Arrays.asList(standardCfDef, superCfDef))); + keyspace.setCf_defs(new ArrayList(Arrays.asList(standardCfDef, superCfDef, counterCfDef, counterSuperCfDef))); + Cassandra.Client client = getClient(false); diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java index b57174cfd7..29c2859383 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java @@ -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) { diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterAdder.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterAdder.java new file mode 100644 index 0000000000..3a34cd372e --- /dev/null +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterAdder.java @@ -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 columns = new ArrayList(); + List superColumns = new ArrayList(); + + // 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>> record = new HashMap>>(); + + 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> getSuperColumnsMutationMap(List superColumns) + { + List mutations = new ArrayList(); + Map> mutationMap = new HashMap>(); + + 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> getColumnsMutationMap(List columns) + { + List mutations = new ArrayList(); + Map> mutationMap = new HashMap>(); + + for (CounterColumn c : columns) + { + Counter counter = new Counter().setColumn(c); + mutations.add(new CounterMutation().setCounter(counter)); + } + + mutationMap.put("Counter1", mutations); + + return mutationMap; + } +} diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterGetter.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterGetter.java new file mode 100644 index 0000000000..decf3377e3 --- /dev/null +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/CounterGetter.java @@ -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 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 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); + } + +} diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java index 8b25aa799c..2959c80d3b 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java @@ -31,7 +31,6 @@ import java.util.Map; public class Inserter extends Operation { - public Inserter(int index) { super(index); diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/util/Operation.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/util/Operation.java index 1048298727..7267c8d205 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/util/Operation.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/util/Operation.java @@ -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(); }