mirror of https://github.com/apache/cassandra
Add support for multiple table operations to cassandra-stress
Patch by Ben Slater; reviewed by tjake for CASSANDRA-8780
This commit is contained in:
Ben Slater
T Jake Luciani
parent
cff8dadbe8
commit
07c11ca45b
|
|
@ -1,4 +1,5 @@
|
|||
4.0
|
||||
* Add multiple table operation support to cassandra-stress (CASSANDRA-8780)
|
||||
* Fix incorrect cqlsh results when selecting same columns multiple times (CASSANDRA-13262)
|
||||
* Fix WriteResponseHandlerTest is sensitive to test execution order (CASSANDRA-13421)
|
||||
* Improve incremental repair logging (CASSANDRA-13468)
|
||||
|
|
|
|||
|
|
@ -107,7 +107,12 @@ doesn't scale.
|
|||
Profile
|
||||
+++++++
|
||||
|
||||
User mode requires a profile defined in YAML.
|
||||
User mode requires a profile defined in YAML.
|
||||
Multiple YAML files may be specified in which case operations in the ops argument are referenced as specname.opname.
|
||||
|
||||
An identifier for the profile::
|
||||
|
||||
specname: staff_activities
|
||||
|
||||
The keyspace for the test::
|
||||
|
||||
|
|
@ -209,6 +214,12 @@ queries. Additionally the table will be truncated once before the test.
|
|||
|
||||
The full example can be found here :download:`yaml <./stress-example.yaml>`
|
||||
|
||||
Running a user mode test with multiple yaml files::
|
||||
cassandra-stress user profile=./example.yaml,./example2.yaml duration=1m "ops(ex1.insert=1,ex1.latest_event=1,ex2.insert=2)" truncate=once
|
||||
|
||||
This will run operations as specified in both the example.yaml and example2.yaml files. example.yaml and example2.yaml can reference the same table
|
||||
although care must be taken that the table definition is identical (data generation specs can be different).
|
||||
|
||||
Graphing
|
||||
^^^^^^^^
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
spacenam: example # idenitifier for this spec if running with multiple yaml files
|
||||
keyspace: example
|
||||
|
||||
# Would almost always be network topology unless running something locally
|
||||
|
|
|
|||
|
|
@ -66,6 +66,7 @@ public class StressProfile implements Serializable
|
|||
private List<String> extraSchemaDefinitions;
|
||||
public final String seedStr = "seed for stress";
|
||||
|
||||
public String specName;
|
||||
public String keyspaceName;
|
||||
public String tableName;
|
||||
private Map<String, GeneratorConfig> columnConfigs;
|
||||
|
|
@ -144,6 +145,8 @@ public class StressProfile implements Serializable
|
|||
queries = yaml.queries;
|
||||
tokenRangeQueries = yaml.token_range_queries;
|
||||
insert = yaml.insert;
|
||||
specName = yaml.specname;
|
||||
if (specName == null){specName = keyspaceName + "." + tableName;}
|
||||
|
||||
extraSchemaDefinitions = yaml.extra_definitions;
|
||||
|
||||
|
|
@ -216,7 +219,7 @@ public class StressProfile implements Serializable
|
|||
{
|
||||
if (!schemaCreated)
|
||||
{
|
||||
JavaDriverClient client = settings.getJavaDriverClient(false);
|
||||
JavaDriverClient client = settings.getJavaDriverClient();
|
||||
|
||||
if (keyspaceCql != null)
|
||||
{
|
||||
|
|
@ -283,7 +286,7 @@ public class StressProfile implements Serializable
|
|||
{
|
||||
if (tableMetaData == null)
|
||||
{
|
||||
JavaDriverClient client = settings.getJavaDriverClient();
|
||||
JavaDriverClient client = settings.getJavaDriverClient(keyspaceName);
|
||||
|
||||
synchronized (client)
|
||||
{
|
||||
|
|
@ -317,7 +320,7 @@ public class StressProfile implements Serializable
|
|||
{
|
||||
maybeLoadSchemaInfo(settings); // ensure table metadata is available
|
||||
|
||||
JavaDriverClient client = settings.getJavaDriverClient();
|
||||
JavaDriverClient client = settings.getJavaDriverClient(keyspaceName);
|
||||
synchronized (client)
|
||||
{
|
||||
if (tokenRanges != null)
|
||||
|
|
@ -361,7 +364,7 @@ public class StressProfile implements Serializable
|
|||
{
|
||||
if (queryStatements == null)
|
||||
{
|
||||
JavaDriverClient jclient = settings.getJavaDriverClient();
|
||||
JavaDriverClient jclient = settings.getJavaDriverClient(keyspaceName);
|
||||
|
||||
Map<String, PreparedStatement> stmts = new HashMap<>();
|
||||
Map<String, SchemaQuery.ArgSelect> args = new HashMap<>();
|
||||
|
|
@ -588,7 +591,7 @@ public class StressProfile implements Serializable
|
|||
System.err.printf("WARNING: You have defined a schema that permits very large batches (%.0f max rows (>100K)). This may OOM this stress client, or the server.%n",
|
||||
selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount);
|
||||
|
||||
JavaDriverClient client = settings.getJavaDriverClient();
|
||||
JavaDriverClient client = settings.getJavaDriverClient(keyspaceName);
|
||||
String query = sb.toString();
|
||||
|
||||
insertStatement = client.prepare(query);
|
||||
|
|
@ -772,6 +775,7 @@ public class StressProfile implements Serializable
|
|||
|
||||
StressYaml profileYaml = yaml.loadAs(yamlStream, StressYaml.class);
|
||||
|
||||
|
||||
StressProfile profile = new StressProfile();
|
||||
profile.init(profileYaml);
|
||||
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ import java.util.Map;
|
|||
|
||||
public class StressYaml
|
||||
{
|
||||
public String specname;
|
||||
public String keyspace;
|
||||
public String keyspace_definition;
|
||||
public String table;
|
||||
|
|
|
|||
|
|
@ -46,17 +46,15 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
|
|||
this.clustering = clustering;
|
||||
}
|
||||
|
||||
protected abstract List<? extends Operation> get(Timer timer, PartitionGenerator generator, T key, boolean isWarmup);
|
||||
protected abstract PartitionGenerator newGenerator();
|
||||
protected abstract List<? extends Operation> get(Timer timer, T key, boolean isWarmup);
|
||||
|
||||
public OpDistribution get(boolean isWarmup, MeasurementSink sink)
|
||||
{
|
||||
PartitionGenerator generator = newGenerator();
|
||||
List<Pair<Operation, Double>> operations = new ArrayList<>();
|
||||
for (Map.Entry<T, Double> ratio : ratios.entrySet())
|
||||
{
|
||||
List<? extends Operation> ops = get(new Timer(ratio.getKey().toString(), sink),
|
||||
generator, ratio.getKey(), isWarmup);
|
||||
ratio.getKey(), isWarmup);
|
||||
for (Operation op : ops)
|
||||
operations.add(new Pair<>(op, ratio.getValue() / ops.size()));
|
||||
}
|
||||
|
|
@ -81,7 +79,6 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
|
|||
public OpDistribution get(boolean isWarmup, MeasurementSink sink)
|
||||
{
|
||||
List<? extends Operation> ops = SampledOpDistributionFactory.this.get(new Timer(ratio.getKey().toString(), sink),
|
||||
newGenerator(),
|
||||
ratio.getKey(),
|
||||
isWarmup);
|
||||
if (ops.size() == 1)
|
||||
|
|
|
|||
|
|
@ -339,9 +339,9 @@ public class StressMetrics implements MeasurementSink
|
|||
|
||||
|
||||
// PRINT FORMATTING
|
||||
public static final String HEADFORMAT = "%-50s%10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s,%7s,%7s,%8s,%8s,%8s,%8s";
|
||||
public static final String ROWFORMAT = "%-50s%10d,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f,%7d,%7.0f,%8.0f,%8.0f,%8.0f,%8.0f";
|
||||
|
||||
public static final String HEADFORMAT = "%-10s%10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s,%7s,%7s,%8s,%8s,%8s,%8s";
|
||||
public static final String ROWFORMAT = "%-10s%10d,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f,%7d,%7.0f,%8.0f,%8.0f,%8.0f,%8.0f";
|
||||
public static final String[] HEADMETRICS = new String[]{"type", "total ops","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr", "errors", "gc: #", "max ms", "sum ms", "sdv ms", "mb"};
|
||||
public static final String HEAD = String.format(HEADFORMAT, (Object[]) HEADMETRICS);
|
||||
|
||||
|
|
|
|||
|
|
@ -58,14 +58,9 @@ public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined
|
|||
final SeedManager seeds = new SeedManager(settings);
|
||||
return new SampledOpDistributionFactory<Command>(ratios, clustering)
|
||||
{
|
||||
protected List<? extends Operation> get(Timer timer, PartitionGenerator generator, Command key, boolean isWarmup)
|
||||
protected List<? extends Operation> get(Timer timer, Command key, boolean isWarmup)
|
||||
{
|
||||
return Collections.singletonList(PredefinedOperation.operation(key, timer, generator, seeds, settings, add));
|
||||
}
|
||||
|
||||
protected PartitionGenerator newGenerator()
|
||||
{
|
||||
return SettingsCommandPreDefinedMixed.this.newGenerator(settings);
|
||||
return Collections.singletonList(PredefinedOperation.operation(key, timer, SettingsCommandPreDefinedMixed.this.newGenerator(settings), seeds, settings, add));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,6 +27,10 @@ import java.util.Arrays;
|
|||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
|
||||
import org.apache.cassandra.stress.Operation;
|
||||
import org.apache.cassandra.stress.StressProfile;
|
||||
|
|
@ -46,8 +50,11 @@ public class SettingsCommandUser extends SettingsCommand
|
|||
// Ratios for selecting commands - index for each Command, NaN indicates the command is not requested
|
||||
private final Map<String, Double> ratios;
|
||||
private final DistributionFactory clustering;
|
||||
public final StressProfile profile;
|
||||
public final Map<String, StressProfile> profiles;
|
||||
private final Options options;
|
||||
private String default_profile_name;
|
||||
private static final Pattern EXTRACT_SPEC_CMD = Pattern.compile("(.+)\\.(.+)");
|
||||
|
||||
|
||||
public SettingsCommandUser(Options options)
|
||||
{
|
||||
|
|
@ -56,10 +63,26 @@ public class SettingsCommandUser extends SettingsCommand
|
|||
this.options = options;
|
||||
clustering = options.clustering.get();
|
||||
ratios = options.ops.ratios();
|
||||
default_profile_name=null;
|
||||
|
||||
|
||||
String yamlPath = options.profile.value();
|
||||
File yamlFile = new File(yamlPath);
|
||||
profile = StressProfile.load(yamlFile.exists() ? yamlFile.toURI() : URI.create(yamlPath));
|
||||
profiles = new LinkedHashMap<>();
|
||||
|
||||
String[] yamlPaths = yamlPath.split(",");
|
||||
for (String curYamlPath : yamlPaths)
|
||||
{
|
||||
File yamlFile = new File(curYamlPath);
|
||||
StressProfile profile = StressProfile.load(yamlFile.exists() ? yamlFile.toURI() : URI.create(curYamlPath));
|
||||
String specName = profile.specName;
|
||||
if (default_profile_name == null) {default_profile_name=specName;} //first file is default
|
||||
if (profiles.containsKey(specName))
|
||||
{
|
||||
throw new IllegalArgumentException("Must only specify a singe YAML file per table (including keyspace qualifier).");
|
||||
}
|
||||
profiles.put(specName, profile);
|
||||
}
|
||||
|
||||
|
||||
if (ratios.size() == 0)
|
||||
throw new IllegalArgumentException("Must specify at least one command with a non-zero ratio");
|
||||
|
|
@ -73,36 +96,54 @@ public class SettingsCommandUser extends SettingsCommand
|
|||
public OpDistributionFactory getFactory(final StressSettings settings)
|
||||
{
|
||||
final SeedManager seeds = new SeedManager(settings);
|
||||
final TokenRangeIterator tokenRangeIterator = profile.tokenRangeQueries.isEmpty()
|
||||
? null
|
||||
: new TokenRangeIterator(settings,
|
||||
profile.maybeLoadTokenRanges(settings));
|
||||
|
||||
final Map<String, TokenRangeIterator> tokenRangeIterators = new LinkedHashMap<>();
|
||||
profiles.forEach((k,v)->tokenRangeIterators.put(k, (v.tokenRangeQueries.isEmpty()
|
||||
? null
|
||||
: new TokenRangeIterator(settings,
|
||||
v.maybeLoadTokenRanges(settings)))));
|
||||
|
||||
return new SampledOpDistributionFactory<String>(ratios, clustering)
|
||||
{
|
||||
protected List<? extends Operation> get(Timer timer, PartitionGenerator generator, String key, boolean isWarmup)
|
||||
protected List<? extends Operation> get(Timer timer, String key, boolean isWarmup)
|
||||
{
|
||||
if (key.equalsIgnoreCase("insert"))
|
||||
Matcher m = EXTRACT_SPEC_CMD.matcher(key);
|
||||
final String profile_name;
|
||||
final String sub_key;
|
||||
if (m.matches())
|
||||
{
|
||||
profile_name = m.group(1);
|
||||
sub_key = m.group(2);
|
||||
}
|
||||
else
|
||||
{
|
||||
profile_name = default_profile_name;
|
||||
sub_key = key;
|
||||
}
|
||||
|
||||
if (!profiles.containsKey(profile_name))
|
||||
{
|
||||
throw new IllegalArgumentException(String.format("Op name %s contains an invalid profile specname: %s", key, profile_name));
|
||||
}
|
||||
StressProfile profile = profiles.get(profile_name);
|
||||
TokenRangeIterator tokenRangeIterator = tokenRangeIterators.get(profile_name);
|
||||
PartitionGenerator generator = profile.newGenerator(settings);
|
||||
if (sub_key.equalsIgnoreCase("insert"))
|
||||
return Collections.singletonList(profile.getInsert(timer, generator, seeds, settings));
|
||||
if (key.equalsIgnoreCase("validate"))
|
||||
if (sub_key.equalsIgnoreCase("validate"))
|
||||
return profile.getValidate(timer, generator, seeds, settings);
|
||||
|
||||
if (profile.tokenRangeQueries.containsKey(key))
|
||||
return Collections.singletonList(profile.getBulkReadQueries(key, timer, settings, tokenRangeIterator, isWarmup));
|
||||
if (profile.tokenRangeQueries.containsKey(sub_key))
|
||||
return Collections.singletonList(profile.getBulkReadQueries(sub_key, timer, settings, tokenRangeIterator, isWarmup));
|
||||
|
||||
return Collections.singletonList(profile.getQuery(key, timer, generator, seeds, settings, isWarmup));
|
||||
}
|
||||
|
||||
protected PartitionGenerator newGenerator()
|
||||
{
|
||||
return profile.newGenerator(settings);
|
||||
return Collections.singletonList(profile.getQuery(sub_key, timer, generator, seeds, settings, isWarmup));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public void truncateTables(StressSettings settings)
|
||||
{
|
||||
profile.truncateTable(settings);
|
||||
profiles.forEach((k,v)-> v.truncateTable(settings));
|
||||
}
|
||||
|
||||
static final class Options extends GroupedOptions
|
||||
|
|
@ -113,8 +154,8 @@ public class SettingsCommandUser extends SettingsCommand
|
|||
this.parent = parent;
|
||||
}
|
||||
final OptionDistribution clustering = new OptionDistribution("clustering=", "gaussian(1..10)", "Distribution clustering runs of operations of the same kind");
|
||||
final OptionSimple profile = new OptionSimple("profile=", ".*", null, "Specify the path to a yaml cql3 profile", true);
|
||||
final OptionAnyProbabilities ops = new OptionAnyProbabilities("ops", "Specify the ratios for inserts/queries to perform; e.g. ops(insert=2,<query1>=1) will perform 2 inserts for each query1");
|
||||
final OptionSimple profile = new OptionSimple("profile=", ".*", null, "Specify the path to a yaml cql3 profile. Multiple comma separated files can be added.", true);
|
||||
final OptionAnyProbabilities ops = new OptionAnyProbabilities("ops", "Specify the ratios for inserts/queries to perform; e.g. ops(insert=2,<query1>=1) will perform 2 inserts for each query1. When using multiple files, specify as keyspace.table.op.");
|
||||
|
||||
@Override
|
||||
public List<? extends Option> options()
|
||||
|
|
@ -130,8 +171,7 @@ public class SettingsCommandUser extends SettingsCommand
|
|||
super.printSettings(out);
|
||||
out.printf(" Command Ratios: %s%n", ratios);
|
||||
out.printf(" Command Clustering Distribution: %s%n", options.clustering.getOptionAsString());
|
||||
out.printf(" Profile File: %s%n", options.profile.value());
|
||||
// profile.noSettings(out);
|
||||
out.printf(" Profile File(s): %s%n", options.profile.value());
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ public class SettingsSchema implements Serializable
|
|||
public SettingsSchema(Options options, SettingsCommand command)
|
||||
{
|
||||
if (command instanceof SettingsCommandUser)
|
||||
keyspace = ((SettingsCommandUser) command).profile.keyspaceName;
|
||||
keyspace = null; //this should never be used - StressProfile passes keyspace name directly
|
||||
else
|
||||
keyspace = options.keyspace.value();
|
||||
|
||||
|
|
|
|||
|
|
@ -108,6 +108,17 @@ public class StressSettings implements Serializable
|
|||
}
|
||||
|
||||
public JavaDriverClient getJavaDriverClient(boolean setKeyspace)
|
||||
{
|
||||
if (setKeyspace)
|
||||
{
|
||||
return getJavaDriverClient(schema.keyspace);
|
||||
} else {
|
||||
return getJavaDriverClient(null);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public JavaDriverClient getJavaDriverClient(String keyspace)
|
||||
{
|
||||
if (client != null)
|
||||
return client;
|
||||
|
|
@ -126,8 +137,8 @@ public class StressSettings implements Serializable
|
|||
EncryptionOptions.ClientEncryptionOptions encOptions = transport.getEncryptionOptions();
|
||||
JavaDriverClient c = new JavaDriverClient(this, currentNode, port.nativePort, encOptions);
|
||||
c.connect(mode.compression());
|
||||
if (setKeyspace)
|
||||
c.execute("USE \"" + schema.keyspace + "\";", org.apache.cassandra.db.ConsistencyLevel.ONE);
|
||||
if (keyspace != null)
|
||||
c.execute("USE \"" + keyspace + "\";", org.apache.cassandra.db.ConsistencyLevel.ONE);
|
||||
|
||||
return client = c;
|
||||
}
|
||||
|
|
@ -144,7 +155,7 @@ public class StressSettings implements Serializable
|
|||
if (command.type == Command.WRITE || command.type == Command.COUNTER_WRITE)
|
||||
schema.createKeySpaces(this);
|
||||
else if (command.type == Command.USER)
|
||||
((SettingsCommandUser) command).profile.maybeCreateSchema(this);
|
||||
((SettingsCommandUser) command).profiles.forEach((k,v) -> v.maybeCreateSchema(this));
|
||||
}
|
||||
|
||||
public static StressSettings parse(String[] args)
|
||||
|
|
@ -296,8 +307,8 @@ public class StressSettings implements Serializable
|
|||
if (command.type == Command.USER)
|
||||
{
|
||||
out.println();
|
||||
out.println("******************** Profile ********************");
|
||||
((SettingsCommandUser) command).profile.printSettings(out, this);
|
||||
out.println("******************** Profile(s) ********************");
|
||||
((SettingsCommandUser) command).profiles.forEach((k,v) -> v.printSettings(out, this));
|
||||
}
|
||||
out.println();
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue