diff --git a/CHANGES.txt b/CHANGES.txt index 6fdc04a226..097e9c1d84 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.10 + * Cassandra stress should dump all setting on startup (CASSANDRA-11914) * Make it possible to compact a given token range (CASSANDRA-10643) * Allow updating DynamicEndpointSnitch properties via JMX (CASSANDRA-12179) * Collect metrics on queries by consistency level (CASSANDRA-7384) diff --git a/tools/stress/src/org/apache/cassandra/stress/Stress.java b/tools/stress/src/org/apache/cassandra/stress/Stress.java index 50b1b810a6..581ec0c611 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Stress.java +++ b/tools/stress/src/org/apache/cassandra/stress/Stress.java @@ -86,7 +86,14 @@ public final class Stress } MultiPrintStream logout = settings.log.getOutput(); - if (settings.graph.inGraphMode()) { + + if (! settings.log.noSettings) + { + settings.printSettings(logout); + } + + if (settings.graph.inGraphMode()) + { logout.addStream(new PrintStream(settings.graph.temporaryLogFile)); } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 1964c2774d..3a20215040 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -52,6 +52,7 @@ import org.apache.cassandra.stress.operations.userdefined.SchemaQuery; import org.apache.cassandra.stress.operations.userdefined.ValidatingSchemaQuery; import org.apache.cassandra.stress.settings.*; import org.apache.cassandra.stress.util.JavaDriverClient; +import org.apache.cassandra.stress.util.MultiPrintStream; import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.Compression; @@ -76,6 +77,7 @@ public class StressProfile implements Serializable private Map queries; public Map tokenRangeQueries; private Map insert; + private boolean schemaCreated=false; transient volatile TableMetadata tableMetaData; transient volatile Set tokenRanges; @@ -96,6 +98,38 @@ public class StressProfile implements Serializable private static final Pattern lowercaseAlphanumeric = Pattern.compile("[a-z0-9_]+"); + + public void printSettings(MultiPrintStream out, StressSettings stressSettings) + { + out.printf(" Keyspace Name: %s%n", keyspaceName); + out.printf(" Keyspace CQL: %n***%n%s***%n%n", keyspaceCql); + out.printf(" Table Name: %s%n", tableName); + out.printf(" Table CQL: %n***%n%s***%n%n", tableCql); + out.printf(" Extra Schema Definitions: %s%n", extraSchemaDefinitions); + out.printf(" Generator Configs:%n"); + columnConfigs.forEach((k,v)->out.printf(" %s: %s%n", k, v.getConfigAsString())); + out.printf(" Query Definitions:%n"); + queries.forEach((k,v)->out.printf(" %s: %s%n", k, v.getConfigAsString())); + out.printf(" Token Range Queries:%n"); + tokenRangeQueries.forEach((k,v)->out.printf(" %s: %s%n", k, v.getConfigAsString())); + out.printf(" Insert Settings:%n"); + insert.forEach((k,v)->out.printf(" %s: %s%n", k, v)); + + PartitionGenerator generator = newGenerator(stressSettings); + Distribution visits = stressSettings.insert.visits.get(); + SchemaInsert tmp = getInsert(null, generator, null, stressSettings); //just calling this to initialize selectchance and partitions vals for calc below + + double minBatchSize = selectchance.get().min() * partitions.get().minValue() * generator.minRowCount * (1d / visits.maxValue()); + double maxBatchSize = selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount * (1d / visits.minValue()); + out.printf("Generating batches with [%d..%d] partitions and [%.0f..%.0f] rows (of [%.0f..%.0f] total rows in the partitions)%n", + partitions.get().minValue(), partitions.get().maxValue(), + minBatchSize, maxBatchSize, + partitions.get().minValue() * generator.minRowCount, + partitions.get().maxValue() * generator.maxRowCount); + + } + + private void init(StressYaml yaml) throws RequestValidationException { keyspaceName = yaml.keyspace; @@ -175,54 +209,58 @@ public class StressProfile implements Serializable public void maybeCreateSchema(StressSettings settings) { - JavaDriverClient client = settings.getJavaDriverClient(false); - - if (keyspaceCql != null) + if (!schemaCreated) { - try - { - client.execute(keyspaceCql, org.apache.cassandra.db.ConsistencyLevel.ONE); - } - catch (AlreadyExistsException e) - { - } - } + JavaDriverClient client = settings.getJavaDriverClient(false); - client.execute("use " + keyspaceName, org.apache.cassandra.db.ConsistencyLevel.ONE); - - if (tableCql != null) - { - try + if (keyspaceCql != null) { - client.execute(tableCql, org.apache.cassandra.db.ConsistencyLevel.ONE); - } - catch (AlreadyExistsException e) - { - } - - System.out.println(String.format("Created schema. Sleeping %ss for propagation.", settings.node.nodes.size())); - Uninterruptibles.sleepUninterruptibly(settings.node.nodes.size(), TimeUnit.SECONDS); - } - - if (extraSchemaDefinitions != null) - { - for (String extraCql : extraSchemaDefinitions) - { - try { - client.execute(extraCql, org.apache.cassandra.db.ConsistencyLevel.ONE); + client.execute(keyspaceCql, org.apache.cassandra.db.ConsistencyLevel.ONE); } catch (AlreadyExistsException e) { } } - System.out.println(String.format("Created extra schema. Sleeping %ss for propagation.", settings.node.nodes.size())); - Uninterruptibles.sleepUninterruptibly(settings.node.nodes.size(), TimeUnit.SECONDS); - } + client.execute("use " + keyspaceName, org.apache.cassandra.db.ConsistencyLevel.ONE); + if (tableCql != null) + { + try + { + client.execute(tableCql, org.apache.cassandra.db.ConsistencyLevel.ONE); + } + catch (AlreadyExistsException e) + { + } + + System.out.println(String.format("Created schema. Sleeping %ss for propagation.", settings.node.nodes.size())); + Uninterruptibles.sleepUninterruptibly(settings.node.nodes.size(), TimeUnit.SECONDS); + } + + if (extraSchemaDefinitions != null) + { + for (String extraCql : extraSchemaDefinitions) + { + + try + { + client.execute(extraCql, org.apache.cassandra.db.ConsistencyLevel.ONE); + } + catch (AlreadyExistsException e) + { + } + } + + System.out.println(String.format("Created extra schema. Sleeping %ss for propagation.", settings.node.nodes.size())); + Uninterruptibles.sleepUninterruptibly(settings.node.nodes.size(), TimeUnit.SECONDS); + } + schemaCreated = true; + } maybeLoadSchemaInfo(settings); + } public void truncateTable(StressSettings settings) @@ -550,11 +588,7 @@ public class StressProfile implements Serializable // guarantee the vast majority of actions occur in these bounds double minBatchSize = selectchance.get().min() * partitions.get().minValue() * generator.minRowCount * (1d / visits.maxValue()); double maxBatchSize = selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount * (1d / visits.minValue()); - System.out.printf("Generating batches with [%d..%d] partitions and [%.0f..%.0f] rows (of [%.0f..%.0f] total rows in the partitions)%n", - partitions.get().minValue(), partitions.get().maxValue(), - minBatchSize, maxBatchSize, - partitions.get().minValue() * generator.minRowCount, - partitions.get().maxValue() * generator.maxRowCount); + if (generator.maxRowCount > 100 * 1000 * 1000) System.err.printf("WARNING: You have defined a schema that permits very large partitions (%.0f max rows (>100M))%n", generator.maxRowCount); if (batchType == BatchStatement.Type.LOGGED && maxBatchSize > 65535) @@ -628,6 +662,7 @@ public class StressProfile implements Serializable { synchronized (this) { + maybeCreateSchema(settings); maybeLoadSchemaInfo(settings); if (generatorFactory == null) generatorFactory = new GeneratorFactory(); diff --git a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java index 214e56ae41..6727f62690 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java @@ -42,12 +42,20 @@ public class StressYaml { public String cql; public String fields; + public String getConfigAsString() + { + return String.format("CQL:%s;Fields:%s;", cql, fields); + } } public static class TokenRangeQueryDef { public String columns; public int page_size = 5000; + public String getConfigAsString() + { + return String.format("Columns:%s;", columns); + } } } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/DistributionFactory.java b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionFactory.java index d0dfa893a4..4fa46faf59 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/DistributionFactory.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionFactory.java @@ -27,5 +27,6 @@ public interface DistributionFactory extends Serializable { Distribution get(); + String getConfigAsString(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistributionFactory.java b/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistributionFactory.java index 16474d8451..5a01d044ef 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistributionFactory.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistributionFactory.java @@ -27,5 +27,6 @@ public interface RatioDistributionFactory extends Serializable { RatioDistribution get(); + String getConfigAsString(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java index 3522338b87..52977db1e6 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java @@ -62,4 +62,18 @@ public class GeneratorConfig implements Serializable return (sizeDistributions == null ? deflt : sizeDistributions).get(); } + public String getConfigAsString() + { + StringBuilder sb = new StringBuilder(); + if (clusteringDistributions != null){ + sb.append(String.format("Clustering: %s;", clusteringDistributions.getConfigAsString())); + } + if (sizeDistributions != null){ + sb.append(String.format("Size: %s;", sizeDistributions.getConfigAsString())); + } + if (identityDistributions != null){ + sb.append(String.format("Identity: %s;", identityDistributions.getConfigAsString())); + } + return sb.toString(); + } } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/GroupedOptions.java b/tools/stress/src/org/apache/cassandra/stress/settings/GroupedOptions.java index f3924c0ebc..c0a5d35844 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/GroupedOptions.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/GroupedOptions.java @@ -28,6 +28,8 @@ import java.util.Set; import com.google.common.collect.ImmutableList; +import org.apache.cassandra.stress.util.MultiPrintStream; + public abstract class GroupedOptions { @@ -114,6 +116,18 @@ public abstract class GroupedOptions } } + public String getOptionAsString() + { + StringBuilder sb = new StringBuilder(); + for (Option option : options()) + { + sb.append(option.getOptionAsString()); + sb.append("; "); + } + return sb.toString(); + } + + public static List merge(List ... optionss) { ImmutableList.Builder