natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GCInspector more closely tracks GC; cassandra-stress and nodetool report it 

patch by benedict; reviewed by tjake
This commit is contained in:
Benedict Elliott Smith 2014-09-14 09:09:37 +01:00
parent d002c7edc5
commit 0e652e7548
10 changed files with 331 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.1
* GCInspector more closely tracks GC; cassandra-stress and nodetool report it
* nodetool won't output bogus ownership info without a keyspace (CASSANDRA-7173)
* Add human readable option to nodetool commands (CASSANDRA-5433)
* Don't try to set repairedAt on old sstables (CASSANDRA-7913)

83
src/java/org/apache/cassandra/service/GCInspector.java
View File

@ -22,6 +22,8 @@ import java.lang.management.MemoryUsage;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationListener;
@ -35,11 +37,55 @@ import com.sun.management.GarbageCollectionNotificationInfo;
import org.apache.cassandra.io.sstable.SSTableDeletingTask;
import org.apache.cassandra.utils.StatusLogger;
public class GCInspector implements NotificationListener
public class GCInspector implements NotificationListener, GCInspectorMXBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.service:type=GCInspector";
private static final Logger logger = LoggerFactory.getLogger(GCInspector.class);
final static long MIN_DURATION = 200;
final static long MIN_DURATION_TPSTATS = 1000;
final static long MIN_LOG_DURATION = 200;
final static long MIN_LOG_DURATION_TPSTATS = 1000;
static final class State
{
final double maxRealTimeElapsed;
final double totalRealTimeElapsed;
final double sumSquaresRealTimeElapsed;
final double totalBytesReclaimed;
final double count;
final long startNanos;
State(double extraElapsed, double extraBytes, State prev)
{
this.totalRealTimeElapsed = prev.totalRealTimeElapsed + extraElapsed;
this.totalBytesReclaimed = prev.totalBytesReclaimed + extraBytes;
this.sumSquaresRealTimeElapsed = prev.sumSquaresRealTimeElapsed + (extraElapsed * extraElapsed);
this.startNanos = prev.startNanos;
this.count = prev.count + 1;
this.maxRealTimeElapsed = Math.max(prev.maxRealTimeElapsed, extraElapsed);
}
State()
{
count = maxRealTimeElapsed = sumSquaresRealTimeElapsed = totalRealTimeElapsed = totalBytesReclaimed = 0;
startNanos = System.nanoTime();
}
}
final AtomicReference<State> state = new AtomicReference<>(new State());
public GCInspector()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
public static void register() throws Exception
{
@ -66,6 +112,7 @@ public class GCInspector implements NotificationListener
StringBuilder sb = new StringBuilder();
sb.append(info.getGcName()).append(" GC in ").append(duration).append("ms. ");
long bytes = 0;
List<String> keys = new ArrayList<>(info.getGcInfo().getMemoryUsageBeforeGc().keySet());
Collections.sort(keys);
for (String key : keys)
@ -79,16 +126,24 @@ public class GCInspector implements NotificationListener
sb.append(after.getUsed());
if (!key.equals(keys.get(keys.size() - 1)))
sb.append("; ");
bytes += before.getUsed() - after.getUsed();
}
}
while (true)
{
State prev = state.get();
if (state.compareAndSet(prev, new State(duration, bytes, prev)))
break;
}
String st = sb.toString();
if (duration > MIN_DURATION)
if (duration > MIN_LOG_DURATION)
logger.info(st);
else if (logger.isDebugEnabled())
logger.debug(st);
if (duration > MIN_DURATION_TPSTATS)
if (duration > MIN_LOG_DURATION_TPSTATS)
StatusLogger.log();
// if we just finished a full collection and we're still using a lot of memory, try to reduce the pressure
@ -96,4 +151,22 @@ public class GCInspector implements NotificationListener
SSTableDeletingTask.rescheduleFailedTasks();
}
}
public State getTotalSinceLastCheck()
{
return state.getAndSet(new State());
}
public double[] getAndResetStats()
{
State state = getTotalSinceLastCheck();
double[] r = new double[6];
r[0] = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - state.startNanos);
r[1] = state.maxRealTimeElapsed;
r[2] = state.totalRealTimeElapsed;
r[3] = state.sumSquaresRealTimeElapsed;
r[4] = state.totalBytesReclaimed;
r[5] = state.count;
return r;
}
}

25
src/java/org/apache/cassandra/service/GCInspectorMXBean.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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.service;
public interface GCInspectorMXBean
{
// returns { interval (ms), max(gc real time (ms)), sum(gc real time (ms)), sum((gc real time (ms))^2), sum(gc bytes), count(gc) }
public double[] getAndResetStats();
}

11
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -79,6 +79,7 @@ public class NodeProbe implements AutoCloseable
private CompactionManagerMBean compactionProxy;
private StorageServiceMBean ssProxy;
private MemoryMXBean memProxy;
private GCInspectorMXBean gcProxy;
private RuntimeMXBean runtimeProxy;
private StreamManagerMBean streamProxy;
public MessagingServiceMBean msProxy;
@ -169,7 +170,10 @@ public class NodeProbe implements AutoCloseable
spProxy = JMX.newMBeanProxy(mbeanServerConn, name, StorageProxyMBean.class);
name = new ObjectName(HintedHandOffManager.MBEAN_NAME);
hhProxy = JMX.newMBeanProxy(mbeanServerConn, name, HintedHandOffManagerMBean.class);
} catch (MalformedObjectNameException e)
name = new ObjectName(GCInspector.MBEAN_NAME);
gcProxy = JMX.newMBeanProxy(mbeanServerConn, name, GCInspectorMXBean.class);
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException(
"Invalid ObjectName? Please report this as a bug.", e);
@ -374,6 +378,11 @@ public class NodeProbe implements AutoCloseable
}
}
public double[] getAndResetGCStats()
{
return gcProxy.getAndResetStats();
}
public Iterator<Map.Entry<String, ColumnFamilyStoreMBean>> getColumnFamilyStoreMBeanProxies()
{
try

15
src/java/org/apache/cassandra/tools/NodeTool.java
View File

@ -96,6 +96,7 @@ public class NodeTool
DisableGossip.class,
EnableHandoff.class,
EnableThrift.class,
GcStats.class,
GetCompactionThreshold.class,
GetCompactionThroughput.class,
GetStreamThroughput.class,
@ -2361,6 +2362,20 @@ public class NodeTool
}
}
@Command(name = "gcstats", description = "Print GC Statistics")
public static class GcStats extends NodeTool.NodeToolCmd
{
@Override
public void execute(NodeProbe probe)
{
double[] stats = probe.getAndResetGCStats();
double mean = stats[2] / stats[5];
double stdev = Math.sqrt((stats[3] / stats[5]) - (mean * mean));
System.out.printf("%20s%20s%20s%20s%20s%n", "Interval (ms)", "Max GC Elapsed (ms)", "Total GC Elapsed (ms)", "Stdev GC Elapsed (ms)", "GC Reclaimed (MB)", "Collections");
System.out.printf("%20.0d%20.0d%20.0d%20.0d%20.0d%n", stats[0], stats[1], stats[2], stdev, stats[4], stats[5]);
}
}
@Command(name = "truncatehints", description = "Truncate all hints on the local node, or truncate hints for the endpoint(s) specified.")
public static class TruncateHints extends NodeToolCmd
{

2
tools/stress/src/org/apache/cassandra/stress/StressAction.java
View File

@ -182,7 +182,7 @@ public class StressAction implements Runnable
if (settings.rate.opRateTargetPerSecond > 0)
rateLimiter = RateLimiter.create(settings.rate.opRateTargetPerSecond);
final StressMetrics metrics = new StressMetrics(output, settings.log.intervalMillis);
final StressMetrics metrics = new StressMetrics(output, settings.log.intervalMillis, settings);
final CountDownLatch done = new CountDownLatch(threadCount);
final Consumer[] consumers = new Consumer[threadCount];

61
tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
View File

@ -23,12 +23,16 @@ package org.apache.cassandra.stress;
import java.io.PrintStream;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.JmxCollector;
import org.apache.cassandra.stress.util.Timing;
import org.apache.cassandra.stress.util.TimingInterval;
import org.apache.cassandra.stress.util.Uncertainty;
@ -45,10 +49,33 @@ public class StressMetrics
private final Uncertainty rowRateUncertainty = new Uncertainty();
private final CountDownLatch stopped = new CountDownLatch(1);
private final Timing timing = new Timing();
private final Callable<JmxCollector.GcStats> gcStatsCollector;
private volatile JmxCollector.GcStats totalGcStats;
public StressMetrics(PrintStream output, final long logIntervalMillis)
public StressMetrics(PrintStream output, final long logIntervalMillis, StressSettings settings)
{
this.output = output;
Callable<JmxCollector.GcStats> gcStatsCollector;
try
{
gcStatsCollector = new JmxCollector(settings.node.nodes, settings.port.jmxPort);
totalGcStats = new JmxCollector.GcStats(0);
}
catch (Throwable t)
{
t.printStackTrace();
System.err.println("Failed to connect over JMX; not collecting these stats");
totalGcStats = new JmxCollector.GcStats(Double.POSITIVE_INFINITY);
gcStatsCollector = new Callable<JmxCollector.GcStats>()
{
public JmxCollector.GcStats call() throws Exception
{
return totalGcStats;
}
};
}
this.gcStatsCollector = gcStatsCollector;
printHeader("", output);
thread = tf.newThread(new Runnable()
{
@ -121,10 +148,10 @@ public class StressMetrics
private void update() throws InterruptedException
{
TimingInterval interval = timing.snapInterval();
if (interval.partitionCount != 0)
printRow("", interval, timing.getHistory(), rowRateUncertainty, output);
rowRateUncertainty.update(interval.adjustedRowRate());
Timing.TimingResult<JmxCollector.GcStats> result = timing.snap(gcStatsCollector);
if (result.timing.partitionCount != 0)
printRow("", result.timing, timing.getHistory(), result.extra, rowRateUncertainty, output);
rowRateUncertainty.update(result.timing.adjustedRowRate());
if (timing.done())
stop = true;
}
@ -132,15 +159,15 @@ public class StressMetrics
// PRINT FORMATTING
public static final String HEADFORMAT = "%-10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s";
public static final String ROWFORMAT = "%-10d,%8.0f,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f";
public static final String HEADFORMAT = "%-10s,%10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s,%7s,%8s,%8s,%8s,%8s";
public static final String ROWFORMAT = "%-10d,%10.0f,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f,%7.0f,%8.0f,%8.0f,%8.0f,%8.0f";
private static void printHeader(String prefix, PrintStream output)
{
output.println(prefix + String.format(HEADFORMAT, "total ops","adj row/s","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr"));
output.println(prefix + String.format(HEADFORMAT, "total ops","adj row/s","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr", "gc: #", "max ms", "sum ms", "sdv ms", "mb"));
}
private static void printRow(String prefix, TimingInterval interval, TimingInterval total, Uncertainty opRateUncertainty, PrintStream output)
private static void printRow(String prefix, TimingInterval interval, TimingInterval total, JmxCollector.GcStats gcStats, Uncertainty opRateUncertainty, PrintStream output)
{
output.println(prefix + String.format(ROWFORMAT,
total.operationCount,
@ -155,7 +182,13 @@ public class StressMetrics
interval.rankLatency(0.999f),
interval.maxLatency(),
total.runTime() / 1000f,
opRateUncertainty.getUncertainty()));
opRateUncertainty.getUncertainty(),
gcStats.count,
gcStats.maxms,
gcStats.summs,
gcStats.sdvms,
gcStats.bytes / (1 << 20)
));
}
public void summarise()
@ -172,11 +205,16 @@ public class StressMetrics
output.println(String.format("latency 99th percentile : %.1f", history.rankLatency(0.99f)));
output.println(String.format("latency 99.9th percentile : %.1f", history.rankLatency(0.999f)));
output.println(String.format("latency max : %.1f", history.maxLatency()));
output.println(String.format("total gc count : %.0f", totalGcStats.count));
output.println(String.format("total gc mb : %.0f", totalGcStats.bytes / (1 << 20)));
output.println(String.format("total gc time (s) : %.0f", totalGcStats.summs / 1000));
output.println(String.format("avg gc time(ms) : %.0f", totalGcStats.summs / totalGcStats.count));
output.println(String.format("stdev gc time(ms) : %.0f", totalGcStats.sdvms));
output.println("Total operation time : " + DurationFormatUtils.formatDuration(
history.runTime(), "HH:mm:ss", true));
}
public static final void summarise(List<String> ids, List<StressMetrics> summarise, PrintStream out)
public static void summarise(List<String> ids, List<StressMetrics> summarise, PrintStream out)
{
int idLen = 0;
for (String id : ids)
@ -187,6 +225,7 @@ public class StressMetrics
printRow(String.format(formatstr, ids.get(i)),
summarise.get(i).timing.getHistory(),
summarise.get(i).timing.getHistory(),
summarise.get(i).totalGcStats,
summarise.get(i).rowRateUncertainty,
out
);

3
tools/stress/src/org/apache/cassandra/stress/settings/SettingsPort.java
View File

@ -31,11 +31,13 @@ public class SettingsPort implements Serializable
public final int nativePort;
public final int thriftPort;
public final int jmxPort;
public SettingsPort(PortOptions options)
{
nativePort = Integer.parseInt(options.nativePort.value());
thriftPort = Integer.parseInt(options.thriftPort.value());
jmxPort = Integer.parseInt(options.jmxPort.value());
}
// Option Declarations
@ -44,6 +46,7 @@ public class SettingsPort implements Serializable
{
final OptionSimple nativePort = new OptionSimple("native=", "[0-9]+", "9042", "Use this port for the Cassandra native protocol", false);
final OptionSimple thriftPort = new OptionSimple("thrift=", "[0-9]+", "9160", "Use this port for the thrift protocol", false);
final OptionSimple jmxPort = new OptionSimple("jmx=", "[0-9]+", "7199", "Use this port for retrieving statistics over jmx", false);
@Override
public List<? extends Option> options()

119
tools/stress/src/org/apache/cassandra/stress/util/JmxCollector.java Normal file
View File

@ -0,0 +1,119 @@
/*
* 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.util;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.tools.NodeProbe;
public class JmxCollector implements Callable<JmxCollector.GcStats>
{
public static class GcStats
{
public final double count;
public final double bytes;
public final double maxms;
public final double summs;
public final double sumsqms;
public final double sdvms;
public GcStats(double count, double bytes, double maxms, double summs, double sumsqms)
{
this.count = count;
this.bytes = bytes;
this.maxms = maxms;
this.summs = summs;
this.sumsqms = sumsqms;
double mean = summs / count;
double stdev = Math.sqrt((sumsqms / count) - (mean * mean));
if (Double.isNaN(stdev))
stdev = 0;
this.sdvms = stdev;
}
public GcStats(double fill)
{
this(fill, fill, fill, fill, fill);
}
public static GcStats aggregate(List<GcStats> stats)
{
double count = 0, bytes = 0, maxms = 0, summs = 0, sumsqms = 0;
for (GcStats stat : stats)
{
count += stat.count;
bytes += stat.bytes;
maxms += stat.maxms;
summs += stat.summs;
sumsqms += stat.sumsqms;
}
return new GcStats(count, bytes, maxms, summs, sumsqms);
}
}
final NodeProbe[] probes;
// TODO: should expand to whole cluster
public JmxCollector(List<String> hosts, int port)
{
probes = new NodeProbe[hosts.size()];
for (int i = 0 ; i < hosts.size() ; i++)
probes[i] = connect(hosts.get(i), port);
}
private static NodeProbe connect(String host, int port)
{
try
{
return new NodeProbe(host, port);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
public GcStats call() throws Exception
{
final List<Future<GcStats>> futures = new ArrayList<>();
for (final NodeProbe probe : probes)
{
futures.add(TPE.submit(new Callable<GcStats>()
{
public GcStats call() throws Exception
{
final double[] stats = probe.getAndResetGCStats();
return new GcStats(stats[5], stats[4], stats[1], stats[2], stats[3]);
}
}));
}
List<GcStats> results = new ArrayList<>();
for (Future<GcStats> future : futures)
results.add(future.get());
return GcStats.aggregate(results);
}
private static final ExecutorService TPE = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new NamedThreadFactory("JmxCollector"));
}

36
tools/stress/src/org/apache/cassandra/stress/util/Timing.java
View File

@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -44,7 +45,18 @@ public class Timing
// TIMING
private TimingInterval snapInterval(Random rnd) throws InterruptedException
public static class TimingResult<E>
{
public final E extra;
public final TimingInterval timing;
public TimingResult(E extra, TimingInterval timing)
{
this.extra = extra;
this.timing = timing;
}
}
private <E> TimingResult<E> snap(Random rnd, Callable<E> call) throws InterruptedException
{
final Timer[] timers = this.timers.toArray(new Timer[0]);
final CountDownLatch ready = new CountDownLatch(timers.length);
@ -54,8 +66,18 @@ public class Timing
timer.requestReport(ready);
}
E extra;
try
{
extra = call.call();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
// TODO fail gracefully after timeout if a thread is stuck
if (!ready.await(2L, TimeUnit.MINUTES))
if (!ready.await(5L, TimeUnit.MINUTES))
throw new RuntimeException("Timed out waiting for a timer thread - seems one got stuck");
boolean done = true;
@ -68,7 +90,7 @@ public class Timing
}
this.done = done;
return TimingInterval.merge(rnd, intervals, Integer.MAX_VALUE, history.endNanos());
return new TimingResult<>(extra, TimingInterval.merge(rnd, intervals, Integer.MAX_VALUE, history.endNanos()));
}
// build a new timer and add it to the set of running timers
@ -89,11 +111,11 @@ public class Timing
return done;
}
public TimingInterval snapInterval() throws InterruptedException
public <E> TimingResult<E> snap(Callable<E> call) throws InterruptedException
{
final TimingInterval interval = snapInterval(rnd);
history = TimingInterval.merge(rnd, Arrays.asList(interval, history), 200000, history.startNanos());
return interval;
final TimingResult<E> result = snap(rnd, call);
history = TimingInterval.merge(rnd, Arrays.asList(result.timing, history), 200000, history.startNanos());
return result;
}
public TimingInterval getHistory()