[Improvement-18039][Metrics] Add missing metrics for workflow and task state transitions

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/statemachine/AbstractTaskStateAction.java

@@ -196,7 +196,6 @@ private void persistentTaskInstanceKilledEventToDB(final ITaskExecutionRunnable
         taskInstance.setState(TaskExecutionStatus.KILL);
         taskInstance.setEndTime(taskKilledEvent.getEndTime());
         taskInstanceDao.updateById(taskInstance);
-
     }
 
     @Override

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/runnable/WorkflowExecutionRunnableFactory.java

@@ -23,6 +23,7 @@
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.server.master.engine.command.ICommandHandler;
 import org.apache.dolphinscheduler.server.master.engine.exceptions.CommandDuplicateHandleException;
+import org.apache.dolphinscheduler.server.master.metrics.WorkflowInstanceMetrics;
 
 import java.util.List;
 
@@ -52,8 +53,11 @@ public class WorkflowExecutionRunnableFactory {
      */
     @Transactional
     public IWorkflowExecutionRunnable createWorkflowExecuteRunnable(Command command) {
+        long startTime = System.currentTimeMillis();
         deleteCommandOrThrow(command);
-        return doCreateWorkflowExecutionRunnable(command);
+        IWorkflowExecutionRunnable workflowExecutionRunnable = doCreateWorkflowExecutionRunnable(command);
+        WorkflowInstanceMetrics.recordWorkflowInstanceGenerateTime(System.currentTimeMillis() - startTime);
+        return workflowExecutionRunnable;
     }
 
     /**

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java

@@ -176,6 +176,9 @@ protected void transformWorkflowInstanceState(final IWorkflowExecutionRunnable w
             workflowInstanceDao.updateById(workflowInstance);
             log.info("Success set WorkflowExecuteRunnable: {} state from: {} to {}",
                     workflowInstance.getName(), originState.name(), targetState.name());
+            WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                    targetState,
+                    String.valueOf(workflowInstance.getWorkflowDefinitionCode()));
         } catch (Exception ex) {
             workflowInstance.setState(originState);
             throw ex;

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetrics.java

@@ -50,39 +50,12 @@ public class TaskMetrics {
 
     }
 
-    private final Counter taskDispatchCounter =
-            Counter.builder("ds.task.dispatch.count")
-                    .description("Task dispatch count")
-                    .register(Metrics.globalRegistry);
-
-    private final Counter taskDispatchFailCounter =
-            Counter.builder("ds.task.dispatch.failure.count")
-                    .description("Task dispatch failures count, retried ones included")
-                    .register(Metrics.globalRegistry);
-
-    private final Counter taskDispatchErrorCounter =
-            Counter.builder("ds.task.dispatch.error.count")
-                    .description("Number of errors during task dispatch")
-                    .register(Metrics.globalRegistry);
-
     public synchronized void registerTaskPrepared(Supplier<Number> consumer) {
         Gauge.builder("ds.task.prepared", consumer)
                 .description("Task prepared count")
                 .register(Metrics.globalRegistry);
     }
 
-    public void incTaskDispatchFailed(int failedCount) {
-        taskDispatchFailCounter.increment(failedCount);
-    }
-
-    public void incTaskDispatchError() {
-        taskDispatchErrorCounter.increment();
-    }
-
-    public void incTaskDispatch() {
-        taskDispatchCounter.increment();
-    }
-
     public void incTaskInstanceByState(final String state) {
         if (taskInstanceCounters.get(state) == null) {
             return;

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetricsTest.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+
+class TaskMetricsTest {
+
+    @Test
+    void testIncTaskInstanceByState_validStates() {
+        List<String> validStates = Arrays.asList(
+                "submit", "timeout", "finish", "failover", "retry", "dispatch", "success", "kill", "fail", "stop");
+
+        for (String state : validStates) {
+            Counter counter = Metrics.globalRegistry.find("ds.task.instance.count")
+                    .tag("state", state)
+                    .counter();
+            assertNotNull(counter, "Counter should exist for state: " + state);
+            double before = counter.count();
+            TaskMetrics.incTaskInstanceByState(state);
+            assertEquals(before + 1, counter.count(), 0.001,
+                    "Counter should be incremented for state: " + state);
+        }
+    }
+
+    @Test
+    void testIncTaskInstanceByState_invalidState() {
+        TaskMetrics.incTaskInstanceByState("nonexistent_state");
+        Counter counter = Metrics.globalRegistry.find("ds.task.instance.count")
+                .tag("state", "nonexistent_state")
+                .counter();
+        assertTrue(counter == null || counter.count() == 0,
+                "Counter should not exist or be zero for invalid state");
+    }
+
+    @Test
+    void testIncTaskInstanceByState_multipleIncrements() {
+        Counter counter = Metrics.globalRegistry.find("ds.task.instance.count")
+                .tag("state", "submit")
+                .counter();
+        assertNotNull(counter);
+        double before = counter.count();
+
+        TaskMetrics.incTaskInstanceByState("submit");
+        TaskMetrics.incTaskInstanceByState("submit");
+        TaskMetrics.incTaskInstanceByState("submit");
+
+        assertEquals(before + 3, counter.count(), 0.001,
+                "Counter should be incremented by 3 after three calls");
+    }
+
+    @Test
+    void testRegisterTaskPrepared() {
+        TaskMetrics.registerTaskPrepared(() -> 5);
+        assertNotNull(Metrics.globalRegistry.find("ds.task.prepared").gauge(),
+                "Task prepared gauge should be registered");
+    }
+
+}

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/WorkflowInstanceMetricsTest.java

@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+
+import org.junit.jupiter.api.Test;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Timer;
+
+class WorkflowInstanceMetricsTest {
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_submitState() {
+        String defCode = "test_submit_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.SUBMITTED_SUCCESS, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "submit")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for submit state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_failureState() {
+        String defCode = "test_failure_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.FAILURE, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "fail")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for fail state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_successState() {
+        String defCode = "test_success_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.SUCCESS, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "success")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for success state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_stopState() {
+        String defCode = "test_stop_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.STOP, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "stop")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for stop state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_pauseState() {
+        String defCode = "test_pause_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.PAUSE, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "pause")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for pause state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_failoverState() {
+        String defCode = "test_failover_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.FAILOVER, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "failover")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for failover state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_defaultMapping() {
+        String defCode = "test_running_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.RUNNING_EXECUTION, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "running_execution")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for default-mapped state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testRecordCommandQueryTime() {
+        WorkflowInstanceMetrics.recordCommandQueryTime(100L);
+        Timer timer = Metrics.globalRegistry.find("ds.workflow.command.query.duration").timer();
+        assertNotNull(timer, "Command query timer should be registered");
+        assertEquals(1, timer.count(), "Timer should have recorded one event");
+    }
+
+    @Test
+    void testRecordWorkflowInstanceGenerateTime() {
+        WorkflowInstanceMetrics.recordWorkflowInstanceGenerateTime(200L);
+        Timer timer = Metrics.globalRegistry.find("ds.workflow.instance.generate.duration").timer();
+        assertNotNull(timer, "Workflow instance generate timer should be registered");
+        assertEquals(1, timer.count(), "Timer should have recorded one event");
+    }
+
+    @Test
+    void testRegisterWorkflowInstanceRunningGauge() {
+        WorkflowInstanceMetrics.registerWorkflowInstanceRunningGauge(() -> 10);
+        assertNotNull(Metrics.globalRegistry.find("ds.workflow.instance.running").gauge(),
+                "Running gauge should be registered");
+    }
+
+    @Test
+    void testRegisterWorkflowInstanceResubmitGauge() {
+        WorkflowInstanceMetrics.registerWorkflowInstanceResubmitGauge(() -> 3);
+        assertNotNull(Metrics.globalRegistry.find("ds.workflow.instance.resubmit").gauge(),
+                "Resubmit gauge should be registered");
+    }
+
+    @Test
+    void testCleanUpWorkflowInstanceCountMetricsByDefinitionCode() {
+        String defCode = "99999";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.SUCCESS, defCode);
+        Counter counterBefore = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "success")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counterBefore, "Counter should exist before cleanup");
+
+        WorkflowInstanceMetrics.cleanUpWorkflowInstanceCountMetricsByDefinitionCode(99999L);
+
+        Counter counterAfter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", "success")
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNull(counterAfter, "Counter should be removed after cleanup");
+    }
+
+}