diff --git a/CHANGES.txt b/CHANGES.txt index ed69d14524..6b5d11421c 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) diff --git a/doc/source/tools/cassandra_stress.rst b/doc/source/tools/cassandra_stress.rst index 417288fa7f..322a981891 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 ^^^^^^^^ diff --git a/doc/source/tools/stress-example.yaml b/doc/source/tools/stress-example.yaml index 0384de2d1f..17161af27e 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 366263235f..2420d68fb7 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 extraSchemaDefinitions; public final String seedStr = "seed for stress"; + public String specName; public String keyspaceName; public String tableName; private Map 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 stmts = new HashMap<>(); Map 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); diff --git a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java index 6727f62690..afb932cf8c 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; 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 180003919f..59f2394d57 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 implements OpDistributionF this.clustering = clustering; } - protected abstract List get(Timer timer, PartitionGenerator generator, T key, boolean isWarmup); - protected abstract PartitionGenerator newGenerator(); + protected abstract List get(Timer timer, T key, boolean isWarmup); public OpDistribution get(boolean isWarmup, MeasurementSink sink) { - PartitionGenerator generator = newGenerator(); List> operations = new ArrayList<>(); for (Map.Entry ratio : ratios.entrySet()) { List 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 implements OpDistributionF public OpDistribution get(boolean isWarmup, MeasurementSink sink) { List ops = SampledOpDistributionFactory.this.get(new Timer(ratio.getKey().toString(), sink), - newGenerator(), ratio.getKey(), isWarmup); if (ops.size() == 1) 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 52e90e2bcb..a4058f22ee 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); 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 72f6c8684d..1df2a063ae 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(ratios, clustering) { - protected List get(Timer timer, PartitionGenerator generator, Command key, boolean isWarmup) + protected List 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)); } }; } 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 4c7ad91e93..fd34d7450d 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 ratios; private final DistributionFactory clustering; - public final StressProfile profile; + public final Map 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 tokenRangeIterators = new LinkedHashMap<>(); + profiles.forEach((k,v)->tokenRangeIterators.put(k, (v.tokenRangeQueries.isEmpty() + ? null + : new TokenRangeIterator(settings, + v.maybeLoadTokenRanges(settings))))); return new SampledOpDistributionFactory(ratios, clustering) { - protected List get(Timer timer, PartitionGenerator generator, String key, boolean isWarmup) + protected List 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,=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,=1) will perform 2 inserts for each query1. When using multiple files, specify as keyspace.table.op."); @Override public List 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()); } 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 872f7dd3ed..26d159e73f 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(); 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 b74e93f046..a27b986a41 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java @@ -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();