Add ignore_idle_threads (default: true) to hot threads

docs/reference/cluster/nodes-hot-threads.asciidoc

@@ -14,3 +14,5 @@ threads. Parameters allowed are:
 				Defaults to 500ms.
 `type`:: 		The type to sample, defaults to cpu, but supports wait and
 				block to see hot threads that are in wait or block state.
+`ignore_idle_threads`::    If true, known idle threads (e.g. waiting in a socket select, or to
+			   get a task from an empty queue) are filtered out.  Defaults to true.

rest-api-spec/api/nodes.hot_threads.json

@@ -24,6 +24,10 @@
           "type" : "number",
           "description" : "Specify the number of threads to provide information for (default: 3)"
         },
+	"ignore_idle_threads": {
+          "type" : "boolean",
+          "description" : "Don't show threads that are in known-idle places, such as waiting on a socket select or pulling from an empty task queue (default: true)"
+        },
         "type": {
           "type" : "enum",
           "options" : ["cpu", "wait", "block"],

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.admin.cluster.node.hotthreads;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.support.nodes.NodesOperationRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -35,6 +36,7 @@ public class NodesHotThreadsRequest extends NodesOperationRequest<NodesHotThread
     String type = "cpu";
     TimeValue interval = new TimeValue(500, TimeUnit.MILLISECONDS);
     int snapshots = 10;
+    boolean ignoreIdleThreads = true;
 
     /**
      * Get hot threads from nodes based on the nodes ids specified. If none are passed, hot
@@ -53,6 +55,15 @@ public NodesHotThreadsRequest threads(int threads) {
         return this;
     }
 
+    public boolean ignoreIdleThreads() {
+        return this.ignoreIdleThreads;
+    }
+
+    public NodesHotThreadsRequest ignoreIdleThreads(boolean ignoreIdleThreads) {
+        this.ignoreIdleThreads = ignoreIdleThreads;
+        return this;
+    }
+
     public NodesHotThreadsRequest type(String type) {
         this.type = type;
         return this;
@@ -84,6 +95,12 @@ public NodesHotThreadsRequest snapshots(int snapshots) {
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         threads = in.readInt();
+        if (in.getVersion().before(Version.V_1_5_0)) {
+            // Pre-1.5.0 did not filter hot threads, so we shouldn't:
+            ignoreIdleThreads = false;
+        } else {
+            ignoreIdleThreads = in.readBoolean();
+        }
         type = in.readString();
         interval = TimeValue.readTimeValue(in);
         snapshots = in.readInt();
@@ -93,6 +110,9 @@ public void readFrom(StreamInput in) throws IOException {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeInt(threads);
+        if (out.getVersion().onOrAfter(Version.V_1_5_0)) {
+            out.writeBoolean(ignoreIdleThreads);
+        }
         out.writeString(type);
         interval.writeTo(out);
         out.writeInt(snapshots);

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequestBuilder.java

@@ -37,6 +37,11 @@ public NodesHotThreadsRequestBuilder setThreads(int threads) {
         return this;
     }
 
+    public NodesHotThreadsRequestBuilder setIgnoreIdleThreads(boolean ignoreIdleThreads) {
+        request.ignoreIdleThreads(ignoreIdleThreads);
+        return this;
+    }
+
     public NodesHotThreadsRequestBuilder setType(String type) {
         request.type(type);
         return this;

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java

@@ -92,7 +92,8 @@ protected NodeHotThreads nodeOperation(NodeRequest request) throws Elasticsearch
                 .busiestThreads(request.request.threads)
                 .type(request.request.type)
                 .interval(request.request.interval)
-                .threadElementsSnapshotCount(request.request.snapshots);
+                .threadElementsSnapshotCount(request.request.snapshots)
+                .ignoreIdleThreads(request.request.ignoreIdleThreads);
         try {
             return new NodeHotThreads(clusterService.localNode(), hotThreads.detect());
         } catch (Exception e) {
@@ -130,4 +131,4 @@ public void writeTo(StreamOutput out) throws IOException {
             request.writeTo(out);
         }
     }
-}
+}

src/main/java/org/elasticsearch/monitor/jvm/HotThreads.java

@@ -40,6 +40,7 @@ public class HotThreads {
     private TimeValue threadElementsSnapshotDelay = new TimeValue(10);
     private int threadElementsSnapshotCount = 10;
     private String type = "cpu";
+    private boolean ignoreIdleThreads = true;
 
     public HotThreads interval(TimeValue interval) {
         this.interval = interval;
@@ -51,6 +52,11 @@ public HotThreads busiestThreads(int busiestThreads) {
         return this;
     }
 
+    public HotThreads ignoreIdleThreads(boolean ignoreIdleThreads) {
+        this.ignoreIdleThreads = ignoreIdleThreads;
+        return this;
+    }
+
     public HotThreads threadElementsSnapshotDelay(TimeValue threadElementsSnapshotDelay) {
         this.threadElementsSnapshotDelay = threadElementsSnapshotDelay;
         return this;
@@ -76,6 +82,44 @@ public String detect() throws Exception {
         }
     }
 
+    private static boolean isIdleThread(ThreadInfo threadInfo) {
+        String threadName = threadInfo.getThreadName();
+
+        // NOTE: these are likely JVM dependent
+        if (threadName.equals("Signal Dispatcher") ||
+            threadName.equals("Finalizer") ||
+            threadName.equals("Reference Handler")) {
+            return true;
+        }
+
+        for (StackTraceElement frame : threadInfo.getStackTrace()) {
+            String className = frame.getClassName();
+            String methodName = frame.getMethodName();
+            if (className.equals("java.util.concurrent.ThreadPoolExecutor") &&
+                methodName.equals("getTask")) {
+                return true;
+            }
+            if (className.equals("sun.nio.ch.SelectorImpl") &&
+                methodName.equals("select")) {
+                return true;
+            }
+            if (className.equals("org.elasticsearch.threadpool.ThreadPool$EstimatedTimeThread") &&
+                methodName.equals("run")) {
+                return true;
+            }
+            if (className.equals("org.elasticsearch.indices.ttl.IndicesTTLService$Notifier") &&
+                methodName.equals("await")) {
+                return true;
+            }
+            if (className.equals("java.util.concurrent.LinkedTransferQueue") &&
+                methodName.equals("poll")) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
     private String innerDetect() throws Exception {
         StringBuilder sb = new StringBuilder();
         ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
@@ -168,18 +212,18 @@ public int compare(MyThreadInfo o1, MyThreadInfo o2) {
                     time = hotties.get(t).blockedTime;
                 }
                 String threadName = null;
-                if (allInfos[0][t] == null) {
-                    for (ThreadInfo[] info : allInfos) {
-                        if (info != null && info[t] != null) {
-                            threadName = info[t].getThreadName();
-                            break;
+                for (ThreadInfo[] info : allInfos) {
+                    if (info != null && info[t] != null) {
+                        if (ignoreIdleThreads && isIdleThread(info[t])) {
+                            info[t] = null;
+                            continue;
                         }
+                        threadName = info[t].getThreadName();
+                        break;
                     }
-                    if (threadName == null) {
-                        continue; // thread is not alive yet or died before the first snapshot - ignore it!
-                    }
-                } else {
-                    threadName = allInfos[0][t].getThreadName();
+                }
+                if (threadName == null) {
+                    continue; // thread is not alive yet or died before the first snapshot - ignore it!
                 }
                 double percent = (((double) time) / interval.nanos()) * 100;
                 sb.append(String.format(Locale.ROOT, "%n%4.1f%% (%s out of %s) %s usage by thread '%s'%n", percent, TimeValue.timeValueNanos(time), interval, type, threadName));
@@ -277,4 +321,4 @@ void setDelta(long cpuTime, ThreadInfo info) {
             this.info = info;
         }
     }
-}
+}

src/main/java/org/elasticsearch/rest/action/admin/cluster/node/hotthreads/RestNodesHotThreadsAction.java

@@ -54,6 +54,7 @@ public void handleRequest(final RestRequest request, final RestChannel channel,
         String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodeId"));
         NodesHotThreadsRequest nodesHotThreadsRequest = new NodesHotThreadsRequest(nodesIds);
         nodesHotThreadsRequest.threads(request.paramAsInt("threads", nodesHotThreadsRequest.threads()));
+        nodesHotThreadsRequest.ignoreIdleThreads(request.paramAsBoolean("ignore_idle_threads", nodesHotThreadsRequest.ignoreIdleThreads()));
         nodesHotThreadsRequest.type(request.param("type", nodesHotThreadsRequest.type()));
         nodesHotThreadsRequest.interval(TimeValue.parseTimeValue(request.param("interval"), nodesHotThreadsRequest.interval()));
         nodesHotThreadsRequest.snapshots(request.paramAsInt("snapshots", nodesHotThreadsRequest.snapshots()));

src/test/java/org/elasticsearch/action/admin/HotThreadsTest.java

@@ -40,6 +40,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.Matchers.lessThan;
 
 /**
  */
@@ -63,6 +64,7 @@ public void testHotThreadsDontFail() throws ExecutionException, InterruptedExcep
             if (randomBoolean()) {
                 nodesHotThreadsRequestBuilder.setThreads(rarely() ? randomIntBetween(500, 5000) : randomIntBetween(1, 500));
             }
+            nodesHotThreadsRequestBuilder.setIgnoreIdleThreads(randomBoolean());
             if (randomBoolean()) {
                 switch (randomIntBetween(0, 2)) {
                     case 2:
@@ -131,4 +133,34 @@ public void onFailure(Throwable e) {
             assertThat(hasErrors.get(), is(false));
         }
     }
+
+    public void testIgnoreIdleThreads() throws ExecutionException, InterruptedException {
+
+        // First time, don't ignore idle threads:
+        NodesHotThreadsRequestBuilder builder = client().admin().cluster().prepareNodesHotThreads();
+        builder.setIgnoreIdleThreads(false);
+        builder.setThreads(Integer.MAX_VALUE);
+        NodesHotThreadsResponse response = builder.execute().get();
+
+        int totSizeAll = 0;
+        for (NodeHotThreads node : response.getNodesMap().values()) {
+            totSizeAll += node.getHotThreads().length();
+        }
+
+        // Second time, do ignore idle threads:
+        builder = client().admin().cluster().prepareNodesHotThreads();
+        builder.setThreads(Integer.MAX_VALUE);
+
+        // Make sure default is true:
+        assertEquals(true, builder.request().ignoreIdleThreads());
+        response = builder.execute().get();
+
+        int totSizeIgnoreIdle = 0;
+        for (NodeHotThreads node : response.getNodesMap().values()) {
+            totSizeIgnoreIdle += node.getHotThreads().length();
+        }
+
+        // The filtered stacks should be smaller than unfiltered ones:
+        assertThat(totSizeIgnoreIdle, lessThan(totSizeAll));
+    }
 }