This is an automated email from the ASF dual-hosted git repository.
leventov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new 2803fda Added an allocation rate metric #6604 (#6710)
2803fda is described below
commit 2803fda8b781ffd13739ce49779e0e0578332d91
Author: Egor Riashin <[email protected]>
AuthorDate: Tue Jan 29 16:16:35 2019 +0300
Added an allocation rate metric #6604 (#6710)
Addressing #6604
---
.../util/metrics/AllocationMetricCollector.java | 90 +++++++++++++++++
.../util/metrics/AllocationMetricCollectors.java | 59 +++++++++++
.../apache/druid/java/util/metrics/JvmMonitor.java | 14 +++
.../metrics/AllocationMetricCollectorTest.java | 108 +++++++++++++++++++++
4 files changed, 271 insertions(+)
diff --git
a/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollector.java
b/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollector.java
new file mode 100644
index 0000000..a0a9d7f
--- /dev/null
+++
b/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollector.java
@@ -0,0 +1,90 @@
+/*
+ * 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.druid.java.util.metrics;
+
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.lang.management.ThreadMXBean;
+import java.lang.reflect.Method;
+
+class AllocationMetricCollector
+{
+ private static final Logger log = new
Logger(AllocationMetricCollector.class);
+
+ private static final int NO_DATA = -1;
+
+ private final Method getThreadAllocatedBytes;
+ private final ThreadMXBean threadMXBean;
+
+ private Long2LongMap previousResults;
+
+ AllocationMetricCollector(Method method, ThreadMXBean threadMXBean)
+ {
+ this.getThreadAllocatedBytes = method;
+ this.threadMXBean = threadMXBean;
+
+ previousResults = new Long2LongOpenHashMap();
+ previousResults.defaultReturnValue(NO_DATA);
+ }
+
+ /**
+ * Uses getThreadAllocatedBytes internally {@link
com.sun.management.ThreadMXBean#getThreadAllocatedBytes}.
+ *
+ * Tests show the call to getThreadAllocatedBytes for a single thread ID out
of 500 threads running takes around
+ * 9000 ns (in the worst case), which for 500 IDs should take
500*9000/1000/1000 = 4.5 ms to the max.
+ * AllocationMetricCollector takes linear time to calculate delta, for 500
threads it's negligible.
+ * See the default emitting period {@link
MonitorSchedulerConfig#getEmitterPeriod}.
+ *
+ * @return all threads summed allocated bytes delta
+ */
+ long calculateDelta()
+ {
+ try {
+ long[] allThreadIds = threadMXBean.getAllThreadIds();
+ // the call time depends on number of threads, for 500 threads the
estimated time is 4 ms
+ long[] bytes = (long[]) getThreadAllocatedBytes.invoke(threadMXBean,
(Object) allThreadIds);
+ long sum = 0;
+ Long2LongMap newResults = new Long2LongOpenHashMap();
+ newResults.defaultReturnValue(NO_DATA);
+ for (int i = 0; i < allThreadIds.length; i++) {
+ long threadId = allThreadIds[i];
+ long previous = previousResults.get(threadId);
+ long current = bytes[i];
+ newResults.put(threadId, current);
+ // a) some threads can be terminated and their ids won't be present
+ // b) if new threads ids can collide with terminated threads ids then
the current allocation can be lesser than
+ // before
+ if (previous == NO_DATA || previous > current) {
+ sum += current;
+ } else {
+ sum += current - previous;
+ }
+ }
+ previousResults = newResults;
+ return sum;
+ }
+ catch (ReflectiveOperationException e) {
+ log.error(e, "Cannot calculate delta"); // it doesn't make sense after
initialization is complete
+ }
+ return 0;
+ }
+}
diff --git
a/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollectors.java
b/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollectors.java
new file mode 100644
index 0000000..36ffb09
--- /dev/null
+++
b/core/src/main/java/org/apache/druid/java/util/metrics/AllocationMetricCollectors.java
@@ -0,0 +1,59 @@
+/*
+ * 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.druid.java.util.metrics;
+
+import org.apache.druid.java.util.common.logger.Logger;
+
+import javax.annotation.Nullable;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
+import java.lang.reflect.Method;
+
+class AllocationMetricCollectors
+{
+ private static final Logger log = new
Logger(AllocationMetricCollectors.class);
+ private static Method getThreadAllocatedBytes;
+ private static ThreadMXBean threadMXBean;
+ private static boolean initialized = false;
+
+ static {
+ try {
+ // classes in the sun.* packages are not part of the public/supported
Java API and should not be used directly.
+ threadMXBean = ManagementFactory.getThreadMXBean();
+ getThreadAllocatedBytes =
threadMXBean.getClass().getMethod("getThreadAllocatedBytes", long[].class);
+ getThreadAllocatedBytes.setAccessible(true);
+ getThreadAllocatedBytes.invoke(threadMXBean, (Object)
threadMXBean.getAllThreadIds());
+ initialized = true;
+ }
+ catch (Exception e) {
+ log.warn(e, "Cannot initialize %s",
AllocationMetricCollector.class.getName());
+ }
+ }
+
+ @Nullable
+ static AllocationMetricCollector getAllocationMetricCollector()
+ {
+ if (initialized) {
+ return new AllocationMetricCollector(getThreadAllocatedBytes,
threadMXBean);
+ }
+ return null;
+ }
+}
diff --git
a/core/src/main/java/org/apache/druid/java/util/metrics/JvmMonitor.java
b/core/src/main/java/org/apache/druid/java/util/metrics/JvmMonitor.java
index ea47e34..118ed31 100644
--- a/core/src/main/java/org/apache/druid/java/util/metrics/JvmMonitor.java
+++ b/core/src/main/java/org/apache/druid/java/util/metrics/JvmMonitor.java
@@ -45,6 +45,8 @@ public class JvmMonitor extends FeedDefiningMonitor
private final GcCounters gcCounters = new GcCounters();
+ private final AllocationMetricCollector collector;
+
public JvmMonitor()
{
this(ImmutableMap.of());
@@ -66,6 +68,7 @@ public class JvmMonitor extends FeedDefiningMonitor
Preconditions.checkNotNull(dimensions);
this.dimensions = ImmutableMap.copyOf(dimensions);
this.pid = Preconditions.checkNotNull(pidDiscoverer).getPid();
+ this.collector = AllocationMetricCollectors.getAllocationMetricCollector();
}
@Override
@@ -74,10 +77,21 @@ public class JvmMonitor extends FeedDefiningMonitor
emitJvmMemMetrics(emitter);
emitDirectMemMetrics(emitter);
emitGcMetrics(emitter);
+ emitThreadAllocationMetrics(emitter);
return true;
}
+ private void emitThreadAllocationMetrics(ServiceEmitter emitter)
+ {
+ final ServiceMetricEvent.Builder builder = builder();
+ MonitorUtils.addDimensionsToBuilder(builder, dimensions);
+ if (collector != null) {
+ long delta = collector.calculateDelta();
+ emitter.emit(builder.build("jvm/heapAlloc/bytes", delta));
+ }
+ }
+
/**
* These metrics are going to be replaced by new jvm/gc/mem/* metrics
*/
diff --git
a/core/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
b/core/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
new file mode 100644
index 0000000..c32029a
--- /dev/null
+++
b/core/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.druid.java.util.metrics;
+
+import org.apache.druid.java.util.common.logger.Logger;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+public class AllocationMetricCollectorTest
+{
+ private static final Logger log = new
Logger(AllocationMetricCollectorTest.class);
+ private final List<Thread> threads = new ArrayList<>();
+ private final int objectHeader64BitSize = 16;
+
+ /**
+ * Test a calculated delta is larger than objects size generated by this
method.
+ * @throws InterruptedException
+ */
+ @SuppressWarnings("OptionalIsPresent")
+ @Test
+ public void testDelta() throws InterruptedException
+ {
+ AllocationMetricCollector collector =
AllocationMetricCollectors.getAllocationMetricCollector();
+ if (collector == null) {
+ return;
+ }
+
+ long delta = collector.calculateDelta();
+ Assert.assertNotNull(delta);
+ Assert.assertTrue(delta > 0);
+ log.info("First delta: %s", delta);
+
+ int generatedSize2 = generateObjectsConcurrently(1000);
+ long delta2 = collector.calculateDelta();
+ Assert.assertTrue(delta2 > generatedSize2);
+ log.info("Second delta: %s", delta2);
+
+ int generatedSize3 = generateObjectsConcurrently(100000);
+ long delta3 = collector.calculateDelta();
+ Assert.assertTrue(delta3 > generatedSize3);
+ log.info("Third delta: %s", delta3);
+ }
+
+ private int generateObjectsConcurrently(int countPerThread) throws
InterruptedException
+ {
+ int threads = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);
+ log.info("Threads: %s", threads);
+ int objectsCount = countPerThread * threads;
+ log.info("Total objects: %s", objectsCount);
+ int totalSize = objectsCount * objectHeader64BitSize;
+ log.info("Total size: %s", totalSize);
+ CountDownLatch countDownLatch = new CountDownLatch(threads);
+ for (int i = 0; i < threads; i++) {
+ Thread thread = new Thread(() -> {
+ @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
+ List<Object> list = new ArrayList<>(countPerThread);
+ for (int j = 0; j < countPerThread; j++) {
+ list.add(new Object());
+ }
+ countDownLatch.countDown();
+ //noinspection EmptyCatchBlock,UnusedCatchParameter
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ }
+ catch (InterruptedException e) {
+ }
+ });
+ thread.setDaemon(true);
+ thread.start();
+ this.threads.add(thread);
+
+ }
+ countDownLatch.await();
+ return totalSize;
+ }
+
+ @After
+ public void stopThreads()
+ {
+ // threads are in sleep so that their ids are still present in JVM
allocation "registry"
+ // so stop them manually
+ for (Thread thread : threads) {
+ thread.interrupt();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
