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"); + } + +}