Add counter for tasks created on worker

trinodb · Jan 8, 2025 · b27dd3b · b27dd3b
1 parent 509508a
commit b27dd3b
Showing 1 changed file with 27 additions and 16 deletions.
diff --git a/core/trino-main/src/main/java/io/trino/execution/SqlTaskManager.java b/core/trino-main/src/main/java/io/trino/execution/SqlTaskManager.java
@@ -136,6 +136,7 @@ public class SqlTaskManager
 
     private final long queryMaxMemoryPerNode;
 
+    private final CounterStat createdTasks = new CounterStat();
     private final CounterStat failedTasks = new CounterStat();
     private final Optional<StuckSplitTasksInterrupter> stuckSplitTasksInterrupter;
     private final LanguageFunctionProvider languageFunctionProvider;
@@ -230,22 +231,25 @@ public SqlTaskManager(
                 queryId -> createQueryContext(queryId, localMemoryManager, localSpillManager, gcMonitor, maxQueryMemoryPerNode, maxQuerySpillPerNode)));
 
         tasks = buildNonEvictableCache(CacheBuilder.newBuilder(), CacheLoader.from(
-                taskId -> createSqlTask(
-                        taskId,
-                        locationFactory.createLocalTaskLocation(taskId),
-                        nodeInfo.getNodeId(),
-                        queryContexts.getUnchecked(taskId.getQueryId()),
-                        tracer,
-                        sqlTaskExecutionFactory,
-                        taskNotificationExecutor,
-                        sqlTask -> {
-                            languageFunctionProvider.unregisterTask(taskId);
-                            finishedTaskStats.merge(sqlTask.getIoStats());
-                        },
-                        maxBufferSize,
-                        maxBroadcastBufferSize,
-                        requireNonNull(exchangeManagerRegistry, "exchangeManagerRegistry is null"),
-                        failedTasks)));
+                taskId -> {
+                    createdTasks.update(1);
+                    return createSqlTask(
+                            taskId,
+                            locationFactory.createLocalTaskLocation(taskId),
+                            nodeInfo.getNodeId(),
+                            queryContexts.getUnchecked(taskId.getQueryId()),
+                            tracer,
+                            sqlTaskExecutionFactory,
+                            taskNotificationExecutor,
+                            sqlTask -> {
+                                languageFunctionProvider.unregisterTask(taskId);
+                                finishedTaskStats.merge(sqlTask.getIoStats());
+                            },
+                            maxBufferSize,
+                            maxBroadcastBufferSize,
+                            requireNonNull(exchangeManagerRegistry, "exchangeManagerRegistry is null"),
+                            failedTasks);
+                }));
 
         stuckSplitTasksInterrupter = createStuckSplitTasksInterrupter(
                 config.isInterruptStuckSplitTasksEnabled(),
@@ -355,6 +359,13 @@ public ThreadPoolExecutorMBean getTaskNotificationExecutor()
         return taskNotificationExecutorMBean;
     }
 
+    @Managed(description = "Created tasks counter")
+    @Nested
+    public CounterStat getCreatedTasks()
+    {
+        return createdTasks;
+    }
+
     @Managed(description = "Failed tasks counter")
     @Nested
     public CounterStat getFailedTasks()