Skip to content

Commit

Permalink
Print header and statistics for cassandra-stress output with arbitrar…
Browse files Browse the repository at this point in the history
…y frequency

patch by Stefan Miklosovic; reviewed by Brandon Williams and Berenguer Blasi for CASSANDRA-12972
  • Loading branch information
smiklosovic committed May 30, 2023
1 parent fad1f74 commit 18cc821
Show file tree
Hide file tree
Showing 6 changed files with 141 additions and 5 deletions.
1 change: 1 addition & 0 deletions CHANGES.txt
@@ -1,4 +1,5 @@
5.0
* Print header and statistics for cassandra-stress output with arbitrary frequency (CASSANDRA-12972)
* CEP-25: Trie-indexed SSTable format (CASSANDRA-18398)
* Make cassandra-stress able to read all credentials from a file (CASSANDRA-18544)
* Add guardrail for partition size and deprecate compaction_large_partition_warning_threshold (CASSANDRA-18500)
Expand Down
Expand Up @@ -75,6 +75,8 @@ Primary Options:::
Username and password for JMX connection
-credentials-file <path>:;;
Credentials file to specify for CQL, JMX and transport
-reporting:;;
Frequency of printing statistics and header for stress output
Suboptions:::
Every command and primary option has its own collection of suboptions.
These are too numerous to list here. For information on the suboptions
Expand Down
Expand Up @@ -83,11 +83,14 @@ public class StressMetrics implements MeasurementSink
private final Queue<OpMeasurement> leftovers = new ArrayDeque<>();
private final TimingInterval totalCurrentInterval;
private final TimingInterval totalSummaryInterval;
private final int outputFrequencyInSeconds;
private final int headerFrequencyInSeconds;
private int outputLines = 0;

public StressMetrics(ResultLogger output, final long logIntervalMillis, StressSettings settings)
{
this.output = output;
if(settings.log.hdrFile != null)
if (settings.log.hdrFile != null)
{
try
{
Expand Down Expand Up @@ -134,6 +137,8 @@ public StressMetrics(ResultLogger output, final long logIntervalMillis, StressSe
reportingLoop(logIntervalMillis);
});
thread.setName("StressMetrics");
headerFrequencyInSeconds = settings.reporting.headerFrequency;
outputFrequencyInSeconds = settings.reporting.outputFrequency;
}
public void start()
{
Expand Down Expand Up @@ -263,7 +268,12 @@ private void recordInterval(long intervalEnd, long parkIntervalNs)
opInterval.reset();
}

printRow("", "total", totalCurrentInterval, totalSummaryInterval, gcStats, rowRateUncertainty, output);
++outputLines;
if (outputFrequencyInSeconds == 0 || outputLines % outputFrequencyInSeconds == 0)
printRow("", "total", totalCurrentInterval, totalSummaryInterval, gcStats, rowRateUncertainty, output);
if (headerFrequencyInSeconds != 0 && outputLines % headerFrequencyInSeconds == 0)
printHeader("\n", output);

totalCurrentInterval.reset();
}
}
Expand Down
Expand Up @@ -40,7 +40,8 @@ public enum CliOption
JMX("JMX credentials", SettingsJMX.helpPrinter()),
GRAPH("-graph", "Graph recorded metrics", SettingsGraph.helpPrinter()),
TOKENRANGE("Token range settings", SettingsTokenRange.helpPrinter()),
CREDENTIALS_FILE("Credentials file for CQL, JMX and transport", SettingsCredentials.helpPrinter());
CREDENTIALS_FILE("Credentials file for CQL, JMX and transport", SettingsCredentials.helpPrinter()),
REPORTING("Frequency of printing statistics and header for stress output", SettingsReporting.helpPrinter());
;

private static final Map<String, CliOption> LOOKUP;
Expand Down
@@ -0,0 +1,116 @@
/*
* 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.stress.settings;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.stress.util.ResultLogger;

public class SettingsReporting implements Serializable
{
public final int outputFrequency;
private final String outputFrequencyString;
public final int headerFrequency;
private final String headerFrequencyString;

public SettingsReporting(SettingsReporting.Options reporting)
{
if (reporting.headerFrequency.present())
{
headerFrequencyString = reporting.headerFrequency.value();
headerFrequency = new DurationSpec.IntSecondsBound(headerFrequencyString).toSeconds();
}
else
{
headerFrequency = 0;
headerFrequencyString = "*not set*";
}

if (reporting.outputFrequency.present())
{
outputFrequencyString = reporting.outputFrequency.value();
outputFrequency = new DurationSpec.IntSecondsBound(outputFrequencyString).toSeconds();
}
else
{
outputFrequency = 0;
outputFrequencyString = "*not set*";
}
}

// Option Declarations

public static final class Options extends GroupedOptions
{
final OptionSimple outputFrequency = new OptionSimple("output-frequency=",
".*",
"1s",
"Frequency each line of output will be printed out when running a stress test, defaults to '1s'.",
false);

final OptionSimple headerFrequency = new OptionSimple("header-frequency=",
".*",
null,
"Frequency the header for the statistics will be printed out. " +
"If not specified, the header will be printed at the beginning of the test only.",
false);

@Override
public List<? extends Option> options()
{
return Arrays.asList(outputFrequency, headerFrequency);
}
}

public static SettingsReporting get(Map<String, String[]> clArgs)
{
String[] params = clArgs.remove("-reporting");
if (params == null)
return new SettingsReporting(new SettingsReporting.Options());

GroupedOptions options = GroupedOptions.select(params, new SettingsReporting.Options());
if (options == null)
{
printHelp();
System.out.println("Invalid -reporting options provided, see output for valid options");
System.exit(1);
}
return new SettingsReporting((SettingsReporting.Options) options);
}

public void printSettings(ResultLogger out)
{
out.printf(" Output frequency: %s%n", outputFrequencyString);
out.printf(" Header frequency: %s%n", headerFrequencyString);
}

public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-reporting", new SettingsReporting.Options());
}

public static Runnable helpPrinter()
{
return SettingsReporting::printHelp;
}
}
Expand Up @@ -47,6 +47,7 @@ public class StressSettings implements Serializable
public final SettingsJMX jmx;
public final SettingsGraph graph;
public final SettingsTokenRange tokenRange;
public final SettingsReporting reporting;

public StressSettings(SettingsCommand command,
SettingsRate rate,
Expand All @@ -63,7 +64,8 @@ public StressSettings(SettingsCommand command,
SettingsPort port,
SettingsJMX jmx,
SettingsGraph graph,
SettingsTokenRange tokenRange)
SettingsTokenRange tokenRange,
SettingsReporting reporting)
{
this.command = command;
this.rate = rate;
Expand All @@ -81,6 +83,7 @@ public StressSettings(SettingsCommand command,
this.jmx = jmx;
this.graph = graph;
this.tokenRange = tokenRange;
this.reporting = reporting;
}

public SimpleClient getSimpleNativeClient()
Expand Down Expand Up @@ -207,6 +210,7 @@ public static StressSettings get(Map<String, String[]> clArgs)
SettingsTransport transport = SettingsTransport.get(clArgs, credentials);
SettingsJMX jmx = SettingsJMX.get(clArgs, credentials);
SettingsGraph graph = SettingsGraph.get(clArgs, command);
SettingsReporting reporting = SettingsReporting.get(clArgs);
if (!clArgs.isEmpty())
{
printHelp();
Expand All @@ -224,7 +228,7 @@ public static StressSettings get(Map<String, String[]> clArgs)
System.exit(1);
}

return new StressSettings(command, rate, generate, insert, columns, errors, log, credentials, mode, node, schema, transport, port, jmx, graph, tokenRange);
return new StressSettings(command, rate, generate, insert, columns, errors, log, credentials, mode, node, schema, transport, port, jmx, graph, tokenRange, reporting);
}

private static Map<String, String[]> parseMap(String[] args)
Expand Down Expand Up @@ -306,6 +310,8 @@ public void printSettings(ResultLogger out)
tokenRange.printSettings(out);
out.println("Credentials file:");
credentials.printSettings(out);
out.println("Reporting:");
reporting.printSettings(out);

if (command.type == Command.USER)
{
Expand Down

0 comments on commit 18cc821

Please sign in to comment.