From 65164d6cf631738e804589394385d989ddc38fb0 Mon Sep 17 00:00:00 2001 From: Sanjana Date: Tue, 9 Jun 2026 20:49:00 +0530 Subject: [PATCH] [Improvement-18039][Metrics] Add missing metrics for workflow and task state transitions Record workflow instance state changes in AbstractWorkflowStateAction, workflow generation duration in WorkflowExecutionFactory, and task lifecycle metrics via TaskLifecycleEventType in WorkflowEventBusFireWorker. Add unit tests for TaskMetrics and WorkflowInstanceMetrics. --- .../engine/WorkflowEventBusFireWorker.java | 12 ++ .../execution/WorkflowExecutionFactory.java | 6 +- .../AbstractWorkflowStateAction.java | 3 + .../server/master/metrics/TaskMetrics.java | 35 ++-- .../master/metrics/TaskMetricsTest.java | 57 +++++++ .../metrics/WorkflowInstanceMetricsTest.java | 154 ++++++++++++++++++ 6 files changed, 244 insertions(+), 23 deletions(-) create mode 100644 dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetricsTest.java create mode 100644 dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/WorkflowInstanceMetricsTest.java diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java index 46c5cb32b02c..b95165139e69 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java +++ b/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); } } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/execution/WorkflowExecutionFactory.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/execution/WorkflowExecutionFactory.java index 1c495c6a139e..9ac19f5d29ef 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/execution/WorkflowExecutionFactory.java +++ b/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; } /** diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java index 48b329bc771c..2789fbabb546 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java +++ b/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; diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetrics.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetrics.java index 8ea9f13c4dae..db1b6d597d98 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetrics.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetrics.java @@ -17,15 +17,11 @@ 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; @@ -33,21 +29,13 @@ @UtilityClass public class TaskMetrics { - private final Map taskInstanceCounters = new HashMap<>(); - - private final Set 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(); } } diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetricsTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/TaskMetricsTest.java new file mode 100644 index 000000000000..e45135d490f8 --- /dev/null +++ b/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); + } + +} diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/WorkflowInstanceMetricsTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/metrics/WorkflowInstanceMetricsTest.java new file mode 100644 index 000000000000..b9918536cc5a --- /dev/null +++ b/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"); + } + +}