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

Add a virtual table to expose thread pools 

patch by Chris Lohfink; reviewed by Aleksey Yeschenko for
CASSANDRA-14523
This commit is contained in:
Chris Lohfink 2018-06-22 10:48:01 -05:00 committed by Aleksey Yeshchenko
parent fcb87a2d1c
commit d5ae2ae481
20 changed files with 355 additions and 304 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0
* Add a virtual table to expose thread pools (CASSANDRA-14523)
* Add a virtual table to expose caches (CASSANDRA-14538)
* Fix toDate function for timestamp arguments (CASSANDRA-14502)
* Revert running dtests by default in circleci (CASSANDRA-14614)

12
src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java
View File

@ -213,6 +213,18 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor implements
super.beforeExecute(t, r);
}
@Override
public int getActiveTaskCount()
{
return getActiveCount();
}
@Override
public int getPendingTaskCount()
{
return getQueue().size();
}
/**
* Send @param t and any exception wrapped by @param r to the default uncaught exception handler,
* or log them if none such is set up

2
src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java
View File

@ -79,7 +79,7 @@ public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor i
{
super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, threadFactory);
super.prestartAllCoreThreads();
metrics = new ThreadPoolMetrics(this, jmxPath, threadFactory.id);
metrics = new ThreadPoolMetrics(this, jmxPath, threadFactory.id).register();
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
mbeanName = "org.apache.cassandra." + jmxPath + ":type=" + threadFactory.id;

46
src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java
View File

@ -27,8 +27,50 @@ public interface LocalAwareExecutorService extends ExecutorService
{
// we need a way to inject a TraceState directly into the Executor context without going through
// the global Tracing sessions; see CASSANDRA-5668
public void execute(Runnable command, ExecutorLocals locals);
void execute(Runnable command, ExecutorLocals locals);
// permits executing in the context of the submitting thread
public void maybeExecuteImmediately(Runnable command);
void maybeExecuteImmediately(Runnable command);
/**
* Returns the approximate number of threads that are actively
* executing tasks.
*
* @return the number of threads
*/
int getActiveTaskCount();
/**
* Returns the approximate total number of tasks that have
* completed execution. Because the states of tasks and threads
* may change dynamically during computation, the returned value
* is only an approximation, but one that does not ever decrease
* across successive calls.
*
* @return the number of tasks
*/
long getCompletedTaskCount();
/**
* Returns the approximate total of tasks waiting to be executed.
* Because the states of tasks and threads
* may change dynamically during computation, the returned value
* is only an approximation, but one that does not ever decrease
* across successive calls.
*
* @return the number of tasks
*/
int getPendingTaskCount();
/**
* Returns the maximum allowed number of threads.
*
* @return the maximum allowed number of threads
*/
int getMaximumPoolSize();
default int getMaxTasksQueued()
{
return -1;
}
}

28
src/java/org/apache/cassandra/concurrent/SEPExecutor.java
View File

@ -23,7 +23,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.metrics.SEPMetrics;
import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.concurrent.WaitQueue;
@ -36,7 +36,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
public final int maxWorkers;
public final String name;
public final int maxTasksQueued;
private final SEPMetrics metrics;
private final ThreadPoolMetrics metrics;
// stores both a set of work permits and task permits:
// bottom 32 bits are number of queued tasks, in the range [0..maxTasksQueued] (initially 0)
@ -60,7 +60,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
this.maxWorkers = maxWorkers;
this.maxTasksQueued = maxTasksQueued;
this.permits.set(combine(0, maxWorkers));
this.metrics = new SEPMetrics(this, jmxPath, name);
this.metrics = new ThreadPoolMetrics(this, jmxPath, name).register();
}
protected void onCompletion()
@ -68,6 +68,12 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
completedTasks.incrementAndGet();
}
@Override
public int getMaxTasksQueued()
{
return maxTasksQueued;
}
// schedules another worker for this pool if there is work outstanding and there are no spinning threads that
// will self-assign to it in the immediate future
boolean maybeSchedule()
@ -208,7 +214,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
{
shuttingDown = true;
pool.executors.remove(this);
if (getActiveCount() == 0)
if (getActiveTaskCount() == 0)
shutdown.signalAll();
// release metrics
@ -240,21 +246,29 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
return isTerminated();
}
public long getPendingTasks()
@Override
public int getPendingTaskCount()
{
return taskPermits(permits.get());
}
public long getCompletedTasks()
@Override
public long getCompletedTaskCount()
{
return completedTasks.get();
}
public int getActiveCount()
public int getActiveTaskCount()
{
return maxWorkers - workPermits(permits.get());
}
@Override
public int getMaximumPoolSize()
{
return maxWorkers;
}
private static int taskPermits(long both)
{
return (int) both;

2
src/java/org/apache/cassandra/concurrent/SEPWorker.java
View File

@ -118,7 +118,7 @@ final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnabl
// return our work permit, and maybe signal shutdown
assigned.returnWorkPermit();
if (shutdown && assigned.getActiveCount() == 0)
if (shutdown && assigned.getActiveTaskCount() == 0)
assigned.shutdown.signalAll();
assigned = null;

12
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -132,5 +132,17 @@ public class StageManager
{
execute(command);
}
@Override
public int getActiveTaskCount()
{
return getActiveCount();
}
@Override
public int getPendingTaskCount()
{
return getQueue().size();
}
}
}

5
src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
View File

@ -27,6 +27,9 @@ public final class SystemViewsKeyspace extends VirtualKeyspace
private SystemViewsKeyspace()
{
super(NAME, ImmutableList.of(new SSTableTasksTable(NAME), new ClientsTable(NAME), new CachesTable(NAME)));
super(NAME, ImmutableList.of(new CachesTable(NAME),
new ClientsTable(NAME),
new SSTableTasksTable(NAME),
new ThreadPoolsTable(NAME)));
}
}

85
src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java Normal file
View File

@ -0,0 +1,85 @@
/*
* 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.db.virtual;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.apache.cassandra.schema.TableMetadata;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
final class ThreadPoolsTable extends AbstractVirtualTable
{
private static final String NAME = "name";
private static final String ACTIVE_TASKS = "active_tasks";
private static final String ACTIVE_TASKS_LIMIT = "active_tasks_limit";
private static final String PENDING_TASKS = "pending_tasks";
private static final String COMPLETED_TASKS = "completed_tasks";
private static final String BLOCKED_TASKS = "blocked_tasks";
private static final String BLOCKED_TASKS_ALL_TIME = "blocked_tasks_all_time";
ThreadPoolsTable(String keyspace)
{
super(TableMetadata.builder(keyspace, "thread_pools")
.kind(TableMetadata.Kind.VIRTUAL)
.partitioner(new LocalPartitioner(UTF8Type.instance))
.addPartitionKeyColumn(NAME, UTF8Type.instance)
.addRegularColumn(ACTIVE_TASKS, Int32Type.instance)
.addRegularColumn(ACTIVE_TASKS_LIMIT, Int32Type.instance)
.addRegularColumn(PENDING_TASKS, Int32Type.instance)
.addRegularColumn(COMPLETED_TASKS, LongType.instance)
.addRegularColumn(BLOCKED_TASKS, LongType.instance)
.addRegularColumn(BLOCKED_TASKS_ALL_TIME, LongType.instance)
.build());
}
@Override
public DataSet data(DecoratedKey partitionKey)
{
String poolName = UTF8Type.instance.compose(partitionKey.getKey());
SimpleDataSet result = new SimpleDataSet(metadata());
Metrics.getThreadPoolMetrics(poolName)
.ifPresent(metrics -> addRow(result, metrics));
return result;
}
@Override
public DataSet data()
{
SimpleDataSet result = new SimpleDataSet(metadata());
Metrics.allThreadPoolMetrics()
.forEach(metrics -> addRow(result, metrics));
return result;
}
private void addRow(SimpleDataSet dataSet, ThreadPoolMetrics metrics)
{
dataSet.row(metrics.poolName)
.column(ACTIVE_TASKS, metrics.activeTasks.getValue())
.column(ACTIVE_TASKS_LIMIT, metrics.maxPoolSize.getValue())
.column(PENDING_TASKS, metrics.pendingTasks.getValue())
.column(COMPLETED_TASKS, metrics.completedTasks.getValue())
.column(BLOCKED_TASKS, metrics.currentBlocked.getCount())
.column(BLOCKED_TASKS_ALL_TIME, metrics.totalBlocked.getCount());
}
}

33
src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
View File

@ -19,14 +19,21 @@ package org.apache.cassandra.metrics;
import java.lang.management.ManagementFactory;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import com.codahale.metrics.*;
import com.google.common.annotations.VisibleForTesting;
import javax.management.*;
/**
* Makes integrating 3.0 metrics API with 2.0.
* <p>
@ -36,6 +43,7 @@ import javax.management.*;
public class CassandraMetricsRegistry extends MetricRegistry
{
public static final CassandraMetricsRegistry Metrics = new CassandraMetricsRegistry();
private final Map<String, ThreadPoolMetrics> threadPoolMetrics = new ConcurrentHashMap<>();
private final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
@ -119,6 +127,27 @@ public class CassandraMetricsRegistry extends MetricRegistry
}
}
public Collection<ThreadPoolMetrics> allThreadPoolMetrics()
{
return Collections.unmodifiableCollection(threadPoolMetrics.values());
}
public Optional<ThreadPoolMetrics> getThreadPoolMetrics(String poolName)
{
return Optional.ofNullable(threadPoolMetrics.get(poolName));
}
ThreadPoolMetrics register(ThreadPoolMetrics metrics)
{
threadPoolMetrics.put(metrics.poolName, metrics);
return metrics;
}
void remove(ThreadPoolMetrics metrics)
{
threadPoolMetrics.remove(metrics.poolName, metrics);
}
public <T extends Metric> T register(MetricName name, MetricName aliasName, T metric)
{
T ret = register(name, metric);

109
src/java/org/apache/cassandra/metrics/SEPMetrics.java
View File

@ -1,109 +0,0 @@
/*
* 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.metrics;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Gauge;
import org.apache.cassandra.concurrent.SEPExecutor;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class SEPMetrics
{
/** Number of active tasks. */
public final Gauge<Integer> activeTasks;
/** Number of tasks that had blocked before being accepted (or rejected). */
public final Counter totalBlocked;
/**
* Number of tasks currently blocked, waiting to be accepted by
* the executor (because all threads are busy and the backing queue is full).
*/
public final Counter currentBlocked;
/** Number of completed tasks. */
public final Gauge<Long> completedTasks;
/** Number of tasks waiting to be executed. */
public final Gauge<Long> pendingTasks;
/** Maximum number of threads before it will start queuing tasks */
public final Gauge<Integer> maxPoolSize;
/** Maximum number of tasks queued before a task get blocked */
public final Gauge<Integer> maxTasksQueued;
private MetricNameFactory factory;
/**
* Create metrics for the given LowSignalExecutor.
*
* @param executor Thread pool
* @param path Type of thread pool
* @param poolName Name of thread pool to identify metrics
*/
public SEPMetrics(final SEPExecutor executor, String path, String poolName)
{
this.factory = new ThreadPoolMetricNameFactory("ThreadPools", path, poolName);
activeTasks = Metrics.register(factory.createMetricName("ActiveTasks"), new Gauge<Integer>()
{
public Integer getValue()
{
return executor.getActiveCount();
}
});
pendingTasks = Metrics.register(factory.createMetricName("PendingTasks"), new Gauge<Long>()
{
public Long getValue()
{
return executor.getPendingTasks();
}
});
totalBlocked = Metrics.counter(factory.createMetricName("TotalBlockedTasks"));
currentBlocked = Metrics.counter(factory.createMetricName("CurrentlyBlockedTasks"));
completedTasks = Metrics.register(factory.createMetricName("CompletedTasks"), new Gauge<Long>()
{
public Long getValue()
{
return executor.getCompletedTasks();
}
});
maxPoolSize = Metrics.register(factory.createMetricName("MaxPoolSize"), new Gauge<Integer>()
{
public Integer getValue()
{
return executor.maxWorkers;
}
});
maxTasksQueued = Metrics.register(factory.createMetricName("MaxTasksQueued"), new Gauge<Integer>()
{
public Integer getValue()
{
return executor.maxTasksQueued;
}
});
}
public void release()
{
Metrics.remove(factory.createMetricName("ActiveTasks"));
Metrics.remove(factory.createMetricName("PendingTasks"));
Metrics.remove(factory.createMetricName("CompletedTasks"));
Metrics.remove(factory.createMetricName("TotalBlockedTasks"));
Metrics.remove(factory.createMetricName("CurrentlyBlockedTasks"));
Metrics.remove(factory.createMetricName("MaxPoolSize"));
Metrics.remove(factory.createMetricName("MaxTasksQueued"));
}
}

45
src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java
View File

@ -1,45 +0,0 @@
/*
* 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.metrics;
class ThreadPoolMetricNameFactory implements MetricNameFactory
{
private final String type;
private final String path;
private final String poolName;
ThreadPoolMetricNameFactory(String type, String path, String poolName)
{
this.type = type;
this.path = path;
this.poolName = poolName;
}
public CassandraMetricsRegistry.MetricName createMetricName(String metricName)
{
String groupName = ThreadPoolMetrics.class.getPackage().getName();
StringBuilder mbeanName = new StringBuilder();
mbeanName.append(groupName).append(":");
mbeanName.append("type=").append(type);
mbeanName.append(",path=").append(path);
mbeanName.append(",scope=").append(poolName);
mbeanName.append(",name=").append(metricName);
return new CassandraMetricsRegistry.MetricName(groupName, type, metricName, path + "." + poolName, mbeanName.toString());
}
}

173
src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java
View File

@ -17,47 +17,56 @@
*/
package org.apache.cassandra.metrics;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.JmxReporter;
import org.apache.cassandra.concurrent.LocalAwareExecutorService;
import org.apache.cassandra.metrics.CassandraMetricsRegistry.MetricName;
import javax.management.JMX;
import javax.management.MBeanServerConnection;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import static java.lang.String.format;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
/**
* Metrics for {@link ThreadPoolExecutor}.
*/
public class ThreadPoolMetrics
{
public static final String ACTIVE_TASKS = "ActiveTasks";
public static final String PENDING_TASKS = "PendingTasks";
public static final String COMPLETED_TASKS = "CompletedTasks";
public static final String CURRENTLY_BLOCKED_TASKS = "CurrentlyBlockedTasks";
public static final String TOTAL_BLOCKED_TASKS = "TotalBlockedTasks";
public static final String MAX_POOL_SIZE = "MaxPoolSize";
public static final String MAX_TASKS_QUEUED = "MaxTasksQueued";
/** Number of active tasks. */
public final Gauge<Integer> activeTasks;
/** Number of tasks that had blocked before being accepted (or rejected). */
public final Counter totalBlocked;
/** Number of tasks waiting to be executed. */
public final Gauge<Integer> pendingTasks;
/** Number of completed tasks. */
public final Gauge<Long> completedTasks;
/**
* Number of tasks currently blocked, waiting to be accepted by
* the executor (because all threads are busy and the backing queue is full).
*/
public final Counter currentBlocked;
/** Number of completed tasks. */
public final Gauge<Long> completedTasks;
/** Number of tasks waiting to be executed. */
public final Gauge<Long> pendingTasks;
/** Number of tasks that had blocked before being accepted (or rejected). */
public final Counter totalBlocked;
/** Maximum number of threads before it will start queuing tasks */
public final Gauge<Integer> maxPoolSize;
private MetricNameFactory factory;
/** Maximum number of tasks queued before a task get blocked */
public final Gauge<Integer> maxTasksQueued;
public final String path;
public final String poolName;
/**
* Create metrics for given ThreadPoolExecutor.
@ -66,105 +75,51 @@ public class ThreadPoolMetrics
* @param path Type of thread pool
* @param poolName Name of thread pool to identify metrics
*/
public ThreadPoolMetrics(final ThreadPoolExecutor executor, String path, String poolName)
public ThreadPoolMetrics(LocalAwareExecutorService executor, String path, String poolName)
{
this.factory = new ThreadPoolMetricNameFactory("ThreadPools", path, poolName);
this.path = path;
this.poolName = poolName;
activeTasks = Metrics.register(factory.createMetricName("ActiveTasks"), new Gauge<Integer>()
{
public Integer getValue()
{
return executor.getActiveCount();
}
});
totalBlocked = Metrics.counter(factory.createMetricName("TotalBlockedTasks"));
currentBlocked = Metrics.counter(factory.createMetricName("CurrentlyBlockedTasks"));
completedTasks = Metrics.register(factory.createMetricName("CompletedTasks"), new Gauge<Long>()
{
public Long getValue()
{
return executor.getCompletedTaskCount();
}
});
pendingTasks = Metrics.register(factory.createMetricName("PendingTasks"), new Gauge<Long>()
{
public Long getValue()
{
return executor.getTaskCount() - executor.getCompletedTaskCount();
}
});
maxPoolSize = Metrics.register(factory.createMetricName("MaxPoolSize"), new Gauge<Integer>()
{
public Integer getValue()
{
return executor.getMaximumPoolSize();
}
});
totalBlocked = new Counter();
currentBlocked = new Counter();
activeTasks = executor::getActiveTaskCount;
pendingTasks = executor::getPendingTaskCount;
completedTasks = executor::getCompletedTaskCount;
maxPoolSize = executor::getMaximumPoolSize;
maxTasksQueued = executor::getMaxTasksQueued;
}
public ThreadPoolMetrics register()
{
Metrics.register(makeMetricName(path, poolName, ACTIVE_TASKS), activeTasks);
Metrics.register(makeMetricName(path, poolName, PENDING_TASKS), pendingTasks);
Metrics.register(makeMetricName(path, poolName, COMPLETED_TASKS), completedTasks);
Metrics.register(makeMetricName(path, poolName, CURRENTLY_BLOCKED_TASKS), currentBlocked);
Metrics.register(makeMetricName(path, poolName, TOTAL_BLOCKED_TASKS), totalBlocked);
Metrics.register(makeMetricName(path, poolName, MAX_POOL_SIZE), maxPoolSize);
Metrics.register(makeMetricName(path, poolName, MAX_TASKS_QUEUED), maxTasksQueued);
return Metrics.register(this);
}
public void release()
{
Metrics.remove(factory.createMetricName("ActiveTasks"));
Metrics.remove(factory.createMetricName("PendingTasks"));
Metrics.remove(factory.createMetricName("CompletedTasks"));
Metrics.remove(factory.createMetricName("TotalBlockedTasks"));
Metrics.remove(factory.createMetricName("CurrentlyBlockedTasks"));
Metrics.remove(factory.createMetricName("MaxPoolSize"));
Metrics.remove(makeMetricName(path, poolName, ACTIVE_TASKS));
Metrics.remove(makeMetricName(path, poolName, PENDING_TASKS));
Metrics.remove(makeMetricName(path, poolName, COMPLETED_TASKS));
Metrics.remove(makeMetricName(path, poolName, CURRENTLY_BLOCKED_TASKS));
Metrics.remove(makeMetricName(path, poolName, TOTAL_BLOCKED_TASKS));
Metrics.remove(makeMetricName(path, poolName, MAX_POOL_SIZE));
Metrics.remove(makeMetricName(path, poolName, MAX_TASKS_QUEUED));
Metrics.remove(this);
}
public static Object getJmxMetric(MBeanServerConnection mbeanServerConn, String jmxPath, String poolName, String metricName)
private static MetricName makeMetricName(String path, String poolName, String metricName)
{
String name = String.format("org.apache.cassandra.metrics:type=ThreadPools,path=%s,scope=%s,name=%s", jmxPath, poolName, metricName);
try
{
ObjectName oName = new ObjectName(name);
if (!mbeanServerConn.isRegistered(oName))
{
return "N/A";
}
switch (metricName)
{
case "ActiveTasks":
case "PendingTasks":
case "CompletedTasks":
return JMX.newMBeanProxy(mbeanServerConn, oName, JmxReporter.JmxGaugeMBean.class).getValue();
case "TotalBlockedTasks":
case "CurrentlyBlockedTasks":
return JMX.newMBeanProxy(mbeanServerConn, oName, JmxReporter.JmxCounterMBean.class).getCount();
default:
throw new AssertionError("Unknown metric name " + metricName);
}
}
catch (Exception e)
{
throw new RuntimeException("Error reading: " + name, e);
}
return new MetricName("org.apache.cassandra.metrics",
"ThreadPools",
metricName,
path + '.' + poolName,
format("org.apache.cassandra.metrics:type=ThreadPools,path=%s,scope=%s,name=%s",
path, poolName, metricName));
}
public static Multimap<String, String> getJmxThreadPools(MBeanServerConnection mbeanServerConn)
{
try
{
Multimap<String, String> threadPools = HashMultimap.create();
Set<ObjectName> threadPoolObjectNames = mbeanServerConn.queryNames(new ObjectName("org.apache.cassandra.metrics:type=ThreadPools,*"),
null);
for (ObjectName oName : threadPoolObjectNames)
{
threadPools.put(oName.getKeyProperty("path"), oName.getKeyProperty("scope"));
}
return threadPools;
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException("Bad query to JMX server: ", e);
}
catch (IOException e)
{
throw new RuntimeException("Error getting threadpool names from JMX", e);
}
}
}

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

@ -87,8 +87,10 @@ import org.apache.cassandra.streaming.StreamManagerMBean;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.streaming.management.StreamStateCompositeData;
import com.codahale.metrics.JmxReporter;
import com.google.common.base.Function;
import com.google.common.base.Strings;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
@ -1349,9 +1351,64 @@ public class NodeProbe implements AutoCloseable
}
}
private static Multimap<String, String> getJmxThreadPools(MBeanServerConnection mbeanServerConn)
{
try
{
Multimap<String, String> threadPools = HashMultimap.create();
Set<ObjectName> threadPoolObjectNames = mbeanServerConn.queryNames(
new ObjectName("org.apache.cassandra.metrics:type=ThreadPools,*"),
null);
for (ObjectName oName : threadPoolObjectNames)
{
threadPools.put(oName.getKeyProperty("path"), oName.getKeyProperty("scope"));
}
return threadPools;
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException("Bad query to JMX server: ", e);
}
catch (IOException e)
{
throw new RuntimeException("Error getting threadpool names from JMX", e);
}
}
public Object getThreadPoolMetric(String pathName, String poolName, String metricName)
{
return ThreadPoolMetrics.getJmxMetric(mbeanServerConn, pathName, poolName, metricName);
String name = String.format("org.apache.cassandra.metrics:type=ThreadPools,path=%s,scope=%s,name=%s",
pathName, poolName, metricName);
try
{
ObjectName oName = new ObjectName(name);
if (!mbeanServerConn.isRegistered(oName))
{
return "N/A";
}
switch (metricName)
{
case ThreadPoolMetrics.ACTIVE_TASKS:
case ThreadPoolMetrics.PENDING_TASKS:
case ThreadPoolMetrics.COMPLETED_TASKS:
case ThreadPoolMetrics.MAX_POOL_SIZE:
return JMX.newMBeanProxy(mbeanServerConn, oName, JmxReporter.JmxGaugeMBean.class).getValue();
case ThreadPoolMetrics.TOTAL_BLOCKED_TASKS:
case ThreadPoolMetrics.CURRENTLY_BLOCKED_TASKS:
return JMX.newMBeanProxy(mbeanServerConn, oName, JmxReporter.JmxCounterMBean.class).getCount();
default:
throw new AssertionError("Unknown metric name " + metricName);
}
}
catch (Exception e)
{
throw new RuntimeException("Error reading: " + name, e);
}
}
/**
@ -1360,7 +1417,7 @@ public class NodeProbe implements AutoCloseable
*/
public Multimap<String, String> getThreadPools()
{
return ThreadPoolMetrics.getJmxThreadPools(mbeanServerConn);
return getJmxThreadPools(mbeanServerConn);
}
public int getNumberOfTables()

25
src/java/org/apache/cassandra/utils/StatusLogger.java
View File

@ -17,13 +17,10 @@
*/
package org.apache.cassandra.utils;
import java.lang.management.ManagementFactory;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import javax.management.*;
import org.apache.cassandra.cache.*;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.slf4j.Logger;
@ -64,20 +61,18 @@ public class StatusLogger
private static void logStatus()
{
MBeanServer server = ManagementFactory.getPlatformMBeanServer();
// everything from o.a.c.concurrent
logger.info(String.format("%-25s%10s%10s%15s%10s%18s", "Pool Name", "Active", "Pending", "Completed", "Blocked", "All Time Blocked"));
logger.info(String.format("%-28s%10s%10s%15s%10s%18s", "Pool Name", "Active", "Pending", "Completed", "Blocked", "All Time Blocked"));
for (Map.Entry<String, String> tpool : ThreadPoolMetrics.getJmxThreadPools(server).entries())
for (ThreadPoolMetrics tpool : CassandraMetricsRegistry.Metrics.allThreadPoolMetrics())
{
logger.info(String.format("%-25s%10s%10s%15s%10s%18s%n",
tpool.getValue(),
ThreadPoolMetrics.getJmxMetric(server, tpool.getKey(), tpool.getValue(), "ActiveTasks"),
ThreadPoolMetrics.getJmxMetric(server, tpool.getKey(), tpool.getValue(), "PendingTasks"),
ThreadPoolMetrics.getJmxMetric(server, tpool.getKey(), tpool.getValue(), "CompletedTasks"),
ThreadPoolMetrics.getJmxMetric(server, tpool.getKey(), tpool.getValue(), "CurrentlyBlockedTasks"),
ThreadPoolMetrics.getJmxMetric(server, tpool.getKey(), tpool.getValue(), "TotalBlockedTasks")));
logger.info(String.format("%-28s%10s%10s%15s%10s%18s",
tpool.poolName,
tpool.activeTasks.getValue(),
tpool.pendingTasks.getValue(),
tpool.completedTasks.getValue(),
tpool.currentBlocked.getCount(),
tpool.totalBlocked.getCount()));
}
// one offs

4
test/long/org/apache/cassandra/cql3/ViewLongTest.java
View File

@ -407,8 +407,8 @@ public class ViewLongTest extends CQLTester
private void updateViewWithFlush(String query, boolean flush, Object... params) throws Throwable
{
executeNet(protocolVersion, query, params);
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTasks() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveCount() == 0))
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTaskCount() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveTaskCount() == 0))
{
Thread.sleep(1);
}

4
test/unit/org/apache/cassandra/cql3/ViewComplexTest.java
View File

@ -71,8 +71,8 @@ public class ViewComplexTest extends CQLTester
private void updateViewWithFlush(String query, boolean flush, Object... params) throws Throwable
{
executeNet(protocolVersion, query, params);
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTasks() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveCount() == 0))
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTaskCount() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveTaskCount() == 0))
{
Thread.sleep(1);
}

4
test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java
View File

@ -80,8 +80,8 @@ public class ViewFilteringTest extends CQLTester
private void updateView(String query, Object... params) throws Throwable
{
executeNet(protocolVersion, query, params);
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTasks() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveCount() == 0))
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTaskCount() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveTaskCount() == 0))
{
Thread.sleep(1);
}

4
test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java
View File

@ -84,8 +84,8 @@ public class ViewSchemaTest extends CQLTester
private void updateView(String query, Object... params) throws Throwable
{
executeNet(protocolVersion, query, params);
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTasks() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveCount() == 0))
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTaskCount() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveTaskCount() == 0))
{
Thread.sleep(1);
}

4
test/unit/org/apache/cassandra/cql3/ViewTest.java
View File

@ -83,8 +83,8 @@ public class ViewTest extends CQLTester
private void updateView(String query, Object... params) throws Throwable
{
executeNet(protocolVersion, query, params);
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTasks() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveCount() == 0))
while (!(((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getPendingTaskCount() == 0
&& ((SEPExecutor) StageManager.getStage(Stage.VIEW_MUTATION)).getActiveTaskCount() == 0))
{
Thread.sleep(1);
}