[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

@@ -25,8 +25,10 @@
 import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
 import org.apache.dolphinscheduler.server.master.engine.exceptions.WorkflowEventFireException;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
+import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFatalLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+import org.apache.dolphinscheduler.server.master.metrics.TaskMetrics;
 import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
 import org.apache.dolphinscheduler.server.master.utils.ExceptionUtils;
 
@@ -156,6 +158,16 @@ private void doFireSingleEvent(final IWorkflowExecution workflowExecution,
             throw new RuntimeException("No EventHandler found for event: " + event.getEventType());
         }
         lifecycleEventHandler.handle(workflowExecution, event);
+        recordTaskInstanceMetrics(event);
+    }
+
+    private void recordTaskInstanceMetrics(AbstractLifecycleEvent event) {
+        final ILifecycleEventType eventType = event.getEventType();
+        if (!(eventType instanceof TaskLifecycleEventType)) {
+            return;
+        }
+        final TaskLifecycleEventType taskLifecycleEventType = (TaskLifecycleEventType) eventType;
+        TaskMetrics.incTaskInstanceByLifecycleEvent(taskLifecycleEventType);
     }
 
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/execution/WorkflowExecutionFactory.java

@@ -22,6 +22,7 @@
 import org.apache.dolphinscheduler.dao.repository.CommandDao;
 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;
 
@@ -48,8 +49,11 @@ public class WorkflowExecutionFactory {
      */
     @Transactional
     public IWorkflowExecution createWorkflowExecuteRunnable(Command command) {
+        long startTime = System.currentTimeMillis();
         deleteCommandOrThrow(command);
-        return doCreateWorkflowExecution(command);
+        IWorkflowExecution workflowExecution = doCreateWorkflowExecution(command);
+        WorkflowInstanceMetrics.recordWorkflowInstanceGenerateTime(System.currentTimeMillis() - startTime);
+        return workflowExecution;
     }
 
     /**

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

@@ -179,6 +179,9 @@ protected void transformWorkflowInstanceState(final IWorkflowExecution workflowE
             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

@@ -17,37 +17,25 @@
 
 package org.apache.dolphinscheduler.server.master.metrics;
 
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
+import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
+
 import java.util.function.Supplier;
 
 import lombok.experimental.UtilityClass;
-
-import com.google.common.collect.ImmutableSet;
-
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.Metrics;
 
 @UtilityClass
 public class TaskMetrics {
 
-    private final Map<String, Counter> taskInstanceCounters = new HashMap<>();
-
-    private final Set<String> taskInstanceStates = ImmutableSet.of(
-            "submit", "timeout", "finish", "failover", "retry", "dispatch", "success", "kill", "fail", "stop");
-
     static {
-        for (final String state : taskInstanceStates) {
-            taskInstanceCounters.put(
-                    state,
-                    Counter.builder("ds.task.instance.count")
-                            .tags("state", state)
-                            .description(String.format("Workflow instance %s total count", state))
-                            .register(Metrics.globalRegistry));
+        for (final TaskLifecycleEventType eventType : TaskLifecycleEventType.values()) {
+            Counter.builder("ds.task.instance.count")
+                    .tags("state", eventType.name())
+                    .description(String.format("Task instance %s total count", eventType.name()))
+                    .register(Metrics.globalRegistry);
         }
-
     }
 
     private final Counter taskDispatchCounter =
@@ -83,11 +71,14 @@ public void incTaskDispatch() {
         taskDispatchCounter.increment();
     }
 
-    public void incTaskInstanceByState(final String state) {
-        if (taskInstanceCounters.get(state) == null) {
+    public void incTaskInstanceByLifecycleEvent(final TaskLifecycleEventType eventType) {
+        if (eventType == null) {
             return;
         }
-        taskInstanceCounters.get(state).increment();
+        Metrics.globalRegistry.counter(
+                "ds.task.instance.count",
+                "state", eventType.name())
+                .increment();
     }
 
 }

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

@@ -0,0 +1,57 @@
+/*
+ * 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 org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
+
+import org.junit.jupiter.api.Test;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+
+class TaskMetricsTest {
+
+    @Test
+    void testRegisterTaskPrepared() {
+        TaskMetrics.registerTaskPrepared(() -> 5);
+        assertNotNull(Metrics.globalRegistry.find("ds.task.prepared").gauge(),
+                "Task prepared gauge should be registered");
+    }
+
+    @Test
+    void testIncTaskInstanceByLifecycleEvent() {
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.DISPATCHED);
+        Counter counter = Metrics.globalRegistry.find("ds.task.instance.count")
+                .tag("state", TaskLifecycleEventType.DISPATCHED.name())
+                .counter();
+        assertNotNull(counter);
+        assertEquals(1, counter.count(), 0.001);
+
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.SUCCEEDED);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.FAILED);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.KILLED);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.RETRY);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.TIMEOUT);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(TaskLifecycleEventType.FATAL);
+        TaskMetrics.incTaskInstanceByLifecycleEvent(null);
+    }
+
+}

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

@@ -0,0 +1,154 @@
+/*
+ * 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 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", WorkflowExecutionStatus.SUBMITTED_SUCCESS.name())
+                .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", WorkflowExecutionStatus.FAILURE.name())
+                .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", WorkflowExecutionStatus.SUCCESS.name())
+                .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", WorkflowExecutionStatus.STOP.name())
+                .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", WorkflowExecutionStatus.PAUSE.name())
+                .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", WorkflowExecutionStatus.FAILOVER.name())
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for failover state");
+        assertEquals(1, counter.count(), 0.001);
+    }
+
+    @Test
+    void testIncWorkflowInstanceByStateAndWorkflowDefinitionCode_runningState() {
+        String defCode = "test_running_1";
+        WorkflowInstanceMetrics.incWorkflowInstanceByStateAndWorkflowDefinitionCode(
+                WorkflowExecutionStatus.RUNNING_EXECUTION, defCode);
+        Counter counter = Metrics.globalRegistry.find("ds.workflow.instance.count")
+                .tag("state", WorkflowExecutionStatus.RUNNING_EXECUTION.name())
+                .tag("workflow.definition.code", defCode)
+                .counter();
+        assertNotNull(counter, "Counter should be registered for running 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");
+    }
+
+}