Repository: cassandra Updated Branches: refs/heads/trunk cff8dadbe -> 07c11ca45
Add support for multiple table operations to cassandra-stress Patch by Ben Slater; reviewed by tjake for CASSANDRA-8780 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/07c11ca4 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/07c11ca4 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/07c11ca4 Branch: refs/heads/trunk Commit: 07c11ca45b8b55b11a91812a741e85a930ac537e Parents: cff8dad Author: Ben Slater <ben.sla...@instaclustr.com> Authored: Mon May 8 10:18:34 2017 -0400 Committer: T Jake Luciani <j...@apache.org> Committed: Mon May 8 13:36:18 2017 -0400 ---------------------------------------------------------------------- CHANGES.txt | 1 + doc/source/tools/cassandra_stress.rst | 13 ++- doc/source/tools/stress-example.yaml | 1 + .../apache/cassandra/stress/StressProfile.java | 14 ++-- .../org/apache/cassandra/stress/StressYaml.java | 1 + .../SampledOpDistributionFactory.java | 7 +- .../cassandra/stress/report/StressMetrics.java | 4 +- .../SettingsCommandPreDefinedMixed.java | 9 +- .../stress/settings/SettingsCommandUser.java | 86 ++++++++++++++------ .../stress/settings/SettingsSchema.java | 2 +- .../stress/settings/StressSettings.java | 21 +++-- 11 files changed, 110 insertions(+), 49 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index ed69d14..6b5d114 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/doc/source/tools/cassandra_stress.rst ---------------------------------------------------------------------- diff --git a/doc/source/tools/cassandra_stress.rst b/doc/source/tools/cassandra_stress.rst index 417288f..322a981 100644 --- a/doc/source/tools/cassandra_stress.rst +++ b/doc/source/tools/cassandra_stress.rst @@ -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 ^^^^^^^^ http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/doc/source/tools/stress-example.yaml ---------------------------------------------------------------------- diff --git a/doc/source/tools/stress-example.yaml b/doc/source/tools/stress-example.yaml index 0384de2..17161af 100644 --- a/doc/source/tools/stress-example.yaml +++ b/doc/source/tools/stress-example.yaml @@ -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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/StressProfile.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 3662632..2420d68 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -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); http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/StressYaml.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java index 6727f62..afb932c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java @@ -26,6 +26,7 @@ import java.util.Map; public class StressYaml { + public String specname; public String keyspace; public String keyspace_definition; public String table; http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java index 1800039..59f2394 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java index 52e90e2..a4058f2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java +++ b/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java @@ -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); http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java index 72f6c86..1df2a06 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java @@ -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)); } }; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java index 4c7ad91..fd34d74 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java @@ -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()); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java index 872f7dd..26d159e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java @@ -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(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/07c11ca4/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java index b74e93f..a27b986 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java @@ -109,6 +109,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(); --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org
