Refactor TaskMetrics tracking to event-driven approach

Author: sanjana2505006

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java

@@ -156,6 +156,40 @@ private void doFireSingleEvent(final IWorkflowExecutionRunnable workflowExecutio
             throw new RuntimeException("No EventHandler found for event: " + event.getEventType());
         }
         lifecycleEventHandler.handle(workflowExecutionRunnable, event);
+
+        recordTaskInstanceMetrics(event);
+    }
+
+    private void recordTaskInstanceMetrics(AbstractLifecycleEvent event) {
+        if (!(event
+                .getEventType() instanceof org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType)) {
+            return;
+        }
+
+        switch ((org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType) event
+                .getEventType()) {
+            case DISPATCHED:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("dispatch");
+                break;
+            case SUCCEEDED:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("success");
+                break;
+            case FAILED:
+            case FATAL:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("fail");
+                break;
+            case KILLED:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("kill");
+                break;
+            case RETRY:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("retry");
+                break;
+            case TIMEOUT:
+                org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("timeout");
+                break;
+            default:
+                break;
+        }
     }
 
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskTimeoutLifecycleEventHandler.java

@@ -54,7 +54,6 @@ public void handle(final ITaskStateAction taskStateAction,
             // The task instance is not active, means it is already finished.
             return;
         }
-        org.apache.dolphinscheduler.server.master.metrics.TaskMetrics.incTaskInstanceByState("timeout");
         final String taskName = taskExecutionRunnable.getName();
         final TaskTimeoutStrategy timeoutNotifyStrategy = taskTimeoutLifecycleEvent.getTimeoutStrategy();
         if (timeoutNotifyStrategy == null) {

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java

@@ -46,7 +46,6 @@
 import org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
-import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -110,7 +109,6 @@ public void onFatalEvent(final IWorkflowExecutionRunnable workflowExecutionRunna
 
         if (taskExecutionRunnable.isTaskInstanceCanRetry()) {
             taskExecutionRunnable.getWorkflowEventBus().publish(TaskRetryLifecycleEvent.of(taskExecutionRunnable));
-            TaskMetrics.incTaskInstanceByState("retry");
             return;
         }
 
@@ -133,7 +131,6 @@ private void persistentTaskInstanceFatalEventToDB(final ITaskExecutionRunnable t
         taskInstance.setState(TaskExecutionStatus.FAILURE);
         taskInstance.setEndTime(taskFatalEvent.getEndTime());
         taskInstanceDao.updateById(taskInstance);
-        TaskMetrics.incTaskInstanceByState("fail");
     }
 
     @Override
@@ -144,7 +141,6 @@ public void onDispatchedEvent(final IWorkflowExecutionRunnable workflowExecution
         taskInstance.setState(DISPATCH);
         taskInstance.setHost(taskDispatchedEvent.getExecutorHost());
         taskInstanceDao.updateById(taskInstance);
-        TaskMetrics.incTaskInstanceByState("dispatch");
     }
 
     @Override
@@ -200,7 +196,6 @@ private void persistentTaskInstanceKilledEventToDB(final ITaskExecutionRunnable
         taskInstance.setState(TaskExecutionStatus.KILL);
         taskInstance.setEndTime(taskKilledEvent.getEndTime());
         taskInstanceDao.updateById(taskInstance);
-        TaskMetrics.incTaskInstanceByState("kill");
     }
 
     @Override
@@ -212,7 +207,6 @@ public void onFailedEvent(final IWorkflowExecutionRunnable workflowExecutionRunn
 
         if (taskExecutionRunnable.isTaskInstanceCanRetry()) {
             taskExecutionRunnable.getWorkflowEventBus().publish(TaskRetryLifecycleEvent.of(taskExecutionRunnable));
-            TaskMetrics.incTaskInstanceByState("retry");
             return;
         }
         // If all successors are condition tasks, then the task will not be marked as failure.
@@ -234,7 +228,6 @@ private void persistentTaskInstanceFailedEventToDB(final ITaskExecutionRunnable
         taskInstance.setState(TaskExecutionStatus.FAILURE);
         taskInstance.setEndTime(taskFailedEvent.getEndTime());
         taskInstanceDao.updateById(taskInstance);
-        TaskMetrics.incTaskInstanceByState("fail");
     }
 
     @Override
@@ -265,7 +258,6 @@ protected void persistentTaskInstanceSuccessEventToDB(final ITaskExecutionRunnab
                 JSONUtils.toJsonString(taskSuccessEvent.getVarPool()));
         taskInstance.setVarPool(VarPoolUtils.serializeVarPool(finalVarPool));
         taskInstanceDao.updateById(taskInstance);
-        TaskMetrics.incTaskInstanceByState("success");
     }
 
     /**