Repository: cassandra Updated Branches: refs/heads/trunk fcb87a2d1 -> d5ae2ae48
Add a virtual table to expose thread pools patch by Chris Lohfink; reviewed by Aleksey Yeschenko for CASSANDRA-14523 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d5ae2ae4 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d5ae2ae4 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d5ae2ae4 Branch: refs/heads/trunk Commit: d5ae2ae481545b1fb2332b46013088f2f8cea636 Parents: fcb87a2 Author: Chris Lohfink <[email protected]> Authored: Fri Jun 22 10:48:01 2018 -0500 Committer: Aleksey Yeshchenko <[email protected]> Committed: Mon Aug 6 19:06:16 2018 +0100 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../DebuggableThreadPoolExecutor.java | 12 ++ .../JMXEnabledThreadPoolExecutor.java | 2 +- .../concurrent/LocalAwareExecutorService.java | 46 ++++- .../cassandra/concurrent/SEPExecutor.java | 28 ++- .../apache/cassandra/concurrent/SEPWorker.java | 2 +- .../cassandra/concurrent/StageManager.java | 12 ++ .../db/virtual/SystemViewsKeyspace.java | 5 +- .../cassandra/db/virtual/ThreadPoolsTable.java | 85 +++++++++ .../metrics/CassandraMetricsRegistry.java | 33 +++- .../apache/cassandra/metrics/SEPMetrics.java | 109 ------------ .../metrics/ThreadPoolMetricNameFactory.java | 45 ----- .../cassandra/metrics/ThreadPoolMetrics.java | 171 +++++++------------ .../org/apache/cassandra/tools/NodeProbe.java | 61 ++++++- .../apache/cassandra/utils/StatusLogger.java | 25 ++- .../org/apache/cassandra/cql3/ViewLongTest.java | 4 +- .../apache/cassandra/cql3/ViewComplexTest.java | 4 +- .../cassandra/cql3/ViewFilteringTest.java | 4 +- .../apache/cassandra/cql3/ViewSchemaTest.java | 4 +- .../org/apache/cassandra/cql3/ViewTest.java | 4 +- 20 files changed, 354 insertions(+), 303 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 3b35608..103da7b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 92cbbf4..ef0f43c 100644 --- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java @@ -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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java index 2dafb4f..278b399 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java b/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java index 5577d59..4f07c6c 100644 --- a/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java +++ b/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorService.java @@ -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; + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/SEPExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java index add850a..4563fca 100644 --- a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/SEPWorker.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/SEPWorker.java b/src/java/org/apache/cassandra/concurrent/SEPWorker.java index 29a2557..4549b48 100644 --- a/src/java/org/apache/cassandra/concurrent/SEPWorker.java +++ b/src/java/org/apache/cassandra/concurrent/SEPWorker.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/concurrent/StageManager.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index 84b8da6..c102042 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -132,5 +132,17 @@ public class StageManager { execute(command); } + + @Override + public int getActiveTaskCount() + { + return getActiveCount(); + } + + @Override + public int getPendingTaskCount() + { + return getQueue().size(); + } } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java index e803d3d..3505a30 100644 --- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java @@ -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))); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java b/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java new file mode 100644 index 0000000..778e46f --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/ThreadPoolsTable.java @@ -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()); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 97a932b..79d3c3e 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -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); http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/metrics/SEPMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/SEPMetrics.java b/src/java/org/apache/cassandra/metrics/SEPMetrics.java deleted file mode 100644 index dd1d2d6..0000000 --- a/src/java/org/apache/cassandra/metrics/SEPMetrics.java +++ /dev/null @@ -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")); - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java b/src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java deleted file mode 100644 index 7810108..0000000 --- a/src/java/org/apache/cassandra/metrics/ThreadPoolMetricNameFactory.java +++ /dev/null @@ -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()); - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java b/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java index 268e878..3ba984a 100644 --- a/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ThreadPoolMetrics.java @@ -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); - - 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(); - } - }); + this.path = path; + this.poolName = poolName; + + totalBlocked = new Counter(); + currentBlocked = new Counter(); + activeTasks = executor::getActiveTaskCount; + pendingTasks = executor::getPendingTaskCount; + completedTasks = executor::getCompletedTaskCount; + maxPoolSize = executor::getMaximumPoolSize; + maxTasksQueued = executor::getMaxTasksQueued; } - public void release() + public ThreadPoolMetrics register() { - 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.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 static Object getJmxMetric(MBeanServerConnection mbeanServerConn, String jmxPath, String poolName, String metricName) + public void release() { - 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); - } + 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 Multimap<String, String> getJmxThreadPools(MBeanServerConnection mbeanServerConn) + private static MetricName makeMetricName(String path, String poolName, String metricName) { - 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); - } + 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)); } - } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/tools/NodeProbe.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index caaa337..63efe93 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -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() http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/src/java/org/apache/cassandra/utils/StatusLogger.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/StatusLogger.java b/src/java/org/apache/cassandra/utils/StatusLogger.java index 9f9d869..dcb1135 100644 --- a/src/java/org/apache/cassandra/utils/StatusLogger.java +++ b/src/java/org/apache/cassandra/utils/StatusLogger.java @@ -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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/test/long/org/apache/cassandra/cql3/ViewLongTest.java ---------------------------------------------------------------------- diff --git a/test/long/org/apache/cassandra/cql3/ViewLongTest.java b/test/long/org/apache/cassandra/cql3/ViewLongTest.java index dd8fc9a..9d97729 100644 --- a/test/long/org/apache/cassandra/cql3/ViewLongTest.java +++ b/test/long/org/apache/cassandra/cql3/ViewLongTest.java @@ -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); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java b/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java index 32319a9..d24ab52 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewComplexTest.java @@ -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); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java b/test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java index 65d66b5..8b4a556 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewFilteringTest.java @@ -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); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java b/test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java index 093f122..2422ef5 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewSchemaTest.java @@ -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); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/d5ae2ae4/test/unit/org/apache/cassandra/cql3/ViewTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/ViewTest.java b/test/unit/org/apache/cassandra/cql3/ViewTest.java index aae7d83..4f99d03 100644 --- a/test/unit/org/apache/cassandra/cql3/ViewTest.java +++ b/test/unit/org/apache/cassandra/cql3/ViewTest.java @@ -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); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
