Add runtime metrics for task time on event loop and log slow execution (prestodb#25009)

## Description
1. Add runtime metrics for task execution time on event loop
2. Add logging for slow execution where the logging threshold can be
controlled by config and we will log the query id, task id, and the
method name if the runnable is taking too long to finish.

## Motivation and Context
1. This increases the observability on what can be running slow on event
loop

## Impact
<!---Describe any public API or user-facing feature change or any
performance impact-->

## Test Plan
1. running verifier

## Contributor checklist

- [ ] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [ ] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [ ] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.

## Release Notes
Please follow [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines)
and fill in the release notes below.


```
== NO RELEASE NOTE ==
```

Author: shangm2

presto-common/src/main/java/com/facebook/presto/common/RuntimeMetricName.java

@@ -68,6 +68,8 @@ private RuntimeMetricName()
     public static final String TASK_UPDATE_DELIVERED_WALL_TIME_NANOS = "taskUpdateDeliveredWallTimeNanos";
     public static final String TASK_UPDATE_SERIALIZED_CPU_TIME_NANOS = "taskUpdateSerializedCpuNanos";
     public static final String TASK_PLAN_SERIALIZED_CPU_TIME_NANOS = "taskPlanSerializedCpuNanos";
+    // Time for event loop to execute a method
+    public static final String EVENT_LOOP_METHOD_EXECUTION_CPU_TIME_NANOS = "eventLoopMethodExecutionCpuNanos";
     // Time taken for a read call to storage
     public static final String STORAGE_READ_TIME_NANOS = "storageReadTimeNanos";
     // Size of the data retrieved by read call to storage

presto-main-base/src/main/java/com/facebook/presto/execution/SafeEventLoopGroup.java

@@ -14,17 +14,22 @@
 package com.facebook.presto.execution;
 
 import com.facebook.airlift.log.Logger;
+import com.sun.management.ThreadMXBean;
 import io.netty.channel.DefaultEventLoop;
 import io.netty.channel.DefaultEventLoopGroup;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadFactory;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
 /***
  *  One observation about event loop is if submitted task fails, it could kill the thread but the event loop group will not create a new one.
@@ -35,10 +40,15 @@ public class SafeEventLoopGroup
         extends DefaultEventLoopGroup
 {
     private static final Logger log = Logger.get(SafeEventLoopGroup.class);
+    private static final ThreadMXBean THREAD_MX_BEAN = (ThreadMXBean) ManagementFactory.getThreadMXBean();
+    private static final List<GarbageCollectorMXBean> gcBeans = ManagementFactory.getGarbageCollectorMXBeans();
 
-    public SafeEventLoopGroup(int nThreads, ThreadFactory threadFactory)
+    private final long slowMethodThresholdOnEventLoopInNanos;
+
+    public SafeEventLoopGroup(int nThreads, ThreadFactory threadFactory, long slowMethodThresholdOnEventLoopInNanos)
     {
         super(nThreads, threadFactory);
+        this.slowMethodThresholdOnEventLoopInNanos = slowMethodThresholdOnEventLoopInNanos;
     }
 
     @Override
@@ -47,7 +57,7 @@ protected EventLoop newChild(Executor executor, Object... args)
         return new SafeEventLoop(this, executor);
     }
 
-    public static class SafeEventLoop
+    public class SafeEventLoop
             extends DefaultEventLoop
     {
         public SafeEventLoop(EventLoopGroup parent, Executor executor)
@@ -73,9 +83,11 @@ protected void run()
             while (!this.confirmShutdown());
         }
 
-        public void execute(Runnable task, Consumer<Throwable> failureHandler)
+        public void execute(Runnable task, Consumer<Throwable> failureHandler, SchedulerStatsTracker statsTracker, String methodSignature)
         {
             requireNonNull(task, "task is null");
+            long initialGCTime = getTotalGCTime();
+            long start = THREAD_MX_BEAN.getCurrentThreadCpuTime();
             this.execute(() -> {
                 try {
                     task.run();
@@ -86,6 +98,15 @@ public void execute(Runnable task, Consumer<Throwable> failureHandler)
                         failureHandler.accept(t);
                     }
                 }
+                finally {
+                    long currentGCTime = getTotalGCTime();
+                    long cpuTimeInNanos = THREAD_MX_BEAN.getCurrentThreadCpuTime() - start - (currentGCTime - initialGCTime);
+
+                    statsTracker.recordEventLoopMethodExecutionCpuTime(cpuTimeInNanos);
+                    if (cpuTimeInNanos > slowMethodThresholdOnEventLoopInNanos) {
+                        log.warn("Slow method execution on event loop: %s took %s milliseconds", methodSignature, NANOSECONDS.toMillis(cpuTimeInNanos));
+                    }
+                }
             });
         }
 
@@ -108,4 +129,9 @@ public <T> void execute(Supplier<T> task, Consumer<T> successHandler, Consumer<T
             });
         }
     }
+
+    private long getTotalGCTime()
+    {
+        return gcBeans.stream().mapToLong(GarbageCollectorMXBean::getCollectionTime).sum();
+    }
 }

presto-main-base/src/main/java/com/facebook/presto/execution/SchedulerStatsTracker.java

@@ -25,11 +25,16 @@ public void recordTaskUpdateSerializedCpuTime(long nanos) {}
 
         @Override
         public void recordTaskPlanSerializedCpuTime(long nanos) {}
+
+        @Override
+        public void recordEventLoopMethodExecutionCpuTime(long nanos) {}
     };
 
     void recordTaskUpdateDeliveredTime(long nanos);
 
     void recordTaskUpdateSerializedCpuTime(long nanos);
 
     void recordTaskPlanSerializedCpuTime(long nanos);
+
+    void recordEventLoopMethodExecutionCpuTime(long nanos);
 }

presto-main-base/src/main/java/com/facebook/presto/execution/StageExecutionStateMachine.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
+import static com.facebook.presto.common.RuntimeMetricName.EVENT_LOOP_METHOD_EXECUTION_CPU_TIME_NANOS;
 import static com.facebook.presto.common.RuntimeMetricName.GET_SPLITS_TIME_NANOS;
 import static com.facebook.presto.common.RuntimeMetricName.SCAN_STAGE_SCHEDULER_BLOCKED_TIME_NANOS;
 import static com.facebook.presto.common.RuntimeMetricName.SCAN_STAGE_SCHEDULER_CPU_TIME_NANOS;
@@ -413,6 +414,12 @@ public void recordTaskPlanSerializedCpuTime(long nanos)
         runtimeStats.addMetricValue(TASK_PLAN_SERIALIZED_CPU_TIME_NANOS, NANO, max(nanos, 0));
     }
 
+    @Override
+    public void recordEventLoopMethodExecutionCpuTime(long nanos)
+    {
+        runtimeStats.addMetricValue(EVENT_LOOP_METHOD_EXECUTION_CPU_TIME_NANOS, NANO, max(nanos, 0));
+    }
+
     @Override
     public String toString()
     {

presto-main-base/src/main/java/com/facebook/presto/execution/TaskManagerConfig.java

@@ -33,6 +33,7 @@
 import java.math.BigDecimal;
 import java.util.concurrent.TimeUnit;
 
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 @DefunctConfig({
@@ -101,6 +102,20 @@ public class TaskManagerConfig
     private Duration highMemoryTaskKillerFrequentFullGCDurationThreshold = new Duration(1, SECONDS);
     private double highMemoryTaskKillerHeapMemoryThreshold = 0.9;
     private boolean enableEventLoop;
+    private Duration slowMethodThresholdOnEventLoop = new Duration(10, SECONDS);
+
+    @Min(50_000_000L)
+    public long getSlowMethodThresholdOnEventLoop()
+    {
+        return slowMethodThresholdOnEventLoop.roundTo(NANOSECONDS);
+    }
+
+    @Config("task.event-loop-slow-method-threshold")
+    public TaskManagerConfig setSlowMethodThresholdOnEventLoop(Duration slowMethodThresholdOnEventLoop)
+    {
+        this.slowMethodThresholdOnEventLoop = slowMethodThresholdOnEventLoop;
+        return this;
+    }
 
     @Config("task.enable-event-loop")
     public TaskManagerConfig setEventLoopEnabled(boolean enableEventLoop)

presto-main-base/src/test/java/com/facebook/presto/execution/TestTaskManagerConfig.java

@@ -79,6 +79,7 @@ public void testDefaults()
                 .setHighMemoryTaskKillerFrequentFullGCDurationThreshold(new Duration(1, SECONDS))
                 .setHighMemoryTaskKillerHeapMemoryThreshold(0.9)
                 .setTaskUpdateSizeTrackingEnabled(true)
+                .setSlowMethodThresholdOnEventLoop(new Duration(10, SECONDS))
                 .setEventLoopEnabled(false));
     }
 
@@ -129,6 +130,7 @@ public void testExplicitPropertyMappings()
                 .put("experimental.task.high-memory-task-killer-heap-memory-threshold", "0.8")
                 .put("task.update-size-tracking-enabled", "false")
                 .put("task.enable-event-loop", "true")
+                .put("task.event-loop-slow-method-threshold", "1s")
                 .build();
 
         TaskManagerConfig expected = new TaskManagerConfig()
@@ -174,7 +176,8 @@ public void testExplicitPropertyMappings()
                 .setHighMemoryTaskKillerFrequentFullGCDurationThreshold(new Duration(2, SECONDS))
                 .setHighMemoryTaskKillerHeapMemoryThreshold(0.8)
                 .setTaskUpdateSizeTrackingEnabled(false)
-                .setEventLoopEnabled(true);
+                .setEventLoopEnabled(true)
+                .setSlowMethodThresholdOnEventLoop(new Duration(1, SECONDS));
 
         assertFullMapping(properties, expected);
     }

presto-main/src/main/java/com/facebook/presto/server/remotetask/HttpRemoteTaskFactory.java

@@ -109,6 +109,7 @@ public class HttpRemoteTaskFactory
     private final DecayCounter taskUpdateRequestSize;
     private final boolean taskUpdateSizeTrackingEnabled;
     private final Optional<SafeEventLoopGroup> eventLoopGroup;
+    private final long slowMethodThresholdOnEventLoopInNanos;
 
     @Inject
     public HttpRemoteTaskFactory(
@@ -194,8 +195,9 @@ else if (binaryTransportEnabled) {
         this.taskUpdateRequestSize = new DecayCounter(ExponentialDecay.oneMinute());
         this.taskUpdateSizeTrackingEnabled = taskConfig.isTaskUpdateSizeTrackingEnabled();
 
+        this.slowMethodThresholdOnEventLoopInNanos = taskConfig.getSlowMethodThresholdOnEventLoop();
         this.eventLoopGroup = taskConfig.isEventLoopEnabled() ? Optional.of(new SafeEventLoopGroup(config.getRemoteTaskMaxCallbackThreads(),
-                new ThreadFactoryBuilder().setNameFormat("task-event-loop-%s").setDaemon(true).build())
+                new ThreadFactoryBuilder().setNameFormat("task-event-loop-%s").setDaemon(true).build(), this.slowMethodThresholdOnEventLoopInNanos)
         {
             @Override
             protected EventLoop newChild(Executor executor, Object... args)

presto-main/src/main/java/com/facebook/presto/server/remotetask/HttpRemoteTaskWithEventLoop.java

@@ -232,6 +232,7 @@ public final class HttpRemoteTaskWithEventLoop
     private final SchedulerStatsTracker schedulerStatsTracker;
 
     private final SafeEventLoopGroup.SafeEventLoop taskEventLoop;
+    private final String loggingPrefix;
 
     public static HttpRemoteTaskWithEventLoop createHttpRemoteTaskWithEventLoop(
             Session session,
@@ -469,6 +470,7 @@ private HttpRemoteTaskWithEventLoop(Session session,
                 handleResolver,
                 connectorTypeSerdeManager,
                 thriftProtocol);
+        this.loggingPrefix = format("Query: %s, Task: %s", session.getQueryId(), taskId);
     }
 
     // this is a separate method to ensure that the `this` reference is not leaked during construction
@@ -488,7 +490,7 @@ private void initialize()
         });
 
         updateTaskStats();
-        safeExecuteOnEventLoop(this::updateSplitQueueSpace);
+        safeExecuteOnEventLoop(this::updateSplitQueueSpace, "updateSplitQueueSpace");
     }
 
     @Override
@@ -537,7 +539,7 @@ public void start()
 
             taskStatusFetcher.start();
             taskInfoFetcher.start();
-        });
+        }, "start");
     }
 
     @Override
@@ -581,7 +583,7 @@ public void addSplits(Multimap<PlanNodeId, Split> splitsBySource)
                 needsUpdate = true;
                 scheduleUpdate();
             }
-        });
+        }, "addSplits");
     }
 
     @Override
@@ -595,7 +597,7 @@ public void noMoreSplits(PlanNodeId sourceId)
             noMoreSplits.put(sourceId, true);
             needsUpdate = true;
             scheduleUpdate();
-        });
+        }, "noMoreSplits");
     }
 
     @Override
@@ -606,7 +608,7 @@ public void noMoreSplits(PlanNodeId sourceId, Lifespan lifespan)
                 needsUpdate = true;
                 scheduleUpdate();
             }
-        });
+        }, "noMoreSplits with lifeSpan");
     }
 
     @Override
@@ -622,7 +624,7 @@ public void setOutputBuffers(OutputBuffers newOutputBuffers)
                 needsUpdate = true;
                 scheduleUpdate();
             }
-        });
+        }, "setOutputBuffers");
     }
 
     @Override
@@ -793,7 +795,7 @@ public ListenableFuture<?> whenSplitQueueHasSpace(long weightThreshold)
                 future.set(null);
             }
             whenSplitQueueHasSpace.createNewListener().addListener(() -> future.set(null), taskEventLoop);
-        });
+        }, "whenSplitQueueHasSpace");
         return future;
     }
 
@@ -1042,7 +1044,7 @@ private void sendUpdate()
                     future,
                     new SimpleHttpResponseHandler<>(new UpdateResponseHandler(sources), request.getUri(), stats.getHttpResponseStats(), REMOTE_TASK_ERROR),
                     taskEventLoop);
-        });
+        }, "sendUpdate");
     }
 
     private String getExceededTaskUpdateSizeMessage(byte[] taskUpdateRequestJson)
@@ -1091,7 +1093,7 @@ public void cancel()
             Request request = builder.setUri(uriBuilder.build())
                     .build();
             scheduleAsyncCleanupRequest(createCleanupBackoff(), request, "cancel");
-        });
+        }, "cancel");
     }
 
     private void cleanUpTask()
@@ -1129,7 +1131,7 @@ private void cleanUpTask()
                     .build();
 
             scheduleAsyncCleanupRequest(createCleanupBackoff(), request, "cleanup");
-        });
+        }, "cleanUpTask");
     }
 
     @Override
@@ -1159,7 +1161,7 @@ private void abort(TaskStatus status)
             Request request = builder.setUri(uriBuilder.build())
                     .build();
             scheduleAsyncCleanupRequest(createCleanupBackoff(), request, "abort");
-        });
+        }, "abort");
     }
 
     private void scheduleAsyncCleanupRequest(Backoff cleanupBackoff, Request request, String action)
@@ -1407,8 +1409,8 @@ public void onFailure(Throwable throwable)
         }
     }
 
-    private void safeExecuteOnEventLoop(Runnable r)
+    private void safeExecuteOnEventLoop(Runnable r, String methodName)
     {
-        taskEventLoop.execute(r, this::failTask);
+        taskEventLoop.execute(r, this::failTask, schedulerStatsTracker, loggingPrefix + ", method: " + methodName);
     }
 }