Add a way to log hot threads in plain text (#111053)

This adds a way to log current threads in plain text to logs. This way we do not need to decode them and can search by stack trace in logs (for example to know if the issue is recurring). Please note, this produces a multi-line log entry.

Author: idegtiarenko

server/src/internalClusterTest/java/org/elasticsearch/action/admin/HotThreadsIT.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.monitor.jvm.HotThreads;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.hamcrest.Matcher;
 
@@ -31,6 +32,7 @@
 import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
 import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import static org.elasticsearch.test.MockLog.assertThatLogger;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
 import static org.hamcrest.CoreMatchers.equalTo;
@@ -211,4 +213,25 @@ public void testLogLocalHotThreads() {
             )
         );
     }
+
+    @TestLogging(reason = "testing logging at various levels", value = "org.elasticsearch.action.admin.HotThreadsIT:TRACE")
+    public void testLogLocalCurrentThreadsInPlainText() {
+        final var level = randomFrom(Level.TRACE, Level.DEBUG, Level.INFO, Level.WARN, Level.ERROR);
+        assertThatLogger(
+            () -> HotThreads.logLocalCurrentThreads(logger, level, getTestName()),
+            HotThreadsIT.class,
+            new MockLog.SeenEventExpectation(
+                "Should log hot threads header in plain text",
+                HotThreadsIT.class.getCanonicalName(),
+                level,
+                "testLogLocalCurrentThreadsInPlainText: Hot threads at"
+            ),
+            new MockLog.SeenEventExpectation(
+                "Should log hot threads cpu usage in plain text",
+                HotThreadsIT.class.getCanonicalName(),
+                level,
+                "cpu usage by thread"
+            )
+        );
+    }
 }

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

@@ -28,6 +28,7 @@
 
 import java.io.IOException;
 import java.io.OutputStreamWriter;
+import java.io.StringWriter;
 import java.io.Writer;
 import java.lang.management.ManagementFactory;
 import java.lang.management.ThreadInfo;
@@ -105,6 +106,33 @@ public static void logLocalHotThreads(Logger logger, Level level, String prefix,
         }
     }
 
+    /**
+     * Capture and log the current threads on the local node. Unlike hot threads this does not sample and captures current state only.
+     * Useful for capturing stack traces for unexpectedly-slow operations in production. The resulting message might be large, so it is
+     * split per thread and logged as multiple entries.
+     *
+     * @param logger        The logger to use for the logging
+     * @param level         The log level to use for the logging.
+     * @param prefix        The prefix to emit on each chunk of the logging.
+     */
+    public static void logLocalCurrentThreads(Logger logger, Level level, String prefix) {
+        if (logger.isEnabled(level) == false) {
+            return;
+        }
+
+        try (var writer = new StringWriter()) {
+            new HotThreads().busiestThreads(500).threadElementsSnapshotCount(1).detect(writer, () -> {
+                logger.log(level, "{}: {}", prefix, writer.toString());
+                writer.getBuffer().setLength(0);
+            });
+        } catch (Exception e) {
+            logger.error(
+                () -> org.elasticsearch.common.Strings.format("failed to write local current threads with prefix [%s]", prefix),
+                e
+            );
+        }
+    }
+
     public enum ReportType {
 
         CPU("cpu"),
@@ -192,11 +220,12 @@ public HotThreads sortOrder(SortOrder order) {
     }
 
     public void detect(Writer writer) throws Exception {
+        detect(writer, () -> {});
+    }
+
+    public void detect(Writer writer, Runnable onNextThread) throws Exception {
         synchronized (mutex) {
-            innerDetect(ManagementFactory.getThreadMXBean(), SunThreadInfo.INSTANCE, Thread.currentThread().getId(), (interval) -> {
-                Thread.sleep(interval);
-                return null;
-            }, writer);
+            innerDetect(ManagementFactory.getThreadMXBean(), SunThreadInfo.INSTANCE, Thread.currentThread().getId(), writer, onNextThread);
         }
     }
 
@@ -245,13 +274,15 @@ Map<Long, ThreadTimeAccumulator> getAllValidThreadInfos(ThreadMXBean threadBean,
 
     ThreadInfo[][] captureThreadStacks(ThreadMXBean threadBean, long[] threadIds) throws InterruptedException {
         ThreadInfo[][] result = new ThreadInfo[threadElementsSnapshotCount][];
-        for (int j = 0; j < threadElementsSnapshotCount; j++) {
-            // NOTE, javadoc of getThreadInfo says: If a thread of the given ID is not alive or does not exist,
-            // null will be set in the corresponding element in the returned array. A thread is alive if it has
-            // been started and has not yet died.
+
+        // NOTE, javadoc of getThreadInfo says: If a thread of the given ID is not alive or does not exist,
+        // null will be set in the corresponding element in the returned array. A thread is alive if it has
+        // been started and has not yet died.
+        for (int j = 0; j < threadElementsSnapshotCount - 1; j++) {
             result[j] = threadBean.getThreadInfo(threadIds, Integer.MAX_VALUE);
             Thread.sleep(threadElementsSnapshotDelay.millis());
         }
+        result[threadElementsSnapshotCount - 1] = threadBean.getThreadInfo(threadIds, Integer.MAX_VALUE);
 
         return result;
     }
@@ -267,13 +298,8 @@ private double getTimeSharePercentage(long time) {
         return (((double) time) / interval.nanos()) * 100;
     }
 
-    void innerDetect(
-        ThreadMXBean threadBean,
-        SunThreadInfo sunThreadInfo,
-        long currentThreadId,
-        SleepFunction<Long, Void> threadSleep,
-        Writer writer
-    ) throws Exception {
+    void innerDetect(ThreadMXBean threadBean, SunThreadInfo sunThreadInfo, long currentThreadId, Writer writer, Runnable onNextThread)
+        throws Exception {
         if (threadBean.isThreadCpuTimeSupported() == false) {
             throw new ElasticsearchException("thread CPU time is not supported on this JDK");
         }
@@ -297,10 +323,11 @@ void innerDetect(
             .append(", ignoreIdleThreads=")
             .append(Boolean.toString(ignoreIdleThreads))
             .append(":\n");
+        onNextThread.run();
 
         // Capture before and after thread state with timings
         Map<Long, ThreadTimeAccumulator> previousThreadInfos = getAllValidThreadInfos(threadBean, sunThreadInfo, currentThreadId);
-        threadSleep.apply(interval.millis());
+        Thread.sleep(interval.millis());
         Map<Long, ThreadTimeAccumulator> latestThreadInfos = getAllValidThreadInfos(threadBean, sunThreadInfo, currentThreadId);
 
         latestThreadInfos.forEach((threadId, accumulator) -> accumulator.subtractPrevious(previousThreadInfos.get(threadId)));
@@ -430,6 +457,7 @@ void innerDetect(
                     }
                 }
             }
+            onNextThread.run();
         }
     }
 

server/src/test/java/org/elasticsearch/monitor/jvm/HotThreadsTests.java

@@ -947,7 +947,7 @@ private static String innerDetect(
         long currentThreadId
     ) throws Exception {
         try (var writer = new StringWriter()) {
-            hotThreads.innerDetect(mockedMthreadMXBeanBean, sunThreadInfo, currentThreadId, (interval) -> null, writer);
+            hotThreads.innerDetect(mockedMthreadMXBeanBean, sunThreadInfo, currentThreadId, writer, () -> {});
             return writer.toString();
         }
     }